Merge "The executor doesn't need to set the timeout"

oslo/messaging/_drivers/amqpdriver.py

@@ -17,7 +17,6 @@ __all__ = ['AMQPDriverBase']
 
 import logging
 import threading
-import time
 import uuid
 
 from six import moves
@@ -94,6 +93,7 @@ class AMQPListener(base.Listener):
         self.conn = conn
         self.msg_id_cache = rpc_amqp._MsgIdCache()
         self.incoming = []
+        self._stopped = threading.Event()
 
     def __call__(self, message):
         # FIXME(markmc): logging isn't driver specific
@@ -110,23 +110,17 @@ class AMQPListener(base.Listener):
                                                  ctxt.reply_q))
 
     def poll(self, timeout=None):
-        if timeout is not None:
+        while not self._stopped.is_set():
-            deadline = time.time() + timeout
-        else:
-            deadline = None
-        while True:
             if self.incoming:
                 return self.incoming.pop(0)
-            if deadline is not None:
+            try:
-                timeout = deadline - time.time()
+                self.conn.consume(limit=1, timeout=timeout)
-                if timeout < 0:
+            except rpc_common.Timeout:
-                    return None
+                return None
-                try:
+
-                    self.conn.consume(limit=1, timeout=timeout)
+    def stop(self):
-                except rpc_common.Timeout:
+        self._stopped.set()
-                    return None
+        self.conn.stop_consuming()
-            else:
-                self.conn.consume(limit=1)
 
     def cleanup(self):
         # Closes listener connection

oslo/messaging/_drivers/base.py

@@ -56,9 +56,15 @@ class Listener(object):
     def poll(self, timeout=None):
         """Blocking until a message is pending and return IncomingMessage.
         Return None after timeout seconds if timeout is set and no message is
-        ending.
+        ending or if the listener have been stopped.
         """
 
