Merge pull request #2 from K-T0BIAS/experimental

fix #1 and add a Barrier in cthreads.sync

Author: K-T0BIAS

.gitignore

@@ -54,6 +54,7 @@ __Threadable__/
 .cthreads_cache.json
 cthreads_kernels.dll
 cthreads_kernels.so
+cthreads_kernels.lib
 libcthreads_kernels.so
 libcthreads_kernels.dylib
 # <<< cthreads (auto)

src/cthreads/cpp/bindings/module.cpp

@@ -10,6 +10,7 @@
 #include "../headers/math/clamps.hpp"
 #include "../headers/math/random.hpp"
 #include "../headers/pyThread.hpp"
+#include "../headers/sync/pyBarrier.hpp"
 #include "../headers/sync/pyEvent.hpp"
 #include "../headers/sync/pyLock.hpp"
 #include "../headers/sync/pyRWLock.hpp"
@@ -675,16 +676,28 @@ std::shared_ptr<SpawnedKernel> spawn_from_meta(
     auto job = make_kernel_job(self, call_fn, params_keep);
     if (pool) {
         // Keep SpawnedKernel alive until the task runs or is dropped from the queue.
+        // Python-owned handles in `job` / `spawned` must only be destroyed while the
+        // GIL is held (pool workers are bare OS threads).
         struct PoolTaskState {
             std::shared_ptr<SpawnedKernel> spawned;
             std::function<void()> job;
             std::atomic<bool> started{false};
+
+            void release_python_state() {
+                py::gil_scoped_acquire gil;
+                job = nullptr;
+                spawned.reset();
+            }
+
             ~PoolTaskState() {
+                py::gil_scoped_acquire gil;
                 if (!started.load(std::memory_order_acquire) && spawned) {
                     spawned->mark_done(std::make_exception_ptr(std::runtime_error(
                         "cthreads.pool: job dropped (pool stopped before run)"
                     )));
                 }
+                job = nullptr;
+                spawned.reset();
             }
         };
         auto st = std::make_shared<PoolTaskState>();
@@ -699,6 +712,7 @@ std::shared_ptr<SpawnedKernel> spawn_from_meta(
             } catch (...) {
                 st->spawned->mark_done(std::current_exception());
             }
+            st->release_python_state();
         });
     } else {
         // Dedicated path (unchanged): one OS thread via CThread.
@@ -837,8 +851,11 @@ std::uintptr_t sync_native_ptr(py::object obj) {
     if (py::isinstance<cthreads::sync::RWLock>(obj)) {
         return reinterpret_cast<std::uintptr_t>(&obj.cast<cthreads::sync::RWLock&>());
     }
+    if (py::isinstance<cthreads::sync::Barrier>(obj)) {
+        return reinterpret_cast<std::uintptr_t>(&obj.cast<cthreads::sync::Barrier&>());
+    }
     throw std::runtime_error(
-        "cthreads.sync_native_ptr: expected Lock, Event, or RWLock");
+        "cthreads.sync_native_ptr: expected Lock, Event, RWLock, or Barrier");
 }
 
 #include "pool.tpp"
@@ -877,7 +894,7 @@ PYBIND11_MODULE(_ext, m) {
         "sync_native_ptr",
         &sync_native_ptr,
         py::arg("obj"),
-        "Return the native address of a cthreads.sync Lock/Event/RWLock "
+        "Return the native address of a cthreads.sync Lock/Event/RWLock/Barrier "
         "for kernel marshalling."
     );
 
