wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-11 15:31:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'new_world_order' of github.com:quantopian/zipline into new_world_order

Browse Source
This commit is contained in:
scottsanderson
2012-08-02 09:26:19 -04:00
parent 56177a7c4f dd1056bf30
commit bf3d9cef02
12 changed files with 458 additions and 482 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_components.py
+106
View File
@@ -0,0 +1,106 @@
import zmq
import pytz
from datetime import datetime, timedelta
from unittest2 import TestCase
from collections import defaultdict
from zipline.finance.trading import SIMULATION_STYLE
from zipline.core.devsimulator import AddressAllocator
from zipline.lines import SimulatedTrading
from zipline.utils.test_utils import (
drain_zipline,
check,
setup_logger,
teardown_logger,
launch_component,
create_monitor,
launch_monitor
)
from zipline.core import Component
from zipline.core.component import ComponentSocketArgs
from zipline.protocol import (
DATASOURCE_FRAME,
DATASOURCE_UNFRAME
)
from zipline.gens.tradegens import SpecificEquityTrades
from zipline.gens.utils import hash_args
from zipline.gens.zmqgen import gen_from_poller
import logbook
log = logbook.Logger('ComponentTestCase')
allocator = AddressAllocator(1000)
class ComponentTestCase(TestCase):
leased_sockets = defaultdict(list)
def setUp(self):
self.zipline_test_config = {
'allocator' : allocator,
'sid' : 133,
'devel' : False,
'results_socket' : allocator.lease(1)[0],
'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE
}
self.ctx = zmq.Context()
setup_logger(self)
def tearDown(self):
self.ctx.term()
teardown_logger(self)
def test_specific_equity_source(self):
filter = [1,2,3,4]
#Set up source a. One minute between events.
args_a = tuple()
kwargs_a = {
'sids' : [1,2],
'start' : datetime(2012,6,6,0,tzinfo=pytz.utc),
'delta' : timedelta(minutes = 1),
'filter' : filter,
'count' : 100
}
c_id = SpecificEquityTrades.__name__ + hash_args(args_a, kwargs_a)
mon = create_monitor(allocator)
out_socket_args = ComponentSocketArgs(
style=zmq.PUSH,
uri=allocator.lease(1)[0],
bind=True
)
c = Component(
SpecificEquityTrades,
args_a,
kwargs_a,
c_id,
out_socket_args,
DATASOURCE_FRAME,
mon
)
mon.manage(set([c.get_id]))
mon_proc = launch_monitor(mon)
# launch in a process
proc = launch_component(c)
pull_socket = self.ctx.socket(zmq.PULL)
pull_socket.connect(out_socket_args.uri)
poller = zmq.Poller()
poller.register(pull_socket, zmq.POLLIN)
unframe = DATASOURCE_UNFRAME
for msg in gen_from_poller(poller, pull_socket, unframe):
# assert things about the messages.
log.info(msg)
pull_socket.close()
log.info("DONE!")
tests/test_monitor.py
+5 -5
View File
@@ -1,7 +1,7 @@
from zipline.utils.test_utils import setup_logger, teardown_logger
from unittest2 import TestCase, skip
from zipline.core.monitor import Controller
from zipline.core.monitor import Monitor
class TestMonitor(TestCase):
def setUp(self):
@@ -15,12 +15,12 @@ class TestMonitor(TestCase):
pub_socket = 'tcp://127.0.0.1:5000'
route_socket = 'tcp://127.0.0.1:5001'
con = Controller(pub_socket, route_socket)
con.manage([])
mon = Monitor(pub_socket, route_socket)
mon.manage([])
def test_init_topology(self):
pub_socket = 'tcp://127.0.0.1:5000'
route_socket = 'tcp://127.0.0.1:5001'
con = Controller(pub_socket, route_socket, )
con.manage([ 'a', 'b', 'c', 'd' ])
mon = Monitor(pub_socket, route_socket, )
mon.manage([ 'a', 'b', 'c', 'd' ])
zipline/__init__.py
+2 -2
View File
@@ -6,14 +6,14 @@ Zipline
# it is a place to expose the public interfaces.
import protocol # namespace
from core.monitor import Controller
from core.monitor import Monitor
from lines import SimulatedTrading
from core.host import ComponentHost
from utils.protocol_utils import ndict
__all__ = [
SimulatedTrading,
Controller,
Monitor,
ComponentHost,
protocol,
ndict
zipline/core/__init__.py
+2 -2
View File
@@ -1,9 +1,9 @@
from host import ComponentHost
from component import Component
from monitor import Controller
from monitor import Monitor
__all__ = [
Component,
Controller,
Monitor,
ComponentHost
]
zipline/core/component.py
+267 -449
View File
@@ -11,10 +11,13 @@ import logbook
import traceback
import humanhash
from setproctitle import setproctitle
from collections import namedtuple
# pyzmq
import zmq
from zipline.gens.zmqgen import gen_from_poller
from zipline.core.monitor import PARAMETERS
from zipline.protocol import (
@@ -26,146 +29,85 @@ from zipline.protocol import (
EXCEPTION_FRAME
)
log = logbook.Logger('Component')
from zipline.exceptions import ComponentNoInit
log = logbook.Logger('Component')
class KillSignal(Exception):
def __init__(self):
pass
ComponentSocketArgs = namedtuple('ComponentSocket',['uri','style','bind'])
class Component(object):
"""
Base class for components. Defines the the base messaging
interface for components.
:param addresses: a dict of name_string -> zmq port address strings.
Must have the following entries
:param data_address: socket address used for data sources to stream
their records. Will be used in PUSH/PULL sockets
between data sources and a Feed. Bind will always
be on the PULL side (we always have N producers and
1 consumer)
:param feed_address: socket address used to publish consolidated feed
from serialization of data sources
will be used in PUB/SUB sockets between Feed and
Transforms. Bind is always on the PUB side.
:param merge_address: socket address used to publish transformed
values. will be used in PUSH/PULL from many
transforms to one Merge Bind will always be on
the PULL side (we always have N producers and
1 consumer)
:param results_address: socket address used to publish merged data
source feed and transforms to clients will be
used in PUB/SUB from one Merge to one or many
clients. Bind is always on the PUB side.
bind/connect methods will return the correct socket type for each
address.
"""
# ------------
# Construction
# ------------
abstract = True
#__metaclass__ = WorkflowMeta
def __init__(self,
gen_func,
gen_args,
gen_kwargs,
component_id,
out_socket_args,
frame,
monitor,
in_socket_args=None,
unframe=None
):
def __init__(self, *args, **kwargs):
self.zmq = None
self.context = None
self.addresses = None
self.waiting = None
assert component_id, \
"Every component needs a unique and invariant identifier"
assert isinstance(component_id, basestring), \
"Components must have string IDs"
assert isinstance(out_socket_args, ComponentSocketArgs), \
"out_socket_args args must be ComponentSocketArgs"
if in_socket_args:
assert isinstance(in_socket_args, ComponentSocketArgs), \
"in_socket_args args must be ComponentSocketArgs"
# -----------------
# Generator
# -----------------
self.component_id = component_id
self.gen_args = gen_args
self.gen_kwargs = gen_kwargs
self.gen_func = gen_func
self.generator = None
self.frame = frame
# lock for waiting on monitor "GO"
self.waiting = None
# -----------------
# ZMQ properties
# -----------------
self.in_socket_args = in_socket_args
self.out_socket_args = out_socket_args
self.zmq = None
self.context = None
self.out_socket = None
self.in_socket = None
self.monitor = monitor
self.unframe = unframe
self.out_socket = None
self.killed = False
self.controller = None
# timeout on heartbeat is very short to avoid burning
# cycles on heartbeating. unit is milliconds
self.heartbeat_timeout = 0
# TODO: state_flag is deprecated, remove
# TODO: error_state is deprecated, remove
self.state_flag = COMPONENT_STATE.OK
self.error_state = COMPONENT_FAILURE.NOFAILURE
self.on_done = None
self.state_flag = COMPONENT_STATE.OK
self._exception = None
self.fail_time = None
self.start_tic = None
self.stop_tic = None
self.note = None
self.confirmed = False
self.devel = False
self.socks = None
self.last_ping = None
# track time of last ping we received from monitor
self.last_ping = None
# 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)
# This is where component specific constructors should be
# defined. Arguments passed to init are threaded through.
self.init(*args, **kwargs)
def init(self):
"""
Subclasses should override this to extend the setup for the
class. Shouldn't have side effects.
"""
raise ComponentNoInit(self.__class__)
self.guid = uuid.uuid4()
self.huid = humanhash.humanize(self.guid.hex)
# ------------
# Core Methods
# ------------
def open(self):
"""
Open the connections needed to start doing work.
"""
raise NotImplementedError
def ready(self):
"""
Return ``True`` if and only if the component has finished
execution.
"""
return self.state_flag in [COMPONENT_STATE.DONE, \
COMPONENT_STATE.EXCEPTION]
def successful(self):
"""
Return ``True`` if and only if the component has finished
execution successfully, that is, without raising an error.
"""
return self.state_flag == COMPONENT_STATE.DONE and not \
self.exception
@property
def exception(self):
"""
Holds the exception that the component failed on, or ``None`` if
the component has not failed.
"""
return self._exception
def do_work(self):
raise NotImplementedError
def init_zmq(self):
self.zmq = zmq
self.context = self.zmq.Context()
self.zmq_poller = self.zmq.Poller
# The the process title so you can watch it in top
setproctitle(self.__class__.__name__)
return
def _run(self):
"""
@@ -174,33 +116,32 @@ class Component(object):
The core logic of the all components is run here.
"""
# The process title so you can watch it in top, ps.
setproctitle(self.gen_func.__name__)
log.info("Start %r" % self)
log.info("Pid %s" % os.getpid())
log.info("Group %s" % os.getpgrp())
self.start_tic = time.time()
self.done = False # TODO: use state flag
self.sockets = []
self.init_zmq()
self.setup_poller()
self.setup_control()
self.open()
self.signal_ready()
self.lock_ready()
self.wait_ready()
# -----------------------
# YOU SHALL NOT PASS!!!!!
# -----------------------
# ... until the controller signals GO
# ... until the monitor signals GO
self.loop()
for event in self.generator:
self.heartbeat()
msg = self.frame(event)
self.out_socket.send(msg)
self.stop_tic = time.time()
self.signal_done()
def run(self, catch_exceptions=True):
"""
@@ -214,119 +155,10 @@ class Component(object):
else:
# if we get a kill signal, forcibly close all the
# sockets.
# exc_info = sys.exc_info()
# self.relay_exception(exc_info[0], exc_info[1], exc_info[2])
self.teardown_sockets()
finally:
self.shutdown()
log.info("Exiting %r" % self)
def working(self):
"""
Controls when the work loop will start and end
If we encounter an exception or signal done exit.
Overload for higher order behavior.
"""
return (not self.done)
def loop(self, lockstep=True):
"""
Loop to do work while we still have work to do.
"""
while self.working():
self.heartbeat()
self.do_work()
def runtime(self):
if self.ready() and self.start_tic and self.stop_tic:
return self.stop_tic - self.start_tic
def heartbeat(self, timeout=0):
# wait for synchronization reply from the host
self.socks = dict(self.poll.poll(timeout))
# ----------------
# Control Dispatch
# ----------------
assert self.control_in, 'Component does not have a control_in socket'
# If we're in devel mode drop out because the controller
# isn't guaranteed to be around anymore
if self.devel:
log.warn("Skipping heartbeat because of devel flag")
return
if self.socks.get(self.control_in) == zmq.POLLIN:
msg = self.control_in.recv()
event, payload = CONTROL_UNFRAME(msg)
# ===========
# Heartbeat
# ===========
# The controller will send out a single number packed in
# a CONTROL_FRAME with ``heartbeat`` event every
# (n)-seconds. The component then has n seconds to
# respond to it. If not then it will be considered as
# malfunctioning or maybe CPU bound.
if event == CONTROL_PROTOCOL.HEARTBEAT:
# Heart outgoing
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.OK,
payload
)
self.last_ping = float(payload)
# Echo back the heartbeat identifier to tell the
# controller that this component is still alive and
# doing work
self.control_out.send(heartbeat_frame)
# =========
# Soft Kill
# =========
# Try and clean up properly and send out any reports or
# data that are done during a clean shutdown. Inform the
# controller that we're done.
elif event == CONTROL_PROTOCOL.SHUTDOWN:
self.signal_done()
# =========
# Hard Kill
# =========
# Just exit.
elif event == CONTROL_PROTOCOL.KILL:
self.kill()
# In case we didn't receive a ping, send a pre-emptive
# pong to the monitor.
elif hasattr(self, 'control_out') and \
self.last_ping and \
time.time() - self.last_ping > 1:
# send a ping ahead of schedule
pre_pong = time.time()
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.OK,
str(pre_pong)
)
# Echo back the heartbeat identifier to tell the
# controller that this component is still alive and
# doing work
self.control_out.send(heartbeat_frame, self.zmq.NOBLOCK)
self.last_ping = pre_pong
elif self.last_ping and \
time.time() - self.last_ping > PARAMETERS.MAX_COMPONENT_WAIT:
# monitor is gone without sending the shutdown
# signal, do a hard exit.
self.kill()
# ----------------------------
# Cleanup & Modes of Failure
@@ -342,16 +174,6 @@ class Component(object):
for sock in self.sockets:
sock.close()
def shutdown(self):
"""
Clean shutdown.
Tear down after normal operation.
"""
if self.on_done:
log.warn("{id} calling done.".format(id=self.get_id))
self.on_done()
def kill(self):
"""
Unclean shutdown.
@@ -359,9 +181,64 @@ class Component(object):
Tear down ( fast ) as a mode of failure in the simulation or on
service halt.
"""
# sys.exit(1)
raise KillSignal()
def signal_exception(self, exc=None, scope=None):
"""
All exceptions inside any component should boil back to
this handler.
Will inform the system that the component has failed and how it
has failed.
"""
self.state_flag = COMPONENT_STATE.EXCEPTION
exc_type, exc_value, exc_traceback = sys.exc_info()
# if a downstream component fails, this component may try
# sending when there are zero connections to the socket,
# which will raise ZMQError(EAGAIN). So, it doesn't make
# sense to relay this exception to Monitor and the rest
# of the zipline.
if isinstance(exc, zmq.ZMQError) and exc.errno == zmq.EAGAIN:
log.warn("{id} raised a ZMQError(EAGAIN) not relaying"\
.format(id=self.get_id))
return
# sys.stdout.write(trace)
log.exception("Unexpected error in run for {id}.".format(id=self.get_id))
try:
log.info('{id} sending exception to monitor'\
.format(id=self.get_id))
msg = EXCEPTION_FRAME(
exc_traceback,
exc_type.__name__,
exc_value.message
)
exception_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.EXCEPTION,
msg
)
self.control_out.send(exception_frame, self.zmq.NOBLOCK)
# The monitor should relay the exception back
# to all zipline components. Wait here until the
# notice arrives, and we can assume other zipline
# components have broken out of their message
# loops.
for i in xrange(PARAMETERS.MAX_COMPONENT_WAIT):
self.heartbeat(timeout=1000)
log.warn("{id} never heard back from monitor."\
.format(id=self.get_id))
except KillSignal:
log.info("{id} received confirmation from monitor"\
.format(id=self.get_id))
except:
log.exception("Exception waiting for monitor reply")
# ----------------------
# Internal Maintenance
# ----------------------
@@ -405,7 +282,7 @@ class Component(object):
# Go
# ====
# A distributed lock from the controller to ensure
# A distributed lock from the monitor to ensure
# synchronized start.
if event == CONTROL_PROTOCOL.HEARTBEAT:
@@ -417,7 +294,7 @@ class Component(object):
log.info('Prestart Heartbeat ' + self.get_id)
elif event == CONTROL_PROTOCOL.GO:
# Side effectful call from the controller to unlock
# Side effectful call from the monitor to unlock
# and begin doing work only when the entire topology
# of the system beings to come online
log.info('Unlocking ' + self.__class__.__name__)
@@ -429,7 +306,7 @@ class Component(object):
# Try and clean up properly and send out any reports or
# data that are done during a clean shutdown. Inform the
# controller that we're done.
# monitor that we're done.
elif event == CONTROL_PROTOCOL.SHUTDOWN:
self.signal_done()
break
@@ -449,97 +326,90 @@ class Component(object):
self.kill()
break
def heartbeat(self, timeout=0):
# wait for synchronization reply from the host
socks = dict(self.poll.poll(timeout))
# ----------------
# Control Dispatch
# ----------------
assert self.control_in, 'Component does not have a control_in socket'
if socks.get(self.control_in) == zmq.POLLIN:
msg = self.control_in.recv()
event, payload = CONTROL_UNFRAME(msg)
# ===========
# Heartbeat
# ===========
# The monitor will send out a single number packed in
# a CONTROL_FRAME with ``heartbeat`` event every
# (n)-seconds. The component then has n seconds to
# respond to it. If not then it will be considered as
# malfunctioning or maybe CPU bound.
if event == CONTROL_PROTOCOL.HEARTBEAT:
# Heart outgoing
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.OK,
payload
)
self.last_ping = float(payload)
# Echo back the heartbeat identifier to tell the
# monitor that this component is still alive and
# doing work
self.control_out.send(heartbeat_frame)
# =========
# Soft Kill
# =========
# Try and clean up properly and send out any reports or
# data that are done during a clean shutdown. Inform the
# monitor that we're done.
elif event == CONTROL_PROTOCOL.SHUTDOWN:
self.signal_done()
# =========
# Hard Kill
# =========
# Just exit.
elif event == CONTROL_PROTOCOL.KILL:
self.kill()
# In case we didn't receive a ping, send a pre-emptive
# pong to the monitor.
elif self.last_ping and time.time() - self.last_ping > 1:
# send a ping ahead of schedule
pre_pong = time.time()
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.OK,
str(pre_pong)
)
# Echo back the heartbeat identifier to tell the
# monitor that this component is still alive and
# doing work
self.control_out.send(heartbeat_frame, self.zmq.NOBLOCK)
self.last_ping = pre_pong
elif self.last_ping and \
time.time() - self.last_ping > PARAMETERS.MAX_COMPONENT_WAIT:
# monitor is gone without sending the shutdown
# signal, do a hard exit.
self.kill()
def signal_ready(self):
log.info(self.__class__.__name__ + ' is ready')
if hasattr(self, 'control_out'):
frame = CONTROL_FRAME(
CONTROL_PROTOCOL.READY,
''
)
self.control_out.send(frame)
def signal_cancel(self):
self.done = True
# TODO: no hasattr hacks
#if not self.controller:
if hasattr(self, 'control_out'):
frame = CONTROL_FRAME(
CONTROL_PROTOCOL.SHUTDOWN,
None
)
self.control_out.send(frame)
# then proceeds to do shutdown(), and teardown_sockets()
# to complete the process
def signal_exception(self, exc=None, scope=None):
"""
All exceptions inside any component should boil back to
this handler.
Will inform the system that the component has failed and how it
has failed.
"""
if scope == 'algo':
self.error_state = COMPONENT_FAILURE.ALGOEXCEPT
else:
self.error_state = COMPONENT_FAILURE.HOSTEXCEPT
self.state_flag = COMPONENT_STATE.EXCEPTION
# mark the time of failure so we can track the failure
# progogation through the system.
self.stop_tic = time.time()
self._exception = exc
exc_type, exc_value, exc_traceback = sys.exc_info()
# if a downstream component fails, this component may try
# sending when there are zero connections to the socket,
# which will raise ZMQError(EAGAIN). So, it doesn't make
# sense to relay this exception to Monitor and the rest
# of the zipline.
if isinstance(exc, zmq.ZMQError) and exc.errno == zmq.EAGAIN:
log.warn("{id} raised a ZMQError(EAGAIN) not relaying"\
.format(id=self.get_id))
return
# sys.stdout.write(trace)
log.exception("Unexpected error in run for {id}.".format(id=self.get_id))
if hasattr(self, 'control_out') and self.control_out:
try:
log.info('{id} sending exception to controller'\
.format(id=self.get_id))
msg = EXCEPTION_FRAME(
exc_traceback,
exc_type.__name__,
exc_value.message
)
exception_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.EXCEPTION,
msg
)
self.control_out.send(exception_frame, self.zmq.NOBLOCK)
# The controller should relay the exception back
# to all zipline components. Wait here until the
# notice arrives, and we can assume other zipline
# components have broken out of their message
# loops.
for i in xrange(PARAMETERS.MAX_COMPONENT_WAIT):
self.heartbeat(timeout=1000)
log.warn("{id} never heard back from monitor."\
.format(id=self.get_id))
except KillSignal:
log.info("{id} received confirmation from controller"\
.format(id=self.get_id))
except:
log.exception("Exception waiting for controller reply")
frame = CONTROL_FRAME(
CONTROL_PROTOCOL.READY,
''
)
self.control_out.send(frame)
def signal_done(self):
"""
@@ -550,20 +420,18 @@ class Component(object):
# notify internal work loop that we're done
self.done = True # TODO: use state flag
if hasattr(self, 'out_socket') and self.out_socket:
msg = zmq.Message(str(CONTROL_PROTOCOL.DONE))
self.out_socket.send(msg)
msg = zmq.Message(str(CONTROL_PROTOCOL.DONE))
self.out_socket.send(msg)
if hasattr(self, 'control_out'):
# notify controller we're done
done_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.DONE,
''
)
# notify monitor we're done
done_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.DONE,
''
)
self.control_out.send(done_frame)
log.info("[%s] sent control done" % self.get_id)
self.control_out.send(done_frame)
log.info("[%s] sent control done" % self.get_id)
# there is a narrow race condition where we finish just
# after the Monitor accepts our prior heartbeat, but just
@@ -571,43 +439,64 @@ class Component(object):
# last heartbeat, and wait an unusually long time.
self.heartbeat(timeout=5000)
# -----------
# Messaging
# -----------
def setup_poller(self):
def open(self):
"""
Setup the poller used for multiplexing the incoming data
handling sockets.
Open the connections needed to start doing work.
Perform any setup that must be done within process.
"""
self.poll = self.zmq_poller()
def bind_data(self):
return self.bind_pull_socket(self.addresses['data_address'])
self.zmq = zmq
self.context = self.zmq.Context()
self.poll = self.zmq.Poller()
def connect_data(self):
return self.connect_push_socket(self.addresses['data_address'])
self.setup_control()
def bind_feed(self):
return self.bind_pub_socket(self.addresses['feed_address'])
if self.in_socket_args:
self.in_socket = self.open_socket(self.in_socket_args)
poller_gen = gen_from_poller(
self.poller,
self.in_socket,
self.unframe
)
self.generator = self.gen_func(
poller_gen,
*self.gen_args,
**self.gen_kwargs
)
else:
self.generator = self.gen_func(*self.gen_args, **self.gen_kwargs)
def connect_feed(self):
return self.connect_sub_socket(self.addresses['feed_address'])
self.out_socket = self.open_socket(self.out_socket_args)
def bind_merge(self):
return self.bind_pull_socket(self.addresses['merge_address'])
def open_socket(self, sock_args):
if sock_args.bind:
return self.bind_socket(sock_args)
else:
return self.connect_socket(sock_args)
def connect_merge(self):
return self.connect_push_socket(self.addresses['merge_address'])
def bind_socket(self, sock_args):
if sock_args.style == zmq.PULL:
return self.bind_pull_socket(sock_args.uri)
if sock_args.style == zmq.PUSH:
return self.bind_push_socket(sock_args.uri)
if sock_args.style == zmq.PUB:
return self.bind_pub_socket(sock_args.uri)
def bind_result(self):
return self.bind_push_socket(self.addresses['results_address'])
raise Exception("Invalid socket arguments")
def connect_result(self):
return self.connect_pull_socket(self.addresses['results_address'])
def connect_socket(self, sock_args):
if sock_args.style == zmq.PULL:
return self.connect_pull_socket(sock_args.uri)
if sock_args.style == zmq.PUSH:
return self.connect_push_socket(sock_args.uri)
if sock_args.style == zmq.SUB:
return self.connect_sub_socket(sock_args.uri)
raise Exception("Invalid socket arguments")
def bind_push_socket(self, addr):
push_socket = self.context.socket(self.zmq.PUSH)
@@ -638,7 +527,6 @@ class Component(object):
def connect_push_socket(self, addr):
push_socket = self.context.socket(self.zmq.PUSH)
push_socket.connect(addr)
#push_socket.setsockopt(self.zmq.LINGER,0)
self.sockets.append(push_socket)
self.out_socket = push_socket
@@ -647,7 +535,6 @@ class Component(object):
def bind_pub_socket(self, addr):
pub_socket = self.context.socket(self.zmq.PUB)
pub_socket.bind(addr)
#pub_socket.setsockopt(self.zmq.LINGER, 0)
self.out_socket = pub_socket
return pub_socket
@@ -668,92 +555,29 @@ class Component(object):
of the simulation and to forcefully tear down the simulation in
case of a failure.
"""
# Allow for the possibility of not having a controller,
# possibly the zipline devsimulator may not want this.
if not self.controller:
return
self.control_out = self.controller.message_sender(
self.control_out = self.monitor.message_sender(
identity = self.get_id,
context = self.context,
)
self.control_in = self.controller.message_listener(
self.control_in = self.monitor.message_listener(
context = self.context
)
self.poll.register(self.control_in, self.zmq.POLLIN)
self.sockets.extend([self.control_in, self.control_out])
# -----------
# FSM Actions
# -----------
#@property
#def state(self):
#if not hasattr(self, '_state'):
#self._state = self.initial_state
#else:
#return self._state
#@state.setter
#def state(self, new):
#if not hasattr(self, '_state'):
#self._state = self.initial_state
#old = self._state
#if (old, new) in self.workflow:
#self._state = new
#else:
#raise RuntimeError("Invalid State Transition : %s -> %s" %(old, new))
# ---------------------
# Description and Debug
# ---------------------
def extern_logger(self):
"""
Pipe logs out to a provided logging interface.
"""
pass
def setup_extern_logger(self):
"""
Pipe logs out to a provided logging interface.
"""
pass
@property
def get_id(self):
"""
The descriptive name of the component.
The time invariant name for this component.
Must be unique within this zipline.
"""
# Prevents the bug that Thomas ran into
raise NotImplementedError
@property
def get_type(self):
"""
The data flow type of the component.
- ``SOURCE``
- ``CONDUIT``
- ``SINK``
"""
raise NotImplementedError
@property
def get_pure(self):
"""
Describes whehter this component purely functional, i.e. for a
given set of inputs is it guaranteed to always give the same
output . Components that are side-effectful are, generally, not
pure.
"""
return False
return self.component_id
def debug(self):
"""
@@ -766,18 +590,12 @@ class Component(object):
'pid' : os.getpid() ,
'memaddress' : hex(id(self)) ,
'ready' : self.successful() ,
'succesfull' : self.ready() ,
'successful' : self.ready() ,
}
def __len__(self):
"""
Some components overload this for debug purposes
"""
raise NotImplementedError
def __repr__(self):
"""
Return a usefull string representation of the component to
Return a useful string representation of the component to
indicate its type, unique identifier, and computational context
identifier name.
"""
zipline/core/host.py
+1 -1
View File
@@ -103,7 +103,7 @@ class ComponentHost(object):
log.info('== Roll Call ==')
log.info('Controller')
log.info('Monitor')
self.launch_controller()
zipline/core/monitor.py
+2 -9
View File
@@ -38,7 +38,7 @@ class UnknownChatter(Exception):
return """Component calling itself "%s" talking on unexpected channel""" % self.named
log = logbook.Logger('Controller')
log = logbook.Logger('Monitor')
# The scalars determining the timing of the monitor behavior for
# the system.
@@ -56,7 +56,7 @@ PARAMETERS = ndict(dict(
SYSTEM_TIMEOUT = 50,
))
class Controller(object):
class Monitor(object):
"""
A N to M messaging system for inter component communication.
@@ -103,8 +103,6 @@ class Controller(object):
self.route_socket = route_socket
self.exception_socket = exception_socket
self.error_replay = OrderedDict()
self.missed_beats = Counter()
self.send_sighup = send_sighup
@@ -499,7 +497,6 @@ class Controller(object):
# Error Handling
# --------------
def exception(self, component, exception_data):
self.error_replay[(component, time.time())] = exception_data
log.error('Component in exception state: %s. Shutting down system and sending exception data to listeners.'\
% component)
# Send the exception message out to listeners.
@@ -616,10 +613,6 @@ class Controller(object):
self.associated.append(s)
return s
def do_error_replay(self):
for (component, time), error in self.error_replay.iteritems():
log.info('Component Log for -- %s --:\n%s' % (component, error))
def kill(self):
"""Aggressively exit the whole zipline.
"""
zipline/core/process.py
+4 -4
View File
@@ -40,13 +40,13 @@ class ProcessSimulator(ComponentHost):
# invoked by the host's open()
def launch_controller(self):
proc = multiprocessing.Process(target=self.controller.run)
proc = multiprocessing.Process(target=self.monitor.run)
proc.start()
self.con = proc
# Process specific
self.controller_process = proc
self.mapping[proc.pid] = 'Controller'
self.monitor_process = proc
self.mapping[proc.pid] = 'Monitor'
def launch_component(self, component):
proc = multiprocessing.Process(target=component.run)
@@ -81,7 +81,7 @@ class ProcessSimulator(ComponentHost):
process.join(timeout=1)
process.terminate()
self.controller.shutdown(soft=True)
self.monitor.shutdown(soft=True)
self.running = False
self.con.terminate()
zipline/gens/utils.py
+3 -3
View File
@@ -19,8 +19,8 @@ def mock_raw_event(sid, dt):
def mock_done(id):
return ndict({
'dt' : "DONE",
"source_id" : id,
'dt' : "DONE",
"source_id" : id,
'tnfm_id' : id,
'tnfm_value': None,
'type' : 0
@@ -43,7 +43,7 @@ def roundrobin(sources, namestrings):
"""
assert len(sources) == len(namestrings)
mapping = OrderedDict(zip(namestrings, sources))
# While our generators have not been exhausted, pull elements
while mapping.keys() != []:
for namestring, source in mapping.iteritems():
zipline/gens/zmqgen.py
+27
View File
@@ -0,0 +1,27 @@
import zmq
import zipline.protocol as zp
def gen_from_pull_socket(socket_uri, context, unframe):
"""
A generator that takes a socket_uri, and yields
messages from the poller until it gets a zp.CONTROL_PROTOCOL.DONE.
"""
pull_socket = context.socket(zmq.PULL)
pull_socket.connect(socket_uri)
poller = zmq.Poller()
poller.register(pull_socket, zmq.POLLIN)
return gen_from_poller(poller, pull_socket, unframe)
def gen_from_poller(poller, in_socket, unframe):
while True:
socks = dict(poller.poll(1000))
if socks.get(in_socket) == zmq.POLLIN:
message = in_socket.recv()
if message == str(zp.CONTROL_PROTOCOL.DONE):
break
else:
event = unframe(message)
yield event
zipline/lines.py
+7 -7
View File
@@ -70,7 +70,7 @@ from zipline.transforms import BaseTransform
from zipline.test_algorithms import TestAlgorithm
from zipline.components import TradeSimulationClient
from zipline.core.process import ProcessSimulator
from zipline.core.monitor import Controller
from zipline.core.monitor import Monitor
from zipline.finance.trading import SIMULATION_STYLE
log = logbook.Logger('Lines')
@@ -131,7 +131,7 @@ class SimulatedTrading(object):
'results_address' : sockets[4],
}
self.con = Controller(
self.monitor = Monitor(
# pub socket
sockets[5],
# route socket
@@ -163,7 +163,7 @@ class SimulatedTrading(object):
#setup transforms
self.transforms = {}
self.sim.register_controller( self.con )
self.sim.register_monitor( self.monitor )
@staticmethod
@@ -348,15 +348,15 @@ class SimulatedTrading(object):
return base | transforms | sources
def setup_controller(self):
def setup_monitor(self):
"""
Prepare the controller to manage the topology specified
Prepare the monitor to manage the topology specified
by this line.
"""
self.con.manage(self.topology)
self.monitor.manage(self.topology)
def simulate(self, blocking=True):
self.setup_controller()
self.setup_monitor()
self.started = True
self.sim_context = self.sim.simulate()
zipline/utils/test_utils.py
+32
View File
@@ -1,3 +1,4 @@
import multiprocessing
import zmq
import time
import zipline.protocol as zp
@@ -6,6 +7,7 @@ import blist
from zipline.utils.date_utils import EPOCH
from itertools import izip
from logbook import FileHandler
from zipline.core.monitor import Monitor
def setup_logger(test, path='/var/log/zipline/zipline.log'):
test.log_handler = FileHandler(path)
@@ -140,3 +142,33 @@ def assert_single_position(test, zipline):
sid,
"Portfolio should have one position in " + str(sid)
)
def launch_component(component):
proc = multiprocessing.Process(target=component.run)
proc.start()
return proc
def launch_monitor(monitor):
proc = multiprocessing.Process(target=monitor.run)
proc.start()
return proc
def create_monitor(allocator):
sockets = allocator.lease(3)
mon = Monitor(
# pub socket
sockets[0],
# route socket
sockets[1],
# exception socket to match tradesimclient's result
# socket, because we want to relay exceptions to the
# same listener
sockets[2],
# this controller is expected to run in a test, so no
# need to signal the parent process on success or error.
send_sighup=False
)
return mon