+    def stop(self):
+        """Stop listener.
+        Stop the listener message polling
+        """
+        pass
+
     def cleanup(self):
         """Cleanup listener.
         Close connection used by listener if any. For some listeners like

oslo/messaging/_drivers/impl_fake.py

@@ -46,6 +46,7 @@ class FakeListener(base.Listener):
         self._exchange_manager = exchange_manager
         self._targets = targets
         self._pool = pool
+        self._stopped = threading.Event()
 
         # NOTE(sileht): Ensure that all needed queues exists even the listener
         # have not been polled yet
@@ -58,7 +59,7 @@ class FakeListener(base.Listener):
             deadline = time.time() + timeout
         else:
             deadline = None
-        while True:
+        while not self._stopped.is_set():
             for target in self._targets:
                 exchange = self._exchange_manager.get_exchange(target.exchange)
                 (ctxt, message, reply_q, requeue) = exchange.poll(target,
@@ -77,6 +78,9 @@ class FakeListener(base.Listener):
             time.sleep(pause)
         return None
 
+    def stop(self):
+        self._stopped.set()
+
 
 class FakeExchange(object):
 

oslo/messaging/_drivers/impl_qpid.py

@@ -460,6 +460,8 @@ class Connection(object):
         self.consumers = {}
         self.conf = conf
 
+        self._consume_loop_stopped = False
+
         self.brokers_params = []
         if url.hosts:
             for host in url.hosts:
@@ -651,8 +653,16 @@ class Connection(object):
             LOG.exception(_('Failed to consume message from queue: %s'), exc)
 
         def _consume():
+            # NOTE(sileht):
+            # maximun value choosen according the best practice from kombu:
+            # http://kombu.readthedocs.org/en/latest/reference/kombu.common.html#kombu.common.eventloop
             poll_timeout = 1 if timeout is None else min(timeout, 1)
+
             while True:
+                if self._consume_loop_stopped:
+                    self._consume_loop_stopped = False
+                    raise StopIteration
+
                 try:
                     nxt_receiver = self.session.next_receiver(
                         timeout=poll_timeout)
@@ -745,6 +755,9 @@ class Connection(object):
             except StopIteration:
                 return
 
+    def stop_consuming(self):
+        self._consume_loop_stopped = True
+
 
 class QpidDriver(amqpdriver.AMQPDriverBase):
 

oslo/messaging/_drivers/impl_rabbit.py

@@ -497,6 +497,7 @@ class Connection(object):
         self._initial_pid = os.getpid()
 
         self.do_consume = True
+        self._consume_loop_stopped = False
 
         self.channel = None
         self.connection = kombu.connection.Connection(
@@ -715,8 +716,15 @@ class Connection(object):
                 queues_tail.consume(nowait=False)
                 self.do_consume = False
 
+            # NOTE(sileht):
+            # maximun value choosen according the best practice from kombu:
+            # http://kombu.readthedocs.org/en/latest/reference/kombu.common.html#kombu.common.eventloop
             poll_timeout = 1 if timeout is None else min(timeout, 1)
             while True:
+                if self._consume_loop_stopped:
+                    self._consume_loop_stopped = False
+                    raise StopIteration
+
                 try:
                     return self.connection.drain_events(timeout=poll_timeout)
                 except socket.timeout as exc:
@@ -790,6 +798,9 @@ class Connection(object):
             except StopIteration:
                 return
 
+    def stop_consuming(self):
+        self._consume_loop_stopped = True
+
 
 class RabbitDriver(amqpdriver.AMQPDriverBase):
 

oslo/messaging/_executors/base.py

@@ -16,10 +16,6 @@ import abc
 
 import six
 
-# NOTE(sileht): value choosen according the best practice from kombu
-# http://kombu.readthedocs.org/en/latest/reference/kombu.common.html#kombu.common.eventloop
-POLL_TIMEOUT = 1
-
 
 @six.add_metaclass(abc.ABCMeta)
 class ExecutorBase(object):

oslo/messaging/_executors/impl_blocking.py

@@ -42,7 +42,7 @@ class BlockingExecutor(base.ExecutorBase):
         self._running = True
         while self._running:
             try:
-                incoming = self.listener.poll(timeout=base.POLL_TIMEOUT)
+                incoming = self.listener.poll()
                 if incoming is not None:
                     with self.dispatcher(incoming) as callback:
                         callback()
@@ -51,6 +51,7 @@ class BlockingExecutor(base.ExecutorBase):
 
     def stop(self):
         self._running = False
+        self.listener.stop()
 
     def wait(self):
         pass

oslo/messaging/_executors/impl_eventlet.py

@@ -85,7 +85,7 @@ class EventletExecutor(base.ExecutorBase):
         def _executor_thread():
             try:
                 while self._running:
-                    incoming = self.listener.poll(timeout=base.POLL_TIMEOUT)
+                    incoming = self.listener.poll()
                     if incoming is not None:
                         spawn_with(ctxt=self.dispatcher(incoming),
                                    pool=self._greenpool)
@@ -99,6 +99,7 @@ class EventletExecutor(base.ExecutorBase):
         if self._thread is None:
             return
         self._running = False
+        self.listener.stop()
         self._thread.cancel()
 
     def wait(self):

tests/executors/test_executor.py

@@ -39,12 +39,10 @@ class TestExecutor(test_utils.BaseTestCase):
 
     @classmethod
     def generate_scenarios(cls):
-        impl = [('blocking', dict(executor=impl_blocking.BlockingExecutor,
+        impl = [('blocking', dict(executor=impl_blocking.BlockingExecutor))]
-                                  stop_before_return=True))]
         if impl_eventlet is not None:
             impl.append(
-                ('eventlet', dict(executor=impl_eventlet.EventletExecutor,
+                ('eventlet', dict(executor=impl_eventlet.EventletExecutor)))
-                                  stop_before_return=False)))
         cls.scenarios = testscenarios.multiply_scenarios(impl)
 
     @staticmethod
@@ -72,13 +70,9 @@ class TestExecutor(test_utils.BaseTestCase):
                                           message={'payload': 'data'})
 
         def fake_poll(timeout=None):
-            if self.stop_before_return:
+            if listener.poll.call_count == 1:
-                executor.stop()
                 return incoming_message
-            else:
+            executor.stop()
-                if listener.poll.call_count == 1:
-                    return incoming_message
-                executor.stop()
 
         listener.poll.side_effect = fake_poll