Not tested enough, but the primary election is implemented.

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

connection.py

@@ -130,7 +130,6 @@ class Connection:
             for event in self.event_dict.itervalues():
                 self.cm.removeIdleEvent(event)
             self.event_dict.clear()
-            self.connectionClosed()
 
     def abort(self):
         """Abort dealing with this connection."""
@@ -194,7 +193,7 @@ class Connection:
                         pass
 
                 self.packetReceived(packet)
-                msg = msg[len(packet)]
+                msg = msg[len(packet):]
 
             if msg:
                 self.read_buf = [msg]
@@ -211,13 +210,19 @@ class Connection:
             r = s.recv(4096)
             if not r:
                 logging.error('cannot read')
+                self.connectionClosed()
                 self.close()
             else:
                 self.read_buf.append(r)
         except socket.error, m:
             if m[0] == errno.EAGAIN:
-                return []
-            raise
+                pass
+            elif m[0] == errno.ECONNRESET:
+                logging.error('cannot read')
+                self.connectionClosed()
+                self.close()
+            else:
+                raise
 
     def send(self):
         """Send data to a socket."""
@@ -231,6 +236,7 @@ class Connection:
                 r = s.send(msg)
                 if not r:
                     logging.error('cannot write')
+                    self.connectionClosed()
                     self.close()
                 elif r == len(msg):
                     del self.write_buf[:]
@@ -309,9 +315,10 @@ class Connection:
 
     def packetMalformed(self, packet, error_message):
         """Called when a packet is malformed."""
-        self.peerBroken()
+        logging.info('malformed packet: %s', error_message)
         self.addPacket(Packet().protocolError(packet.getId(), error_message))
         self.abort()
+        self.peerBroken()
 
 class ConnectionManager:
     """This class manages connections and sockets."""
@@ -337,6 +344,9 @@ class ConnectionManager:
         self.listening_socket = s
         self.reader_set.add(s)
 
+    def getConnectionList(self):
+        return self.connection_dict.values()
+
     def register(self, conn):
         self.connection_dict[conn.getSocket()] = conn
 
@@ -355,9 +365,11 @@ class ConnectionManager:
         for s in rlist:
             if s == self.listening_socket:
                 try:
-                    conn, addr = s.accept()
+                    new_s, addr = s.accept()
                     logging.info('accepted a connection from %s:%d', addr[0], addr[1])
-                    self.register(self.klass(self, conn, addr))
+                    conn = self.klass(self, new_s, addr)
+                    self.register(conn)
+                    conn.connectionAccepted()
                 except socket.error, m:
                     if m[0] == errno.EAGAIN:
                         continue
@@ -377,7 +389,7 @@ class ConnectionManager:
             t = time()
             if t - self.prev_time >= 1:
                 self.prev_time = t
-                event_list.sort(key = lambda event: event.getTimeout())
+                event_list.sort(key = lambda event: event.getTime())
                 for event in tuple(event_list):
                     if event(t):
                         event_list.pop(0)
@@ -397,7 +409,7 @@ class ConnectionManager:
         self.reader_set.add(s)
 
     def removeReader(self, s):
-        self.read_set.discard(s)
+        self.reader_set.discard(s)
 
     def addWriter(self, s):
         self.writer_set.add(s)

neo.conf

-[DEFAULTS]
-#master_nodes: 
+[DEFAULT]
+master_nodes: 127.0.0.1:10010 127.0.0.1:10011
 #replicas: 1
 #partitions: 1009
 #name: main
 
-[master]
-#database: test
-#user: test
+[master1]
+database: master1
+user: neo
 #password:
-#server: 127.0.0.1:10010
+server: 127.0.0.1:10010
+
+[master2]
+database: master2
+user: neo
+server: 127.0.0.1:10011

node.py

+from time import time
+
 RUNNING_STATE = 0
 TEMPORARILY_DOWN_STATE = 2
 DOWN_STATE = 3
@@ -12,14 +14,26 @@ class Node(object):
         self.port = port
         self.uuid = uuid
         self.manager = None
+        self.last_state_change = time()
+
+    def setManager(self, manager):
+        self.manager = manager
+
+    def getLastStateChange(self):
+        return self.last_state_change
 
     def getState(self):
         return self.state
 
     def setState(self, new_state):
