From 0f8ebc97dcd917f415547a1f0ce7db42a1b75795 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 19 Jan 2012 15:59:57 -0500 Subject: [PATCH] merge working --- backtest/util.py | 20 +++++++++++++----- data/transforms.py | 51 ++++++++++++++++++++++++++++++---------------- qbt_client.py | 1 - 3 files changed, 48 insertions(+), 24 deletions(-) diff --git a/backtest/util.py b/backtest/util.py index 8d874238..f4ec3652 100644 --- a/backtest/util.py +++ b/backtest/util.py @@ -67,7 +67,10 @@ class ParallelBuffer(object): 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 @@ -117,15 +120,22 @@ class ParallelBuffer(object): 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 = {} + + result = self.feed.pop(0) for source, events in self.data_buffer.iteritems(): + if(source == "feed"): + continue if(len(events) > 0): - result[source] = events.pop(0) - + cur = events.pop(0) + result[source] = cur['value'] return result diff --git a/data/transforms.py b/data/transforms.py index da509440..78e160cf 100644 --- a/data/transforms.py +++ b/data/transforms.py @@ -55,12 +55,13 @@ class Transform(object): while True: message = self.feed_socket.recv() - self.received_count += 1 if(message == "DONE"): + self.logger.info("{name} received the Done message from the feed".format(name=self.name)) self.result_socket.send("DONE") break; + self.received_count += 1 event = json.loads(message) - cur_state = self.update(event) + cur_state = self.transform(event) cur_state['dt'] = event['dt'] cur_state['name'] = self.name self.result_socket.send(json.dumps(cur_state)) @@ -73,13 +74,13 @@ class Transform(object): self.result_socket.close() self.context.term() - def update(self, event): + def transform(self, event): return {} class MovingAverage(Transform): - def __init__(self, feed, props, result_address="tcp://127.0.0.1:20202"): + def __init__(self, feed, props, result_address): Transform.__init__(self, feed, props, result_address) self.events = [] @@ -94,7 +95,7 @@ class MovingAverage(Transform): - def update(self, event): + def transform(self, event): self.events.append(event) #filter the event list to the window length. @@ -109,7 +110,7 @@ class MovingAverage(Transform): self.average = total/len(self.events) - self.state['avg'] = self.average + self.state['value'] = self.average return self.state @@ -143,11 +144,6 @@ class MergedTransformsFeed(Transform): keys = copy.copy(self.transforms.keys()) keys.append("feed") #for the raw feed self.data_buffer = MergedParallelBuffer(keys) - - for name, transform in self.transforms.iteritems(): - self.logger.info("starting {name}".format(name=name)) - proc = multiprocessing.Process(target=transform.run) - proc.start() self.buffers = {} for name, transform in self.transforms.iteritems(): @@ -176,36 +172,55 @@ class MergedTransformsFeed(Transform): self.poller.register(self.feed_socket, zmq.POLLIN) self.poller.register(self.transform_socket, zmq.POLLIN) + for name, transform in self.transforms.iteritems(): + self.logger.info("starting {name}".format(name=name)) + proc = multiprocessing.Process(target=transform.run) + proc.start() + + self.sync.confirm() + def close(self): self.transform_socket.close() Transform.close(self) def process_all(self): - self.sync.confirm() + done_count = 0 while True: socks = dict(self.poller.poll()) if self.feed_socket in socks and socks[self.feed_socket] == zmq.POLLIN: message = self.feed_socket.recv() - self.received_count += 1 - if(message != "DONE"): + if(message == "DONE"): + self.logger.info("finished receiving feed to merge") + done_count += 1 + else: + self.received_count += 1 event = json.loads(message) self.data_buffer.append("feed",event) - + if self.transform_socket in socks and socks[self.transform_socket] == zmq.POLLIN: t_message = self.transform_socket.recv() - if(message != "DONE"): + if(t_message == "DONE"): + self.logger.info("finished receiving a transform to merge") + done_count += 1 + else: + self.received_count += 1 t_event = json.loads(t_message) self.data_buffer.append(t_event['name'], t_event) - + if(done_count >= len(self.data_buffer)): + break #done! + self.data_buffer.send_next() + + self.logger.info("Transform {name} received {r} and sent {s}".format(name=self.name, r=self.data_buffer.received_count, s=self.data_buffer.sent_count)) + self.logger.info("about to drain {n} messages from merger's buffer".format(n=self.data_buffer.pending_messages())) #drain any remaining messages in the buffer self.data_buffer.drain() #signal to client that we're done self.result_socket.send("DONE") - self.logger.info("Transform {name} recieved {r} and sent {s}".format(name=self.name, r=self.data_buffer.received_count, s=self.data_buffer.sent_count)) + self.logger.info("Transform {name} received {r} and sent {s}".format(name=self.name, r=self.data_buffer.received_count, s=self.data_buffer.sent_count)) \ No newline at end of file diff --git a/qbt_client.py b/qbt_client.py index 097dfaa9..eda31fb6 100644 --- a/qbt_client.py +++ b/qbt_client.py @@ -41,7 +41,6 @@ class TestClient(object): self.logger.info("DONE!") break self.received_count += 1 - self.logger.info(msg) event = json.loads(msg) if(prev_dt != None): if(not event['dt'] >= prev_dt):