Merge pull request #4 from filipcacky/async_runner_improvements_fixes

Async runner improvements fixes

metaflow/runner/metaflow_runner.py

@@ -292,9 +292,10 @@ async def __async_get_executing_run(self, attribute_file_fd, command_obj):
 
         # Set the correct metadata from the runner_attribute file corresponding to this run.
         metadata_for_flow = content.get("metadata")
-        metadata(metadata_for_flow)
 
-        run_object = Run(pathspec, _namespace_check=False)
+        run_object = Run(
+            pathspec, _namespace_check=False, _current_metadata=metadata_for_flow
+        )
         return ExecutingRun(self, command_obj, run_object)
 
     def run(self, **kwargs) -> ExecutingRun:

metaflow/runner/subprocess_manager.py

@@ -37,19 +37,31 @@ async def async_kill_processes_and_descendants(
     # TODO: there's a race condition that new descendants might
     # spawn b/w the invocations of 'pkill' and 'kill'.
     # Needs to be fixed in future.
-    sub_term = await asyncio.create_subprocess_exec("pkill", "-TERM", "-P", *pids)
-    await sub_term.wait()
+    try:
+        sub_term = await asyncio.create_subprocess_exec("pkill", "-TERM", "-P", *pids)
+        await sub_term.wait()
+    except Exception:
+        pass
 
-    main_term = await asyncio.create_subprocess_exec("kill", "-TERM", *pids)
-    await main_term.wait()
+    try:
+        main_term = await asyncio.create_subprocess_exec("kill", "-TERM", *pids)
+        await main_term.wait()
+    except Exception:
+        pass
 
     await asyncio.sleep(termination_timeout)
 
-    sub_kill = await asyncio.create_subprocess_exec("pkill", "-KILL", "-P", *pids)
-    await sub_kill.wait()
+    try:
+        sub_kill = await asyncio.create_subprocess_exec("pkill", "-KILL", "-P", *pids)
+        await sub_kill.wait()
+    except Exception:
+        pass
 
-    main_kill = await asyncio.create_subprocess_exec("kill", "-KILL", *pids)
-    await main_kill.wait()
+    try:
+        main_kill = await asyncio.create_subprocess_exec("kill", "-KILL", *pids)
+        await main_kill.wait()
+    except Exception:
+        pass
 
 
 class LogReadTimeoutError(Exception):