[PATCH] libports: implement lockless management of threads

[Justus Winter]

ports_manage_port_operations_multithread uses two values, totalthreads
and nreqthreads, to manage the threads it creates.

Previously the two values were stored in two variables and a lock was
used to synchronize the access to them.

Use a single variable thread_counts to hold both values. This way,
both values can be manipulated consistently using atomic operations.

* libports/manage-multithread.c (ports_manage_port_operations_multithread):
  Combine totalthreads and nreqthreads and use atomic operations
  instead of a lock to synchronize the access.
---
 libports/manage-multithread.c |   73 +++++++++++++++++++++--------------------
 1 file changed, 38 insertions(+), 35 deletions(-)

diff --git a/libports/manage-multithread.c b/libports/manage-multithread.c
index 0c2da00..012059e 100644
--- a/libports/manage-multithread.c
+++ b/libports/manage-multithread.c
@@ -91,9 +91,18 @@ ports_manage_port_operations_multithread (struct port_bucket 
*bucket,
                                          int global_timeout,
                                          void (*hook)())
 {
-  volatile unsigned int nreqthreads;
-  volatile unsigned int totalthreads;
-  pthread_spinlock_t lock = PTHREAD_SPINLOCK_INITIALIZER;
+  /* The variable thread_counts holds two values, totalthreads and
+     nreqthreads.  This way, both values can be manipulated
+     consistently using atomic operations.  The most significant
+     half-word holds the value totalthreads, the least significant
+     half-word nreqthreads.  The initial value accounts for the main
+     thread.  */
+  unsigned int thread_counts = 1 << 16 | 1;
+
+  /* Provide convenience macros to make the code more readable.  */
+  #define NREQTHREADS(x)       ((x) & 0xffff)
+  #define TOTALTHREADS(x)      ((x) >> 16)
+
   pthread_attr_t attr;
 
   auto void * thread_function (void *);
@@ -120,30 +129,26 @@ ports_manage_port_operations_multithread (struct 
port_bucket *bucket,
                /* msgt_unused = */             0
        };
 
-      pthread_spin_lock (&lock);
-      assert (nreqthreads);
-      nreqthreads--;
-      if (nreqthreads != 0)
-       pthread_spin_unlock (&lock);
-      else
+      /* Decrement nreqthreads.  */
+      unsigned int tc = __atomic_sub_fetch (&thread_counts, 1,
+                                           __ATOMIC_RELAXED);
+      if (NREQTHREADS (tc) == 0)
        /* No thread would be listening for requests, spawn one. */
        {
          pthread_t pthread_id;
          error_t err;
 
-         totalthreads++;
-         nreqthreads++;
-         pthread_spin_unlock (&lock);
+         /* Increment totalthreads and nreqthreads.  */
+         __atomic_add_fetch (&thread_counts, 1 << 16 | 1, __ATOMIC_RELAXED);
 
          err = pthread_create (&pthread_id, &attr, thread_function, NULL);
          if (!err)
            pthread_detach (pthread_id);
          else
            {
-             pthread_spin_lock (&lock);
-             totalthreads--;
-             nreqthreads--;
-             pthread_spin_unlock (&lock);
+             /* Decrement totalthreads and nreqthreads.  */
+             __atomic_sub_fetch (&thread_counts, 1 << 16 | 1,
+                                 __ATOMIC_RELAXED);
              /* There is not much we can do at this point.  The code
                 and design of the Hurd servers just don't handle
                 thread creation failure.  */
@@ -189,9 +194,8 @@ ports_manage_port_operations_multithread (struct 
port_bucket *bucket,
          status = 1;
        }
 
-      pthread_spin_lock (&lock);
-      nreqthreads++;
-      pthread_spin_unlock (&lock);
+      /* Increment nreqthreads.  */
+      __atomic_add_fetch (&thread_counts, 1, __ATOMIC_RELAXED);
 
       return status;
     }
@@ -203,8 +207,8 @@ ports_manage_port_operations_multithread (struct 
port_bucket *bucket,
       int timeout;
       error_t err;
 
-      /* No need to lock as an approximation is sufficient. */
-      adjust_priority (totalthreads);
+      adjust_priority (TOTALTHREADS (__atomic_load_n (&thread_counts,
+                                                     __ATOMIC_RELAXED)));
 
       if (hook)
        (*hook) ();
@@ -224,30 +228,29 @@ ports_manage_port_operations_multithread (struct 
port_bucket *bucket,
 
       if (master)
        {
-         pthread_spin_lock (&lock);
-         if (totalthreads != 1)
-           {
-             pthread_spin_unlock (&lock);
-             goto startover;
-           }
+         if (TOTALTHREADS (__atomic_load_n (&thread_counts,
+                                            __ATOMIC_RELAXED)) != 1)
+           goto startover;
        }
       else
        {
-         pthread_spin_lock (&lock);
-         if (nreqthreads == 1)
+         unsigned int tc;
+         /* Decrement totalthreads and nreqthreads.  */
+         tc = __atomic_sub_fetch (&thread_counts, 1 << 16 | 1,
+                                  __ATOMIC_RELAXED);
+         if (NREQTHREADS (tc) == 0)
            {
              /* No other thread is listening for requests, continue. */
-             pthread_spin_unlock (&lock);
+             /* Increment totalthreads and nreqthreads.  */
+             __atomic_add_fetch (&thread_counts, 1 << 16 | 1,
+                                 __ATOMIC_RELAXED);
              goto startover;
            }
-         nreqthreads--;
-         totalthreads--;
-         pthread_spin_unlock (&lock);
        }
       return NULL;
     }
 
-  nreqthreads = 1;
-  totalthreads = 1;
   thread_function ((void *) 1);
+  #undef NREQTHREADS
+  #undef TOTALTHREADS
 }
-- 
1.7.10.4