Add AtomicPartitionedQueue from moodycamel's ConcurrentQueue

docs/api/AtomicPartitionedQueue.md

@@ -0,0 +1,29 @@
+::: cereggii._cereggii.AtomicPartitionedQueue
+    options:
+        members:
+            - __init__
+            - put
+            - get
+            - try_get
+            - put_many
+            - get_many
+            - try_get_many
+            - close
+            - closed
+            - approx_len
+            - producer
+            - consumer
+
+::: cereggii._cereggii.AtomicPartitionedQueueProducer
+    options:
+        members:
+            - put
+            - put_many
+
+::: cereggii._cereggii.AtomicPartitionedQueueConsumer
+    options:
+        members:
+            - get
+            - try_get
+            - get_many
+            - try_get_many

docs/api/index.md

@@ -6,6 +6,7 @@
 - [AtomicDict](AtomicDict.md) – A lock-free, atomic dictionary implementation
 - [AtomicCache](AtomicCache.md) – A lock-free, atomic key-value cache with invalidation support
 - [AtomicInt64](AtomicInt64.md) – 64-bit atomic integer operations
+- [AtomicPartitionedQueue](AtomicPartitionedQueue.md) – High throughput thread-safe queue
 - [AtomicRef](AtomicRef.md) – Atomic reference to an object with thread-safe operations
 
 ## Concurrency Primitives

examples/atomic_partitioned_queue/bench.py

@@ -0,0 +1,169 @@
+"""Microbenchmark for AtomicPartitionedQueue.
+
+Exercises the four main modes of operation with N producer threads and N
+consumer threads each:
+
+  1. plain   put / get
+  2. ctx     put / get with producer/consumer contexts
+  3. bulk    put_many / get_many
+  4. bulk+ctx put_many / get_many with producer/consumer contexts
+
+Each mode enqueues NUM_PRODUCERS * ITEMS_PER_PRODUCER items in total and
+the consumers cooperatively drain them all. Reported numbers are wall-clock
+time and total items per second (producer + consumer side combined).
+"""
+
+import collections
+import queue
+import threading
+import time
+from threading import Barrier
+
+from cereggii import AtomicInt64, AtomicPartitionedQueue
+
+
+NUM_PRODUCERS = 4
+NUM_CONSUMERS = 4
+ITEMS_PER_PRODUCER = 200_000
+BATCH = 64
+
+TOTAL_ITEMS = NUM_PRODUCERS * ITEMS_PER_PRODUCER
+assert TOTAL_ITEMS % NUM_CONSUMERS == 0
+
+
+def _run(label, producer_fn, consumer_fn, queue_factory=AtomicPartitionedQueue):
+    q = queue_factory()
+    barrier = Barrier(NUM_PRODUCERS + NUM_CONSUMERS + 1)
+    consumed = AtomicInt64(0)
+
+    producers = [threading.Thread(target=producer_fn, args=(q, barrier, i)) for i in range(NUM_PRODUCERS)]
+    consumers = [threading.Thread(target=consumer_fn, args=(q, barrier, consumed)) for _ in range(NUM_CONSUMERS)]
+
+    for t in producers + consumers:
+        t.start()
+
+    barrier.wait()
+    started = time.perf_counter()
+    for t in producers + consumers:
+        t.join()
+    elapsed = time.perf_counter() - started
+
+    got = consumed.get()
+    rate = got / elapsed if elapsed > 0 else float("inf")
+    status = "" if got == TOTAL_ITEMS else f"[MISMATCH (got {got}/{TOTAL_ITEMS})]"
+    print(f"  {label:<25} {elapsed:7.3f}s   {rate/1e6:7.2f} M items/s   {status}")
+
+
+# 1. plain put / get
+def _prod_plain(q, barrier, pid):
+    base = pid * ITEMS_PER_PRODUCER
+    barrier.wait()
+    for i in range(ITEMS_PER_PRODUCER):
+        q.put(base + i)
+
+
+def _cons_plain(q, barrier, consumed):
+    local_consumed = 0
+    barrier.wait()
+    while local_consumed < TOTAL_ITEMS // NUM_CONSUMERS:
+        _ = q.get()
+        local_consumed += 1
+    consumed += local_consumed
+
+
+# 2. put / get with contexts
+def _prod_ctx(q, barrier, pid):
+    base = pid * ITEMS_PER_PRODUCER
+    with q.producer() as p:
+        barrier.wait()
+        for i in range(ITEMS_PER_PRODUCER):
+            p.put(base + i)
+
+
+def _cons_ctx(q, barrier, consumed):
+    local_consumed = 0
+    with q.consumer() as c:
+        barrier.wait()
+        while local_consumed < TOTAL_ITEMS // NUM_CONSUMERS:
+            _ = c.get()
+            local_consumed += 1
+    consumed += local_consumed
+
+
+# 3. put_many / get_many
+def _prod_bulk(q, barrier, pid):
+    base = pid * ITEMS_PER_PRODUCER
+    barrier.wait()
+    for start in range(0, ITEMS_PER_PRODUCER, BATCH):
+        q.put_many(range(base + start, base + start + BATCH))
+
+
+def _cons_bulk(q, barrier, consumed):
+    local_consumed = 0
+    target_consumed = TOTAL_ITEMS // NUM_CONSUMERS
+    barrier.wait()
+    while local_consumed < target_consumed:
+        items = q.get_many(min(BATCH, target_consumed - local_consumed))
+        if items:
+            local_consumed += len(items)
+    consumed += local_consumed
+
+
+# 4. put_many / get_many with contexts
+def _prod_bulk_ctx(q, barrier, pid):
+    base = pid * ITEMS_PER_PRODUCER
+    with q.producer() as p:
+        barrier.wait()
+        for start in range(0, ITEMS_PER_PRODUCER, BATCH):
+            p.put_many(range(base + start, base + start + BATCH))
+
+
+def _cons_bulk_ctx(q, barrier, consumed):
+    local_consumed = 0
+    target_consumed = TOTAL_ITEMS // NUM_CONSUMERS
+    with q.consumer() as c:
+        barrier.wait()
+        while local_consumed < target_consumed:
+            items = c.try_get_many(min(BATCH, target_consumed - local_consumed))
+            if items:
+                local_consumed += len(items)
+    consumed += local_consumed
+
+
+class WrappedDeque(collections.deque):
+    put = collections.deque.append
+    put_many = collections.deque.extend
+
+    def get(self):
+        try:
+            return self.popleft()
+        except IndexError:
+            return None
+
+    def get_many(self, n):
+        try:
+            return [self.popleft() for _ in range(n)]
+        except IndexError:
+            return []
+
+
+def main():
+    print(
+        f"AtomicPartitionedQueue bench: "
+        f"{NUM_PRODUCERS} producers x {NUM_CONSUMERS} consumers, "
+        f"{TOTAL_ITEMS:,} total items, batch={BATCH}\n"
+    )
+    print(f"  {'mode':<25} {'time':>7}     {'throughput':>16}")
+    print(f"  {'-'*25} {'-'*7}     {'-'*16}")
+
+    _run("stdlib queue.Queue", _prod_plain, _cons_plain, queue_factory=queue.Queue)
+    _run("stdlib deque", _prod_plain, _cons_plain, queue_factory=WrappedDeque)
+    _run("stdlib deque many", _prod_bulk, _cons_bulk, queue_factory=WrappedDeque)
+    _run("plain put/get", _prod_plain, _cons_plain)
+    _run("ctx put/get", _prod_ctx, _cons_ctx)
+    _run("plain put_many/get_many", _prod_bulk, _cons_bulk)
+    _run("ctx put_many/get_many", _prod_bulk_ctx, _cons_bulk_ctx)
+
+
+if __name__ == "__main__":
+    main()

