Merge branch 'remove_client_eof'

Author: langmm

yggdrasil/communication/CommBase.py

@@ -894,6 +894,16 @@ def get_status_message(self, nindent=0, extra_lines_before=None,
         lines += ['%s%s' % (prefix, x) for x in extra_lines_after]
         return lines, prefix
 
+    # Re-enable this once the environment is crystalized on initialization
+    # @property
+    # def print_name(self):
+    #     r"""str: Name of the class object."""
+    #     out = super(CommBase, self).print_name
+    #     model_name = self.full_model_name
+    #     if model_name:
+    #         out += '[%s]' % model_name
+    #     return out
+        
     def printStatus(self, *args, **kwargs):
         r"""Print status of the communicator."""
         nindent = kwargs.get('nindent', 0)
@@ -1986,8 +1996,8 @@ def prepare_message(self, *args, header_kwargs=None, skip_serialization=False,
             elif isinstance(msg.args, bytes) and (msg.args == YGG_CLIENT_EOF):
                 once_per_partner = True
             if once_per_partner and (self.partner_copies > 1):
-                self.info("Sending %s to %d model(s)", msg.args,
-                          self.partner_copies)
+                self.debug("Sending %s to %d model(s)", msg.args,
+                           self.partner_copies)
                 for i in range(self.partner_copies - 1):
                     msg.add_message(args=msg.args,
                                     header=copy.deepcopy(msg.header))

yggdrasil/config.py

@@ -571,8 +571,10 @@ def resolve_config_parser(args):
         args.validate_components = True
         args.validate_messages = True
     else:
-        args.loglevel = 'INFO'
-        args.client_loglevel = 'INFO'
+        if args.loglevel is None:
+            args.loglevel = 'INFO'
+        if args.client_loglevel is None:
+            args.client_loglevel = 'INFO'
     if args.validate_messages in ['True', 'False']:
         args.validate_messages = (args.validate_messages == 'True')
     return args

yggdrasil/drivers/RPCRequestDriver.py

@@ -117,16 +117,16 @@ def remove_model(self, direction, name):
                 self.send_eof()
             return out
 
-    def send_eof(self):
-        r"""Send EOF message.
+    # def send_eof(self):
+    #     r"""Send EOF message.
 
-        Returns:
-            bool: Success or failure of send.
+    #     Returns:
+    #         bool: Success or failure of send.
 
-        """
-        if self.ocomm.partner_copies > 1:
-            self.ocomm.partner_copies = len(self.servers_recvd)
-        return super(RPCRequestDriver, self).send_eof()
+    #     """
+    #     if self.ocomm.partner_copies > 1:
+    #         self.ocomm.partner_copies = len(self.servers_recvd)
+    #     return super(RPCRequestDriver, self).send_eof()
 
     def on_eof(self, msg):
         r"""On EOF, decrement number of clients. Only send EOF if the number

yggdrasil/drivers/RPCResponseDriver.py

@@ -43,6 +43,16 @@ def response_address(self):
         r"""str: Address of response comm."""
         return self.icomm.address
 
+    def send_eof(self):
+        r"""Send EOF message.
+
+        Returns:
+            bool: Success or failure of send.
+
+        """
+        # Don't send EOF
+        return False
+        
     def send_message(self, msg, **kwargs):
         r"""Propagate the request_id.
 

yggdrasil/languages/C/communication/ZMQComm.h

@@ -790,14 +790,20 @@ int free_zmq_comm(comm_t *x) {
     if (_ygg_error_flag == 0) {
       size_t data_len = 100;
       char *data = (char*)malloc(data_len);
+      comm_head_t head;
+      bool is_eof_flag = false;
       while (zmq_comm_nmsg(x) > 0) {
         ret = zmq_comm_recv(x, &data, data_len, 1);
-        if (ret < 0) {
-          if (ret == -2) {
+	if (ret >= 0) {
+	  head = parse_comm_header(data, ret);
+	  if (strncmp(YGG_MSG_EOF, data + head.bodybeg, strlen(YGG_MSG_EOF)) == 0)
+	    is_eof_flag = true;
+	  destroy_header(&head);
+	  if ((head.flags & HEAD_FLAG_VALID) && is_eof_flag) {
 	    x->const_flags[0] = x->const_flags[0] | COMM_EOF_RECV;
-            break;
-          }
-        }
+	    break;
+	  }
+	}
       }
       free(data);
     }