[busyclosure] Rename ReadyDeque to BusyClosure and update struct members and functions to better reflect the semantics of this new structure, as storing pointers to the busy closures that workers are currently working on.

neboat · neboat · commit 065eb29cd0cf · 2026-01-09T17:06:59.000-05:00
diff --git a/runtime/busyclosure.h b/runtime/busyclosure.h
@@ -0,0 +1,126 @@
+#ifndef _BUSYCLOSURE_H
+#define _BUSYCLOSURE_H
+
+#include "closure.h"
+#include "debug.h"
+#include "global.h"
+#include "rts-config.h"
+#include "worker_coord.h"
+#include <atomic>
+
+struct alignas(CILK_CACHE_LINE) BusyClosure {
+    Closure *closure;
+    alignas(CILK_CACHE_LINE) std::atomic<worker_id> mutex_owner;
+
+    /*********************************************************
+     * Management of BusyClosure
+     *********************************************************/
+
+    void assert_ownership([[maybe_unused]] worker_id self) {
+        CILK_ASSERT(mutex_owner.load(std::memory_order_relaxed) == self);
+    }
+
+    static void assert_ownership(BusyClosure *busy, worker_id self,
+                                 worker_id pn) {
+        busy[pn].assert_ownership(self);
+    }
+
+    void lock(worker_id id) {
+        while (true) {
+            worker_id current_owner =
+                mutex_owner.load(std::memory_order_relaxed);
+            if ((current_owner == NO_WORKER) &&
+                mutex_owner.compare_exchange_weak(current_owner, id,
+                                                  std::memory_order_acq_rel,
+                                                  std::memory_order_relaxed))
+                return;
+            busy_loop_pause();
+        }
+    }
+
+    static void lock(BusyClosure *busy, worker_id self, worker_id pn) {
+        busy[pn].lock(self);
+    }
+
+    static void lock_self(BusyClosure *busy, worker_id self) {
+        busy[self].lock(self);
+    }
+    static void unlock_self(BusyClosure *busy, worker_id self) {
+        worker_id id = self;
+        busy[id].mutex_owner.store(NO_WORKER, std::memory_order_release);
+    }
+
+    bool trylock(worker_id id) {
+        worker_id current_owner = mutex_owner.load(std::memory_order_relaxed);
+        if (current_owner == NO_WORKER)
+            return mutex_owner.compare_exchange_weak(current_owner, id,
+                                                     std::memory_order_acq_rel,
+                                                     std::memory_order_relaxed);
+        return false;
+    }
+
+    static bool trylock(BusyClosure *busy, worker_id self, worker_id pn) {
+        return busy[pn].trylock(self);
+    }
+
+    void unlock([[maybe_unused]] worker_id self) {
+        mutex_owner.store(NO_WORKER, std::memory_order_release);
+    }
+
+    static void unlock(BusyClosure *busy, worker_id self, worker_id pn) {
+        busy[pn].unlock(self);
+    }
+
+    /*
+     * Add/remove the busy closure
+     *
+     * The precondition of these functions is that the worker self
+     * must have locked worker pn's busy closure before entering the function.
+     */
+
+    static Closure *xtract(BusyClosure *busy, worker_id self, worker_id pn) {
+        return busy[pn].xtract(self);
+    }
+
+    Closure *xtract(worker_id self) {
+        // make sure w has the lock on worker pn's busy closure
+        assert_ownership(self);
+
+        if (Closure *cl = closure) {
+            closure = nullptr;
+            cl->owner = NO_WORKER;
+            return cl;
+        }
+        return nullptr;
+    }
+
+    Closure *peek(worker_id self) {
+        // make sure w has the lock on worker pn's busy closure
+        assert_ownership(self);
+        // returns the closure but does not unlink
+        return closure;
+    }
+
+    static Closure *peek(BusyClosure *busy, worker_id self, worker_id pn) {
+        return busy[pn].peek(self);
+    }
+
+    /*
+     * This allows self to make Closure cl worker pn's busy closure.
+     */
+    void set(Closure *cl, worker_id self, worker_id pn) {
+        assert_ownership(self);
+
+        CILK_ASSERT(cl->owner == NO_WORKER);
+        CILK_ASSERT_NULL(closure);
+        closure = cl;
+        cl->owner = pn;
+    }
+
+    static void set(BusyClosure *busy, Closure *cl, worker_id self,
+                    worker_id pn) {
+        return busy[pn].set(cl, self, pn);
+    }
+};
+
+#endif
diff --git a/runtime/closure.h b/runtime/closure.h
@@ -3,10 +3,12 @@
 
 #include "cilk-internal.h"
 #include "local-hypertable.h"
+#include "rts-config.h"
 #include <atomic>
 
