src/urcu-call-rcu-impl.h

   1 /*
   2  * urcu-call-rcu.c
   3  *
   4  * Userspace RCU library - batch memory reclamation with kernel API
   5  *
   6  * Copyright (c) 2010 Paul E. McKenney <paulmck@linux.vnet.ibm.com>
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2.1 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Lesser General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Lesser General Public
  19  * License along with this library; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  21  */
  22
  23 #define _LGPL_SOURCE
  24 #include <stdio.h>
  25 #include <pthread.h>
  26 #include <signal.h>
  27 #include <assert.h>
  28 #include <stdlib.h>
  29 #include <stdint.h>
  30 #include <string.h>
  31 #include <errno.h>
  32 #include <poll.h>
  33 #include <sys/time.h>
  34 #include <unistd.h>
  35 #include <sched.h>
  36
  37 #include "compat-getcpu.h"
  38 #include <urcu/wfcqueue.h>
  39 #include <urcu/call-rcu.h>
  40 #include <urcu/pointer.h>
  41 #include <urcu/list.h>
  42 #include <urcu/futex.h>
  43 #include <urcu/tls-compat.h>
  44 #include <urcu/ref.h>
  45 #include "urcu-die.h"
  46 #include "urcu-utils.h"
  47 #include "compat-smp.h"
  48
  49 #define SET_AFFINITY_CHECK_PERIOD               (1U << 8)       /* 256 */
  50 #define SET_AFFINITY_CHECK_PERIOD_MASK          (SET_AFFINITY_CHECK_PERIOD - 1)
  51
  52 /* Data structure that identifies a call_rcu thread. */
  53
  54 struct call_rcu_data {
  55         /*
  56          * We do not align head on a different cache-line than tail
  57          * mainly because call_rcu callback-invocation threads use
  58          * batching ("splice") to get an entire list of callbacks, which
  59          * effectively empties the queue, and requires to touch the tail
  60          * anyway.
  61          */
  62         struct cds_wfcq_tail cbs_tail;
  63         struct cds_wfcq_head cbs_head;
  64         unsigned long flags;
  65         int32_t futex;
  66         unsigned long qlen; /* maintained for debugging. */
  67         pthread_t tid;
  68         int cpu_affinity;
  69         unsigned long gp_count;
  70         struct cds_list_head list;
  71 } __attribute__((aligned(CAA_CACHE_LINE_SIZE)));
  72
  73 struct call_rcu_completion {
  74         int barrier_count;
  75         int32_t futex;
  76         struct urcu_ref ref;
  77 };
  78
  79 struct call_rcu_completion_work {
  80         struct rcu_head head;
  81         struct call_rcu_completion *completion;
  82 };
  83
  84 /*
  85  * List of all call_rcu_data structures to keep valgrind happy.
  86  * Protected by call_rcu_mutex.
  87  */
  88
  89 static CDS_LIST_HEAD(call_rcu_data_list);
  90
  91 /* Link a thread using call_rcu() to its call_rcu thread. */
  92
  93 static DEFINE_URCU_TLS(struct call_rcu_data *, thread_call_rcu_data);
  94
  95 /*
  96  * Guard call_rcu thread creation and atfork handlers.
  97  */
  98 static pthread_mutex_t call_rcu_mutex = PTHREAD_MUTEX_INITIALIZER;
  99
 100 /* If a given thread does not have its own call_rcu thread, this is default. */
 101
 102 static struct call_rcu_data *default_call_rcu_data;
 103
 104 static struct urcu_atfork *registered_rculfhash_atfork;
 105 static unsigned long registered_rculfhash_atfork_refcount;
 106
 107 /*
 108  * If the sched_getcpu() and sysconf(_SC_NPROCESSORS_CONF) calls are
 109  * available, then we can have call_rcu threads assigned to individual
 110  * CPUs rather than only to specific threads.
 111  */
 112
 113 #if defined(HAVE_SYSCONF) && (defined(HAVE_SCHED_GETCPU) || defined(HAVE_GETCPUID))
 114
 115 /*
 116  * Pointer to array of pointers to per-CPU call_rcu_data structures
 117  * and # CPUs. per_cpu_call_rcu_data is a RCU-protected pointer to an
 118  * array of RCU-protected pointers to call_rcu_data. call_rcu acts as a
 119  * RCU read-side and reads per_cpu_call_rcu_data and the per-cpu pointer
 120  * without mutex. The call_rcu_mutex protects updates.
 121  */
 122
 123 static struct call_rcu_data **per_cpu_call_rcu_data;
 124 static long cpus_array_len;
 125
 126 static void cpus_array_len_reset(void)
 127 {
 128         cpus_array_len = 0;
 129 }
 130
 131 /* Allocate the array if it has not already been allocated. */
 132
 133 static void alloc_cpu_call_rcu_data(void)
 134 {
 135         struct call_rcu_data **p;
 136         static int warned = 0;
 137
 138         if (cpus_array_len != 0)
 139                 return;
 140         cpus_array_len = get_possible_cpus_array_len();
 141         if (cpus_array_len <= 0) {
 142                 return;
 143         }
 144         p = malloc(cpus_array_len * sizeof(*per_cpu_call_rcu_data));
 145         if (p != NULL) {
 146                 memset(p, '\0', cpus_array_len * sizeof(*per_cpu_call_rcu_data));
 147                 rcu_set_pointer(&per_cpu_call_rcu_data, p);
 148         } else {
 149                 if (!warned) {
 150                         fprintf(stderr, "[error] liburcu: unable to allocate per-CPU pointer array\n");
 151                 }
 152                 warned = 1;
 153         }
 154 }
 155
 156 #else /* #if defined(HAVE_SYSCONF) && defined(HAVE_SCHED_GETCPU) */
 157
 158 /*
 159  * per_cpu_call_rcu_data should be constant, but some functions below, used both
 160  * for cases where cpu number is available and not available, assume it it not
 161  * constant.
 162  */
 163 static struct call_rcu_data **per_cpu_call_rcu_data = NULL;
 164 static const long cpus_array_len = -1;
 165
 166 static void cpus_array_len_reset(void)
 167 {
 168 }
 169
 170 static void alloc_cpu_call_rcu_data(void)
 171 {
 172 }
 173
 174 #endif /* #else #if defined(HAVE_SYSCONF) && defined(HAVE_SCHED_GETCPU) */
 175
 176 /* Acquire the specified pthread mutex. */
 177
 178 static void call_rcu_lock(pthread_mutex_t *pmp)
 179 {
 180         int ret;
 181
 182         ret = pthread_mutex_lock(pmp);
 183         if (ret)
 184                 urcu_die(ret);
 185 }
 186
 187 /* Release the specified pthread mutex. */
 188
 189 static void call_rcu_unlock(pthread_mutex_t *pmp)
 190 {
 191         int ret;
 192
 193         ret = pthread_mutex_unlock(pmp);
 194         if (ret)
 195                 urcu_die(ret);
 196 }
 197
 198 /*
 199  * Periodically retry setting CPU affinity if we migrate.
 200  * Losing affinity can be caused by CPU hotunplug/hotplug, or by
 201  * cpuset(7).
 202  */
 203 #ifdef HAVE_SCHED_SETAFFINITY
 204 static
 205 int set_thread_cpu_affinity(struct call_rcu_data *crdp)
 206 {
 207         cpu_set_t mask;
 208         int ret;
 209
 210         if (crdp->cpu_affinity < 0)
 211                 return 0;
 212         if (++crdp->gp_count & SET_AFFINITY_CHECK_PERIOD_MASK)
 213                 return 0;
 214         if (urcu_sched_getcpu() == crdp->cpu_affinity)
 215                 return 0;
 216
 217         CPU_ZERO(&mask);
 218         CPU_SET(crdp->cpu_affinity, &mask);
 219 #if SCHED_SETAFFINITY_ARGS == 2
 220         ret = sched_setaffinity(0, &mask);
 221 #else
 222         ret = sched_setaffinity(0, sizeof(mask), &mask);
 223 #endif
 224         /*
 225          * EINVAL is fine: can be caused by hotunplugged CPUs, or by
 226          * cpuset(7). This is why we should always retry if we detect
 227          * migration.
 228          */
 229         if (ret && errno == EINVAL) {
 230                 ret = 0;
 231                 errno = 0;
 232         }
 233         return ret;
 234 }
 235 #else
 236 static
 237 int set_thread_cpu_affinity(struct call_rcu_data *crdp __attribute__((unused)))
 238 {
 239         return 0;
 240 }
 241 #endif
 242
 243 static void call_rcu_wait(struct call_rcu_data *crdp)
 244 {
 245         /* Read call_rcu list before read futex */
 246         cmm_smp_mb();
 247         while (uatomic_read(&crdp->futex) == -1) {
 248                 if (!futex_async(&crdp->futex, FUTEX_WAIT, -1, NULL, NULL, 0)) {
 249                         /*
 250                          * Prior queued wakeups queued by unrelated code
 251                          * using the same address can cause futex wait to
 252                          * return 0 even through the futex value is still
 253                          * -1 (spurious wakeups). Check the value again
 254                          * in user-space to validate whether it really
 255                          * differs from -1.
 256                          */
 257                         continue;
 258                 }
 259                 switch (errno) {
 260                 case EAGAIN:
 261                         /* Value already changed. */
 262                         return;
 263                 case EINTR:
 264                         /* Retry if interrupted by signal. */
 265                         break;  /* Get out of switch. Check again. */
 266                 default:
 267                         /* Unexpected error. */
 268                         urcu_die(errno);
 269                 }
 270         }
 271 }
 272
 273 static void call_rcu_wake_up(struct call_rcu_data *crdp)
 274 {
 275         /* Write to call_rcu list before reading/writing futex */
 276         cmm_smp_mb();
 277         if (caa_unlikely(uatomic_read(&crdp->futex) == -1)) {
 278                 uatomic_set(&crdp->futex, 0);
 279                 if (futex_async(&crdp->futex, FUTEX_WAKE, 1,
 280                                 NULL, NULL, 0) < 0)
 281                         urcu_die(errno);
 282         }
 283 }
 284
 285 static void call_rcu_completion_wait(struct call_rcu_completion *completion)
 286 {
 287         /* Read completion barrier count before read futex */
 288         cmm_smp_mb();
 289         while (uatomic_read(&completion->futex) == -1) {
 290                 if (!futex_async(&completion->futex, FUTEX_WAIT, -1, NULL, NULL, 0)) {
 291                         /*
 292                          * Prior queued wakeups queued by unrelated code
 293                          * using the same address can cause futex wait to
 294                          * return 0 even through the futex value is still
 295                          * -1 (spurious wakeups). Check the value again
 296                          * in user-space to validate whether it really
 297                          * differs from -1.
 298                          */
 299                         continue;
 300                 }
 301                 switch (errno) {
 302                 case EAGAIN:
 303                         /* Value already changed. */
 304                         return;
 305                 case EINTR:
 306                         /* Retry if interrupted by signal. */
 307                         break;  /* Get out of switch. Check again. */
 308                 default:
 309                         /* Unexpected error. */
 310                         urcu_die(errno);
 311                 }
 312         }
 313 }
 314
 315 static void call_rcu_completion_wake_up(struct call_rcu_completion *completion)
 316 {
 317         /* Write to completion barrier count before reading/writing futex */
 318         cmm_smp_mb();
 319         if (caa_unlikely(uatomic_read(&completion->futex) == -1)) {
 320                 uatomic_set(&completion->futex, 0);
 321                 if (futex_async(&completion->futex, FUTEX_WAKE, 1,
 322                                 NULL, NULL, 0) < 0)
 323                         urcu_die(errno);
 324         }
 325 }
 326
 327 /* This is the code run by each call_rcu thread. */
 328
 329 static void *call_rcu_thread(void *arg)
 330 {
 331         unsigned long cbcount;
 332         struct call_rcu_data *crdp = (struct call_rcu_data *) arg;
 333         int rt = !!(uatomic_read(&crdp->flags) & URCU_CALL_RCU_RT);
 334
 335         if (set_thread_cpu_affinity(crdp))
 336                 urcu_die(errno);
 337
 338         /*
 339          * If callbacks take a read-side lock, we need to be registered.
 340          */
 341         rcu_register_thread();
 342
 343         URCU_TLS(thread_call_rcu_data) = crdp;
 344         if (!rt) {
 345                 uatomic_dec(&crdp->futex);
 346                 /* Decrement futex before reading call_rcu list */
 347                 cmm_smp_mb();
 348         }
 349         for (;;) {
 350                 struct cds_wfcq_head cbs_tmp_head;
 351                 struct cds_wfcq_tail cbs_tmp_tail;
 352                 struct cds_wfcq_node *cbs, *cbs_tmp_n;
 353                 enum cds_wfcq_ret splice_ret;
 354
 355                 if (set_thread_cpu_affinity(crdp))
 356                         urcu_die(errno);
 357
 358                 if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_PAUSE) {
 359                         /*
 360                          * Pause requested. Become quiescent: remove
 361                          * ourself from all global lists, and don't
 362                          * process any callback. The callback lists may
 363                          * still be non-empty though.
 364                          */
 365                         rcu_unregister_thread();
 366                         cmm_smp_mb__before_uatomic_or();
 367                         uatomic_or(&crdp->flags, URCU_CALL_RCU_PAUSED);
 368                         while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_PAUSE) != 0)
 369                                 (void) poll(NULL, 0, 1);
 370                         uatomic_and(&crdp->flags, ~URCU_CALL_RCU_PAUSED);
 371                         cmm_smp_mb__after_uatomic_and();
 372                         rcu_register_thread();
 373                 }
 374
 375                 cds_wfcq_init(&cbs_tmp_head, &cbs_tmp_tail);
 376                 splice_ret = __cds_wfcq_splice_blocking(&cbs_tmp_head,
 377                         &cbs_tmp_tail, &crdp->cbs_head, &crdp->cbs_tail);
 378                 assert(splice_ret != CDS_WFCQ_RET_WOULDBLOCK);
 379                 assert(splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
 380                 if (splice_ret != CDS_WFCQ_RET_SRC_EMPTY) {
 381                         synchronize_rcu();
 382                         cbcount = 0;
 383                         __cds_wfcq_for_each_blocking_safe(&cbs_tmp_head,
 384                                         &cbs_tmp_tail, cbs, cbs_tmp_n) {
 385                                 struct rcu_head *rhp;
 386
 387                                 rhp = caa_container_of(cbs,
 388                                         struct rcu_head, next);
 389                                 rhp->func(rhp);
 390                                 cbcount++;
 391                         }
 392                         uatomic_sub(&crdp->qlen, cbcount);
 393                 }
 394                 if (uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOP)
 395                         break;
 396                 rcu_thread_offline();
 397                 if (!rt) {
 398                         if (cds_wfcq_empty(&crdp->cbs_head,
 399                                         &crdp->cbs_tail)) {
 400                                 call_rcu_wait(crdp);
 401                                 (void) poll(NULL, 0, 10);
 402                                 uatomic_dec(&crdp->futex);
 403                                 /*
 404                                  * Decrement futex before reading
 405                                  * call_rcu list.
 406                                  */
 407                                 cmm_smp_mb();
 408                         } else {
 409                                 (void) poll(NULL, 0, 10);
 410                         }
 411                 } else {
 412                         (void) poll(NULL, 0, 10);
 413                 }
 414                 rcu_thread_online();
 415         }
 416         if (!rt) {
 417                 /*
 418                  * Read call_rcu list before write futex.
 419                  */
 420                 cmm_smp_mb();
 421                 uatomic_set(&crdp->futex, 0);
 422         }
 423         uatomic_or(&crdp->flags, URCU_CALL_RCU_STOPPED);
 424         rcu_unregister_thread();
 425         return NULL;
 426 }
 427
 428 /*
 429  * Create both a call_rcu thread and the corresponding call_rcu_data
 430  * structure, linking the structure in as specified.  Caller must hold
 431  * call_rcu_mutex.
 432  */
 433
 434 static void call_rcu_data_init(struct call_rcu_data **crdpp,
 435                                unsigned long flags,
 436                                int cpu_affinity)
 437 {
 438         struct call_rcu_data *crdp;
 439         int ret;
 440
 441         crdp = malloc(sizeof(*crdp));
 442         if (crdp == NULL)
 443                 urcu_die(errno);
 444         memset(crdp, '\0', sizeof(*crdp));
 445         cds_wfcq_init(&crdp->cbs_head, &crdp->cbs_tail);
 446         crdp->qlen = 0;
 447         crdp->futex = 0;
 448         crdp->flags = flags;
 449         cds_list_add(&crdp->list, &call_rcu_data_list);
 450         crdp->cpu_affinity = cpu_affinity;
 451         crdp->gp_count = 0;
 452         cmm_smp_mb();  /* Structure initialized before pointer is planted. */
 453         *crdpp = crdp;
 454         ret = pthread_create(&crdp->tid, NULL, call_rcu_thread, crdp);
 455         if (ret)
 456                 urcu_die(ret);
 457 }
 458
 459 /*
 460  * Return a pointer to the call_rcu_data structure for the specified
 461  * CPU, returning NULL if there is none.  We cannot automatically
 462  * created it because the platform we are running on might not define
 463  * urcu_sched_getcpu().
 464  *
 465  * The call to this function and use of the returned call_rcu_data
 466  * should be protected by RCU read-side lock.
 467  */
 468
 469 struct call_rcu_data *get_cpu_call_rcu_data(int cpu)
 470 {
 471         static int warned = 0;
 472         struct call_rcu_data **pcpu_crdp;
 473
 474         pcpu_crdp = rcu_dereference(per_cpu_call_rcu_data);
 475         if (pcpu_crdp == NULL)
 476                 return NULL;
 477         if (!warned && cpus_array_len > 0 && (cpu < 0 || cpus_array_len <= cpu)) {
 478                 fprintf(stderr, "[error] liburcu: get CPU # out of range\n");
 479                 warned = 1;
 480         }
 481         if (cpu < 0 || cpus_array_len <= cpu)
 482                 return NULL;
 483         return rcu_dereference(pcpu_crdp[cpu]);
 484 }
 485 URCU_ATTR_ALIAS(urcu_stringify(get_cpu_call_rcu_data))
 486 struct call_rcu_data *alias_get_cpu_call_rcu_data();
 487
 488 /*
 489  * Return the tid corresponding to the call_rcu thread whose
 490  * call_rcu_data structure is specified.
 491  */
 492
 493 pthread_t get_call_rcu_thread(struct call_rcu_data *crdp)
 494 {
 495         return crdp->tid;
 496 }
 497 URCU_ATTR_ALIAS(urcu_stringify(get_call_rcu_thread))
 498 pthread_t alias_get_call_rcu_thread();
 499
 500 /*
 501  * Create a call_rcu_data structure (with thread) and return a pointer.
 502  */
 503
 504 static struct call_rcu_data *__create_call_rcu_data(unsigned long flags,
 505                                                     int cpu_affinity)
 506 {
 507         struct call_rcu_data *crdp;
 508
 509         call_rcu_data_init(&crdp, flags, cpu_affinity);
 510         return crdp;
 511 }
 512
 513 URCU_ATTR_ALIAS(urcu_stringify(create_call_rcu_data))
 514 struct call_rcu_data *alias_create_call_rcu_data();
 515 struct call_rcu_data *create_call_rcu_data(unsigned long flags,
 516                                            int cpu_affinity)
 517 {
 518         struct call_rcu_data *crdp;
 519
 520         call_rcu_lock(&call_rcu_mutex);
 521         crdp = __create_call_rcu_data(flags, cpu_affinity);
 522         call_rcu_unlock(&call_rcu_mutex);
 523         return crdp;
 524 }
 525
 526 /*
 527  * Set the specified CPU to use the specified call_rcu_data structure.
 528  *
 529  * Use NULL to remove a CPU's call_rcu_data structure, but it is
 530  * the caller's responsibility to dispose of the removed structure.
 531  * Use get_cpu_call_rcu_data() to obtain a pointer to the old structure
 532  * (prior to NULLing it out, of course).
 533  *
 534  * The caller must wait for a grace-period to pass between return from
 535  * set_cpu_call_rcu_data() and call to call_rcu_data_free() passing the
 536  * previous call rcu data as argument.
 537  */
 538
 539 int set_cpu_call_rcu_data(int cpu, struct call_rcu_data *crdp)
 540 {
 541         static int warned = 0;
 542
 543         call_rcu_lock(&call_rcu_mutex);
 544         alloc_cpu_call_rcu_data();
 545         if (cpu < 0 || cpus_array_len <= cpu) {
 546                 if (!warned) {
 547                         fprintf(stderr, "[error] liburcu: set CPU # out of range\n");
 548                         warned = 1;
 549                 }
 550                 call_rcu_unlock(&call_rcu_mutex);
 551                 errno = EINVAL;
 552                 return -EINVAL;
 553         }
 554
 555         if (per_cpu_call_rcu_data == NULL) {
 556                 call_rcu_unlock(&call_rcu_mutex);
 557                 errno = ENOMEM;
 558                 return -ENOMEM;
 559         }
 560
 561         if (per_cpu_call_rcu_data[cpu] != NULL && crdp != NULL) {
 562                 call_rcu_unlock(&call_rcu_mutex);
 563                 errno = EEXIST;
 564                 return -EEXIST;
 565         }
 566
 567         rcu_set_pointer(&per_cpu_call_rcu_data[cpu], crdp);
 568         call_rcu_unlock(&call_rcu_mutex);
 569         return 0;
 570 }
 571 URCU_ATTR_ALIAS(urcu_stringify(set_cpu_call_rcu_data))
 572 int alias_set_cpu_call_rcu_data();
 573
 574 /*
 575  * Return a pointer to the default call_rcu_data structure, creating
 576  * one if need be.  Because we never free call_rcu_data structures,
 577  * we don't need to be in an RCU read-side critical section.
 578  */
 579
 580 struct call_rcu_data *get_default_call_rcu_data(void)
 581 {
 582         if (default_call_rcu_data != NULL)
 583                 return rcu_dereference(default_call_rcu_data);
 584         call_rcu_lock(&call_rcu_mutex);
 585         if (default_call_rcu_data != NULL) {
 586                 call_rcu_unlock(&call_rcu_mutex);
 587                 return default_call_rcu_data;
 588         }
 589         call_rcu_data_init(&default_call_rcu_data, 0, -1);
 590         call_rcu_unlock(&call_rcu_mutex);
 591         return default_call_rcu_data;
 592 }
 593 URCU_ATTR_ALIAS(urcu_stringify(get_default_call_rcu_data))
 594 struct call_rcu_data *alias_get_default_call_rcu_data();
 595
 596 /*
 597  * Return the call_rcu_data structure that applies to the currently
 598  * running thread.  Any call_rcu_data structure assigned specifically
 599  * to this thread has first priority, followed by any call_rcu_data
 600  * structure assigned to the CPU on which the thread is running,
 601  * followed by the default call_rcu_data structure.  If there is not
 602  * yet a default call_rcu_data structure, one will be created.
 603  *
 604  * Calls to this function and use of the returned call_rcu_data should
 605  * be protected by RCU read-side lock.
 606  */
 607 struct call_rcu_data *get_call_rcu_data(void)
 608 {
 609         struct call_rcu_data *crd;
 610
 611         if (URCU_TLS(thread_call_rcu_data) != NULL)
 612                 return URCU_TLS(thread_call_rcu_data);
 613
 614         if (cpus_array_len > 0) {
 615                 crd = get_cpu_call_rcu_data(urcu_sched_getcpu());
 616                 if (crd)
 617                         return crd;
 618         }
 619
 620         return get_default_call_rcu_data();
 621 }
 622 URCU_ATTR_ALIAS(urcu_stringify(get_call_rcu_data))
 623 struct call_rcu_data *alias_get_call_rcu_data();
 624
 625 /*
 626  * Return a pointer to this task's call_rcu_data if there is one.
 627  */
 628
 629 struct call_rcu_data *get_thread_call_rcu_data(void)
 630 {
 631         return URCU_TLS(thread_call_rcu_data);
 632 }
 633 URCU_ATTR_ALIAS(urcu_stringify(get_thread_call_rcu_data))
 634 struct call_rcu_data *alias_get_thread_call_rcu_data();
 635
 636 /*
 637  * Set this task's call_rcu_data structure as specified, regardless
 638  * of whether or not this task already had one.  (This allows switching
 639  * to and from real-time call_rcu threads, for example.)
 640  *
 641  * Use NULL to remove a thread's call_rcu_data structure, but it is
 642  * the caller's responsibility to dispose of the removed structure.
 643  * Use get_thread_call_rcu_data() to obtain a pointer to the old structure
 644  * (prior to NULLing it out, of course).
 645  */
 646
 647 void set_thread_call_rcu_data(struct call_rcu_data *crdp)
 648 {
 649         URCU_TLS(thread_call_rcu_data) = crdp;
 650 }
 651 URCU_ATTR_ALIAS(urcu_stringify(set_thread_call_rcu_data))
 652 void alias_set_thread_call_rcu_data();
 653
 654 /*
 655  * Create a separate call_rcu thread for each CPU.  This does not
 656  * replace a pre-existing call_rcu thread -- use the set_cpu_call_rcu_data()
 657  * function if you want that behavior. Should be paired with
 658  * free_all_cpu_call_rcu_data() to teardown these call_rcu worker
 659  * threads.
 660  */
 661
 662 int create_all_cpu_call_rcu_data(unsigned long flags)
 663 {
 664         int i;
 665         struct call_rcu_data *crdp;
 666         int ret;
 667
 668         call_rcu_lock(&call_rcu_mutex);
 669         alloc_cpu_call_rcu_data();
 670         call_rcu_unlock(&call_rcu_mutex);
 671         if (cpus_array_len <= 0) {
 672                 errno = EINVAL;
 673                 return -EINVAL;
 674         }
 675         if (per_cpu_call_rcu_data == NULL) {
 676                 errno = ENOMEM;
 677                 return -ENOMEM;
 678         }
 679         for (i = 0; i < cpus_array_len; i++) {
 680                 call_rcu_lock(&call_rcu_mutex);
 681                 if (get_cpu_call_rcu_data(i)) {
 682                         call_rcu_unlock(&call_rcu_mutex);
 683                         continue;
 684                 }
 685                 crdp = __create_call_rcu_data(flags, i);
 686                 if (crdp == NULL) {
 687                         call_rcu_unlock(&call_rcu_mutex);
 688                         errno = ENOMEM;
 689                         return -ENOMEM;
 690                 }
 691                 call_rcu_unlock(&call_rcu_mutex);
 692                 if ((ret = set_cpu_call_rcu_data(i, crdp)) != 0) {
 693                         call_rcu_data_free(crdp);
 694
 695                         /* it has been created by other thread */
 696                         if (ret == -EEXIST)
 697                                 continue;
 698
 699                         return ret;
 700                 }
 701         }
 702         return 0;
 703 }
 704 URCU_ATTR_ALIAS(urcu_stringify(create_all_cpu_call_rcu_data))
 705 int alias_create_all_cpu_call_rcu_data();
 706
 707 /*
 708  * Wake up the call_rcu thread corresponding to the specified
 709  * call_rcu_data structure.
 710  */
 711 static void wake_call_rcu_thread(struct call_rcu_data *crdp)
 712 {
 713         if (!(_CMM_LOAD_SHARED(crdp->flags) & URCU_CALL_RCU_RT))
 714                 call_rcu_wake_up(crdp);
 715 }
 716
 717 static void _call_rcu(struct rcu_head *head,
 718                       void (*func)(struct rcu_head *head),
 719                       struct call_rcu_data *crdp)
 720 {
 721         cds_wfcq_node_init(&head->next);
 722         head->func = func;
 723         cds_wfcq_enqueue(&crdp->cbs_head, &crdp->cbs_tail, &head->next);
 724         uatomic_inc(&crdp->qlen);
 725         wake_call_rcu_thread(crdp);
 726 }
 727
 728 /*
 729  * Schedule a function to be invoked after a following grace period.
 730  * This is the only function that must be called -- the others are
 731  * only present to allow applications to tune their use of RCU for
 732  * maximum performance.
 733  *
 734  * Note that unless a call_rcu thread has not already been created,
 735  * the first invocation of call_rcu() will create one.  So, if you
 736  * need the first invocation of call_rcu() to be fast, make sure
 737  * to create a call_rcu thread first.  One way to accomplish this is
 738  * "get_call_rcu_data();", and another is create_all_cpu_call_rcu_data().
 739  *
 740  * call_rcu must be called by registered RCU read-side threads.
 741  */
 742 void call_rcu(struct rcu_head *head,
 743               void (*func)(struct rcu_head *head))
 744 {
 745         struct call_rcu_data *crdp;
 746
 747         /* Holding rcu read-side lock across use of per-cpu crdp */
 748         _rcu_read_lock();
 749         crdp = get_call_rcu_data();
 750         _call_rcu(head, func, crdp);
 751         _rcu_read_unlock();
 752 }
 753 URCU_ATTR_ALIAS(urcu_stringify(call_rcu)) void alias_call_rcu();
 754
 755 /*
 756  * Free up the specified call_rcu_data structure, terminating the
 757  * associated call_rcu thread.  The caller must have previously
 758  * removed the call_rcu_data structure from per-thread or per-CPU
 759  * usage.  For example, set_cpu_call_rcu_data(cpu, NULL) for per-CPU
 760  * call_rcu_data structures or set_thread_call_rcu_data(NULL) for
 761  * per-thread call_rcu_data structures.
 762  *
 763  * We silently refuse to free up the default call_rcu_data structure
 764  * because that is where we put any leftover callbacks.  Note that
 765  * the possibility of self-spawning callbacks makes it impossible
 766  * to execute all the callbacks in finite time without putting any
 767  * newly spawned callbacks somewhere else.  The "somewhere else" of
 768  * last resort is the default call_rcu_data structure.
 769  *
 770  * We also silently refuse to free NULL pointers.  This simplifies
 771  * the calling code.
 772  *
 773  * The caller must wait for a grace-period to pass between return from
 774  * set_cpu_call_rcu_data() and call to call_rcu_data_free() passing the
 775  * previous call rcu data as argument.
 776  *
 777  * Note: introducing __cds_wfcq_splice_blocking() in this function fixed
 778  * a list corruption bug in the 0.7.x series. The equivalent fix
 779  * appeared in 0.6.8 for the stable-0.6 branch.
 780  */
 781 void call_rcu_data_free(struct call_rcu_data *crdp)
 782 {
 783         if (crdp == NULL || crdp == default_call_rcu_data) {
 784                 return;
 785         }
 786         if ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0) {
 787                 uatomic_or(&crdp->flags, URCU_CALL_RCU_STOP);
 788                 wake_call_rcu_thread(crdp);
 789                 while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_STOPPED) == 0)
 790                         (void) poll(NULL, 0, 1);
 791         }
 792         call_rcu_lock(&call_rcu_mutex);
 793         if (!cds_wfcq_empty(&crdp->cbs_head, &crdp->cbs_tail)) {
 794                 call_rcu_unlock(&call_rcu_mutex);
 795                 /* Create default call rcu data if need be. */
 796                 /* CBs queued here will be handed to the default list. */
 797                 (void) get_default_call_rcu_data();
 798                 call_rcu_lock(&call_rcu_mutex);
 799                 __cds_wfcq_splice_blocking(&default_call_rcu_data->cbs_head,
 800                         &default_call_rcu_data->cbs_tail,
 801                         &crdp->cbs_head, &crdp->cbs_tail);
 802                 uatomic_add(&default_call_rcu_data->qlen,
 803                             uatomic_read(&crdp->qlen));
 804                 wake_call_rcu_thread(default_call_rcu_data);
 805         }
 806
 807         cds_list_del(&crdp->list);
 808         call_rcu_unlock(&call_rcu_mutex);
 809
 810         free(crdp);
 811 }
 812 URCU_ATTR_ALIAS(urcu_stringify(call_rcu_data_free))
 813 void alias_call_rcu_data_free();
 814
 815 /*
 816  * Clean up all the per-CPU call_rcu threads.
 817  */
 818 void free_all_cpu_call_rcu_data(void)
 819 {
 820         int cpu;
 821         struct call_rcu_data **crdp;
 822         static int warned = 0;
 823
 824         if (cpus_array_len <= 0)
 825                 return;
 826
 827         crdp = malloc(sizeof(*crdp) * cpus_array_len);
 828         if (!crdp) {
 829                 if (!warned) {
 830                         fprintf(stderr, "[error] liburcu: unable to allocate per-CPU pointer array\n");
 831                 }
 832                 warned = 1;
 833                 return;
 834         }
 835
 836         for (cpu = 0; cpu < cpus_array_len; cpu++) {
 837                 crdp[cpu] = get_cpu_call_rcu_data(cpu);
 838                 if (crdp[cpu] == NULL)
 839                         continue;
 840                 set_cpu_call_rcu_data(cpu, NULL);
 841         }
 842         /*
 843          * Wait for call_rcu sites acting as RCU readers of the
 844          * call_rcu_data to become quiescent.
 845          */
 846         synchronize_rcu();
 847         for (cpu = 0; cpu < cpus_array_len; cpu++) {
 848                 if (crdp[cpu] == NULL)
 849                         continue;
 850                 call_rcu_data_free(crdp[cpu]);
 851         }
 852         free(crdp);
 853 }
 854 #ifdef RCU_QSBR
 855 /* ABI6 has a non-namespaced free_all_cpu_call_rcu_data for qsbr */
 856 #undef free_all_cpu_call_rcu_data
 857 URCU_ATTR_ALIAS("urcu_qsbr_free_all_cpu_call_rcu_data")
 858 void free_all_cpu_call_rcu_data();
 859 #define free_all_cpu_call_rcu_data urcu_qsbr_free_all_cpu_call_rcu_data
 860 #else
 861 URCU_ATTR_ALIAS(urcu_stringify(free_all_cpu_call_rcu_data))
 862 void alias_free_all_cpu_call_rcu_data();
 863 #endif
 864
 865 static
 866 void free_completion(struct urcu_ref *ref)
 867 {
 868         struct call_rcu_completion *completion;
 869
 870         completion = caa_container_of(ref, struct call_rcu_completion, ref);
 871         free(completion);
 872 }
 873
 874 static
 875 void _rcu_barrier_complete(struct rcu_head *head)
 876 {
 877         struct call_rcu_completion_work *work;
 878         struct call_rcu_completion *completion;
 879
 880         work = caa_container_of(head, struct call_rcu_completion_work, head);
 881         completion = work->completion;
 882         if (!uatomic_sub_return(&completion->barrier_count, 1))
 883                 call_rcu_completion_wake_up(completion);
 884         urcu_ref_put(&completion->ref, free_completion);
 885         free(work);
 886 }
 887
 888 /*
 889  * Wait for all in-flight call_rcu callbacks to complete execution.
 890  */
 891 void rcu_barrier(void)
 892 {
 893         struct call_rcu_data *crdp;
 894         struct call_rcu_completion *completion;
 895         int count = 0;
 896         int was_online;
 897
 898         /* Put in offline state in QSBR. */
 899         was_online = _rcu_read_ongoing();
 900         if (was_online)
 901                 rcu_thread_offline();
 902         /*
 903          * Calling a rcu_barrier() within a RCU read-side critical
 904          * section is an error.
 905          */
 906         if (_rcu_read_ongoing()) {
 907                 static int warned = 0;
 908
 909                 if (!warned) {
 910                         fprintf(stderr, "[error] liburcu: rcu_barrier() called from within RCU read-side critical section.\n");
 911                 }
 912                 warned = 1;
 913                 goto online;
 914         }
 915
 916         completion = calloc(sizeof(*completion), 1);
 917         if (!completion)
 918                 urcu_die(errno);
 919
 920         call_rcu_lock(&call_rcu_mutex);
 921         cds_list_for_each_entry(crdp, &call_rcu_data_list, list)
 922                 count++;
 923
 924         /* Referenced by rcu_barrier() and each call_rcu thread. */
 925         urcu_ref_set(&completion->ref, count + 1);
 926         completion->barrier_count = count;
 927
 928         cds_list_for_each_entry(crdp, &call_rcu_data_list, list) {
 929                 struct call_rcu_completion_work *work;
 930
 931                 work = calloc(sizeof(*work), 1);
 932                 if (!work)
 933                         urcu_die(errno);
 934                 work->completion = completion;
 935                 _call_rcu(&work->head, _rcu_barrier_complete, crdp);
 936         }
 937         call_rcu_unlock(&call_rcu_mutex);
 938
 939         /* Wait for them */
 940         for (;;) {
 941                 uatomic_dec(&completion->futex);
 942                 /* Decrement futex before reading barrier_count */
 943                 cmm_smp_mb();
 944                 if (!uatomic_read(&completion->barrier_count))
 945                         break;
 946                 call_rcu_completion_wait(completion);
 947         }
 948
 949         urcu_ref_put(&completion->ref, free_completion);
 950
 951 online:
 952         if (was_online)
 953                 rcu_thread_online();
 954 }
 955 URCU_ATTR_ALIAS(urcu_stringify(rcu_barrier))
 956 void alias_rcu_barrier();
 957
 958 /*
 959  * Acquire the call_rcu_mutex in order to ensure that the child sees
 960  * all of the call_rcu() data structures in a consistent state. Ensure
 961  * that all call_rcu threads are in a quiescent state across fork.
 962  * Suitable for pthread_atfork() and friends.
 963  */
 964 void call_rcu_before_fork(void)
 965 {
 966         struct call_rcu_data *crdp;
 967         struct urcu_atfork *atfork;
 968
 969         call_rcu_lock(&call_rcu_mutex);
 970
 971         atfork = registered_rculfhash_atfork;
 972         if (atfork)
 973                 atfork->before_fork(atfork->priv);
 974
 975         cds_list_for_each_entry(crdp, &call_rcu_data_list, list) {
 976                 uatomic_or(&crdp->flags, URCU_CALL_RCU_PAUSE);
 977                 cmm_smp_mb__after_uatomic_or();
 978                 wake_call_rcu_thread(crdp);
 979         }
 980         cds_list_for_each_entry(crdp, &call_rcu_data_list, list) {
 981                 while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_PAUSED) == 0)
 982                         (void) poll(NULL, 0, 1);
 983         }
 984 }
 985 URCU_ATTR_ALIAS(urcu_stringify(call_rcu_before_fork))
 986 void alias_call_rcu_before_fork();
 987
 988 /*
 989  * Clean up call_rcu data structures in the parent of a successful fork()
 990  * that is not followed by exec() in the child.  Suitable for
 991  * pthread_atfork() and friends.
 992  */
 993 void call_rcu_after_fork_parent(void)
 994 {
 995         struct call_rcu_data *crdp;
 996         struct urcu_atfork *atfork;
 997
 998         cds_list_for_each_entry(crdp, &call_rcu_data_list, list)
 999                 uatomic_and(&crdp->flags, ~URCU_CALL_RCU_PAUSE);
1000         cds_list_for_each_entry(crdp, &call_rcu_data_list, list) {
1001                 while ((uatomic_read(&crdp->flags) & URCU_CALL_RCU_PAUSED) != 0)
1002                         (void) poll(NULL, 0, 1);
1003         }
1004         atfork = registered_rculfhash_atfork;
1005         if (atfork)
1006                 atfork->after_fork_parent(atfork->priv);
1007         call_rcu_unlock(&call_rcu_mutex);
1008 }
1009 URCU_ATTR_ALIAS(urcu_stringify(call_rcu_after_fork_parent))
1010 void alias_call_rcu_after_fork_parent();
1011
1012 /*
1013  * Clean up call_rcu data structures in the child of a successful fork()
1014  * that is not followed by exec().  Suitable for pthread_atfork() and
1015  * friends.
1016  */
1017 void call_rcu_after_fork_child(void)
1018 {
1019         struct call_rcu_data *crdp, *next;
1020         struct urcu_atfork *atfork;
1021
1022         /* Release the mutex. */
1023         call_rcu_unlock(&call_rcu_mutex);
1024
1025         atfork = registered_rculfhash_atfork;
1026         if (atfork)
1027                 atfork->after_fork_child(atfork->priv);
1028
1029         /* Do nothing when call_rcu() has not been used */
1030         if (cds_list_empty(&call_rcu_data_list))
1031                 return;
1032
1033         /*
1034          * Allocate a new default call_rcu_data structure in order
1035          * to get a working call_rcu thread to go with it.
1036          */
1037         default_call_rcu_data = NULL;
1038         (void)get_default_call_rcu_data();
1039
1040         /* Cleanup call_rcu_data pointers before use */
1041         cpus_array_len_reset();
1042         free(per_cpu_call_rcu_data);
1043         rcu_set_pointer(&per_cpu_call_rcu_data, NULL);
1044         URCU_TLS(thread_call_rcu_data) = NULL;
1045
1046         /*
1047          * Dispose of all of the rest of the call_rcu_data structures.
1048          * Leftover call_rcu callbacks will be merged into the new
1049          * default call_rcu thread queue.
1050          */
1051         cds_list_for_each_entry_safe(crdp, next, &call_rcu_data_list, list) {
1052                 if (crdp == default_call_rcu_data)
1053                         continue;
1054                 uatomic_set(&crdp->flags, URCU_CALL_RCU_STOPPED);
1055                 call_rcu_data_free(crdp);
1056         }
1057 }
1058 URCU_ATTR_ALIAS(urcu_stringify(call_rcu_after_fork_child))
1059 void alias_call_rcu_after_fork_child();
1060
1061 void urcu_register_rculfhash_atfork(struct urcu_atfork *atfork)
1062 {
1063         call_rcu_lock(&call_rcu_mutex);
1064         if (registered_rculfhash_atfork_refcount++)
1065                 goto end;
1066         registered_rculfhash_atfork = atfork;
1067 end:
1068         call_rcu_unlock(&call_rcu_mutex);
1069 }
1070 URCU_ATTR_ALIAS(urcu_stringify(urcu_register_rculfhash_atfork))
1071 void alias_urcu_register_rculfhash_atfork();
1072
1073 void urcu_unregister_rculfhash_atfork(struct urcu_atfork *atfork __attribute__((unused)))
1074 {
1075         call_rcu_lock(&call_rcu_mutex);
1076         if (--registered_rculfhash_atfork_refcount)
1077                 goto end;
1078         registered_rculfhash_atfork = NULL;
1079 end:
1080         call_rcu_unlock(&call_rcu_mutex);
1081 }
1082 URCU_ATTR_ALIAS(urcu_stringify(urcu_unregister_rculfhash_atfork))
1083 void alias_urcu_unregister_rculfhash_atfork();