[src/trunk]: src/lib/librumpuser Fix a few locking problems with multithreade...

[pooka <pooka%NetBSD.org@localhost>]

details:   https://anonhg.NetBSD.org/src/rev/6ef9ec7a2013
branches:  trunk
changeset: 759052:6ef9ec7a2013
user:      pooka <pooka%NetBSD.org@localhost>
date:      Fri Nov 26 18:51:03 2010 +0000

description:
Fix a few locking problems with multithreaded clients.

TODO: make server deal graciously with out-of-resources conditions

diffstat:

 lib/librumpuser/rumpuser_sp.c |  38 +++++++++++++++++---------------------
 lib/librumpuser/sp_common.c   |  38 ++++++++++++++++++++++++++++++--------
 2 files changed, 47 insertions(+), 29 deletions(-)

diffs (184 lines):

diff -r 45b73e8a5147 -r 6ef9ec7a2013 lib/librumpuser/rumpuser_sp.c
--- a/lib/librumpuser/rumpuser_sp.c     Fri Nov 26 15:14:29 2010 +0000
+++ b/lib/librumpuser/rumpuser_sp.c     Fri Nov 26 18:51:03 2010 +0000
@@ -1,4 +1,4 @@
-/*      $NetBSD: rumpuser_sp.c,v 1.17 2010/11/26 14:37:08 pooka Exp $  */
+/*      $NetBSD: rumpuser_sp.c,v 1.18 2010/11/26 18:51:03 pooka Exp $  */
 
 /*
  * Copyright (c) 2010 Antti Kantee.  All Rights Reserved.
@@ -38,7 +38,7 @@
  */
 
 #include <sys/cdefs.h>
-__RCSID("$NetBSD: rumpuser_sp.c,v 1.17 2010/11/26 14:37:08 pooka Exp $");
+__RCSID("$NetBSD: rumpuser_sp.c,v 1.18 2010/11/26 18:51:03 pooka Exp $");
 
 #include <sys/types.h>
 #include <sys/atomic.h>
@@ -65,7 +65,7 @@
 
 #include "sp_common.c"
 
-#define MAXCLI 4
+#define MAXCLI 256
 
 static struct pollfd pfdlist[MAXCLI];
 static struct spclient spclist[MAXCLI];
@@ -329,7 +329,6 @@
        spc->spc_dying = 0;
 
        atomic_inc_uint(&disco);
-
 }
 
 static void
