Remove the 'queue' parameter in MTClientConnection.ask by giving the thread safe

data manager at MTClientConnection's constructor. This allows keep the same
method signature for ask() and to not crash when the client try to send a ping.


git-svn-id: https://svn.erp5.org/repos/neo/branches/prototype3@864 71dcc9de-d417-0410-9af5-da40c76e7ee4

neo/client/app.py

@@ -73,9 +73,8 @@ class ConnectionPool(object):
         while True:
             logging.info('trying to connect to %s - %s', node, node.getState())
             app.setNodeReady()
-            conn = MTClientConnection(app.em, app.storage_event_handler, addr,
-                                      connector_handler=app.connector_handler,
-                                      dispatcher=app.dispatcher)
+            conn = MTClientConnection(self.app.local_var, app.em, app.storage_event_handler, 
+                    addr, connector_handler=app.connector_handler, dispatcher=app.dispatcher)
             conn.lock()
             try:
                 if conn.getConnector() is None:
@@ -85,7 +84,7 @@ class ConnectionPool(object):
 
                 p = protocol.requestNodeIdentification(CLIENT_NODE_TYPE,
                             app.uuid, '0.0.0.0', 0, app.name)
-                msg_id = conn.ask(app.local_var.queue, p)
+                msg_id = conn.ask(p)
             finally:
                 conn.unlock()
 
@@ -341,8 +340,7 @@ class Application(object):
     def _askStorage(self, conn, packet, timeout=5, additional_timeout=30):
         """ Send a request to a storage node and process it's answer """
         try:
-            msg_id = conn.ask(self.local_var.queue, packet, timeout,
-                              additional_timeout)
+            msg_id = conn.ask(packet, timeout, additional_timeout)
         finally:
             # assume that the connection was already locked
             conn.unlock()
@@ -353,8 +351,7 @@ class Application(object):
         conn = self._getMasterConnection()
         conn.lock()
         try:
-            msg_id = conn.ask(self.local_var.queue, packet, timeout,
-                              additional_timeout)
+            msg_id = conn.ask(packet, timeout, additional_timeout)
         finally:
             conn.unlock()
         self._waitMessage(conn, msg_id, self.primary_handler)
@@ -405,7 +402,7 @@ class Application(object):
                         self.trying_master_node = master_list[0]
                     index += 1
                 # Connect to master
-                conn = MTClientConnection(self.em, self.notifications_handler,
+                conn = MTClientConnection(self.local_var, self.em, self.notifications_handler,
                                           addr=self.trying_master_node.getServer(),
                                           connector_handler=self.connector_handler,
                                           dispatcher=self.dispatcher)
@@ -417,7 +414,7 @@ class Application(object):
                         logging.error('Connection to master node %s failed',
                                       self.trying_master_node)
                         continue
-                    msg_id = conn.ask(self.local_var.queue, protocol.askPrimaryMaster())
+                    msg_id = conn.ask(protocol.askPrimaryMaster())
                 finally:
                     conn.unlock()
                 try:
@@ -440,7 +437,7 @@ class Application(object):
                         break
                     p = protocol.requestNodeIdentification(CLIENT_NODE_TYPE,
                             self.uuid, '0.0.0.0', 0, self.name)
-                    msg_id = conn.ask(self.local_var.queue, p)
+                    msg_id = conn.ask(p)
                 finally:
                     conn.unlock()
                 try:
@@ -466,15 +463,13 @@ class Application(object):
                 # wait on one message at a time
                 conn.lock()
                 try:
-                    msg_id = conn.ask(self.local_var.queue,
-                                      protocol.askPartitionTable([]))
+                    msg_id = conn.ask(protocol.askPartitionTable([]))
                 finally:
                     conn.unlock()
                 self._waitMessage(conn, msg_id, handler=self.primary_bootstrap_handler)
                 conn.lock()
                 try:
-                    msg_id = conn.ask(self.local_var.queue,
-                                      protocol.askNodeInformation())
+                    msg_id = conn.ask(protocol.askNodeInformation())
                 finally:
                     conn.unlock()
                 self._waitMessage(conn, msg_id, handler=self.primary_bootstrap_handler)
@@ -881,8 +876,7 @@ class Application(object):
                 continue
 
             try:
-                conn.ask(self.local_var.queue,
-                         protocol.askTIDs(first, last, INVALID_PARTITION))
+                conn.ask(protocol.askTIDs(first, last, INVALID_PARTITION))
             finally:
                 conn.unlock()
 

neo/connection.py

@@ -462,12 +462,14 @@ class ServerConnection(Connection):
 
 class MTClientConnection(ClientConnection):
     """A Multithread-safe version of ClientConnection."""
-    def __init__(self, *args, **kwargs):
+
+    def __init__(self, local_var, *args, **kwargs):
         # _lock is only here for lock debugging purposes. Do not use.
         self._lock = lock = RLock()
         self.acquire = lock.acquire
         self.release = lock.release
         self.dispatcher = kwargs.pop('dispatcher')
+        self.local_var = local_var
         self.lock()
         try:
             super(MTClientConnection, self).__init__(*args, **kwargs)
@@ -501,10 +503,10 @@ class MTClientConnection(ClientConnection):
         return super(MTClientConnection, self).notify(*args, **kw)
 
     @lockCheckWrapper
-    def ask(self, queue, packet, timeout=5, additional_timeout=30):
+    def ask(self, packet, timeout=5, additional_timeout=30):
         msg_id = self._getNextId()
         packet.setId(msg_id)
-        self.dispatcher.register(self, msg_id, queue)
+        self.dispatcher.register(self, msg_id, self.local_var.queue)
         self.expectMessage(msg_id)
         self._addPacket(packet)
         return msg_id