Python zmq 模块，Poller() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用zmq.Poller()。
def receive_message(self, event, event_data, listener_data):
"""
Receives a messages from another processes.
:param * event: Not used.
:param * event_data: Not used.
:param * listener_data: Not used.
"""
del event, event_data, listener_data
# Make a poller for all incoming sockets.
poller = zmq.Poller()
for socket in self.__end_points.values():
if socket.type in [zmq.PULL, zmq.REP]:
poller.register(socket, zmq.POLLIN)
# Wait for socket is ready for reading.
socks = dict(poller.poll())
for name, socket in self.__end_points.items():
if socket in socks:
self._receive_message(name, socket)
# ------------------------------------------------------------------------------------------------------------------
def no_barking(self, seconds):
"""
During start up of ZMQ the incoming file descriptors become 'ready for reading' while there is no message on
the socket. This method prevent incoming sockets barking that the are ready the for reading.
:param int seconds: The number of seconds the give the other ZMQ thread to start up.
"""
sleep(seconds)
for _ in range(1, len(self.end_points)):
poller = zmq.Poller()
for socket in self.end_points.values():
if socket.type in [zmq.PULL, zmq.REP]:
poller.register(socket, zmq.POLLIN)
poller.poll(1)
# ----------------------------------------------------------------------------------------------------------------------
def run(self):
""" Start the Authentication Agent thread task """
self.authenticator.start()
zap = self.authenticator.zap_socket
poller = zmq.Poller()
poller.register(self.pipe, zmq.POLLIN)
poller.register(zap, zmq.POLLIN)
while True:
try:
socks = dict(poller.poll())
except zmq.ZMQError:
break # interrupted
if self.pipe in socks and socks[self.pipe] == zmq.POLLIN:
terminate = self._handle_pipe()
if terminate:
break
if zap in socks and socks[zap] == zmq.POLLIN:
self._handle_zap()
self.pipe.close()
self.authenticator.stop()
def test_timeout(self):
"""make sure Poller.poll timeout has the right units (milliseconds)."""
s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
poller = self.Poller()
poller.register(s1, zmq.POLLIN)
tic = time.time()
evt = poller.poll(.005)
toc = time.time()
self.assertTrue(toc-tic < 0.1)
tic = time.time()
evt = poller.poll(5)
toc = time.time()
self.assertTrue(toc-tic < 0.1)
self.assertTrue(toc-tic > .001)
tic = time.time()
evt = poller.poll(500)
toc = time.time()
self.assertTrue(toc-tic < 1)
self.assertTrue(toc-tic > 0.1)
def __init__(self, socket, io_loop=None):
self.socket = socket
self.io_loop = io_loop or IOLoop.instance()
self.poller = zmq.Poller()
self._send_queue = Queue()
self._recv_callback = None
self._send_callback = None
self._close_callback = None
self._recv_copy = False
self._flushed = False
self._state = self.io_loop.ERROR
self._init_io_state()
# shortcircuit some socket methods
self.bind = self.socket.bind
self.bind_to_random_port = self.socket.bind_to_random_port
self.connect = self.socket.connect
self.setsockopt = self.socket.setsockopt
self.getsockopt = self.socket.getsockopt
self.setsockopt_string = self.socket.setsockopt_string
self.getsockopt_string = self.socket.getsockopt_string
self.setsockopt_unicode = self.socket.setsockopt_unicode
self.getsockopt_unicode = self.socket.getsockopt_unicode
def connect(self):
self.context = zmq.Context()
if not self.context:
raise RuntimeError('Failed to create ZMQ context!')
self.socket = self.context.socket(zmq.REQ)
if not self.socket:
raise RuntimeError('Failed to create ZMQ socket!')
self.socket.connect(self.endpoint)
self.poller = zmq.Poller()
self.poller.register(self.socket, zmq.POLLIN)
self.is_connected = True
def connect(self):
self.context = zmq.Context()
if not self.context:
raise RuntimeError('Failed to create ZMQ context!')
self.socket = self.context.socket(zmq.PULL)
if not self.socket:
raise RuntimeError('Failed to create ZMQ socket!')
self.socket.bind(self.endpoint)
self.poller = zmq.Poller()
self.poller.register(self.socket, zmq.POLLIN)
self.is_connected = True
def connect(self):
self.context = zmq.Context()
if not self.context:
raise RuntimeError('Failed to create ZMQ context!')
self.socket = self.context.socket(zmq.REP)
if not self.socket:
raise RuntimeError('Failed to create ZMQ socket!')
self.socket.bind(self.endpoint)
self.poller = zmq.Poller()
self.poller.register(self.socket, zmq.POLLIN)
self.is_connected = True
def run(self):
""" Contents of the infinite loop. """
# Create zmq sockets
sockets = SupvisorsZmq(self.supvisors)
# create poller
poller = zmq.Poller()
# register sockets
poller.register(sockets.internal_subscriber.socket, zmq.POLLIN)
poller.register(sockets.puller.socket, zmq.POLLIN)
# poll events forever
while not self.stopping():
socks = dict(poller.poll(500))
# test stop condition again: if Supervisor is stopping,
# any XML-RPC call would block this thread, and the other
# because of the join
if not self.stopping():
self.check_requests(sockets, socks)
self.check_events(sockets.internal_subscriber, socks)
# close resources gracefully
poller.unregister(sockets.puller.socket)
poller.unregister(sockets.internal_subscriber.socket)
sockets.close()
def support_test_send_to_multiple_addresses(self, address1, address2):
poller = zmq.Poller()
socket1 = self.context.socket(roles['listener'])
socket2 = self.context.socket(roles['listener'])
try:
socket1.bind("tcp://%s" % address1)
socket2.bind("tcp://%s" % address2)
poller.register(socket1, zmq.POLLIN)
poller.register(socket2, zmq.POLLIN)
polled = dict(poller.poll(2000))
if socket1 in polled:
socket1.recv()
socket1.send(nw0.sockets._serialise(address1))
elif socket2 in polled:
socket2.recv()
socket2.send(nw0.sockets._serialise(address2))
else:
raise RuntimeError("Nothing found")
finally:
socket1.close()
socket2.close()
def __init__(self, data_dir=bqueryd.DEFAULT_DATA_DIR, redis_url='redis://127.0.0.1:6379/0', loglevel=logging.DEBUG):
if not os.path.exists(data_dir) or not os.path.isdir(data_dir):
raise Exception("Datadir %s is not a valid directory" % data_dir)
self.worker_id = binascii.hexlify(os.urandom(8))
self.node_name = socket.gethostname()
self.data_dir = data_dir
self.data_files = set()
context = zmq.Context()
self.socket = context.socket(zmq.ROUTER)
self.socket.setsockopt(zmq.LINGER, 500)
self.socket.identity = self.worker_id
self.poller = zmq.Poller()
self.poller.register(self.socket, zmq.POLLIN | zmq.POLLOUT)
self.redis_server = redis.from_url(redis_url)
self.controllers = {} # Keep a dict of timestamps when you last spoke to controllers
self.check_controllers()
self.last_wrm = 0
self.start_time = time.time()
self.logger = bqueryd.logger.getChild('worker ' + self.worker_id)
self.logger.setLevel(loglevel)
self.msg_count = 0
signal.signal(signal.SIGTERM, self.term_signal())
def run(self):
""" Start the Authentication Agent thread task """
self.authenticator.start()
zap = self.authenticator.zap_socket
poller = zmq.Poller()
poller.register(self.pipe, zmq.POLLIN)
poller.register(zap, zmq.POLLIN)
while True:
try:
socks = dict(poller.poll())
except zmq.ZMQError:
break # interrupted
if self.pipe in socks and socks[self.pipe] == zmq.POLLIN:
terminate = self._handle_pipe()
if terminate:
break
if zap in socks and socks[zap] == zmq.POLLIN:
self._handle_zap()
self.pipe.close()
self.authenticator.stop()
def __init__(self, socket, io_loop=None):
self.socket = socket
self.io_loop = io_loop or IOLoop.instance()
self.poller = zmq.Poller()
self._send_queue = Queue()
self._recv_callback = None
self._send_callback = None
self._close_callback = None
self._recv_copy = False
self._flushed = False
self._state = self.io_loop.ERROR
self._init_io_state()
# shortcircuit some socket methods
self.bind = self.socket.bind
self.bind_to_random_port = self.socket.bind_to_random_port
self.connect = self.socket.connect
self.setsockopt = self.socket.setsockopt
self.getsockopt = self.socket.getsockopt
self.setsockopt_string = self.socket.setsockopt_string
self.getsockopt_string = self.socket.getsockopt_string
self.setsockopt_unicode = self.socket.setsockopt_unicode
self.getsockopt_unicode = self.socket.getsockopt_unicode
def run(self):
""" Start the Authentication Agent thread task """
self.authenticator.start()
zap = self.authenticator.zap_socket
poller = zmq.Poller()
poller.register(self.pipe, zmq.POLLIN)
poller.register(zap, zmq.POLLIN)
while True:
try:
socks = dict(poller.poll())
except zmq.ZMQError:
break # interrupted
if self.pipe in socks and socks[self.pipe] == zmq.POLLIN:
terminate = self._handle_pipe()
if terminate:
break
if zap in socks and socks[zap] == zmq.POLLIN:
self._handle_zap()
self.pipe.close()
self.authenticator.stop()
def test_timeout(self):
"""make sure Poller.poll timeout has the right units (milliseconds)."""
s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
poller = self.Poller()
poller.register(s1, zmq.POLLIN)
tic = time.time()
evt = poller.poll(.005)
toc = time.time()
self.assertTrue(toc-tic < 0.1)
tic = time.time()
evt = poller.poll(5)
toc = time.time()
self.assertTrue(toc-tic < 0.1)
self.assertTrue(toc-tic > .001)
tic = time.time()
evt = poller.poll(500)
toc = time.time()
self.assertTrue(toc-tic < 1)
self.assertTrue(toc-tic > 0.1)
def __init__(self, socket, io_loop=None):
self.socket = socket
self.io_loop = io_loop or IOLoop.instance()
self.poller = zmq.Poller()
self._send_queue = Queue()
self._recv_callback = None
self._send_callback = None
self._close_callback = None
self._recv_copy = False
self._flushed = False
self._state = self.io_loop.ERROR
self._init_io_state()
# shortcircuit some socket methods
self.bind = self.socket.bind
self.bind_to_random_port = self.socket.bind_to_random_port
self.connect = self.socket.connect
self.setsockopt = self.socket.setsockopt
self.getsockopt = self.socket.getsockopt
self.setsockopt_string = self.socket.setsockopt_string
self.getsockopt_string = self.socket.getsockopt_string
self.setsockopt_unicode = self.socket.setsockopt_unicode
self.getsockopt_unicode = self.socket.getsockopt_unicode
def __init__(self, socket, io_loop=None):
self.socket = socket
self.io_loop = io_loop or IOLoop.instance()
self.poller = zmq.Poller()
self._send_queue = Queue()
self._recv_callback = None
self._send_callback = None
self._close_callback = None
self._recv_copy = False
self._flushed = False
self._state = self.io_loop.ERROR
self._init_io_state()
# shortcircuit some socket methods
self.bind = self.socket.bind
self.bind_to_random_port = self.socket.bind_to_random_port
self.connect = self.socket.connect
self.setsockopt = self.socket.setsockopt
self.getsockopt = self.socket.getsockopt
self.setsockopt_string = self.socket.setsockopt_string
self.getsockopt_string = self.socket.getsockopt_string
self.setsockopt_unicode = self.socket.setsockopt_unicode
self.getsockopt_unicode = self.socket.getsockopt_unicode
def run(self):
""" Start the Authentication Agent thread task """
self.authenticator.start()
zap = self.authenticator.zap_socket
poller = zmq.Poller()
poller.register(self.pipe, zmq.POLLIN)
poller.register(zap, zmq.POLLIN)
while True:
try:
socks = dict(poller.poll())
except zmq.ZMQError:
break # interrupted
if self.pipe in socks and socks[self.pipe] == zmq.POLLIN:
terminate = self._handle_pipe()
if terminate:
break
if zap in socks and socks[zap] == zmq.POLLIN:
self._handle_zap()
self.pipe.close()
self.authenticator.stop()
def test_timeout(self):
"""make sure Poller.poll timeout has the right units (milliseconds)."""
s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
poller = self.Poller()
poller.register(s1, zmq.POLLIN)
tic = time.time()
evt = poller.poll(.005)
toc = time.time()
self.assertTrue(toc-tic < 0.1)
tic = time.time()
evt = poller.poll(5)
toc = time.time()
self.assertTrue(toc-tic < 0.1)
self.assertTrue(toc-tic > .001)
tic = time.time()
evt = poller.poll(500)
toc = time.time()
self.assertTrue(toc-tic < 1)
self.assertTrue(toc-tic > 0.1)
def __init__(self, socket, io_loop=None):
self.socket = socket
self.io_loop = io_loop or IOLoop.instance()
self.poller = zmq.Poller()
self._send_queue = Queue()
self._recv_callback = None
self._send_callback = None
self._close_callback = None
self._recv_copy = False
self._flushed = False
self._state = self.io_loop.ERROR
self._init_io_state()
# shortcircuit some socket methods
self.bind = self.socket.bind
self.bind_to_random_port = self.socket.bind_to_random_port
self.connect = self.socket.connect
self.setsockopt = self.socket.setsockopt
self.getsockopt = self.socket.getsockopt
self.setsockopt_string = self.socket.setsockopt_string
self.getsockopt_string = self.socket.getsockopt_string
self.setsockopt_unicode = self.socket.setsockopt_unicode
self.getsockopt_unicode = self.socket.getsockopt_unicode
def __init__(self, socket, io_loop=None):
self.socket = socket
self.io_loop = io_loop or IOLoop.instance()
self.poller = zmq.Poller()
self._send_queue = Queue()
self._recv_callback = None
self._send_callback = None
self._close_callback = None
self._recv_copy = False
self._flushed = False
self._state = self.io_loop.ERROR
self._init_io_state()
# shortcircuit some socket methods
self.bind = self.socket.bind
self.bind_to_random_port = self.socket.bind_to_random_port
self.connect = self.socket.connect
self.setsockopt = self.socket.setsockopt
self.getsockopt = self.socket.getsockopt
self.setsockopt_string = self.socket.setsockopt_string
self.getsockopt_string = self.socket.getsockopt_string
self.setsockopt_unicode = self.socket.setsockopt_unicode
self.getsockopt_unicode = self.socket.getsockopt_unicode
def run(self):
""" Start the Authentication Agent thread task """
self.authenticator.start()
zap = self.authenticator.zap_socket
poller = zmq.Poller()
poller.register(self.pipe, zmq.POLLIN)
poller.register(zap, zmq.POLLIN)
while True:
try:
socks = dict(poller.poll())
except zmq.ZMQError:
break # interrupted
if self.pipe in socks and socks[self.pipe] == zmq.POLLIN:
terminate = self._handle_pipe()
if terminate:
break
if zap in socks and socks[zap] == zmq.POLLIN:
self._handle_zap()
self.pipe.close()
self.authenticator.stop()
def test_timeout(self):
"""make sure Poller.poll timeout has the right units (milliseconds)."""
s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
poller = self.Poller()
poller.register(s1, zmq.POLLIN)
tic = time.time()
evt = poller.poll(.005)
toc = time.time()
self.assertTrue(toc-tic < 0.1)
tic = time.time()
evt = poller.poll(5)
toc = time.time()
self.assertTrue(toc-tic < 0.1)
self.assertTrue(toc-tic > .001)
tic = time.time()
evt = poller.poll(500)
toc = time.time()
self.assertTrue(toc-tic < 1)
self.assertTrue(toc-tic > 0.1)
def __init__(self, socket, io_loop=None):
self.socket = socket
self.io_loop = io_loop or IOLoop.instance()
self.poller = zmq.Poller()
self._send_queue = Queue()
self._recv_callback = None
self._send_callback = None
self._close_callback = None
self._recv_copy = False
self._flushed = False
self._state = self.io_loop.ERROR
self._init_io_state()
# shortcircuit some socket methods
self.bind = self.socket.bind
self.bind_to_random_port = self.socket.bind_to_random_port
self.connect = self.socket.connect
self.setsockopt = self.socket.setsockopt
self.getsockopt = self.socket.getsockopt
self.setsockopt_string = self.socket.setsockopt_string
self.getsockopt_string = self.socket.getsockopt_string
self.setsockopt_unicode = self.socket.setsockopt_unicode
self.getsockopt_unicode = self.socket.getsockopt_unicode
def __init__(self, server_ip, server_port, task_id='', debug=False):
if debug:
l.setLevel(logging.DEBUG)
l.debug("Hydra Analyser initiated...")
self.server_ip = server_ip
self.port = server_port
self.task_id = task_id
self.data = {} # This is where all received data will be stored
self.context = zmq.Context.instance()
self.poller = zmq.Poller()
self.req_msg = hdaemon_pb2.CommandMessage()
self.resp_msg = hdaemon_pb2.ResponseMessage()
l.debug("Connecting to server at [%s:%s]", self.server_ip, self.port)
self.socket = self.context.socket(zmq.REQ)
self.socket.connect("tcp://%s:%s" % (self.server_ip, self.port))
l.debug("Connected...")
def run(self):
self.log.debug("Broker starts XPUB:{}, XSUB:{}"
.format(self.xpub_url, self.xsub_url))
# self.proxy.start()
poller = zmq.Poller()
poller.register(self.xpub, zmq.POLLIN)
poller.register(self.xsub, zmq.POLLIN)
self.running = True
while self.running:
events = dict(poller.poll(1000))
if self.xpub in events:
message = self.xpub.recv_multipart()
self.log.debug("subscription message: {}".format(message[0]))
self.xsub.send_multipart(message)
if self.xsub in events:
message = self.xsub.recv_multipart()
self.log.debug("publishing message: {}".format(message))
self.xpub.send_multipart(message)
def __init__(self, port, pipeline=100, host='localhost', log_file=None):
"""Create a new ZMQDealer object.
"""
context = zmq.Context.instance()
# noinspection PyUnresolvedReferences
self.socket = context.socket(zmq.DEALER)
self.socket.hwm = pipeline
self.socket.connect('tcp://%s:%d' % (host, port))
self._log_file = log_file
self.poller = zmq.Poller()
# noinspection PyUnresolvedReferences
self.poller.register(self.socket, zmq.POLLIN)
if self._log_file is not None:
self._log_file = os.path.abspath(self._log_file)
# If log file directory does not exists, create it
log_dir = os.path.dirname(self._log_file)
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# clears any existing log
if os.path.exists(self._log_file):
os.remove(self._log_file)
def __init__(self, targname, cfg, isServer=False):
self.targname = targname
self.cfg = cfg
self.isServer = isServer
self.fnCallName = ''
self.ctx = zmq.Context()
self.ctx.linger = 100
if not self.isServer:
self.sock = self.ctx.socket(zmq.DEALER)
self.sock.linger = 100
self.sock.connect('tcp://%s:%s' % (self.cfg['server'],self.cfg.get('port',7677))) # this times out with EINVAL when no internet
self.poller = zmq.Poller()
self.poller.register(self.sock, zmq.POLLIN)
else:
self.sock = self.ctx.socket(zmq.ROUTER)
self.sock.linger = 100
self.sock.bind('tcp://*:%s' % (self.cfg.get('port',7677)))
self.poller = zmq.Poller()
self.poller.register(self.sock, zmq.POLLIN)
self.be = GetBackend(self.cfg['backend'])(self.targname, self.cfg)
self.inTime = time.time()
self.inactiveLimit = int(self.cfg.get('inactivelimit',0))
print 'inactivelimit ',self.inactiveLimit
def ready(self, name, polling_mechanism, timeout=None):
"""Polls all sockets and checks if the socket with the given name is ready for either ``zmq.POLLIN`` or ``zmq.POLLOUT``.
returns the socket if available, or ``None``
:param socket_name: the socket name
:param polling_mechanism: either ``zmq.POLLIN`` or ``zmq.POLLOUT``
:param timeout: the polling timeout in miliseconds that will
be passed to ``zmq.Poller().poll()`` (optional, defaults to
``core.DEFAULT_POLLING_TIMEOUT``)
"""
socket = self.get_by_name(name)
available_mechanism = self.engage(timeout is None and self.timeout or timeout).pop(socket, None)
if polling_mechanism == available_mechanism:
return socket
def wait_until_ready(self, name, polling_mechanism, timeout=None, polling_timeout=None):
"""Briefly waits until the socket is ready to be used, yields to other
greenlets until the socket becomes available.
returns the socket if available within the given timeout, or ``None``
:param socket_name: the socket name
:param polling_mechanism: either ``zmq.POLLIN`` or ``zmq.POLLOUT``
:param timeout: the timeout in seconds (accepts float) in which it
should wait for the socket to become available
(optional, defaults to ``core.DEFAULT_TIMEOUT_IN_SECONDS``)
:param polling_timeout: the polling timeout in miliseconds that will
be passed to ``zmq.Poller().poll()``.
(optional, defaults to ``core.DEFAULT_POLLING_TIMEOUT``)
"""
timeout = timeout is None and self.timeout or timeout
polling_timeout = polling_timeout is None and self.polling_timeout or polling_timeout
start = time.time()
current = start
while current - start < timeout:
self.engage(polling_timeout)
socket = self.ready(name, polling_mechanism, timeout=timeout)
current = time.time()
if socket:
return socket
def _send_raw(self, serialized):
self.create_socket()
self._socket.send_string(serialized, zmq.NOBLOCK)
poller = zmq.Poller()
poller.register(self._socket, zmq.POLLIN)
if poller.poll(self._timeout * 1000):
msg = self._socket.recv()
self.on_message(msg)
self.cleanup_socket()
else:
self._transport.log("Peer " + self._address + " timed out.")
self.cleanup_socket()
self._transport.remove_peer(self._address)
def zmq_request(self, msg_type, msg_content, timeout=__DEFAULT_REQUEST_TIMEOUT):
# new socket to talk to server
self.__socket = zmq.Context().socket(zmq.REQ)
self.__socket.connect("tcp://localhost:" + ZMQPort.RQ)
# init poller and register to socket that web can poll socket to check is it has messages
poller = zmq.Poller()
poller.register(self.__socket, zmq.POLLIN)
send_flatbuf_msg(self.__socket, msg_type, msg_content)
reqs = 0
while reqs * self.__POLL_INTERVAL <= timeout:
socks = dict(poller.poll(self.__POLL_INTERVAL))
if self.__socket in socks and socks[self.__socket] == zmq.POLLIN:
msg = self.__socket.recv()
msgObj = TransMsg.GetRootAsTransMsg(msg, 0)
return msgObj.Content()
reqs = reqs + 1
return False
def test_getsockopt_events(self):
sock1, sock2, _port = self.create_bound_pair(zmq.DEALER, zmq.DEALER)
eventlet.sleep()
poll_out = zmq.Poller()
poll_out.register(sock1, zmq.POLLOUT)
sock_map = poll_out.poll(100)
self.assertEqual(len(sock_map), 1)
events = sock1.getsockopt(zmq.EVENTS)
self.assertEqual(events & zmq.POLLOUT, zmq.POLLOUT)
sock1.send(b'')
poll_in = zmq.Poller()
poll_in.register(sock2, zmq.POLLIN)
sock_map = poll_in.poll(100)
self.assertEqual(len(sock_map), 1)
events = sock2.getsockopt(zmq.EVENTS)
self.assertEqual(events & zmq.POLLIN, zmq.POLLIN)
def __init__(self, *args, **kwargs):
super(ZMQPlaybook, self).__init__(*args, **kwargs)
self._sockets = []
self._hooks = {}
self.context = kwargs.get("context", zmq.Context.instance())
self.socket = self.context.socket(zmq.SUB)
self.socket_dir = tempfile.mkdtemp()
self._env['DAUBER_SOCKET_URI'] = "ipc://{}/dauber.socket".format(self.socket_dir)
self.socket.connect(self._env['DAUBER_SOCKET_URI'])
self._env['DAUBER_CONTROL_SOCKET_URI'] = "ipc://{}/control.socket".format(self.socket_dir)
self.poller = zmq.Poller()
self._register_socket(self.socket, self.__class__._zmq_socket_handler)
self.add_callback_plugin_dir(
pr.resource_filename(__name__, 'ansible/callback_plugins'))
self.logger.setLevel(logging.WARNING)
self.verbosity = 4
def _wait_for_goahead(self):
control_socket = self.context.socket(zmq.REP)
control_socket.bind(os.environ['DAUBER_CONTROL_SOCKET_URI'])
poller = zmq.Poller()
poller.register(control_socket)
timeout = 500
t_last = time.time()
while (time.time() - t_last) < timeout:
ready = dict(poller.poll(10))
if ready.get(control_socket):
control_socket.recv()
control_socket.send(b'')
break
self.socket.send_multipart(['hello', b''])
t_last = time.time()
assert (time.time() - t_last) < timeout, \
"Timed out before recieving a signal to continue"
del control_socket
def monitored_queue(in_socket, out_socket, mon_socket,
in_prefix=b'in', out_prefix=b'out'):
swap_ids = in_socket.type == zmq.ROUTER and out_socket.type == zmq.ROUTER
poller = zmq.Poller()
poller.register(in_socket, zmq.POLLIN)
poller.register(out_socket, zmq.POLLIN)
while True:
events = dict(poller.poll())
if in_socket in events:
_relay(in_socket, out_socket, mon_socket, in_prefix, swap_ids)
if out_socket in events:
_relay(out_socket, in_socket, mon_socket, out_prefix, swap_ids)
def test_retry_poll(self):
x, y = self.create_bound_pair()
poller = zmq.Poller()
poller.register(x, zmq.POLLIN)
self.alarm()
def send():
time.sleep(2 * self.signal_delay)
y.send(b('ping'))
t = Thread(target=send)
t.start()
evts = dict(poller.poll(2 * self.timeout_ms))
t.join()
assert x in evts
assert self.timer_fired
x.recv()
def test_pair(self):
s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
# Sleep to allow sockets to connect.
wait()
poller = self.Poller()
poller.register(s1, zmq.POLLIN|zmq.POLLOUT)
poller.register(s2, zmq.POLLIN|zmq.POLLOUT)
# Poll result should contain both sockets
socks = dict(poller.poll())
# Now make sure that both are send ready.
self.assertEqual(socks[s1], zmq.POLLOUT)
self.assertEqual(socks[s2], zmq.POLLOUT)
# Now do a send on both, wait and test for zmq.POLLOUT|zmq.POLLIN
s1.send(b'msg1')
s2.send(b'msg2')
wait()
socks = dict(poller.poll())
self.assertEqual(socks[s1], zmq.POLLOUT|zmq.POLLIN)
self.assertEqual(socks[s2], zmq.POLLOUT|zmq.POLLIN)
# Make sure that both are in POLLOUT after recv.
s1.recv()
s2.recv()
socks = dict(poller.poll())
self.assertEqual(socks[s1], zmq.POLLOUT)
self.assertEqual(socks[s2], zmq.POLLOUT)
poller.unregister(s1)
poller.unregister(s2)
# Wait for everything to finish.
wait()
def test_pubsub(self):
s1, s2 = self.create_bound_pair(zmq.PUB, zmq.SUB)
s2.setsockopt(zmq.SUBSCRIBE, b'')
# Sleep to allow sockets to connect.
wait()
poller = self.Poller()
poller.register(s1, zmq.POLLIN|zmq.POLLOUT)
poller.register(s2, zmq.POLLIN)
# Now make sure that both are send ready.
socks = dict(poller.poll())
self.assertEqual(socks[s1], zmq.POLLOUT)
self.assertEqual(s2 in socks, 0)
# Make sure that s1 stays in POLLOUT after a send.
s1.send(b'msg1')
socks = dict(poller.poll())
self.assertEqual(socks[s1], zmq.POLLOUT)
# Make sure that s2 is POLLIN after waiting.
wait()
socks = dict(poller.poll())
self.assertEqual(socks[s2], zmq.POLLIN)
# Make sure that s2 goes into 0 after recv.
s2.recv()
socks = dict(poller.poll())
self.assertEqual(s2 in socks, 0)
poller.unregister(s1)
poller.unregister(s2)
# Wait for everything to finish.
wait()
def test_wakeup(self):
s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
poller = self.Poller()
poller.register(s2, zmq.POLLIN)
tic = time.time()
r = gevent.spawn(lambda: poller.poll(10000))
s = gevent.spawn(lambda: s1.send(b'msg1'))
r.join()
toc = time.time()
self.assertTrue(toc-tic < 1)
def __init__(self):
super().__init__()
# this maps file descriptors to keys
self._fd_to_key = {}
# read-only mapping returned by get_map()
self._map = _SelectorMapping(self)
self._zmq_poller = _zmq.Poller()
def thread_loop(self, context, pipe):
poller = zmq.Poller()
ipc_pub = zmq_tools.Msg_Dispatcher(context, self.g_pool.ipc_push_url)
poller.register(pipe, zmq.POLLIN)
remote_socket = None
while True:
items = dict(poller.poll())
if pipe in items:
cmd = pipe.recv_string()
if cmd == 'Exit':
break
elif cmd == 'Bind':
new_url = pipe.recv_string()
if remote_socket:
poller.unregister(remote_socket)
remote_socket.close(linger=0)
try:
remote_socket = context.socket(zmq.REP)
remote_socket.bind(new_url)
except zmq.ZMQError as e:
remote_socket = None
pipe.send_string("Error", flags=zmq.SNDMORE)
pipe.send_string("Could not bind to Socket: {}. Reason: {}".format(new_url, e))
else:
pipe.send_string("Bind OK", flags=zmq.SNDMORE)
# `.last_endpoint` is already of type `bytes`
pipe.send(remote_socket.last_endpoint.replace(b"tcp://", b""))
poller.register(remote_socket, zmq.POLLIN)
if remote_socket in items:
self.on_recv(remote_socket, ipc_pub)
self.thread_pipe = None
def __init__(self, host, port=7772):
QtCore.QObject.__init__(self)
self.context = zmq.Context()
self.socket = self.context.socket(zmq.SUB)
self.socket.connect("tcp://{}:{}".format(host, port))
self.socket.setsockopt_string(zmq.SUBSCRIBE, "data")
self.socket.setsockopt_string(zmq.SUBSCRIBE, "done")
self.poller = zmq.Poller()
self.poller.register(self.socket, zmq.POLLIN)
self.running = True
def open_connection(self, address, status_port, data_port):
self.statusBar().showMessage("Open session to {}:{}".format(address, status_port), 2000)
socket = self.context.socket(zmq.DEALER)
socket.identity = "Matplotlib_Qt_Client".encode()
socket.connect("tcp://{}:{}".format(address, status_port))
socket.send(b"WHATSUP")
poller = zmq.Poller()
poller.register(socket, zmq.POLLOUT)
time.sleep(0.1)
evts = dict(poller.poll(100))
if socket in evts:
try:
reply, desc = [e.decode() for e in socket.recv_multipart()]
desc = json.loads(desc)
self.statusBar().showMessage("Connection established. Pulling plot information.", 2000)
except:
self.statusBar().showMessage("Could not connect to server.", 2000)
return
else:
self.statusBar().showMessage("Server did not respond.", 2000)
socket.close()
self.construct_plots(desc)
# Actual data listener
if self.listener_thread:
self.Datalistener.running = False
self.listener_thread.quit()
self.listener_thread.wait()
self.listener_thread = QtCore.QThread()
self.Datalistener = DataListener(address, data_port)
self.Datalistener.moveToThread(self.listener_thread)
self.listener_thread.started.connect(self.Datalistener.loop)
self.Datalistener.message.connect(self.data_signal_received)
self.Datalistener.finished.connect(self.stop_listening)
QtCore.QTimer.singleShot(0, self.listener_thread.start)
def run(self):
""" Main loop of the thread. """
# create event socket
self.subscriber = EventSubscriber(self.zmq_context,
self.event_port,
self.logger)
self.configure()
# create poller and register event subscriber
poller = zmq.Poller()
poller.register(self.subscriber.socket, zmq.POLLIN)
# poll events every seconds
self.logger.info('entering main loop')
while not self.stop_event.is_set():
socks = dict(poller.poll(self._Poll_timeout))
# check if something happened on the socket
if self.subscriber.socket in socks and \
socks[self.subscriber.socket] == zmq.POLLIN:
self.logger.debug('got message on subscriber')
try:
message = self.subscriber.receive()
except Exception, e:
self.logger.error(
'failed to get data from subscriber: {}'.format(
e.message))
else:
if message[0] == EventHeaders.SUPVISORS:
self.on_supvisors_status(message[1])
elif message[0] == EventHeaders.ADDRESS:
self.on_address_status(message[1])
elif message[0] == EventHeaders.APPLICATION:
self.on_application_status(message[1])
elif message[0] == EventHeaders.PROCESS_EVENT:
self.on_process_event(message[1])
elif message[0] == EventHeaders.PROCESS_STATUS:
self.on_process_status(message[1])
self.logger.warn('exiting main loop')
self.subscriber.close()
def _receive_with_timeout(self, socket, timeout_s, use_multipart=False):
"""Check for socket activity and either return what's
received on the socket or time out if timeout_s expires
without anything on the socket.
This is implemented in loops of self.try_length_ms milliseconds
to allow Ctrl-C handling to take place.
"""
if timeout_s is config.FOREVER:
timeout_ms = config.FOREVER
else:
timeout_ms = int(1000 * timeout_s)
poller = zmq.Poller()
poller.register(socket, zmq.POLLIN)
ms_so_far = 0
try:
for interval_ms in self.intervals_ms(timeout_ms):
sockets = dict(poller.poll(interval_ms))
ms_so_far += interval_ms
if socket in sockets:
if use_multipart:
return socket.recv_multipart()
else:
return socket.recv()
else:
raise core.SocketTimedOutError(timeout_s)
except KeyboardInterrupt:
raise core.SocketInterruptedError(ms_so_far / 1000.0)
def __init__(self, redis_url='redis://127.0.0.1:6379/0', loglevel=logging.INFO):
self.redis_url = redis_url
self.redis_server = redis.from_url(redis_url)
self.context = zmq.Context()
self.socket = self.context.socket(zmq.ROUTER)
self.socket.setsockopt(zmq.LINGER, 500)
self.socket.setsockopt(zmq.ROUTER_MANDATORY, 1) # Paranoid for debugging purposes
self.socket.setsockopt(zmq.SNDTIMEO, 1000) # Short timeout
self.poller = zmq.Poller()
self.poller.register(self.socket, zmq.POLLIN | zmq.POLLOUT)
self.node_name = socket.gethostname()
self.address = bind_to_random_port(self.socket, 'tcp://' + get_my_ip(), min_port=14300, max_port=14399,
max_tries=100)
with open(os.path.join(RUNFILES_LOCATION, 'bqueryd_controller.address'), 'w') as F:
F.write(self.address)
with open(os.path.join(RUNFILES_LOCATION, 'bqueryd_controller.pid'), 'w') as F:
F.write(str(os.getpid()))
self.logger = bqueryd.logger.getChild('controller').getChild(self.address)
self.logger.setLevel(loglevel)
self.msg_count_in = 0
self.rpc_results = [] # buffer of results that are ready to be returned to callers
self.rpc_segments = {} # Certain RPC calls get split and divided over workers, this dict tracks the original RPCs
self.worker_map = {} # maintain a list of connected workers TODO get rid of unresponsive ones...
self.files_map = {} # shows on which workers a file is available on
self.worker_out_messages = {None: []} # A dict of buffers, used to round-robin based on message affinity
self.worker_out_messages_sequence = [None] # used to round-robin the outgoing messages
self.is_running = True
self.last_heartbeat = 0
self.others = {} # A dict of other Controllers running on other DQE nodes
self.start_time = time.time()
def test_pair(self):
s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
# Sleep to allow sockets to connect.
wait()
poller = self.Poller()
poller.register(s1, zmq.POLLIN|zmq.POLLOUT)
poller.register(s2, zmq.POLLIN|zmq.POLLOUT)
# Poll result should contain both sockets
socks = dict(poller.poll())
# Now make sure that both are send ready.
self.assertEqual(socks[s1], zmq.POLLOUT)
self.assertEqual(socks[s2], zmq.POLLOUT)
# Now do a send on both, wait and test for zmq.POLLOUT|zmq.POLLIN
s1.send(b'msg1')
s2.send(b'msg2')
wait()
socks = dict(poller.poll())
self.assertEqual(socks[s1], zmq.POLLOUT|zmq.POLLIN)
self.assertEqual(socks[s2], zmq.POLLOUT|zmq.POLLIN)
# Make sure that both are in POLLOUT after recv.
s1.recv()
s2.recv()
socks = dict(poller.poll())
self.assertEqual(socks[s1], zmq.POLLOUT)
self.assertEqual(socks[s2], zmq.POLLOUT)
poller.unregister(s1)
poller.unregister(s2)
# Wait for everything to finish.
wait()
def test_no_events(self):
s1, s2 = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
poller = self.Poller()
poller.register(s1, zmq.POLLIN|zmq.POLLOUT)
poller.register(s2, 0)
self.assertTrue(s1 in poller)
self.assertFalse(s2 in poller)
poller.register(s1, 0)
self.assertFalse(s1 in poller)
def test_pubsub(self):
s1, s2 = self.create_bound_pair(zmq.PUB, zmq.SUB)
s2.setsockopt(zmq.SUBSCRIBE, b'')
# Sleep to allow sockets to connect.
wait()
poller = self.Poller()
poller.register(s1, zmq.POLLIN|zmq.POLLOUT)
poller.register(s2, zmq.POLLIN)
# Now make sure that both are send ready.
socks = dict(poller.poll())
self.assertEqual(socks[s1], zmq.POLLOUT)
self.assertEqual(s2 in socks, 0)
# Make sure that s1 stays in POLLOUT after a send.
s1.send(b'msg1')
socks = dict(poller.poll())
self.assertEqual(socks[s1], zmq.POLLOUT)
# Make sure that s2 is POLLIN after waiting.
wait()
socks = dict(poller.poll())
self.assertEqual(socks[s2], zmq.POLLIN)
# Make sure that s2 goes into 0 after recv.
s2.recv()
socks = dict(poller.poll())
self.assertEqual(s2 in socks, 0)
poller.unregister(s1)
poller.unregister(s2)
# Wait for everything to finish.
wait()