feat: update ConcurrentBuffer and ConcurrentQueue for improved performance and add usage guidelines

Author: Synaptic724

CHANGELOG.md

@@ -14,7 +14,7 @@ This project adheres to [Semantic Versioning](https://semver.org/).
 
 ---
 
-## [1.1.0] - 2025-03-23
+## [1.1.0] - 2025-03-26
 ### Classes Added
 ### 1. Dynaphore
 - A dynamic semaphore class that allows for dynamic tuning of semaphore limits.

src/thread_factory/concurrency/__init__.py

@@ -6,6 +6,7 @@
 from src.thread_factory.concurrency.concurrent_stack import ConcurrentStack
 from src.thread_factory.concurrency.concurrent_buffer import ConcurrentBuffer
 
+
 __all__ = [
     "ConcurrentBag",
     "ConcurrentDict",

src/thread_factory/concurrency/concurrent_buffer.py

@@ -147,9 +147,10 @@ class ConcurrentBuffer(Generic[_T]):
 
     **NOTE**: This buffer is not designed for high-contention scenarios.
     ConcurrentQueue or ConcurrentStack Outperforms this object in high-contention scenarios.
+    DO NOT EXCEED 20 THREADS OVERALL (for producer and consumer pattern) WHEN USING THIS OBJECT.
 
-    4-6 Shards have given quite good results in most scenarios with under 10 threads, it can
-    outperform the ConcurrentQueue in some scenarios by 2x
+    Please follow the rule of dividing the number of threads by 2 for the producer and consumer pattern
+    to obtain shard count. For example, if you have 10 threads, you should use 5 shards.
     """
 
     def __init__(
@@ -183,13 +184,12 @@ def __init__(
 
     def enqueue(self, item: _T) -> None:
         """
-        Adds a new item to the buffer. The item is randomly assigned to one of the internal shards.
+        Adds a new item to the buffer. The item is to the queue with the least length.
 
         Args:
             item (_T): The item to add.
         """
-        # Choose a random shard to enqueue the item into. This helps distribute the load.
-        shard_idx = random.randint(0, self._num_shards - 1)
+        shard_idx = min(range(self._num_shards), key=lambda i: self._length_array[i])
         shard = self._shards[shard_idx]
         shard.enqueue_item(item)
 

src/thread_factory/concurrency/concurrent_queue.py

@@ -56,7 +56,7 @@ def enqueue(self, item: _T) -> None:
         with self._lock:
             self._deque.append(item)
 
-    def dequeue(self) -> _T:
+    def dequeue1(self) -> _T:
         """
         Remove and return an item from the front of the queue.
 
@@ -72,9 +72,24 @@ def dequeue(self) -> _T:
                     raise Empty("dequeue from empty ConcurrentQueue")
                 return self._deque.popleft()
         except Empty:
-            time.sleep(0.10)
+            #time.sleep(0.001)
             raise
 
+    def dequeue(self) -> _T:
+        """
+        Remove and return an item from the front of the queue.
+
+        Raises:
+            IndexError: If the queue is empty.
+
+        Returns:
+            _T: The item dequeued.
+        """
+        with self._lock:
+            if not self._deque:
+                raise Empty("dequeue from empty ConcurrentQueue")
+            return self._deque.popleft()
+
     def peek(self) -> _T:
         """
         Return (but do not remove) the item at the front of the queue.

src/thread_factory/threads/testing.py

@@ -1,47 +1,3 @@
 import threading
 import cProfile
 import pstats
-
-
-def test():
-    print("Test begins")
-    print("Lock Acquired")
-    print("This is the thread ID " + str(threading.get_ident()))
-    x = threading.RLock()
-    x.acquire()
-    print("Lock Released")
-    x.release()
-    print("Deadlock")
-    x.acquire()
-    x.acquire()
-    print("Test ends")
-
-test()
-
-# profiler = cProfile.Profile()
-# profiler.enable()
-#
-# # 🔁 Run your test or logic here
-# my_buffer.stress_test()
-#
-# profiler.disable()
-#
-# # 📊 Print top 30 slowest functions by cumulative time
-# stats = pstats.Stats(profiler).sort_stats("cumtime")
-# stats.print_stats(30)
-
-
-
-testing = threading.Thread(target=test)
-testing2 = threading.Thread(target=test)
-testing3 = threading.Thread(target=test)
-testing4 = threading.Thread(target=test)
-
-
-testing.start()
-testing2.start()
-testing3.start()
-testing4.start()
-
-
-

tests/concurrency/test_concurrent_buffer.py

@@ -550,10 +550,10 @@ def test_threaded_concurrent_buffer_performance(self):
         High-performance producer/consumer test using threads and ConcurrentBuffer.
         Balanced consumer workload to avoid stalling.
         """
-        self.buffer = ConcurrentBuffer(4)
-        producers = 8
-        consumers = 8
-        self.items_per_producer = 50_000
+        self.buffer = ConcurrentBuffer(10)
+        producers = 10
+        consumers = 10
+        self.items_per_producer = 100_000
         total_items = producers * self.items_per_producer
 
         try:
@@ -580,7 +580,8 @@ def consumer(target):
                     _ = self.buffer.dequeue()
                     consumed += 1
                 except Empty:
-                    time.sleep(0.001)
+                    pass
+                    #time.sleep(0.001)
 
         threads = []
         for pid in range(producers):
@@ -664,9 +665,9 @@ def test_threaded_concurrent_queue_performance(self):
         """
         from thread_factory import ConcurrentQueue
         q = ConcurrentQueue()
-        producers = 8
-        consumers = 8
-        items_per_producer = 50_000
+        producers = 10
+        consumers = 10
+        items_per_producer = 100_000
         total_items = producers * items_per_producer
 
         try:
@@ -693,7 +694,8 @@ def consumer(target):
                     _ = q.dequeue()
                     consumed += 1
                 except Empty:
-                    time.sleep(0.001)
+                    pass
+                    #time.sleep(0.001)
 
         threads = []
         for pid in range(producers):

tests/concurrency/test_concurrent_queue.py

@@ -520,7 +520,7 @@ def test_threaded_concurrent_queue_performance(self):
         q = ConcurrentQueue()
         producers = 10
         consumers = 10
-        items_per_producer = 100000
+        items_per_producer = 100_000
         total_items = producers * items_per_producer
 
         try:
@@ -574,7 +574,7 @@ def test_multiprocessing_queue_performance(self):
         queue = multiprocessing.Queue()
         producers = 10
         consumers = 10
-        items_per_producer = 100000
+        items_per_producer = 100_000
         total_items = producers * items_per_producer
 
         try: