Move execute_wait into parsl.utils and remove parsl.channels remnant

parsl/providers/cluster_provider.py

@@ -2,11 +2,11 @@
 from abc import abstractmethod
 from string import Template
 
-from parsl.channels.local.local import execute_wait
 from parsl.launchers.base import Launcher
 from parsl.launchers.errors import BadLauncher
 from parsl.providers.base import ExecutionProvider
 from parsl.providers.errors import SchedulerMissingArgs, ScriptPathError
+from parsl.utils import execute_wait
 
 logger = logging.getLogger(__name__)
 

parsl/providers/local/local.py

@@ -2,7 +2,6 @@
 import os
 import time
 
-from parsl.channels.local.local import execute_wait
 from parsl.jobs.states import JobState, JobStatus
 from parsl.launchers import SingleNodeLauncher
 from parsl.providers.base import ExecutionProvider
@@ -11,7 +10,7 @@
     ScriptPathError,
     SubmitException,
 )
-from parsl.utils import RepresentationMixin
+from parsl.utils import RepresentationMixin, execute_wait
 
 logger = logging.getLogger(__name__)
 

parsl/tests/test_channels/test_large_output.py

@@ -1,6 +1,6 @@
 import pytest
 
-from parsl.channels.local.local import execute_wait
+from parsl.utils import execute_wait
 
 
 @pytest.mark.local

parsl/tests/test_channels/test_local_channel.py

@@ -1,6 +1,6 @@
 import pytest
 
-from parsl.channels.local.local import execute_wait
+from parsl.utils import execute_wait
 
 
 @pytest.mark.local

parsl/utils.py

@@ -458,3 +458,38 @@ def sanitize_dns_subdomain_rfc1123(raw_string: str) -> str:
         raise ValueError(f"Sanitized DNS subdomain is empty for input '{raw_string}'")
 
     return sanitized
+
+
+def execute_wait(cmd: str, walltime: Union[float, int, None] = None) -> Tuple[int, str, str]:
+    ''' Synchronously execute a commandline string on the shell.
+
+    Args:
+        - cmd (string) : Commandline string to execute
+        - walltime (int) : walltime in seconds
+
+    Returns:
+        - retcode : Return code from the execution
+        - stdout  : stdout string
+        - stderr  : stderr string
+    '''
+    try:
+        logger.debug("Creating process with command '%s'", cmd)
+        proc = subprocess.Popen(
+            cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            shell=True,
+            preexec_fn=os.setpgrp
+        )
+        logger.debug("Created process with pid %s. Performing communicate", proc.pid)
+        (stdout, stderr) = proc.communicate(timeout=walltime)
+        retcode = proc.returncode
+        logger.debug("Process %s returned %s", proc.pid, proc.returncode)
+
+    except Exception:
+        logger.exception(f"Execution of command failed:\n{cmd}")
+        raise
+    else:
+        logger.debug("Execution of command in process %s completed normally", proc.pid)
+
+    return (retcode, stdout.decode("utf-8"), stderr.decode("utf-8"))