Fix issue where run_command blocks when it shouldn't

metaflow/plugins/argo/argo_workflows_deployer_objects.py

@@ -97,6 +97,7 @@ def suspend(self, **kwargs) -> bool:
         )
 
         command_obj = self.deployer.spm.get(pid)
+        command_obj.sync_wait()
         return command_obj.process.returncode == 0
 
     def unsuspend(self, **kwargs) -> bool:
@@ -131,6 +132,7 @@ def unsuspend(self, **kwargs) -> bool:
         )
 
         command_obj = self.deployer.spm.get(pid)
+        command_obj.sync_wait()
         return command_obj.process.returncode == 0
 
     def terminate(self, **kwargs) -> bool:
@@ -165,6 +167,7 @@ def terminate(self, **kwargs) -> bool:
         )
 
         command_obj = self.deployer.spm.get(pid)
+        command_obj.sync_wait()
         return command_obj.process.returncode == 0
 
     @property
@@ -319,6 +322,7 @@ def delete(self, **kwargs) -> bool:
         )
 
         command_obj = self.deployer.spm.get(pid)
+        command_obj.sync_wait()
         return command_obj.process.returncode == 0
 
     def trigger(self, **kwargs) -> ArgoWorkflowsTriggeredRun:
@@ -361,7 +365,7 @@ def trigger(self, **kwargs) -> ArgoWorkflowsTriggeredRun:
             content = handle_timeout(
                 attribute_file_fd, command_obj, self.deployer.file_read_timeout
             )
-
+            command_obj.sync_wait()
             if command_obj.process.returncode == 0:
                 return ArgoWorkflowsTriggeredRun(
                     deployer=self.deployer, content=content

metaflow/plugins/aws/step_functions/step_functions_deployer_objects.py

@@ -46,6 +46,7 @@ def terminate(self, **kwargs) -> bool:
         )
 
         command_obj = self.deployer.spm.get(pid)
+        command_obj.sync_wait()
         return command_obj.process.returncode == 0
 
 
@@ -174,6 +175,7 @@ def delete(self, **kwargs) -> bool:
         )
 
         command_obj = self.deployer.spm.get(pid)
+        command_obj.sync_wait()
         return command_obj.process.returncode == 0
 
     def trigger(self, **kwargs) -> StepFunctionsTriggeredRun:
@@ -217,6 +219,7 @@ def trigger(self, **kwargs) -> StepFunctionsTriggeredRun:
                 attribute_file_fd, command_obj, self.deployer.file_read_timeout
             )
 
+            command_obj.sync_wait()
             if command_obj.process.returncode == 0:
                 return StepFunctionsTriggeredRun(
                     deployer=self.deployer, content=content

metaflow/runner/deployer.py

@@ -64,7 +64,7 @@ class Deployer(metaclass=DeployerMeta):
         The directory to run the subprocess in; if not specified, the current
         directory is used.
     file_read_timeout : int, default 3600
-        The timeout until which we try to read the deployer attribute file.
+        The timeout until which we try to read the deployer attribute file (in seconds).
     **kwargs : Any
         Additional arguments that you would pass to `python myflow.py` before
         the deployment command.

metaflow/runner/deployer_impl.py

@@ -37,7 +37,7 @@ class variable that matches the name of the CLI group.
         The directory to run the subprocess in; if not specified, the current
         directory is used.
     file_read_timeout : int, default 3600
-        The timeout until which we try to read the deployer attribute file.
+        The timeout until which we try to read the deployer attribute file (in seconds).
     **kwargs : Any
         Additional arguments that you would pass to `python myflow.py` before
         the deployment command.
@@ -144,7 +144,7 @@ def _create(
             # Additional info is used to pass additional deployer specific information.
             # It is used in non-OSS deployers (extensions).
             self.additional_info = content.get("additional_info", {})
-
+            command_obj.sync_wait()
             if command_obj.process.returncode == 0:
                 return create_class(deployer=self)
 

metaflow/runner/metaflow_runner.py

@@ -221,7 +221,7 @@ class Runner(object):
         The directory to run the subprocess in; if not specified, the current
         directory is used.
     file_read_timeout : int, default 3600
-        The timeout until which we try to read the runner attribute file.
+        The timeout until which we try to read the runner attribute file (in seconds).
     **kwargs : Any
         Additional arguments that you would pass to `python myflow.py` before
         the `run` command.
@@ -326,6 +326,7 @@ def run(self, **kwargs) -> ExecutingRun:
                 show_output=self.show_output,
             )
             command_obj = self.spm.get(pid)
+            command_obj.sync_wait()
 
             return self.__get_executing_run(attribute_file_fd, command_obj)
 
@@ -357,6 +358,7 @@ def resume(self, **kwargs):
                 show_output=self.show_output,
             )
             command_obj = self.spm.get(pid)
