Repository: Parsl/parsl (language: python)

Issue:
Remove channels
# Overview

Parsl has a facility called Channels intended to support use cases such as running a workflow on your laptop but with task execution on a supercomputer. This facility never evolved beyond prototype stage, and effort from University of Chicago and University of Illinois was rapidly distracted into the funcX project which aims (with substantially more code and developer time) to provide roughly the same facility - now named Globus Compute and developed by professional programmers rather than an academic research team.

The presence of channels as quasi-abandonware inside the modern Parsl codebase is a painful drain on both user and developer time: Users continue to be fooled into believing that this abandoned prototype facility, rather than the results of the funcX project, should be used for remote execution use cases. Developer time continues to be taken up dealing with the intricate lacing of channel handling through the core codebase.

Channels should be removed from Parsl, with the default behaviour of `LocalChannel` becoming the only behaviour.

# Proposed timeline

I propose this timeline as a default, unless a consensus forms for some other timeline. In the absence of any other consensus forming in the comments of this issue, I'll make this timeline happen.


now .. 7th August 2024: people paying attention to issues get to comment on this issue, including offering entirely different alternative paths forward
7th August 2024: channel code is marked as going away in the codebase, in some way that is hard to avoid for users (for example, renaming all the user-facing channel classes). Users should be encouraged to visit this issue and comment, and to seek alternatives such as Globus Compute.
7th November 2024: channel code is removed from Parsl

# Why can't this become an "abandoned component sitting in a directory"

The channel facility is not componentised - it contains prototype quality code inside the core of Parsl

# What about the AdHocProvider which attempts to use multiple channels to run on a cluster which has no resource manager?

This should go away too - without channels it is of little use. Users should be encouraged to find other ways to run a cluster without any resource manager. It should not be parsl's job to manage this aspect of a user's cluster.

# What if someone wants to take on Channel tech sponsorship?

This could happen, but I think it's unlikely.

The tech sponsor would need to be responsible for tidying up the architecture and implementing those fixes, with noise such as "why are channels initialised or not based on the presence of `script_dir` attribute on a provider?", test implementations and fixups such as testing parsl without shared file systems in a channel-like environment, and substantially more fundamental questions such as "how is htex supposed to work when channels do not provide a channel for htex network connections?".

For the more fundamental problems, this starts to sound like making a re-run of Globus Compute, which the funcX project has shown can occupy several full time developers for several years. I think probably that tech sponsors time would be better places working to get Parsl playing nicely with other remote execution technology - for example Globus Compute.

I just tagged a load of issues with I just tagged a bunch of issues with https://github.com/Parsl/parsl/labels/channels to give an overview of the sort of stuff a Channel tech sponsor would need to address.

An indicator for a successful channel tech sponsor would be substantial progress on these issues by the removal date in the timetable above.


Discussion / hints:
@benclifford I just wanted to put my vote here to remove channels entirely. 

From the user perspective the only folks I think we need to make sure know of the change would be:
1) folks using `SSHChannels` ought to gently pushed to make a switch over to `globus-compute`. 
2) `AdHocChannels` will have no alternative really beyond your recommendation to use a cluster resource manager (slurm..)

@benclifford I agree that channels are no longer needed and can be very finnicky to use.

Fix patch:
diff --git a/parsl/channels/base.py b/parsl/channels/base.py
index 754e89c385..dffe6d3be5 100644
--- a/parsl/channels/base.py
+++ b/parsl/channels/base.py
@@ -20,8 +20,8 @@ class Channel(metaclass=ABCMeta):
     such as push_file, pull_file and makedirs.
 
     Channels should ensure that each launched command runs in a new process