@@ -995,6 +1012,14 @@ PYBIND11_MODULE(_ext, m) {
 
     rwlock.attr("__cthreads_internal__") = true;
 
+    auto barrier = py::class_<cthreads::sync::Barrier>(sync, "Barrier")
+        .def(py::init<std::size_t>(), py::arg("parties"))
+        .def("parties", &cthreads::sync::Barrier::parties)
+        .def("arrive_and_wait", &cthreads::sync::Barrier::arrive_and_wait,
+             py::call_guard<py::gil_scoped_release>());
+
+    barrier.attr("__cthreads_internal__") = true;
+
     // Fixed-capacity triple buffers (baseline bindings for primitive/container types).
     // Thread-side writes are typically codegen'd as: buf[i].field = ... (for Threadables)
     // or buf[i] = value (for scalar/container slots).

src/cthreads/cpp/headers/pool/threadPool.hpp

@@ -69,11 +69,15 @@ namespace cthreads::pool {
                 this->cv.notify_all(); // notify all waiting worker threads that the pool is stopping
                 this->join(); // wait for in-flight tasks; queued tasks are dropped below
                 this->threads.clear();
+                // Destroy dropped tasks outside the queue mutex: their destructors may
+                // acquire the GIL, and Python code can wait on this mutex while holding it.
+                std::queue<std::function<void()>> dropped;
                 {
                     std::lock_guard<std::mutex> lock(this->tasks_queue_mutex);
-                    while (!this->tasks_queue.empty()) {
-                        this->tasks_queue.pop();
-                    }
+                    dropped.swap(this->tasks_queue);
+                }
+                while (!dropped.empty()) {
+                    dropped.pop();
                 }
                 this->started = false;
                 this->stop_signal.store(false);

src/cthreads/cpp/headers/sync/pyBarrier.hpp

@@ -0,0 +1,52 @@
+// Copyright (c) 2026 Tobias Karusseit
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+#pragma once
+
+#include <condition_variable>
+#include <cstddef>
+#include <mutex>
+#include <stdexcept>
+
+namespace cthreads::sync {
+
+/**
+ * Fixed-party generation barrier for long-lived @Thread workers.
+ *
+ * All `parties` threads must call `arrive_and_wait()` before any proceeds.
+ * Reusable across phases (grid → density → dynamics → …).
+ */
+class Barrier {
+    std::mutex _mu;
+    std::condition_variable _cv;
+    const std::size_t _parties;
+    std::size_t _count = 0;
+    std::size_t _generation = 0;
+
+public:
+    explicit Barrier(std::size_t parties) : _parties(parties) {
+        if (parties == 0) {
+            throw std::invalid_argument("cthreads.sync.Barrier: parties must be >= 1");
+        }
+    }
+
+    Barrier(const Barrier&) = delete;
+    Barrier& operator=(const Barrier&) = delete;
+
+    std::size_t parties() const { return _parties; }
+
+    void arrive_and_wait() {
+        std::unique_lock<std::mutex> g(_mu);
+        const std::size_t gen = _generation;
+        if (++_count == _parties) {
+            _count = 0;
+            ++_generation;
+            _cv.notify_all();
+            return;
+        }
+        _cv.wait(g, [this, gen] { return gen != _generation; });
+    }
+};
+
+}  // namespace cthreads::sync

src/cthreads/python/cthreads/compiler/translation/plugins/sync/Sync.py

@@ -45,6 +45,10 @@ class SyncMethodPlugin(MethodTablePlugin):
             "release_write": method_op("release_write", 0),
             "try_acquire_write": method_op("try_acquire_write", 0),
         },
+        "Barrier": {
+            "arrive_and_wait": method_op("arrive_and_wait", 0),
+            "parties": method_op("parties", 0),
+        },
     }
 
     def type_key(self, py_type: PyType) -> str | None:

src/cthreads/python/cthreads/sync/__init__.py

@@ -29,11 +29,13 @@
     Lock = _native.Lock
     Event = _native.Event
     RWLock = getattr(_native, "RWLock", None)
+    Barrier = getattr(_native, "Barrier", None)
     TBufferI64 = getattr(_native, "TBufferI64", None)
 else:
     Lock = None  # type: ignore[assignment,misc]
     Event = None  # type: ignore[assignment,misc]
     RWLock = None  # type: ignore[assignment,misc]
+    Barrier = None  # type: ignore[assignment,misc]
     TBufferI64 = None  # type: ignore[assignment,misc]
 
 __all__ = [
@@ -48,5 +50,6 @@
     "Lock",
     "Event",
     "RWLock",
+    "Barrier",
     "TBufferI64",
 ]

src/cthreads/python/cthreads/types/pyType/internal/TBuffer.py

@@ -64,7 +64,7 @@ def is_tbuffer_pytype(py_type: PyType) -> bool:
 
 
 def is_sync_pytype(py_type: PyType) -> bool:
-    """True for `Lock` / `Event` / `RWLock` kernel params (non-copyable)."""
+    """True for `Lock` / `Event` / `RWLock` / `Barrier` kernel params (non-copyable)."""
     return (
         isinstance(py_type, PyCThreadsInternalType)
         and py_type.name in SYNC_INTERNAL_NAMES

src/cthreads/python/cthreads/types/pyType/internal/include_map.py

@@ -1,6 +1,6 @@
 # Native types shipped in the pybind module (not codegen'd as Threadables).
 # keyed by Python class __name__ as exposed on cthreads.sync / cthreads.linalg
-SYNC_INTERNAL_NAMES: frozenset[str] = frozenset({"Lock", "Event", "RWLock"})
+SYNC_INTERNAL_NAMES: frozenset[str] = frozenset({"Lock", "Event", "RWLock", "Barrier"})
 
 CTHREADS_INTERNAL_TYPES: dict[str, dict[str, str]] = {
     "Lock": {
@@ -15,6 +15,10 @@
         "cpp_name": "cthreads::sync::RWLock",
         "cpp_include": "sync/pyRWLock.hpp",
     },
+    "Barrier": {
+        "cpp_name": "cthreads::sync::Barrier",
+        "cpp_include": "sync/pyBarrier.hpp",
+    },
     # Fixed-capacity triple buffers (cthreads.sync.TBuffer*).
     "TBufferF64": {
         "cpp_name": "cthreads::sync::tripple_buffer<double>",