Commit 37c75a72 by Grégory Wisniewski

Remove 'packet' parameter of handler's methods.

Almost any use of packet was to retreive the msg_id of the request and pass it to answer().
In storage's replicator, critical TIDs are indexed with UUID instead of remote connection ID (full replicator review required).
In admin node, expected answers are still queued with the msg_id but it seems useless as no concurrent queries should happen.

git-svn-id: https://svn.erp5.org/repos/neo/trunk@1569 71dcc9de-d417-0410-9af5-da40c76e7ee4
1 parent 91ff3fef
......@@ -133,8 +133,6 @@ RC - Review output of pylint (CODE)
an incoming packet that trigger the poll() system call.
- Allow daemonize NEO processes, re-use code from TIDStorage and support
start/stop/restart/status commands.
- Remove 'packet' parameter from handler methods. Set the last_received_id
attribute on the connection and reload it from answer() method.
- Consider don't close the connection after sending a packet but wait (a
bit) for the closure from the remote peer.
- Rename packets:
......
......@@ -141,7 +141,7 @@ class Application(object):
self.master_conn.ask(Packets.AskNodeInformation())
self.master_conn.ask(Packets.AskPartitionTable([]))
def sendPartitionTable(self, conn, min_offset, max_offset, uuid, msg_id):
def sendPartitionTable(self, conn, min_offset, max_offset, uuid):
# we have a pt
self.pt.log()
row_list = []
......@@ -164,4 +164,4 @@ class Application(object):
conn.notify(p)
return
p = Packets.AnswerPartitionList(self.ptid, row_list)
conn.answer(p, msg_id)
conn.answer(p)
......@@ -26,7 +26,7 @@ from neo.util import dump
class AdminEventHandler(EventHandler):
"""This class deals with events for administrating cluster."""
def askPartitionList(self, conn, packet, min_offset, max_offset, uuid):
def askPartitionList(self, conn, min_offset, max_offset, uuid):
logging.info("ask partition list from %s to %s for %s" %
(min_offset, max_offset, dump(uuid)))
app = self.app
......@@ -41,22 +41,21 @@ class AdminEventHandler(EventHandler):
{'min_offset' : min_offset,
'max_offset' : max_offset,
'uuid' : uuid,
'msg_id' : packet.getId()})
'msg_id' : conn.getPeerId()})
else:
app.sendPartitionTable(conn, min_offset, max_offset, uuid,
packet.getId())
app.sendPartitionTable(conn, min_offset, max_offset, uuid)
def askNodeList(self, conn, packet, node_type):
def askNodeList(self, conn, node_type):
logging.info("ask node list for %s" %(node_type))
def node_filter(n):
return n.getType() is node_type
node_list = self.app.nm.getList(node_filter)
node_information_list = [node.asTuple() for node in node_list ]
p = Packets.AnswerNodeList(node_information_list)
conn.answer(p, packet.getId())
conn.answer(p)
def setNodeState(self, conn, packet, uuid, state, modify_partition_table):
def setNodeState(self, conn, uuid, state, modify_partition_table):
logging.info("set node state for %s-%s" %(dump(uuid), state))
node = self.app.nm.getByUUID(uuid)
if node is None:
......@@ -64,32 +63,32 @@ class AdminEventHandler(EventHandler):
if node.getState() == state and modify_partition_table is False:
# no change
p = protocol.ack('no change')
conn.answer(p, packet.getId())
conn.answer(p)
return
# forward to primary master node
if self.app.master_conn is None:
raise protocol.NotReadyError('Not connected to a primary master.')
p = Packets.SetNodeState(uuid, state, modify_partition_table)
msg_id = self.app.master_conn.ask(p)
self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()})
self.app.dispatcher.register(msg_id, conn, {'msg_id' : conn.getPeerId()})
def setClusterState(self, conn, packet, state):
def setClusterState(self, conn, state):
# forward to primary
if self.app.master_conn is None:
raise protocol.NotReadyError('Not connected to a primary master.')
p = Packets.SetClusterState(state)
msg_id = self.app.master_conn.ask(p)
self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()})
self.app.dispatcher.register(msg_id, conn, {'msg_id' : conn.getPeerId()})
def addPendingNodes(self, conn, packet, uuid_list):
def addPendingNodes(self, conn, uuid_list):
if self.app.master_conn is None:
raise protocol.NotReadyError('Not connected to a primary master.')
logging.info('Add nodes %s' % [dump(uuid) for uuid in uuid_list])
# forward the request to primary
msg_id = self.app.master_conn.ask(Packets.AddPendingNodes(uuid_list))
self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()})
self.app.dispatcher.register(msg_id, conn, {'msg_id' : conn.getPeerId()})
def askClusterState(self, conn, packet):
def askClusterState(self, conn):
if self.app.cluster_state is None:
if self.app.master_conn is None:
raise protocol.NotReadyError('Not connected to a primary ' \
......@@ -97,17 +96,15 @@ class AdminEventHandler(EventHandler):
# required it from PMN first
msg_id = self.app.master_conn.ask(Packets.AskClusterState())
self.app.dispatcher.register(msg_id, conn,
{'msg_id' : packet.getId()})
{'msg_id' : conn.getPeerId()})
else:
conn.answer(Packets.AnswerClusterState(self.app.cluster_state),
packet.getId())
conn.answer(Packets.AnswerClusterState(self.app.cluster_state))
def askPrimary(self, conn, packet):
def askPrimary(self, conn):
if self.app.master_conn is None:
raise protocol.NotReadyError('Not connected to a primary master.')
master_node = self.app.master_node
conn.answer(Packets.AnswerPrimary(master_node.getUUID(), []),
packet.getId())
conn.answer(Packets.AnswerPrimary(master_node.getUUID(), []))
class MasterEventHandler(EventHandler):
""" This class is just used to dispacth message to right handler"""
......@@ -141,17 +138,17 @@ class MasterEventHandler(EventHandler):
# unexpectexd answers and notifications
super(MasterEventHandler, self).dispatch(conn, packet)
def answerNodeInformation(self, conn, packet):
def answerNodeInformation(self, conn):
# XXX: This will no more exists when the initialization module will be
# implemented for factorize code (as done for bootstrap)
logging.debug("answerNodeInformation")
def answerPartitionTable(self, conn, packet, ptid, row_list):
def answerPartitionTable(self, conn, ptid, row_list):
# XXX: This will no more exists when the initialization module will be
# implemented for factorize code (as done for bootstrap)
logging.debug("answerPartitionTable")
def notifyPartitionChanges(self, conn, packet, ptid, cell_list):
def notifyPartitionChanges(self, conn, ptid, cell_list):
app = self.app
if ptid < app.ptid:
# Ignore this packet.
......@@ -159,7 +156,7 @@ class MasterEventHandler(EventHandler):
app.ptid = ptid
app.pt.update(ptid, cell_list, app.nm)
def sendPartitionTable(self, conn, packet, ptid, row_list):
def sendPartitionTable(self, conn, ptid, row_list):
uuid = conn.getUUID()
app = self.app
nm = app.nm
......@@ -176,10 +173,10 @@ class MasterEventHandler(EventHandler):
pt.setCell(offset, node, state)
pt.log()
def notifyClusterInformation(self, conn, packet, cluster_state):
def notifyClusterInformation(self, conn, cluster_state):
self.app.cluster_state = cluster_state
def notifyNodeInformation(self, conn, packet, node_list):
def notifyNodeInformation(self, conn, node_list):
app = self.app
app.nm.update(node_list)
if not app.pt.filled():
......@@ -191,34 +188,33 @@ class MasterEventHandler(EventHandler):
class MasterRequestEventHandler(EventHandler):
""" This class handle all answer from primary master node"""
def __answerNeoCTL(self, msg_id, packet):
def __answerNeoCTL(self, conn, packet):
msg_id = conn.getPeerId()
client_conn, kw = self.app.dispatcher.pop(msg_id)
client_conn.answer(packet, kw['msg_id'])
client_conn.answer(packet)
def answerClusterState(self, conn, packet, state):
def answerClusterState(self, conn, state):
logging.info("answerClusterState for a conn")
self.app.cluster_state = state
self.__answerNeoCTL(packet.getId(),
Packets.AnswerClusterState(state))
self.__answerNeoCTL(conn, Packets.AnswerClusterState(state))
def answerNewNodes(self, conn, packet, uuid_list):
def answerNewNodes(self, conn, uuid_list):
logging.info("answerNewNodes for a conn")
self.__answerNeoCTL(packet.getId(),
Packets.AnswerNewNodes(uuid_list))
self.__answerNeoCTL(conn, Packets.AnswerNewNodes(uuid_list))
def answerPartitionTable(self, conn, packet, ptid, row_list):
def answerPartitionTable(self, conn, ptid, row_list):
logging.info("answerPartitionTable for a conn")
client_conn, kw = self.app.dispatcher.pop(packet.getId())
client_conn, kw = self.app.dispatcher.pop(conn.getPeerId())
# sent client the partition table
self.app.sendPartitionTable(client_conn, **kw)
self.app.sendPartitionTable(client_conn)
def answerNodeState(self, conn, packet, uuid, state):
self.__answerNeoCTL(packet.getId(),
def answerNodeState(self, conn, uuid, state):
self.__answerNeoCTL(conn,
Packets.AnswerNodeState(uuid, state))
def ack(self, conn, packet, msg):
self.__answerNeoCTL(packet.getId(), protocol.ack(msg))
def ack(self, conn, msg):
self.__answerNeoCTL(conn, protocol.ack(msg))
def protocolError(self, conn, packet, msg):
self.__answerNeoCTL(packet.getId(), protocol.protocolError(msg))
def protocolError(self, conn, msg):
self.__answerNeoCTL(conn, protocol.protocolError(msg))
......@@ -69,7 +69,7 @@ class BootstrapManager(EventHandler):
"""
self.current = None
def notReady(self, conn, packet, message):
def notReady(self, conn, message):
"""
The primary master send this message when it is still not ready to
handle the client node.
......@@ -79,7 +79,7 @@ class BootstrapManager(EventHandler):
self.current = None
conn.close()
def answerPrimary(self, conn, packet, primary_uuid, known_master_list):
def answerPrimary(self, conn, primary_uuid, known_master_list):
"""
A master answer who's the primary. If it's another node, connect to it.
If it's itself then the primary is successfully found, ask
......@@ -108,7 +108,7 @@ class BootstrapManager(EventHandler):
conn.ask(Packets.RequestIdentification(self.node_type,
self.uuid, self.server, self.name))
def acceptIdentification(self, conn, packet, node_type,
def acceptIdentification(self, conn, node_type,
uuid, num_partitions, num_replicas, your_uuid):
"""
The primary master has accepted the node.
......
......@@ -25,12 +25,12 @@ from neo.util import dump
class PrimaryBootstrapHandler(AnswerBaseHandler):
""" Bootstrap handler used when looking for the primary master """
def notReady(self, conn, packet, message):
def notReady(self, conn, message):
app = self.app
app.trying_master_node = None
app.setNodeNotReady()
def acceptIdentification(self, conn, packet, node_type,
def acceptIdentification(self, conn, node_type,
uuid, num_partitions, num_replicas, your_uuid):
app = self.app
node = app.nm.getByAddress(conn.getAddress())
......@@ -49,7 +49,7 @@ class PrimaryBootstrapHandler(AnswerBaseHandler):
# Always create partition table
app.pt = PartitionTable(num_partitions, num_replicas)
def answerPrimary(self, conn, packet, primary_uuid,
def answerPrimary(self, conn, primary_uuid,
known_master_list):
app = self.app
# Register new master nodes.
......@@ -81,10 +81,10 @@ class PrimaryBootstrapHandler(AnswerBaseHandler):
app.trying_master_node = None
conn.close()
def answerPartitionTable(self, conn, packet, ptid, row_list):
def answerPartitionTable(self, conn, ptid, row_list):
pass
def answerNodeInformation(self, conn, packet):
def answerNodeInformation(self, conn):
pass
class PrimaryNotificationsHandler(BaseHandler):
......@@ -117,10 +117,10 @@ class PrimaryNotificationsHandler(BaseHandler):
logging.critical("primary master node is broken")
BaseHandler.peerBroken(self, conn)
def stopOperation(self, conn, packet):
def stopOperation(self, conn):
logging.critical("master node ask to stop operation")
def invalidateObjects(self, conn, packet, oid_list, tid):
def invalidateObjects(self, conn, oid_list, tid):
app = self.app
app._cache_lock_acquire()
try:
......@@ -142,15 +142,15 @@ class PrimaryNotificationsHandler(BaseHandler):
# to avoid a dead lock. It is safe to not check the master connection
# because it's in the master handler, so the connection is already
# established.
def notifyPartitionChanges(self, conn, packet, ptid, cell_list):
def notifyPartitionChanges(self, conn, ptid, cell_list):
pt = self.app.pt
if pt.filled():
pt.update(ptid, cell_list, self.app.nm)
def sendPartitionTable(self, conn, packet, ptid, row_list):
def sendPartitionTable(self, conn, ptid, row_list):
self.app.pt.load(ptid, row_list, self.app.nm)
def notifyNodeInformation(self, conn, packet, node_list):
def notifyNodeInformation(self, conn, node_list):
app = self.app
self.app.nm.update(node_list)
for node_type, addr, uuid, state in node_list:
......@@ -169,16 +169,16 @@ class PrimaryNotificationsHandler(BaseHandler):
class PrimaryAnswersHandler(AnswerBaseHandler):
""" Handle that process expected packets from the primary master """
def answerBeginTransaction(self, conn, packet, tid):
def answerBeginTransaction(self, conn, tid):
app = self.app
app.setTID(tid)
def answerNewOIDs(self, conn, packet, oid_list):
def answerNewOIDs(self, conn, oid_list):
app = self.app
app.new_oid_list = oid_list
app.new_oid_list.reverse()
def answerTransactionFinished(self, conn, packet, tid):
def answerTransactionFinished(self, conn, tid):
app = self.app
if tid == app.getTID():
app.setTransactionFinished()
......
......@@ -48,11 +48,11 @@ class StorageEventHandler(BaseHandler):
class StorageBootstrapHandler(AnswerBaseHandler):
""" Handler used when connecting to a storage node """
def notReady(self, conn, packet, message):
def notReady(self, conn, message):
app = self.app
app.setNodeNotReady()
def acceptIdentification(self, conn, packet, node_type,
def acceptIdentification(self, conn, node_type,
uuid, num_partitions, num_replicas, your_uuid):
app = self.app
node = app.nm.getByAddress(conn.getAddress())
......@@ -67,24 +67,24 @@ class StorageBootstrapHandler(AnswerBaseHandler):
class StorageAnswersHandler(AnswerBaseHandler):
""" Handle all messages related to ZODB operations """
def answerObject(self, conn, packet, oid, start_serial, end_serial,
def answerObject(self, conn, oid, start_serial, end_serial,
compression, checksum, data):
app = self.app
app.local_var.asked_object = (oid, start_serial, end_serial,
compression, checksum, data)
def answerStoreObject(self, conn, packet, conflicting, oid, serial):
def answerStoreObject(self, conn, conflicting, oid, serial):
app = self.app
if conflicting:
app.local_var.object_stored = -1, serial
else:
app.local_var.object_stored = oid, serial
def answerStoreTransaction(self, conn, packet, tid):
def answerStoreTransaction(self, conn, tid):
app = self.app
app.setTransactionVoted()
def answerTransactionInformation(self, conn, packet, tid,
def answerTransactionInformation(self, conn, tid,
user, desc, ext, oid_list):
app = self.app
# transaction information are returned as a dict
......@@ -96,12 +96,12 @@ class StorageAnswersHandler(AnswerBaseHandler):
info['oids'] = oid_list
app.local_var.txn_info = info
def answerObjectHistory(self, conn, packet, oid, history_list):
def answerObjectHistory(self, conn, oid, history_list):
app = self.app
# history_list is a list of tuple (serial, size)
app.local_var.history = oid, history_list
def oidNotFound(self, conn, packet, message):
def oidNotFound(self, conn, message):
app = self.app
# This can happen either when :
# - loading an object
......@@ -109,12 +109,12 @@ class StorageAnswersHandler(AnswerBaseHandler):
app.local_var.asked_object = -1
app.local_var.history = -1
def tidNotFound(self, conn, packet, message):
def tidNotFound(self, conn, message):
app = self.app
# This can happen when requiring txn informations
app.local_var.txn_info = -1
def answerTIDs(self, conn, packet, tid_list):
def answerTIDs(self, conn, tid_list):
app = self.app
app.local_var.node_tids[conn.getUUID()] = tid_list
......@@ -44,208 +44,208 @@ class PacketLogger(EventHandler):
except PacketMalformedError:
logging.warning("Can't decode packet for logging")
return
log_message = logger(conn, packet, *args)
log_message = logger(conn, *args)
if log_message is not None:
logging.debug('#0x%08x %s', packet.getId(), log_message)
# Packet loggers
def error(self, conn, packet, code, message):
def error(self, conn, code, message):
return "%s (%s)" % (code, message)
def requestIdentification(self, conn, packet, node_type,
def requestIdentification(self, conn, node_type,
uuid, address, name):
logging.debug('Request identification for cluster %s' % (name, ))
pass
def acceptIdentification(self, conn, packet, node_type,
def acceptIdentification(self, conn, node_type,
uuid, num_partitions, num_replicas, your_uuid):
pass
def askPrimary(self, conn, packet):
def askPrimary(self, conn):
pass
def answerPrimary(self, conn, packet, primary_uuid,
def answerPrimary(self, conn, primary_uuid,
known_master_list):
pass
def announcePrimary(self, conn, packet):
def announcePrimary(self, conn):
pass
def reelectPrimary(self, conn, packet):
def reelectPrimary(self, conn):
pass
def notifyNodeInformation(self, conn, packet, node_list):
def notifyNodeInformation(self, conn, node_list):
pass
def askLastIDs(self, conn, packet):
def askLastIDs(self, conn):
pass
def answerLastIDs(self, conn, packet, loid, ltid, lptid):
def answerLastIDs(self, conn, loid, ltid, lptid):
pass
def askPartitionTable(self, conn, packet, offset_list):
def askPartitionTable(self, conn, offset_list):
pass
def answerPartitionTable(self, conn, packet, ptid, row_list):
def answerPartitionTable(self, conn, ptid, row_list):
pass
def sendPartitionTable(self, conn, packet, ptid, row_list):
def sendPartitionTable(self, conn, ptid, row_list):
pass
def notifyPartitionChanges(self, conn, packet, ptid, cell_list):
def notifyPartitionChanges(self, conn, ptid, cell_list):
pass
def startOperation(self, conn, packet):
def startOperation(self, conn):
pass
def stopOperation(self, conn, packet):
def stopOperation(self, conn):
pass
def askUnfinishedTransactions(self, conn, packet):
def askUnfinishedTransactions(self, conn):
pass
def answerUnfinishedTransactions(self, conn, packet, tid_list):
def answerUnfinishedTransactions(self, conn, tid_list):
pass
def askObjectPresent(self, conn, packet, oid, tid):
def askObjectPresent(self, conn, oid, tid):
pass
def answerObjectPresent(self, conn, packet, oid, tid):
def answerObjectPresent(self, conn, oid, tid):
pass
def deleteTransaction(self, conn, packet, tid):
def deleteTransaction(self, conn, tid):
pass
def commitTransaction(self, conn, packet, tid):
def commitTransaction(self, conn, tid):
pass
def askBeginTransaction(self, conn, packet, tid):
def askBeginTransaction(self, conn, tid):
pass
def answerBeginTransaction(self, conn, packet, tid):
def answerBeginTransaction(self, conn, tid):
pass
def askNewOIDs(self, conn, packet, num_oids):
def askNewOIDs(self, conn, num_oids):
pass
def answerNewOIDs(self, conn, packet, num_oids):
def answerNewOIDs(self, conn, num_oids):
pass
def finishTransaction(self, conn, packet, oid_list, tid):
def finishTransaction(self, conn, oid_list, tid):
pass
def answerTransactionFinished(self, conn, packet, tid):
def answerTransactionFinished(self, conn, tid):
pass
def lockInformation(self, conn, packet, tid):
def lockInformation(self, conn, tid):
pass
def notifyInformationLocked(self, conn, packet, tid):
def notifyInformationLocked(self, conn, tid):
pass
def invalidateObjects(self, conn, packet, oid_list, tid):
def invalidateObjects(self, conn, oid_list, tid):
pass
def notifyUnlockInformation(self, conn, packet, tid):
def notifyUnlockInformation(self, conn, tid):
pass
def askStoreObject(self, conn, packet, oid, serial,
def askStoreObject(self, conn, oid, serial,
compression, checksum, data, tid):
pass
def answerStoreObject(self, conn, packet, conflicting, oid, serial):
def answerStoreObject(self, conn, conflicting, oid, serial):
pass
def abortTransaction(self, conn, packet, tid):
def abortTransaction(self, conn, tid):
pass
def askStoreTransaction(self, conn, packet, tid, user, desc,
def askStoreTransaction(self, conn, tid, user, desc,
ext, oid_list):
pass
def answerStoreTransaction(self, conn, packet, tid):
def answerStoreTransaction(self, conn, tid):
pass
def askObject(self, conn, packet, oid, serial, tid):
def askObject(self, conn, oid, serial, tid):
pass
def answerObject(self, conn, packet, oid, serial_start,
def answerObject(self, conn, oid, serial_start,
serial_end, compression, checksum, data):
pass
def askTIDs(self, conn, packet, first, last, partition):
def askTIDs(self, conn, first, last, partition):
pass
def answerTIDs(self, conn, packet, tid_list):
def answerTIDs(self, conn, tid_list):
pass
def askTransactionInformation(self, conn, packet, tid):
def askTransactionInformation(self, conn, tid):
pass
def answerTransactionInformation(self, conn, packet, tid,
def answerTransactionInformation(self, conn, tid,
user, desc, ext, oid_list):
pass
def askObjectHistory(self, conn, packet, oid, first, last):
def askObjectHistory(self, conn, oid, first, last):
pass
def answerObjectHistory(self, conn, packet, oid, history_list):
def answerObjectHistory(self, conn, oid, history_list):
pass
def askOIDs(self, conn, packet, first, last, partition):
def askOIDs(self, conn, first, last, partition):
pass
def answerOIDs(self, conn, packet, oid_list):
def answerOIDs(self, conn, oid_list):
pass
def askPartitionList(self, conn, packet, min_offset, max_offset, uuid):
def askPartitionList(self, conn, min_offset, max_offset, uuid):
pass
def answerPartitionList(self, conn, packet, ptid, row_list):
def answerPartitionList(self, conn, ptid, row_list):
pass
def askNodeList(self, conn, packet, offset_list):
def askNodeList(self, conn, offset_list):
pass
def answerNodeList(self, conn, packet, node_list):
def answerNodeList(self, conn, node_list):
pass
def setNodeState(self, conn, packet, uuid, state, modify_partition_table):
def setNodeState(self, conn, uuid, state, modify_partition_table):
pass
def answerNodeState(self, conn, packet, uuid, state):
def answerNodeState(self, conn, uuid, state):
pass
def addPendingNodes(self, conn, packet, uuid_list):
def addPendingNodes(self, conn, uuid_list):
pass
def answerNewNodes(self, conn, packet, uuid_list):
def answerNewNodes(self, conn, uuid_list):
pass
def askNodeInformation(self, conn, packet):
def askNodeInformation(self, conn):
pass
def answerNodeInformation(self, conn, packet):
def answerNodeInformation(self, conn):
pass
def askClusterState(self, conn, packet):
def askClusterState(self, conn):
pass
def answerClusterState(self, conn, packet, state):
def answerClusterState(self, conn, state):
pass
def setClusterState(self, conn, packet, state):
def setClusterState(self, conn, state):
pass
def notifyClusterInformation(self, conn, packet, state):
def notifyClusterInformation(self, conn, state):
pass
def notifyLastOID(self, conn, packet, oid):
def notifyLastOID(self, conn, oid):
pass
def notifyReplicationDone(self, conn, packet, offset):
def notifyReplicationDone(self, conn, offset):
pass
......
......@@ -23,10 +23,10 @@ from neo.protocol import NodeTypes, NodeStates, Packets
class MasterHandler(EventHandler):
"""This class implements a generic part of the event handlers."""
def protocolError(self, conn, packet, message):
def protocolError(self, conn, message):
logging.error('Protocol error %s %s' % (message, conn.getAddress()))
def askPrimary(self, conn, packet):
def askPrimary(self, conn):
if conn.getConnector() is None:
# Connection can be closed by peer after he sent AskPrimary
# if he finds the primary master before we answer him.
......@@ -50,24 +50,22 @@ class MasterHandler(EventHandler):
conn.answer(Packets.AnswerPrimary(
primary_uuid,
known_master_list),
packet.getId(),
)
def askClusterState(self, conn, packet):
def askClusterState(self, conn):
assert conn.getUUID() is not None
state = self.app.getClusterState()
conn.answer(Packets.AnswerClusterState(state), packet.getId())
conn.answer(Packets.AnswerClusterState(state))
def askNodeInformation(self, conn, packet):
def askNodeInformation(self, conn):
self.app.sendNodesInformations(conn)
conn.answer(Packets.AnswerNodeInformation(), packet.getId())
conn.answer(Packets.AnswerNodeInformation())
def askPartitionTable(self, conn, packet, offset_list):
def askPartitionTable(self, conn, offset_list):
assert len(offset_list) == 0
app = self.app
app.sendPartitionTable(conn)
conn.answer(Packets.AnswerPartitionTable(app.pt.getID(), []),
packet.getId())
conn.answer(Packets.AnswerPartitionTable(app.pt.getID(), []))
DISCONNECTED_STATE_DICT = {
......
......@@ -29,20 +29,20 @@ class AdministrationHandler(MasterHandler):
node = self.app.nm.getByUUID(conn.getUUID())
self.app.nm.remove(node)
def askPrimary(self, conn, packet):
def askPrimary(self, conn):
app = self.app
# I'm the primary
conn.answer(Packets.AnswerPrimary(app.uuid, []), packet.getId())
conn.answer(Packets.AnswerPrimary(app.uuid, []))
def setClusterState(self, conn, packet, state):
def setClusterState(self, conn, state):
self.app.changeClusterState(state)
p = protocol.ack('cluster state changed')
conn.answer(p, packet.getId())
conn.answer(p)
if state == ClusterStates.STOPPING:
self.app.cluster_state = state
self.app.shutdown()
def setNodeState(self, conn, packet, uuid, state, modify_partition_table):
def setNodeState(self, conn, uuid, state, modify_partition_table):
logging.info("set node state for %s-%s : %s" %
(dump(uuid), state, modify_partition_table))
app = self.app
......@@ -55,13 +55,13 @@ class AdministrationHandler(MasterHandler):
# get message for self
if state != NodeStates.RUNNING:
p = protocol.ack('node state changed')
conn.answer(p, packet.getId())
conn.answer(p)
app.shutdown()
if node.getState() == state:
# no change, just notify admin node
p = protocol.ack('node state changed')
conn.answer(p, packet.getId())
conn.answer(p)
return
if state == NodeStates.RUNNING:
......@@ -88,10 +88,10 @@ class AdministrationHandler(MasterHandler):
# /!\ send the node information *after* the partition table change
node.setState(state)
p = protocol.ack('state changed')
conn.answer(p, packet.getId())
conn.answer(p)
app.broadcastNodesInformation([node])
def addPendingNodes(self, conn, packet, uuid_list):
def addPendingNodes(self, conn, uuid_list):
uuids = ', '.join([dump(uuid) for uuid in uuid_list])
logging.debug('Add nodes %s' % uuids)
app, nm, em, pt = self.app, self.app.nm, self.app.em, self.app.pt
......@@ -108,7 +108,7 @@ class AdministrationHandler(MasterHandler):
if not uuid_set:
logging.warning('No nodes added')
p = protocol.ack('no nodes added')
conn.answer(p, packet.getId())
conn.answer(p)
return
uuids = ', '.join([dump(uuid) for uuid in uuid_set])
logging.info('Adding nodes %s' % uuids)
......@@ -127,4 +127,4 @@ class AdministrationHandler(MasterHandler):
# broadcast the new partition table
app.broadcastPartitionChanges(cell_list)
p = protocol.ack('node added')
conn.answer(p, packet.getId())
conn.answer(p)
......@@ -33,22 +33,22 @@ class ClientServiceHandler(BaseServiceHandler):
self.app.tm.abortFor(node)
self.app.nm.remove(node)
def abortTransaction(self, conn, packet, tid):
def abortTransaction(self, conn, tid):
if tid in self.app.tm:
self.app.tm.remove(tid)
else:
logging.warn('aborting transaction %s does not exist', dump(tid))
def askBeginTransaction(self, conn, packet, tid):
def askBeginTransaction(self, conn, tid):
node = self.app.nm.getByUUID(conn.getUUID())
tid = self.app.tm.begin(node, tid)
conn.answer(Packets.AnswerBeginTransaction(tid), packet.getId())
conn.answer(Packets.AnswerBeginTransaction(tid))
def askNewOIDs(self, conn, packet, num_oids):
def askNewOIDs(self, conn, num_oids):
oid_list = self.app.getNewOIDList(num_oids)
conn.answer(Packets.AnswerNewOIDs(oid_list), packet.getId())
conn.answer(Packets.AnswerNewOIDs(oid_list))
def finishTransaction(self, conn, packet, oid_list, tid):
def finishTransaction(self, conn, oid_list, tid):
app = self.app
# If the given transaction ID is later than the last TID, the peer
# is crazy.
......@@ -76,5 +76,5 @@ class ClientServiceHandler(BaseServiceHandler):
c.ask(Packets.LockInformation(tid), timeout=60)
used_uuid_set.add(c.getUUID())
app.tm.prepare(tid, oid_list, used_uuid_set, packet.getId())
app.tm.prepare(tid, oid_list, used_uuid_set, conn.getPeerId())
......@@ -25,7 +25,7 @@ from neo.exception import ElectionFailure
class ElectionHandler(MasterHandler):
"""This class deals with events for a primary master election."""
def notifyNodeInformation(self, conn, packet, node_list):
def notifyNodeInformation(self, conn, node_list):
uuid = conn.getUUID()
if uuid is None:
raise protocol.ProtocolError('Not identified')
......@@ -66,7 +66,7 @@ class ClientElectionHandler(ElectionHandler):