We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Incorrect pgid identification in LocalProvider kills too many processes
This code in LocalProvider probably should have a space added to give `-o pgid= {}`. This should be two parameters, one '-o pgid=` meaning output only pgids, and give that single column an empty heading; and secondly the process ID to inspect.

```
                cmd = "kill -- -$(ps -o pgid={} | grep -o '[0-9]*')".format(self.resources[job]['remote_pid'])
```

The code as it is probably doesn't shut down the expected process - it might sometimes; it might kill other processes or process groups, as the resulting ps output contains a list of many PGIDs, some of which might be valid PIDs to kill; or it might not kill anything. This might be related to bad shutdown behaviour described in #674 

I have not tested this but I encountered similar behaviour reviewing  #1297 

--- 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 signal
4 import time
5 
6 from parsl.channels import LocalChannel
7 from parsl.launchers import SingleNodeLauncher
8 from parsl.providers.provider_base import ExecutionProvider
9 from parsl.providers.error import SchedulerMissingArgs, ScriptPathError
10 from parsl.utils import RepresentationMixin
11 
12 logger = logging.getLogger(__name__)
13 
14 translate_table = {
15     'PD': 'PENDING',
16     'R': 'RUNNING',
17     'CA': 'CANCELLED',
18     'CF': 'PENDING',  # (configuring),
19     'CG': 'RUNNING',  # (completing),
20     'CD': 'COMPLETED',
21     'F': 'FAILED',
22     'TO': 'TIMEOUT',
23     'NF': 'FAILED',  # (node failure),
24     'RV': 'FAILED',  # (revoked) and
25     'SE': 'FAILED'
26 }  # (special exit state
27 
28 
29 class LocalProvider(ExecutionProvider, RepresentationMixin):
30     """ Local Execution Provider
31 
32     This provider is used to provide execution resources from the localhost.
33 
34     Parameters
35     ----------
36 
37     min_blocks : int
38         Minimum number of blocks to maintain.
39     max_blocks : int
40         Maximum number of blocks to maintain.
41     parallelism : float
42         Ratio of provisioned task slots to active tasks. A parallelism value of 1 represents aggressive
43         scaling where as many resources as possible are used; parallelism close to 0 represents
44         the opposite situation in which as few resources as possible (i.e., min_blocks) are used.
45     move_files : Optional[Bool]: should files be moved? by default, Parsl will try to figure
46         this out itself (= None). If True, then will always move. If False, will never move.
47     worker_init : str
48         Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.
49     """
50 
51     def __init__(self,
52                  channel=LocalChannel(),
53                  nodes_per_block=1,
54                  launcher=SingleNodeLauncher(),
55                  init_blocks=4,
56                  min_blocks=0,
57                  max_blocks=10,
58                  walltime="00:15:00",
59                  worker_init='',
60                  cmd_timeout=30,
61                  parallelism=1,
62                  move_files=None):
63         self.channel = channel
64         self._label = 'local'
65         self.provisioned_blocks = 0
66         self.nodes_per_block = nodes_per_block
67         self.launcher = launcher
68         self.worker_init = worker_init
69         self.init_blocks = init_blocks
70         self.min_blocks = min_blocks
71         self.max_blocks = max_blocks
72         self.parallelism = parallelism
73         self.walltime = walltime
74         self.script_dir = None
75         self.cmd_timeout = cmd_timeout
76         self.move_files = move_files
77 
78         # Dictionary that keeps track of jobs, keyed on job_id
79         self.resources = {}
80 
81     def status(self, job_ids):
82         '''  Get the status of a list of jobs identified by their ids.
83 
84         Args:
85             - job_ids (List of ids) : List of identifiers for the jobs
86 
87         Returns:
88             - List of status codes.
89 
90         '''
91 
92         logger.debug("Checking status of: {0}".format(job_ids))
93         for job_id in self.resources:
94 
95             if self.resources[job_id]['proc']:
96 
97                 poll_code = self.resources[job_id]['proc'].poll()
98                 if self.resources[job_id]['status'] in ['COMPLETED', 'FAILED']:
99                     continue
100 
101                 if poll_code is None:
102                     self.resources[job_id]['status'] = 'RUNNING'
103                 elif poll_code == 0:
104                     self.resources[job_id]['status'] = 'COMPLETED'
105                 elif poll_code != 0:
106                     self.resources[job_id]['status'] = 'FAILED'
107                 else:
108                     logger.error("Internal consistency error: unexpected case in local provider state machine")
109 
110             elif self.resources[job_id]['remote_pid']:
111 
112                 retcode, stdout, stderr = self.channel.execute_wait('ps -p {} &> /dev/null; echo "STATUS:$?" '.format(self.resources[job_id]['remote_pid']),
113                                                                     self.cmd_timeout)
114                 for line in stdout.split('\n'):
115                     if line.startswith("STATUS:"):
116                         status = line.split("STATUS:")[1].strip()
117                         if status == "0":
118                             self.resources[job_id]['status'] = 'RUNNING'
119                         else:
120                             self.resources[job_id]['status'] = 'FAILED'
121 
122         return [self.resources[jid]['status'] for jid in job_ids]
123 
124     def _write_submit_script(self, script_string, script_filename):
125         '''
126         Load the template string with config values and write the generated submit script to
127         a submit script file.
128 
129         Args:
130               - template_string (string) : The template string to be used for the writing submit script
131               - script_filename (string) : Name of the submit script
132 
133         Returns:
134               - True: on success
135 
136         Raises:
137               SchedulerMissingArgs : If template is missing args
138               ScriptPathError : Unable to write submit script out
139         '''
140 
141         try:
142             with open(script_filename, 'w') as f:
143                 f.write(script_string)
144 
145         except KeyError as e:
146             logger.error("Missing keys for submit script: %s", e)
147             raise (SchedulerMissingArgs(e.args, self.label))
148 
149         except IOError as e:
150             logger.error("Failed writing to submit script: %s", script_filename)
151             raise (ScriptPathError(script_filename, e))
152 
153         return True
154 
155     def submit(self, command, tasks_per_node, job_name="parsl.auto"):
156         ''' Submits the command onto an Local Resource Manager job.
157         Submit returns an ID that corresponds to the task that was just submitted.
158 
159         If tasks_per_node <  1:
160              1/tasks_per_node is provisioned
161 
162         If tasks_per_node == 1:
163              A single node is provisioned
164 
165         If tasks_per_node >  1 :
166              tasks_per_node nodes are provisioned.
167 
168         Args:
169              - command  :(String) Commandline invocation to be made on the remote side.
170              - tasks_per_node (int) : command invocations to be launched per node
171 
172         Kwargs:
173              - job_name (String): Name for job, must be unique
174 
175         Returns:
176              - None: At capacity, cannot provision more
177              - job_id: (string) Identifier for the job
178 
179         '''
180 
181         job_name = "{0}.{1}".format(job_name, time.time())
182 
183         # Set script path
184         script_path = "{0}/{1}.sh".format(self.script_dir, job_name)
185         script_path = os.path.abspath(script_path)
186 
187         wrap_command = self.worker_init + '\n' + self.launcher(command, tasks_per_node, self.nodes_per_block)
188 
189         self._write_submit_script(wrap_command, script_path)
190 
191         job_id = None
192         proc = None
193         remote_pid = None
194         if (self.move_files is None and not isinstance(self.channel, LocalChannel)) or (self.move_files):
195             logger.debug("Pushing start script")
196             script_path = self.channel.push_file(script_path, self.channel.script_dir)
197 
198         if not isinstance(self.channel, LocalChannel):
199             logger.debug("Launching in remote mode")
200             # Bash would return until the streams are closed. So we redirect to a outs file
201             cmd = 'bash {0} &> {0}.out & \n echo "PID:$!" '.format(script_path)
202             retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)
203             for line in stdout.split('\n'):
204                 if line.startswith("PID:"):
205                     remote_pid = line.split("PID:")[1].strip()
206                     job_id = remote_pid
207             if job_id is None:
208                 logger.warning("Channel failed to start remote command/retrieve PID")
209         else:
210 
211             try:
212                 job_id, proc = self.channel.execute_no_wait('bash {0}'.format(script_path), self.cmd_timeout)
213             except Exception as e:
214                 logger.debug("Channel execute failed for: {}, {}".format(self.channel, e))
215                 raise
216 
217         self.resources[job_id] = {'job_id': job_id, 'status': 'RUNNING',
218                                   'remote_pid': remote_pid,
219                                   'proc': proc}
220 
221         return job_id
222 
223     def cancel(self, job_ids):
224         ''' Cancels the jobs specified by a list of job ids
225 
226         Args:
227         job_ids : [<job_id> ...]
228 
229         Returns :
230         [True/False...] : If the cancel operation fails the entire list will be False.
231         '''
232         for job in job_ids:
233             logger.debug("Terminating job/proc_id: {0}".format(job))
234             # Here we are assuming that for local, the job_ids are the process id's
235             if self.resources[job]['proc']:
236                 proc = self.resources[job]['proc']
237                 os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
238                 self.resources[job]['status'] = 'CANCELLED'
239 
240             elif self.resources[job]['remote_pid']:
241                 cmd = "kill -- -$(ps -o pgid={} | grep -o '[0-9]*')".format(self.resources[job]['remote_pid'])
242                 retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)
243                 if retcode != 0:
244                     logger.warning("Failed to kill PID: {} and child processes on {}".format(self.resources[job]['remote_pid'],
245                                                                                              self.label))
246 
247         rets = [True for i in job_ids]
248         return rets
249 
250     @property
251     def current_capacity(self):
252         return len(self.resources)
253 
254     @property
255     def label(self):
256         return self._label
257 
258 
259 if __name__ == "__main__":
260 
261     print("Nothing here")
262
```
--- 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
@@ -238,7 +238,7 @@
                 self.resources[job]['status'] = 'CANCELLED'
 
             elif self.resources[job]['remote_pid']:
-                cmd = "kill -- -$(ps -o pgid={} | grep -o '[0-9]*')".format(self.resources[job]['remote_pid'])
+                cmd = "kill -- -$(ps -o pgid= {} | grep -o '[0-9]*')".format(self.resources[job]['remote_pid'])
                 retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)
                 if retcode != 0:
                     logger.warning("Failed to kill PID: {} and child processes on {}".format(self.resources[job]['remote_pid'],

{"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@@ -238,7 +238,7 @@\n                 self.resources[job]['status'] = 'CANCELLED'\n \n             elif self.resources[job]['remote_pid']:\n-                cmd = \"kill -- -$(ps -o pgid={} | grep -o '[0-9]*')\".format(self.resources[job]['remote_pid'])\n+                cmd = \"kill -- -$(ps -o pgid= {} | grep -o '[0-9]*')\".format(self.resources[job]['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(self.resources[job]['remote_pid'],\n", "issue": "Incorrect pgid identification in LocalProvider kills too many processes\nThis code in LocalProvider probably should have a space added to give `-o pgid= {}`. This should be two parameters, one '-o pgid=` meaning output only pgids, and give that single column an empty heading; and secondly the process ID to inspect.\r\n\r\n```\r\n                cmd = \"kill -- -$(ps -o pgid={} | grep -o '[0-9]*')\".format(self.resources[job]['remote_pid'])\r\n```\r\n\r\nThe code as it is probably doesn't shut down the expected process - it might sometimes; it might kill other processes or process groups, as the resulting ps output contains a list of many PGIDs, some of which might be valid PIDs to kill; or it might not kill anything. This might be related to bad shutdown behaviour described in #674 \r\n\r\nI have not tested this but I encountered similar behaviour reviewing  #1297 \n", "before_files": [{"content": "import logging\nimport os\nimport signal\nimport time\n\nfrom parsl.channels import LocalChannel\nfrom parsl.launchers import SingleNodeLauncher\nfrom parsl.providers.provider_base import ExecutionProvider\nfrom parsl.providers.error import SchedulerMissingArgs, ScriptPathError\nfrom parsl.utils import RepresentationMixin\n\nlogger = logging.getLogger(__name__)\n\ntranslate_table = {\n    'PD': 'PENDING',\n    'R': 'RUNNING',\n    'CA': 'CANCELLED',\n    'CF': 'PENDING',  # (configuring),\n    'CG': 'RUNNING',  # (completing),\n    'CD': 'COMPLETED',\n    'F': 'FAILED',\n    'TO': 'TIMEOUT',\n    'NF': 'FAILED',  # (node failure),\n    'RV': 'FAILED',  # (revoked) and\n    'SE': 'FAILED'\n}  # (special exit state\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]: should files be moved? by default, Parsl will try to figure\n        this out itself (= None). 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=4,\n                 min_blocks=0,\n                 max_blocks=10,\n                 walltime=\"00:15:00\",\n                 worker_init='',\n                 cmd_timeout=30,\n                 parallelism=1,\n                 move_files=None):\n        self.channel = channel\n        self._label = 'local'\n        self.provisioned_blocks = 0\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.walltime = walltime\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        logger.debug(\"Checking status of: {0}\".format(job_ids))\n        for job_id in self.resources:\n\n            if self.resources[job_id]['proc']:\n\n                poll_code = self.resources[job_id]['proc'].poll()\n                if self.resources[job_id]['status'] in ['COMPLETED', 'FAILED']:\n                    continue\n\n                if poll_code is None:\n                    self.resources[job_id]['status'] = 'RUNNING'\n                elif poll_code == 0:\n                    self.resources[job_id]['status'] = 'COMPLETED'\n                elif poll_code != 0:\n                    self.resources[job_id]['status'] = 'FAILED'\n                else:\n                    logger.error(\"Internal consistency error: unexpected case in local provider state machine\")\n\n            elif self.resources[job_id]['remote_pid']:\n\n                retcode, stdout, stderr = self.channel.execute_wait('ps -p {} &> /dev/null; echo \"STATUS:$?\" '.format(self.resources[job_id]['remote_pid']),\n                                                                    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                            self.resources[job_id]['status'] = 'RUNNING'\n                        else:\n                            self.resources[job_id]['status'] = 'FAILED'\n\n        return [self.resources[jid]['status'] for jid in job_ids]\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.auto\"):\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 + '\\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        proc = None\n        remote_pid = None\n        if (self.move_files is None and not isinstance(self.channel, LocalChannel)) or (self.move_files):\n            logger.debug(\"Pushing start script\")\n            script_path = self.channel.push_file(script_path, self.channel.script_dir)\n\n        if not isinstance(self.channel, LocalChannel):\n            logger.debug(\"Launching in remote mode\")\n            # Bash would return until the streams are closed. So we redirect to a outs file\n            cmd = 'bash {0} &> {0}.out & \\n echo \"PID:$!\" '.format(script_path)\n            retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)\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                logger.warning(\"Channel failed to start remote command/retrieve PID\")\n        else:\n\n            try:\n                job_id, proc = self.channel.execute_no_wait('bash {0}'.format(script_path), self.cmd_timeout)\n            except Exception as e:\n                logger.debug(\"Channel execute failed for: {}, {}\".format(self.channel, e))\n                raise\n\n        self.resources[job_id] = {'job_id': job_id, 'status': 'RUNNING',\n                                  'remote_pid': remote_pid,\n                                  'proc': proc}\n\n        return job_id\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 :\n        [True/False...] : If the cancel operation fails the entire list will be False.\n        '''\n        for job in job_ids:\n            logger.debug(\"Terminating job/proc_id: {0}\".format(job))\n            # Here we are assuming that for local, the job_ids are the process id's\n            if self.resources[job]['proc']:\n                proc = self.resources[job]['proc']\n                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)\n                self.resources[job]['status'] = 'CANCELLED'\n\n            elif self.resources[job]['remote_pid']:\n                cmd = \"kill -- -$(ps -o pgid={} | grep -o '[0-9]*')\".format(self.resources[job]['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(self.resources[job]['remote_pid'],\n                                                                                             self.label))\n\n        rets = [True for i in job_ids]\n        return rets\n\n    @property\n    def current_capacity(self):\n        return len(self.resources)\n\n    @property\n    def label(self):\n        return self._label\n\n\nif __name__ == \"__main__\":\n\n    print(\"Nothing here\")\n", "path": "parsl/providers/local/local.py"}], "after_files": [{"content": "import logging\nimport os\nimport signal\nimport time\n\nfrom parsl.channels import LocalChannel\nfrom parsl.launchers import SingleNodeLauncher\nfrom parsl.providers.provider_base import ExecutionProvider\nfrom parsl.providers.error import SchedulerMissingArgs, ScriptPathError\nfrom parsl.utils import RepresentationMixin\n\nlogger = logging.getLogger(__name__)\n\ntranslate_table = {\n    'PD': 'PENDING',\n    'R': 'RUNNING',\n    'CA': 'CANCELLED',\n    'CF': 'PENDING',  # (configuring),\n    'CG': 'RUNNING',  # (completing),\n    'CD': 'COMPLETED',\n    'F': 'FAILED',\n    'TO': 'TIMEOUT',\n    'NF': 'FAILED',  # (node failure),\n    'RV': 'FAILED',  # (revoked) and\n    'SE': 'FAILED'\n}  # (special exit state\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]: should files be moved? by default, Parsl will try to figure\n        this out itself (= None). 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=4,\n                 min_blocks=0,\n                 max_blocks=10,\n                 walltime=\"00:15:00\",\n                 worker_init='',\n                 cmd_timeout=30,\n                 parallelism=1,\n                 move_files=None):\n        self.channel = channel\n        self._label = 'local'\n        self.provisioned_blocks = 0\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.walltime = walltime\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        logger.debug(\"Checking status of: {0}\".format(job_ids))\n        for job_id in self.resources:\n\n            if self.resources[job_id]['proc']:\n\n                poll_code = self.resources[job_id]['proc'].poll()\n                if self.resources[job_id]['status'] in ['COMPLETED', 'FAILED']:\n                    continue\n\n                if poll_code is None:\n                    self.resources[job_id]['status'] = 'RUNNING'\n                elif poll_code == 0:\n                    self.resources[job_id]['status'] = 'COMPLETED'\n                elif poll_code != 0:\n                    self.resources[job_id]['status'] = 'FAILED'\n                else:\n                    logger.error(\"Internal consistency error: unexpected case in local provider state machine\")\n\n            elif self.resources[job_id]['remote_pid']:\n\n                retcode, stdout, stderr = self.channel.execute_wait('ps -p {} &> /dev/null; echo \"STATUS:$?\" '.format(self.resources[job_id]['remote_pid']),\n                                                                    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                            self.resources[job_id]['status'] = 'RUNNING'\n                        else:\n                            self.resources[job_id]['status'] = 'FAILED'\n\n        return [self.resources[jid]['status'] for jid in job_ids]\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.auto\"):\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 + '\\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        proc = None\n        remote_pid = None\n        if (self.move_files is None and not isinstance(self.channel, LocalChannel)) or (self.move_files):\n            logger.debug(\"Pushing start script\")\n            script_path = self.channel.push_file(script_path, self.channel.script_dir)\n\n        if not isinstance(self.channel, LocalChannel):\n            logger.debug(\"Launching in remote mode\")\n            # Bash would return until the streams are closed. So we redirect to a outs file\n            cmd = 'bash {0} &> {0}.out & \\n echo \"PID:$!\" '.format(script_path)\n            retcode, stdout, stderr = self.channel.execute_wait(cmd, self.cmd_timeout)\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                logger.warning(\"Channel failed to start remote command/retrieve PID\")\n        else:\n\n            try:\n                job_id, proc = self.channel.execute_no_wait('bash {0}'.format(script_path), self.cmd_timeout)\n            except Exception as e:\n                logger.debug(\"Channel execute failed for: {}, {}\".format(self.channel, e))\n                raise\n\n        self.resources[job_id] = {'job_id': job_id, 'status': 'RUNNING',\n                                  'remote_pid': remote_pid,\n                                  'proc': proc}\n\n        return job_id\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 :\n        [True/False...] : If the cancel operation fails the entire list will be False.\n        '''\n        for job in job_ids:\n            logger.debug(\"Terminating job/proc_id: {0}\".format(job))\n            # Here we are assuming that for local, the job_ids are the process id's\n            if self.resources[job]['proc']:\n                proc = self.resources[job]['proc']\n                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)\n                self.resources[job]['status'] = 'CANCELLED'\n\n            elif self.resources[job]['remote_pid']:\n                cmd = \"kill -- -$(ps -o pgid= {} | grep -o '[0-9]*')\".format(self.resources[job]['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(self.resources[job]['remote_pid'],\n                                                                                             self.label))\n\n        rets = [True for i in job_ids]\n        return rets\n\n    @property\n    def current_capacity(self):\n        return len(self.resources)\n\n    @property\n    def label(self):\n        return self._label\n\n\nif __name__ == \"__main__\":\n\n    print(\"Nothing here\")\n", "path": "parsl/providers/local/local.py"}]}