Replacement of broadcastNodeInformation() with broadcastNodesInformation()

- Handle a node list instead of a single node
- Send only one packet per peer, reduce bandwidth and latency
- Remove the related XXX

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

neo/master/app.py

@@ -265,32 +265,37 @@ class Application(object):
                     conn.close()
                 bootstrap = False
 
-    # XXX: should accept a node list and send at most one packet per peer
-    def broadcastNodeInformation(self, node):
-        """Broadcast a Notify Node Information packet."""
-        logging.debug('broadcasting node information')
-        node_type = node.getType()
-        state = node.getState()
-        uuid = node.getUUID()
-
-        # The server address may be None.
+    def broadcastNodesInformation(self, node_list):
+        """
+          Broadcast changes for a set a nodes
+          Send only one packet per connection to reduce bandwidth
+        """
+        node_dict = {}
+        # group modified nodes by destination node type
+        for node in node_list:
             address = node.getAddress()
-
-        if node.isClient():
-            # Only to master nodes and storage nodes.
-            for c in self.em.getConnectionList():
-                if c.getUUID() is not None:
-                    n = self.nm.getByUUID(c.getUUID())
-                    if n.isMaster() or n.isStorage() or n.isAdmin():
-                        node_list = [(node_type, address, uuid, state)]
-                        c.notify(Packets.NotifyNodeInformation(node_list))
-        elif node.isMaster() or node.isStorage():
-            for c in self.em.getConnectionList():
-                if c.getUUID() is not None:
-                    node_list = [(node_type, address, uuid, state)]
-                    c.notify(Packets.NotifyNodeInformation(node_list))
-        elif not node.isAdmin():
-            raise RuntimeError('unknown node type')
+            uuid = node.getUUID()
+            state = node.getState()
+            node_info = (node.getType(), address, uuid, state)
+            def assign_for_notification(node_type):
+                # helper function
+                node_dict.setdefault(node_type, []).append(node_info)
+            if node.isMaster() or node.isStorage():
+                # client get notifications for master and storage only
+                assign_for_notification(NodeTypes.CLIENT)
+            if node.isMaster() or node.isStorage() or node.isClient():
+                assign_for_notification(NodeTypes.MASTER)
+                assign_for_notification(NodeTypes.STORAGE)
+                assign_for_notification(NodeTypes.ADMIN)
+
+        # send at most one non-empty notification packet per node
+        for conn in self.em.getConnectionList():
+            if conn.getUUID() is None:
+                continue
+            node = self.nm.getByUUID(conn.getUUID())
+            node_list = node_dict.get(node.getType(), [])
+            if node_list:
+                conn.notify(Packets.NotifyNodeInformation(node_list))
 
     def broadcastPartitionChanges(self, cell_list):
         """Broadcast a Notify Partition Changes packet."""
@@ -363,7 +368,7 @@ class Application(object):
         node_list = nm.getStorageList()[:REQUIRED_NODE_NUMBER]
         for node in node_list:
             node.setRunning()
-            self.broadcastNodeInformation(node)
+        self.broadcastNodesInformation(node_list)
         # resert IDs generators
         self.loid = '\0' * 8
         self.ltid = '\0' * 8
@@ -404,7 +409,7 @@ class Application(object):
         refused_node_set = set(self.nm.getStorageList()) - allowed_node_set
         for node in refused_node_set:
             node.setPending()
-            self.broadcastNodeInformation(node)
+        self.broadcastNodesInformation(refused_node_set)
 
         logging.debug('cluster starts with loid=%s and this partition table :',
                 dump(self.loid))

neo/master/handlers/__init__.py

@@ -96,7 +96,7 @@ class BaseServiceHandler(MasterHandler):
             logging.info('drop a pending node from the node manager')
             self.app.nm.remove(node)
         node.setState(new_state)
-        self.app.broadcastNodeInformation(node)
+        self.app.broadcastNodesInformation([node])
         # clean node related data in specialized handlers
         self.nodeLost(conn, node)
 

neo/master/handlers/administration.py

@@ -89,7 +89,7 @@ class AdministrationHandler(MasterHandler):
         node.setState(state)
         p = protocol.noError('state changed')
         conn.answer(p, packet.getId())
-        app.broadcastNodeInformation(node)
+        app.broadcastNodesInformation([node])
 
     def addPendingNodes(self, conn, packet, uuid_list):
         uuids = ', '.join([dump(uuid) for uuid in uuid_list])
@@ -113,12 +113,12 @@ class AdministrationHandler(MasterHandler):
         uuids = ', '.join([dump(uuid) for uuid in uuid_set])
         logging.info('Adding nodes %s' % uuids)
         # switch nodes to running state
-        for uuid in uuid_set:
-            node = nm.getByUUID(uuid)
+        node_list = [nm.getByUUID(uuid) for uuid in uuid_set]
+        for node in node_list:
             new_cells = pt.addNode(node)
             cell_list.extend(new_cells)
             node.setRunning()
-            app.broadcastNodeInformation(node)
+        app.broadcastNodesInformation(node_list)
         # start nodes
         for s_conn in em.getConnectionList():
             if s_conn.getUUID() in uuid_set:

neo/master/handlers/identification.py

@@ -80,5 +80,5 @@ class IdentificationHandler(MasterHandler):
         conn.answer(Packets.AcceptIdentification(*args), packet.getId())
         # trigger the event
         handler.connectionCompleted(conn)
-        app.broadcastNodeInformation(node)
+        app.broadcastNodesInformation([node])
 

neo/master/handlers/secondary.py

@@ -29,7 +29,7 @@ class SecondaryMasterHandler(MasterHandler):
         node = self.app.nm.getByUUID(conn.getUUID())
         assert node is not None
         node.setDown()
-        self.app.broadcastNodeInformation(node)
+        self.app.broadcastNodesInformation([node])
 
     def connectionCompleted(self, conn):
         pass