@@ -622,25 +621,22 @@
        DPRINTF(("rump_sp: server mainloop\n"));
 
        for (;;) {
-               /* g/c hangarounds (eventually) */
-               if (disco) {
-                       int discoed;
+               int discoed;
 
-                       discoed = atomic_swap_uint(&disco, 0);
-                       while (discoed--) {
-                               nfds--;
-                               idx = maxidx;
-                               while (idx) {
-                                       if (pfdlist[idx].fd != -1) {
-                                               maxidx = idx;
-                                               break;
-                                       }
-                                       idx--;
+               /* g/c hangarounds (eventually) */
+               discoed = atomic_swap_uint(&disco, 0);
+               while (discoed--) {
+                       nfds--;
+                       idx = maxidx;
+                       while (idx) {
+                               if (pfdlist[idx].fd != -1) {
+                                       maxidx = idx;
+                                       break;
                                }
-                               DPRINTF(("rump_sp: set maxidx to [%u]\n",
-                                   maxidx));
-                               assert(maxidx+1 >= nfds);
+                               idx--;
                        }
+                       DPRINTF(("rump_sp: set maxidx to [%u]\n",
+                           maxidx));
                }
 
                DPRINTF(("rump_sp: loop nfd %d\n", maxidx+1));
@@ -744,7 +740,7 @@
                fprintf(stderr, "rump_sp: server bind failed\n");
                return errno;
        }
-       if (listen(s, 20) == -1) {
+       if (listen(s, MAXCLI) == -1) {
                fprintf(stderr, "rump_sp: server listen failed\n");
                return errno;
        }
diff -r 45b73e8a5147 -r 6ef9ec7a2013 lib/librumpuser/sp_common.c
--- a/lib/librumpuser/sp_common.c       Fri Nov 26 15:14:29 2010 +0000
+++ b/lib/librumpuser/sp_common.c       Fri Nov 26 18:51:03 2010 +0000
@@ -1,4 +1,4 @@
-/*      $NetBSD: sp_common.c,v 1.11 2010/11/26 14:37:08 pooka Exp $    */
+/*      $NetBSD: sp_common.c,v 1.12 2010/11/26 18:51:03 pooka Exp $    */
 
 /*
  * Copyright (c) 2010 Antti Kantee.  All Rights Reserved.
@@ -133,7 +133,7 @@
        TAILQ_HEAD(, respwait) spc_respwait;
 
        /* rest of the fields are zeroed upon disconnect */
-#define SPC_ZEROFF offsetof(struct spclient, spc_pid)
+#define SPC_ZEROFF offsetof(struct spclient, spc_pfd)
        struct pollfd *spc_pfd;
 
        struct rsp_hdr spc_hdr;
@@ -154,26 +154,42 @@
 static void handlereq(struct spclient *);
 
 static void
-sendlock(struct spclient *spc)
+sendlockl(struct spclient *spc)
 {
 
-       pthread_mutex_lock(&spc->spc_mtx);
+       /* assert(pthread_mutex_owned) */
        while (spc->spc_ostatus != SPCSTATUS_FREE) {
                spc->spc_ostatus = SPCSTATUS_WANTED;
                pthread_cond_wait(&spc->spc_cv, &spc->spc_mtx);
        }
        spc->spc_ostatus = SPCSTATUS_BUSY;
+}
+
+static void
+sendlock(struct spclient *spc)
+{
+
+       pthread_mutex_lock(&spc->spc_mtx);
+       sendlockl(spc);
        pthread_mutex_unlock(&spc->spc_mtx);
 }
 
 static void
+sendunlockl(struct spclient *spc)
+{
+
+       /* assert(pthread_mutex_owned) */
+       if (spc->spc_ostatus == SPCSTATUS_WANTED)
+               pthread_cond_broadcast(&spc->spc_cv);
+       spc->spc_ostatus = SPCSTATUS_FREE;
+}
+
+static void
 sendunlock(struct spclient *spc)
 {
 
        pthread_mutex_lock(&spc->spc_mtx);
-       if (spc->spc_ostatus == SPCSTATUS_WANTED)
-               pthread_cond_broadcast(&spc->spc_cv);
-       spc->spc_ostatus = SPCSTATUS_FREE;
+       sendunlockl(spc);
        pthread_mutex_unlock(&spc->spc_mtx);
 }
 
@@ -224,12 +240,16 @@
        pthread_mutex_lock(&spc->spc_mtx);
        rw->rw_reqno = rhdr->rsp_reqno = spc->spc_nextreq++;
        TAILQ_INSERT_TAIL(&spc->spc_respwait, rw, rw_entries);
+
+       sendlockl(spc);
 }
 
 static void
 unputwait(struct spclient *spc, struct respwait *rw)
 {
 
+       sendunlockl(spc);
+
        TAILQ_REMOVE(&spc->spc_respwait, rw, rw_entries);
        pthread_mutex_unlock(&spc->spc_mtx);
        pthread_cond_destroy(&rw->rw_cv);
@@ -267,7 +287,7 @@
 
        /* DIAGASSERT(mutex_owned(spc_lock)) */
        TAILQ_FOREACH(rw, &spc->spc_respwait, rw_entries)
-               pthread_cond_signal(&rw->rw_cv);
+               pthread_cond_broadcast(&rw->rw_cv);
 }
 
 static int
@@ -276,6 +296,8 @@
        struct pollfd pfd;
        int rv = 0;
 
+       sendunlockl(spc);
+
        while (rw->rw_data == NULL && spc->spc_dying == 0) {
                /* are we free to receive? */
                if (spc->spc_istatus == SPCSTATUS_FREE) {