mkdocs.yml

@@ -13,6 +13,7 @@ nav:
       - 'api/AtomicDict.md'
       - 'api/AtomicCache.md'
       - 'api/AtomicInt64.md'
+      - 'api/AtomicPartitionedQueue.md'
       - 'api/CountDownLatch.md'
       - 'api/AtomicRef.md'
       - 'api/ThreadHandle.md'

src/CMakeLists.txt

@@ -22,6 +22,11 @@ if (MSVC)
     add_compile_options(/wd4232)  # address of dllimport 'PyType_GenericNew' is not static, identity not guaranteed
     add_compile_options(/wd5045)  # compiler will insert Spectre mitigation for memory load
     add_compile_options(/wd4710)  # function not inlined
+    add_compile_options(/wd4625)  # copy constructor was implicitly defined as deleted
+    add_compile_options(/wd4626)  # assignment operator was implicitly defined as deleted
+    add_compile_options(/wd5026)  # move constructor was implicitly defined as deleted
+    add_compile_options(/wd5027)  # move constructor was implicitly defined as deleted
+    add_compile_options(/wd4514)  # unreferenced inline function has been removed
 
     # from pythoncapi-compat PyAPI_FUNC
     add_compile_options(/wd4210)  # nonstandard extension used: function given file scope
@@ -36,6 +41,7 @@ else()
 
     # warnings ignored because causing many false positives
     add_compile_options(-Wno-cast-function-type)
+    add_compile_options(-Wno-write-strings)
 endif()
 
 message(STATUS "CEREGGII_TSAN=$ENV{CEREGGII_TSAN}")
@@ -50,8 +56,8 @@ endif ()
 message(STATUS "CEREGGII_ASAN=$ENV{CEREGGII_ASAN}")
 
 if (DEFINED ENV{CEREGGII_ASAN})
-    add_compile_options("-fsanitize=address,undefined")
-    add_link_options("-fsanitize=address")
+    add_compile_options("-fsanitize=address,undefined" "-fno-sanitize=vptr")
+    add_link_options("-fsanitize=address,undefined" "-fno-sanitize=vptr")
     add_compile_options("-g")
     add_compile_options("-fno-omit-frame-pointer")
 endif ()
@@ -125,6 +131,7 @@ Python3_add_library(_cereggii MODULE
         "cereggii/atomic_event.c"
         "cereggii/atomic_int.c"
         "cereggii/atomic_ref.c"
+        "cereggii/atomic_partitioned_queue.cpp"
         "cereggii/cereggii.c"
         "cereggii/constants.c"
         "cereggii/thread_handle.c"

src/cereggii/__init__.py

@@ -13,6 +13,7 @@
 from .atomic_event import AtomicEvent  # noqa: F401
 from .atomic_int import AtomicInt64  # noqa: F401
 from .atomic_ref import AtomicRef  # noqa: F401
+from .atomic_partitioned_queue import AtomicPartitionedQueue  # noqa: F401
 from .constants import *  # noqa: F403
 from .count_down_latch import CountDownLatch  # noqa: F401
 from .misc import call_once  # noqa: F401