use a thread mixing class for storage and implement a dispatcher to manage packets

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

neo/client/NEOStorage.py

-
+from Queue import Queue
+from threading import Lock
 from ZODB import BaseStorage, ConflictResolution, POSException
 from ZODB.utils import p64, u64, cp, z64
 
+from thread import get_ident
+from neo.client.dispatcher import Dispatcher
+from neo.event import EventManager
+import logging
+
 class NEOStorageError(POSException.StorageError):
     pass
 
-class NEOStorageConflictError(NEOStorageError):  
+class NEOStorageConflictError(NEOStorageError):
     pass
 
 class NEOStorageNotFoundError(NEOStorageError):
@@ -14,28 +20,47 @@ class NEOStorageNotFoundError(NEOStorageError):
 class NEOStorage(BaseStorage.BaseStorage,
                  ConflictResolution.ConflictResolvingStorage):
     """Wrapper class for neoclient."""
-  
-    def __init__(self, master_addr, master_port, read_only=False, **kw):
+
+    def __init__(self, master_addr, master_port, name, read_only=False, **kw):
         self._is_read_only = read_only
-        from neo.client.app import Application # here to prevent recursive import
-        self.app = Application(master_addr, master_port)
+        # Transaction must be under protection of lock
+        l = Lock()
+        self._txn_lock_acquire = l.acquire
+        self._txn_lock_release = l.release
+        # Create two queue for message between thread and dispatcher
+        # - message queue is for message that has to be send to other node
+        # through the dispatcher
+        # - request queue is for message receive from other node which have to
+        # be processed
+        message_queue = Queue()
+        request_queue = Queue()
+        # Create the event manager
+        em = EventManager()
+        # Create dispatcher thread
+        dispatcher = Dispatcher(em, message_queue, request_queue)
+        dispatcher.setDaemon(True)
+        dispatcher.start()
+        # Import here to prevent recursive import
+        from neo.client.app import Application
+        self.app = Application(master_addr, master_port, name, em, dispatcher,
+                               message_queue, request_queue)
 
     def load(self, oid, version=None):
         try:
-            self.app.load(oid)
+            return self.app.process_method('load', oid=oid)
         except NEOStorageNotFoundError:
             raise POSException.POSKeyError (oid)
-          
+
     def close(self):
-        self.app.close()
+        return self.app.process_method('close')
 
     def cleanup(self):
         raise NotImplementedError
-  
+
     def lastSerial(self):
         # does not seem to be used
         raise NotImplementedError
-  
+
     def lastTransaction(self):
         # does not seem to be used
         raise NotImplementedError
