Fix threading zmq poller and proxy

- Fixed universal proxy to not get stuck with multiple backends
- Fixed threading pollers/executors (proxy side)
    - Driver option to switch green/no-green impl.
    - Swtiched to no-green in real-world proxy (green left for unit tests)
- Minor names fixes in serializer

Change-Id: Id6508101521d8914228c639ed58ecd29db0ef456

oslo_messaging/_cmd/zmq_receiver.py

@@ -14,9 +14,6 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import eventlet
-eventlet.monkey_patch()
-
 import contextlib
 import logging
 import sys
@@ -30,6 +27,9 @@ from oslo_messaging._executors import base  # FIXME(markmc)
 CONF = cfg.CONF
 CONF.register_opts(impl_zmq.zmq_opts)
 CONF.register_opts(base._pool_opts)
+# TODO(ozamiatin): Move this option assignment to an external config file
+# Use efficient zmq poller in real-world deployment
+CONF.rpc_zmq_native = True
 
 
 def main():

oslo_messaging/_drivers/base.py

@@ -75,8 +75,7 @@ class Listener(object):
 
     def cleanup(self):
         """Cleanup listener.
-        Close connection used by listener if any. For some listeners like
+        Close connection (socket) used by listener if any.
-        zmq there is no connection so no need to close connection.
         As this is listener specific method, overwrite it in to derived class
         if cleanup of listener required.
         """

oslo_messaging/_drivers/impl_zmq.py

@@ -45,9 +45,13 @@ zmq_opts = [
 
     cfg.BoolOpt('rpc_zmq_all_req_rep',
                 default=True,
-                deprecated_group='DEFAULT',
                 help='Use REQ/REP pattern for all methods CALL/CAST/FANOUT.'),
 
+    cfg.BoolOpt('rpc_zmq_native',
+                default=False,
+                help='Switches ZeroMQ eventlet/threading way of usage.'
+                     'Affects pollers, executors etc.'),
+
     # The following port is unassigned by IANA as of 2012-05-21
     cfg.IntOpt('rpc_zmq_port', default=9501,
                help='ZeroMQ receiver listening port.'),

oslo_messaging/_drivers/zmq_driver/broker/zmq_base_proxy.py

@@ -41,7 +41,8 @@ class BaseProxy(object):
         super(BaseProxy, self).__init__()
         self.conf = conf
         self.context = context
-        self.executor = zmq_async.get_executor(self.run)
+        self.executor = zmq_async.get_executor(
+            self.run, native_zmq=conf.rpc_zmq_native)
 
     @abc.abstractmethod
     def run(self):
@@ -132,9 +133,8 @@ class DirectBackendMatcher(BaseBackendMatcher):
     def _match_backend(self, message):
         topic = self._get_topic(message)
         ipc_address = self._get_ipc_address(topic)
-        if ipc_address not in self.backends:
+        backend = self._create_backend(ipc_address)
-            self._create_backend(ipc_address)
+        return backend, topic
-        return self.backend, topic
 
     @abc.abstractmethod
     def _get_topic(self, message):

oslo_messaging/_drivers/zmq_driver/broker/zmq_broker.py

@@ -24,8 +24,6 @@ from oslo_messaging._i18n import _LE, _LI
 
 LOG = logging.getLogger(__name__)
 
-zmq = zmq_async.import_zmq()
-
 
 class ZmqBroker(object):
     """Local messaging IPC broker (nodes are still peers).
@@ -42,6 +40,7 @@ class ZmqBroker(object):
 
     def __init__(self, conf):
         super(ZmqBroker, self).__init__()
+        zmq = zmq_async.import_zmq(native_zmq=conf.rpc_zmq_native)
         self.conf = conf
         self.context = zmq.Context()
         proxy = zmq_universal_proxy.UniversalProxy(conf, self.context)

oslo_messaging/_drivers/zmq_driver/broker/zmq_call_proxy.py

@@ -46,12 +46,11 @@ class CallProxy(base_proxy.BaseProxy):
 
 class DealerBackend(base_proxy.DirectBackendMatcher):
 
-    def __init__(self, conf, context):
+    def __init__(self, conf, context, poller=None):
-        super(DealerBackend, self).__init__(conf,
+        if poller is None:
-                                            zmq_async.get_poller(),
+            poller = zmq_async.get_poller(
-                                            context)
+                native_zmq=conf.rpc_zmq_native)
-        self.backend = self.context.socket(zmq.DEALER)
+        super(DealerBackend, self).__init__(conf, poller, context)
-        self.poller.register(self.backend)
 
     def receive_outgoing_reply(self):
         reply_message = self.poller.poll(1)
@@ -71,16 +70,22 @@ class DealerBackend(base_proxy.DirectBackendMatcher):
         backend.send_multipart(message)
 
     def _create_backend(self, ipc_address):
-        self.backend.connect(ipc_address)
+        if ipc_address in self.backends:
-        self.backends[str(ipc_address)] = True
+            return self.backends[ipc_address]
+        backend = self.context.socket(zmq.DEALER)
+        backend.connect(ipc_address)
+        self.poller.register(backend)
+        self.backends[ipc_address] = backend
+        return backend
 
 
 class FrontendTcpRouter(base_proxy.BaseTcpFrontend):
 
-    def __init__(self, conf, context):
+    def __init__(self, conf, context, poller=None):
-        super(FrontendTcpRouter, self).__init__(conf,
+        if poller is None:
-                                                zmq_async.get_poller(),
+            poller = zmq_async.get_poller(
-                                                context,
+                native_zmq=conf.rpc_zmq_native)
+        super(FrontendTcpRouter, self).__init__(conf, poller, context,
                                                 socket_type=zmq.ROUTER,
                                                 port_number=conf.rpc_zmq_port)
 

oslo_messaging/_drivers/zmq_driver/broker/zmq_cast_proxy.py

@@ -42,8 +42,8 @@ class CastProxy(base_proxy.BaseProxy):
 class FrontendTcpPull(base_proxy.BaseTcpFrontend):
 
     def __init__(self, conf, context):
-        super(FrontendTcpPull, self).__init__(conf, zmq_async.get_poller(),
+        poller = zmq_async.get_poller(native_zmq=conf.rpc_zmq_native)
-                                              context)
+        super(FrontendTcpPull, self).__init__(conf, poller, context)
         self.frontend = self.context.socket(zmq.PULL)
         address = zmq_topic.get_tcp_bind_address(conf.rpc_zmq_fanout_port)
         LOG.info(_LI("Binding to TCP PULL %s") % address)
@@ -58,9 +58,8 @@ class FrontendTcpPull(base_proxy.BaseTcpFrontend):
 class CastPushBackendMatcher(base_proxy.BaseBackendMatcher):
 
     def __init__(self, conf, context):
-        super(CastPushBackendMatcher, self).__init__(conf,
+        poller = zmq_async.get_poller(native_zmq=conf.rpc_zmq_native)
-                                                     zmq_async.get_poller(),
+        super(CastPushBackendMatcher, self).__init__(conf, poller, context)
-                                                     context)
         self.backend = self.context.socket(zmq.PUSH)
 
     def _get_topic(self, message):

oslo_messaging/_drivers/zmq_driver/broker/zmq_fanout_proxy.py

@@ -24,9 +24,8 @@ zmq = zmq_async.import_zmq()
 class PublisherBackend(base_proxy.BaseBackendMatcher):
 
     def __init__(self, conf, context):
-        super(PublisherBackend, self).__init__(conf,
+        poller = zmq_async.get_poller(native_zmq=conf.rpc_zmq_native)
-                                               zmq_async.get_poller(),
+        super(PublisherBackend, self).__init__(conf, poller, context)
-                                               context)
         self.backend = self.context.socket(zmq.PUB)
         self.backend.bind(zmq_topic.get_ipc_address_fanout(conf))
 

oslo_messaging/_drivers/zmq_driver/broker/zmq_universal_proxy.py

@@ -17,6 +17,7 @@ import logging
 import oslo_messaging._drivers.zmq_driver.broker.zmq_base_proxy as base_proxy
 from oslo_messaging._drivers.zmq_driver.broker import zmq_call_proxy
 from oslo_messaging._drivers.zmq_driver.broker import zmq_fanout_proxy
+from oslo_messaging._drivers.zmq_driver import zmq_async
 from oslo_messaging._drivers.zmq_driver import zmq_serializer
 from oslo_messaging._i18n import _LI
 
@@ -27,27 +28,34 @@ class UniversalProxy(base_proxy.BaseProxy):
 
     def __init__(self, conf, context):
         super(UniversalProxy, self).__init__(conf, context)
-        self.tcp_frontend = zmq_call_proxy.FrontendTcpRouter(conf, context)
+        self.poller = zmq_async.get_poller(
-        self.backend_matcher = BackendMatcher(conf, context)
+            native_zmq=conf.rpc_zmq_native)
+        self.tcp_frontend = zmq_call_proxy.FrontendTcpRouter(
+            conf, context, poller=self.poller)
+        self.backend_matcher = BackendMatcher(
+            conf, context, poller=self.poller)
         call = zmq_serializer.CALL_TYPE
         self.call_backend = self.backend_matcher.backends[call]
         LOG.info(_LI("Starting universal-proxy thread"))
 
     def run(self):
-        message = self.tcp_frontend.receive_incoming()
+        message, socket = self.poller.poll(self.conf.rpc_poll_timeout)
-        if message is not None:
+        if message is None:
-            self.backend_matcher.redirect_to_backend(message)
+            return
 
-        reply, socket = self.call_backend.receive_outgoing_reply()
+        LOG.info(_LI("Received message at universal proxy: %s") % str(message))
-        if reply is not None:
+
-            self.tcp_frontend.redirect_outgoing_reply(reply)
+        if socket == self.tcp_frontend.frontend:
+            self.backend_matcher.redirect_to_backend(message)
+        else:
+            self.tcp_frontend.redirect_outgoing_reply(message)
 
 
 class BackendMatcher(base_proxy.BaseBackendMatcher):
 
-    def __init__(self, conf, context):
+    def __init__(self, conf, context, poller=None):
-        super(BackendMatcher, self).__init__(conf, None, context)
+        super(BackendMatcher, self).__init__(conf, poller, context)
-        direct_backend = zmq_call_proxy.DealerBackend(conf, context)
+        direct_backend = zmq_call_proxy.DealerBackend(conf, context, poller)
         self.backends[zmq_serializer.CALL_TYPE] = direct_backend
         self.backends[zmq_serializer.CAST_TYPE] = direct_backend
         fanout_backend = zmq_fanout_proxy.PublisherBackend(conf, context)

oslo_messaging/_drivers/zmq_driver/poller/threading_poller.py

@@ -15,36 +15,60 @@
 import logging
 import threading
 
+from oslo_utils import eventletutils
 import zmq
 
 from oslo_messaging._drivers.zmq_driver import zmq_poller
 
 LOG = logging.getLogger(__name__)
 
+_threading = threading
+
+if eventletutils.EVENTLET_AVAILABLE:
+    import eventlet
+    _threading = eventlet.patcher.original('threading')
+
 
 class ThreadingPoller(zmq_poller.ZmqPoller):
 
     def __init__(self):
         self.poller = zmq.Poller()
+        self.recv_methods = {}
 
-    def register(self, socket):
+    def register(self, socket, recv_method=None):
-        self.poller.register(socket, zmq.POLLOUT)
+        if recv_method is not None:
+            self.recv_methods[socket] = recv_method
+        self.poller.register(socket, zmq.POLLIN)
 
     def poll(self, timeout=None):
-        socks = dict(self.poller.poll(timeout))
+        timeout = timeout * 1000  # zmq poller waits milliseconds
-        for socket in socks:
+        sockets = dict(self.poller.poll(timeout=timeout))
-            incoming = socket.recv()
+        if not sockets:
-            return incoming
+            return None, None
+        for socket in sockets:
+            if socket in self.recv_methods:
+                return self.recv_methods[socket](socket)
+            else:
+                return socket.recv_multipart(), socket
 
 
 class ThreadingExecutor(zmq_poller.Executor):
 
     def __init__(self, method):
-        thread = threading.Thread(target=method)
+        self._method = method
-        super(ThreadingExecutor, self).__init__(thread)
+        super(ThreadingExecutor, self).__init__(
+            _threading.Thread(target=self._loop))
+        self._stop = _threading.Event()
+
+    def _loop(self):
+        while not self._stop.is_set():
+            self._method()
 
     def execute(self):
         self.thread.start()
 
+    def stop(self):
+        self._stop.set()
+
     def wait(self):
         self.thread.join()

oslo_messaging/_drivers/zmq_driver/rpc/client/zmq_call_request.py

@@ -60,8 +60,9 @@ class CallRequest(Request):
             raise oslo_messaging.MessagingTimeout(
                 "Timeout %s seconds was reached" % self.timeout)
 
-        if reply['failure']:
+        if reply[zmq_serializer.FIELD_FAILURE]:
             raise rpc_common.deserialize_remote_exception(
-                reply['failure'], self.allowed_remote_exmods)
+                reply[zmq_serializer.FIELD_FAILURE],
+                self.allowed_remote_exmods)
         else:
-            return reply['reply']
+            return reply[zmq_serializer.FIELD_REPLY]

oslo_messaging/_drivers/zmq_driver/rpc/server/zmq_call_responder.py

@@ -19,6 +19,7 @@ from oslo_messaging._drivers import base
 from oslo_messaging._drivers import common as rpc_common
 from oslo_messaging._drivers.zmq_driver.rpc.server import zmq_base_consumer
 from oslo_messaging._drivers.zmq_driver import zmq_async
+from oslo_messaging._drivers.zmq_driver import zmq_serializer
 from oslo_messaging._drivers.zmq_driver import zmq_topic as topic_utils
 from oslo_messaging._i18n import _LE
 
@@ -41,9 +42,9 @@ class ZmqIncomingRequest(base.IncomingMessage):
         if failure is not None:
             failure = rpc_common.serialize_remote_exception(failure,
                                                             log_failure)
-        message_reply = {u'reply': reply,
+        message_reply = {zmq_serializer.FIELD_REPLY: reply,
-                         u'failure': failure,
+                         zmq_serializer.FIELD_FAILURE: failure,
-                         u'log_failure': log_failure}
+                         zmq_serializer.FIELD_LOG_FAILURE: log_failure}
         LOG.debug("Replying %s REP", (str(message_reply)))
         self.received = True
         self.reply_socket.send(self.reply_id, zmq.SNDMORE)

oslo_messaging/_drivers/zmq_driver/zmq_async.py

@@ -23,8 +23,12 @@ LOG = logging.getLogger(__name__)
 green_zmq = importutils.try_import('eventlet.green.zmq')
 
 
-def import_zmq():
+def import_zmq(native_zmq=False):
-    imported_zmq = green_zmq or importutils.try_import('zmq')
+    if native_zmq:
+        imported_zmq = importutils.try_import('zmq')
+    else:
+        imported_zmq = green_zmq or importutils.try_import('zmq')
+
     if imported_zmq is None:
         errmsg = _LE("ZeroMQ not found!")
         LOG.error(errmsg)
@@ -32,28 +36,28 @@ def import_zmq():
     return imported_zmq
 
 
-def get_poller():
+def get_poller(native_zmq=False):
-    if green_zmq:
+    if native_zmq or green_zmq is None:
+        from oslo_messaging._drivers.zmq_driver.poller import threading_poller
+        return threading_poller.ThreadingPoller()
+    else:
         from oslo_messaging._drivers.zmq_driver.poller import green_poller
         return green_poller.GreenPoller()
-    else:
+
+
+def get_reply_poller(native_zmq=False):
+    if native_zmq or green_zmq is None:
         from oslo_messaging._drivers.zmq_driver.poller import threading_poller
         return threading_poller.ThreadingPoller()
-
+    else:
-
-def get_reply_poller():
-    if green_zmq:
         from oslo_messaging._drivers.zmq_driver.poller import green_poller
         return green_poller.HoldReplyPoller()
-    else:
+
+
+def get_executor(method, native_zmq=False):
+    if native_zmq or green_zmq is None:
         from oslo_messaging._drivers.zmq_driver.poller import threading_poller
-        return threading_poller.ThreadingPoller()
+        return threading_poller.ThreadingExecutor(method)
-
+    else:
-
-def get_executor(method):
-    if green_zmq is not None:
         from oslo_messaging._drivers.zmq_driver.poller import green_poller
         return green_poller.GreenExecutor(method)
-    else:
-        from oslo_messaging._drivers.zmq_driver.poller import threading_poller
-        return threading_poller.ThreadingExecutor()

oslo_messaging/_drivers/zmq_driver/zmq_serializer.py

@@ -26,6 +26,10 @@ LOG = logging.getLogger(__name__)
 MESSAGE_CALL_TYPE_POSITION = 2
 MESSAGE_CALL_TOPIC_POSITION = 3
 
+FIELD_FAILURE = 'failure'
+FIELD_REPLY = 'reply'
+FIELD_LOG_FAILURE = 'log_failure'
+
 CALL_TYPE = 'call'
 CAST_TYPE = 'cast'
 FANOUT_TYPE = 'fanout'

oslo_messaging/tests/drivers/test_impl_zmq.py

@@ -150,7 +150,7 @@ class TestZmqBasics(ZmqBaseTestCase):
             target, {},
             {'method': 'hello-world', 'tx_id': 1},
             wait_for_reply=True)
-        self.assertIsNotNone(result)
+        self.assertTrue(result)
 
     def test_send_noreply(self):
         """Cast() with topic."""

setup-test-env-zmq.sh

@@ -24,7 +24,4 @@ redis-server --port $ZMQ_REDIS_PORT &
 
 oslo-messaging-zmq-receiver --config-file ${DATADIR}/zmq.conf > ${DATADIR}/receiver.log 2>&1 &
 
-# FIXME(sileht): This does the same kind of setup that devstack does
-# But this doesn't work yet, a zeromq maintener should take a look on that
-
 $*