We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Parsl creates hidden working files, doesn't clean them up
**Is your feature request related to a problem? Please describe.**
I didn't know Parsl was creating so many log files because they were hidden. 
I'm OK with files appearing in my run directory, but not accumulating them endlessly without my notice.

To be specific, I don't know what is responsible for `.workers.block-0.17158*` files but I'd like them to be visible.

**Describe the solution you'd like**
No hidden files.

**Describe alternatives you've considered**
Adding a "clean up Parsl's logging files" CLI tool

**Additional context**
Is anyone else concerned?


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `parsl/providers/local/local.py`
Content:
```
1 import logging
2 import os
3 import time
4 
5 from parsl.channels import LocalChannel
6 from parsl.jobs.states import JobState, JobStatus
7 from parsl.launchers import SingleNodeLauncher
8 from parsl.providers.base import ExecutionProvider
9 from parsl.providers.errors import SchedulerMissingArgs, ScriptPathError, SubmitException
10 from parsl.utils import RepresentationMixin
11 
12 logger = logging.getLogger(__name__)
13 
14 
15 class LocalProvider(ExecutionProvider, RepresentationMixin):
16     """ Local Execution Provider
17 
18     This provider is used to provide execution resources from the localhost.
19 
20     Parameters
21     ----------
22 
23     min_blocks : int
24         Minimum number of blocks to maintain.
25     max_blocks : int
26         Maximum number of blocks to maintain.
27     parallelism : float
28         Ratio of provisioned task slots to active tasks. A parallelism value of 1 represents aggressive
29         scaling where as many resources as possible are used; parallelism close to 0 represents
30         the opposite situation in which as few resources as possible (i.e., min_blocks) are used.
31     move_files : Optional[Bool]
32         Should files be moved? By default, Parsl will try to figure this out itself (= None).
33         If True, then will always move. If False, will never move.
34     worker_init : str
35         Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.
36     """
37 
38     def __init__(self,
39                  channel=LocalChannel(),
40                  nodes_per_block=1,
41                  launcher=SingleNodeLauncher(),
42                  init_blocks=1,
43                  min_blocks=0,
44                  max_blocks=1,
45                  worker_init='',
46                  cmd_timeout=30,
47                  parallelism=1,
48                  move_files=None):
49         self.channel = channel
50         self._label = 'local'
51         self.nodes_per_block = nodes_per_block
52         self.launcher = launcher
53         self.worker_init = worker_init
54         self.init_blocks = init_blocks
55         self.min_blocks = min_blocks
56         self.max_blocks = max_blocks
57         self.parallelism = parallelism
58         self.script_dir = None
59         self.cmd_timeout = cmd_timeout
60         self.move_files = move_files
61 
62         # Dictionary that keeps track of jobs, keyed on job_id
63         self.resources = {}
64 
65     def status(self, job_ids):
66         '''  Get the status of a list of jobs identified by their ids.
67 
68         Args:
69             - job_ids (List of ids) : List of identifiers for the jobs
70 
71         Returns:
72             - List of status codes.
73 
74         '''
75 
76         for job_id in self.resources:
77             # This job dict should really be a class on its own
78             job_dict = self.resources[job_id]
79             if job_dict['status'] and job_dict['status'].terminal:
80                 # We already checked this and it can't change after that
81                 continue
82             # Script path should point to remote path if _should_move_files() is True
83             script_path = job_dict['script_path']
84 
85             alive = self._is_alive(job_dict)
86             str_ec = self._read_job_file(script_path, '.ec').strip()
87 
88             status = None
89             if str_ec == '-':
90                 if alive:
91                     status = JobStatus(JobState.RUNNING)
92                 else:
93                     # not alive but didn't get to write an exit code
94                     if 'cancelled' in job_dict:
95                         # because we cancelled it
96                         status = JobStatus(JobState.CANCELLED)
97                     else:
98                         # we didn't cancel it, so it must have been killed by something outside
99                         # parsl; we don't have a state for this, but we'll use CANCELLED with
100                         # a specific message
101                         status = JobStatus(JobState.CANCELLED, message='Killed')
102             else:
103                 try:
104                     # TODO: ensure that these files are only read once and clean them
105                     ec = int(str_ec)
106                     stdout_path = self._job_file_path(script_path, '.out')
107                     stderr_path = self._job_file_path(script_path, '.err')
108                     if ec == 0:
109                         state = JobState.COMPLETED
110                     else:
111                         state = JobState.FAILED
112                     status = JobStatus(state, exit_code=ec,
113                                        stdout_path=stdout_path, stderr_path=stderr_path)
114                 except Exception:
115                     status = JobStatus(JobState.FAILED,
116                                        'Cannot parse exit code: {}'.format(str_ec))
117 
118             job_dict['status'] = status
119 
120         return [self.resources[jid]['status'] for jid in job_ids]
121 
122     def _is_alive(self, job_dict):
123         retcode, stdout, stderr = self.channel.execute_wait(
124             'ps -p {} > /dev/null 2> /dev/null; echo "STATUS:$?" '.format(
125                 job_dict['remote_pid']), self.cmd_timeout)
126         for line in stdout.split('\n'):
127             if line.startswith("STATUS:"):
128                 status = line.split("STATUS:")[1].strip()
129                 if status == "0":
130                     return True
131                 else:
132                     return False
133 
134     def _job_file_path(self, script_path: str, suffix: str) -> str:
135         path = '{0}{1}'.format(script_path, suffix)
136         if self._should_move_files():
137             path = self.channel.pull_file(path, self.script_dir)
138         return path
139 
140     def _read_job_file(self, script_path: str, suffix: str) -> str:
141         path = self._job_file_path(script_path, suffix)
142 
143         with open(path, 'r') as f:
144             return f.read()
145 
146     def _write_submit_script(self, script_string, script_filename):
147         '''
148         Load the template string with config values and write the generated submit script to
149         a submit script file.
150 
151         Args:
152               - template_string (string) : The template string to be used for the writing submit script
153               - script_filename (string) : Name of the submit script
154 
155         Returns:
156               - True: on success
157 
158         Raises:
159               SchedulerMissingArgs : If template is missing args
160               ScriptPathError : Unable to write submit script out
161         '''
162 
163         try:
164             with open(script_filename, 'w') as f:
165                 f.write(script_string)
166 
167         except KeyError as e:
168             logger.error("Missing keys for submit script: %s", e)
169             raise SchedulerMissingArgs(e.args, self.label)
170 
171         except IOError as e:
172             logger.error("Failed writing to submit script: %s", script_filename)
173             raise ScriptPathError(script_filename, e)
174 
175         return True
176 
177     def submit(self, command, tasks_per_node, job_name="parsl.localprovider"):
178         ''' Submits the command onto an Local Resource Manager job.
179         Submit returns an ID that corresponds to the task that was just submitted.
180 
181         If tasks_per_node <  1:
182              1/tasks_per_node is provisioned
183 
184         If tasks_per_node == 1:
185              A single node is provisioned
186 
187         If tasks_per_node >  1 :
188              tasks_per_node nodes are provisioned.
189 
190         Args:
191              - command  :(String) Commandline invocation to be made on the remote side.
192              - tasks_per_node (int) : command invocations to be launched per node
193 
194         Kwargs:
195              - job_name (String): Name for job, must be unique
196 
197         Returns:
198              - None: At capacity, cannot provision more
199              - job_id: (string) Identifier for the job
200 
201         '''
202 
203         job_name = "{0}.{1}".format(job_name, time.time())
204 
205         # Set script path
206         script_path = "{0}/{1}.sh".format(self.script_dir, job_name)
207         script_path = os.path.abspath(script_path)
208 
209         wrap_command = self.worker_init + f'\nexport JOBNAME=${job_name}\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)
210 
211         self._write_submit_script(wrap_command, script_path)
212 
213         job_id = None
214         remote_pid = None
215         if self._should_move_files():
216             logger.debug("Pushing start script")
217             script_path = self.channel.push_file(script_path, self.channel.script_dir)
218 
219         logger.debug("Launching")
220         # We need to capture the exit code and the streams, so we put them in files. We also write
221         # '-' to the exit code file to isolate potential problems with writing to files in the
222         # script directory
223         #
224         # The basic flow is:
225         #   1. write "-" to the exit code file. If this fails, exit
226         #   2. Launch the following sequence in the background:
227         #      a. the command to run
228         #      b. write the exit code of the command from (a) to the exit code file
229         #   3. Write the PID of the background sequence on stdout. The PID is needed if we want to
230         #      cancel the task later.
231         #
232         # We need to do the >/dev/null 2>&1 so that bash closes stdout, otherwise
233         # channel.execute_wait hangs reading the process stdout until all the
234         # background commands complete.
235         cmd = '/bin/bash -c \'echo - >{0}.ec && {{ {{ bash {0} 1>{0}.out 2>{0}.err ; ' \
236               'echo $? > {0}.ec ; }} >/dev/null 2>&1 & echo "PID:$!" ; }}\''.format(script_path)
237         retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)
238         if retcode != 0:
239             raise SubmitException(job_name, "Launch command exited with code {0}".format(retcode),
240                                   stdout, stderr)
241         for line in stdout.split('\n'):
242             if line.startswith("PID:"):
243                 remote_pid = line.split("PID:")[1].strip()
244                 job_id = remote_pid
245         if job_id is None:
246             raise SubmitException(job_name, "Channel failed to start remote command/retrieve PID")
247 
248         self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.RUNNING),
249                                   'remote_pid': remote_pid, 'script_path': script_path}
250 
251         return job_id
252 
253     def _should_move_files(self):
254         return (self.move_files is None and not isinstance(self.channel, LocalChannel)) or (self.move_files)
255 
256     def cancel(self, job_ids):
257         ''' Cancels the jobs specified by a list of job ids
258 
259         Args:
260         job_ids : [<job_id> ...]
261 
262         Returns: [True]  Always returns true for every job_id, regardless of
263                          whether an individual cancel failed (unless an
264                          exception is raised)
265         '''
266         for job in job_ids:
267             job_dict = self.resources[job]
268             job_dict['cancelled'] = True
269             logger.debug("Terminating job/process ID: {0}".format(job))
270             cmd = "kill -- -$(ps -o pgid= {} | grep -o '[0-9]*')".format(job_dict['remote_pid'])
271             retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)
272             if retcode != 0:
273                 logger.warning("Failed to kill PID: {} and child processes on {}".format(job_dict['remote_pid'],
274                                                                                          self.label))
275 
276         rets = [True for i in job_ids]
277         return rets
278 
279     @property
280     def label(self):
281         return self._label
282 
283     @property
284     def status_polling_interval(self):
285         return 5
286
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/parsl/providers/local/local.py b/parsl/providers/local/local.py
--- a/parsl/providers/local/local.py
+++ b/parsl/providers/local/local.py
@@ -206,7 +206,7 @@
         script_path = "{0}/{1}.sh".format(self.script_dir, job_name)
         script_path = os.path.abspath(script_path)
 
-        wrap_command = self.worker_init + f'\nexport JOBNAME=${job_name}\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)
+        wrap_command = self.worker_init + f'\nexport JOBNAME={job_name}\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)
 
         self._write_submit_script(wrap_command, script_path)

{"golden_diff": "diff --git a/parsl/providers/local/local.py b/parsl/providers/local/local.py\n--- a/parsl/providers/local/local.py\n+++ b/parsl/providers/local/local.py\n@@ -206,7 +206,7 @@\n         script_path = \"{0}/{1}.sh\".format(self.script_dir, job_name)\n         script_path = os.path.abspath(script_path)\n \n-        wrap_command = self.worker_init + f'\\nexport JOBNAME=${job_name}\\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)\n+        wrap_command = self.worker_init + f'\\nexport JOBNAME={job_name}\\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)\n \n         self._write_submit_script(wrap_command, script_path)\n", "issue": "Parsl creates hidden working files, doesn't clean them up\n**Is your feature request related to a problem? Please describe.**\r\nI didn't know Parsl was creating so many log files because they were hidden. \r\nI'm OK with files appearing in my run directory, but not accumulating them endlessly without my notice.\r\n\r\nTo be specific, I don't know what is responsible for `.workers.block-0.17158*` files but I'd like them to be visible.\r\n\r\n**Describe the solution you'd like**\r\nNo hidden files.\r\n\r\n**Describe alternatives you've considered**\r\nAdding a \"clean up Parsl's logging files\" CLI tool\r\n\r\n**Additional context**\r\nIs anyone else concerned?\r\n\n", "before_files": [{"content": "import logging\nimport os\nimport time\n\nfrom parsl.channels import LocalChannel\nfrom parsl.jobs.states import JobState, JobStatus\nfrom parsl.launchers import SingleNodeLauncher\nfrom parsl.providers.base import ExecutionProvider\nfrom parsl.providers.errors import SchedulerMissingArgs, ScriptPathError, SubmitException\nfrom parsl.utils import RepresentationMixin\n\nlogger = logging.getLogger(__name__)\n\n\nclass LocalProvider(ExecutionProvider, RepresentationMixin):\n    \"\"\" Local Execution Provider\n\n    This provider is used to provide execution resources from the localhost.\n\n    Parameters\n    ----------\n\n    min_blocks : int\n        Minimum number of blocks to maintain.\n    max_blocks : int\n        Maximum number of blocks to maintain.\n    parallelism : float\n        Ratio of provisioned task slots to active tasks. A parallelism value of 1 represents aggressive\n        scaling where as many resources as possible are used; parallelism close to 0 represents\n        the opposite situation in which as few resources as possible (i.e., min_blocks) are used.\n    move_files : Optional[Bool]\n        Should files be moved? By default, Parsl will try to figure this out itself (= None).\n        If True, then will always move. If False, will never move.\n    worker_init : str\n        Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.\n    \"\"\"\n\n    def __init__(self,\n                 channel=LocalChannel(),\n                 nodes_per_block=1,\n                 launcher=SingleNodeLauncher(),\n                 init_blocks=1,\n                 min_blocks=0,\n                 max_blocks=1,\n                 worker_init='',\n                 cmd_timeout=30,\n                 parallelism=1,\n                 move_files=None):\n        self.channel = channel\n        self._label = 'local'\n        self.nodes_per_block = nodes_per_block\n        self.launcher = launcher\n        self.worker_init = worker_init\n        self.init_blocks = init_blocks\n        self.min_blocks = min_blocks\n        self.max_blocks = max_blocks\n        self.parallelism = parallelism\n        self.script_dir = None\n        self.cmd_timeout = cmd_timeout\n        self.move_files = move_files\n\n        # Dictionary that keeps track of jobs, keyed on job_id\n        self.resources = {}\n\n    def status(self, job_ids):\n        '''  Get the status of a list of jobs identified by their ids.\n\n        Args:\n            - job_ids (List of ids) : List of identifiers for the jobs\n\n        Returns:\n            - List of status codes.\n\n        '''\n\n        for job_id in self.resources:\n            # This job dict should really be a class on its own\n            job_dict = self.resources[job_id]\n            if job_dict['status'] and job_dict['status'].terminal:\n                # We already checked this and it can't change after that\n                continue\n            # Script path should point to remote path if _should_move_files() is True\n            script_path = job_dict['script_path']\n\n            alive = self._is_alive(job_dict)\n            str_ec = self._read_job_file(script_path, '.ec').strip()\n\n            status = None\n            if str_ec == '-':\n                if alive:\n                    status = JobStatus(JobState.RUNNING)\n                else:\n                    # not alive but didn't get to write an exit code\n                    if 'cancelled' in job_dict:\n                        # because we cancelled it\n                        status = JobStatus(JobState.CANCELLED)\n                    else:\n                        # we didn't cancel it, so it must have been killed by something outside\n                        # parsl; we don't have a state for this, but we'll use CANCELLED with\n                        # a specific message\n                        status = JobStatus(JobState.CANCELLED, message='Killed')\n            else:\n                try:\n                    # TODO: ensure that these files are only read once and clean them\n                    ec = int(str_ec)\n                    stdout_path = self._job_file_path(script_path, '.out')\n                    stderr_path = self._job_file_path(script_path, '.err')\n                    if ec == 0:\n                        state = JobState.COMPLETED\n                    else:\n                        state = JobState.FAILED\n                    status = JobStatus(state, exit_code=ec,\n                                       stdout_path=stdout_path, stderr_path=stderr_path)\n                except Exception:\n                    status = JobStatus(JobState.FAILED,\n                                       'Cannot parse exit code: {}'.format(str_ec))\n\n            job_dict['status'] = status\n\n        return [self.resources[jid]['status'] for jid in job_ids]\n\n    def _is_alive(self, job_dict):\n        retcode, stdout, stderr = self.channel.execute_wait(\n            'ps -p {} > /dev/null 2> /dev/null; echo \"STATUS:$?\" '.format(\n                job_dict['remote_pid']), self.cmd_timeout)\n        for line in stdout.split('\\n'):\n            if line.startswith(\"STATUS:\"):\n                status = line.split(\"STATUS:\")[1].strip()\n                if status == \"0\":\n                    return True\n                else:\n                    return False\n\n    def _job_file_path(self, script_path: str, suffix: str) -> str:\n        path = '{0}{1}'.format(script_path, suffix)\n        if self._should_move_files():\n            path = self.channel.pull_file(path, self.script_dir)\n        return path\n\n    def _read_job_file(self, script_path: str, suffix: str) -> str:\n        path = self._job_file_path(script_path, suffix)\n\n        with open(path, 'r') as f:\n            return f.read()\n\n    def _write_submit_script(self, script_string, script_filename):\n        '''\n        Load the template string with config values and write the generated submit script to\n        a submit script file.\n\n        Args:\n              - template_string (string) : The template string to be used for the writing submit script\n              - script_filename (string) : Name of the submit script\n\n        Returns:\n              - True: on success\n\n        Raises:\n              SchedulerMissingArgs : If template is missing args\n              ScriptPathError : Unable to write submit script out\n        '''\n\n        try:\n            with open(script_filename, 'w') as f:\n                f.write(script_string)\n\n        except KeyError as e:\n            logger.error(\"Missing keys for submit script: %s\", e)\n            raise SchedulerMissingArgs(e.args, self.label)\n\n        except IOError as e:\n            logger.error(\"Failed writing to submit script: %s\", script_filename)\n            raise ScriptPathError(script_filename, e)\n\n        return True\n\n    def submit(self, command, tasks_per_node, job_name=\"parsl.localprovider\"):\n        ''' Submits the command onto an Local Resource Manager job.\n        Submit returns an ID that corresponds to the task that was just submitted.\n\n        If tasks_per_node <  1:\n             1/tasks_per_node is provisioned\n\n        If tasks_per_node == 1:\n             A single node is provisioned\n\n        If tasks_per_node >  1 :\n             tasks_per_node nodes are provisioned.\n\n        Args:\n             - command  :(String) Commandline invocation to be made on the remote side.\n             - tasks_per_node (int) : command invocations to be launched per node\n\n        Kwargs:\n             - job_name (String): Name for job, must be unique\n\n        Returns:\n             - None: At capacity, cannot provision more\n             - job_id: (string) Identifier for the job\n\n        '''\n\n        job_name = \"{0}.{1}\".format(job_name, time.time())\n\n        # Set script path\n        script_path = \"{0}/{1}.sh\".format(self.script_dir, job_name)\n        script_path = os.path.abspath(script_path)\n\n        wrap_command = self.worker_init + f'\\nexport JOBNAME=${job_name}\\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)\n\n        self._write_submit_script(wrap_command, script_path)\n\n        job_id = None\n        remote_pid = None\n        if self._should_move_files():\n            logger.debug(\"Pushing start script\")\n            script_path = self.channel.push_file(script_path, self.channel.script_dir)\n\n        logger.debug(\"Launching\")\n        # We need to capture the exit code and the streams, so we put them in files. We also write\n        # '-' to the exit code file to isolate potential problems with writing to files in the\n        # script directory\n        #\n        # The basic flow is:\n        #   1. write \"-\" to the exit code file. If this fails, exit\n        #   2. Launch the following sequence in the background:\n        #      a. the command to run\n        #      b. write the exit code of the command from (a) to the exit code file\n        #   3. Write the PID of the background sequence on stdout. The PID is needed if we want to\n        #      cancel the task later.\n        #\n        # We need to do the >/dev/null 2>&1 so that bash closes stdout, otherwise\n        # channel.execute_wait hangs reading the process stdout until all the\n        # background commands complete.\n        cmd = '/bin/bash -c \\'echo - >{0}.ec && {{ {{ bash {0} 1>{0}.out 2>{0}.err ; ' \\\n              'echo $? > {0}.ec ; }} >/dev/null 2>&1 & echo \"PID:$!\" ; }}\\''.format(script_path)\n        retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)\n        if retcode != 0:\n            raise SubmitException(job_name, \"Launch command exited with code {0}\".format(retcode),\n                                  stdout, stderr)\n        for line in stdout.split('\\n'):\n            if line.startswith(\"PID:\"):\n                remote_pid = line.split(\"PID:\")[1].strip()\n                job_id = remote_pid\n        if job_id is None:\n            raise SubmitException(job_name, \"Channel failed to start remote command/retrieve PID\")\n\n        self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.RUNNING),\n                                  'remote_pid': remote_pid, 'script_path': script_path}\n\n        return job_id\n\n    def _should_move_files(self):\n        return (self.move_files is None and not isinstance(self.channel, LocalChannel)) or (self.move_files)\n\n    def cancel(self, job_ids):\n        ''' Cancels the jobs specified by a list of job ids\n\n        Args:\n        job_ids : [<job_id> ...]\n\n        Returns: [True]  Always returns true for every job_id, regardless of\n                         whether an individual cancel failed (unless an\n                         exception is raised)\n        '''\n        for job in job_ids:\n            job_dict = self.resources[job]\n            job_dict['cancelled'] = True\n            logger.debug(\"Terminating job/process ID: {0}\".format(job))\n            cmd = \"kill -- -$(ps -o pgid= {} | grep -o '[0-9]*')\".format(job_dict['remote_pid'])\n            retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)\n            if retcode != 0:\n                logger.warning(\"Failed to kill PID: {} and child processes on {}\".format(job_dict['remote_pid'],\n                                                                                         self.label))\n\n        rets = [True for i in job_ids]\n        return rets\n\n    @property\n    def label(self):\n        return self._label\n\n    @property\n    def status_polling_interval(self):\n        return 5\n", "path": "parsl/providers/local/local.py"}], "after_files": [{"content": "import logging\nimport os\nimport time\n\nfrom parsl.channels import LocalChannel\nfrom parsl.jobs.states import JobState, JobStatus\nfrom parsl.launchers import SingleNodeLauncher\nfrom parsl.providers.base import ExecutionProvider\nfrom parsl.providers.errors import SchedulerMissingArgs, ScriptPathError, SubmitException\nfrom parsl.utils import RepresentationMixin\n\nlogger = logging.getLogger(__name__)\n\n\nclass LocalProvider(ExecutionProvider, RepresentationMixin):\n    \"\"\" Local Execution Provider\n\n    This provider is used to provide execution resources from the localhost.\n\n    Parameters\n    ----------\n\n    min_blocks : int\n        Minimum number of blocks to maintain.\n    max_blocks : int\n        Maximum number of blocks to maintain.\n    parallelism : float\n        Ratio of provisioned task slots to active tasks. A parallelism value of 1 represents aggressive\n        scaling where as many resources as possible are used; parallelism close to 0 represents\n        the opposite situation in which as few resources as possible (i.e., min_blocks) are used.\n    move_files : Optional[Bool]\n        Should files be moved? By default, Parsl will try to figure this out itself (= None).\n        If True, then will always move. If False, will never move.\n    worker_init : str\n        Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.\n    \"\"\"\n\n    def __init__(self,\n                 channel=LocalChannel(),\n                 nodes_per_block=1,\n                 launcher=SingleNodeLauncher(),\n                 init_blocks=1,\n                 min_blocks=0,\n                 max_blocks=1,\n                 worker_init='',\n                 cmd_timeout=30,\n                 parallelism=1,\n                 move_files=None):\n        self.channel = channel\n        self._label = 'local'\n        self.nodes_per_block = nodes_per_block\n        self.launcher = launcher\n        self.worker_init = worker_init\n        self.init_blocks = init_blocks\n        self.min_blocks = min_blocks\n        self.max_blocks = max_blocks\n        self.parallelism = parallelism\n        self.script_dir = None\n        self.cmd_timeout = cmd_timeout\n        self.move_files = move_files\n\n        # Dictionary that keeps track of jobs, keyed on job_id\n        self.resources = {}\n\n    def status(self, job_ids):\n        '''  Get the status of a list of jobs identified by their ids.\n\n        Args:\n            - job_ids (List of ids) : List of identifiers for the jobs\n\n        Returns:\n            - List of status codes.\n\n        '''\n\n        for job_id in self.resources:\n            # This job dict should really be a class on its own\n            job_dict = self.resources[job_id]\n            if job_dict['status'] and job_dict['status'].terminal:\n                # We already checked this and it can't change after that\n                continue\n            # Script path should point to remote path if _should_move_files() is True\n            script_path = job_dict['script_path']\n\n            alive = self._is_alive(job_dict)\n            str_ec = self._read_job_file(script_path, '.ec').strip()\n\n            status = None\n            if str_ec == '-':\n                if alive:\n                    status = JobStatus(JobState.RUNNING)\n                else:\n                    # not alive but didn't get to write an exit code\n                    if 'cancelled' in job_dict:\n                        # because we cancelled it\n                        status = JobStatus(JobState.CANCELLED)\n                    else:\n                        # we didn't cancel it, so it must have been killed by something outside\n                        # parsl; we don't have a state for this, but we'll use CANCELLED with\n                        # a specific message\n                        status = JobStatus(JobState.CANCELLED, message='Killed')\n            else:\n                try:\n                    # TODO: ensure that these files are only read once and clean them\n                    ec = int(str_ec)\n                    stdout_path = self._job_file_path(script_path, '.out')\n                    stderr_path = self._job_file_path(script_path, '.err')\n                    if ec == 0:\n                        state = JobState.COMPLETED\n                    else:\n                        state = JobState.FAILED\n                    status = JobStatus(state, exit_code=ec,\n                                       stdout_path=stdout_path, stderr_path=stderr_path)\n                except Exception:\n                    status = JobStatus(JobState.FAILED,\n                                       'Cannot parse exit code: {}'.format(str_ec))\n\n            job_dict['status'] = status\n\n        return [self.resources[jid]['status'] for jid in job_ids]\n\n    def _is_alive(self, job_dict):\n        retcode, stdout, stderr = self.channel.execute_wait(\n            'ps -p {} > /dev/null 2> /dev/null; echo \"STATUS:$?\" '.format(\n                job_dict['remote_pid']), self.cmd_timeout)\n        for line in stdout.split('\\n'):\n            if line.startswith(\"STATUS:\"):\n                status = line.split(\"STATUS:\")[1].strip()\n                if status == \"0\":\n                    return True\n                else:\n                    return False\n\n    def _job_file_path(self, script_path: str, suffix: str) -> str:\n        path = '{0}{1}'.format(script_path, suffix)\n        if self._should_move_files():\n            path = self.channel.pull_file(path, self.script_dir)\n        return path\n\n    def _read_job_file(self, script_path: str, suffix: str) -> str:\n        path = self._job_file_path(script_path, suffix)\n\n        with open(path, 'r') as f:\n            return f.read()\n\n    def _write_submit_script(self, script_string, script_filename):\n        '''\n        Load the template string with config values and write the generated submit script to\n        a submit script file.\n\n        Args:\n              - template_string (string) : The template string to be used for the writing submit script\n              - script_filename (string) : Name of the submit script\n\n        Returns:\n              - True: on success\n\n        Raises:\n              SchedulerMissingArgs : If template is missing args\n              ScriptPathError : Unable to write submit script out\n        '''\n\n        try:\n            with open(script_filename, 'w') as f:\n                f.write(script_string)\n\n        except KeyError as e:\n            logger.error(\"Missing keys for submit script: %s\", e)\n            raise SchedulerMissingArgs(e.args, self.label)\n\n        except IOError as e:\n            logger.error(\"Failed writing to submit script: %s\", script_filename)\n            raise ScriptPathError(script_filename, e)\n\n        return True\n\n    def submit(self, command, tasks_per_node, job_name=\"parsl.localprovider\"):\n        ''' Submits the command onto an Local Resource Manager job.\n        Submit returns an ID that corresponds to the task that was just submitted.\n\n        If tasks_per_node <  1:\n             1/tasks_per_node is provisioned\n\n        If tasks_per_node == 1:\n             A single node is provisioned\n\n        If tasks_per_node >  1 :\n             tasks_per_node nodes are provisioned.\n\n        Args:\n             - command  :(String) Commandline invocation to be made on the remote side.\n             - tasks_per_node (int) : command invocations to be launched per node\n\n        Kwargs:\n             - job_name (String): Name for job, must be unique\n\n        Returns:\n             - None: At capacity, cannot provision more\n             - job_id: (string) Identifier for the job\n\n        '''\n\n        job_name = \"{0}.{1}\".format(job_name, time.time())\n\n        # Set script path\n        script_path = \"{0}/{1}.sh\".format(self.script_dir, job_name)\n        script_path = os.path.abspath(script_path)\n\n        wrap_command = self.worker_init + f'\\nexport JOBNAME={job_name}\\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)\n\n        self._write_submit_script(wrap_command, script_path)\n\n        job_id = None\n        remote_pid = None\n        if self._should_move_files():\n            logger.debug(\"Pushing start script\")\n            script_path = self.channel.push_file(script_path, self.channel.script_dir)\n\n        logger.debug(\"Launching\")\n        # We need to capture the exit code and the streams, so we put them in files. We also write\n        # '-' to the exit code file to isolate potential problems with writing to files in the\n        # script directory\n        #\n        # The basic flow is:\n        #   1. write \"-\" to the exit code file. If this fails, exit\n        #   2. Launch the following sequence in the background:\n        #      a. the command to run\n        #      b. write the exit code of the command from (a) to the exit code file\n        #   3. Write the PID of the background sequence on stdout. The PID is needed if we want to\n        #      cancel the task later.\n        #\n        # We need to do the >/dev/null 2>&1 so that bash closes stdout, otherwise\n        # channel.execute_wait hangs reading the process stdout until all the\n        # background commands complete.\n        cmd = '/bin/bash -c \\'echo - >{0}.ec && {{ {{ bash {0} 1>{0}.out 2>{0}.err ; ' \\\n              'echo $? > {0}.ec ; }} >/dev/null 2>&1 & echo \"PID:$!\" ; }}\\''.format(script_path)\n        retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)\n        if retcode != 0:\n            raise SubmitException(job_name, \"Launch command exited with code {0}\".format(retcode),\n                                  stdout, stderr)\n        for line in stdout.split('\\n'):\n            if line.startswith(\"PID:\"):\n                remote_pid = line.split(\"PID:\")[1].strip()\n                job_id = remote_pid\n        if job_id is None:\n            raise SubmitException(job_name, \"Channel failed to start remote command/retrieve PID\")\n\n        self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.RUNNING),\n                                  'remote_pid': remote_pid, 'script_path': script_path}\n\n        return job_id\n\n    def _should_move_files(self):\n        return (self.move_files is None and not isinstance(self.channel, LocalChannel)) or (self.move_files)\n\n    def cancel(self, job_ids):\n        ''' Cancels the jobs specified by a list of job ids\n\n        Args:\n        job_ids : [<job_id> ...]\n\n        Returns: [True]  Always returns true for every job_id, regardless of\n                         whether an individual cancel failed (unless an\n                         exception is raised)\n        '''\n        for job in job_ids:\n            job_dict = self.resources[job]\n            job_dict['cancelled'] = True\n            logger.debug(\"Terminating job/process ID: {0}\".format(job))\n            cmd = \"kill -- -$(ps -o pgid= {} | grep -o '[0-9]*')\".format(job_dict['remote_pid'])\n            retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)\n            if retcode != 0:\n                logger.warning(\"Failed to kill PID: {} and child processes on {}\".format(job_dict['remote_pid'],\n                                                                                         self.label))\n\n        rets = [True for i in job_ids]\n        return rets\n\n    @property\n    def label(self):\n        return self._label\n\n    @property\n    def status_polling_interval(self):\n        return 5\n", "path": "parsl/providers/local/local.py"}]}