Repository: alex-petrenko/faster-fifo (language: python)

Issue:
Queue() constructor - missing `maxsize` argument.
Not fully compatible with standard Queue. Missing `maxsize` argument. 


Discussion / hints:
Indeed this should be implemented. Feel free to submit a PR!
This issue is stale because it has been open for 30 days with no activity.
is there an alternative?  need to set limit on max queue size in my application.  if queue is full, i want the producer to drop data.  its a video analytics application and I need to drop frames to keep up with processing demand.
Why not catch the `Full` exception that's raised when there's no more space in the queue?
You can drop a frame every time the exception is raised

Fix patch:
diff --git a/cpp_faster_fifo/cpp_lib/faster_fifo.cpp b/cpp_faster_fifo/cpp_lib/faster_fifo.cpp
index 197450e..48408e6 100644
--- a/cpp_faster_fifo/cpp_lib/faster_fifo.cpp
+++ b/cpp_faster_fifo/cpp_lib/faster_fifo.cpp
@@ -19,7 +19,7 @@
 
 
 struct Queue {
-    explicit Queue(size_t max_size_bytes) : max_size_bytes(max_size_bytes) {
+    explicit Queue(size_t max_size_bytes, size_t maxsize) : max_size_bytes(max_size_bytes), maxsize(maxsize) {
         pthread_mutexattr_init(&mutex_attr);
         pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED);
         pthread_mutex_init(&mutex, &mutex_attr);
@@ -37,10 +37,17 @@ struct Queue {
         return max_size_bytes;
     }
 
-    [[nodiscard]] bool can_fit(size_t data_size) const {
-        return size + data_size <= max_size_bytes;
+    [[nodiscard]] size_t get_max_size() const {
+        return maxsize;
     }
 
+[[nodiscard]] bool can_fit(size_t data_size, size_t additional_size) const {
+    const bool cond_size = size + data_size <= max_size_bytes;
+    const bool cond_num = num_elem + additional_size <= maxsize;
+
+    return cond_size && cond_num;
+}
+
     /// This function does not check if there is enough space in the circular buffer, assuming the check
     /// has been performed.
     void circular_buffer_write(uint8_t *buffer, const uint8_t *data, const size_t data_size) {
@@ -78,7 +85,7 @@ struct Queue {
         const auto new_size = size - read_size;
 
         LOG_ASSERT(new_head < max_size_bytes, "Circular buffer head pointer is incorrect");
-        LOG_ASSERT(new_size >= 0 && new_size < max_size_bytes, "New size is incorrect after reading from buffer");
+        LOG_ASSERT(new_size >= 0 && new_size <= max_size_bytes, "New size is incorrect after reading from buffer");
 
         if (pop_message) {
             head = new_head;
@@ -90,6 +97,7 @@ struct Queue {
     // 9 bytes is the min message size. 8 bytes for the size and 1 for the minimal message
     static const size_t MIN_MSG_SIZE = sizeof(size_t) + 1;
     size_t max_size_bytes;
+    size_t maxsize;
     size_t head = 0, tail = 0, size = 0;
     size_t num_elem = 0;
 
@@ -120,8 +128,8 @@ size_t queue_object_size() {
     return sizeof(Queue);
 }
 
-void create_queue(void *queue_obj_memory, size_t max_size_bytes) {
-    new(queue_obj_memory) Queue(max_size_bytes);  // placement new
+void create_queue(void *queue_obj_memory, size_t max_size_bytes, size_t maxsize) {
+    new(queue_obj_memory) Queue(max_size_bytes, maxsize); 
 }
 
 struct timeval float_seconds_to_timeval(float seconds) {
@@ -169,7 +177,8 @@ int queue_put(void *queue_obj, void *buffer, const void **msgs_data, const size_
             total_size += msg_sizes[i];
 
         auto wait_remaining = float_seconds_to_timeval(timeout);
-        while (!q->can_fit(total_size)) {
+        while (!q->can_fit(total_size, num_msgs)) {
+            
             if (!block || !timer_positive(wait_remaining))
                 return Q_FULL;
 
@@ -194,7 +203,7 @@ int queue_put(void *queue_obj, void *buffer, const void **msgs_data, const size_
     
     if (q->not_empty_n_waiters > 0)
         pthread_cond_signal(&q->not_empty);
-    else if (q->not_full_n_waiters && q->can_fit(Queue::MIN_MSG_SIZE)) {
+    else if (q->not_full_n_waiters && q->can_fit(Queue::MIN_MSG_SIZE, 1)) {
         // In the case of many producers and one batched consumer, producers
         // should wake each other up as the batched consumer is only guaranteed to
         // wake up 1 producer its pthread_cond_signal(&q->not_full).
@@ -283,5 +292,6 @@ size_t get_data_size(void *queue_obj) {
 bool is_queue_full(void *queue_obj) {
     auto q = (Queue *)queue_obj;
     constexpr size_t min_message_size = 1;
-    return !q->can_fit(min_message_size + sizeof(min_message_size));
+    constexpr size_t min_messages_count = 1;
+    return !q->can_fit(min_message_size + sizeof(min_message_size), min_messages_count);
 }
diff --git a/cpp_faster_fifo/cpp_lib/faster_fifo.hpp b/cpp_faster_fifo/cpp_lib/faster_fifo.hpp
index 37413ed..73247a7 100644
--- a/cpp_faster_fifo/cpp_lib/faster_fifo.hpp
+++ b/cpp_faster_fifo/cpp_lib/faster_fifo.hpp
@@ -8,7 +8,7 @@ constexpr int Q_SUCCESS = 0,
 
 
 size_t queue_object_size();
-void create_queue(void *queue_obj, size_t max_size_bytes);
+void create_queue(void *queue_obj, size_t max_size_bytes, size_t maxsize);
 
 int queue_put(void *queue_obj, void *buffer, const void **msgs_data, const size_t *msg_sizes, size_t num_msgs, int block, float timeout);
 
diff --git a/faster_fifo.pyx b/faster_fifo.pyx
index abeba09..21285a7 100644
--- a/faster_fifo.pyx
+++ b/faster_fifo.pyx
@@ -60,8 +60,9 @@ cdef size_t bytes_to_ptr(b):
 
 
 class Queue:
-    def __init__(self, max_size_bytes=DEFAULT_CIRCULAR_BUFFER_SIZE, loads=None, dumps=None):
+    def __init__(self, max_size_bytes=DEFAULT_CIRCULAR_BUFFER_SIZE, maxsize=int(1e9), loads=None, dumps=None):
         self.max_size_bytes = max_size_bytes
+        self.maxsize = maxsize  # default maxsize
         self.max_bytes_to_read = self.max_size_bytes  # by default, read the whole queue if necessary
 
         # allow per-instance serializer overriding
@@ -76,7 +77,7 @@ class Queue:
         self.queue_obj_buffer = multiprocessing.RawArray(ctypes.c_ubyte, queue_obj_size)
         self.shared_memory = multiprocessing.RawArray(ctypes.c_ubyte, max_size_bytes)
 
-        Q.create_queue(<void *> q_addr(self), max_size_bytes)
+        Q.create_queue(<void *> q_addr(self), max_size_bytes, maxsize)
 
         self.message_buffer: TLSBuffer = TLSBuffer(None)
 
@@ -150,13 +151,23 @@ class Queue:
             raise Exception(f'Unexpected queue error {status}')
 
     def put(self, x, block=True, timeout=DEFAULT_TIMEOUT):
-        return self.put_many([x], block, timeout)
+        status = self.put_many([x], block, timeout)
+        if status == Q.Q_FULL:
+            raise Full()
+        return status
 
     def put_many_nowait(self, xs):
-        return self.put_many(xs, block=False)
+        status = self.put_many(xs, block=False)
+        if status == Q.Q_FULL:
+            raise Full()
+        return status
 
     def put_nowait(self, x):
-        return self.put_many_nowait([x])
+        status = self.put_many_nowait([x])
+        if status == Q.Q_FULL:
+            raise Full()
+        return status
+
 
     def get_many(self, block=True, timeout=DEFAULT_TIMEOUT, max_messages_to_get=int(1e9)):
         if self.message_buffer.val is None:
diff --git a/faster_fifo_def.pxd b/faster_fifo_def.pxd
index c3fc938..1b92a01 100644
--- a/faster_fifo_def.pxd
+++ b/faster_fifo_def.pxd
@@ -5,7 +5,7 @@ cdef extern from 'cpp_faster_fifo/cpp_lib/faster_fifo.hpp':
     int Q_SUCCESS = 0, Q_EMPTY = -1, Q_FULL = -2, Q_MSG_BUFFER_TOO_SMALL = -3;
 
     size_t queue_object_size();
-    void create_queue(void *queue_obj_memory, size_t max_size_bytes);
+    void create_queue(void *queue_obj_memory, size_t max_size_bytes, size_t maxsize);
 
     int queue_put(void *queue_obj, void *buffer, const void **msgs_data, const size_t *msg_sizes, size_t num_msgs, int block, float timeout) nogil;
     int queue_get(void *queue_obj, void *buffer,


Test patch:
diff --git a/cpp_faster_fifo/tests/test_faster_fifo.py b/cpp_faster_fifo/tests/test_faster_fifo.py
index 06e1189..3467533 100644
--- a/cpp_faster_fifo/tests/test_faster_fifo.py
+++ b/cpp_faster_fifo/tests/test_faster_fifo.py
@@ -13,7 +13,7 @@
 ch = logging.StreamHandler()
 ch.setLevel(logging.DEBUG)
 
-log = logging.getLogger('rl')
+log = logging.getLogger("rl")
 log.setLevel(logging.DEBUG)
 log.handlers = []  # No duplicated handlers
 log.propagate = False  # workaround for duplicated logs in ipython
@@ -38,12 +38,12 @@ def produce(q, p_idx, num_messages, make_msg_fn: Callable = make_msg):
         try:
             q.put(make_msg_fn(i), timeout=0.01)
             if i % 50000 == 0:
-                log.info('Produce: %d %d', i, p_idx)
+                log.info("Produce: %d %d", i, p_idx)
             i += 1
         except Full:
             # time.sleep(0.001)
             pass
-    log.info('Done! %d', p_idx)
+    log.info("Done! %d", p_idx)
 
 
 def consume(q, p_idx, consume_many, total_num_messages=int(1e9)):
@@ -54,13 +54,13 @@ def consume(q, p_idx, consume_many, total_num_messages=int(1e9)):
             for msg in msgs:
                 messages_received += 1
                 if msg[0] % 50000 == 0:
-                    log.info('Consume: %r %d num_msgs: %d', msg, p_idx, len(msgs))
+                    log.info("Consume: %r %d num_msgs: %d", msg, p_idx, len(msgs))
             if messages_received >= total_num_messages:
                 break
         except Empty:
             if q.is_closed():
                 break
-    log.info('Done! %d', p_idx)
+    log.info("Done! %d", p_idx)
 
 
 class TestFastQueue(TestCase):
@@ -72,8 +72,12 @@ def test_singleproc(self):
         self.assertIsNone(q.last_error)
 
     def run_producer_consumer(
-            self, n_producers: int, n_consumers: int, n_msg: int, execution_medium: type,
-            make_msg_fn: Callable,
+        self,
+        n_producers: int,
+        n_consumers: int,
+        n_msg: int,
+        execution_medium: type,
+        make_msg_fn: Callable,
     ):
         q = Queue()
         consume_many = 1000
@@ -96,48 +100,67 @@ def run_producer_consumer(
             c.join()
 
         self.assertIsNone(q.last_error)
-        log.info('Exit...')
+        log.info("Exit...")
 
     def test_multiprocessing(self):
         self.run_producer_consumer(
-            20, 3, 100001, execution_medium=multiprocessing.Process, make_msg_fn=make_msg,
+            20,
+            3,
+            100001,
+            execution_medium=multiprocessing.Process,
+            make_msg_fn=make_msg,
         )
 
     def test_multithreading(self):
         self.run_producer_consumer(
-            20, 3, 100001, execution_medium=threading.Thread, make_msg_fn=make_msg,
+            20,
+            3,
+            100001,
+            execution_medium=threading.Thread,
+            make_msg_fn=make_msg,
         )
 
     def test_multiprocessing_big_msg(self):
         self.run_producer_consumer(
-            20, 3, 1001, execution_medium=multiprocessing.Process, make_msg_fn=make_big_msg,
+            20,
+            3,
+            1001,
+            execution_medium=multiprocessing.Process,
+            make_msg_fn=make_big_msg,
         )
 
     def test_multithreading_big_msg(self):
         self.run_producer_consumer(
-            20, 20, 101, execution_medium=threading.Thread, make_msg_fn=make_big_msg,
+            20,
+            20,
+            101,
+            execution_medium=threading.Thread,
+            make_msg_fn=make_big_msg,
         )
 
     def test_msg(self):
         q = Queue(max_size_bytes=1000)
 
-        py_obj = dict(a=42, b=33, c=(1, 2, 3), d=[1, 2, 3], e='123', f=b'kkk')
+        py_obj = dict(a=42, b=33, c=(1, 2, 3), d=[1, 2, 3], e="123", f=b"kkk")
         q.put_nowait(py_obj)
         res = q.get_nowait()
-        log.debug('got object %r', res)
+        log.debug("got object %r", res)
         self.assertEqual(py_obj, res)
 
     def test_msg_many(self):
         q = Queue(max_size_bytes=100000)
 
-        py_objs = [dict(a=42, b=33, c=(1, 2, 3), d=[1, 2, 3], e='123', f=b'kkk') for _ in range(5)]
+        py_objs = [
+            dict(a=42, b=33, c=(1, 2, 3), d=[1, 2, 3], e="123", f=b"kkk")
+            for _ in range(5)
+        ]
         q.put_many_nowait(py_objs)
         res = q.get_many_nowait()
 
         while not q.empty():
             res.extend(q.get_many_nowait())
 
-        log.debug('Got object %r', res)
+        log.debug("Got object %r", res)
         self.assertEqual(py_objs, res)
 
         q.put_nowait(py_objs)
@@ -151,7 +174,7 @@ def test_queue_size(self):
         q.put_nowait(py_obj_1)
         q.put_nowait(py_obj_2)
         q_size_bef = q.qsize()
-        log.debug('Queue size after put -  %d', q_size_bef)
+        log.debug("Queue size after put -  %d", q_size_bef)
         num_messages = 0
         want_to_read = 2
         while num_messages < want_to_read:
@@ -160,13 +183,13 @@ def test_queue_size(self):
             num_messages += len(msgs)
         self.assertEqual(type(q_size_bef), int)
         q_size_af = q.qsize()
-        log.debug('Queue size after get -  %d', q_size_af)
+        log.debug("Queue size after get -  %d", q_size_af)
         self.assertEqual(q_size_af, 0)
 
     def test_queue_empty(self):
         q = Queue(max_size_bytes=1000)
         self.assertTrue(q.empty())
-        py_obj = dict(a=42, b=33, c=(1, 2, 3), d=[1, 2, 3], e='123', f=b'kkk')
+        py_obj = dict(a=42, b=33, c=(1, 2, 3), d=[1, 2, 3], e="123", f=b"kkk")
         q.put_nowait(py_obj)
         q_empty = q.empty()
         self.assertFalse(q_empty)
@@ -176,9 +199,9 @@ def test_queue_data_size(self):
         py_obj = dict(a=10, b=20)
         q.put_nowait(py_obj)
         py_obj_size = q.data_size()
-        log.debug('Queue data size after put -  %d', py_obj_size)
+        log.debug("Queue data size after put -  %d", py_obj_size)
         q.put_nowait(py_obj)
-        self.assertTrue(q.data_size(), 2*py_obj_size)
+        self.assertTrue(q.data_size(), 2 * py_obj_size)
 
     def test_queue_full(self):
         q = Queue(max_size_bytes=60)
@@ -195,7 +218,7 @@ def test_queue_usage(self):
         q = Queue(1000 * 1000)  # specify the size of the circular buffer in the ctor
 
         # any pickle-able Python object can be added to the queue
-        py_obj = dict(a=42, b=33, c=(1, 2, 3), d=[1, 2, 3], e='123', f=b'kkk')
+        py_obj = dict(a=42, b=33, c=(1, 2, 3), d=[1, 2, 3], e="123", f=b"kkk")
         q.put(py_obj)
         assert q.qsize() == 1
 
@@ -207,7 +230,7 @@ def test_queue_usage(self):
             try:
                 q.put(py_obj, timeout=0.1)
             except Full:
-                log.debug('Queue is full!')
+                log.debug("Queue is full!")
 
         num_received = 0
         while num_received < 100:
@@ -221,9 +244,79 @@ def test_queue_usage(self):
 
         try:
             q.get(timeout=0.1)
-            assert True, 'This won\'t be called'
+            assert True, "This won't be called"
         except Empty:
-            log.debug('Queue is empty')
+            log.debug("Queue is empty")
+
+    def test_max_size(self):
+        q = Queue(
+            max_size_bytes=1000, maxsize=5
+        )  # Create a queue with a maximum of 5 messages
+
+        for i in range(5):
+            q.put_nowait(make_msg(i))  # Add 5 messages to the queue
+
+        q_size_bef = q.qsize()
+        log.debug("Queue size after put -  %d", q_size_bef)
+
+        self.assertTrue(q.full())  # Check that the queue is full
+
+        with self.assertRaises(
+            Full
+        ):  # Check that adding another message raises the Full exception
+            q.put_nowait(make_msg(5))
+        q.get_many()
+        self.assertFalse(q.full())  # Check that the queue is not full
+
+    def test_queue_full_msgs(self):
+        q = Queue(maxsize=5)
+        self.assertFalse(q.full())
+        py_obj = (1, 2)
+        for _ in range(5):
+            q.put_nowait(py_obj)
+        self.assertTrue(q.full())
+        with self.assertRaises(Full):
+            q.put_nowait(py_obj)
+
+    def test_producer_consumer_msgs(self):
+        self.run_producer_consumer_msgs(
+            1,
+            1,
+            10,
+            threading.Thread,
+            make_msg,
+        )
+
+    def run_producer_consumer_msgs(
+        self,
+        n_producers: int,
+        n_consumers: int,
+        n_msg: int,
+        execution_medium: type,
+        make_msg_fn: Callable,
+    ):
+        q = Queue(maxsize=n_msg * n_producers)
+        consume_many = 5
+        producers = []
+        consumers = []
+        for j in range(n_producers):
+            p = execution_medium(target=produce, args=(q, j, n_msg, make_msg_fn))
+            producers.append(p)
+        for j in range(n_consumers):
+            p = execution_medium(target=consume, args=(q, j, consume_many))
+            consumers.append(p)
+        for c in consumers:
+            c.start()
+        for p in producers:
+            p.start()
+        for p in producers:
+            p.join()
+        q.close()
+        for c in consumers:
+            c.join()
+
+        self.assertIsNone(q.last_error)
+        log.info("Exit...")
 
 
 def spawn_producer(data_q_):
@@ -240,17 +333,15 @@ def spawn_consumer(data_q_):
             print(data)
             i += 1
         except Empty:
-            print('Read', i, 'messages')
+            print("Read", i, "messages")
             break
 
 
 class TestSpawn(TestCase):
     def test_spawn_ctx(self):
-        ctx = multiprocessing.get_context('spawn')
+        ctx = multiprocessing.get_context("spawn")
         data_q = Queue(1000 * 1000)
-        procs = [
-            ctx.Process(target=spawn_producer, args=(data_q,)) for _ in range(2)
-        ]
+        procs = [ctx.Process(target=spawn_producer, args=(data_q,)) for _ in range(2)]
         procs.append(ctx.Process(target=spawn_consumer, args=(data_q,)))
 
         # add data to the queue and read some of it back to make sure all buffers are initialized before
@@ -269,16 +360,20 @@ def test_spawn_ctx(self):
 # this can actually be used instead of Pickle if we know that we need to support only specific data types
 # should be significantly faster
 def custom_int_deserializer(msg_bytes):
-    return int.from_bytes(msg_bytes, 'big')
+    return int.from_bytes(msg_bytes, "big")
 
 
 def custom_int_serializer(x):
-    return x.to_bytes(4, 'big')
+    return x.to_bytes(4, "big")
 
 
 class TestCustomSerializer(TestCase):
     def test_custom_loads_dumps(self):
-        q = Queue(max_size_bytes=100000, loads=custom_int_deserializer, dumps=custom_int_serializer)
+        q = Queue(
+            max_size_bytes=100000,
+            loads=custom_int_deserializer,
+            dumps=custom_int_serializer,
+        )
         for i in range(32767):
             q.put(i)
             deserialized_i = q.get()
@@ -295,7 +390,7 @@ def worker_test_subclass(_x: Queue, _y: Queue):
 
 class TestSubclass(TestCase):
     def test_subclass(self):
-        ctx = multiprocessing.get_context('spawn')
+        ctx = multiprocessing.get_context("spawn")
 
         q = Queue()
         q.put(1)
@@ -305,6 +400,9 @@ def test_subclass(self):
         qs.get()
 
         from multiprocessing.pool import Pool as MpPool
-        pool = MpPool(2, initializer=worker_test_subclass, initargs=(q, qs), context=ctx)
+
+        pool = MpPool(
+            2, initializer=worker_test_subclass, initargs=(q, qs), context=ctx
+        )
         pool.close()
         pool.join()