kernel/rcupdate.c

   1 /*
   2  * Read-Copy Update mechanism for mutual exclusion
   3  *
   4  * This program is free software; you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License as published by
   6  * the Free Software Foundation; either version 2 of the License, or
   7  * (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  * GNU General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License
  15  * along with this program; if not, write to the Free Software
  16  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  17  *
  18  * Copyright (C) IBM Corporation, 2001
  19  *
  20  * Author: Dipankar Sarma <dipankar@in.ibm.com>
  21  *
  22  * Based on the original work by Paul McKenney <paul.mckenney@us.ibm.com>
  23  * and inputs from Rusty Russell, Andrea Arcangeli and Andi Kleen.
  24  * Papers:
  25  * http://www.rdrop.com/users/paulmck/paper/rclockpdcsproof.pdf
  26  * http://lse.sourceforge.net/locking/rclock_OLS.2001.05.01c.sc.pdf (OLS2001)
  27  *
  28  * For detailed explanation of Read-Copy Update mechanism see -
  29  *              http://lse.sourceforge.net/locking/rcupdate.html
  30  *
  31  */
  32 #include <linux/types.h>
  33 #include <linux/kernel.h>
  34 #include <linux/init.h>
  35 #include <linux/spinlock.h>
  36 #include <linux/smp.h>
  37 #include <linux/interrupt.h>
  38 #include <linux/sched.h>
  39 #include <asm/atomic.h>
  40 #include <asm/bitops.h>
  41 #include <linux/module.h>
  42 #include <linux/completion.h>
  43 #include <linux/percpu.h>
  44 #include <linux/notifier.h>
  45 #include <linux/rcupdate.h>
  46 #include <linux/cpu.h>
  47
  48 /* Definition for rcupdate control block. */
  49 struct rcu_ctrlblk rcu_ctrlblk =
  50         { .mutex = SPIN_LOCK_UNLOCKED, .curbatch = 1,
  51           .maxbatch = 1, .rcu_cpu_mask = CPU_MASK_NONE };
  52 DEFINE_PER_CPU(struct rcu_data, rcu_data) = { 0L };
  53
  54 /* Fake initialization required by compiler */
  55 static DEFINE_PER_CPU(struct tasklet_struct, rcu_tasklet) = {NULL};
  56 #define RCU_tasklet(cpu) (per_cpu(rcu_tasklet, cpu))
  57
  58 /**
  59  * call_rcu - Queue an RCU update request.
  60  * @head: structure to be used for queueing the RCU updates.
  61  * @func: actual update function to be invoked after the grace period
  62  * @arg: argument to be passed to the update function
  63  *
  64  * The update function will be invoked as soon as all CPUs have performed
  65  * a context switch or been seen in the idle loop or in a user process.
  66  * The read-side of critical section that use call_rcu() for updation must
  67  * be protected by rcu_read_lock()/rcu_read_unlock().
  68  */
  69 void fastcall call_rcu(struct rcu_head *head, void (*func)(void *arg), void *arg)
  70 {
  71         int cpu;
  72         unsigned long flags;
  73
  74         head->func = func;
  75         head->arg = arg;
  76         local_irq_save(flags);
  77         cpu = smp_processor_id();
  78         list_add_tail(&head->list, &RCU_nxtlist(cpu));
  79         local_irq_restore(flags);
  80 }
  81
  82 /*
  83  * Invoke the completed RCU callbacks. They are expected to be in
  84  * a per-cpu list.
  85  */
  86 static void rcu_do_batch(struct list_head *list)
  87 {
  88         struct list_head *entry;
  89         struct rcu_head *head;
  90
  91         while (!list_empty(list)) {
  92                 entry = list->next;
  93                 list_del(entry);
  94                 head = list_entry(entry, struct rcu_head, list);
  95                 head->func(head->arg);
  96         }
  97 }
  98
  99 /*
 100  * Register a new batch of callbacks, and start it up if there is currently no
 101  * active batch and the batch to be registered has not already occurred.
 102  * Caller must hold the rcu_ctrlblk lock.
 103  */
 104 static void rcu_start_batch(long newbatch)
 105 {
 106         cpumask_t active;
 107
 108         if (rcu_batch_before(rcu_ctrlblk.maxbatch, newbatch)) {
 109                 rcu_ctrlblk.maxbatch = newbatch;
 110         }
 111         if (rcu_batch_before(rcu_ctrlblk.maxbatch, rcu_ctrlblk.curbatch) ||
 112             !cpus_empty(rcu_ctrlblk.rcu_cpu_mask)) {
 113                 return;
 114         }
 115         /* Can't change, since spin lock held. */
 116         active = nohz_cpu_mask;
 117         cpus_complement(active);
 118         cpus_and(rcu_ctrlblk.rcu_cpu_mask, cpu_online_map, active);
 119 }
 120
 121 /*
 122  * Check if the cpu has gone through a quiescent state (say context
 123  * switch). If so and if it already hasn't done so in this RCU
 124  * quiescent cycle, then indicate that it has done so.
 125  */
 126 static void rcu_check_quiescent_state(void)
 127 {
 128         int cpu = smp_processor_id();
 129
 130         if (!cpu_isset(cpu, rcu_ctrlblk.rcu_cpu_mask))
 131                 return;
 132
 133         /*
 134          * Races with local timer interrupt - in the worst case
 135          * we may miss one quiescent state of that CPU. That is
 136          * tolerable. So no need to disable interrupts.
 137          */
 138         if (RCU_last_qsctr(cpu) == RCU_QSCTR_INVALID) {
 139                 RCU_last_qsctr(cpu) = RCU_qsctr(cpu);
 140                 return;
 141         }
 142         if (RCU_qsctr(cpu) == RCU_last_qsctr(cpu))
 143                 return;
 144
 145         spin_lock(&rcu_ctrlblk.mutex);
 146         if (!cpu_isset(cpu, rcu_ctrlblk.rcu_cpu_mask))
 147                 goto out_unlock;
 148
 149         cpu_clear(cpu, rcu_ctrlblk.rcu_cpu_mask);
 150         RCU_last_qsctr(cpu) = RCU_QSCTR_INVALID;
 151         if (!cpus_empty(rcu_ctrlblk.rcu_cpu_mask))
 152                 goto out_unlock;
 153
 154         rcu_ctrlblk.curbatch++;
 155         rcu_start_batch(rcu_ctrlblk.maxbatch);
 156
 157 out_unlock:
 158         spin_unlock(&rcu_ctrlblk.mutex);
 159 }
 160
 161
 162 #ifdef CONFIG_HOTPLUG_CPU
 163
 164 /* warning! helper for rcu_offline_cpu. do not use elsewhere without reviewing
 165  * locking requirements, the list it's pulling from has to belong to a cpu
 166  * which is dead and hence not processing interrupts.
 167  */
 168 static void rcu_move_batch(struct list_head *list)
 169 {
 170         struct list_head *entry;
 171         int cpu = smp_processor_id();
 172
 173         local_irq_disable();
 174         while (!list_empty(list)) {
 175                 entry = list->next;
 176                 list_del(entry);
 177                 list_add_tail(entry, &RCU_nxtlist(cpu));
 178         }
 179         local_irq_enable();
 180 }
 181
 182 static void rcu_offline_cpu(int cpu)
 183 {
 184         /* if the cpu going offline owns the grace period
 185          * we can block indefinitely waiting for it, so flush
 186          * it here
 187          */
 188         spin_lock_irq(&rcu_ctrlblk.mutex);
 189         if (cpus_empty(rcu_ctrlblk.rcu_cpu_mask))
 190                 goto unlock;
 191
 192         cpu_clear(cpu, rcu_ctrlblk.rcu_cpu_mask);
 193         if (cpus_empty(rcu_ctrlblk.rcu_cpu_mask)) {
 194                 rcu_ctrlblk.curbatch++;
 195                 /* We may avoid calling start batch if
 196                  * we are starting the batch only
 197                  * because of the DEAD CPU (the current
 198                  * CPU will start a new batch anyway for
 199                  * the callbacks we will move to current CPU).
 200                  * However, we will avoid this optimisation
 201                  * for now.
 202                  */
 203                 rcu_start_batch(rcu_ctrlblk.maxbatch);
 204         }
 205 unlock:
 206         spin_unlock_irq(&rcu_ctrlblk.mutex);
 207
 208         rcu_move_batch(&RCU_curlist(cpu));
 209         rcu_move_batch(&RCU_nxtlist(cpu));
 210
 211         tasklet_kill_immediate(&RCU_tasklet(cpu), cpu);
 212 }
 213
 214 #endif
 215
 216 /*
 217  * This does the RCU processing work from tasklet context.
 218  */
 219 static void rcu_process_callbacks(unsigned long unused)
 220 {
 221         int cpu = smp_processor_id();
 222         LIST_HEAD(list);
 223
 224         if (!list_empty(&RCU_curlist(cpu)) &&
 225             rcu_batch_after(rcu_ctrlblk.curbatch, RCU_batch(cpu))) {
 226                 __list_splice(&RCU_curlist(cpu), &list);
 227                 INIT_LIST_HEAD(&RCU_curlist(cpu));
 228         }
 229
 230         local_irq_disable();
 231         if (!list_empty(&RCU_nxtlist(cpu)) && list_empty(&RCU_curlist(cpu))) {
 232                 __list_splice(&RCU_nxtlist(cpu), &RCU_curlist(cpu));
 233                 INIT_LIST_HEAD(&RCU_nxtlist(cpu));
 234                 local_irq_enable();
 235
 236                 /*
 237                  * start the next batch of callbacks
 238                  */
 239                 spin_lock(&rcu_ctrlblk.mutex);
 240                 RCU_batch(cpu) = rcu_ctrlblk.curbatch + 1;
 241                 rcu_start_batch(RCU_batch(cpu));
 242                 spin_unlock(&rcu_ctrlblk.mutex);
 243         } else {
 244                 local_irq_enable();
 245         }
 246         rcu_check_quiescent_state();
 247         if (!list_empty(&list))
 248                 rcu_do_batch(&list);
 249 }
 250
 251 void rcu_check_callbacks(int cpu, int user)
 252 {
 253         if (user ||
 254             (idle_cpu(cpu) && !in_softirq() &&
 255                                 hardirq_count() <= (1 << HARDIRQ_SHIFT)))
 256                 RCU_qsctr(cpu)++;
 257         tasklet_schedule(&RCU_tasklet(cpu));
 258 }
 259
 260 static void __devinit rcu_online_cpu(int cpu)
 261 {
 262         memset(&per_cpu(rcu_data, cpu), 0, sizeof(struct rcu_data));
 263         tasklet_init(&RCU_tasklet(cpu), rcu_process_callbacks, 0UL);
 264         INIT_LIST_HEAD(&RCU_nxtlist(cpu));
 265         INIT_LIST_HEAD(&RCU_curlist(cpu));
 266 }
 267
 268 static int __devinit rcu_cpu_notify(struct notifier_block *self,
 269                                 unsigned long action, void *hcpu)
 270 {
 271         long cpu = (long)hcpu;
 272         switch (action) {
 273         case CPU_UP_PREPARE:
 274                 rcu_online_cpu(cpu);
 275                 break;
 276 #ifdef CONFIG_HOTPLUG_CPU
 277         case CPU_DEAD:
 278                 rcu_offline_cpu(cpu);
 279                 break;
 280 #endif
 281         default:
 282                 break;
 283         }
 284         return NOTIFY_OK;
 285 }
 286
 287 static struct notifier_block __devinitdata rcu_nb = {
 288         .notifier_call  = rcu_cpu_notify,
 289 };
 290
 291 /*
 292  * Initializes rcu mechanism.  Assumed to be called early.
 293  * That is before local timer(SMP) or jiffie timer (uniproc) is setup.
 294  * Note that rcu_qsctr and friends are implicitly
 295  * initialized due to the choice of ``0'' for RCU_CTR_INVALID.
 296  */
 297 void __init rcu_init(void)
 298 {
 299         rcu_cpu_notify(&rcu_nb, CPU_UP_PREPARE,
 300                         (void *)(long)smp_processor_id());
 301         /* Register notifier for non-boot CPUs */
 302         register_cpu_notifier(&rcu_nb);
 303 }
 304
 305
 306 /* Because of FASTCALL declaration of complete, we use this wrapper */
 307 static void wakeme_after_rcu(void *completion)
 308 {
 309         complete(completion);
 310 }
 311
 312 /**
 313  * synchronize-kernel - wait until all the CPUs have gone
 314  * through a "quiescent" state. It may sleep.
 315  */
 316 void synchronize_kernel(void)
 317 {
 318         struct rcu_head rcu;
 319         DECLARE_COMPLETION(completion);
 320
 321         /* Will wake me after RCU finished */
 322         call_rcu(&rcu, wakeme_after_rcu, &completion);
 323
 324         /* Wait for it */
 325         wait_for_completion(&completion);
 326 }
 327
 328
 329 EXPORT_SYMBOL(call_rcu);
 330 EXPORT_SYMBOL(synchronize_kernel);