Remove script_dir state from channels

parsl/channels/base.py

@@ -1,22 +1,5 @@
-from abc import ABCMeta, abstractproperty
+from abc import ABCMeta
 
 
 class Channel(metaclass=ABCMeta):
-    @abstractproperty
-    def script_dir(self) -> str:
-        ''' This is a property. Returns the directory assigned for storing all internal scripts such as
-        scheduler submit scripts. This is usually where error logs from the scheduler would reside on the
-        channel destination side.
-
-        Args:
-            - None
-
-        Returns:
-            - Channel script dir
-        '''
-        pass
-
-    # DFK expects to be able to modify this, so it needs to be in the abstract class
-    @script_dir.setter
-    def script_dir(self, value: str) -> None:
-        pass
+    pass

parsl/channels/local/local.py

@@ -1,5 +1,4 @@
 import logging
-import os
 
 from parsl.channels.base import Channel
 from parsl.utils import RepresentationMixin
@@ -8,24 +7,4 @@
 
 
 class LocalChannel(Channel, RepresentationMixin):
-    ''' This is not even really a channel, since opening a local shell is not heavy
-    and done so infrequently that they do not need a persistent channel
-    '''
-
-    def __init__(self):
-        ''' Initialize the local channel. script_dir is required by set to a default.
-
-        KwArgs:
-            - script_dir (string): Directory to place scripts
-        '''
-        self.script_dir = None
-
-    @property
-    def script_dir(self):
-        return self._script_dir
-
-    @script_dir.setter
-    def script_dir(self, value):
-        if value is not None:
-            value = os.path.abspath(value)
-        self._script_dir = value
+    pass

parsl/dataflow/dflow.py

@@ -1151,8 +1151,6 @@ def add_executors(self, executors: Sequence[ParslExecutor]) -> None:
                     executor.provider.script_dir = os.path.join(self.run_dir, 'submit_scripts')
                     os.makedirs(executor.provider.script_dir, exist_ok=True)
 
-                    executor.provider.channel.script_dir = executor.provider.script_dir
-
             self.executors[executor.label] = executor
             executor.start()
         block_executors = [e for e in executors if isinstance(e, BlockProviderExecutor)]

parsl/providers/condor/condor.py

@@ -226,7 +226,7 @@ def submit(self, command, tasks_per_node, job_name="parsl.condor"):
 
         job_config = {}
         job_config["job_name"] = job_name
-        job_config["submit_script_dir"] = self.channel.script_dir
+        job_config["submit_script_dir"] = self.script_dir
         job_config["project"] = self.project
         job_config["nodes"] = self.nodes_per_block
         job_config["scheduler_options"] = scheduler_options

parsl/providers/grid_engine/grid_engine.py

@@ -100,7 +100,7 @@ def get_configs(self, command, tasks_per_node):
             self.nodes_per_block, tasks_per_node))
 
         job_config = {}
-        job_config["submit_script_dir"] = self.channel.script_dir
+        job_config["submit_script_dir"] = self.script_dir
         job_config["nodes"] = self.nodes_per_block
         job_config["walltime"] = self.walltime
         job_config["scheduler_options"] = self.scheduler_options

parsl/providers/lsf/lsf.py

@@ -211,7 +211,7 @@ def submit(self, command, tasks_per_node, job_name="parsl.lsf"):
         logger.debug("Requesting one block with {} nodes".format(self.nodes_per_block))
 
         job_config = {}
-        job_config["submit_script_dir"] = self.channel.script_dir
+        job_config["submit_script_dir"] = self.script_dir
         job_config["nodes"] = self.nodes_per_block
         job_config["tasks_per_node"] = tasks_per_node
         job_config["walltime"] = wtime_to_minutes(self.walltime)

parsl/providers/pbspro/pbspro.py

@@ -159,7 +159,7 @@ def submit(self, command, tasks_per_node, job_name="parsl"):
         )
 
         job_config = {}
-        job_config["submit_script_dir"] = self.channel.script_dir
+        job_config["submit_script_dir"] = self.script_dir
         job_config["nodes_per_block"] = self.nodes_per_block
         job_config["ncpus"] = self.cpus_per_node
         job_config["walltime"] = self.walltime

parsl/providers/slurm/slurm.py

@@ -3,7 +3,7 @@
 import os
 import re
 import time
-from typing import Optional
+from typing import Any, Dict, Optional
 
 import typeguard
 
@@ -286,8 +286,8 @@ def submit(self, command: str, tasks_per_node: int, job_name="parsl.slurm") -> s
 
         logger.debug("Requesting one block with {} nodes".format(self.nodes_per_block))
 
-        job_config = {}
-        job_config["submit_script_dir"] = self.channel.script_dir
+        job_config: Dict[str, Any] = {}
+        job_config["submit_script_dir"] = self.script_dir
         job_config["nodes"] = self.nodes_per_block
         job_config["tasks_per_node"] = tasks_per_node
         job_config["walltime"] = wtime_to_minutes(self.walltime)

parsl/providers/torque/torque.py

@@ -171,7 +171,7 @@ def submit(self, command, tasks_per_node, job_name="parsl.torque"):
 
         job_config = {}
         # TODO : script_path might need to change to accommodate script dir set via channels
-        job_config["submit_script_dir"] = self.channel.script_dir
+        job_config["submit_script_dir"] = self.script_dir
         job_config["nodes"] = self.nodes_per_block
         job_config["task_blocks"] = self.nodes_per_block * tasks_per_node
         job_config["nodes_per_block"] = self.nodes_per_block

parsl/tests/test_providers/test_pbspro_template.py

@@ -15,7 +15,6 @@ def test_submit_script_basic(tmp_path):
         queue="debug", channel=LocalChannel()
     )
     provider.script_dir = tmp_path
-    provider.channel.script_dir = tmp_path
     job_id = str(random.randint(55000, 59000))
     provider.execute_wait = mock.Mock(spec=PBSProProvider.execute_wait)
     provider.execute_wait.return_value = (0, job_id, "")

parsl/tests/test_providers/test_slurm_template.py

@@ -16,7 +16,6 @@ def test_submit_script_basic(tmp_path):
         partition="debug", channel=LocalChannel()
     )
     provider.script_dir = tmp_path
-    provider.channel.script_dir = tmp_path
     job_id = str(random.randint(55000, 59000))
     provider.execute_wait = mock.MagicMock(spec=SlurmProvider.execute_wait)
     provider.execute_wait.return_value = (0, f"Submitted batch job {job_id}", "")