From f2042b7de60aa4a63d96414c5802f88e7766cbf9 Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 8 Feb 2012 22:54:35 -0500 Subject: [PATCH] more refactoring, tests passing again --- nosetests.xml | 2 +- qsim/messaging.py | 109 ++++++++++++++++++++++++++++++++++++ qsim/simulator/config.py | 14 ++++- qsim/simulator/feed.py | 19 ++++--- qsim/simulator/sources.py | 38 +++++++------ qsim/test/client.py | 18 +++--- qsim/transforms/core.py | 3 +- qsim/transforms/merge.py | 3 +- qsim/util.py | 113 +------------------------------------- 9 files changed, 168 insertions(+), 151 deletions(-) create mode 100644 qsim/messaging.py diff --git a/nosetests.xml b/nosetests.xml index 3841115e..ff889808 100644 --- a/nosetests.xml +++ b/nosetests.xml @@ -1 +1 @@ - \ No newline at end of file + \ No newline at end of file diff --git a/qsim/messaging.py b/qsim/messaging.py new file mode 100644 index 00000000..b2c23778 --- /dev/null +++ b/qsim/messaging.py @@ -0,0 +1,109 @@ +import json +import uuid +import zmq + +import util as qutil + +class ParallelBuffer(object): + """ holds several queues of events by key, allows retrieval in date order or by merging""" + def __init__(self, key_list): + self.out_socket = None + self.sent_count = 0 + self.received_count = 0 + self.draining = False + self.data_buffer = {} + for key in key_list: + self.data_buffer[key] = [] + + def __len__(self): + return len(self.data_buffer) + + def append(self, key, value): + self.data_buffer[key].append(value) + self.received_count += 1 + + def next(self): + if(not(self.is_full() or self.draining)): + return + + cur = None + earliest = None + for source, events in self.data_buffer.iteritems(): + if len(events) == 0: + continue + cur = events + if(earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']): + earliest = cur + + if(earliest != None): + return earliest.pop(0) + + def is_full(self): + for source, events in self.data_buffer.iteritems(): + if (len(events) == 0): + return False + return True + + def pending_messages(self): + total = 0 + for source, events in self.data_buffer.iteritems(): + total += len(events) + return total + + def drain(self): + self.draining = True + while(self.pending_messages() > 0): + self.send_next() + + def send_next(self): + if(not(self.is_full() or self.draining)): + return + + event = self.next() + if(event != None): + self.out_socket.send(json.dumps(event)) + self.sent_count += 1 + + +class MergedParallelBuffer(ParallelBuffer): + + def __init__(self, keys): + ParallelBuffer.__init__(self, keys) + self.feed = [] + self.data_buffer["feed"] = self.feed + + def next(self): + if(not(self.is_full() or self.draining)): + return + + result = self.feed.pop(0) + for source, events in self.data_buffer.iteritems(): + if(source == "feed"): + continue + if(len(events) > 0): + cur = events.pop(0) + result[source] = cur['value'] + return result + + +class FeedSync(object): + + def __init__(self, feed, name): + self.feed = feed + self.id = "{name}-{id}".format(name=name, id=uuid.uuid1()) + self.feed.register_sync(self.id) + #qutil.logger.info("registered {id} with feed".format(id=self.id)) + + def confirm(self): + context = zmq.Context() + #synchronize with feed + sync_socket = context.socket(zmq.REQ) + sync_socket.connect(self.feed.sync_address) + # send a synchronization request to the feed + sync_socket.send(self.id) + # wait for synchronization reply from the feed + sync_socket.recv() + sync_socket.close() + context.term() + qutil.logger.info("sync'd feed from {id}".format(id = self.id)) + \ No newline at end of file diff --git a/qsim/simulator/config.py b/qsim/simulator/config.py index 584e41b3..c460d60d 100644 --- a/qsim/simulator/config.py +++ b/qsim/simulator/config.py @@ -1,3 +1,4 @@ +"""Tools for managing configuration data for sources and transforms.""" import json class Config(object): @@ -6,29 +7,36 @@ class Config(object): def __init__(self, props): self.store = props - def __setitem__(self,key,value): + def __setitem__(self, key, value): self.store[key] = value def __getitem__(self, key): if self.store.has_key(key): return self.store[key] - def __getattr__(self,attrname): + def __getattr__(self, attrname): if self.store.has_key(attrname): return self.store[attrname] else: raise AttributeError("No attribute named {name}".format(name=attrname)) def get_integer(self, name, default=0): + """get the named config property as an integer""" return self.get_value(name, default, type(1)) def get_string(self, name, default=''): + """get the named config property as a string""" return self.get_value(name, default, type('')) def get_float(self, name, default=0.0): + """get the named config property as a float""" return self.get_value(name, default, type(1.0)) def get_value(self, name, default, expected_type): + """ + return the named config property as the expected_type. + if the property is missing, or is not of the right type, return default. + """ if(self.store.has_key(name)): val = self.store[name] if isinstance(val, expected_type): @@ -37,7 +45,9 @@ class Config(object): return default def to_json(self): + """convert this config to a json string""" return json.dumps(self.store) def from_json(self, json_string): + """parse a json string into this config object's properties""" self.store = json.loads(json_string) \ No newline at end of file diff --git a/qsim/simulator/feed.py b/qsim/simulator/feed.py index 04e435d6..086f151b 100644 --- a/qsim/simulator/feed.py +++ b/qsim/simulator/feed.py @@ -1,6 +1,7 @@ import qsim.simulator.sources as sources import qsim.util as qutil +import qsim.messaging as qmsg import zmq import time import logging @@ -9,7 +10,7 @@ import json class DataFeed(object): def __init__(self, config): - self.logger = qutil.logger + qutil.logger = qutil.logger self.data_address = "tcp://127.0.0.1:{port}".format(port=10101) self.sync_address = "tcp://127.0.0.1:{port}".format(port=10102) @@ -27,14 +28,14 @@ class DataFeed(object): ret = sources.RandomEquityTrades(info['sid'], self, name, info['count']) self.data_workers[name] = ret - self.data_buffer = qutil.ParallelBuffer(self.data_workers.keys()) + self.data_buffer = qmsg.ParallelBuffer(self.data_workers.keys()) def start_data_workers(self): """Start a sub-process for each datasource.""" for source_id, source in self.data_workers.iteritems(): - self.logger.info("starting {id}".format(id=source_id)) + qutil.logger.info("starting {id}".format(id=source_id)) source.start() - self.logger.info("ds processes launched") + qutil.logger.info("ds processes launched") def register_sync(self, sync_id): self.client_register[sync_id] = "UNCONFIRMED" @@ -48,7 +49,7 @@ class DataFeed(object): def sync_clients(self): # Socket to receive signals - self.logger.info("waiting for all datasources and clients to be ready") + qutil.logger.info("waiting for all datasources and clients to be ready") self.syncservice = self.context.socket(zmq.REP) self.syncservice.bind(self.sync_address) @@ -56,12 +57,12 @@ class DataFeed(object): # wait for synchronization request msg = self.syncservice.recv() self.client_register[msg] = "CONFIRMED" - #self.logger.info("confirmed {id}".format(id=msg)) + #qutil.logger.info("confirmed {id}".format(id=msg)) # send synchronization reply self.syncservice.send('CONFIRMED') self.syncservice.close() - self.logger.info("sync'd all datasources and clients") + qutil.logger.info("sync'd all datasources and clients") def run(self): # Prepare our context and sockets @@ -86,7 +87,7 @@ class DataFeed(object): #wait for all feed subscribers self.sync_clients() - self.logger.info("entering feed loop on {addr}".format(addr=self.data_address)) + qutil.logger.info("entering feed loop on {addr}".format(addr=self.data_address)) while True: message = self.data_socket.recv() @@ -105,7 +106,7 @@ class DataFeed(object): #send the DONE message self.feed_socket.send("DONE") - self.logger.info("received {n} messages, sent {m} messages".format(n=self.data_buffer.received_count, m=self.data_buffer.sent_count)) + qutil.logger.info("received {n} messages, sent {m} messages".format(n=self.data_buffer.received_count, m=self.data_buffer.sent_count)) self.data_socket.close() self.feed_socket.close() self.context.term() diff --git a/qsim/simulator/sources.py b/qsim/simulator/sources.py index 1c59240e..aa7ccc64 100644 --- a/qsim/simulator/sources.py +++ b/qsim/simulator/sources.py @@ -1,5 +1,5 @@ """ -Provides data source handlers that can push messages to a qsim.simulator.DataFeed +Provides data handlers that can push messages to a qsim.simulator.DataFeed """ import datetime import zmq @@ -8,6 +8,7 @@ import multiprocessing import random import qsim.util as qutil +import qsim.messaging as qmsg class DataSource(object): """ @@ -18,40 +19,41 @@ class DataSource(object): def __init__(self, feed, source_id): self.source_id = source_id self.feed = feed - self.sync = qutil.FeedSync(self.feed, str(source_id)) - self.data_address = self.feed.data_address - qutil.logger.info("data address is {ds}".format(ds=self.feed.data_address)) - self.cur_event = None + self.cur_event = None + self.context = None + self.data_socket = None def start(self): """Launch the datasource in a separate process.""" - self.proc = multiprocessing.Process(target=self.run) - self.proc.start() + proc = multiprocessing.Process(target=self.run) + proc.start() def open(self): """create zmq context and socket""" - qutil.logger.info("starting data source:{source_id} on {addr}" - .format(source_id=self.source_id, addr=self.feed.data_address)) + qutil.logger.info( + "starting data source:{source_id} on {addr}" + .format(source_id=self.source_id, addr=self.feed.data_address)) self.context = zmq.Context() #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 self.data_socket = self.context.socket(zmq.PUSH) - self.data_socket.connect(self.data_address) + self.data_socket.connect(self.feed.data_address) #signal we are ready - self.sync.confirm() + sync = qmsg.FeedSync(self.feed, str(self.source_id)) + sync.confirm() def run(self): """Fully execute this datasource.""" - try: - self.open() - self.send_all() - self.close() - except Exception as err: - qutil.logger.exception("Unexpected failure running datasource - {name}." - .format(name=self.source_id)) + self.open() + self.send_all() + self.close() + + def send_all(self): + """Subclasses must implement this method.""" + raise NotImplementedError() def send(self, event): """ diff --git a/qsim/test/client.py b/qsim/test/client.py index c86fd024..0efb074b 100644 --- a/qsim/test/client.py +++ b/qsim/test/client.py @@ -5,40 +5,40 @@ import logging import json import qsim.util as qutil +import qsim.messaging as qmsg class TestClient(object): def __init__(self,feed, address, bind=False): - self.logger = qutil.logger self.feed = feed self.address = address - self.sync = qutil.FeedSync(feed, "testclient") + self.sync = qmsg.FeedSync(feed, "testclient") self.bind = bind self.received_count = 0 def run(self): - self.logger.info("running the client") + qutil.logger.info("running the client") self.context = zmq.Context() self.data_feed = self.context.socket(zmq.PULL) if(self.bind): - self.logger.info("binding to {address}".format(address=self.address)) + qutil.logger.info("binding to {address}".format(address=self.address)) self.data_feed.bind(self.address) else: - self.logger.info("connecting to {address}".format(address=self.address)) + qutil.logger.info("connecting to {address}".format(address=self.address)) self.data_feed.connect(self.address) self.sync.confirm() - self.logger.info("Starting the client loop") + qutil.logger.info("Starting the client loop") prev_dt = None while True: msg = self.data_feed.recv() if(msg == "DONE"): - self.logger.info("DONE!") + qutil.logger.info("DONE!") break self.received_count += 1 event = json.loads(msg) @@ -48,8 +48,8 @@ class TestClient(object): prev_dt = event['dt'] if(self.received_count % 100 == 0): - self.logger.info("received {n} messages".format(n=self.received_count)) + qutil.logger.info("received {n} messages".format(n=self.received_count)) - self.logger.info("received {n} messages".format(n=self.received_count)) + qutil.logger.info("received {n} messages".format(n=self.received_count)) self.data_feed.close() self.context.term() \ No newline at end of file diff --git a/qsim/transforms/core.py b/qsim/transforms/core.py index 4217aa92..bd1ed398 100644 --- a/qsim/transforms/core.py +++ b/qsim/transforms/core.py @@ -19,6 +19,7 @@ all the transforms of that event into a single new message. """ import zmq import json +import qsim.messaging as qmsg import qsim.util as qutil import qsim.simulator.config as config @@ -42,7 +43,7 @@ class BaseTransform(object): self.config = config.Config(config_dict) self.state = {} self.state['name'] = self.config.name - self.sync = qutil.FeedSync(feed, self.state['name']) + self.sync = qmsg.FeedSync(feed, self.state['name']) self.received_count = 0 self.sent_count = 0 self.context = None diff --git a/qsim/transforms/merge.py b/qsim/transforms/merge.py index 2a8c9ebe..6406a9cb 100644 --- a/qsim/transforms/merge.py +++ b/qsim/transforms/merge.py @@ -5,6 +5,7 @@ import zmq import technical as ta from core import BaseTransform import qsim.util as qutil +import qsim.messaging as qmsg class MergedTransformsFeed(BaseTransform): """ Merge data feed and array of transform feeds into a single result vector. @@ -40,7 +41,7 @@ class MergedTransformsFeed(BaseTransform): keys = copy.copy(self.transforms.keys()) keys.append("feed") #for the raw feed - self.data_buffer = qutil.MergedParallelBuffer(keys) + self.data_buffer = qmsg.MergedParallelBuffer(keys) self.buffers = {} for name, transform in self.transforms.iteritems(): diff --git a/qsim/util.py b/qsim/util.py index 08c49e87..53d54572 100644 --- a/qsim/util.py +++ b/qsim/util.py @@ -1,13 +1,12 @@ """ -Small classes to assist with db access, timezone calculations, and so on. +Small classes to assist with timezone calculations, logger configuration, +and other common operations. """ import datetime import pytz -import json import logging -import uuid -import zmq + logger = logging.getLogger('QSimLogger') @@ -26,112 +25,6 @@ def format_date(dt): dt_str = dt.strftime('%Y/%m/%d-%H:%M:%S') + "." + str(dt.microsecond / 1000) return dt_str - -class ParallelBuffer(object): - """ holds several queues of events by key, allows retrieval in date order or by merging""" - def __init__(self, key_list): - self.out_socket = None - self.sent_count = 0 - self.received_count = 0 - self.draining = False - self.data_buffer = {} - for key in key_list: - self.data_buffer[key] = [] - - def __len__(self): - return len(self.data_buffer) - - def append(self, key, value): - self.data_buffer[key].append(value) - self.received_count += 1 - - def next(self): - if(not(self.is_full() or self.draining)): - return - - cur = None - earliest = None - for source, events in self.data_buffer.iteritems(): - if len(events) == 0: - continue - cur = events - if(earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']): - earliest = cur - - if(earliest != None): - return earliest.pop(0) - - def is_full(self): - for source, events in self.data_buffer.iteritems(): - if (len(events) == 0): - return False - return True - - def pending_messages(self): - total = 0 - for source, events in self.data_buffer.iteritems(): - total += len(events) - return total - - def drain(self): - self.draining = True - while(self.pending_messages() > 0): - self.send_next() - - def send_next(self): - if(not(self.is_full() or self.draining)): - return - - event = self.next() - if(event != None): - self.out_socket.send(json.dumps(event)) - self.sent_count += 1 - - -class MergedParallelBuffer(ParallelBuffer): - - def __init__(self, keys): - ParallelBuffer.__init__(self, keys) - self.feed = [] - self.data_buffer["feed"] = self.feed - - def next(self): - if(not(self.is_full() or self.draining)): - return - - result = self.feed.pop(0) - for source, events in self.data_buffer.iteritems(): - if(source == "feed"): - continue - if(len(events) > 0): - cur = events.pop(0) - result[source] = cur['value'] - return result - - -class FeedSync(object): - - def __init__(self, feed, name): - self.feed = feed - self.id = "{name}-{id}".format(name=name, id=uuid.uuid1()) - self.feed.register_sync(self.id) - self.logger = logger - #self.logger.info("registered {id} with feed".format(id=self.id)) - - def confirm(self): - context = zmq.Context() - #synchronize with feed - sync_socket = context.socket(zmq.REQ) - sync_socket.connect(self.feed.sync_address) - # send a synchronization request to the feed - sync_socket.send(self.id) - # wait for synchronization reply from the feed - sync_socket.recv() - sync_socket.close() - context.term() - self.logger.info("sync'd feed from {id}".format(id = self.id)) - - def configure_logging(loglevel=logging.DEBUG): logger.setLevel(loglevel) handler = logging.handlers.RotatingFileHandler("/tmp/{lfn}.log".format(lfn="qsim-log"), maxBytes=10*1024*1024, backupCount=5)