-        self.state = new_state
+        if self.state != new_state:
+            self.state = new_state
+            self.last_state_change = time()
 
     def setServer(self, ip_address, port):
+        if self.ip_address is not None:
+            self.manager.unregisterServer(self)
+
         self.ip_address = ip_address
         self.port = port
         self.manager.registerServer(self)
@@ -28,6 +42,9 @@ class Node(object):
         return self.ip_address, self.port
 
     def setUUID(self, uuid):
+        if self.uuid is not None:
+            self.manager.unregisterUUID(self)
+
         self.uuid = uuid
         self.manager.registerUUID(self)
 
@@ -55,6 +72,7 @@ class NodeManager(object):
         self.uuid_dict = {}
 
     def add(self, node):
+        node.setManager(self)
         self.node_list.append(node)   
         if node.getServer()[0] is not None:
             self.registerServer(node)

protocol.py

 from struct import pack, unpack
 from socket import inet_ntoa, inet_aton
+import logging
+
+from util import dump
 
 # The protocol version (major, minor).
 PROTOCOL_VERSION = (4, 0)
@@ -9,7 +12,7 @@ MIN_PACKET_SIZE = 10
 MAX_PACKET_SIZE = 0x100000
 
 # Message types.
-ERROR = 0x8000
+ERROR = 0x0000
 REQUEST_NODE_IDENTIFICATION = 0x0001
 ACCEPT_NODE_IDENTIFICATION = 0x8001
 PING = 0x0002
@@ -17,6 +20,7 @@ PONG = 0x8002
 ASK_PRIMARY_MASTER = 0x0003
 ANSWER_PRIMARY_MASTER = 0x8003
 ANNOUNCE_PRIMARY_MASTER = 0x0004
+REELECT_PRIMARY_MASTER = 0x0005
 
 # Error codes.
 NOT_READY_CODE = 1
@@ -35,6 +39,12 @@ CLIENT_NODE_TYPE = 3
 
 VALID_NODE_TYPE_LIST = (MASTER_NODE_TYPE, STORAGE_NODE_TYPE, CLIENT_NODE_TYPE)
 
+# Other constants.
+INVALID_UUID = '\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0'
+INVALID_TID = '\0\0\0\0\0\0\0\0'
+INVALID_SERIAL = '\0\0\0\0\0\0\0\0'
+INVALID_OID = '\0\0\0\0\0\0\0\0'
+
 class ProtocolError(Exception): pass
 
 class Packet:
@@ -46,6 +56,7 @@ class Packet:
 
     @classmethod
     def parse(cls, msg):
+        # logging.debug('parsing %s', dump(msg))
         if len(msg) < MIN_PACKET_SIZE:
             return None
         msg_id, msg_type, msg_len, reserved = unpack('!HHLH', msg[:10])
@@ -60,7 +71,7 @@ class Packet:
         if len(msg) < msg_len:
             # Not enough.
             return None
-        return cls(msg_id, msg_type, msg_len, msg[10:msg_len])
+        return cls(msg_id, msg_type, msg[10:msg_len])
 
     def __init__(self, msg_id = None, msg_type = None, body = None):
         self._id = msg_id
@@ -74,7 +85,7 @@ class Packet:
         return self._type
 
     def __len__(self):
-        return len(self._body)
+        return 10 + len(self._body)
 
     # Encoders.
     def encode(self):
@@ -89,33 +100,68 @@ class Packet:
         self._id = msg_id
         self._type = ERROR
         self._body = pack('!HL', error_code, len(error_message)) + error_message
+        return self
 
     def protocolError(self, msg_id, error_message):
-        self.error(msg_id, PROTOCOL_ERROR_CODE, 'protocol error: ' + error_message)
+        return self.error(msg_id, PROTOCOL_ERROR_CODE, 'protocol error: ' + error_message)
 
     def internalError(self, msg_id, error_message):
-        self.error(msg_id, INTERNAL_ERROR_CODE, 'internal error: ' + error_message)
+        return self.error(msg_id, INTERNAL_ERROR_CODE, 'internal error: ' + error_message)
+
+    def notReady(self, msg_id, error_message):
+        return self.error(msg_id, NOT_READY, 'not ready: ' + error_message)
 
     def ping(self, msg_id):
         self._id = msg_id
         self._type = PING
         self._body = ''