+            command_obj.sync_wait()
 
             return self.__get_executing_run(attribute_file_fd, command_obj)
 

metaflow/runner/nbdeploy.py

@@ -46,6 +46,8 @@ class NBDeployer(object):
     base_dir : str, optional, default None
         The directory to run the subprocess in; if not specified, the current
         working directory is used.
+    file_read_timeout : int, default 3600
+        The timeout until which we try to read the deployer attribute file (in seconds).
     **kwargs : Any
         Additional arguments that you would pass to `python myflow.py` i.e. options
         listed in `python myflow.py --help`

metaflow/runner/nbrun.py

@@ -44,7 +44,7 @@ class NBRunner(object):
         The directory to run the subprocess in; if not specified, the current
         working directory is used.
     file_read_timeout : int, default 3600
-        The timeout until which we try to read the runner attribute file.
+        The timeout until which we try to read the runner attribute file (in seconds).
     **kwargs : Any
         Additional arguments that you would pass to `python myflow.py` before
         the `run` command.

metaflow/runner/subprocess_manager.py

@@ -120,6 +120,9 @@ def run_command(
         """
         Run a command synchronously and return its process ID.
 
+        Note: in no case does this wait for the process to *finish*. Use sync_wait()
+        to wait for the command to finish.
+
         Parameters
         ----------
         command : List[str]
@@ -145,7 +148,6 @@ def run_command(
         command_obj = CommandManager(command, env, cwd)
         pid = command_obj.run(show_output=show_output)
         self.commands[pid] = command_obj
-        command_obj.sync_wait()
         return pid
 
     async def async_run_command(

metaflow/runner/utils.py

@@ -91,7 +91,7 @@ def read_from_fifo_when_ready(
     encoding : str, optional
         Encoding to use while reading the file, by default "utf-8".
     timeout : int, optional
-        Timeout for reading the file in milliseconds, by default 3600.
+        Timeout for reading the file in seconds, by default 3600.
 
     Returns
     -------
@@ -113,24 +113,24 @@ def read_from_fifo_when_ready(
 
     while True:
         poll_begin = time.time()
-        poll.poll(timeout)
-        timeout -= 1000 * (time.time() - poll_begin)
-
-        if timeout <= 0:
-            raise TimeoutError("Timeout while waiting for the file content")
+        poll.poll(1000 * timeout)
+        timeout -= time.time() - poll_begin
 
         try:
-            data = os.read(fifo_fd, 128)
+            data = os.read(fifo_fd, 8192)
             while data:
                 content += data
-                data = os.read(fifo_fd, 128)
+                data = os.read(fifo_fd, 8192)
 
             # Read from a non-blocking closed FIFO returns an empty byte array
             break
 
         except BlockingIOError:
             # FIFO is open but no data is available yet
-            continue
+            pass
+        finally:
+            if timeout <= 0:
+                raise TimeoutError("Timeout while waiting for the file content")
 
     if not content and check_process_exited(command_obj):
         raise CalledProcessError(command_obj.process.returncode, command_obj.command)
@@ -156,7 +156,7 @@ async def async_read_from_fifo_when_ready(
     encoding : str, optional
         Encoding to use while reading the file, by default "utf-8".
     timeout : int, optional
-        Timeout for reading the file in milliseconds, by default 3600.
+        Timeout for reading the file in seconds, by default 3600.
 
     Returns
     -------
@@ -206,7 +206,7 @@ def handle_timeout(
     command_obj : CommandManager
         Command manager object that encapsulates the running command details.
     file_read_timeout : int
-        Timeout for reading the file.
+        Timeout for reading the file, in seconds
 
     Returns
     -------
@@ -243,7 +243,7 @@ async def async_handle_timeout(
     command_obj : CommandManager
         Command manager object that encapsulates the running command details.
     file_read_timeout : int
-        Timeout for reading the file.
+        Timeout for reading the file, in seconds
 
     Returns
     -------