Merge lp:~codehelp/lava-dispatcher/multinode into lp:lava-dispatcher/multinode

[Neil Williams]

This is the reworking of the group key=value pair handling done at Connect and which complements Fu's change for wait_all.

This results in:

<LAVA_WAIT_ALL_DEBUG lava_multi_node_send preparing Sun Jul 14 13:00:04 UTC 2013>
<LAVA_WAIT_ALL_DEBUG _lava_multi_node_send started Sun Jul 14 13:00:04 UTC 2013>
<LAVA_MULTI_NODE> <LAVA_WAIT_ALL get_source client>
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: Received Multi_Node API <LAVA_WAIT_ALL>
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: Handling signal <LAVA_WAIT_ALL get_source>
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: transport handler for NodeDispatcher {"role": "client", "request": "lava_wait_all", "messageID": "get_source"}
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM INFO: requesting lava_wait_all {"request": "lava_wait_all", "role": "client", "messageID": "get_source"}
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: sending Message {"client_name": "kvm01", "hostname": "sylvester", "request": "lava_wait_all", "group_name": "group01", "host": "sylvester", "role": "client", "messageID": "get_source", "port": 3079}
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: polling {"client_name": "kvm01", "hostname": "sylvester", "group_name": "group01", "host": "sylvester", "role": "client", "port": 3079}
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: socket created for host:sylvester port:3079
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: read message: {"client_name": "kvm01", "hostname": "sylvester", "request": "lava_wait_all", "group_name": "group01", "host": "sylvester", "role": "client", "messageID": "get_source", "port": 3079}
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM INFO: Response: ack
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: Node transport replied with {u'kvm01': {u'source': u'kvm01', u'role': u'client'}, u'kvm02': {u'source': u'kvm02', u'role': u'client'}}
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: sendline : <LAVA_WAIT_ALL_COMPLETE kvm01:source=kvm01 kvm01:role=client kvm02:source=kvm02 kvm02:role=client>
<LAVA_DISPATCHER>2013-07-14 02:00:04 PM DEBUG: send : <LAVA_WAIT_ALL_COMPLETE kvm01:source=kvm01 kvm01:role=client kvm02:source=kvm02 kvm02:role=client>
<LAVA_DISPATCHER>2013-07-14 02:00:09 PM DEBUG: send :

The output sent from the GroupDispatcher is:
{
    "kvm01": {
        "source": "kvm01",
        "role": "client"
    },
    "kvm02": {
        "source": "kvm02",
        "role": "client"
    }
}

This converts to a message of:
kvm01:source=kvm01 kvm01:role=client kvm02:source=kvm02 kvm02:role=client

[Neil Williams (codehelp)]

There's a parallel change in signals/__init__.py which goes along with this change:

=== modified file 'lava_dispatcher/signals/__init__.py'
--- lava_dispatcher/signals/__init__.py 2013-07-10 10:36:36 +0000
+++ lava_dispatcher/signals/__init__.py 2013-07-16 08:14:35 +0000
@@ -207,8 +207,11 @@
         reply = self.context.transport(json.dumps(msg))
         logging.debug("Node transport replied with %s" % reply)
         message_str = ""
- for key, value in reply[0].items():
- message_str += " %s=%s" % (key, value)
+ for target, messages in reply.items():
+ for key, value in messages.items():
+ message_str += " %s:%s=%s" % (target, key, value)
+# for key, value in reply[0].items():
+# message_str += " %s=%s" % (key, value)
         self.connection.sendline("<LAVA_WAIT_COMPLETE%s>" % message_str)

     def _on_WAIT_ALL(self, message_id, role=None):

With this change, lava-wait gives this kind of output:

<LAVA_DISPATCHER>2013-07-16 08:37:18 AM DEBUG: Node transport replied with
{
    "panda21": {
        "ip": "192.168.23.11",
        "hostname": "panda21.localdomain"
    },
    "multinode-arndale01": {
        "ip": "192.168.1.79"
    },
    "panda19": {
        "ip": "192.168.127.216",
        "hostname": "panda19.localdomain"
    }
}

<LAVA_DISPATCHER>2013-07-16 08:37:18 AM DEBUG: send : <LAVA_WAIT_COMPLETE
panda21:ip=192.168.23.11
panda21:hostname=panda21.localdomain

multinode-arndale01:ip=192.168.1.79

panda19:ip=192.168.127.216
panda19:hostname=panda19.localdomain
>

[Neil Williams (codehelp)]

(I added the formatting of the output manually, the actual output in each case is on a single line)

[Neil Williams (codehelp)]

(for whatever reason, the arndale does not currently support hostname -f, so the fqdn is not available and the code omits a key which has no value.)

[Fu Wei (fu-wei)]

I have tested it , it works fine. Good patch!

review: Approve

=== modified file 'lava/dispatcher/group.py'
--- lava/dispatcher/group.py	2013-07-05 11:48:41 +0000
+++ lava/dispatcher/group.py	2013-07-14 13:26:24 +0000
@@ -20,7 +20,6 @@
 import json
 import time
 import socket
-import copy
 
 
 class GroupDispatcher(object):
@@ -88,7 +87,6 @@
             logging.error("Missing client_name in request: %s" % json_data)
             return None
         if json_data['group_name'] not in self.all_groups:
-            print json.dumps(json_data)
             if "group_size" not in json_data or json_data["group_size"] == 0:
                 logging.error('%s asked for a new group %s without specifying the size of the group'
                               % (client_name, json_data['group_name']))
@@ -151,7 +149,8 @@
             self._badRequest()
             return
         logging.info("Sending messageID '%s' to %s in group %s: %s" %
-                     (messageID, client_name, self.group['group'], json.dumps(self.group['messages'][client_name][messageID])))
+                     (messageID, client_name, self.group['group'],
+                      json.dumps(self.group['messages'][client_name][messageID])))
         msg = {"response": "ack", "message": self.group['messages'][client_name][messageID]}
         logging.info("Sending response to %s in group %s: %s" %
                      (client_name, self.group['group'], json.dumps(msg)))
@@ -192,13 +191,15 @@
         Global synchronization primitive. Sends a message and waits for the same
         message from all of the other devices.
         """
-        logging.debug("GroupDispatcher:lavaSync %s from %s in group %s" %(json.dumps(json_data), client_name, self.group['group']))
+        logging.debug("GroupDispatcher:lavaSync %s from %s in group %s" %
+                      (json.dumps(json_data), client_name, self.group['group']))
         messageID = self._getMessageID(json_data)
         message = self._getMessage(json_data)
         # FIXME: in _sendMessage, be sure to send the messageID if message is empty
         if not message:
             message = messageID
-        logging.info("LavaSync request for '%s' at stage '%s' in group '%s'" % (client_name, messageID, self.group['group']))
+        logging.info("LavaSync request for '%s' at stage '%s' in group '%s'" %
+                     (client_name, messageID, self.group['group']))
         self.group['syncs'].setdefault(messageID, {})
         self.group['messages'].setdefault(client_name, {}).setdefault(messageID, {})
         if len(self.group['syncs'][messageID]) >= self.group['count']:
@@ -230,21 +231,13 @@
         """
         messageID = self._getMessageID(json_data)
         if 'role' in json_data:
-            role_msg = {}
+            logging.debug("setting message: %s for %s" % (self.group['messages'][client_name][messageID], client_name))
             for client in self.group['roles'][json_data['role']]:
                 if client not in self.group['messages'] or messageID not in self.group['messages'][client]:
                     self._waitResponse()
                     return
-                # combine all messages for this messageID into a single message for the entire role.
-                role_msg[client] = copy.deepcopy(self.group['messages'][client][messageID])
-            msg = {}
-            # build a single structure with all of the data for all clients in the role
-            for client in self.group['roles'][json_data['role']]:
-                msg[client] = role_msg[client]
-            del self.group['messages'][client_name][messageID]
-            # now put all of the data in msg into the messageID for all clients with this role
-            for client in self.group['roles'][json_data['role']]:
-                self.group['messages'][client][messageID] = copy.deepcopy(msg)
+                logging.debug("broadcasting: %s for %s" % (self.group['messages'][client][messageID], client))
+            logging.debug("lavaWaitAll message: %s" % json.dumps(self.group['messages'][client_name][messageID]))
         else:
             for client in self.group['clients']:
                 if client not in self.group['messages'] or messageID not in self.group['messages'][client]:
@@ -266,7 +259,7 @@
             return
         self._sendMessage(client_name, messageID)
 
-    def lavaSend(self, json_data):
+    def lavaSend(self, json_data, client_name):
         """
         A message list won't be seen by the destination until the destination
         calls lava_wait or lava_wait_all with the messageID
@@ -274,14 +267,26 @@
         """
         message = self._getMessage(json_data)
         messageID = self._getMessageID(json_data)
-        logging.info("lavaSend handler in GroupDispatcher received a messageID '%s' for group '%s'"
-                     % (messageID, self.group['group']))
+        logging.info("lavaSend handler in GroupDispatcher received a messageID '%s' for group '%s' from %s"
+                     % (messageID, self.group['group'], client_name))
+        if client_name not in self.group['messages']:
+            self.group['messages'][client_name] = {}
+        # construct the message hash which stores the data from each client separately
+        # but which gets returned as a complete hash upon request
+        msg_hash = {}
+        msg_hash.update({client_name: message})
+        # always set this client data if the call is made to update the broadcast
+        if messageID not in self.group['messages'][client_name]:
+            self.group['messages'][client_name][messageID] = {}
+        self.group['messages'][client_name][messageID].update(msg_hash)
+        logging.debug("message %s for %s" % (json.dumps(self.group['messages'][client_name][messageID]), client_name))
         for client in self.group['clients']:
             if client not in self.group['messages']:
                 self.group['messages'][client] = {}
             if messageID not in self.group['messages'][client]:
-                self.group['messages'][client][messageID] = []
-            self.group['messages'][client][messageID].append(message)
+                self.group['messages'][client][messageID] = {}
+            self.group['messages'][client][messageID].update(msg_hash)
+            logging.debug("broadcast %s for %s" % (json.dumps(self.group['messages'][client][messageID]), client))
         self._ackResponse()
 
     def dataReceived(self, json_data):
@@ -315,7 +320,7 @@
             self.lavaWait(json_data, client_name)
         elif request == 'lava_send':
             logging.info("lava_send: %s" % json_data)
-            self.lavaSend(json_data)
+            self.lavaSend(json_data, client_name)
         elif request == "complete":
             logging.info("dispatcher communication for '%s' in group '%s' is complete, closing." %
                          (client_name, self.group['group']))