-    group, so that providers (such as AdHocProvider and LocalProvider) which
-    terminate long running commands using process groups can do so.
+    group, so that providers (such as LocalProvider) which terminate long
+    running commands using process groups can do so.
     """
 
     @abstractmethod
diff --git a/parsl/providers/ad_hoc/__init__.py b/parsl/providers/ad_hoc/__init__.py
deleted file mode 100644
index e69de29bb2..0000000000
diff --git a/parsl/providers/ad_hoc/ad_hoc.py b/parsl/providers/ad_hoc/ad_hoc.py
deleted file mode 100644
index 9059648101..0000000000
--- a/parsl/providers/ad_hoc/ad_hoc.py
+++ /dev/null
@@ -1,252 +0,0 @@
-import logging
-import os
-import time
-
-from parsl.channels import LocalChannel
-from parsl.jobs.states import JobState, JobStatus
-from parsl.launchers import SimpleLauncher
-from parsl.providers.base import ExecutionProvider
-from parsl.providers.errors import ScriptPathError
-from parsl.utils import RepresentationMixin
-
-logger = logging.getLogger(__name__)
-
-
-class DeprecatedAdHocProvider(ExecutionProvider, RepresentationMixin):
-    """ Deprecated ad-hoc execution provider
-
-    The (former) AdHocProvider is deprecated. See
-    `issue #3515 <https://github.com/Parsl/parsl/issues/3515>`_
-    for further discussion.
-
-    This provider is used to provision execution resources over one or more ad hoc nodes
-    that are each accessible over a Channel (say, ssh) but otherwise lack a cluster scheduler.
-
-    Parameters
-    ----------
-
-    channels : list of Channel ojects
-      Each channel represents a connection to a remote node
-
-    worker_init : str
-      Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.
-      Since this provider calls the same worker_init across all nodes in the ad-hoc cluster, it is
-      recommended that a single script is made available across nodes such as ~/setup.sh that can
-      be invoked.
-
-    cmd_timeout : int
-      Duration for which the provider will wait for a command to be invoked on a remote system.
-      Defaults to 30s
-
-    parallelism : float
-      Determines the ratio of workers to tasks as managed by the strategy component
-
-    """
-
-    def __init__(self,
-                 channels=[],
-                 worker_init='',
-                 cmd_timeout=30,
-                 parallelism=1,
-                 move_files=None):
-
-        self.channels = channels
-        self._label = 'ad-hoc'
-        self.worker_init = worker_init
-        self.cmd_timeout = cmd_timeout
-        self.parallelism = 1
-        self.move_files = move_files
-        self.launcher = SimpleLauncher()
-        self.init_blocks = self.min_blocks = self.max_blocks = len(channels)
-
-        # This will be overridden by the DFK to the rundirs.
-        self.script_dir = "."
-
-        # In ad-hoc mode, nodes_per_block should be 1
-        self.nodes_per_block = 1
-
-        # Dictionary that keeps track of jobs, keyed on job_id
-        self.resources = {}
-
-        self.least_loaded = self._least_loaded()
-        logger.debug("AdHoc provider initialized")
-
-    def _write_submit_script(self, script_string, script_filename):
-        '''
-        Load the template string with config values and write the generated submit script to
-        a submit script file.
-
-        Parameters
-        ----------
-        script_string: (string)
-          The template string to be used for the writing submit script
-
-        script_filename: (string)
-          Name of the submit script
-
-        Returns
-        -------
-        None: on success
-
-        Raises
-        ------
-        ScriptPathError
-          Unable to write submit script out
-        '''
-
-        try:
-            with open(script_filename, 'w') as f:
-                f.write(script_string)
-
-        except IOError as e:
-            logger.error("Failed writing to submit script: %s", script_filename)
-            raise ScriptPathError(script_filename, e)
-
-        return None
-
-    def _least_loaded(self):
-        """ Find channels that are not in use
-
-        Returns
-        -------
-        channel : Channel object
-        None : When there are no more available channels
-        """
-        while True:
-            channel_counts = {channel: 0 for channel in self.channels}
-            for job_id in self.resources:
-                channel = self.resources[job_id]['channel']
-                if self.resources[job_id]['status'].state == JobState.RUNNING:
-                    channel_counts[channel] = channel_counts.get(channel, 0) + 1
-                else:
-                    channel_counts[channel] = channel_counts.get(channel, 0)
-
-            logger.debug("Channel_counts : {}".format(channel_counts))
-            if 0 not in channel_counts.values():
-                yield None
-
-            for channel in channel_counts:
-                if channel_counts[channel] == 0:
-                    yield channel
-
-    def submit(self, command, tasks_per_node, job_name="parsl.adhoc"):
-        ''' Submits the command onto a channel from the list of channels
-
-        Submit returns an ID that corresponds to the task that was just submitted.
-
-        Parameters
-        ----------
-        command: (String)
-          Commandline invocation to be made on the remote side.
-
-        tasks_per_node: (int)
-          command invocations to be launched per node
-
-        job_name: (String)
-          Name of the job. Default : parsl.adhoc
-
-
-        Returns
-        -------
-        None
-          At capacity, cannot provision more
-
-        job_id: (string)
-          Identifier for the job
-
-        '''
-        channel = next(self.least_loaded)
-        if channel is None:
-            logger.warning("All Channels in Ad-Hoc provider are in use")
-            return None
-
-        job_name = "{0}.{1}".format(job_name, time.time())
-
-        # Set script path
-        script_path = "{0}/{1}.sh".format(self.script_dir, job_name)
-        script_path = os.path.abspath(script_path)
-
-        wrap_command = self.worker_init + '\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)
-
-        self._write_submit_script(wrap_command, script_path)
-
-        job_id = None
-        remote_pid = None
-        final_cmd = None
-
-        if (self.move_files is None and not isinstance(channel, LocalChannel)) or (self.move_files):
-            logger.debug("Pushing start script")
-            script_path = channel.push_file(script_path, channel.script_dir)
-
-        # Bash would return until the streams are closed. So we redirect to a outs file
-        final_cmd = 'bash {0} > {0}.out 2>&1 & \n echo "PID:$!" '.format(script_path)
-        retcode, stdout, stderr = channel.execute_wait(final_cmd, self.cmd_timeout)
-        for line in stdout.split('\n'):
-            if line.startswith("PID:"):
-                remote_pid = line.split("PID:")[1].strip()
-                job_id = remote_pid
-        if job_id is None:
-            logger.warning("Channel failed to start remote command/retrieve PID")
-
-        self.resources[job_id] = {'job_id': job_id,
-                                  'status': JobStatus(JobState.RUNNING),
-                                  'cmd': final_cmd,
-                                  'channel': channel,
-                                  'remote_pid': remote_pid}
-
-        return job_id
-
-    def status(self, job_ids):
-        """ Get status of the list of jobs with job_ids
-
-        Parameters
-        ----------
-        job_ids : list of strings
-          List of job id strings
-
-        Returns
-        -------
-        list of JobStatus objects
-        """
-        for job_id in job_ids:
-            channel = self.resources[job_id]['channel']
-            status_command = "ps --pid {} | grep {}".format(self.resources[job_id]['job_id'],
-                                                            self.resources[job_id]['cmd'].split()[0])
-            retcode, stdout, stderr = channel.execute_wait(status_command)
-            if retcode != 0 and self.resources[job_id]['status'].state == JobState.RUNNING:
-                self.resources[job_id]['status'] = JobStatus(JobState.FAILED)
-
-        return [self.resources[job_id]['status'] for job_id in job_ids]
-
-    def cancel(self, job_ids):
-        """ Cancel a list of jobs with job_ids
-
-        Parameters
-        ----------
-        job_ids : list of strings
-          List of job id strings
-
-        Returns
-        -------
-        list of confirmation bools: [True, False...]
-        """
-        logger.debug("Cancelling jobs: {}".format(job_ids))
-        rets = []
-        for job_id in job_ids:
-            channel = self.resources[job_id]['channel']
-            cmd = "kill -TERM -$(ps -o pgid= {} | grep -o '[0-9]*')".format(self.resources[job_id]['job_id'])
-            retcode, stdout, stderr = channel.execute_wait(cmd)
-            if retcode == 0:
-                rets.append(True)
-            else:
-                rets.append(False)
-            self.resources[job_id]['status'] = JobStatus(JobState.COMPLETED)
-        return rets
-
-    @property
-    def label(self):
-        return self._label
-
-    @property
-    def status_polling_interval(self):
-        return 10


Test patch:
diff --git a/parsl/tests/configs/local_adhoc.py b/parsl/tests/configs/local_adhoc.py
deleted file mode 100644
index 9b1f951842..0000000000
--- a/parsl/tests/configs/local_adhoc.py
+++ /dev/null
@@ -1,18 +0,0 @@
-from parsl.channels import LocalChannel
-from parsl.config import Config
-from parsl.executors import HighThroughputExecutor
-from parsl.providers.ad_hoc.ad_hoc import DeprecatedAdHocProvider
-
-
-def fresh_config():
-    return Config(
-        executors=[
-            HighThroughputExecutor(
-                label='AdHoc',
-                encrypted=True,
-                provider=DeprecatedAdHocProvider(
-                    channels=[LocalChannel(), LocalChannel()]
-                )
-            )
-        ]
-    )
diff --git a/parsl/tests/configs/user_opts.py b/parsl/tests/configs/user_opts.py
index ec39d4e7ab..540d13eae5 100644
--- a/parsl/tests/configs/user_opts.py
+++ b/parsl/tests/configs/user_opts.py
@@ -135,13 +135,6 @@
     #     # For example:
     #     'remote_writeable': 'globus://af7bda53-6d04-11e5-ba46-22000b92c6ec/home/bzc/'
     # },
-    # 'adhoc': {
-    #    # This specifies configuration parameters when testing an ad-hoc SSH based cluster
-    #    'username': 'fixme', # username on remote systems
-    #    'remote_hostnames': ['hostname1', 'hostname2'], # addresses of remote systems
-    #    'worker_init': 'init commands',  # worker_init for remote systems
-    #    'script_dir': "/path"  # script directory on remote systems
-    # }
     #
 }  # type: Dict[str, Any]
 
diff --git a/parsl/tests/sites/test_local_adhoc.py b/parsl/tests/sites/test_local_adhoc.py
deleted file mode 100644
index bdb922773c..0000000000
--- a/parsl/tests/sites/test_local_adhoc.py
+++ /dev/null
@@ -1,62 +0,0 @@
-import logging
-
-import pytest
-
-from parsl import python_app
-from parsl.tests.configs.local_adhoc import fresh_config as local_config
-
-logger = logging.getLogger(__name__)
-
-
-@python_app
-def python_app_2():
-    import os
-    import threading
-    import time
-    time.sleep(1)
-    return "Hello from PID[{}] TID[{}]".format(os.getpid(), threading.current_thread())
-
-
-@python_app
-def python_app_1():
-    import os
-    import threading
-    import time
-    time.sleep(1)
-    return "Hello from PID[{}] TID[{}]".format(os.getpid(), threading.current_thread())
-
-
-@python_app
-def bash_app(stdout=None, stderr=None):
-    return 'echo "Hello from $(uname -a)" ; sleep 2'
-
-
-@pytest.mark.local
-def test_python(N=2):
-    """Testing basic python functionality."""
-
-    r1 = {}
-    r2 = {}
-    for i in range(0, N):
-        r1[i] = python_app_1()
-        r2[i] = python_app_2()
-    print("Waiting ....")
-
-    for x in r1:
-        print("python_app_1 : ", r1[x].result())
-    for x in r2:
-        print("python_app_2 : ", r2[x].result())
-
-    return
-
-
-@pytest.mark.local
-def test_bash():
-    """Testing basic bash functionality."""
-
-    import os
-    fname = os.path.basename(__file__)
-
-    x = bash_app(stdout="{0}.out".format(fname))
-    print("Waiting ....")
-    print(x.result())