+        return self
 
     def pong(self, msg_id):
         self._id = msg_id
         self._type = PONG
         self._body = ''
+        return self
 
     def requestNodeIdentification(self, msg_id, node_type, uuid, ip_address, port, name):
         self._id = msg_id
         self._type = REQUEST_NODE_IDENTIFICATION
-        self._body = pack('!LLH16s4sHL', protocol_version[0], protocol_version[1],
+        self._body = pack('!LLH16s4sHL', PROTOCOL_VERSION[0], PROTOCOL_VERSION[1],
                           node_type, uuid, inet_aton(ip_address), port, len(name)) + name
+        return self
 
     def acceptNodeIdentification(self, msg_id, node_type, uuid, ip_address, port):
         self._id = msg_id
         self._type = ACCEPT_NODE_IDENTIFICATION
         self._body = pack('!H16s4sH', node_type, uuid, inet_aton(ip_address), port)
+        return self
+
+    def askPrimaryMaster(self, msg_id, ltid, loid):
+        self._id = msg_id
+        self._type = ASK_PRIMARY_MASTER
+        self._body = ltid + loid
+        return self
+
+    def answerPrimaryMaster(self, msg_id, ltid, loid, primary_uuid, known_master_list):
+        self._id = msg_id
+        self._type = ANSWER_PRIMARY_MASTER
+        body = [ltid, loid, primary_uuid, pack('!L', len(known_master_list))]
+        for master in known_master_list:
+            body.append(pack('!4sH16s', inet_aton(master[0]), master[1], master[2]))
+        self._body = ''.join(body)
+        return self
+
+    def announcePrimaryMaster(self, msg_id):
+        self._id = msg_id
+        self._type = ANNOUNCE_PRIMARY_MASTER
+        self._body = ''
+        return self
+
+    def reelectPrimaryMaster(self, msg_id):
+        self._id = msg_id
+        self._type = REELECT_PRIMARY_MASTER
+        self._body = ''
+        return self
 
     # Decoders.
     def decode(self):
@@ -160,7 +206,7 @@ class Packet:
             raise ProtocolError(self, 'invalid name size')
         if node_type not in VALID_NODE_TYPE_LIST:
             raise ProtocolError(self, 'invalid node type %d' % node_type)
-        if (major, minor) != protocol_version:
+        if (major, minor) != PROTOCOL_VERSION:
             raise ProtocolError(self, 'protocol version mismatch')
         return node_type, uuid, ip_address, port, name
     decode_table[REQUEST_NODE_IDENTIFICATION] = _decodeRequestNodeIdentification
@@ -176,3 +222,31 @@ class Packet:
         return node_type, uuid, ip_address, port
     decode_table[ACCEPT_NODE_IDENTIFICATION] = _decodeAcceptNodeIdentification
 
+    def _decodeAskPrimaryMaster(self):
+        try:
+            ltid, loid = unpack('!8s8s', self._body)
+        except:
+            raise ProtocolError(self, 'invalid ask primary master')
+        return ltid, loid
+    decode_table[ASK_PRIMARY_MASTER] = _decodeAskPrimaryMaster
+
+    def _decodeAnswerPrimaryMaster(self):
+        try:
+            ltid, loid, primary_uuid, n = unpack('!8s8s16sL', self._body[:36])
+            known_master_list = []
+            for i in xrange(n):
+                ip_address, port, uuid = unpack('!4sH16s', self._body[36+i*22:58+i*22])
+                ip_address = inet_ntoa(ip_address)
+                known_master_list.append((ip_address, port, uuid))
+        except:
+            raise ProtocolError(self, 'invalid answer primary master')
+        return ltid, loid, primary_uuid, known_master_list
+    decode_table[ANSWER_PRIMARY_MASTER] = _decodeAnswerPrimaryMaster
+
+    def _decodeAnnouncePrimaryMaster(self):
+        pass
+    decode_table[ANNOUNCE_PRIMARY_MASTER] = _decodeAnnouncePrimaryMaster
+
+    def _decodeReelectPrimaryMaster(self):
+        pass
+    decode_table[REELECT_PRIMARY_MASTER] = _decodeReelectPrimaryMaster