Update rapidjson/python-rapidjson to use the yggdrasil specific versions

Add option for running without redirecting output from models to queue via dont_queue_output
Fix bug where cpp was not checked as language to disable use of direct connection
Fix bug in broker where responses could get sent to the wrong client

Author: langmm

.gitmodules

@@ -1,6 +1,6 @@
 [submodule "yggdrasil/rapidjson"]
 	path = yggdrasil/rapidjson
-	url = https://github.com/cropsinsilico/rapidjson
+	url = https://github.com/cropsinsilico/yggdrasil-rapidjson
 	branch = yggdrasil
 [submodule "yggdrasil/FSPM2020_yggdrasil_workshop"]
 	path = yggdrasil/demos/fspm2020
@@ -11,5 +11,5 @@
 	url = https://github.com/cropsinsilico/CiS2021-hackathon
 [submodule "python-rapidjson"]
 	path = yggdrasil/python-rapidjson
-	url = https://github.com/cropsinsilico/python-rapidjson.git
+	url = https://github.com/cropsinsilico/yggdrasil-python-rapidjson
 	branch = yggdrasil

yggdrasil/.ygg_schema.yml

@@ -3621,6 +3621,10 @@ definitions:
         default: false
         description: If True, the Python C API will be disabled. Defaults to False.
         type: boolean
+      dont_queue_output:
+        default: false
+        description: If True, model output will not be captured.
+        type: boolean
       driver:
         deprecated: true
         description: '[DEPRECATED] Name of driver class that should be used.'

yggdrasil/broker.py

@@ -49,6 +49,7 @@ def __init__(self, yaml=None):
             response_kwargs={'datatype': {'type': 'any'}},
         )
         self._delayed_requests = []
+        self._awaiting_comm = []
         self._current_request = None
         self._comms = None
         self._connections = {}
@@ -230,12 +231,14 @@ def run_loop(self):
             timeout=0.1, return_message_object=True,
             quiet_timeout=True)
         if msg.flag in [CommBase.FLAG_EOF, CommBase.FLAG_EMPTY]:
+            if self._delayed_requests:
+                self.complete_request(self._delayed_requests.pop(0))
             return
         elif msg.flag != CommBase.FLAG_SUCCESS:
             raise BrokerError(f"Error receiving inside loop: "
                               f"{CommBase.FLAG_TO_STRING[msg.flag]}")
-        if self._delayed_requests:
-            self.complete_request(self._delayed_requests.pop(0))
+        msg.args['id'] = msg.header['__meta__']['request_id']
+        # self.server_comm.info(f"REQUEST [{msg.args['action']}]")
         self.complete_request(msg.args)
 
     def run_finally(self):
@@ -264,15 +267,21 @@ def complete_request(self, request):
         """
         error = None
         try:
+            # self.server_comm.info(f"PROCESS REQUEST "
+            #                       f"[{request['action']}]: {request}")
             response = self.process_request(request)
         except DelayedRequestError:
+            # self.server_comm.info(f"DELAYED REQUEST "
+            #                       f"[{request['action']}]")
             self._delayed_requests.append(request)
             return
         except BaseException as e:
             error = e
             tb = traceback.format_exc()
             response = {'error': str(e), 'traceback': tb}
-        flag = self.server_comm.send(response)
+        # self.server_comm.info(f"RESPONSE [{request['action']}]: "
+        #                       f"{response}")
+        flag = self.server_comm.send_to(request['id'], response)
         if error:
             raise error
         if not flag:
@@ -483,15 +492,22 @@ def _send_request(cls, action, *args, **kwargs):
             'kwargs': kwargs,
         }
         client = cls.get_client()
-        flag, response = client.call(request)
+        # client.info(f"REQUEST [{action}]")
+        flag = client.send(request)
         if not flag:
             raise BrokerError(f"{model}: Failed to send request to "
                               f"the broker {request}")
+        msg = client.recv(return_message_object=True, timeout=False)
+        if msg.flag != CommBase.FLAG_SUCCESS:
+            raise BrokerError(f"{model}: Failed to receive response from "
+                              f"the broker {request} ({msg.flag})")
+        response = msg.args
         if 'error' in response:
             raise BrokerError(f"{model}: Error on broker process during "
                               f"response to request {request}.\n"
                               f"{response['error']}\n"
                               f"{response['traceback']}")
+        # client.info(f"RESPONSE [{action}]: {response['return']}")
         return response['return']
 
     @classmethod
@@ -513,6 +529,10 @@ def update_model_comm_kwargs(cls, name, kwargs, self=None):
         assert isinstance(comm, dict)
         comm.update(kwargs)
         assert comm['direction'] == direction
+        if name in self._awaiting_comm:
+            assert 'address' in comm
+            # self.server_comm.info(f"UNLOCKING: {name}")
+            self._awaiting_comm.remove(name)
 
     @classmethod
     def model_comm_kwargs(cls, name, direction, self=None):
@@ -536,6 +556,11 @@ def model_comm_kwargs(cls, name, direction, self=None):
         out = {}
         is_split_server = False
         language = None
+        if name in self._awaiting_comm:
+            # self.server_comm.info(f"LOCKED {name}")
+            raise DelayedRequestError(
+                f"Delaying creationg of {name} comm until "
+                f"the partner comm finishing creation")
         with self.lock:
             model_driver = self._models[model]
             language = model_driver.language
@@ -564,6 +589,9 @@ def model_comm_kwargs(cls, name, direction, self=None):
         )
         if is_split_server or out['commtype'] == 'model_function':
             out['global_scope'] = model
+        if 'address' not in out and 'partner_name' in out:
+            # self.server_comm.info(f"LOCKING {out['partner_name']}")
+            self._awaiting_comm.append(out['partner_name'])
         return out
 
     @classmethod

yggdrasil/communication/AsyncComm.py

@@ -478,7 +478,7 @@ def recv_message(self, timeout=None, **kwargs):
         """
         # Sleep until there is a message
         if timeout is None:
