Skip to content

N
neoppod
Commit 74ebf6f2 authored by Grégory Wisniewski's avatar Grégory Wisniewski
Browse files
Options

Packet logging improved thanks to Enum class. 


git-svn-id: https://svn.erp5.org/repos/neo/branches/prototype3@428 71dcc9de-d417-0410-9af5-da40c76e7ee4
parent 4dd996c2
Hide whitespace changes
Inline Side-by-side
Showing with 126 additions and 120 deletions
neo/connection.py
View file @ 74ebf6f2
...@@ -284,6 +284,9 @@ class Connection(BaseConnection): ...@@ -284,6 +284,9 @@ class Connection(BaseConnection):
if self.connector is None: if self.connector is None:
return return
ip, port = self.getAddress()
logging.debug('#0x%04x %s to %s:%d', packet.getId(),
packet.getType(), *self.getAddress())
try: try:
self.write_buf.append(packet.encode()) self.write_buf.append(packet.encode())
except ProtocolError, m: except ProtocolError, m:
......
neo/handler.py
View file @ 74ebf6f2
...@@ -75,8 +75,8 @@ class EventHandler(object): ...@@ -75,8 +75,8 @@ class EventHandler(object):
def packetReceived(self, conn, packet): def packetReceived(self, conn, packet):
"""Called when a packet is received.""" """Called when a packet is received."""
logging.debug('packet %d:%x received from %s:%d', logging.debug('#0x%04x %s from %s:%d',
packet.getId(), packet.getType(), *(conn.getAddress())) packet.getId(), packet.getType(), *conn.getAddress())
self.dispatch(conn, packet) self.dispatch(conn, packet)
def packetMalformed(self, conn, packet, error_message): def packetMalformed(self, conn, packet, error_message):
......
neo/protocol.py
View file @ 74ebf6f2
...@@ -72,7 +72,8 @@ class Enum(object): ...@@ -72,7 +72,8 @@ class Enum(object):
# languages harder. # languages harder.
if not isinstance(value, int): if not isinstance(value, int):
raise TypeError, 'Enum class only support integer values.' raise TypeError, 'Enum class only support integer values.'
global_dict[key] = enum_dict[value] = EnumItem(self, key, value) item = EnumItem(self, key, value)
global_dict[key] = enum_dict[value] = item
def get(self, value, default=None): def get(self, value, default=None):
return self.enum_dict.get(value, default) return self.enum_dict.get(value, default)
...@@ -89,171 +90,173 @@ MAX_PACKET_SIZE = 0x100000 ...@@ -89,171 +90,173 @@ MAX_PACKET_SIZE = 0x100000
PACKET_HEADER_SIZE = 10 PACKET_HEADER_SIZE = 10
# Message types. # Message types.
# Error is a special type of message, because this can be sent against any other message, packet_types = Enum({
# even if such a message does not expect a reply usually. Any -> Any.
ERROR = 0x8000
# Check if a peer is still alive. Any -> Any. # Error is a special type of message, because this can be sent against any other message,
PING = 0x0001 # even if such a message does not expect a reply usually. Any -> Any.
'ERROR': 0x8000,
# Notify being alive. Any -> Any. # Check if a peer is still alive. Any -> Any.
PONG = 0x8001 'PING': 0x0001,
# Request a node identification. This must be the first packet for any connection. # Notify being alive. Any -> Any.
# Any -> Any. 'PONG': 0x8001,
REQUEST_NODE_IDENTIFICATION = 0x0002
# Accept a node identification. This should be a reply to Request Node Identification. # Request a node identification. This must be the first packet for any connection.
# Any -> Any. # Any -> Any.
ACCEPT_NODE_IDENTIFICATION = 0x8002 'REQUEST_NODE_IDENTIFICATION': 0x0002,
# Ask a current primary master node. This must be the second message when connecting # Accept a node identification. This should be a reply to Request Node Identification.
# to a master node. Any -> M. # Any -> Any.
ASK_PRIMARY_MASTER = 0x0003 'ACCEPT_NODE_IDENTIFICATION': 0x8002,
# Reply to Ask Primary Master. This message includes a list of known master nodes, # Ask a current primary master node. This must be the second message when connecting
# to make sure that a peer has the same information. M -> Any. # to a master node. Any -> M.
ANSWER_PRIMARY_MASTER = 0x8003 'ASK_PRIMARY_MASTER': 0x0003,
# Announce a primary master node election. PM -> SM. # Reply to Ask Primary Master. This message includes a list of known master nodes,
ANNOUNCE_PRIMARY_MASTER = 0x0004 # to make sure that a peer has the same information. M -> Any.
'ANSWER_PRIMARY_MASTER': 0x8003,
# Force a re-election of a primary master node. M -> M. # Announce a primary master node election. PM -> SM.
REELECT_PRIMARY_MASTER = 0x0005 'ANNOUNCE_PRIMARY_MASTER': 0x0004,
# Notify information about one or more nodes. Any -> PM, PM -> Any. # Force a re-election of a primary master node. M -> M.
NOTIFY_NODE_INFORMATION = 0x0006 'REELECT_PRIMARY_MASTER': 0x0005,
# Ask the last OID, the last TID and the last Partition Table ID that # Notify information about one or more nodes. Any -> PM, PM -> Any.
# a storage node stores. Used to recover information. PM -> S, S -> PM. 'NOTIFY_NODE_INFORMATION': 0x0006,
ASK_LAST_IDS = 0x0007
# Reply to Ask Last IDs. S -> PM, PM -> S. # Ask the last OID, the last TID and the last Partition Table ID that
ANSWER_LAST_IDS = 0x8007 # a storage node stores. Used to recover information. PM -> S, S -> PM.
'ASK_LAST_IDS': 0x0007,
# Ask rows in a partition table that a storage node stores. Used to recover # Reply to Ask Last IDs. S -> PM, PM -> S.
# information. PM -> S. 'ANSWER_LAST_IDS': 0x8007,
ASK_PARTITION_TABLE = 0x0008
# Answer rows in a partition table. S -> PM. # Ask rows in a partition table that a storage node stores. Used to recover
ANSWER_PARTITION_TABLE = 0x8008 # information. PM -> S.
'ASK_PARTITION_TABLE': 0x0008,
# Send rows in a partition table to update other nodes. PM -> S, C. # Answer rows in a partition table. S -> PM.
SEND_PARTITION_TABLE = 0x0009 'ANSWER_PARTITION_TABLE': 0x8008,
# Notify a subset of a partition table. This is used to notify changes. PM -> S, C. # Send rows in a partition table to update other nodes. PM -> S, C.
NOTIFY_PARTITION_CHANGES = 0x000a 'SEND_PARTITION_TABLE': 0x0009,
# Tell a storage nodes to start an operation. Until a storage node receives this # Notify a subset of a partition table. This is used to notify changes. PM -> S, C.
# message, it must not serve client nodes. PM -> S. 'NOTIFY_PARTITION_CHANGES': 0x000a,
START_OPERATION = 0x000b
# Tell a storage node to stop an operation. Once a storage node receives this message, # Tell a storage nodes to start an operation. Until a storage node receives this
# it must not serve client nodes. PM -> S. # message, it must not serve client nodes. PM -> S.
STOP_OPERATION = 0x000c 'START_OPERATION': 0x000b,
# Ask unfinished transactions' IDs. PM -> S. # Tell a storage node to stop an operation. Once a storage node receives this message,
ASK_UNFINISHED_TRANSACTIONS = 0x000d # it must not serve client nodes. PM -> S.
'STOP_OPERATION': 0x000c,
# Answer unfinished transactions' IDs. S -> PM. # Ask unfinished transactions' IDs. PM -> S.
ANSWER_UNFINISHED_TRANSACTIONS = 0x800d 'ASK_UNFINISHED_TRANSACTIONS': 0x000d,
# Ask if an object is present. If not present, OID_NOT_FOUND should be returned. PM -> S. # Answer unfinished transactions' IDs. S -> PM.
ASK_OBJECT_PRESENT = 0x000f 'ANSWER_UNFINISHED_TRANSACTIONS': 0x800d,
# Answer that an object is present. PM -> S. # Ask if an object is present. If not present, OID_NOT_FOUND should be returned. PM -> S.
ANSWER_OBJECT_PRESENT = 0x800f 'ASK_OBJECT_PRESENT': 0x000f,
# Delete a transaction. PM -> S. # Answer that an object is present. PM -> S.
DELETE_TRANSACTION = 0x0010 'ANSWER_OBJECT_PRESENT': 0x800f,
# Commit a transaction. PM -> S. # Delete a transaction. PM -> S.
COMMIT_TRANSACTION = 0x0011 'DELETE_TRANSACTION': 0x0010,
# Ask a new transaction ID. C -> PM. # Commit a transaction. PM -> S.
ASK_NEW_TID = 0x0012 'COMMIT_TRANSACTION': 0x0011,
# Answer a new transaction ID. PM -> C. # Ask a new transaction ID. C -> PM.
ANSWER_NEW_TID = 0x8012 'ASK_NEW_TID': 0x0012,
# Finish a transaction. C -> PM. # Answer a new transaction ID. PM -> C.
FINISH_TRANSACTION = 0x0013 'ANSWER_NEW_TID': 0x8012,
# Notify a transaction finished. PM -> C. # Finish a transaction. C -> PM.
NOTIFY_TRANSACTION_FINISHED = 0x8013 'FINISH_TRANSACTION': 0x0013,
# Lock information on a transaction. PM -> S. # Notify a transaction finished. PM -> C.
LOCK_INFORMATION = 0x0014 'NOTIFY_TRANSACTION_FINISHED': 0x8013,
# Notify information on a transaction locked. S -> PM. # Lock information on a transaction. PM -> S.
NOTIFY_INFORMATION_LOCKED = 0x8014 'LOCK_INFORMATION': 0x0014,
# Invalidate objects. PM -> C. # Notify information on a transaction locked. S -> PM.
INVALIDATE_OBJECTS = 0x0015 'NOTIFY_INFORMATION_LOCKED': 0x8014,
# Unlock information on a transaction. PM -> S. # Invalidate objects. PM -> C.
UNLOCK_INFORMATION = 0x0016 'INVALIDATE_OBJECTS': 0x0015,
# Ask new object IDs. C -> PM. # Unlock information on a transaction. PM -> S.
ASK_NEW_OIDS = 0x0017 'UNLOCK_INFORMATION': 0x0016,
# Answer new object IDs. PM -> C. # Ask new object IDs. C -> PM.
ANSWER_NEW_OIDS = 0x8017 'ASK_NEW_OIDS': 0x0017,
# Ask to store an object. Send an OID, an original serial, a current # Answer new object IDs. PM -> C.
# transaction ID, and data. C -> S. 'ANSWER_NEW_OIDS': 0x8017,
ASK_STORE_OBJECT = 0x0018
# Answer if an object has been stored. If an object is in conflict, # Ask to store an object. Send an OID, an original serial, a current
# a serial of the conflicting transaction is returned. In this case, # transaction ID, and data. C -> S.
# if this serial is newer than the current transaction ID, a client 'ASK_STORE_OBJECT': 0x0018,
# node must not try to resolve the conflict. S -> C.
ANSWER_STORE_OBJECT = 0x8018
# Abort a transaction. C -> S, PM. # Answer if an object has been stored. If an object is in conflict,
ABORT_TRANSACTION = 0x0019 # a serial of the conflicting transaction is returned. In this case,
# if this serial is newer than the current transaction ID, a client
# node must not try to resolve the conflict. S -> C.
'ANSWER_STORE_OBJECT': 0x8018,
# Ask to store a transaction. C -> S. # Abort a transaction. C -> S, PM.
ASK_STORE_TRANSACTION = 0x001a 'ABORT_TRANSACTION': 0x0019,
# Answer if transaction has been stored. S -> C. # Ask to store a transaction. C -> S.
ANSWER_STORE_TRANSACTION = 0x801a 'ASK_STORE_TRANSACTION': 0x001a,
# Ask a stored object by its OID and a serial or a TID if given. If a serial # Answer if transaction has been stored. S -> C.
# is specified, the specified revision of an object will be returned. If 'ANSWER_STORE_TRANSACTION': 0x801a,
# a TID is specified, an object right before the TID will be returned. C -> S.
ASK_OBJECT = 0x001b
# Answer the requested object. S -> C. # Ask a stored object by its OID and a serial or a TID if given. If a serial
ANSWER_OBJECT = 0x801b # is specified, the specified revision of an object will be returned. If
# a TID is specified, an object right before the TID will be returned. C -> S.
'ASK_OBJECT': 0x001b,
# Ask for TIDs between a range of offsets. The order of TIDs is descending, # Answer the requested object. S -> C.
# and the range is [first, last). C, S -> S. 'ANSWER_OBJECT': 0x801b,
ASK_TIDS = 0x001d
# Answer the requested TIDs. S -> C, S. # Ask for TIDs between a range of offsets. The order of TIDs is descending,
ANSWER_TIDS = 0x801d # and the range is [first, last). C, S -> S.
'ASK_TIDS': 0x001d,
# Ask information about a transaction. Any -> S. # Answer the requested TIDs. S -> C, S.
ASK_TRANSACTION_INFORMATION = 0x001e 'ANSWER_TIDS': 0x801d,
# Answer information (user, description) about a transaction. S -> Any. # Ask information about a transaction. Any -> S.
ANSWER_TRANSACTION_INFORMATION = 0x801e 'ASK_TRANSACTION_INFORMATION': 0x001e,
# Ask history information for a given object. The order of serials is # Answer information (user, description) about a transaction. S -> Any.
# descending, and the range is [first, last]. C, S -> S. 'ANSWER_TRANSACTION_INFORMATION': 0x801e,
ASK_OBJECT_HISTORY = 0x001f
# Answer history information (serial, size) for an object. S -> C, S. # Ask history information for a given object. The order of serials is
ANSWER_OBJECT_HISTORY = 0x801f # descending, and the range is [first, last]. C, S -> S.
'ASK_OBJECT_HISTORY': 0x001f,
# Ask for OIDs between a range of offsets. The order of OIDs is descending, # Answer history information (serial, size) for an object. S -> C, S.
# and the range is [first, last). S -> S. 'ANSWER_OBJECT_HISTORY': 0x801f,
ASK_OIDS = 0x0020
# Answer the requested OIDs. S -> S. # Ask for OIDs between a range of offsets. The order of OIDs is descending,
ANSWER_OIDS = 0x8020 # and the range is [first, last). S -> S.
'ASK_OIDS': 0x0020,
# Answer the requested OIDs. S -> S.
'ANSWER_OIDS': 0x8020,
})
# Error codes. # Error codes.
NOT_READY_CODE = 1 NOT_READY_CODE = 1
...@@ -326,6 +329,7 @@ class Packet(object): ...@@ -326,6 +329,7 @@ class Packet(object):
if len(msg) < MIN_PACKET_SIZE: if len(msg) < MIN_PACKET_SIZE:
return None return None
msg_id, msg_type, msg_len = unpack('!LHL', msg[:PACKET_HEADER_SIZE]) msg_id, msg_type, msg_len = unpack('!LHL', msg[:PACKET_HEADER_SIZE])
msg_type = packet_types[msg_type]
if msg_len > MAX_PACKET_SIZE: if msg_len > MAX_PACKET_SIZE:
raise ProtocolError(cls(msg_id, msg_type), raise ProtocolError(cls(msg_id, msg_type),
'message too big (%d)' % msg_len) 'message too big (%d)' % msg_len)
...@@ -357,7 +361,6 @@ class Packet(object): ...@@ -357,7 +361,6 @@ class Packet(object):
# Encoders. # Encoders.
def encode(self): def encode(self):
msg = pack('!LHL', self._id, self._type, PACKET_HEADER_SIZE + len(self._body)) + self._body msg = pack('!LHL', self._id, self._type, PACKET_HEADER_SIZE + len(self._body)) + self._body
logging.debug('encoding %d:%x', self._id, self._type)
if len(msg) > MAX_PACKET_SIZE: if len(msg) > MAX_PACKET_SIZE:
raise ProtocolError('message too big (%d)' % len(msg)) raise ProtocolError('message too big (%d)' % len(msg))
return msg return msg
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7