ipc/sem: add hysteresis

sysv sem has two lock modes: One with per-semaphore locks, one lock mode
with a single global lock for the whole array.  When switching from the
per-semaphore locks to the global lock, all per-semaphore locks must be
scanned for ongoing operations.

The patch adds a hysteresis for switching from the global lock to the
per semaphore locks.  This reduces how often the per-semaphore locks
must be scanned.

Compared to the initial patch, this is a simplified solution: Setting
USE_GLOBAL_LOCK_HYSTERESIS to 1 restores the current behavior.

In theory, a workload with exactly 10 simple sops and then one complex
op now scales a bit worse, but this is pure theory: If there is
concurrency, the it won't be exactly 10:1:10:1:10:1:...  If there is no
concurrency, then there is no need for scalability.

Link: http://lkml.kernel.org/r/1476851896-3590-3-git-send-email-manfred@colorfullife.com
Signed-off-by: Manfred Spraul <manfred@colorfullife.com>
Cc: Peter Zijlstra <peterz@infradead.org>
Cc: Davidlohr Bueso <dave@stgolabs.net>
Cc: Thomas Gleixner <tglx@linutronix.de>
Cc: Ingo Molnar <mingo@elte.hu>
Cc: H. Peter Anvin <hpa@zytor.com>
Cc: <1vier1@web.de>
Cc: kernel test robot <xiaolong.ye@intel.com>
Cc: <felixh@informatik.uni-bremen.de>
Signed-off-by: Andrew Morton <akpm@linux-foundation.org>
Signed-off-by: Linus Torvalds <torvalds@linux-foundation.org>

diff --git a/include/linux/sem.h b/include/linux/sem.h
index d0efd6e6c20a6a6a39273639dbd33f3d77c2e156..4fc222f8755d82113deba19eea85e8c47a844818 100644 (file)
--- a/include/linux/sem.h
+++ b/include/linux/sem.h
@@ -21,7 +21,7 @@ struct sem_array {
        struct list_head        list_id;        /* undo requests on this array */
        int                     sem_nsems;      /* no. of semaphores in array */
        int                     complex_count;  /* pending complex operations */
-       bool                    complex_mode;   /* no parallel simple ops */
+       unsigned int            use_global_lock;/* >0: global lock required */
 };
 
 #ifdef CONFIG_SYSVIPC

diff --git a/ipc/sem.c b/ipc/sem.c
index fe5db1ed081b7d0153897dad133bd7dfafb8f86a..e468cd1c12f0d6bb6d78fc583d4fb83e13a2c927 100644 (file)
--- a/ipc/sem.c
+++ b/ipc/sem.c
@@ -158,23 +158,43 @@ static int sysvipc_sem_proc_show(struct seq_file *s, void *it);
 #define SEMMSL_FAST    256 /* 512 bytes on stack */
 #define SEMOPM_FAST    64  /* ~ 372 bytes on stack */
 
+/*
+ * Switching from the mode suitable for simple ops
+ * to the mode for complex ops is costly. Therefore:
+ * use some hysteresis
+ */
+#define USE_GLOBAL_LOCK_HYSTERESIS     10
+
 /*
  * Locking:
  * a) global sem_lock() for read/write
  *     sem_undo.id_next,
  *     sem_array.complex_count,
- *     sem_array.complex_mode
  *     sem_array.pending{_alter,_const},
  *     sem_array.sem_undo
  *
  * b) global or semaphore sem_lock() for read/write:
  *     sem_array.sem_base[i].pending_{const,alter}:
- *     sem_array.complex_mode (for read)
  *
  * c) special:
  *     sem_undo_list.list_proc:
  *     * undo_list->lock for write
  *     * rcu for read
+ *     use_global_lock:
+ *     * global sem_lock() for write
+ *     * either local or global sem_lock() for read.
+ *
+ * Memory ordering:
+ * Most ordering is enforced by using spin_lock() and spin_unlock().
+ * The special case is use_global_lock:
+ * Setting it from non-zero to 0 is a RELEASE, this is ensured by
+ * using smp_store_release().
+ * Testing if it is non-zero is an ACQUIRE, this is ensured by using
+ * smp_load_acquire().
+ * Setting it from 0 to non-zero must be ordered with regards to
+ * this smp_load_acquire(), this is guaranteed because the smp_load_acquire()
+ * is inside a spin_lock() and after a write from 0 to non-zero a
+ * spin_lock()+spin_unlock() is done.
  */
 
 #define sc_semmsl      sem_ctls[0]
