urcu/workqueue-fifo.h

   1 #ifndef _URCU_WORKQUEUE_FIFO_H
   2 #define _URCU_WORKQUEUE_FIFO_H
   3
   4 /*
   5  * urcu/workqueue-fifo.h
   6  *
   7  * Userspace RCU library - work queue scheme with FIFO semantic
   8  *
   9  * Copyright (c) 2014 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  10  *
  11  * This library is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU Lesser General Public
  13  * License as published by the Free Software Foundation; either
  14  * version 2.1 of the License, or (at your option) any later version.
  15  *
  16  * This library is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  19  * Lesser General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU Lesser General Public
  22  * License along with this library; if not, write to the Free Software
  23  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  24  */
  25
  26 #include <urcu/uatomic.h>
  27 #include <urcu/lfstack.h>
  28 #include <urcu/waitqueue-lifo.h>
  29 #include <urcu/wfcqueue.h>
  30 #include <urcu/rculist.h>
  31 #include <pthread.h>
  32 #include <assert.h>
  33
  34 enum urcu_accept_ret {
  35         URCU_ACCEPT_WORK        = 0,
  36         URCU_ACCEPT_SHUTDOWN    = 1,
  37 };
  38
  39 /*
  40  * We use RCU to steal work from siblings. Therefore, one of RCU flavors
  41  * need to be included before this header. All worker that participate
  42  * in stealing (initialized with the URCU_WORKER_STEAL flag) need to be
  43  * registered RCU readers threads.
  44  */
  45
  46 struct urcu_work {
  47         struct cds_wfcq_node node;
  48 };
  49
  50 struct urcu_workqueue {
  51         /* FIFO work queue */
  52         struct __cds_wfcq_head head;
  53         struct cds_wfcq_tail tail;
  54
  55         /* Associated wait queue for LIFO wait/wakeup */
  56         struct urcu_wait_queue waitqueue;
  57
  58         /* RCU linked list head of siblings for work stealing. */
  59         struct cds_list_head sibling_head;
  60         pthread_mutex_t sibling_lock;   /* Protect sibling list updates */
  61
  62         bool shutdown;                  /* Shutdown performed */
  63 };
  64
  65 struct urcu_worker {
  66         struct cds_wfcq_head head;
  67         struct cds_wfcq_tail tail;
  68
  69         struct urcu_wait_node wait_node;
  70         /* RCU linked list node of siblings for work stealing. */
  71         struct cds_list_head sibling_node;
  72         int flags;      /* enum urcu_worker_flags */
  73 };
  74
  75 enum urcu_worker_flags {
  76         URCU_WORKER_STEAL       = (1 << 0),
  77 };
  78
  79 static inline
  80 void urcu_workqueue_init(struct urcu_workqueue *queue)
  81 {
  82         __cds_wfcq_init(&queue->head, &queue->tail);
  83         urcu_wait_queue_init(&queue->waitqueue);
  84         CDS_INIT_LIST_HEAD(&queue->sibling_head);
  85         queue->shutdown = false;
  86 }
  87
  88 static inline
  89 void urcu_queue_work(struct urcu_workqueue *queue, struct urcu_work *work)
  90 {
  91         bool was_empty;
  92
  93         cds_wfcq_node_init(&work->node);
  94
  95         /* Enqueue work. */
  96         was_empty = !cds_wfcq_enqueue(&queue->head, &queue->tail,
  97                         &work->node);
  98         /*
  99          * If workqueue was previously empty, wakeup one worker thread.
 100          * It will eventually grab the entire content of the work-queue
 101          * (therefore grabbing a "work batch"). After having grabbed the
 102          * work batch, while that thread is running and taking care of
 103          * that work batch, when we enqueue more work, we will wake
 104          * another thread (if there is one waiting), which will
 105          * eventually grab the new batch, and so on. This scheme ensures
 106          * that contiguous batch of work are handled by the same thread
 107          * (for locality), and also ensures that we scale work to many
 108          * worker threads when threads are busy enough to still be
 109          * running when work is enqueued.
 110          */
 111         if (was_empty) {
 112                 rcu_read_lock();        /* Protect stack dequeue */
 113                 (void) urcu_dequeue_wake_single(&queue->waitqueue);
 114                 rcu_read_unlock();      /* Protect stack dequeue */
 115         }
 116 }
 117
 118 static inline
 119 void __urcu_workqueue_wakeup_all(struct urcu_workqueue *queue)
 120 {
 121         struct urcu_waiters waiters;
 122
 123         rcu_read_lock();        /* Protect stack dequeue */
 124         urcu_move_waiters(&waiters, &queue->waitqueue);
 125         rcu_read_unlock();      /* Protect stack dequeue */
 126
 127         (void) urcu_wake_all_waiters(&waiters);
 128 }
 129
 130 static inline
 131 void urcu_worker_init(struct urcu_worker *worker, int flags)
 132 {
 133         cds_wfcq_init(&worker->head, &worker->tail);
 134         worker->flags = flags;
 135         urcu_wait_node_init(&worker->wait_node, URCU_WAIT_RUNNING);
 136 }
 137
 138 static inline
 139 void urcu_worker_register(struct urcu_workqueue *queue,
 140                 struct urcu_worker *worker)
 141 {
 142         if (worker->flags & URCU_WORKER_STEAL) {
 143                 pthread_mutex_lock(&queue->sibling_lock);
 144                 cds_list_add_rcu(&worker->sibling_node, &queue->sibling_head);
 145                 pthread_mutex_unlock(&queue->sibling_lock);
 146         }
 147 }
 148
 149 static inline
 150 void urcu_worker_unregister(struct urcu_workqueue *queue,
 151                 struct urcu_worker *worker)
 152 {
 153         enum cds_wfcq_ret wfcq_ret;
 154
 155         if (worker->flags & URCU_WORKER_STEAL) {
 156                 pthread_mutex_lock(&queue->sibling_lock);
 157                 cds_list_del_rcu(&worker->sibling_node);
 158                 pthread_mutex_unlock(&queue->sibling_lock);
 159         }
 160
 161         /*
 162          * Make sure we are removed from waitqueue.
 163          */
 164         if (CMM_LOAD_SHARED(worker->wait_node.node.next))
 165                 __urcu_workqueue_wakeup_all(queue);
 166
 167         /*
 168          * Put any local work we still have back into the workqueue.
 169          */
 170         wfcq_ret = __cds_wfcq_splice_blocking(&queue->head,
 171                         &queue->tail,
 172                         &worker->head,
 173                         &worker->tail);
 174         if (wfcq_ret != CDS_WFCQ_RET_SRC_EMPTY
 175                         && wfcq_ret == CDS_WFCQ_RET_DEST_EMPTY) {
 176                 /*
 177                  * Wakeup worker thread if we have put work back into
 178                  * workqueue that was previously empty.
 179                  */
 180                 rcu_read_lock();        /* Protect stack dequeue */
 181                 (void) urcu_dequeue_wake_single(&queue->waitqueue);
 182                 rcu_read_unlock();      /* Protect stack dequeue */
 183         }
 184
 185         /*
 186          * Wait for grace period before freeing or reusing
 187          * "worker" because used by RCU linked list.
 188          * Also prevents ABA for waitqueue stack dequeue: matches RCU
 189          * read-side critical sections around dequeue and move all
 190          * operations on waitqueue).
 191          */
 192         synchronize_rcu();
 193 }
 194
 195 /*
 196  * Try stealing work from siblings when we have nothing to do.
 197  */
 198 static inline
 199 bool ___urcu_steal_work(struct urcu_worker *worker,
 200                 struct urcu_worker *sibling)
 201 {
 202         enum cds_wfcq_ret splice_ret;
 203
 204         /*
 205          * Don't bother grabbing the sibling queue lock if it is empty.
 206          */
 207         if (cds_wfcq_empty(&sibling->head, &sibling->tail))
 208                 return false;
 209         splice_ret = cds_wfcq_splice_blocking(&worker->head,
 210                         &worker->tail,
 211                         &sibling->head,
 212                         &sibling->tail);
 213         /* Ensure that we preserve FIFO work order. */
 214         assert(splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
 215         return splice_ret != CDS_WFCQ_RET_SRC_EMPTY;
 216 }
 217
 218 static inline
 219 bool __urcu_steal_work(struct urcu_workqueue *queue,
 220                 struct urcu_worker *worker)
 221 {
 222         struct urcu_worker *sibling_prev, *sibling_next;
 223         struct cds_list_head *sibling_node;
 224         bool steal_performed = 0;
 225
 226         if (!(worker->flags & URCU_WORKER_STEAL))
 227                 return false;
 228
 229         rcu_read_lock();
 230
 231         sibling_node = rcu_dereference(worker->sibling_node.next);
 232         if (sibling_node == &queue->sibling_head)
 233                 sibling_node = rcu_dereference(sibling_node->next);
 234         sibling_next = caa_container_of(sibling_node, struct urcu_worker,
 235                         sibling_node);
 236         if (sibling_next != worker)
 237                 steal_performed = ___urcu_steal_work(worker, sibling_next);
 238         if (steal_performed)
 239                 goto end;
 240
 241         sibling_node = rcu_dereference(worker->sibling_node.prev);
 242         if (sibling_node == &queue->sibling_head)
 243                 sibling_node = rcu_dereference(sibling_node->prev);
 244         sibling_prev = caa_container_of(sibling_node, struct urcu_worker,
 245                         sibling_node);
 246         if (sibling_prev != worker && sibling_prev != sibling_next)
 247                 steal_performed =  ___urcu_steal_work(worker, sibling_prev);
 248 end:
 249         rcu_read_unlock();
 250
 251         return steal_performed;
 252 }
 253
 254 static inline
 255 bool ___urcu_wakeup_sibling(struct urcu_worker *sibling)
 256 {
 257         return urcu_adaptative_wake_up(&sibling->wait_node);
 258 }
 259
 260 static inline
 261 bool __urcu_wakeup_siblings(struct urcu_workqueue *queue,
 262                 struct urcu_worker *worker)
 263 {
 264         struct urcu_worker *sibling_prev, *sibling_next;
 265         struct cds_list_head *sibling_node;
 266         bool wakeup_performed = 0;
 267
 268         if (!(worker->flags & URCU_WORKER_STEAL))
 269                 return;
 270
 271         /* Only wakeup siblings if we have work in our own queue. */
 272         if (cds_wfcq_empty(&worker->head, &worker->tail))
 273                 return;
 274
 275         rcu_read_lock();
 276
 277         sibling_node = rcu_dereference(worker->sibling_node.next);
 278         if (sibling_node == &queue->sibling_head)
 279                 sibling_node = rcu_dereference(sibling_node->next);
 280         sibling_next = caa_container_of(sibling_node, struct urcu_worker,
 281                         sibling_node);
 282         if (sibling_next != worker)
 283                 wakeup_performed = ___urcu_wakeup_sibling(sibling_next);
 284         if (wakeup_performed)
 285                 goto end;
 286
 287         sibling_node = rcu_dereference(worker->sibling_node.prev);
 288         if (sibling_node == &queue->sibling_head)
 289                 sibling_node = rcu_dereference(sibling_node->prev);
 290         sibling_prev = caa_container_of(sibling_node, struct urcu_worker,
 291                         sibling_node);
 292         if (sibling_prev != worker && sibling_prev != sibling_next)
 293                 wakeup_performed = ___urcu_wakeup_sibling(sibling_prev);
 294 end:
 295         rcu_read_unlock();
 296
 297         return wakeup_performed;
 298 }
 299
 300 static inline
 301 enum urcu_accept_ret urcu_accept_work(struct urcu_workqueue *queue,
 302                 struct urcu_worker *worker)
 303 {
 304         enum cds_wfcq_ret wfcq_ret;
 305
 306         wfcq_ret = __cds_wfcq_splice_blocking(&worker->head,
 307                         &worker->tail,
 308                         &queue->head,
 309                         &queue->tail);
 310         /* Don't wait if we have work to do. */
 311         if (wfcq_ret != CDS_WFCQ_RET_SRC_EMPTY
 312                         || !cds_wfcq_empty(&worker->head,
 313                                 &worker->tail))
 314                 goto do_work;
 315         /* Try to steal work from sibling instead of blocking */
 316         if (__urcu_steal_work(queue, worker))
 317                 goto do_work;
 318         /* No more work to do, check shutdown state */
 319         if (CMM_LOAD_SHARED(queue->shutdown))
 320                 return URCU_ACCEPT_SHUTDOWN;
 321         urcu_wait_set_state(&worker->wait_node,
 322                         URCU_WAIT_WAITING);
 323         if (!CMM_LOAD_SHARED(worker->wait_node.node.next)) {
 324                 int was_empty;
 325
 326                 /*
 327                  * NULL next pointer. We are therefore not in
 328                  * the queue.
 329                  */
 330                 cds_lfs_node_init(&worker->wait_node.node);
 331                 /* Protect stack dequeue against ABA */
 332                 synchronize_rcu();
 333                 was_empty = !urcu_wait_add(&queue->waitqueue,
 334                                 &worker->wait_node);
 335                 /*
 336                  * If the wait queue was empty, it means we are the
 337                  * first thread to be put back into an otherwise empty
 338                  * wait queue. Re-check if work queue is empty after
 339                  * adding ourself to wait queue, so we can wakeup the
 340                  * top of wait queue since new work have appeared, and
 341                  * work enqueuer may not have seen that it needed to do
 342                  * a wake up.
 343                  */
 344                 if (was_empty && !cds_wfcq_empty(&queue->head,
 345                                                 &queue->tail)) {
 346                         rcu_read_lock();        /* Protect stack dequeue */
 347                         (void) urcu_dequeue_wake_single(&queue->waitqueue);
 348                         rcu_read_unlock();      /* Protect stack dequeue */
 349                 }
 350         } else {
 351                 /*
 352                  * Non-NULL next pointer. We are therefore in
 353                  * the queue, or the dispatcher just removed us
 354                  * from it (after we read the next pointer), and
 355                  * is therefore awakening us. The state will
 356                  * therefore have been changed from WAITING to
 357                  * some other state, which will let the busy
 358                  * wait pass through.
 359                  */
 360         }
 361         urcu_adaptative_busy_wait(&worker->wait_node);
 362         return;
 363
 364 do_work:
 365         /*
 366          * We will be busy handling the work batch, awaken siblings so
 367          * they can steal from us.
 368          */
 369         (void) __urcu_wakeup_siblings(queue, worker);
 370         return URCU_ACCEPT_WORK;
 371 }
 372
 373 static inline
 374 struct urcu_work *urcu_dequeue_work(struct urcu_worker *worker)
 375 {
 376         struct cds_wfcq_node *node;
 377
 378         /*
 379          * If we are registered for work stealing, we need to dequeue
 380          * safely against siblings.
 381          */
 382         if (worker->flags & URCU_WORKER_STEAL) {
 383                 /*
 384                  * Don't bother grabbing the worker queue lock if it is
 385                  * empty.
 386                  */
 387                 if (cds_wfcq_empty(&worker->head, &worker->tail))
 388                         return NULL;
 389                 node = cds_wfcq_dequeue_blocking(&worker->head,
 390                                 &worker->tail);
 391         } else {
 392                 node = ___cds_wfcq_dequeue_with_state(&worker->head,
 393                                 &worker->tail, NULL, 1, 0);
 394         }
 395         if (!node)
 396                 return NULL;
 397         return caa_container_of(node, struct urcu_work, node);
 398 }
 399
 400 static inline
 401 void urcu_workqueue_shutdown(struct urcu_workqueue *queue)
 402 {
 403         /* Set shutdown */
 404         CMM_STORE_SHARED(queue->shutdown, true);
 405         /* Wakeup all workers */
 406         __urcu_workqueue_wakeup_all(queue);
 407 }
 408
 409 #endif /* _URCU_WORKQUEUE_FIFO_H */