[PATCH] rcu file: use atomic primitives
[linux-3.10.git] / kernel / rcupdate.c
index d00eded..30b0bba 100644 (file)
@@ -35,6 +35,7 @@
 #include <linux/init.h>
 #include <linux/spinlock.h>
 #include <linux/smp.h>
+#include <linux/rcupdate.h>
 #include <linux/interrupt.h>
 #include <linux/sched.h>
 #include <asm/atomic.h>
@@ -60,9 +61,9 @@ struct rcu_state {
                                      /* for current batch to proceed.        */
 };
 
-static struct rcu_state rcu_state ____cacheline_maxaligned_in_smp =
+static struct rcu_state rcu_state ____cacheline_internodealigned_in_smp =
          {.lock = SPIN_LOCK_UNLOCKED, .cpumask = CPU_MASK_NONE };
-static struct rcu_state rcu_bh_state ____cacheline_maxaligned_in_smp =
+static struct rcu_state rcu_bh_state ____cacheline_internodealigned_in_smp =
          {.lock = SPIN_LOCK_UNLOCKED, .cpumask = CPU_MASK_NONE };
 
 DEFINE_PER_CPU(struct rcu_data, rcu_data) = { 0L };
@@ -70,7 +71,7 @@ DEFINE_PER_CPU(struct rcu_data, rcu_bh_data) = { 0L };
 
 /* Fake initialization required by compiler */
 static DEFINE_PER_CPU(struct tasklet_struct, rcu_tasklet) = {NULL};
-static int maxbatch = 10;
+static int maxbatch = 10000;
 
 /**
  * call_rcu - Queue an RCU callback for invocation after a grace period.
@@ -95,9 +96,17 @@ void fastcall call_rcu(struct rcu_head *head,
        rdp = &__get_cpu_var(rcu_data);
        *rdp->nxttail = head;
        rdp->nxttail = &head->next;
+
+       if (unlikely(++rdp->count > 10000))
+               set_need_resched();
+
        local_irq_restore(flags);
 }
 
+static atomic_t rcu_barrier_cpu_count;
+static struct semaphore rcu_barrier_sema;
+static struct completion rcu_barrier_completion;
+
 /**
  * call_rcu_bh - Queue an RCU for invocation after a quicker grace period.
  * @head: structure to be used for queueing the RCU updates.
@@ -126,10 +135,61 @@ void fastcall call_rcu_bh(struct rcu_head *head,
        rdp = &__get_cpu_var(rcu_bh_data);
        *rdp->nxttail = head;
        rdp->nxttail = &head->next;
+       rdp->count++;
+/*
+ *  Should we directly call rcu_do_batch() here ?
+ *  if (unlikely(rdp->count > 10000))
+ *      rcu_do_batch(rdp);
+ */
        local_irq_restore(flags);
 }
 
 /*
+ * Return the number of RCU batches processed thus far.  Useful
+ * for debug and statistics.
+ */
+long rcu_batches_completed(void)
+{
+       return rcu_ctrlblk.completed;
+}
+
+static void rcu_barrier_callback(struct rcu_head *notused)
+{
+       if (atomic_dec_and_test(&rcu_barrier_cpu_count))
+               complete(&rcu_barrier_completion);
+}
+
+/*
+ * Called with preemption disabled, and from cross-cpu IRQ context.
+ */
+static void rcu_barrier_func(void *notused)
+{
+       int cpu = smp_processor_id();
+       struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
+       struct rcu_head *head;
+
+       head = &rdp->barrier;
+       atomic_inc(&rcu_barrier_cpu_count);
+       call_rcu(head, rcu_barrier_callback);
+}
+
+/**
+ * rcu_barrier - Wait until all the in-flight RCUs are complete.
+ */
+void rcu_barrier(void)
+{
+       BUG_ON(in_interrupt());
+       /* Take cpucontrol semaphore to protect against CPU hotplug */
+       down(&rcu_barrier_sema);
+       init_completion(&rcu_barrier_completion);
+       atomic_set(&rcu_barrier_cpu_count, 0);
+       on_each_cpu(rcu_barrier_func, NULL, 0, 1);
+       wait_for_completion(&rcu_barrier_completion);
+       up(&rcu_barrier_sema);
+}
+EXPORT_SYMBOL_GPL(rcu_barrier);
+
+/*
  * Invoke the completed RCU callbacks. They are expected to be in
  * a per-cpu list.
  */
