ipc,sem: fine grained locking for semtimedop

Introduce finer grained locking for semtimedop, to handle the common case
of a program wanting to manipulate one semaphore from an array with
multiple semaphores.

If the call is a semop manipulating just one semaphore in an array with
multiple semaphores, only take the lock for that semaphore itself.

If the call needs to manipulate multiple semaphores, or another caller is
in a transaction that manipulates multiple semaphores, the sem_array lock
is taken, as well as all the locks for the individual semaphores.

On a 24 CPU system, performance numbers with the semop-multi
test with N threads and N semaphores, look like this:

	vanilla		Davidlohr's	Davidlohr's +	Davidlohr's +
threads			patches		rwlock patches	v3 patches
10	610652		726325		1783589		2142206
20	341570		365699		1520453		1977878
30	288102		307037		1498167		2037995
40	290714		305955		1612665		2256484
50	288620		312890		1733453		2650292
60	289987		306043		1649360		2388008
70	291298		306347		1723167		2717486
80	290948		305662		1729545		2763582
90	290996		306680		1736021		2757524
100	292243		306700		1773700		3059159

[davidlohr.bueso@hp.com: do not call sem_lock when bogus sma]
[davidlohr.bueso@hp.com: make refcounter atomic]
Signed-off-by: Rik van Riel <riel@redhat.com>
Suggested-by: Linus Torvalds <torvalds@linux-foundation.org>
Acked-by: Davidlohr Bueso <davidlohr.bueso@hp.com>
Cc: Chegu Vinod <chegu_vinod@hp.com>
Cc: Jason Low <jason.low2@hp.com>
Reviewed-by: Michel Lespinasse <walken@google.com>
Cc: Peter Hurley <peter@hurleysoftware.com>
Cc: Stanislav Kinsbursky <skinsbursky@parallels.com>
Tested-by: Emmanuel Benisty <benisty.e@gmail.com>
Tested-by: Sedat Dilek <sedat.dilek@gmail.com>
Signed-off-by: Andrew Morton <akpm@linux-foundation.org>
Signed-off-by: Linus Torvalds <torvalds@linux-foundation.org>

4 files changed, 203 insertions, 125 deletions

diff --git a/ipc/msg.c b/ipc/msg.c
index a80aaf463d9c..09a1f41e6595 100644
--- a/ipc/msg.c
+++ b/ipc/msg.c
@@ -687,7 +687,12 @@ long do_msgsnd(int msqid, long mtype, void __user *mtext,
                         goto out_unlock_free;
                 }
                 ss_add(msq, &s);
-                ipc_rcu_getref(msq);
+
+                if (!ipc_rcu_getref(msq)) {
+                        err = -EIDRM;
+                        goto out_unlock_free;
+                }
+
                 msg_unlock(msq);
                 schedule();
 
diff --git a/ipc/sem.c b/ipc/sem.c
index f68b61749a85..e78ee3186d1f 100644
--- a/ipc/sem.c
+++ b/ipc/sem.c
@@ -94,6 +94,7 @@
 struct sem {
         int     semval;         /* current value */
         int     sempid;         /* pid of last operation */
+        spinlock_t      lock;   /* spinlock for fine-grained semtimedop */
         struct list_head sem_pending; /* pending single-sop operations */
 };
 
