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          * Wait for grace period before freeing or reusing
 163          * "worker" because used by RCU linked list.
 164          * Also prevents ABA for waitqueue stack dequeue: matches RCU
 165          * read-side critical sections around dequeue and move all
 166          * operations on waitqueue).
 167          */
 168         synchronize_rcu();
 169
 170         /*
 171          * Put any local work we still have back into the workqueue.
 172          */
 173         wfcq_ret = __cds_wfcq_splice_blocking(&queue->head,
 174                         &queue->tail,
 175                         &worker->head,
 176                         &worker->tail);
 177         if (wfcq_ret != CDS_WFCQ_RET_SRC_EMPTY
 178                         && wfcq_ret == CDS_WFCQ_RET_DEST_EMPTY) {
 179                 /*
 180                  * Wakeup worker thread if we have put work back into
 181                  * workqueue that was previously empty.
 182                  */
 183                 rcu_read_lock();        /* Protect stack dequeue */
 184                 (void) urcu_dequeue_wake_single(&queue->waitqueue);
 185                 rcu_read_unlock();      /* Protect stack dequeue */
 186         }
 187 }
 188
 189 /*
 190  * Try stealing work from siblings when we have nothing to do.
 191  */
 192 static inline
 193 bool ___urcu_steal_work(struct urcu_worker *worker,
 194                 struct urcu_worker *sibling)
 195 {
 196         enum cds_wfcq_ret splice_ret;
 197
 198         /*
 199          * Don't bother grabbing the sibling queue lock if it is empty.
 200          */
 201         if (cds_wfcq_empty(&sibling->head, &sibling->tail))
 202                 return false;
 203         splice_ret = cds_wfcq_splice_blocking(&worker->head,
 204                         &worker->tail,
 205                         &sibling->head,
 206                         &sibling->tail);
 207         /* Ensure that we preserve FIFO work order. */
 208         assert(splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
 209         return splice_ret != CDS_WFCQ_RET_SRC_EMPTY;
 210 }
 211
 212 static inline
 213 bool __urcu_steal_work(struct urcu_workqueue *queue,
 214                 struct urcu_worker *worker)
 215 {
 216         struct urcu_worker *sibling_prev, *sibling_next;
 217         struct cds_list_head *sibling_node;
 218         bool steal_performed = 0;
 219
 220         if (!(worker->flags & URCU_WORKER_STEAL))
 221                 return false;
 222
 223         rcu_read_lock();
 224
 225         sibling_node = rcu_dereference(worker->sibling_node.next);
 226         if (sibling_node == &queue->sibling_head)
 227                 sibling_node = rcu_dereference(sibling_node->next);
 228         sibling_next = caa_container_of(sibling_node, struct urcu_worker,
 229                         sibling_node);
 230         if (sibling_next != worker)
 231                 steal_performed = ___urcu_steal_work(worker, sibling_next);
 232         if (steal_performed)
 233                 goto end;
 234
 235         sibling_node = rcu_dereference(worker->sibling_node.prev);
 236         if (sibling_node == &queue->sibling_head)
 237                 sibling_node = rcu_dereference(sibling_node->prev);
 238         sibling_prev = caa_container_of(sibling_node, struct urcu_worker,
 239                         sibling_node);
 240         if (sibling_prev != worker && sibling_prev != sibling_next)
 241                 steal_performed =  ___urcu_steal_work(worker, sibling_prev);
 242 end:
 243         rcu_read_unlock();
 244
 245         return steal_performed;
 246 }
 247
 248 static inline
 249 bool ___urcu_wakeup_sibling(struct urcu_worker *sibling)
 250 {
 251         return urcu_adaptative_wake_up(&sibling->wait_node);
 252 }
 253
 254 static inline
 255 bool __urcu_wakeup_siblings(struct urcu_workqueue *queue,
 256                 struct urcu_worker *worker)
 257 {
 258         struct urcu_worker *sibling_prev, *sibling_next;
 259         struct cds_list_head *sibling_node;
 260         bool wakeup_performed = 0;
 261
 262         if (!(worker->flags & URCU_WORKER_STEAL))
 263                 return;
 264
 265         /* Only wakeup siblings if we have work in our own queue. */
 266         if (cds_wfcq_empty(&worker->head, &worker->tail))
 267                 return;
 268
 269         rcu_read_lock();
 270
 271         sibling_node = rcu_dereference(worker->sibling_node.next);
 272         if (sibling_node == &queue->sibling_head)
 273                 sibling_node = rcu_dereference(sibling_node->next);
 274         sibling_next = caa_container_of(sibling_node, struct urcu_worker,
 275                         sibling_node);
 276         if (sibling_next != worker)
 277                 wakeup_performed = ___urcu_wakeup_sibling(sibling_next);
 278         if (wakeup_performed)
 279                 goto end;
 280
 281         sibling_node = rcu_dereference(worker->sibling_node.prev);
 282         if (sibling_node == &queue->sibling_head)
 283                 sibling_node = rcu_dereference(sibling_node->prev);
 284         sibling_prev = caa_container_of(sibling_node, struct urcu_worker,
 285                         sibling_node);
 286         if (sibling_prev != worker && sibling_prev != sibling_next)
 287                 wakeup_performed = ___urcu_wakeup_sibling(sibling_prev);
 288 end:
 289         rcu_read_unlock();
 290
 291         return wakeup_performed;
 292 }
 293
 294 static inline
 295 enum urcu_accept_ret urcu_accept_work(struct urcu_workqueue *queue,
 296                 struct urcu_worker *worker)
 297 {
 298         enum cds_wfcq_ret wfcq_ret;
 299
 300         wfcq_ret = __cds_wfcq_splice_blocking(&worker->head,
 301                         &worker->tail,
 302                         &queue->head,
 303                         &queue->tail);
 304         /* Don't wait if we have work to do. */
 305         if (wfcq_ret != CDS_WFCQ_RET_SRC_EMPTY
 306                         || !cds_wfcq_empty(&worker->head,
 307                                 &worker->tail))
 308                 goto do_work;
 309         /* Try to steal work from sibling instead of blocking */
 310         if (__urcu_steal_work(queue, worker))
 311                 goto do_work;
 312         /* No more work to do, check shutdown state */
 313         if (CMM_LOAD_SHARED(queue->shutdown))
 314                 return URCU_ACCEPT_SHUTDOWN;
 315         urcu_wait_set_state(&worker->wait_node,
 316                         URCU_WAIT_WAITING);
 317         if (!CMM_LOAD_SHARED(worker->wait_node.node.next)) {
 318                 int was_empty;
 319
 320                 /*
 321                  * NULL next pointer. We are therefore not in
 322                  * the queue.
 323                  */
 324                 cds_lfs_node_init(&worker->wait_node.node);
 325                 /* Protect stack dequeue against ABA */
 326                 synchronize_rcu();
 327                 was_empty = !urcu_wait_add(&queue->waitqueue,
 328                                 &worker->wait_node);
 329                 /*
 330                  * If the wait queue was empty, it means we are the
 331                  * first thread to be put back into an otherwise empty
 332                  * wait queue. Re-check if work queue is empty after
 333                  * adding ourself to wait queue, so we can wakeup the
 334                  * top of wait queue since new work have appeared, and
 335                  * work enqueuer may not have seen that it needed to do
 336                  * a wake up.
 337                  */
 338                 if (was_empty && !cds_wfcq_empty(&queue->head,
 339                                                 &queue->tail)) {
 340                         rcu_read_lock();        /* Protect stack dequeue */
 341                         (void) urcu_dequeue_wake_single(&queue->waitqueue);
 342                         rcu_read_unlock();      /* Protect stack dequeue */
 343                 }
 344         } else {
 345                 /*
 346                  * Non-NULL next pointer. We are therefore in
 347                  * the queue, or the dispatcher just removed us
 348                  * from it (after we read the next pointer), and
 349                  * is therefore awakening us. The state will
 350                  * therefore have been changed from WAITING to
 351                  * some other state, which will let the busy
 352                  * wait pass through.
 353                  */
 354         }
 355         urcu_adaptative_busy_wait(&worker->wait_node);
 356         return;
 357
 358 do_work:
 359         /*
 360          * We will be busy handling the work batch, awaken siblings so
 361          * they can steal from us.
 362          */
 363         (void) __urcu_wakeup_siblings(queue, worker);
 364         return URCU_ACCEPT_WORK;
 365 }
 366
 367 static inline
 368 struct urcu_work *urcu_dequeue_work(struct urcu_worker *worker)
 369 {
 370         struct cds_wfcq_node *node;
 371
 372         /*
 373          * If we are registered for work stealing, we need to dequeue
 374          * safely against siblings.
 375          */
 376         if (worker->flags & URCU_WORKER_STEAL) {
 377                 /*
 378                  * Don't bother grabbing the worker queue lock if it is
 379                  * empty.
 380                  */
 381                 if (cds_wfcq_empty(&worker->head, &worker->tail))
 382                         return NULL;
 383                 node = cds_wfcq_dequeue_blocking(&worker->head,
 384                                 &worker->tail);
 385         } else {
 386                 node = ___cds_wfcq_dequeue_with_state(&worker->head,
 387                                 &worker->tail, NULL, 1, 0);
 388         }
 389         if (!node)
 390                 return NULL;
 391         return caa_container_of(node, struct urcu_work, node);
 392 }
 393
 394 static inline
 395 void urcu_workqueue_shutdown(struct urcu_workqueue *queue)
 396 {
 397         /* Set shutdown */
 398         CMM_STORE_SHARED(queue->shutdown, true);
 399         /* Wakeup all workers */
 400         __urcu_workqueue_wakeup_all(queue);
 401 }
 402
 403 #endif /* _URCU_WORKQUEUE_FIFO_H */