Implement empty-queue timeout to avoid taxing the CPU.

Author: panta

fluent/asyncsender.py

@@ -13,6 +13,8 @@
 
 _global_sender = None
 
+DEFAULT_QUEUE_TIMEOUT = 0.05
+
 
 def _set_global_sender(sender):
     """ [For testing] Function to set global sender directly
@@ -29,6 +31,7 @@ def setup(tag, **kwargs):
 def get_global_sender():
     return _global_sender
 
+
 def close():
     get_global_sender().close()
 
@@ -42,10 +45,12 @@ def __init__(self, tag,
                  verbose=False,
                  buffer_overflow_handler=None,
                  nanosecond_precision=False,
-                 msgpack_kwargs=None, *args, **kwargs):
+                 msgpack_kwargs=None,
+                 queue_timeout=DEFAULT_QUEUE_TIMEOUT, *args, **kwargs):
         super(CommunicatorThread, self).__init__(**kwargs)
         self._queue = Queue()
         self._do_run = True
+        self._queue_timeout = queue_timeout
         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,
@@ -61,7 +66,7 @@ def send(self, bytes_):
     def run(self):
         while self._do_run:
             try:
-                bytes_ = self._queue.get(block=False)
+                bytes_ = self._queue.get(block=True, timeout=self._queue_timeout)
             except Empty:
                 continue
             self._conn_close_lock.acquire()
@@ -101,6 +106,14 @@ def last_error(self, err):
     def clear_last_error(self, _thread_id = None):
         self._sender.clear_last_error(_thread_id=_thread_id)
 
+    @property
+    def queue_timeout(self):
+        return self._queue_timeout
+
+    @queue_timeout.setter
+    def queue_timeout(self, value):
+        self._queue_timeout = value
+
     def __enter__(self):
         return self
 
@@ -119,14 +132,16 @@ def __init__(self,
                  buffer_overflow_handler=None,
                  nanosecond_precision=False,
                  msgpack_kwargs=None,
+                 queue_timeout=DEFAULT_QUEUE_TIMEOUT,
                  **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,
                                            nanosecond_precision=nanosecond_precision, msgpack_kwargs=msgpack_kwargs,
                                            **kwargs)
         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)
+                                                nanosecond_precision=nanosecond_precision, msgpack_kwargs=msgpack_kwargs,
+                                                queue_timeout=queue_timeout)
         self._communicator.start()
 
     def _send(self, bytes_):
@@ -163,6 +178,14 @@ def last_error(self, err):
     def clear_last_error(self, _thread_id = None):
         self._communicator.clear_last_error(_thread_id=_thread_id)
 
+    @property
+    def queue_timeout(self):
+        return self._communicator.queue_timeout
+
+    @queue_timeout.setter
+    def queue_timeout(self, value):
+        self._communicator.queue_timeout = value
+
     def __enter__(self):
         return self
 

tests/test_asyncsender.py

@@ -145,6 +145,51 @@ def test_connect_exception_during_sender_init(self, mock_socket):
         self.assertEqual(self._sender.last_error.args[0], EXCEPTION_MSG)
 
 
+class TestSenderWithTimeout(unittest.TestCase):
+    def setUp(self):
+        super(TestSenderWithTimeout, self).setUp()
+        self._server = mockserver.MockRecvServer('localhost')
+        self._sender = fluent.asyncsender.FluentSender(tag='test',
+                                                       port=self._server.port,
+                                                       queue_timeout=0.04)
+
+    def tearDown(self):
+        self._sender.close()
+
+    def get_data(self):
+        return self._server.get_recieved()
+
+    def test_simple(self):
+        sender = self._sender
+        sender.emit('foo', {'bar': 'baz'})
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+    def test_simple_with_timeout_props(self):
+        sender = self._sender
+        sender.queue_timeout = 0.06
+        assert sender.queue_timeout == 0.06
+        sender.emit('foo', {'bar': 'baz'})
+        time.sleep(0.5)
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(data[0][1])
+        self.assertTrue(isinstance(data[0][1], int))
+
+
 class TestEventTime(unittest.TestCase):
     def test_event_time(self):
         time = fluent.asyncsender.EventTime(1490061367.8616468906402588)