Introduce non-eager task-submission system

environment.yml

@@ -1,7 +1,6 @@
 name: stride
 channels:
   - conda-forge
-  - defaults
 dependencies:
   - python>=3.8, <3.12
   - blosc

mosaic/comms/comms.py

@@ -780,7 +780,7 @@ def _process_send(self, method, cmd=None, reply=False, **kwargs):
 
                 self.logger.debug('Sending cmd %s %s to %s (%s) from %s '
                                   '(size %.2f MB)' % (method, cmd['method'], self.uid, cmd['uid'],
-                                                   self._runtime.uid, msg_size/1024**2))
+                                                      self._runtime.uid, msg_size/1024**2))
             else:
                 self.logger.debug('Sending msg %s to %s from %s '
                                   '(size %.2f MB)' % (method, self.uid, self._runtime.uid,

mosaic/core/base.py

@@ -228,11 +228,17 @@ def cmd(self, method, *args, **kwargs):
 
         """
         wait = kwargs.pop('wait', False)
+        silence = kwargs.pop('silence', None)
+        restrict = kwargs.pop('restrict', None)
 
         remotes, cmd = self._prepare_cmd(method, *args, **kwargs)
 
         result = []
         for remote in remotes:
+            if silence is not None and remote.uid in silence:
+                continue
+            if restrict is not None and remote.uid not in restrict:
+                continue
             result.append(remote.cmd(**cmd, wait=wait, as_async=False))
 
         if len(result) == 1:
@@ -259,11 +265,17 @@ def cmd_recv(self, method, *args, **kwargs):
 
         """
         wait = kwargs.pop('wait', False)
+        silence = kwargs.pop('silence', None)
+        restrict = kwargs.pop('restrict', None)
 
         remotes, cmd = self._prepare_cmd(method, *args, **kwargs)
 
         result = []
         for remote in remotes:
+            if silence is not None and remote.uid in silence:
+                continue
+            if restrict is not None and remote.uid not in restrict:
+                continue
             result.append(remote.cmd(**cmd, wait=wait, reply=True, as_async=False))
 
         if len(result) == 1:
@@ -289,10 +301,17 @@ async def cmd_async(self, method, *args, **kwargs):
         asyncio.Future
 
         """
+        silence = kwargs.pop('silence', None)
+        restrict = kwargs.pop('restrict', None)
+
         remotes, cmd = self._prepare_cmd(method, *args, **kwargs)
 
         result = []
         for remote in remotes:
+            if silence is not None and remote.uid in silence:
+                continue
+            if restrict is not None and remote.uid not in restrict:
+                continue
             result.append(await remote.cmd(**cmd))
 
         if len(result) == 1:
@@ -318,10 +337,17 @@ async def cmd_recv_async(self, method, *args, **kwargs):
         asyncio.Future
 
         """
+        silence = kwargs.pop('silence', None)
+        restrict = kwargs.pop('restrict', None)
+
         remotes, cmd = self._prepare_cmd(method, *args, **kwargs)
 
         result = []
         for remote in remotes:
+            if silence is not None and remote.uid in silence:
+                continue
+            if restrict is not None and remote.uid not in restrict:
+                continue
             result.append(await remote.cmd(**cmd, reply=True))
 
         if len(result) == 1: