Add RTS worker sync pause

Introduce a cooperative sync-pause primitive for RTS worker threads. A worker
can request a pause, wake its peers, wait until they park at worker-loop
safepoints, and release them when the synchronized operation completes.

Pause requests fail before the worker pool is running or while another pause is
active. Active pause waits time out briefly and re-check rts_exit, so graceful
shutdown cannot leave the owner or parked workers asleep on the pause condition.

Author: Kristian Larsson

base/rts/common.h

@@ -35,6 +35,9 @@ void acton_free(void* ptr);
 char *acton_strdup(const char *s);
 char *acton_strndup(const char *s, size_t n);
 
+int acton_sync_pause_begin(void);
+void acton_sync_pause_end(void);
+
 void *acton_gc_malloc(size_t size);
 void *acton_gc_malloc_atomic(size_t size);
 void *acton_gc_realloc(void* ptr, size_t size);

base/rts/rts.c

@@ -234,6 +234,139 @@ void pin_actor_affinity() { }
 void set_actor_affinity(int wthread_id) { }
 #endif // ACTON_THREADS
 
+#ifdef ACTON_THREADS
+static pthread_mutex_t sync_pause_lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t sync_pause_cond = PTHREAD_COND_INITIALIZER;
+static int sync_pause_requested = 0;
+static int sync_pause_owner = -1;
+static int sync_pause_parked_count = 0;
+static int sync_pause_workers_are_started = 0;
+static int sync_pause_parked[MAX_WTHREADS];
+
+static int sync_pause_current_worker_id(void) {
+    WorkerCtx wctx = (WorkerCtx)pthread_getspecific(pkey_wctx);
+    if (wctx == NULL) {
+        return -1;
+    }
+    return (int)wctx->id;
+}
+
+static int sync_pause_target_count(void) {
+    return (int)num_wthreads;
+}
+
+static void sync_pause_clear_parked(void) {
+    for (int i = 0; i <= num_wthreads; i++) {
+        sync_pause_parked[i] = 0;
+    }
+}
+
+static void sync_pause_wake_workers(void) {
+    for (int i = 0; i <= num_wthreads; i++) {
+        if (i != sync_pause_owner) {
+            uv_async_send(&wake_ev[i]);
+        }
+    }
+}
+
+static void sync_pause_wait(void) {
+    struct timespec ts;
+    clock_gettime(CLOCK_REALTIME, &ts);
+    ts.tv_nsec += 10 * 1000 * 1000;
+    if (ts.tv_nsec >= 1000 * 1000 * 1000) {
+        ts.tv_sec++;
+        ts.tv_nsec -= 1000 * 1000 * 1000;
+    }
+    pthread_cond_timedwait(&sync_pause_cond, &sync_pause_lock, &ts);
+}
+
+int acton_sync_pause_begin(void) {
+    int owner = sync_pause_current_worker_id();
+    if (owner < 0 || owner > num_wthreads) {
+        return -1;
+    }
+
+    pthread_mutex_lock(&sync_pause_lock);
+    if (!sync_pause_workers_are_started || sync_pause_requested) {
+        pthread_mutex_unlock(&sync_pause_lock);
+        return -1;
+    }
+    if (rts_exit) {
+        pthread_mutex_unlock(&sync_pause_lock);
+        return -1;
+    }
+
+    sync_pause_requested = 1;
+    sync_pause_owner = owner;
+    sync_pause_parked_count = 0;
+    sync_pause_clear_parked();
+
+    sync_pause_wake_workers();
+    while (sync_pause_parked_count < sync_pause_target_count() && !rts_exit) {
+        sync_pause_wait();
+    }
+    if (rts_exit) {
+        sync_pause_requested = 0;
+        sync_pause_owner = -1;
+        sync_pause_parked_count = 0;
+        pthread_cond_broadcast(&sync_pause_cond);
+        pthread_mutex_unlock(&sync_pause_lock);
+        return -1;
+    }
+
+    pthread_mutex_unlock(&sync_pause_lock);
+    return 0;
+}
+
+void acton_sync_pause_end(void) {
+    int owner = sync_pause_current_worker_id();
+    if (owner < 0 || owner > num_wthreads) {
+        return;
+    }
+
+    pthread_mutex_lock(&sync_pause_lock);
+    if (!sync_pause_requested || sync_pause_owner != owner) {
+        pthread_mutex_unlock(&sync_pause_lock);
+        return;
+    }
+
+    sync_pause_requested = 0;
+    sync_pause_owner = -1;
+    sync_pause_parked_count = 0;
+    pthread_cond_broadcast(&sync_pause_cond);
+    pthread_mutex_unlock(&sync_pause_lock);
+}
+
+static void sync_pause_safepoint(void) {
+    int id = sync_pause_current_worker_id();
+    if (id < 0 || id >= MAX_WTHREADS) {
+        return;
+    }
+
+    pthread_mutex_lock(&sync_pause_lock);
+    while (sync_pause_requested && id != sync_pause_owner && !rts_exit) {
+        if (!sync_pause_parked[id]) {
+            sync_pause_parked[id] = 1;
+            sync_pause_parked_count++;
+            pthread_cond_broadcast(&sync_pause_cond);
+        }
+        sync_pause_wait();
+    }
+    pthread_mutex_unlock(&sync_pause_lock);
+}
+
+static void sync_pause_workers_started(void) {
+    pthread_mutex_lock(&sync_pause_lock);
+    sync_pause_workers_are_started = 1;
+    pthread_mutex_unlock(&sync_pause_lock);
+}
+#else
+int acton_sync_pause_begin(void) { return 0; }
+void acton_sync_pause_end(void) { }
+static void sync_pause_safepoint(void) { }
+static void sync_pause_workers_started(void) { }
+#endif
+
 void wake_wt(int wtid) {
     // We are sometimes optimistically called, i.e. the caller sometimes does
     // not really know whether there is new work or not. We check and if there
@@ -1533,6 +1666,7 @@ void wt_work_cb(uv_check_t *ev) {
 
     uv_clock_gettime(UV_CLOCK_MONOTONIC, &ts_start);
     while (true) {
+        sync_pause_safepoint();
         if (rts_exit) {
             return;
         }
@@ -2605,9 +2739,9 @@ int main(int argc, char **argv) {
     }
 
 #ifdef ACTON_THREADS
-    if (num_wthreads > MAX_WTHREADS) {
-        fprintf(stderr, "ERROR: Maximum of %d worker threads supported.\n", MAX_WTHREADS);
-        fprintf(stderr, "HINT: Run this program with fewer worker threads: %s --rts-wthreads %d\n", argv[0], MAX_WTHREADS);
+    if (num_wthreads >= MAX_WTHREADS) {
+        fprintf(stderr, "ERROR: Maximum of %d worker threads supported.\n", MAX_WTHREADS - 1);
+        fprintf(stderr, "HINT: Run this program with fewer worker threads: %s --rts-wthreads %d\n", argv[0], MAX_WTHREADS - 1);
         exit(1);
     }
     // Determine number of worker threads, normally 1:1 per CPU thread / core
@@ -2858,6 +2992,7 @@ int main(int argc, char **argv) {
             //pthread_setaffinity_np(threads[idx-1], sizeof(cpu_set), &cpu_set);
         }
     }
+    sync_pause_workers_started();
 
     pthread_attr_destroy(&ss_attr);
 #endif