openstack/oslo.messaging
Code Issues Proposed changes

Always use a poll timeout in the executor

Browse Source 
This change allows to stop the eventlet executor without
using eventlet.kill.

And also, the kombu docs actually recommend that drain_events
by default have a 1 timeout.

Co-Authored-by: Joshua Harlow <harlowja@yahoo-inc.com>

Change-Id: I3a3919e38d08267439843a127346300fd2131540
This commit is contained in:
Mehdi Abaakouk 2014-11-27 11:13:50 +01:00
parent b631f46c7f
commit 7bce31a2d1
3 changed files with 25 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
oslo/messaging/_executors/base.py
View File

@ -16,6 +16,10 @@ import abc
import six 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) @six.add_metaclass(abc.ABCMeta)
class ExecutorBase(object): class ExecutorBase(object):

17
oslo/messaging/_executors/impl_blocking.py
View File

@ -13,7 +13,12 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import logging
from oslo.messaging._executors import base from oslo.messaging._executors import base
from oslo.messaging._i18n import _
LOG = logging.getLogger(__name__)
class BlockingExecutor(base.ExecutorBase): class BlockingExecutor(base.ExecutorBase):
@ -36,11 +41,13 @@ class BlockingExecutor(base.ExecutorBase):
def start(self): def start(self):
self._running = True self._running = True
while self._running: while self._running:
message = self.listener.poll(timeout=0.01) try:
if not message: incoming = self.listener.poll(timeout=base.POLL_TIMEOUT)
continue if incoming is not None:
with self.dispatcher(message) as callback: with self.dispatcher(incoming) as callback:
callback() callback()
except Exception:
LOG.exception(_("Unexpected exception occurred."))
def stop(self): def stop(self):
self._running = False self._running = False

14
oslo/messaging/_executors/impl_eventlet.py
View File

@ -75,6 +75,7 @@ class EventletExecutor(base.ExecutorBase):
self.conf.register_opts(_eventlet_opts) self.conf.register_opts(_eventlet_opts)
self._thread = None self._thread = None
self._greenpool = greenpool.GreenPool(self.conf.rpc_thread_pool_size) self._greenpool = greenpool.GreenPool(self.conf.rpc_thread_pool_size)
self._running = False
def start(self): def start(self):
if self._thread is not None: if self._thread is not None:
@ -83,19 +84,22 @@ class EventletExecutor(base.ExecutorBase):
@excutils.forever_retry_uncaught_exceptions @excutils.forever_retry_uncaught_exceptions
def _executor_thread(): def _executor_thread():
try: try:
while True: while self._running:
incoming = self.listener.poll() incoming = self.listener.poll(timeout=base.POLL_TIMEOUT)
spawn_with(ctxt=self.dispatcher(incoming), if incoming is not None:
pool=self._greenpool) spawn_with(ctxt=self.dispatcher(incoming),
pool=self._greenpool)
except greenlet.GreenletExit: except greenlet.GreenletExit:
return return
self._running = True
self._thread = eventlet.spawn(_executor_thread) self._thread = eventlet.spawn(_executor_thread)
def stop(self): def stop(self):
if self._thread is None: if self._thread is None:
return return
self._thread.kill() self._running = False
self._thread.cancel()
def wait(self): def wait(self):
if self._thread is None: if self._thread is None: