x/deb-oslo.messaging
Code Issues Proposed changes

[zmq] Driver optimizations for CALL

Browse Source 
New DEALER-based publisher provided for CALL.
Futures-based reply waiting makes possible to
refuse using of REQ blocking socket and also
reduce number of openned sockets to a socket-per-target
instead of socket-per-message as it was for CALL.

Closes-Bug: #1517999

Optimized redis requests - request once instead of
per each message. This should be elaborated with an
autonomous nodes discovery mechanism to be correct
in general case.

Closes-Bug: #1517993

Reduced number of INFO log messages. Most of them switched
to the DEBUG level.

Closes-Bug: #1517997

Change-Id: Id017e79368cdc68613ddd7adef26411ea422dc8c
This commit is contained in:
Oleksii Zamiatin
2015-11-16 15:07:15 +02:00
parent e5a232ce03
commit eea60cfb36
20 changed files with 528 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
oslo_messaging/_drivers/impl_zmq.py
View File

@@ -13,6 +13,7 @@
# under the License.
import logging
import os
import pprint
import socket
import threading
@@ -23,6 +24,7 @@ from stevedore import driver
from oslo_messaging._drivers import base
from oslo_messaging._drivers import common as rpc_common
from oslo_messaging._drivers.zmq_driver.client import zmq_client
from oslo_messaging._drivers.zmq_driver.client import zmq_client_light
from oslo_messaging._drivers.zmq_driver.server import zmq_server
from oslo_messaging._executors import impl_pooledexecutor # FIXME(markmc)
@@ -78,8 +80,8 @@ zmq_opts = [
'Poll raises timeout exception when timeout expired.'),
cfg.BoolOpt('zmq_use_broker',
default=True,
help='Shows whether zmq-messaging uses broker or not.'),
default=False,
help='Configures zmq-messaging to use broker or not.'),
cfg.PortOpt('rpc_zmq_min_port',
default=49152,
@@ -106,6 +108,7 @@ class LazyDriverItem(object):
self.item_class = item_cls
self.args = args
self.kwargs = kwargs
self.process_id = os.getpid()
def get(self):
# NOTE(ozamiatin): Lazy initialization.
@@ -114,11 +117,12 @@ class LazyDriverItem(object):
# __init__, but 'fork' extensively used by services
# breaks all things.
if self.item is not None:
if self.item is not None and os.getpid() == self.process_id:
return self.item
self._lock.acquire()
if self.item is None:
if self.item is None or os.getpid() != self.process_id:
self.process_id = os.getpid()
self.item = self.item_class(*self.args, **self.kwargs)
self._lock.release()
return self.item
@@ -175,12 +179,15 @@ class ZmqDriver(base.BaseDriver):
self.notify_server = LazyDriverItem(
zmq_server.ZmqServer, self, self.conf, self.matchmaker)
client_cls = zmq_client_light.ZmqClientLight \
if conf.zmq_use_broker else zmq_client.ZmqClient
self.client = LazyDriverItem(
zmq_client.ZmqClient, self.conf, self.matchmaker,
client_cls, self.conf, self.matchmaker,
self.allowed_remote_exmods)
self.notifier = LazyDriverItem(
zmq_client.ZmqClient, self.conf, self.matchmaker,
client_cls, self.conf, self.matchmaker,
self.allowed_remote_exmods)
super(ZmqDriver, self).__init__(conf, url, default_exchange,