Split storage handlers. As for the primary master, all incoming connections are

handled by the identification handler which apply the right handler depending on
the peer node type. Client and storage incoming connections are refused until
the storage reach the operation state, but the listening port is open to ensure
it's available, tryAgain errors are sent to wrong nodes.
Remove decorators because connection type is implied by handler type.


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

neo/storage/app.py

@@ -32,9 +32,10 @@ from neo.connection import ListeningConnection, ClientConnection
 from neo.exception import OperationFailure, PrimaryFailure
 from neo.storage.bootstrap import BootstrapEventHandler
 from neo.storage.verification import VerificationEventHandler
-from neo.storage.operation import OperationEventHandler
+from neo.storage.operation import MasterOperationEventHandler
 from neo.storage.hidden import HiddenEventHandler
 from neo.storage.replicator import Replicator
+from neo.storage.identification import IdentificationEventHandler
 from neo.connector import getConnectorHandler
 from neo.pt import PartitionTable
 from neo.util import dump
@@ -72,6 +73,8 @@ class Application(object):
         self.listening_conn = None
         self.master_conn = None
 
+        # ready is True when operational and got all informations
+        self.ready = False
         self.has_node_information = False
         self.has_partition_table = False
 
@@ -134,23 +137,21 @@ class Application(object):
         for server in self.master_node_list:
             self.nm.add(MasterNode(server = server))
 
+        # Make a listening port
+        handler = IdentificationEventHandler(self)
+        self.listening_conn = ListeningConnection(self.em, handler, 
+            addr=self.server, connector_handler=self.connector_handler)
+
         # Connect to a primary master node, verify data, and
         # start the operation. This cycle will be executed permentnly,
         # until the user explicitly requests a shutdown.
         while 1:
             self.operational = False
-            # refuse any incoming connections for now
-            if self.listening_conn is not None:
-                self.listening_conn.close()
-                self.listening_conn = None
             # look for the primary master
             self.connectToPrimaryMaster()
             assert self.master_conn is not None
             if self.uuid == INVALID_UUID:
                 raise RuntimeError, 'No UUID supplied from the primary master'
-            # Make a listening port when connected to the primary
-            self.listening_conn = ListeningConnection(self.em, None, 
-                    addr=self.server, connector_handler=self.connector_handler)
             try:
                 while 1:
                     try:
@@ -241,32 +242,28 @@ class Application(object):
         logging.info('verifying data')
 
         handler = VerificationEventHandler(self)
+        self.master_conn.setHandler(handler)
         em = self.em
 
-        # Make sure that every connection has the verfication event handler.
-        for conn in em.getConnectionList():
-            conn.setHandler(handler)
-
         while not self.operational:
             em.poll(1)
 
-        # ask node list
+        # ask node list and partition table
         self.master_conn.ask(protocol.askNodeInformation())        
         self.master_conn.ask(protocol.askPartitionTable(()))
         while not self.has_node_information or not self.has_partition_table:
             em.poll(1)
+        self.ready = True
 
     def doOperation(self):
         """Handle everything, including replications and transactions."""
         logging.info('doing operation')
 
-        handler = OperationEventHandler(self)
         em = self.em
         nm = self.nm
 
-        # Make sure that every connection has the verfication event handler.
-        for conn in em.getConnectionList():
-            conn.setHandler(handler)
+        handler = MasterOperationEventHandler(self)
+        self.master_conn.setHandler(handler)
 
         # Forget all unfinished data.
         self.dm.dropUnfinishedData()
@@ -321,7 +318,6 @@ class Application(object):
     def getPartition(self, oid_or_tid):
         return unpack('!Q', oid_or_tid)[0] % self.num_partitions
 
-
     def shutdown(self):
         """Close all connections and exit"""
         for c in self.em.getConnectionList():

neo/storage/identification.py

