Add asyncsender options to get a circular queue and/or with a bounded size.

NOTE: by default the async sender queue is blocking and non-circular.

Enabling the `queue_circular` async sender option, makes the sender queue circular and non-blocking, useful in high throughput low-latency applications.

Please note that with circular mode enabled, back-pressure could cause the queue to fill up and start discarding events (the oldest not yet sent).

Author: panta

fluent/asyncsender.py

@@ -14,6 +14,8 @@
 _global_sender = None
 
 DEFAULT_QUEUE_TIMEOUT = 0.05
+DEFAULT_QUEUE_MAXSIZE = 100
+DEFAULT_QUEUE_CIRCULAR = False
 
 
 def _set_global_sender(sender):
@@ -46,19 +48,29 @@ def __init__(self, tag,
                  buffer_overflow_handler=None,
                  nanosecond_precision=False,
                  msgpack_kwargs=None,
-                 queue_timeout=DEFAULT_QUEUE_TIMEOUT, *args, **kwargs):
+                 queue_timeout=DEFAULT_QUEUE_TIMEOUT,
+                 queue_maxsize=DEFAULT_QUEUE_MAXSIZE,
+                 queue_circular=DEFAULT_QUEUE_CIRCULAR, *args, **kwargs):
         super(CommunicatorThread, self).__init__(**kwargs)
-        self._queue = Queue()
+        self._queue = Queue(maxsize=queue_maxsize)
         self._do_run = True
         self._queue_timeout = queue_timeout
+        self._queue_maxsize = queue_maxsize
+        self._queue_circular = queue_circular
         self._conn_close_lock = threading.Lock()
         self._sender = sender.FluentSender(tag=tag, host=host, port=port, bufmax=bufmax, timeout=timeout,
                                            verbose=verbose, buffer_overflow_handler=buffer_overflow_handler,
                                            nanosecond_precision=nanosecond_precision, msgpack_kwargs=msgpack_kwargs)
 
     def send(self, bytes_):
+        if self._queue_circular and self._queue.full():
+            # discard oldest
+            try:
+                self._queue.get(block=False)
+            except Empty:
+                pass
         try:
-            self._queue.put(bytes_)
+            self._queue.put(bytes_, block=(not self._queue_circular))
         except Full:
             return False
         return True
@@ -114,6 +126,18 @@ def queue_timeout(self):
     def queue_timeout(self, value):
         self._queue_timeout = value
 
+    @property
+    def queue_maxsize(self):
+        return self._queue_maxsize
+
+    @property
+    def queue_blocking(self):
+        return not self._queue_circular
+
+    @property
+    def queue_circular(self):
+        return self._queue_circular
+
     def __enter__(self):
         return self
 