-// Forward declaration
-typedef struct Closure Closure;
+// Forward declarations
+struct Closure;
+struct BusyClosure;
 
 enum ClosureStatus : unsigned char {
     /* Closure.status == 0 is invalid */
@@ -37,7 +39,7 @@ struct __attribute__((visibility("hidden"))) Closure {
     struct cilk_fiber *ext_fiber = nullptr;
     struct cilk_fiber *ext_fiber_child = nullptr;
 
-    worker_id owner_ready_deque = NO_WORKER; /* debug only */
+    worker_id owner = NO_WORKER; /* debug only */
 
     enum ClosureStatus status = CLOSURE_PRE_INVALID;
     bool exception_pending = false;
@@ -92,10 +94,9 @@ struct __attribute__((visibility("hidden"))) Closure {
     static void destroy(Closure *, struct global_state *);
 
     // This method is used for sync.
-    void suspend(struct ReadyDeque *deques, worker_id self);
+    void suspend(BusyClosure *busy, worker_id self);
     // This method is used for steal.
-    void suspend_victim(struct ReadyDeque *deques, worker_id thief,
-                        worker_id victim);
+    void suspend_victim(BusyClosure *busy, worker_id thief, worker_id victim);
 
     void add_callee(Closure *new_callee);
     void remove_callee();
diff --git a/runtime/global.cpp b/runtime/global.cpp
@@ -7,9 +7,10 @@
 #ifdef __FreeBSD__
 #include <pthread_np.h>
 #endif
+#include "busyclosure.h"
 #include "debug.h"
 #include "global.h"
-#include "readydeque.h"
+#include "local.h"
 #include <cstdio>
 #include <cstring>
 #include <sched.h>
@@ -177,8 +178,8 @@ global_state *global_state_init(int argc, char *argv[]) {
         (struct worker_args *)calloc(active_size, sizeof(struct worker_args));
     g->workers =
         (__cilkrts_worker **)calloc(active_size, sizeof(__cilkrts_worker *));
-    g->deques = (ReadyDeque *)cilk_aligned_alloc(
-        __alignof__(ReadyDeque), active_size * sizeof(ReadyDeque));
+    g->busy = (BusyClosure *)cilk_aligned_alloc(
+        __alignof__(BusyClosure), active_size * sizeof(BusyClosure));
     g->threads = new std::thread[active_size];
     g->index_to_worker = (worker_id *)calloc(active_size, sizeof(worker_id));
     g->worker_to_index = (worker_id *)calloc(active_size, sizeof(worker_id));
diff --git a/runtime/global.h b/runtime/global.h
@@ -19,6 +19,7 @@ extern unsigned __cilkrts_nproc;
 
 struct __cilkrts_worker;
 struct Closure;
+struct BusyClosure;
 
 // clang-format off
 #define DEFAULT_OPTIONS                                            \
@@ -64,8 +65,8 @@ struct CHEETAH_INTERNAL global_state {
     unsigned int nworkers; /* size of next 4 arrays */
     struct worker_args *worker_args;
     struct __cilkrts_worker **workers;
-    /* dynamically-allocated array of deques, one per processor */
-    struct ReadyDeque *deques;
+    /* dynamically-allocated array of busy closures, one per processor */
+    BusyClosure *busy;
     std::thread *threads;
     struct Closure *root_closure;
 
diff --git a/runtime/init.cpp b/runtime/init.cpp
@@ -16,13 +16,13 @@
 #endif
 #include <unistd.h>
 
+#include "busyclosure.h"
 #include "cilk-internal.h"
 #include "debug.h"
 #include "fiber.h"
 #include "global.h"
 #include "init.h"
 #include "local.h"
-#include "readydeque.h"
 #include "sched_stats.h"
 #include "scheduler.h"
 
@@ -57,11 +57,11 @@ static void worker_local_destroy(local_state *l, global_state *g) {
     /* currently nothing to do here */
 }
 
-static void deques_init(global_state *g) {
-    cilkrts_alert(BOOT, "(deques_init) Initializing deques");
+static void busy_init(global_state *g) {
+    cilkrts_alert(BOOT, "(busy_init) Initializing busy closures");
     for (unsigned int i = 0; i < g->options.nproc; i++) {
-        g->deques[i].bottom = nullptr;
-        g->deques[i].mutex_owner = NO_WORKER;
+        g->busy[i].closure = nullptr;
+        g->busy[i].mutex_owner = NO_WORKER;
     }
 }
 
@@ -363,7 +363,7 @@ global_state *__cilkrts_startup(int argc, char *argv[]) {
     cilkrts_alert(BOOT, "(__cilkrts_startup) argc %d", argc);
     global_state *g = global_state_init(argc, argv);
     workers_init(g);
-    deques_init(g);
+    busy_init(g);
 
     // Create the root closure and a fiber to go with it.  Use worker 0 to
     // allocate the closure and fiber.
@@ -560,7 +560,7 @@ void __cilkrts_internal_exit_cilkified_root(global_state *g,
 
     worker_id self = w->self;
     const bool is_boss = (0 == self);
-    ReadyDeque *deques = g->deques;
+    BusyClosure *busy = g->busy;
 
     // Mark the computation as done.  Also "sleep" the workers: update global
     // flags so workers who exit the work-stealing loop will return to waiting
@@ -579,15 +579,15 @@ void __cilkrts_internal_exit_cilkified_root(global_state *g,
         w->extension = nullptr;
     }
 
-    // Clear this worker's deque.  Nobody can successfully steal from this deque
-    // at this point, because head == tail, but we still want any subsequent
-    // Cilkified region to start with an empty deque.  We go ahead and grab the
-    // deque lock to make sure no other worker has a lingering pointer to the
-    // closure.
-    ReadyDeque::lock_self(deques, self);
-    deques[self].bottom = nullptr;
-    WHEN_CILK_DEBUG(g->root_closure->owner_ready_deque = NO_WORKER);
-    ReadyDeque::unlock_self(deques, self);
+    // Clear this worker's busy closure.  Nobody can successfully steal from
+    // this worker's deque at this point, because head == tail, but any
+    // subsequent Cilkified region should still start with an empty busy
+    // closure.  We go ahead and grab the lock to make sure no other worker
+    // has a lingering pointer to the closure.
+    BusyClosure::lock_self(busy, self);
+    busy[self].closure = nullptr;
+    WHEN_CILK_DEBUG(g->root_closure->owner = NO_WORKER);
+    BusyClosure::unlock_self(busy, self);
 
     // Clear the flags in sf.  This routine runs before leave_frame in a Cilk
     // function, but leave_frame is executed conditionally in Cilk functions
@@ -682,8 +682,8 @@ static void global_state_deinit(global_state *g) {
     free(g->workers);
     g->workers = nullptr;
     g->nworkers = 0;
-    free(g->deques);
-    g->deques = nullptr;
+    free(g->busy);
+    g->busy = nullptr;
     delete [] g->threads;
     g->threads = nullptr;
     free(g->index_to_worker);
@@ -693,10 +693,10 @@ static void global_state_deinit(global_state *g) {
     free(g);
 }
 
-static void deques_deinit(global_state *g) {
-    cilkrts_alert(BOOT, "(deques_deinit) Clean up deques");
+static void busy_deinit(global_state *g) {
+    cilkrts_alert(BOOT, "(busy_deinit) Clean up busy closures");
     for (unsigned int i = 0; i < g->options.nproc; i++) {
-        CILK_ASSERT(g->deques[i].mutex_owner == NO_WORKER);
+        CILK_ASSERT(g->busy[i].mutex_owner == NO_WORKER);
     }
 }
 
@@ -790,7 +790,7 @@ void __cilkrts_shutdown(global_state *g) {
     // internal-malloc that does not include all the free fibers.
     global_state_terminate(g);
     workers_deinit(g);
-    deques_deinit(g);
+    busy_deinit(g);
     global_state_deinit(g);
 }
 
diff --git a/runtime/personality.cpp b/runtime/personality.cpp
@@ -1,3 +1,4 @@
+#include "busyclosure.h"
 #include "cilk-internal.h"
 #include "cilk2c.h"
 #include "cilk2c_inlined.h"
@@ -7,7 +8,6 @@
 #include "frame.h"
 #include "init.h"
 #include "local-reducer-api.h"
-#include "readydeque.h"
 #include <cilk/cilk_api.h>
 #include <cstdint>
 #include <cstring>
@@ -120,7 +120,7 @@ __attribute__((noinline)) static void
 sync_in_personality(__cilkrts_worker *w, __cilkrts_stack_frame *sf,
                     struct _Unwind_Exception *ue_header) {
     worker_id self = w->self;
-    ReadyDeque *deques = w->g->deques;
+    BusyClosure *busy = w->g->busy;
     // save floating point state
     sysdep_save_fp_ctrl_state(sf);
 
@@ -129,16 +129,16 @@ sync_in_personality(__cilkrts_worker *w, __cilkrts_stack_frame *sf,
         struct closure_exception *exn_r = get_exception_reducer(w);
         exn_r->exn = (char *)ue_header;
 
-        ReadyDeque::lock_self(deques, self);
-        Closure *t = ReadyDeque::peek_bottom(deques, self, self);
+        BusyClosure::lock_self(busy, self);
+        Closure *t = BusyClosure::peek(busy, self, self);
         t->lock(self);
 
         // ensure that we return here after a cilk_sync.
         exn_r->parent_rsp = t->orig_rsp;
         t->orig_rsp = (char *)SP(sf);
 
         t->unlock(self);
-        ReadyDeque::unlock_self(deques, self);
+        BusyClosure::unlock_self(busy, self);
 
         // save the current fiber for further stack unwinding.
         if (exn_r->throwing_fiber == nullptr) {
diff --git a/runtime/readydeque.h b/runtime/readydeque.h
diff --git a/runtime/scheduler.cpp b/runtime/scheduler.cpp