@@ -273,12 +293,16 @@ static void complexmode_enter(struct sem_array *sma)
        int i;
        struct sem *sem;
 
-       if (sma->complex_mode)  {
-               /* We are already in complex_mode. Nothing to do */
+       if (sma->use_global_lock > 0)  {
+               /*
+                * We are already in global lock mode.
+                * Nothing to do, just reset the
+                * counter until we return to simple mode.
+                */
+               sma->use_global_lock = USE_GLOBAL_LOCK_HYSTERESIS;
                return;
        }
-
-       sma->complex_mode = true;
+       sma->use_global_lock = USE_GLOBAL_LOCK_HYSTERESIS;
 
        for (i = 0; i < sma->sem_nsems; i++) {
                sem = sma->sem_base + i;
@@ -299,13 +323,17 @@ static void complexmode_tryleave(struct sem_array *sma)
                 */
                return;
        }
-       /*
-        * Immediately after setting complex_mode to false,
-        * a simple op can start. Thus: all memory writes
-        * performed by the current operation must be visible
-        * before we set complex_mode to false.
-        */
-       smp_store_release(&sma->complex_mode, false);
+       if (sma->use_global_lock == 1) {
+               /*
+                * Immediately after setting use_global_lock to 0,
+                * a simple op can start. Thus: all memory writes
+                * performed by the current operation must be visible
+                * before we set use_global_lock to 0.
+                */
+               smp_store_release(&sma->use_global_lock, 0);
+       } else {
+               sma->use_global_lock--;
+       }
 }
 
 #define SEM_GLOBAL_LOCK        (-1)
@@ -335,22 +363,23 @@ static inline int sem_lock(struct sem_array *sma, struct sembuf *sops,
         * Optimized locking is possible if no complex operation
         * is either enqueued or processed right now.
         *
-        * Both facts are tracked by complex_mode.
+        * Both facts are tracked by use_global_mode.
         */
        sem = sma->sem_base + sops->sem_num;
 
        /*
-        * Initial check for complex_mode. Just an optimization,
+        * Initial check for use_global_lock. Just an optimization,
         * no locking, no memory barrier.
         */
-       if (!sma->complex_mode) {
+       if (!sma->use_global_lock) {
                /*
                 * It appears that no complex operation is around.
                 * Acquire the per-semaphore lock.
                 */
                spin_lock(&sem->lock);
 
-               if (!smp_load_acquire(&sma->complex_mode)) {
+               /* pairs with smp_store_release() */
+               if (!smp_load_acquire(&sma->use_global_lock)) {
                        /* fast path successful! */
                        return sops->sem_num;
                }
@@ -360,19 +389,26 @@ static inline int sem_lock(struct sem_array *sma, struct sembuf *sops,
        /* slow path: acquire the full lock */
        ipc_lock_object(&sma->sem_perm);
 
-       if (sma->complex_count == 0) {
-               /* False alarm:
-                * There is no complex operation, thus we can switch
-                * back to the fast path.
+       if (sma->use_global_lock == 0) {
+               /*
+                * The use_global_lock mode ended while we waited for
+                * sma->sem_perm.lock. Thus we must switch to locking
+                * with sem->lock.
+                * Unlike in the fast path, there is no need to recheck
+                * sma->use_global_lock after we have acquired sem->lock:
+                * We own sma->sem_perm.lock, thus use_global_lock cannot
+                * change.
                 */
                spin_lock(&sem->lock);
+
                ipc_unlock_object(&sma->sem_perm);
                return sops->sem_num;
        } else {
-               /* Not a false alarm, thus complete the sequence for a
-                * full lock.
+               /*
+                * Not a false alarm, thus continue to use the global lock
+                * mode. No need for complexmode_enter(), this was done by
+                * the caller that has set use_global_mode to non-zero.
                 */
-               complexmode_enter(sma);
                return SEM_GLOBAL_LOCK;
        }
 }
@@ -476,7 +512,7 @@ static int newary(struct ipc_namespace *ns, struct ipc_params *params)
        }
 
        sma->complex_count = 0;
-       sma->complex_mode = true; /* dropped by sem_unlock below */
+       sma->use_global_lock = USE_GLOBAL_LOCK_HYSTERESIS;
        INIT_LIST_HEAD(&sma->pending_alter);
        INIT_LIST_HEAD(&sma->pending_const);
        INIT_LIST_HEAD(&sma->list_id);