@@ -143,6 +203,7 @@ static void rcu_do_batch(struct rcu_data *rdp)
                next = rdp->donelist = list->next;
                list->func(list);
                list = next;
+               rdp->count--;
                if (++count >= maxbatch)
                        break;
        }
@@ -183,15 +244,23 @@ static void rcu_start_batch(struct rcu_ctrlblk *rcp, struct rcu_state *rsp,
 
        if (rcp->next_pending &&
                        rcp->completed == rcp->cur) {
-               /* Can't change, since spin lock held. */
-               cpus_andnot(rsp->cpumask, cpu_online_map, nohz_cpu_mask);
-
                rcp->next_pending = 0;
-               /* next_pending == 0 must be visible in __rcu_process_callbacks()
-                * before it can see new value of cur.
+               /*
+                * next_pending == 0 must be visible in
+                * __rcu_process_callbacks() before it can see new value of cur.
                 */
                smp_wmb();
                rcp->cur++;
+
+               /*
+                * Accessing nohz_cpu_mask before incrementing rcp->cur needs a
+                * Barrier  Otherwise it can cause tickless idle CPUs to be
+                * included in rsp->cpumask, which will extend graceperiods
+                * unnecessarily.
+                */
+               smp_mb();
+               cpus_andnot(rsp->cpumask, cpu_online_map, nohz_cpu_mask);
+
        }
 }
 
@@ -423,6 +492,7 @@ static struct notifier_block __devinitdata rcu_nb = {
  */
 void __init rcu_init(void)
 {
+       sema_init(&rcu_barrier_sema, 1);
        rcu_cpu_notify(&rcu_nb, CPU_UP_PREPARE,
                        (void *)(long)smp_processor_id());
        /* Register notifier for non-boot CPUs */
@@ -444,15 +514,18 @@ static void wakeme_after_rcu(struct rcu_head  *head)
 }
 
 /**
- * synchronize_kernel - wait until a grace period has elapsed.
+ * synchronize_rcu - wait until a grace period has elapsed.
  *
  * Control will return to the caller some time after a full grace
  * period has elapsed, in other words after all currently executing RCU
  * read-side critical sections have completed.  RCU read-side critical
  * sections are delimited by rcu_read_lock() and rcu_read_unlock(),
  * and may be nested.
+ *
+ * If your read-side code is not protected by rcu_read_lock(), do -not-
+ * use synchronize_rcu().
  */
-void synchronize_kernel(void)
+void synchronize_rcu(void)
 {
        struct rcu_synchronize rcu;
 
@@ -464,7 +537,17 @@ void synchronize_kernel(void)
        wait_for_completion(&rcu.completion);
 }
 
+/*
+ * Deprecated, use synchronize_rcu() or synchronize_sched() instead.
+ */
+void synchronize_kernel(void)
+{
+       synchronize_rcu();
+}
+
 module_param(maxbatch, int, 0);
-EXPORT_SYMBOL_GPL(call_rcu);
-EXPORT_SYMBOL_GPL(call_rcu_bh);
-EXPORT_SYMBOL_GPL(synchronize_kernel);
+EXPORT_SYMBOL_GPL(rcu_batches_completed);
+EXPORT_SYMBOL(call_rcu);  /* WARNING: GPL-only in April 2006. */
+EXPORT_SYMBOL(call_rcu_bh);  /* WARNING: GPL-only in April 2006. */
+EXPORT_SYMBOL_GPL(synchronize_rcu);
+EXPORT_SYMBOL(synchronize_kernel);  /* WARNING: GPL-only in April 2006. */