-            timeout = kwargs.get('timeout', self.recv_timeout)
+            timeout = self.recv_timeout
         T = self.start_timeout(timeout, key_suffix='.recv:backlog')
         while (not T.is_out) and (not self.backlog_ready.is_set()):
             self.backlog_ready.wait(self.sleeptime)

yggdrasil/drivers/DirectConnectionDriver.py

@@ -60,7 +60,7 @@ def _init_single_comm(self, io, comm_list):
                 x['model_copies'] = x.pop('partner_copies')
             if 'partner_language' in x:
                 x['language'] = x.pop('partner_language')
-                if x['language'] in ['c', 'c++', 'fortran']:
+                if x['language'] in ['c', 'c++', 'cpp', 'fortran']:
                     raise DirectConnectionError(
                         "C use of comm broker not yet implemented")
             x.setdefault('commtype', comm_type)

yggdrasil/drivers/ModelDriver.py

@@ -418,6 +418,8 @@ class ModelDriver(Driver):
         preserve_cache (bool, optional): [DEPRECATED] If True model
             products will be kept following the run, otherwise all
             products will be cleaned up.
+        dont_queue_output (bool, optional): If True, model output will not
+            be captured.
         with_strace (bool, optional): If True, the command is run with strace (on
             Linux) or dtrace (on MacOS). Defaults to False.
         strace_flags (list, optional): Flags to pass to strace (or dtrace).
@@ -673,6 +675,7 @@ class ModelDriver(Driver):
         'overwrite': {'type': 'boolean', 'default': False},
         'remove_products': {'type': 'boolean', 'default': False},
         'preserve_cache': {'type': 'boolean', 'default': True},
+        'dont_queue_output': {'type': 'boolean', 'default': False},
         'function': {'type': 'string'},
         'iter_function_over': {'type': 'array', 'default': [],
                                'items': {'type': 'string'}},
@@ -827,8 +830,10 @@ def __init__(self, name, args, model_index=0, copy_index=-1, clients=[],
         # TODO: Pass removable_source_exts used by Compiled drivers
         # Setup process things
         self.model_process = None
-        self.queue = multitasking.Queue()
+        self.queue = None
         self.queue_thread = None
+        if not self.dont_queue_output:
+            self.queue = multitasking.Queue()
         self.event_process_kill_called = multitasking.Event()
         self.event_process_kill_complete = multitasking.Event()
         # Tools
@@ -1525,7 +1530,8 @@ def run_model(self, command=None, return_process=True, **kwargs):
         # not running locally.
         default_kwargs = dict(env=env, working_dir=self.working_dir,
                               forward_signals=False,
-                              shell=platform._is_win)
+                              shell=platform._is_win,
+                              allow_buffer=self.dont_queue_output)
         for k, v in default_kwargs.items():
             kwargs.setdefault(k, v)
         return self.run_executable(command, return_process=return_process, **kwargs)
@@ -2025,7 +2031,7 @@ def before_start(self, no_queue_thread=False, **kwargs):
         """
         self.model_process = self.run_model(**kwargs)
         # Start thread to queue output
-        if not no_queue_thread:
+        if not (no_queue_thread or self.dont_queue_output):
             self.queue_thread = multitasking.YggTaskLoop(
                 target=self.enqueue_output_loop,
                 name=self.name + '.EnqueueLoop')
@@ -2035,10 +2041,12 @@ def before_start(self, no_queue_thread=False, **kwargs):
 
     def queue_close(self):
         r"""Close the queue for messages from the model process."""
-        self.model_process.stdout.close()
+        if not self.dont_queue_output:
+            self.model_process.stdout.close()
 
     def queue_recv(self):
         r"""Receive a message from the model process."""
+        assert not self.dont_queue_output
         return self.model_process.stdout.readline()
 
     def enqueue_output_loop(self):
@@ -2185,6 +2193,13 @@ def run_loop(self):
             self.set_break_flag()
         for k in self._mpi_requests_checked:
             self.check_mpi_request(k)
+        if self.dont_queue_output:
+            if self.model_process_complete:
+                self.debug("Model process complete")
+                self.set_break_flag()
+            else:
+                self.sleep()
+            return
         try:
             line = self.queue.get_nowait()
         except Empty:

yggdrasil/python-rapidjson

@@ -1576,17 +1576,19 @@ class YggPopen(subprocess.Popen):
         **kwargs: Additional keywords arguments are passed to Popen.
 
     """
-    def __init__(self, cmd_args, forward_signals=True, for_matlab=False, **kwargs):
+    def __init__(self, cmd_args, forward_signals=True, for_matlab=False,
+                 allow_buffer=False, **kwargs):
         # stdbuf only for linux
         if platform._is_linux:
             stdbuf_args = ['stdbuf', '-o0', '-e0']
             if isinstance(cmd_args, str):
                 cmd_args = ' '.join(stdbuf_args + [cmd_args])
             else:
                 cmd_args = stdbuf_args + cmd_args
-        kwargs.setdefault('bufsize', 0)
-        kwargs.setdefault('stdout', subprocess.PIPE)
-        kwargs.setdefault('stderr', subprocess.STDOUT)
+        if not allow_buffer:
+            kwargs.setdefault('bufsize', 0)
+            kwargs.setdefault('stdout', subprocess.PIPE)
+            kwargs.setdefault('stderr', subprocess.STDOUT)
         # To prevent forward of signals, process will have a new process group
         if not forward_signals:
             if platform._is_win:  # pragma: windows