From ffc2e34334f2f498bdacbf232e9c5bd56a8be0b5 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 24 Feb 2012 12:51:46 -0500 Subject: [PATCH 1/9] Unifying protocol and messaging. --- zipline/component.py | 131 +++++++++++++++++++++++++++-------------- zipline/messaging.py | 50 ++++++++++------ zipline/protocol.py | 29 +++++++++ zipline/test/client.py | 7 ++- zipline/util.py | 43 +++++++++----- 5 files changed, 183 insertions(+), 77 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 036a04b5..06a85b5b 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -1,10 +1,13 @@ """ Commonly used messaging components. """ -import json +import os import uuid -import datetime +import socket +import humanhash + import zipline.util as qutil +from zipline.protocol import CONTROL_PROTOCOL class Component(object): @@ -14,8 +17,6 @@ class Component(object): - sync_address: socket address used for synchronizing the start of all workers, heartbeating, and exit notification will be used in REP/REQ sockets. Bind is always on the REP side. - - control_address: socket address used for controlling and - monitoring the status of the simulation - data_address: socket address used for data sources to stream their records. will be used in PUSH/PULL sockets between data sources and a ParallelBuffer (aka the Feed). Bind will always be on the PULL side (we always have N producers and 1 consumer) @@ -31,15 +32,22 @@ class Component(object): will also return a Poller. """ - self.zmq = None - self.context = None - self.addresses = None - self.out_socket = None - self.gevent_needed = False - self.killed = False + self.zmq = None + self.context = None + self.addresses = None + self.out_socket = None + self.gevent_needed = False + self.killed = False + self.heartbeat_timeout = 2000 - # TODO: could probably mkae this into a property instead of a - # method + self.guid = uuid.uuid4() + self.huid = humanhash.humanize(self.guid.hex) + + # ------------ + # Core Methods + # ------------ + + @property def get_id(self): raise NotImplementedError @@ -62,17 +70,12 @@ class Component(object): def do_work(self): raise NotImplementedError - def run(self): - - fail = None - - #try: - #TODO: can't initialize these values in the __init__? + def _run(self): self.done = False self.sockets = [] if self.gevent_needed: - qutil.LOGGER.info("Loading gevent specific zmq for {id}".format(id=self.get_id())) + qutil.LOGGER.info("Loading gevent specific zmq for {id}".format(id=self.get_id)) import gevent_zeromq self.zmq = gevent_zeromq.zmq else: @@ -89,50 +92,68 @@ class Component(object): for sock in self.sockets: sock.close() - #except Exception as e: - #qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id())) - #fail = e + def run(self, catch_exceptions=False): - #finally: + fail = None - #if(self.context != None): - #self.context.destroy() - - #if fail: - #raise fail + # Catching all exceptions makes this really hard to + # debug, is it with care. + if catch_exceptions: + try: + self._run() + except Exception as e: + qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id)) + fail = e + finally: + if(self.context != None): + self.context.destroy() + if fail: + raise fail + else: + self._run() + if(self.context != None): + self.context.destroy() def loop(self): while not self.done: self.confirm() self.do_work() + # ----------- + # Messaging + # ----------- + def signal_done(self): #notify down stream components that we're done if(self.out_socket != None): - self.out_socket.send("DONE") + self.out_socket.send(str(CONTROL_PROTOCOL.DONE)) #notify host we're done - self.sync_socket.send(self.get_id() + ":DONE") + + # TODO: proper framing + self.sync_socket.send(self.get_id + ":" + str(CONTROL_PROTOCOL.DONE)) + self.receive_sync_ack() #notify internal work look that we're done self.done = True - # TODO: probably don't need a method here ... or move into - # higher level framing protocol - def is_done_message(self, message): - return message == "DONE" - def confirm(self): # send a synchronization request to the host - self.sync_socket.send(self.get_id() + ":RUN") - self.receive_sync_ack() + + # TODO: proper framing + self.sync_socket.send(self.get_id + ":RUN") + + self.receive_sync_ack() # blocking def receive_sync_ack(self): - # wait for synchronization reply from the host - socks = dict(self.sync_poller.poll(2000)) #timeout after 2 seconds. + """ + Wait for synchronization reply from the host. + """ + + socks = dict(self.sync_poller.poll(self.heartbeat_timeout)) if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: message = self.sync_socket.recv() else: - raise Exception("Sync ack timed out on response for {id}".format(id=self.get_id())) + raise Exception("Sync ack timed out on response for {id}".format(id=self.get_id)) def bind_data(self): return self.bind_pull_socket(self.addresses['data_address']) @@ -164,6 +185,7 @@ class Component(object): poller = self.zmq.Poller() poller.register(pull_socket, self.zmq.POLLIN) self.sockets.append(pull_socket) + return pull_socket, poller def connect_push_socket(self, addr): @@ -172,6 +194,7 @@ class Component(object): #push_socket.setsockopt(self.zmq.LINGER,0) self.sockets.append(push_socket) self.out_socket = push_socket + return push_socket def bind_pub_socket(self, addr): @@ -179,15 +202,19 @@ class Component(object): pub_socket.bind(addr) #pub_socket.setsockopt(self.zmq.LINGER,0) self.out_socket = pub_socket + return pub_socket def connect_sub_socket(self, addr): sub_socket = self.context.socket(self.zmq.SUB) sub_socket.connect(addr) sub_socket.setsockopt(self.zmq.SUBSCRIBE,'') + self.sockets.append(sub_socket) + poller = self.zmq.Poller() poller.register(sub_socket, self.zmq.POLLIN) - self.sockets.append(sub_socket) + + # TODO: migrate tuple unpacking to be consistent return sub_socket, poller def setup_control(self): @@ -196,10 +223,10 @@ class Component(object): overall status of the simulation and to forcefully tear down the simulation in case of a failure. """ - pass + assert self.controller def setup_sync(self): - qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id())) + qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id)) self.sync_socket = self.context.socket(self.zmq.REQ) self.sync_socket.connect(self.addresses['sync_address']) @@ -208,3 +235,21 @@ class Component(object): self.sync_poller.register(self.sync_socket, self.zmq.POLLIN) self.sockets.append(self.sync_socket) + + def debug(self): + return ( + self.get_id , + self.huid , + socket.gethostname() , + os.getpid() , + hex(id(self)) , + ) + + def __repr__(self): + return "<{name} {uuid} at {host} {pid} {pointer}>".format( + name = self.get_id , + uuid = self.huid , + host = socket.gethostname() , + pid = os.getpid() , + pointer = hex(id(self)) , + ) diff --git a/zipline/messaging.py b/zipline/messaging.py index 970c243f..c7aa1d89 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -7,17 +7,20 @@ import datetime import zipline.util as qutil from zipline.component import Component +from zipline.protocol import CONTROL_PROTOCOL + class ComponentHost(Component): """ - Components that can launch multiple sub-components, synchronize their start, and then wait for all - components to be finished. + Components that can launch multiple sub-components, synchronize their + start, and then wait for all components to be finished. """ def __init__(self, addresses, gevent_needed=False): Component.__init__(self) self.addresses = addresses - #workaround for defect in threaded use of strptime: http://bugs.python.org/issue11108 + # workaround for defect in threaded use of strptime: + # http://bugs.python.org/issue11108 qutil.parse_date("2012/02/13-10:04:28.114") self.components = {} @@ -47,13 +50,13 @@ class ComponentHost(Component): if self.controller: component.controller = self.controller - self.components[component.get_id()] = component - self.sync_register[component.get_id()] = datetime.datetime.utcnow() + self.components[component.get_id] = component + self.sync_register[component.get_id] = datetime.datetime.utcnow() if(isinstance(component, DataSource)): - self.feed.add_source(component.get_id()) + self.feed.add_source(component.get_id) if(isinstance(component, BaseTransform)): - self.merge.add_source(component.get_id()) + self.merge.add_source(component.get_id) def unregister_component(self, component_id): del self.components[component_id] @@ -97,15 +100,19 @@ class ComponentHost(Component): if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: msg = self.sync_socket.recv() parts = msg.split(':') - if(len(parts) < 2): + + if len(parts) != 2: qutil.LOGGER.info("got bad confirm: {msg}".format(msg=msg)) - sync_id = parts[0] - status = parts[1] - if(self.is_done_message(status)): + continue + + sync_id, status = parts + + if status == str(CONTROL_PROTOCOL.DONE): # TODO: other way around qutil.LOGGER.info("{id} is DONE".format(id=sync_id)) self.unregister_component(sync_id) else: self.sync_register[sync_id] = datetime.datetime.utcnow() + #qutil.LOGGER.info("confirmed {id}".format(id=msg)) # send synchronization reply self.sync_socket.send('ack', self.zmq.NOBLOCK) @@ -119,9 +126,10 @@ class ComponentHost(Component): class ParallelBuffer(Component): """ - Connects to N PULL sockets, publishing all messages received to a PUB socket. - Published messages are guaranteed to be in chronological order based on message property dt. - Expects to be instantiated in one execution context (thread, process, etc) and run in another. + Connects to N PULL sockets, publishing all messages received to a PUB + socket. Published messages are guaranteed to be in chronological order + based on message property dt. Expects to be instantiated in one execution + context (thread, process, etc) and run in another. """ def __init__(self): @@ -133,6 +141,7 @@ class ParallelBuffer(Component): self.ds_finished_counter = 0 + @property def get_id(self): return "FEED" @@ -149,7 +158,7 @@ class ParallelBuffer(Component): if self.pull_socket in socks and socks[self.pull_socket] == self.zmq.POLLIN: message = self.pull_socket.recv() - if self.is_done_message(message): + if message == str(CONTROL_PROTOCOL.DONE): self.ds_finished_counter += 1 if len(self.data_buffer) == self.ds_finished_counter: #drain any remaining messages in the buffer @@ -262,6 +271,7 @@ class MergedParallelBuffer(ParallelBuffer): result[source] = cur['value'] return result + @property def get_id(self): return "MERGE" @@ -283,6 +293,7 @@ class BaseTransform(Component): self.state = {} self.state['name'] = name + @property def get_id(self): return self.state['name'] @@ -305,7 +316,7 @@ class BaseTransform(Component): socks = dict(self.poller.poll(2000)) #timeout after 2 seconds. if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN: message = self.feed_socket.recv() - if self.is_done_message(message): + if message == str(CONTROL_PROTOCOL.DONE): self.signal_done() return @@ -313,6 +324,7 @@ class BaseTransform(Component): cur_state = self.transform(event) cur_state['dt'] = event['dt'] cur_state['id'] = self.state['name'] + self.result_socket.send(json.dumps(cur_state), self.zmq.NOBLOCK) def transform(self, event): @@ -321,8 +333,9 @@ class BaseTransform(Component): {name:"name of new transform", value: "value of new field"} - Transforms run in parallel and results are merged into a single map, so transform names must be unique. - Best practice is to use the self.state object initialized from the transform configuration, and only set the + Transforms run in parallel and results are merged into a single map, so + transform names must be unique. Best practice is to use the self.state + object initialized from the transform configuration, and only set the transformed value:: self.state['value'] = transformed_value @@ -350,6 +363,7 @@ class DataSource(Component): self.id = source_id self.cur_event = None + @property def get_id(self): return self.id diff --git a/zipline/protocol.py b/zipline/protocol.py index a992e20a..818feab8 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -1,3 +1,32 @@ #import msgpack #import ujson #import ultrajson_numpy + +from ctypes import Structure, c_ubyte + +def Enum(*options): + """ + Fast enums are very important when we want really tight zmq + loops. These are probably going to evolve into pure C structs + anyways so might as well get going on that. + """ + class cstruct(Structure): + _fields_ = [(o, c_ubyte) for o in options] + return cstruct(*range(len(options))) + +CONTROL_PROTOCOL = Enum( + 'INIT' , # 0 - req + 'INFO' , # 1 - req + 'STATUS' , # 2 - req + 'SHUTDOWN' , # 3 - req + 'KILL' , # 4 - req + + 'OK' , # 5 - rep + 'DONE' , # 6 - rep + 'EXCEPTION' , # 7 - rep +) + +HEARTBEAT_PROTOCOL = { + 'REQ' : '\x01', + 'REP' : '\x02', +} diff --git a/zipline/test/client.py b/zipline/test/client.py index 3b975ec2..8c3b408d 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -2,6 +2,8 @@ import json import zipline.util as qutil import zipline.messaging as qmsg +from zipline.protocol import CONTROL_PROTOCOL + class TestClient(qmsg.Component): def __init__(self, utest, expected_msg_count=0): @@ -11,6 +13,7 @@ class TestClient(qmsg.Component): self.utest = utest self.prev_dt = None + @property def get_id(self): return "TEST_CLIENT" @@ -19,9 +22,11 @@ class TestClient(qmsg.Component): def do_work(self): socks = dict(self.poller.poll(2000)) #timeout after 2 seconds. + if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN: msg = self.data_feed.recv() - if(self.is_done_message(msg)): + + if msg == str(CONTROL_PROTOCOL.DONE): qutil.LOGGER.info("Client is DONE!") self.signal_done() self.utest.assertEqual(self.expected_msg_count, self.received_count, diff --git a/zipline/util.py b/zipline/util.py index e29f75a1..1178b893 100644 --- a/zipline/util.py +++ b/zipline/util.py @@ -7,32 +7,45 @@ import datetime import pytz import logging - LOGGER = logging.getLogger('QSimLogger') +def configure_logging(loglevel=logging.DEBUG): + """ + Configures zipline.util.LOGGER to write a rotating file + (10M per file, 5 files) to `` /var/log/zipline.log ``. + """ + LOGGER.setLevel(loglevel) + handler = logging.handlers.RotatingFileHandler( + "/var/log/zipline/{lfn}.log".format(lfn="zipline"), + maxBytes=10*1024*1024, backupCount=5 + ) + handler.setFormatter(logging.Formatter( + "%(asctime)s %(levelname)s %(filename)s %(funcName)s - %(message)s", + "%Y-%m-%d %H:%M:%S %Z") + ) + LOGGER.addHandler(handler) + LOGGER.info("logging started...") + def parse_date(dt_str): - """parse strings according to the same format as generated by format_date""" + """ + Parse strings according to the same format as generated by + format_date. + """ if(dt_str == None): return None parts = dt_str.split(".") - dt = datetime.datetime.strptime(parts[0], '%Y/%m/%d-%H:%M:%S').replace(microsecond=int(parts[1]+"000")).replace(tzinfo = pytz.utc) + dt = datetime.datetime.strptime(parts[0], '%Y/%m/%d-%H:%M:%S').replace( + microsecond=int(parts[1]+"000")).replace(tzinfo = pytz.utc + ) return dt def format_date(dt): - """Format the date into a date with millesecond resolution and string/alphabetical - sorting that is equivalent to datetime sorting""" + """ + Format the date into a date with millesecond resolution and + string/alphabetical sorting that is equivalent to datetime sorting. + """ if(dt == None): return None dt_str = dt.strftime('%Y/%m/%d-%H:%M:%S') + "." + str(dt.microsecond / 1000) return dt_str -def configure_logging(loglevel=logging.DEBUG): - """configures zipline.util.LOGGER to write a rotating file (10M per file, 5 files) to /var/log/zipline.log""" - LOGGER.setLevel(loglevel) - handler = logging.handlers.RotatingFileHandler( - "/var/log/zipline/{lfn}.log".format(lfn="zipline"), - maxBytes=10*1024*1024, backupCount=5) - handler.setFormatter(logging.Formatter( - "%(asctime)s %(levelname)s %(filename)s %(funcName)s - %(message)s","%Y-%m-%d %H:%M:%S %Z")) - LOGGER.addHandler(handler) - LOGGER.info("logging started...") \ No newline at end of file From 4aad3a2f930059809cdefdb8cb6da0dfb1f0258f Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 24 Feb 2012 12:55:39 -0500 Subject: [PATCH 2/9] Added initial msgpack. --- etc/requirements.txt | 2 ++ zipline/protocol.py | 2 +- zipline/test/test_messaging.py | 2 ++ 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/etc/requirements.txt b/etc/requirements.txt index 688bee9a..744b7288 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -1,3 +1,5 @@ #zeromq related pyzmq==2.1.11 gevent-zeromq==0.2.2 +msgpack-python==0.1.12 +humanhash==0.0.1 diff --git a/zipline/protocol.py b/zipline/protocol.py index 818feab8..a40ecfe7 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -1,4 +1,4 @@ -#import msgpack +import msgpack #import ujson #import ultrajson_numpy diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 3ab7e69b..9a244ee7 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -104,6 +104,8 @@ class SimulatorTestCase(object): sim.register_controller( con ) sim.register_components([ret1, ret2, client]) + assert False + # Simulation # ---------- sim.simulate() From 320524fed997930adf70fda2a9d4f417016b5537 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 24 Feb 2012 13:00:28 -0500 Subject: [PATCH 3/9] Woops, left pdb assert in. --- zipline/test/test_messaging.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index 9a244ee7..3ab7e69b 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -104,8 +104,6 @@ class SimulatorTestCase(object): sim.register_controller( con ) sim.register_components([ret1, ret2, client]) - assert False - # Simulation # ---------- sim.simulate() From bb70efe4ebe45894d38734c3e86ea3a7c1525e60 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 24 Feb 2012 17:49:24 -0500 Subject: [PATCH 4/9] Polling work Checkpiont. --- zipline/component.py | 24 +++++++++++++++++------- zipline/messaging.py | 19 +++++++++++++------ zipline/monitor.py | 39 +++++++++++++++++---------------------- zipline/test/client.py | 5 +++-- 4 files changed, 50 insertions(+), 37 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 06a85b5b..95e887f6 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -83,6 +83,8 @@ class Component(object): self.zmq = zmq self.context = self.zmq.Context() + self.setup_poller() + self.open() self.setup_sync() self.setup_control() @@ -182,11 +184,11 @@ class Component(object): def bind_pull_socket(self, addr): pull_socket = self.context.socket(self.zmq.PULL) pull_socket.bind(addr) - poller = self.zmq.Poller() - poller.register(pull_socket, self.zmq.POLLIN) + self.poll.register(pull_socket, self.zmq.POLLIN) + self.sockets.append(pull_socket) - return pull_socket, poller + return pull_socket, self.poll def connect_push_socket(self, addr): push_socket = self.context.socket(self.zmq.PUSH) @@ -211,11 +213,17 @@ class Component(object): sub_socket.setsockopt(self.zmq.SUBSCRIBE,'') self.sockets.append(sub_socket) - poller = self.zmq.Poller() - poller.register(sub_socket, self.zmq.POLLIN) + self.poll.register(sub_socket, self.zmq.POLLIN) - # TODO: migrate tuple unpacking to be consistent - return sub_socket, poller + return sub_socket + + def setup_poller(self): + """ + Setup the poller used for multiplexing the incoming data + handling sockets. + """ + + self.poll = self.zmq.Poller() def setup_control(self): """ @@ -231,6 +239,8 @@ class Component(object): self.sync_socket = self.context.socket(self.zmq.REQ) self.sync_socket.connect(self.addresses['sync_address']) #self.sync_socket.setsockopt(self.zmq.LINGER,0) + + # Explictly, a different poller for obvious reasons. self.sync_poller = self.zmq.Poller() self.sync_poller.register(self.sync_socket, self.zmq.POLLIN) diff --git a/zipline/messaging.py b/zipline/messaging.py index c7aa1d89..ebe23f57 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -27,6 +27,7 @@ class ComponentHost(Component): self.sync_register = {} self.timeout = datetime.timedelta(seconds=5) self.gevent_needed = gevent_needed + self.heartbeat_timeout = 2000 self.feed = ParallelBuffer() self.merge = MergedParallelBuffer() @@ -86,16 +87,22 @@ class ComponentHost(Component): if len(self.components) == 0: qutil.LOGGER.info("Component register is empty.") return True + for source, last_dt in self.sync_register.iteritems(): - if((cur_time - last_dt) > self.timeout): - qutil.LOGGER.info("Time out for {source}. Current component registery: {reg}".format(source=source, reg=self.components)) + if (cur_time - last_dt) > self.timeout: + qutil.LOGGER.info( + "Time out for {source}. Current component registery: {reg}". + format(source=source, reg=self.components) + ) return True + return False def loop(self): + while not self.is_timed_out(): # wait for synchronization request - socks = dict(self.poller.poll(2000)) #timeout after 2 seconds. + socks = dict(self.poller.poll(self.heartbeat_timeout)) #timeout after 2 seconds. if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: msg = self.sync_socket.recv() @@ -154,7 +161,7 @@ class ParallelBuffer(Component): def do_work(self): # wait for synchronization reply from the host - socks = dict(self.poller.poll(2000)) #timeout after 2 seconds. + socks = dict(self.poller.poll(self.heartbeat_timeout)) #timeout after 2 seconds. if self.pull_socket in socks and socks[self.pull_socket] == self.zmq.POLLIN: message = self.pull_socket.recv() @@ -302,7 +309,7 @@ class BaseTransform(Component): Establishes zmq connections. """ #create the feed. - self.feed_socket, self.poller = self.connect_feed() + self.feed_socket = self.connect_feed() #create the result PUSH self.result_socket = self.connect_merge() @@ -313,7 +320,7 @@ class BaseTransform(Component): - call transform (subclass' method) on event - send the transformed event """ - socks = dict(self.poller.poll(2000)) #timeout after 2 seconds. + socks = dict(self.poll.poll(2000)) #timeout after 2 seconds. if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN: message = self.feed_socket.recv() if message == str(CONTROL_PROTOCOL.DONE): diff --git a/zipline/monitor.py b/zipline/monitor.py index 43ed7974..7c96db31 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -44,15 +44,26 @@ class Controller(object): except zmq.ZMQError: raise Exception('Cannot not bind on %s' % pub_socket) - def run(self, debug_step=False, stats=True): + def run(self, debug=False): self.polling = True - if self.debug or debug_step: - return self._poll_verbose(True, stats) + if debug: + return self._poll(False) else: - return self._poll(False, stats) + return self._poll_fast() + + def _poll_fast(self): + """ + C version of the polling forwarder. + """ + zmq.device(zmq.FORWARDER, self.pull, self.pub) + + def _poll(self): + """ + Python version of the polling forwarder. With logging, + mostly used for debugging. + """ - def _poll(self, debug_step, stats): while self.polling: try: self.logging.info('msg') @@ -61,23 +72,7 @@ class Controller(object): except KeyboardInterrupt: self.polling = False break - except Exception as e: - # Its common to wrap these in wildcard exceptions so - # that we don't loose messages, ever - self.logging.error(str(e)) - self.failed += 1 - continue - - def _poll_verbose(self, debug_step, stats): - while self.polling: - try: - if debug_step: - msg = self.pull.recv(copy=False) - if self.dologging: - self.logging.info(msg) - self.pub.send(msg) - self.success += 1 - except KeyboardInterrupt: + except zmq.ZMQError: self.polling = False break except Exception as e: diff --git a/zipline/test/client.py b/zipline/test/client.py index 8c3b408d..3ccf9e91 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -12,16 +12,17 @@ class TestClient(qmsg.Component): self.expected_msg_count = expected_msg_count self.utest = utest self.prev_dt = None + self.heartbeat_timeout = 2000 @property def get_id(self): return "TEST_CLIENT" def open(self): - self.data_feed, self.poller = self.connect_result() + self.data_feed = self.connect_result() def do_work(self): - socks = dict(self.poller.poll(2000)) #timeout after 2 seconds. + socks = dict(self.poll.poll(self.heartbeat_timeout)) if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN: msg = self.data_feed.recv() From 041b2311df4bd849ededa2402d72799a49a071e7 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 24 Feb 2012 18:18:22 -0500 Subject: [PATCH 5/9] Poller refactor... Checkpoint #2 --- zipline/component.py | 32 ++++++++++++++++---------------- zipline/messaging.py | 13 +++++++++---- zipline/monitor.py | 37 +++++++++++++++++++------------------ 3 files changed, 44 insertions(+), 38 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 95e887f6..5862e7e4 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -246,20 +246,20 @@ class Component(object): self.sockets.append(self.sync_socket) - def debug(self): - return ( - self.get_id , - self.huid , - socket.gethostname() , - os.getpid() , - hex(id(self)) , - ) + #def debug(self): + #return ( + #self.get_id , + #self.huid , + #socket.gethostname() , + #os.getpid() , + #hex(id(self)) , + #) - def __repr__(self): - return "<{name} {uuid} at {host} {pid} {pointer}>".format( - name = self.get_id , - uuid = self.huid , - host = socket.gethostname() , - pid = os.getpid() , - pointer = hex(id(self)) , - ) + #def __repr__(self): + #return "<{name} {uuid} at {host} {pid} {pointer}>".format( + #name = self.get_id , + #uuid = self.huid , + #host = socket.gethostname() , + #pid = os.getpid() , + #pointer = hex(id(self)) , + #) diff --git a/zipline/messaging.py b/zipline/messaging.py index ebe23f57..1613e388 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -65,15 +65,20 @@ class ComponentHost(Component): def setup_sync(self): """ - Start the sync server. """ qutil.LOGGER.debug("Connecting sync server.") self.sync_socket = self.context.socket(self.zmq.REP) self.sync_socket.bind(self.addresses['sync_address']) - self.poller = self.zmq.Poller() - self.poller.register(self.sync_socket, self.zmq.POLLIN) + # There is a namespace collision between three classes + # which use the self.poller property to mean different + # things. + # ===================================================== + self.sync_poller = self.zmq.Poller() + self.sync_poller.register(self.sync_socket, self.zmq.POLLIN) + # ===================================================== + self.sockets.append(self.sync_socket) def open(self): @@ -102,7 +107,7 @@ class ComponentHost(Component): while not self.is_timed_out(): # wait for synchronization request - socks = dict(self.poller.poll(self.heartbeat_timeout)) #timeout after 2 seconds. + socks = dict(self.sync_poller.poll(self.heartbeat_timeout)) #timeout after 2 seconds. if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: msg = self.sync_socket.recv() diff --git a/zipline/monitor.py b/zipline/monitor.py index 7c96db31..1e43ba6e 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -82,24 +82,6 @@ class Controller(object): self.failed += 1 continue - def qos(self): - return float(self.success) / (self.success + self.failed) - - def destroy(self): - """ - Manual cleanup. - """ - self.polling = False - - for asoc in self.associated: - asoc.close() - - #if self._ctx: - #self._ctx.destroy() - - def __del__(self): - self.destroy() - def message_sender(self): """ Spin off a socket used for sending messages to this @@ -121,4 +103,23 @@ class Controller(object): s.setsockopt(zmq.SUBSCRIBE, '') self.associated.append(s) return s + def destroy(self): + """ + Manual cleanup. + """ + self.polling = False + + for asoc in self.associated: + asoc.close() + + #if self._ctx: + #self._ctx.destroy() + + def __del__(self): + self.destroy() + + def qos(self): + if not self.debug: + return + return float(self.success) / (self.success + self.failed) From a63a7c6deae77bbaf15ddb1deaa4da4f89658bc8 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 24 Feb 2012 18:26:57 -0500 Subject: [PATCH 6/9] Untangled the poller knot! :) --- zipline/component.py | 34 +++++++++++++++++----------------- zipline/messaging.py | 12 ++++++------ 2 files changed, 23 insertions(+), 23 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 5862e7e4..4620e1d8 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -188,7 +188,7 @@ class Component(object): self.sockets.append(pull_socket) - return pull_socket, self.poll + return pull_socket def connect_push_socket(self, addr): push_socket = self.context.socket(self.zmq.PUSH) @@ -246,20 +246,20 @@ class Component(object): self.sockets.append(self.sync_socket) - #def debug(self): - #return ( - #self.get_id , - #self.huid , - #socket.gethostname() , - #os.getpid() , - #hex(id(self)) , - #) + def debug(self): + return ( + self.get_id , + self.huid , + socket.gethostname() , + os.getpid() , + hex(id(self)) , + ) - #def __repr__(self): - #return "<{name} {uuid} at {host} {pid} {pointer}>".format( - #name = self.get_id , - #uuid = self.huid , - #host = socket.gethostname() , - #pid = os.getpid() , - #pointer = hex(id(self)) , - #) + def __repr__(self): + return "<{name} {uuid} at {host} {pid} {pointer}>".format( + name = self.get_id , + uuid = self.huid , + host = socket.gethostname() , + pid = os.getpid() , + pointer = hex(id(self)) , + ) diff --git a/zipline/messaging.py b/zipline/messaging.py index 1613e388..afe69a9c 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -161,12 +161,12 @@ class ParallelBuffer(Component): self.data_buffer[source_id] = [] def open(self): - self.pull_socket, self.poller = self.bind_data() - self.feed_socket = self.bind_feed() + self.pull_socket = self.bind_data() + self.feed_socket = self.bind_feed() def do_work(self): # wait for synchronization reply from the host - socks = dict(self.poller.poll(self.heartbeat_timeout)) #timeout after 2 seconds. + socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds. if self.pull_socket in socks and socks[self.pull_socket] == self.zmq.POLLIN: message = self.pull_socket.recv() @@ -265,8 +265,8 @@ class MergedParallelBuffer(ParallelBuffer): ParallelBuffer.__init__(self) def open(self): - self.pull_socket, self.poller = self.bind_merge() - self.feed_socket = self.bind_result() + self.pull_socket = self.bind_merge() + self.feed_socket = self.bind_result() def next(self): """Get the next merged message from the feed buffer.""" @@ -325,7 +325,7 @@ class BaseTransform(Component): - call transform (subclass' method) on event - send the transformed event """ - socks = dict(self.poll.poll(2000)) #timeout after 2 seconds. + socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds. if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN: message = self.feed_socket.recv() if message == str(CONTROL_PROTOCOL.DONE): From 2deb6ba254b5590c7a38c98bcfa4cf7a6acd2372 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 24 Feb 2012 20:06:12 -0500 Subject: [PATCH 7/9] Register control socket in every component. --- zipline/component.py | 6 ++++++ zipline/monitor.py | 39 ++++++++++++++++----------------------- 2 files changed, 22 insertions(+), 23 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 4620e1d8..3ffdc755 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -233,6 +233,12 @@ class Component(object): """ assert self.controller + self.control_out = self.controller.message_sender() + self.control_in = self.controller.message_listener() + + self.poll.register(self.control_in, self.zmq.POLLIN) + self.sockets.extend([self.control_in, self.control_out]) + def setup_sync(self): qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id)) diff --git a/zipline/monitor.py b/zipline/monitor.py index 1e43ba6e..450fc444 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -10,6 +10,7 @@ class Controller(object): def __init__(self, pull_socket, pub_socket, context=None, logging = None): + self.associated = [] if not context: self._ctx = zmq.Context() @@ -19,11 +20,6 @@ class Controller(object): self.pull_socket = pull_socket self.pub_socket = pub_socket - self.pull = self._ctx.socket(zmq.PULL) - self.pub = self._ctx.socket(zmq.PUB) - - self.associated = [self.pull, self.pub] - if logging: self.logging = logging self.dologging = True @@ -34,23 +30,13 @@ class Controller(object): self.success = 0 self.failed = 0 - try: - self.pull.bind(pull_socket) - except zmq.ZMQError: - raise Exception('Cannot not bind on %s' % pull_socket) - - try: - self.pub.bind(pub_socket) - except zmq.ZMQError: - raise Exception('Cannot not bind on %s' % pub_socket) - def run(self, debug=False): self.polling = True - if debug: - return self._poll(False) - else: - return self._poll_fast() + #if debug: + return self._poll() + #else: + #return self._poll_fast() def _poll_fast(self): """ @@ -64,11 +50,17 @@ class Controller(object): mostly used for debugging. """ + self.pull = self._ctx.socket(zmq.PULL) + self.pub = self._ctx.socket(zmq.PUB) + + self.associated.extend([self.pull, self.pub]) + + self.pull.bind(self.pull_socket) + self.pub.bind(self.pub_socket) + while self.polling: try: - self.logging.info('msg') self.pub.send(self.pull.recv()) - #self.pub.send(self.pull.recv(copy=False)) except KeyboardInterrupt: self.polling = False break @@ -78,7 +70,8 @@ class Controller(object): except Exception as e: # Its common to wrap these in wildcard exceptions so # that we don't loose messages, ever - self.logging.error(str(e)) + if self.logging: + self.logging.error(str(e)) self.failed += 1 continue @@ -89,7 +82,6 @@ class Controller(object): """ s = self._ctx.socket(zmq.PUSH) s.connect(self.pull_socket) - s.setsockopt(zmq.LINGER, -1) self.associated.append(s) return s @@ -103,6 +95,7 @@ class Controller(object): s.setsockopt(zmq.SUBSCRIBE, '') self.associated.append(s) return s + def destroy(self): """ Manual cleanup. From c803e68f35cf2fbadd3e920f9925a7a708510b44 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 25 Feb 2012 09:14:11 -0500 Subject: [PATCH 8/9] Docs for the messaging protocol. --- zipline/protocol.py | 113 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 109 insertions(+), 4 deletions(-) diff --git a/zipline/protocol.py b/zipline/protocol.py index a40ecfe7..ae11af04 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -1,3 +1,46 @@ +""" +The messaging protocol for Zipline. + +Asserts are in place because any protocol error corresponds to a +programmer error so we want it to fail fast and in an obvious way +so it doesn't happen again. ZeroMQ follows the same philosophy. + +Notes +===== + +Msgpack +------- +Msgpack is the fastest seriaization protocol in Python at the +moment. Its 100% C is typically orders of magnitude faster than +json and pickle making it awesome for ZeroMQ. + +You can only serialize Python structural primitives: strings, +numeric types, dicts, tuples and lists. Any any recursive +combinations of these. + +Basically every basestring in Python corresponds to valid +msgpack message since the protocol is highly error tolerant. +Just keep in mind that if you ever unpack a raw msgpack string +make sure it looks like what you intend and/or catch ValueError +and TypeError exceptions. + +It also has the nice benefit of never invoking ``eval`` ( unlike +json and pickle) which is a major security boon since it is +impossible to arbitrary code for evaluation through messages. + +UltraJSON +--------- +For anything going to the browser UltraJSON is the fastest +serializer, its mostly C as well. + +The same domain of serialization as msgpack applies: Python +structural primitives. It also has the additional constraint +that anything outside of UTF8 can cause serious problems, so if +you have a strong desire to JSON encode ancient Sanskrit +( admit it, we all do ), just say no. + +""" + import msgpack #import ujson #import ultrajson_numpy @@ -14,6 +57,39 @@ def Enum(*options): _fields_ = [(o, c_ubyte) for o in options] return cstruct(*range(len(options))) +def FrameExceptionFactory(name): + """ + Exception factory with a closure around the frame class name. + """ + class InvalidFrame(Exception): + def __init__(self, got): + self.got = got + def __str__(self): + return "Invalid {framcls} Frame: {got}".format( + framecls = name, + got = self.got, + ) + +class namedict(object): + """ + So that you can use: + + foo.BAR + -- or -- + foo['BAR'] + + For more complex strcuts use collections.namedtuple: + """ + + def __init__(self, dct): + self.__dict__.update(dct) + +# ================ +# Control Protocol +# ================ + +INVALID_CONTROL_FRAME = FrameExceptionFactory('CONTROL') + CONTROL_PROTOCOL = Enum( 'INIT' , # 0 - req 'INFO' , # 1 - req @@ -26,7 +102,36 @@ CONTROL_PROTOCOL = Enum( 'EXCEPTION' , # 7 - rep ) -HEARTBEAT_PROTOCOL = { - 'REQ' : '\x01', - 'REP' : '\x02', -} +def CONTROL_FRAME(id, status): + assert isinstance(basestring, id) + assert isinstance(int, status) + + return msgpack.dumps(tuple([id, status])) + +def CONTORL_UNFRAME(msg): + assert isinstance(basestring, msg) + + try: + id, status = msgpack.loads(msg) + assert isinstance(basestring, id) + assert isinstance(int, status) + + return id, status + except TypeError: + raise INVALID_CONTROL_FRAME(msg) + except ValueError: + raise INVALID_CONTROL_FRAME(msg) + #except AssertionError: + #raise INVALID_CONTROL_FRAME(msg) + +# ================ +# Heartbeat Protocol +# ================ + +# These encode the msgpack equivelant of 1 and 2. The heartbeat +# frame should only be 1 byte on the wire. + +HEARTBEAT_PROTOCOL = namedict({ + 'REQ' : b'\x01', + 'REP' : b'\x02', +}) From f87b36bfe5ca82d6e03b68332ef713cd531aeae3 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 25 Feb 2012 09:41:34 -0500 Subject: [PATCH 9/9] Cleanup, standardizing methods, more docstrings. --- zipline/component.py | 130 ++++++++++++++++++++++++++++++------------- zipline/protocol.py | 14 ++++- 2 files changed, 102 insertions(+), 42 deletions(-) diff --git a/zipline/component.py b/zipline/component.py index 3ffdc755..b644eb5a 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -7,7 +7,7 @@ import socket import humanhash import zipline.util as qutil -from zipline.protocol import CONTROL_PROTOCOL +from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE class Component(object): @@ -39,7 +39,10 @@ class Component(object): self.gevent_needed = False self.killed = False self.heartbeat_timeout = 2000 + self.state_flag = COMPONENT_STATE.OK # OK | DONE | EXCEPTION + # Humanhashes make this way easier to debug because they + # stick in your mind unlike a 32 byte string of random hex. self.guid = uuid.uuid4() self.huid = humanhash.humanize(self.guid.hex) @@ -47,10 +50,6 @@ class Component(object): # Core Methods # ------------ - @property - def get_id(self): - raise NotImplementedError - def open(self): raise NotImplementedError @@ -71,7 +70,7 @@ class Component(object): raise NotImplementedError def _run(self): - self.done = False + self.done = False # TODO: use state flag self.sockets = [] if self.gevent_needed: @@ -95,19 +94,27 @@ class Component(object): sock.close() def run(self, catch_exceptions=False): + """ + Run the component. + + Optionally takes an argument to catch and log all exceptions raised + during execution ues this with care since it makes it very hard to + debug since it mucks up your stacktraces. + """ fail = None - # Catching all exceptions makes this really hard to - # debug, is it with care. if catch_exceptions: try: self._run() - except Exception as e: - qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id)) - fail = e + except Exception as exc: + # TODO, we want to do this grab the stack + # frame so we can preserve stacktraces when we + # reraise the exception. + self.signal_exception(exc) + fail = exc finally: - if(self.context != None): + if self.context: self.context.destroy() if fail: raise fail @@ -117,35 +124,61 @@ class Component(object): self.context.destroy() def loop(self): - while not self.done: + """ + Loop to do work while we still have work to do. + """ + while not self.done: # TODO: use state flag self.confirm() self.do_work() - # ----------- - # Messaging - # ----------- - - def signal_done(self): - #notify down stream components that we're done - if(self.out_socket != None): - self.out_socket.send(str(CONTROL_PROTOCOL.DONE)) - #notify host we're done - - # TODO: proper framing - self.sync_socket.send(self.get_id + ":" + str(CONTROL_PROTOCOL.DONE)) - - self.receive_sync_ack() - #notify internal work look that we're done - self.done = True - def confirm(self): - # send a synchronization request to the host + """ + Send a synchronization request to the host. + """ # TODO: proper framing self.sync_socket.send(self.get_id + ":RUN") self.receive_sync_ack() # blocking + # ---------------------- + # Internal Maintenance + # ---------------------- + + def signal_exception(self, exc=None): + self.state_flag = COMPONENT_STATE.EXCEPTION + qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id)) + + def signal_done(self): + """ + Notify down stream components that we're done. + """ + + self.state_flag = COMPONENT_STATE.DONE + + if self.out_socket: + self.out_socket.send(str(CONTROL_PROTOCOL.DONE)) + + #notify host we're done + # TODO: proper framing + self.sync_socket.send(self.get_id + ":" + str(CONTROL_PROTOCOL.DONE)) + + self.receive_sync_ack() + #notify internal work look that we're done + self.done = True # TODO: use state flag + + # ----------- + # Messaging + # ----------- + + def setup_poller(self): + """ + Setup the poller used for multiplexing the incoming data + handling sockets. + """ + + self.poll = self.zmq.Poller() + def receive_sync_ack(self): """ Wait for synchronization reply from the host. @@ -217,17 +250,9 @@ class Component(object): return sub_socket - def setup_poller(self): - """ - Setup the poller used for multiplexing the incoming data - handling sockets. - """ - - self.poll = self.zmq.Poller() - def setup_control(self): """ - Set up the control socket. Used to monitor the the + Set up the control socket. Used to monitor the overall status of the simulation and to forcefully tear down the simulation in case of a failure. """ @@ -240,6 +265,10 @@ class Component(object): self.sockets.extend([self.control_in, self.control_out]) def setup_sync(self): + """ + Setup the sync socket and poller. + """ + qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id)) self.sync_socket = self.context.socket(self.zmq.REQ) @@ -247,21 +276,42 @@ class Component(object): #self.sync_socket.setsockopt(self.zmq.LINGER,0) # Explictly, a different poller for obvious reasons. + # I'm not fond of having this poller init'd as a side + # effect of a method call. Still thinking about where to + # put it at the moment though... self.sync_poller = self.zmq.Poller() self.sync_poller.register(self.sync_socket, self.zmq.POLLIN) self.sockets.append(self.sync_socket) + # --------------------- + # Description and Debug + # --------------------- + + @property + def get_id(self): + return 'UNKNOWN COMPONENT' + def debug(self): + """ + Debug information about the component. + """ return ( self.get_id , self.huid , socket.gethostname() , os.getpid() , hex(id(self)) , + self.sockets , ) def __repr__(self): + """ + Return a usefull string representation of the component + to indicate its type, unique identifier, and computational + context identifier name. + """ + return "<{name} {uuid} at {host} {pid} {pointer}>".format( name = self.get_id , uuid = self.huid , diff --git a/zipline/protocol.py b/zipline/protocol.py index ae11af04..7bde3fdf 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -124,9 +124,9 @@ def CONTORL_UNFRAME(msg): #except AssertionError: #raise INVALID_CONTROL_FRAME(msg) -# ================ +# ================== # Heartbeat Protocol -# ================ +# ================== # These encode the msgpack equivelant of 1 and 2. The heartbeat # frame should only be 1 byte on the wire. @@ -135,3 +135,13 @@ HEARTBEAT_PROTOCOL = namedict({ 'REQ' : b'\x01', 'REP' : b'\x02', }) + +# ================== +# Component State +# ================== + +COMPONENT_STATE = Enum( + 'OK' , # 0 + 'DONE' , # 1 + 'EXCEPTION' , # 2 +)