+#
+# Copyright (C) 2006-2009  Nexedi SA
+# 
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+# 
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+
+import logging
+
+from neo import protocol
+from neo.storage.handler import StorageEventHandler
+from neo.protocol import INVALID_SERIAL, INVALID_TID, \
+        INVALID_PARTITION, BROKEN_STATE, TEMPORARILY_DOWN_STATE, \
+        MASTER_NODE_TYPE, STORAGE_NODE_TYPE, CLIENT_NODE_TYPE, \
+        DISCARDED_STATE, OUT_OF_DATE_STATE
+from neo.util import dump
+
+class IdentificationEventHandler(StorageEventHandler):
+    """ Handler used for incoming connections during operation state """
+
+    def connectionClosed(self, conn):
+        logging.warning('lost a node in IdentificationEventHandler')
+
+    def timeoutExpired(self, conn):
+        logging.warning('lost a node in IdentificationEventHandler')
+
+    def peerBroken(self, conn):
+        logging.warning('lost a node in IdentificationEventHandler')
+
+    def handleRequestNodeIdentification(self, conn, packet, node_type,
+                                        uuid, ip_address, port, name):
+        self.checkClusterName(name)
+        # reject any incoming connections if not ready
+        if not self.app.ready:
+            raise protocol.notReady('try again')
+        app, nm = self.app, self.app.nm
+        server = (ip_address, port)
+        node = app.nm.getNodeByUUID(uuid)
+        if node is None:
+            logging.error('reject an unknown node %s', dump(uuid))
+            raise protocol.NotReadyError
+        # If this node is broken, reject it.
+        if node.getState() == BROKEN_STATE:
+            raise protocol.BrokenNodeDisallowedError
+        # choose the handler according to the node type
+        if node_type == protocol.CLIENT_NODE_TYPE:
+            from neo.storage.operation import ClientOperationEventHandler 
+            handler = ClientOperationEventHandler
+        elif node_type == protocol.STORAGE_NODE_TYPE:
+            from neo.storage.operation import StorageOperationEventHandler
+            handler = StorageOperationEventHandler
+        else:
+            raise protocol.protocolError('reject non-client-or-storage node')
+        # apply the handler and set up the connection
+        handler = handler(self.app)
+        conn.setHandler(handler)
+        conn.setUUID(uuid)
+        node.setUUID(uuid)
+        # FIXME: here we should use pt.getPartitions() and pt.getReplicas()
+        args = (STORAGE_NODE_TYPE, app.uuid, app.server[0], app.server[1], 
+                app.num_partitions, app.num_replicas, uuid)
+        # accept the identification and trigger an event
+        conn.answer(protocol.acceptNodeIdentification(*args), packet)
+        handler.connectionCompleted(conn)
+

neo/storage/verification.py

@@ -24,10 +24,8 @@ from neo.protocol import INVALID_OID, INVALID_TID, \
         MASTER_NODE_TYPE, STORAGE_NODE_TYPE, \
         Packet, UnexpectedPacketError
 from neo.util import dump
-from neo.node import MasterNode, StorageNode, ClientNode
-from neo.connection import ClientConnection
+from neo.node import StorageNode
 from neo.exception import PrimaryFailure, OperationFailure
-from neo import decorators
 
 class VerificationEventHandler(StorageEventHandler):
     """This class deals with events for a verification phase."""
@@ -39,67 +37,23 @@ class VerificationEventHandler(StorageEventHandler):
         StorageEventHandler.connectionAccepted(self, conn, s, addr)
 
     def timeoutExpired(self, conn):
-        if not conn.isServerConnection():
-            # If a primary master node timeouts, I cannot continue.
-            logging.critical('the primary master node times out')
-            raise PrimaryFailure('the primary master node times out')
-
+        # If a primary master node timeouts, I cannot continue.
+        logging.critical('the primary master node times out')
+        raise PrimaryFailure('the primary master node times out')
         StorageEventHandler.timeoutExpired(self, conn)
 
     def connectionClosed(self, conn):
-        if not conn.isServerConnection():
-            # If a primary master node closes, I cannot continue.
-            logging.critical('the primary master node is dead')
-            raise PrimaryFailure('the primary master node is dead')
-
+        # If a primary master node closes, I cannot continue.
+        logging.critical('the primary master node is dead')
+        raise PrimaryFailure('the primary master node is dead')
         StorageEventHandler.connectionClosed(self, conn)
 
     def peerBroken(self, conn):
-        if not conn.isServerConnection():
-            # If a primary master node gets broken, I cannot continue.
-            logging.critical('the primary master node is broken')
-            raise PrimaryFailure('the primary master node is broken')
-
+        # If a primary master node gets broken, I cannot continue.
+        logging.critical('the primary master node is broken')
+        raise PrimaryFailure('the primary master node is broken')
         StorageEventHandler.peerBroken(self, conn)
 
