Poller refactor... Checkpoint #2

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))        ,
+        #)

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()

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)