@@ -133,6 +157,8 @@ def __init__(self,
                  nanosecond_precision=False,
                  msgpack_kwargs=None,
                  queue_timeout=DEFAULT_QUEUE_TIMEOUT,
+                 queue_maxsize=DEFAULT_QUEUE_MAXSIZE,
+                 queue_circular=DEFAULT_QUEUE_CIRCULAR,
                  **kwargs): # This kwargs argument is not used in __init__. This will be removed in the next major version.
         super(FluentSender, self).__init__(tag=tag, host=host, port=port, bufmax=bufmax, timeout=timeout,
                                            verbose=verbose, buffer_overflow_handler=buffer_overflow_handler,
@@ -141,7 +167,8 @@ def __init__(self,
         self._communicator = CommunicatorThread(tag=tag, host=host, port=port, bufmax=bufmax, timeout=timeout,
                                                 verbose=verbose, buffer_overflow_handler=buffer_overflow_handler,
                                                 nanosecond_precision=nanosecond_precision, msgpack_kwargs=msgpack_kwargs,
-                                                queue_timeout=queue_timeout)
+                                                queue_timeout=queue_timeout, queue_maxsize=queue_maxsize,
+                                                queue_circular=queue_circular)
         self._communicator.start()
 
     def _send(self, bytes_):
@@ -186,6 +213,18 @@ def queue_timeout(self):
     def queue_timeout(self, value):
         self._communicator.queue_timeout = value
 
+    @property
+    def queue_maxsize(self):
+        return self._communicator.queue_maxsize
+
+    @property
+    def queue_blocking(self):
+        return self._communicator.queue_blocking
+
+    @property
+    def queue_circular(self):
+        return self._communicator.queue_circular
+
     def __enter__(self):
         return self
 

tests/test_asyncsender.py

@@ -145,6 +145,27 @@ def test_connect_exception_during_sender_init(self, mock_socket):
         self.assertEqual(self._sender.last_error.args[0], EXCEPTION_MSG)
 
 
+class TestSenderDefaultProperties(unittest.TestCase):
+    def setUp(self):
+        super(TestSenderDefaultProperties, self).setUp()
+        self._server = mockserver.MockRecvServer('localhost')
+        self._sender = fluent.asyncsender.FluentSender(tag='test',
+                                                       port=self._server.port)
+
+    def tearDown(self):
+        self._sender.close()
+
+    def test_default_properties(self):
+        sender = self._sender
+        self.assertTrue(sender.queue_blocking)
+        self.assertFalse(sender.queue_circular)
+        self.assertTrue(isinstance(sender.queue_maxsize, int))
+        self.assertTrue(sender.queue_maxsize > 0)
+        self.assertTrue(isinstance(sender.queue_timeout, (int, float)))
+        self.assertTrue(sender.queue_timeout > 0)
+        sender._close()
+
+
 class TestSenderWithTimeout(unittest.TestCase):
     def setUp(self):
         super(TestSenderWithTimeout, self).setUp()
@@ -195,3 +216,151 @@ def test_event_time(self):
         time = fluent.asyncsender.EventTime(1490061367.8616468906402588)
         self.assertEqual(time.code, 0)
         self.assertEqual(time.data, b'X\xd0\x8873[\xb0*')
+
+
+class TestSenderWithTimeoutAndCircular(unittest.TestCase):
+    Q_SIZE = 3
+
+    def setUp(self):
+        super(TestSenderWithTimeoutAndCircular, self).setUp()
+        self._server = mockserver.MockRecvServer('localhost')
+        self._sender = fluent.asyncsender.FluentSender(tag='test',
+                                                       port=self._server.port,
+                                                       queue_timeout=0.04,
+                                                       queue_maxsize=self.Q_SIZE,
+                                                       queue_circular=True)
+
+    def tearDown(self):
+        self._sender.close()
+
+    def get_data(self):
+        return self._server.get_recieved()
+
+    def test_simple(self):
+        sender = self._sender
+
+        self.assertEqual(self._sender.queue_maxsize, self.Q_SIZE)
+        self.assertEqual(self._sender.queue_circular, True)
+        self.assertEqual(self._sender.queue_blocking, False)
+
+        ok = sender.emit('foo1', {'bar': 'baz1'})
+        self.assertTrue(ok)
+        ok = sender.emit('foo2', {'bar': 'baz2'})
+        self.assertTrue(ok)
+        ok = sender.emit('foo3', {'bar': 'baz3'})
+        self.assertTrue(ok)
+        ok = sender.emit('foo4', {'bar': 'baz4'})
+        self.assertTrue(ok)
+        ok = sender.emit('foo5', {'bar': 'baz5'})
+        self.assertTrue(ok)
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(self.Q_SIZE, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo3', data[0][0])
+        eq({'bar': 'baz3'}, data[0][2])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+        eq(3, len(data[2]))
+        eq('test.foo5', data[2][0])
+        eq({'bar': 'baz5'}, data[2][2])
+
+
+class TestSenderWithTimeoutMaxSizeNonCircular(unittest.TestCase):
+    Q_SIZE = 3
+
+    def setUp(self):
+        super(TestSenderWithTimeoutMaxSizeNonCircular, self).setUp()
+        self._server = mockserver.MockRecvServer('localhost')
+        self._sender = fluent.asyncsender.FluentSender(tag='test',
+                                                       port=self._server.port,
+                                                       queue_timeout=0.04,
+                                                       queue_maxsize=self.Q_SIZE)
+
+    def tearDown(self):
+        self._sender.close()
+
+    def get_data(self):
+        return self._server.get_recieved()
+
+    def test_simple(self):
+        sender = self._sender
+
+        self.assertEqual(self._sender.queue_maxsize, self.Q_SIZE)
+        self.assertEqual(self._sender.queue_blocking, True)
+        self.assertEqual(self._sender.queue_circular, False)
+
+        ok = sender.emit('foo1', {'bar': 'baz1'})
+        self.assertTrue(ok)
+        ok = sender.emit('foo2', {'bar': 'baz2'})
+        self.assertTrue(ok)
+        ok = sender.emit('foo3', {'bar': 'baz3'})
+        self.assertTrue(ok)
+        ok = sender.emit('foo4', {'bar': 'baz4'})
+        self.assertTrue(ok)
+        ok = sender.emit('foo5', {'bar': 'baz5'})
+        self.assertTrue(ok)
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        print(data)
+        eq(5, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo1', data[0][0])
+        eq({'bar': 'baz1'}, data[0][2])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+        eq(3, len(data[2]))
+        eq('test.foo3', data[2][0])
+        eq({'bar': 'baz3'}, data[2][2])
+
+
+class TestSenderUnlimitedSize(unittest.TestCase):
+    Q_SIZE = 3
+
+    def setUp(self):
+        super(TestSenderUnlimitedSize, self).setUp()
+        self._server = mockserver.MockRecvServer('localhost')
+        self._sender = fluent.asyncsender.FluentSender(tag='test',
+                                                       port=self._server.port,
+                                                       queue_timeout=0.04,
+                                                       queue_maxsize=0)
+
+    def tearDown(self):
+        self._sender.close()
+
+    def get_data(self):
+        return self._server.get_recieved()
+
+    def test_simple(self):
+        sender = self._sender
+
+        self.assertEqual(self._sender.queue_maxsize, 0)
+        self.assertEqual(self._sender.queue_blocking, True)
+        self.assertEqual(self._sender.queue_circular, False)
+
+        NUM = 1000
+        for i in range(1, NUM+1):
+            ok = sender.emit("foo{}".format(i), {'bar': "baz{}".format(i)})
+            self.assertTrue(ok)
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(NUM, len(data))
+        el = data[0]
+        eq(3, len(el))
+        eq('test.foo1', el[0])
+        eq({'bar': 'baz1'}, el[2])
+        self.assertTrue(el[1])
+        self.assertTrue(isinstance(el[1], int))
+
+        el = data[NUM-1]
+        eq(3, len(el))
+        eq("test.foo{}".format(NUM), el[0])
+        eq({'bar': "baz{}".format(NUM)}, el[2])