@@ -43,31 +68,39 @@ class NEOStorage(BaseStorage.BaseStorage,
     def new_oid(self):
         if self._is_read_only:
             raise POSException.ReadOnlyError()
-        self.app.new_oid()
+        return self.app.process_method('new_oid')
 
     def tpc_begin(self, transaction, tid=None, status=' '):
         if self._is_read_only:
             raise POSException.ReadOnlyError()
-        self.app.tpc_begin(transaction, tid, status)
+        self._txn_lock_acquire()
+        return self.app.process_method('tpc_begin', transaction=transaction, tid=tid, status=status)
 
     def tpc_vote(self, transaction):
         if self._is_read_only:
             raise POSException.ReadOnlyError()
-        self.app.tpc_vote(transaction)
-  
+        return self.app.process_method('tpc_vote', transaction=transaction)
+
     def tpc_abort(self, transaction):
         if self._is_read_only:
             raise POSException.ReadOnlyError()
-        self.app.tpc_abort(transaction)
-  
+        try:
+            return self.app.process_method('tpc_abort', transaction=transaction)
+        except:
+            self._txn_lock_release()
+
     def tpc_finish(self, transaction, f=None):
-        self.app.tpc_finish(transaction, f)
+        try:
+            return self.app.process_method('tpc_finish', transaction=transaction, f=f)
+        except:
+            self._txn_lock_release()
 
     def store(self, oid, serial, data, version, transaction):
         if self._is_read_only:
             raise POSException.ReadOnlyError()
         try:
-            self.app.store(oid, serial, data, version, transaction)
+            return self.app.process_method('store', oid=oid, serial=serial, data=data,
+                                    version=version, transaction=transaction)
         except NEOStorageConflictError:
             new_data = self.tryToResolveConflict(oid, self.app.tid,
                                                      serial, data)
@@ -78,53 +111,46 @@ class NEOStorage(BaseStorage.BaseStorage,
                 raise POSException.ConflictError(oid=oid,
                                                  serials=(self.app.tid,
                                                           serial),data=data)
-          
+
     def _clear_temp(self):
         raise NotImplementedError
 
     def getSerial(self, oid):
         try:
-            self.app.getSerial(oid)
+            return self.app.process_method('getSerial', oid=oid)
         except NEOStorageNotFoundError:
             raise POSException.POSKeyError (oid)
-    
+
     # mutliple revisions
     def loadSerial(self, oid, serial):
         try:
-            self.app.loadSerial(oid,serial)
+            return self.app.process_method('loadSerial', oid=oid, serial=serial)
         except NEOStorageNotFoundError:
             raise POSException.POSKeyError (oid, serial)
 
     def loadBefore(self, oid, tid):
         try:
-            self.app.loadBefore(self, oid, tid)
+            return self.app.process_method('loadBefore', oid=oid, tid=tid)
         except NEOStorageNotFoundError:
             raise POSException.POSKeyError (oid, tid)
-  
+
     def iterator(self, start=None, stop=None):
         raise NotImplementedError
 
-    # undo  
+    # undo
     def undo(self, transaction_id, txn):
         if self._is_read_only:
             raise POSException.ReadOnlyError()
-        self.app.undo(transaction_id, txn)
+        self._txn_lock_acquire()
+        try:
+            return self.app.process_method('undo', transaction_id=transaction_id, txn=txn)
+        except:
+            self._txn_lock_release()
 
-    def undoInfo(self, first=0, last=-20, specification=None):
-        if self._is_read_only:
-            raise POSException.ReadOnlyError()
-        self.app.undoInfo(first, last, specification)
-    
     def undoLog(self, first, last, filter):
         if self._is_read_only:
             raise POSException.ReadOnlyError()
-        # This should not be used by ZODB
-        # instead it should use undoInfo
-        # Look at ZODB/interface.py for more info
-        if filter is not None:            
-            return []
-        else:
-            return self.undoInfo(first, last)
+        return self.undoLog(first, last, filter)
 
     def supportsUndo(self):
         return 0

neo/client/dispatcher.py

+from threading import Thread
+from Queue import Empty
+
+from neo.protocol import PING, Packet
+
+class Dispatcher(Thread):
+    """Dispatcher class use to redirect request to thread."""
+
+    def __init__(self, em, message_queue, request_queue, **kw):
+        Thread.__init__(self, **kw)
+        self._message_queue = message_queue
+        self._request_queue = request_queue
+        self.em = em
+        # This dict is used to associate conn/message id to client thread queue
+        # and thus redispatch answer to the original thread
+        self.message_table = {}
+
+    def run(self):
+        while 1:
+            # First check if we receive any new message from other node
+            self.message = None
+            m = None
+            self.em.poll(1)
+            if self.message is not None:
+                conn, packet = self.message
+                # now send message to waiting thread
+                key = "%s-%s" %(conn.getUUID(),packet.getId())
+                if self.message_table.has_key(key):
+                    tmp_q = self.message_table.pop(key)
+                    tmp_q.put(self.message, True)
+                else:
+                    conn, packet = self.message
+                    method_type = packet.getType()
+                    if method_type == PING:
+                        # must answer with no delay
+                        conn.addPacket(Packet().pong(packet.getId()))
+                    else:
+                        # put message in request queue
+                        self._request_queue.put(self.message, True)
+
+            # Then check if a client ask me to send a message
+            try:
+                m = self._message_queue.get_nowait()
+                if m is not None:
+                    tmp_q, msg_id, conn, p = m
+                    conn.addPacket(p)
+                    conn.expectMessage(msg_id)
+                    if tmp_q is not None:
+                        key = "%s-%s" %(conn.getUUID(),msg_id)
+                        self.message_table[key] = tmp_q
+            except Empty:
+                continue
+
+
+

neo/client/handler.py

@@ -9,14 +9,20 @@ from neo.pt import PartitionTable
 
 from ZODB.TimeStamp import TimeStamp
 from ZODB.utils import p64
+from thread import get_ident
 
 class ClientEventHandler(EventHandler):
     """This class deals with events for a master."""
 
-    def __init__(self, app):
+    def __init__(self, app, dispatcher):
         self.app = app
+        self.dispatcher = dispatcher
         EventHandler.__init__(self)
 
+    def packetReceived(self, conn, packet):
+        logging.debug("received packet id %s" %(packet.getId(),))
+        self.dispatcher.message = conn, packet
+
     def handleNotReady(self, conn, packet, message):
         if isinstance(conn, ClientConnection):
             app = self.app
@@ -54,8 +60,12 @@ class ClientEventHandler(EventHandler):
 
                 # Ask a primary master.
                 msg_id = conn.getNextId()
-                conn.addPacket(Packet().askPrimaryMaster(msg_id))
-                conn.expectMessage(msg_id)
+                p = Packet()
+                p.askPrimaryMaster(msg_id)
+                # send message to dispatcher
+                app.queue.put((app.local_var.tmp_q, msg_id, conn, p), True)
+            elif node_type == STORAGE_NODE_TYPE:
+                app.storage_node = node
         else:
             self.handleUnexpectedPacket(conn, packet)
 
@@ -206,8 +216,7 @@ class ClientEventHandler(EventHandler):
         if isinstance(conn, ClientConnection):
             app = self.app
             if tid != app.tid:
-                # What's this ?
-                raise NEOStorageError
+                app.txn_finished = -1
             else:
                 app.txn_finished = 1
         else:
@@ -225,16 +234,16 @@ class ClientEventHandler(EventHandler):
             self.handleUnexpectedPacket(conn, packet)
 
     # Storage node handler
-    def handleAnwserObjectByOID(self, oid, start_serial, end_serial, compression,
-                                checksum, data):
+    def handleAnwserObject(self, conn, packet, oid, start_serial, end_serial, compression,
+                           checksum, data):
         if isinstance(conn, ClientConnection):
             app = self.app
-            app.loaded_object = (oid, start_serial, end_serial, compression,
-                                 checksum, data)
+            app.local_var.loaded_object = (oid, start_serial, end_serial, compression,
+                                           checksum, data)
         else:
             self.handleUnexpectedPacket(conn, packet)
 
-    def handleAnswerStoreObject(self, conflicting, oid, serial):
+    def handleAnswerStoreObject(self, conn, packet, conflicting, oid, serial):
         if isinstance(conn, ClientConnection):
             app = self.app
             if conflicting == '1':
@@ -244,14 +253,14 @@ class ClientEventHandler(EventHandler):
         else:
             self.handleUnexpectedPacket(conn, packet)
 
-    def handleAnswerStoreTransaction(self, tid):
+    def handleAnswerStoreTransaction(self, conn, packet, tid):
         if isinstance(conn, ClientConnection):
             app = self.app
             app.txn_stored = 1
         else:
             self.handleUnexpectedPacket(conn, packet)
 
-    def handleAnswerTransactionInformation(self, tid, user, desc, oid_list):
+    def handleAnswerTransactionInformation(self, conn, packet, tid, user, desc, oid_list):
         if isinstance(conn, ClientConnection):
             app = self.app
             # transaction information are returned as a dict
@@ -261,11 +270,11 @@ class ClientEventHandler(EventHandler):
             info['description'] = desc
             info['id'] = p64(long(tid))
             info['oids'] = oid_list
-            app.txn_info = info
+            app.local_var.txn_info = info
         else:
             self.handleUnexpectedPacket(conn, packet)
 
-    def handleAnswerObjectHistory(self, oid, history_list):
+    def handleAnswerObjectHistory(self, conn, packet, oid, history_list):
         if isinstance(conn, ClientConnection):
             app = self.app
             # history_list is a list of tuple (serial, size)
@@ -273,3 +282,22 @@ class ClientEventHandler(EventHandler):
         else:
             self.handleUnexpectedPacket(conn, packet)
 
+    def handleOidNotFound(self, conn, packet, message):
+        if isinstance(conn, ClientConnection):
+            app = self.app
+            # This can happen either when :
+            # - loading an object
+            # - asking for history
+            self.local_var.asked_object = -1
+            self.local_var.history = -1
+        else:
+            self.handleUnexpectedPacket(conn, packet)
+
+    def handleTidNotFound(self, conn, packet, message):
+        if isinstance(conn, ClientConnection):
+            app = self.app
+            # This can happen when requiring txn informations
+            self.local_var.txn_info = -1
+        else:
+            self.handleUnexpectedPacket(conn, packet)
+

neo/client/multithreading.py

+from threading import Thread
+
+class ThreadingMixIn:
+    """Mix-in class to handle each method in a new thread."""
+
+    def process_method_thread(self, method, kw):
+        m = getattr(self, method)
+        try:
+            r = m(**kw)
+        finally:
+            self._return_lock_acquire()
+            self.returned_data = r
+
+    def process_method(self, method, **kw):
+        """Start a new thread to process the method."""
+        t = Thread(target = self.process_method_thread,
+                   args = (method, kw))
+        t.start()
+        # wait for thread to be completed, returned value must be
+        # under protection of a lock
+        try:
+            t.join()
+            return self.returned_data
+        finally:
+            self._return_lock_release()