-    @decorators.server_connection_required
-    def handleRequestNodeIdentification(self, conn, packet, node_type,
-                                        uuid, ip_address, port, name):
-        self.checkClusterName(name)
-        app = self.app
-        if node_type != MASTER_NODE_TYPE:
-            logging.info('reject a connection from a non-master')
-            raise protocol.NotReadyError
-
-        addr = (ip_address, port)
-        node = app.nm.getNodeByServer(addr)
-        if node is None:
-            node = MasterNode(server = addr, uuid = uuid)
-            app.nm.add(node)
-        else:
-            # If this node is broken, reject it.
-            if node.getUUID() == uuid:
-                if node.getState() == BROKEN_STATE:
-                    raise protocol.BrokenNodeDisallowedError
-
-        # Trust the UUID sent by the peer.
-        node.setUUID(uuid)
-        conn.setUUID(uuid)
-
-        p = protocol.acceptNodeIdentification(STORAGE_NODE_TYPE, app.uuid, 
-                app.server[0], app.server[1], app.num_partitions, 
-                app.num_replicas, uuid)
-        conn.answer(p, packet)
-
-        # Now the master node should know that I am not the right one.
-        conn.abort()
-
-    def handleAcceptNodeIdentification(self, conn, packet, node_type,
-                                       uuid, ip_address, port,
-                                       num_partitions, num_replicas, your_uuid):
-        raise UnexpectedPacketError
-
-    @decorators.client_connection_required
     def handleAnswerPrimaryMaster(self, conn, packet, primary_uuid,
                                   known_master_list):
         app = self.app
@@ -109,7 +63,6 @@ class VerificationEventHandler(StorageEventHandler):
         # But a primary master node is supposed not to send any info
         # with this packet, so it would be useless.
 
-    @decorators.client_connection_required
     def handleAskLastIDs(self, conn, packet):
         app = self.app
         oid = app.dm.getLastOID() or INVALID_OID
@@ -124,8 +77,9 @@ class VerificationEventHandler(StorageEventHandler):
     def handleAnswerPartitionTable(self, conn, packet, ptid, row_list):
         assert not row_list
         self.app.has_partition_table = True
+        logging.info('Got the partition table :')
+        self.app.pt.log()
 
-    @decorators.client_connection_required
     def handleAskPartitionTable(self, conn, packet, offset_list):
         app = self.app
         row_list = []
@@ -144,7 +98,6 @@ class VerificationEventHandler(StorageEventHandler):
         p = protocol.answerPartitionTable(app.ptid, row_list)
         conn.answer(p, packet)
 
-    @decorators.client_connection_required
     def handleSendPartitionTable(self, conn, packet, ptid, row_list):
         """A primary master node sends this packet to synchronize a partition
         table. Note that the message can be split into multiple packets."""
@@ -176,7 +129,6 @@ class VerificationEventHandler(StorageEventHandler):
                                       cell.getState()))
             app.dm.setPartitionTable(ptid, cell_list)
 
-    @decorators.client_connection_required
     def handleNotifyPartitionChanges(self, conn, packet, ptid, cell_list):
         """This is very similar to Send Partition Table, except that
         the information is only about changes from the previous."""
@@ -203,15 +155,12 @@ class VerificationEventHandler(StorageEventHandler):
         # Then, the database.
         app.dm.changePartitionTable(ptid, cell_list)
 
-    @decorators.client_connection_required
     def handleStartOperation(self, conn, packet):
         self.app.operational = True
 
-    @decorators.client_connection_required
     def handleStopOperation(self, conn, packet):
         raise OperationFailure('operation stopped')
 
-    @decorators.client_connection_required
     def handleAskUnfinishedTransactions(self, conn, packet):
         tid_list = self.app.dm.getUnfinishedTIDList()
         p = protocol.answerUnfinishedTransactions(tid_list)
@@ -225,6 +174,8 @@ class VerificationEventHandler(StorageEventHandler):
             # not been finished.
             t = app.dm.getTransaction(tid, all = True)
         else:
+            # XXX: this should never be used since we don't accept incoming
+            # connections out of the operation state.
             t = app.dm.getTransaction(tid)
 
         if t is None:
@@ -233,7 +184,6 @@ class VerificationEventHandler(StorageEventHandler):
             p = protocol.answerTransactionInformation(tid, t[1], t[2], t[3], t[0])
         conn.answer(p, packet)
 
-    @decorators.client_connection_required
     def handleAskObjectPresent(self, conn, packet, oid, tid):
         if self.app.dm.objectPresent(oid, tid):
             p = protocol.answerObjectPresent(oid, tid)
@@ -242,16 +192,9 @@ class VerificationEventHandler(StorageEventHandler):
                           '%s:%s do not exist' % (dump(oid), dump(tid)))
         conn.answer(p, packet)
 
-    @decorators.client_connection_required
     def handleDeleteTransaction(self, conn, packet, tid):
         self.app.dm.deleteTransaction(tid, all = True)
 
-    @decorators.client_connection_required
     def handleCommitTransaction(self, conn, packet, tid):
         self.app.dm.finishTransaction(tid)
 
-    def handleLockInformation(self, conn, packet, tid):
-        pass
-
-    def handleUnlockInformation(self, conn, packet, tid):
-        pass