@@ -137,7 +138,6 @@ struct sem_undo_list {
 
 #define sem_ids(ns)     ((ns)->ids[IPC_SEM_IDS])
 
-#define sem_unlock(sma)         ipc_unlock(&(sma)->sem_perm)
 #define sem_checkid(sma, semid) ipc_checkid(&sma->sem_perm, semid)
 
 static int newary(struct ipc_namespace *, struct ipc_params *);
@@ -190,10 +190,89 @@ void __init sem_init (void)
 }
 
 /*
+ * If the request contains only one semaphore operation, and there are
+ * no complex transactions pending, lock only the semaphore involved.
+ * Otherwise, lock the entire semaphore array, since we either have
+ * multiple semaphores in our own semops, or we need to look at
+ * semaphores from other pending complex operations.
+ *
+ * Carefully guard against sma->complex_count changing between zero
+ * and non-zero while we are spinning for the lock. The value of
+ * sma->complex_count cannot change while we are holding the lock,
+ * so sem_unlock should be fine.
+ *
+ * The global lock path checks that all the local locks have been released,
+ * checking each local lock once. This means that the local lock paths
+ * cannot start their critical sections while the global lock is held.
+ */
+static inline int sem_lock(struct sem_array *sma, struct sembuf *sops,
+                              int nsops)
+{
+        int locknum;
+ again:
+        if (nsops == 1 && !sma->complex_count) {
+                struct sem *sem = sma->sem_base + sops->sem_num;
+
+                /* Lock just the semaphore we are interested in. */
+                spin_lock(&sem->lock);
+
+                /*
+                 * If sma->complex_count was set while we were spinning,
+                 * we may need to look at things we did not lock here.
+                 */
+                if (unlikely(sma->complex_count)) {
+                        spin_unlock(&sem->lock);
+                        goto lock_array;
+                }
+
+                /*
+                 * Another process is holding the global lock on the
+                 * sem_array; we cannot enter our critical section,
+                 * but have to wait for the global lock to be released.
+                 */
+                if (unlikely(spin_is_locked(&sma->sem_perm.lock))) {
+                        spin_unlock(&sem->lock);
+                        spin_unlock_wait(&sma->sem_perm.lock);
+                        goto again;
+                }
+
+                locknum = sops->sem_num;
+        } else {
+                int i;
+                /*
+                 * Lock the semaphore array, and wait for all of the
+                 * individual semaphore locks to go away.  The code
+                 * above ensures no new single-lock holders will enter
+                 * their critical section while the array lock is held.
+                 */
+ lock_array:
+                spin_lock(&sma->sem_perm.lock);
+                for (i = 0; i < sma->sem_nsems; i++) {
+                        struct sem *sem = sma->sem_base + i;
+                        spin_unlock_wait(&sem->lock);
+                }
+                locknum = -1;
+        }
+        return locknum;
+}
+
+static inline void sem_unlock(struct sem_array *sma, int locknum)
+{
+        if (locknum == -1) {
+                spin_unlock(&sma->sem_perm.lock);
+        } else {
+                struct sem *sem = sma->sem_base + locknum;
+                spin_unlock(&sem->lock);
+        }
+        rcu_read_unlock();
+}
+
+/*
  * sem_lock_(check_) routines are called in the paths where the rw_mutex
  * is not held.
  */
-static inline struct sem_array *sem_obtain_lock(struct ipc_namespace *ns, int id)
+static inline struct sem_array *sem_obtain_lock(struct ipc_namespace *ns,
+                        int id, struct sembuf *sops, int nsops, int *locknum)
 {
         struct kern_ipc_perm *ipcp;
         struct sem_array *sma;
@@ -205,7 +284,8 @@ static inline struct sem_array *sem_obtain_lock(struct ipc_namespace *ns, int id
                 goto err;
         }
 
-        spin_lock(&ipcp->lock);
+        sma = container_of(ipcp, struct sem_array, sem_perm);
+        *locknum = sem_lock(sma, sops, nsops);
 
         /* ipc_rmid() may have already freed the ID while sem_lock
          * was spinning: verify that the structure is still valid
@@ -213,7 +293,7 @@ static inline struct sem_array *sem_obtain_lock(struct ipc_namespace *ns, int id
         if (!ipcp->deleted)
                 return container_of(ipcp, struct sem_array, sem_perm);
 
-        spin_unlock(&ipcp->lock);
+        sem_unlock(sma, *locknum);
         sma = ERR_PTR(-EINVAL);
 err:
         rcu_read_unlock();
@@ -230,17 +310,6 @@ static inline struct sem_array *sem_obtain_object(struct ipc_namespace *ns, int
         return container_of(ipcp, struct sem_array, sem_perm);
 }
 
-static inline struct sem_array *sem_lock_check(struct ipc_namespace *ns,
-                                                int id)
-{
-        struct kern_ipc_perm *ipcp = ipc_lock_check(&sem_ids(ns), id);
-
-        if (IS_ERR(ipcp))
-                return ERR_CAST(ipcp);
-
-        return container_of(ipcp, struct sem_array, sem_perm);
-}
-
 static inline struct sem_array *sem_obtain_object_check(struct ipc_namespace *ns,
                                                         int id)
 {
@@ -254,21 +323,21 @@ static inline struct sem_array *sem_obtain_object_check(struct ipc_namespace *ns
 
 static inline void sem_lock_and_putref(struct sem_array *sma)
 {
-        ipc_lock_by_ptr(&sma->sem_perm);
+        rcu_read_lock();
+        sem_lock(sma, NULL, -1);
         ipc_rcu_putref(sma);
 }
 
 static inline void sem_getref_and_unlock(struct sem_array *sma)
 {
-        ipc_rcu_getref(sma);
+        WARN_ON_ONCE(!ipc_rcu_getref(sma));