We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`SlurmProvider` is missing a docstring for `init_blocks`
**Describe the bug**
The `SlurmProvider` is missing the docstring for its `init_blocks` keyword argument [here](https://github.com/Parsl/parsl/blob/72df44c105a294b90d1b7f9022ed24309fc41885/parsl/providers/slurm/slurm.py#L37).


--- 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/slurm/slurm.py`
Content:
```
1 import os
2 import math
3 import time
4 import logging
5 import re
6 import typeguard
7 
8 from typing import Optional
9 
10 from parsl.channels import LocalChannel
11 from parsl.channels.base import Channel
12 from parsl.jobs.states import JobState, JobStatus
13 from parsl.launchers import SingleNodeLauncher
14 from parsl.launchers.base import Launcher
15 from parsl.providers.cluster_provider import ClusterProvider
16 from parsl.providers.errors import SubmitException
17 from parsl.providers.slurm.template import template_string
18 from parsl.utils import RepresentationMixin, wtime_to_minutes
19 
20 logger = logging.getLogger(__name__)
21 
22 translate_table = {
23     'PD': JobState.PENDING,
24     'R': JobState.RUNNING,
25     'CA': JobState.CANCELLED,
26     'CF': JobState.PENDING,  # (configuring),
27     'CG': JobState.RUNNING,  # (completing),
28     'CD': JobState.COMPLETED,
29     'F': JobState.FAILED,  # (failed),
30     'TO': JobState.TIMEOUT,  # (timeout),
31     'NF': JobState.FAILED,  # (node failure),
32     'RV': JobState.FAILED,  # (revoked) and
33     'SE': JobState.FAILED   # (special exit state)
34 }
35 
36 
37 class SlurmProvider(ClusterProvider, RepresentationMixin):
38     """Slurm Execution Provider
39 
40     This provider uses sbatch to submit, squeue for status and scancel to cancel
41     jobs. The sbatch script to be used is created from a template file in this
42     same module.
43 
44     Parameters
45     ----------
46     partition : str
47         Slurm partition to request blocks from. If unspecified or ``None``, no partition slurm directive will be specified.
48     account : str
49         Slurm account to which to charge resources used by the job. If unspecified or ``None``, the job will use the
50         user's default account.
51     qos : str
52         Slurm queue to place job in. If unspecified or ``None``, no queue slurm directive will be specified.
53     constraint : str
54         Slurm job constraint, often used to choose cpu or gpu type. If unspecified or ``None``, no constraint slurm directive will be added.
55     channel : Channel
56         Channel for accessing this provider. Possible channels include
57         :class:`~parsl.channels.LocalChannel` (the default),
58         :class:`~parsl.channels.SSHChannel`, or
59         :class:`~parsl.channels.SSHInteractiveLoginChannel`.
60     nodes_per_block : int
61         Nodes to provision per block.
62     cores_per_node : int
63         Specify the number of cores to provision per node. If set to None, executors
64         will assume all cores on the node are available for computation. Default is None.
65     mem_per_node : int
66         Specify the real memory to provision per node in GB. If set to None, no
67         explicit request to the scheduler will be made. Default is None.
68     min_blocks : int
69         Minimum number of blocks to maintain.
70     max_blocks : int
71         Maximum number of blocks to maintain.
72     parallelism : float
73         Ratio of provisioned task slots to active tasks. A parallelism value of 1 represents aggressive
74         scaling where as many resources as possible are used; parallelism close to 0 represents
75         the opposite situation in which as few resources as possible (i.e., min_blocks) are used.
76     walltime : str
77         Walltime requested per block in HH:MM:SS.
78     scheduler_options : str
79         String to prepend to the #SBATCH blocks in the submit script to the scheduler.
80     regex_job_id : str
81         The regular expression used to extract the job ID from the ``sbatch`` standard output.
82         The default is ``r"Submitted batch job (?P<id>\\S*)"``, where ``id`` is the regular expression
83         symbolic group for the job ID.
84     worker_init : str
85         Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.
86     exclusive : bool (Default = True)
87         Requests nodes which are not shared with other running jobs.
88     launcher : Launcher
89         Launcher for this provider. Possible launchers include
90         :class:`~parsl.launchers.SingleNodeLauncher` (the default),
91         :class:`~parsl.launchers.SrunLauncher`, or
92         :class:`~parsl.launchers.AprunLauncher`
93     move_files : Optional[Bool]: should files be moved? by default, Parsl will try to move files.
94     """
95 
96     @typeguard.typechecked
97     def __init__(self,
98                  partition: Optional[str] = None,
99                  account: Optional[str] = None,
100                  qos: Optional[str] = None,
101                  constraint: Optional[str] = None,
102                  channel: Channel = LocalChannel(),
103                  nodes_per_block: int = 1,
104                  cores_per_node: Optional[int] = None,
105                  mem_per_node: Optional[int] = None,
106                  init_blocks: int = 1,
107                  min_blocks: int = 0,
108                  max_blocks: int = 1,
109                  parallelism: float = 1,
110                  walltime: str = "00:10:00",
111                  scheduler_options: str = '',
112                  regex_job_id: str = r"Submitted batch job (?P<id>\S*)",
113                  worker_init: str = '',
114                  cmd_timeout: int = 10,
115                  exclusive: bool = True,
116                  move_files: bool = True,
117                  launcher: Launcher = SingleNodeLauncher()):
118         label = 'slurm'
119         super().__init__(label,
120                          channel,
121                          nodes_per_block,
122                          init_blocks,
123                          min_blocks,
124                          max_blocks,
125                          parallelism,
126                          walltime,
127                          cmd_timeout=cmd_timeout,
128                          launcher=launcher)
129 
130         self.partition = partition
131         self.cores_per_node = cores_per_node
132         self.mem_per_node = mem_per_node
133         self.exclusive = exclusive
134         self.move_files = move_files
135         self.account = account
136         self.qos = qos
137         self.constraint = constraint
138         self.scheduler_options = scheduler_options + '\n'
139         if exclusive:
140             self.scheduler_options += "#SBATCH --exclusive\n"
141         if partition:
142             self.scheduler_options += "#SBATCH --partition={}\n".format(partition)
143         if account:
144             self.scheduler_options += "#SBATCH --account={}\n".format(account)
145         if qos:
146             self.scheduler_options += "#SBATCH --qos={}\n".format(qos)
147         if constraint:
148             self.scheduler_options += "#SBATCH --constraint={}\n".format(constraint)
149 
150         self.regex_job_id = regex_job_id
151         self.worker_init = worker_init + '\n'
152 
153     def _status(self):
154         '''Returns the status list for a list of job_ids
155 
156         Args:
157               self
158 
159         Returns:
160               [status...] : Status list of all jobs
161         '''
162         job_id_list = ','.join(
163             [jid for jid, job in self.resources.items() if not job['status'].terminal]
164         )
165         if not job_id_list:
166             logger.debug('No active jobs, skipping status update')
167             return
168 
169         cmd = "squeue --noheader --format='%i %t' --job '{0}'".format(job_id_list)
170         logger.debug("Executing %s", cmd)
171         retcode, stdout, stderr = self.execute_wait(cmd)
172         logger.debug("squeue returned %s %s", stdout, stderr)
173 
174         # Execute_wait failed. Do no update
175         if retcode != 0:
176             logger.warning("squeue failed with non-zero exit code {}".format(retcode))
177             return
178 
179         jobs_missing = set(self.resources.keys())
180         for line in stdout.split('\n'):
181             if not line:
182                 # Blank line
183                 continue
184             job_id, slurm_state = line.split()
185             if slurm_state not in translate_table:
186                 logger.warning(f"Slurm status {slurm_state} is not recognized")
187             status = translate_table.get(slurm_state, JobState.UNKNOWN)
188             logger.debug("Updating job {} with slurm status {} to parsl state {!s}".format(job_id, slurm_state, status))
189             self.resources[job_id]['status'] = JobStatus(status)
190             jobs_missing.remove(job_id)
191 
192         # squeue does not report on jobs that are not running. So we are filling in the
193         # blanks for missing jobs, we might lose some information about why the jobs failed.
194         for missing_job in jobs_missing:
195             logger.debug("Updating missing job {} to completed status".format(missing_job))
196             self.resources[missing_job]['status'] = JobStatus(JobState.COMPLETED)
197 
198     def submit(self, command: str, tasks_per_node: int, job_name="parsl.slurm") -> str:
199         """Submit the command as a slurm job.
200 
201         Parameters
202         ----------
203         command : str
204             Command to be made on the remote side.
205         tasks_per_node : int
206             Command invocations to be launched per node
207         job_name : str
208             Name for the job
209         Returns
210         -------
211         job id : str
212             A string identifier for the job
213         """
214 
215         scheduler_options = self.scheduler_options
216         worker_init = self.worker_init
217         if self.mem_per_node is not None:
218             scheduler_options += '#SBATCH --mem={}g\n'.format(self.mem_per_node)
219             worker_init += 'export PARSL_MEMORY_GB={}\n'.format(self.mem_per_node)
220         if self.cores_per_node is not None:
221             cpus_per_task = math.floor(self.cores_per_node / tasks_per_node)
222             scheduler_options += '#SBATCH --cpus-per-task={}'.format(cpus_per_task)
223             worker_init += 'export PARSL_CORES={}\n'.format(cpus_per_task)
224 
225         job_name = "{0}.{1}".format(job_name, time.time())
226 
227         script_path = "{0}/{1}.submit".format(self.script_dir, job_name)
228         script_path = os.path.abspath(script_path)
229 
230         logger.debug("Requesting one block with {} nodes".format(self.nodes_per_block))
231 
232         job_config = {}
233         job_config["submit_script_dir"] = self.channel.script_dir
234         job_config["nodes"] = self.nodes_per_block
235         job_config["tasks_per_node"] = tasks_per_node
236         job_config["walltime"] = wtime_to_minutes(self.walltime)
237         job_config["scheduler_options"] = scheduler_options
238         job_config["worker_init"] = worker_init
239         job_config["user_script"] = command
240 
241         # Wrap the command
242         job_config["user_script"] = self.launcher(command,
243                                                   tasks_per_node,
244                                                   self.nodes_per_block)
245 
246         logger.debug("Writing submit script")
247         self._write_submit_script(template_string, script_path, job_name, job_config)
248 
249         if self.move_files:
250             logger.debug("moving files")
251             channel_script_path = self.channel.push_file(script_path, self.channel.script_dir)
252         else:
253             logger.debug("not moving files")
254             channel_script_path = script_path
255 
256         retcode, stdout, stderr = self.execute_wait("sbatch {0}".format(channel_script_path))
257 
258         if retcode == 0:
259             for line in stdout.split('\n'):
260                 match = re.match(self.regex_job_id, line)
261                 if match:
262                     job_id = match.group("id")
263                     self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}
264                     return job_id
265             else:
266                 logger.error("Could not read job ID from submit command standard output.")
267                 logger.error("Retcode:%s STDOUT:%s STDERR:%s", retcode, stdout.strip(), stderr.strip())
268                 raise SubmitException(job_name, "Could not read job ID from submit command standard output", stdout=stdout, stderr=stderr, retcode=retcode)
269         else:
270             logger.error("Submit command failed")
271             logger.error("Retcode:%s STDOUT:%s STDERR:%s", retcode, stdout.strip(), stderr.strip())
272             raise SubmitException(job_name, "Could not read job ID from submit command standard output", stdout=stdout, stderr=stderr, retcode=retcode)
273 
274     def cancel(self, job_ids):
275         ''' Cancels the jobs specified by a list of job ids
276 
277         Args:
278         job_ids : [<job_id> ...]
279 
280         Returns :
281         [True/False...] : If the cancel operation fails the entire list will be False.
282         '''
283 
284         job_id_list = ' '.join(job_ids)
285         retcode, stdout, stderr = self.execute_wait("scancel {0}".format(job_id_list))
286         rets = None
287         if retcode == 0:
288             for jid in job_ids:
289                 self.resources[jid]['status'] = JobStatus(JobState.CANCELLED)  # Setting state to cancelled
290             rets = [True for i in job_ids]
291         else:
292             rets = [False for i in job_ids]
293 
294         return rets
295 
296     @property
297     def status_polling_interval(self):
298         return 60
299
```
--- 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/slurm/slurm.py b/parsl/providers/slurm/slurm.py
--- a/parsl/providers/slurm/slurm.py
+++ b/parsl/providers/slurm/slurm.py
@@ -65,6 +65,8 @@
     mem_per_node : int
         Specify the real memory to provision per node in GB. If set to None, no
         explicit request to the scheduler will be made. Default is None.
+    init_blocks : int
+        Number of blocks to provision at the start of the run. Default is 1.
     min_blocks : int
         Minimum number of blocks to maintain.
     max_blocks : int

{"golden_diff": "diff --git a/parsl/providers/slurm/slurm.py b/parsl/providers/slurm/slurm.py\n--- a/parsl/providers/slurm/slurm.py\n+++ b/parsl/providers/slurm/slurm.py\n@@ -65,6 +65,8 @@\n     mem_per_node : int\n         Specify the real memory to provision per node in GB. If set to None, no\n         explicit request to the scheduler will be made. Default is None.\n+    init_blocks : int\n+        Number of blocks to provision at the start of the run. Default is 1.\n     min_blocks : int\n         Minimum number of blocks to maintain.\n     max_blocks : int\n", "issue": "`SlurmProvider` is missing a docstring for `init_blocks`\n**Describe the bug**\r\nThe `SlurmProvider` is missing the docstring for its `init_blocks` keyword argument [here](https://github.com/Parsl/parsl/blob/72df44c105a294b90d1b7f9022ed24309fc41885/parsl/providers/slurm/slurm.py#L37).\r\n\n", "before_files": [{"content": "import os\nimport math\nimport time\nimport logging\nimport re\nimport typeguard\n\nfrom typing import Optional\n\nfrom parsl.channels import LocalChannel\nfrom parsl.channels.base import Channel\nfrom parsl.jobs.states import JobState, JobStatus\nfrom parsl.launchers import SingleNodeLauncher\nfrom parsl.launchers.base import Launcher\nfrom parsl.providers.cluster_provider import ClusterProvider\nfrom parsl.providers.errors import SubmitException\nfrom parsl.providers.slurm.template import template_string\nfrom parsl.utils import RepresentationMixin, wtime_to_minutes\n\nlogger = logging.getLogger(__name__)\n\ntranslate_table = {\n    'PD': JobState.PENDING,\n    'R': JobState.RUNNING,\n    'CA': JobState.CANCELLED,\n    'CF': JobState.PENDING,  # (configuring),\n    'CG': JobState.RUNNING,  # (completing),\n    'CD': JobState.COMPLETED,\n    'F': JobState.FAILED,  # (failed),\n    'TO': JobState.TIMEOUT,  # (timeout),\n    'NF': JobState.FAILED,  # (node failure),\n    'RV': JobState.FAILED,  # (revoked) and\n    'SE': JobState.FAILED   # (special exit state)\n}\n\n\nclass SlurmProvider(ClusterProvider, RepresentationMixin):\n    \"\"\"Slurm Execution Provider\n\n    This provider uses sbatch to submit, squeue for status and scancel to cancel\n    jobs. The sbatch script to be used is created from a template file in this\n    same module.\n\n    Parameters\n    ----------\n    partition : str\n        Slurm partition to request blocks from. If unspecified or ``None``, no partition slurm directive will be specified.\n    account : str\n        Slurm account to which to charge resources used by the job. If unspecified or ``None``, the job will use the\n        user's default account.\n    qos : str\n        Slurm queue to place job in. If unspecified or ``None``, no queue slurm directive will be specified.\n    constraint : str\n        Slurm job constraint, often used to choose cpu or gpu type. If unspecified or ``None``, no constraint slurm directive will be added.\n    channel : Channel\n        Channel for accessing this provider. Possible channels include\n        :class:`~parsl.channels.LocalChannel` (the default),\n        :class:`~parsl.channels.SSHChannel`, or\n        :class:`~parsl.channels.SSHInteractiveLoginChannel`.\n    nodes_per_block : int\n        Nodes to provision per block.\n    cores_per_node : int\n        Specify the number of cores to provision per node. If set to None, executors\n        will assume all cores on the node are available for computation. Default is None.\n    mem_per_node : int\n        Specify the real memory to provision per node in GB. If set to None, no\n        explicit request to the scheduler will be made. Default is None.\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    walltime : str\n        Walltime requested per block in HH:MM:SS.\n    scheduler_options : str\n        String to prepend to the #SBATCH blocks in the submit script to the scheduler.\n    regex_job_id : str\n        The regular expression used to extract the job ID from the ``sbatch`` standard output.\n        The default is ``r\"Submitted batch job (?P<id>\\\\S*)\"``, where ``id`` is the regular expression\n        symbolic group for the job ID.\n    worker_init : str\n        Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.\n    exclusive : bool (Default = True)\n        Requests nodes which are not shared with other running jobs.\n    launcher : Launcher\n        Launcher for this provider. Possible launchers include\n        :class:`~parsl.launchers.SingleNodeLauncher` (the default),\n        :class:`~parsl.launchers.SrunLauncher`, or\n        :class:`~parsl.launchers.AprunLauncher`\n    move_files : Optional[Bool]: should files be moved? by default, Parsl will try to move files.\n    \"\"\"\n\n    @typeguard.typechecked\n    def __init__(self,\n                 partition: Optional[str] = None,\n                 account: Optional[str] = None,\n                 qos: Optional[str] = None,\n                 constraint: Optional[str] = None,\n                 channel: Channel = LocalChannel(),\n                 nodes_per_block: int = 1,\n                 cores_per_node: Optional[int] = None,\n                 mem_per_node: Optional[int] = None,\n                 init_blocks: int = 1,\n                 min_blocks: int = 0,\n                 max_blocks: int = 1,\n                 parallelism: float = 1,\n                 walltime: str = \"00:10:00\",\n                 scheduler_options: str = '',\n                 regex_job_id: str = r\"Submitted batch job (?P<id>\\S*)\",\n                 worker_init: str = '',\n                 cmd_timeout: int = 10,\n                 exclusive: bool = True,\n                 move_files: bool = True,\n                 launcher: Launcher = SingleNodeLauncher()):\n        label = 'slurm'\n        super().__init__(label,\n                         channel,\n                         nodes_per_block,\n                         init_blocks,\n                         min_blocks,\n                         max_blocks,\n                         parallelism,\n                         walltime,\n                         cmd_timeout=cmd_timeout,\n                         launcher=launcher)\n\n        self.partition = partition\n        self.cores_per_node = cores_per_node\n        self.mem_per_node = mem_per_node\n        self.exclusive = exclusive\n        self.move_files = move_files\n        self.account = account\n        self.qos = qos\n        self.constraint = constraint\n        self.scheduler_options = scheduler_options + '\\n'\n        if exclusive:\n            self.scheduler_options += \"#SBATCH --exclusive\\n\"\n        if partition:\n            self.scheduler_options += \"#SBATCH --partition={}\\n\".format(partition)\n        if account:\n            self.scheduler_options += \"#SBATCH --account={}\\n\".format(account)\n        if qos:\n            self.scheduler_options += \"#SBATCH --qos={}\\n\".format(qos)\n        if constraint:\n            self.scheduler_options += \"#SBATCH --constraint={}\\n\".format(constraint)\n\n        self.regex_job_id = regex_job_id\n        self.worker_init = worker_init + '\\n'\n\n    def _status(self):\n        '''Returns the status list for a list of job_ids\n\n        Args:\n              self\n\n        Returns:\n              [status...] : Status list of all jobs\n        '''\n        job_id_list = ','.join(\n            [jid for jid, job in self.resources.items() if not job['status'].terminal]\n        )\n        if not job_id_list:\n            logger.debug('No active jobs, skipping status update')\n            return\n\n        cmd = \"squeue --noheader --format='%i %t' --job '{0}'\".format(job_id_list)\n        logger.debug(\"Executing %s\", cmd)\n        retcode, stdout, stderr = self.execute_wait(cmd)\n        logger.debug(\"squeue returned %s %s\", stdout, stderr)\n\n        # Execute_wait failed. Do no update\n        if retcode != 0:\n            logger.warning(\"squeue failed with non-zero exit code {}\".format(retcode))\n            return\n\n        jobs_missing = set(self.resources.keys())\n        for line in stdout.split('\\n'):\n            if not line:\n                # Blank line\n                continue\n            job_id, slurm_state = line.split()\n            if slurm_state not in translate_table:\n                logger.warning(f\"Slurm status {slurm_state} is not recognized\")\n            status = translate_table.get(slurm_state, JobState.UNKNOWN)\n            logger.debug(\"Updating job {} with slurm status {} to parsl state {!s}\".format(job_id, slurm_state, status))\n            self.resources[job_id]['status'] = JobStatus(status)\n            jobs_missing.remove(job_id)\n\n        # squeue does not report on jobs that are not running. So we are filling in the\n        # blanks for missing jobs, we might lose some information about why the jobs failed.\n        for missing_job in jobs_missing:\n            logger.debug(\"Updating missing job {} to completed status\".format(missing_job))\n            self.resources[missing_job]['status'] = JobStatus(JobState.COMPLETED)\n\n    def submit(self, command: str, tasks_per_node: int, job_name=\"parsl.slurm\") -> str:\n        \"\"\"Submit the command as a slurm job.\n\n        Parameters\n        ----------\n        command : str\n            Command to be made on the remote side.\n        tasks_per_node : int\n            Command invocations to be launched per node\n        job_name : str\n            Name for the job\n        Returns\n        -------\n        job id : str\n            A string identifier for the job\n        \"\"\"\n\n        scheduler_options = self.scheduler_options\n        worker_init = self.worker_init\n        if self.mem_per_node is not None:\n            scheduler_options += '#SBATCH --mem={}g\\n'.format(self.mem_per_node)\n            worker_init += 'export PARSL_MEMORY_GB={}\\n'.format(self.mem_per_node)\n        if self.cores_per_node is not None:\n            cpus_per_task = math.floor(self.cores_per_node / tasks_per_node)\n            scheduler_options += '#SBATCH --cpus-per-task={}'.format(cpus_per_task)\n            worker_init += 'export PARSL_CORES={}\\n'.format(cpus_per_task)\n\n        job_name = \"{0}.{1}\".format(job_name, time.time())\n\n        script_path = \"{0}/{1}.submit\".format(self.script_dir, job_name)\n        script_path = os.path.abspath(script_path)\n\n        logger.debug(\"Requesting one block with {} nodes\".format(self.nodes_per_block))\n\n        job_config = {}\n        job_config[\"submit_script_dir\"] = self.channel.script_dir\n        job_config[\"nodes\"] = self.nodes_per_block\n        job_config[\"tasks_per_node\"] = tasks_per_node\n        job_config[\"walltime\"] = wtime_to_minutes(self.walltime)\n        job_config[\"scheduler_options\"] = scheduler_options\n        job_config[\"worker_init\"] = worker_init\n        job_config[\"user_script\"] = command\n\n        # Wrap the command\n        job_config[\"user_script\"] = self.launcher(command,\n                                                  tasks_per_node,\n                                                  self.nodes_per_block)\n\n        logger.debug(\"Writing submit script\")\n        self._write_submit_script(template_string, script_path, job_name, job_config)\n\n        if self.move_files:\n            logger.debug(\"moving files\")\n            channel_script_path = self.channel.push_file(script_path, self.channel.script_dir)\n        else:\n            logger.debug(\"not moving files\")\n            channel_script_path = script_path\n\n        retcode, stdout, stderr = self.execute_wait(\"sbatch {0}\".format(channel_script_path))\n\n        if retcode == 0:\n            for line in stdout.split('\\n'):\n                match = re.match(self.regex_job_id, line)\n                if match:\n                    job_id = match.group(\"id\")\n                    self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}\n                    return job_id\n            else:\n                logger.error(\"Could not read job ID from submit command standard output.\")\n                logger.error(\"Retcode:%s STDOUT:%s STDERR:%s\", retcode, stdout.strip(), stderr.strip())\n                raise SubmitException(job_name, \"Could not read job ID from submit command standard output\", stdout=stdout, stderr=stderr, retcode=retcode)\n        else:\n            logger.error(\"Submit command failed\")\n            logger.error(\"Retcode:%s STDOUT:%s STDERR:%s\", retcode, stdout.strip(), stderr.strip())\n            raise SubmitException(job_name, \"Could not read job ID from submit command standard output\", stdout=stdout, stderr=stderr, retcode=retcode)\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\n        job_id_list = ' '.join(job_ids)\n        retcode, stdout, stderr = self.execute_wait(\"scancel {0}\".format(job_id_list))\n        rets = None\n        if retcode == 0:\n            for jid in job_ids:\n                self.resources[jid]['status'] = JobStatus(JobState.CANCELLED)  # Setting state to cancelled\n            rets = [True for i in job_ids]\n        else:\n            rets = [False for i in job_ids]\n\n        return rets\n\n    @property\n    def status_polling_interval(self):\n        return 60\n", "path": "parsl/providers/slurm/slurm.py"}], "after_files": [{"content": "import os\nimport math\nimport time\nimport logging\nimport re\nimport typeguard\n\nfrom typing import Optional\n\nfrom parsl.channels import LocalChannel\nfrom parsl.channels.base import Channel\nfrom parsl.jobs.states import JobState, JobStatus\nfrom parsl.launchers import SingleNodeLauncher\nfrom parsl.launchers.base import Launcher\nfrom parsl.providers.cluster_provider import ClusterProvider\nfrom parsl.providers.errors import SubmitException\nfrom parsl.providers.slurm.template import template_string\nfrom parsl.utils import RepresentationMixin, wtime_to_minutes\n\nlogger = logging.getLogger(__name__)\n\ntranslate_table = {\n    'PD': JobState.PENDING,\n    'R': JobState.RUNNING,\n    'CA': JobState.CANCELLED,\n    'CF': JobState.PENDING,  # (configuring),\n    'CG': JobState.RUNNING,  # (completing),\n    'CD': JobState.COMPLETED,\n    'F': JobState.FAILED,  # (failed),\n    'TO': JobState.TIMEOUT,  # (timeout),\n    'NF': JobState.FAILED,  # (node failure),\n    'RV': JobState.FAILED,  # (revoked) and\n    'SE': JobState.FAILED   # (special exit state)\n}\n\n\nclass SlurmProvider(ClusterProvider, RepresentationMixin):\n    \"\"\"Slurm Execution Provider\n\n    This provider uses sbatch to submit, squeue for status and scancel to cancel\n    jobs. The sbatch script to be used is created from a template file in this\n    same module.\n\n    Parameters\n    ----------\n    partition : str\n        Slurm partition to request blocks from. If unspecified or ``None``, no partition slurm directive will be specified.\n    account : str\n        Slurm account to which to charge resources used by the job. If unspecified or ``None``, the job will use the\n        user's default account.\n    qos : str\n        Slurm queue to place job in. If unspecified or ``None``, no queue slurm directive will be specified.\n    constraint : str\n        Slurm job constraint, often used to choose cpu or gpu type. If unspecified or ``None``, no constraint slurm directive will be added.\n    channel : Channel\n        Channel for accessing this provider. Possible channels include\n        :class:`~parsl.channels.LocalChannel` (the default),\n        :class:`~parsl.channels.SSHChannel`, or\n        :class:`~parsl.channels.SSHInteractiveLoginChannel`.\n    nodes_per_block : int\n        Nodes to provision per block.\n    cores_per_node : int\n        Specify the number of cores to provision per node. If set to None, executors\n        will assume all cores on the node are available for computation. Default is None.\n    mem_per_node : int\n        Specify the real memory to provision per node in GB. If set to None, no\n        explicit request to the scheduler will be made. Default is None.\n    init_blocks : int\n        Number of blocks to provision at the start of the run. Default is 1.\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    walltime : str\n        Walltime requested per block in HH:MM:SS.\n    scheduler_options : str\n        String to prepend to the #SBATCH blocks in the submit script to the scheduler.\n    regex_job_id : str\n        The regular expression used to extract the job ID from the ``sbatch`` standard output.\n        The default is ``r\"Submitted batch job (?P<id>\\\\S*)\"``, where ``id`` is the regular expression\n        symbolic group for the job ID.\n    worker_init : str\n        Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.\n    exclusive : bool (Default = True)\n        Requests nodes which are not shared with other running jobs.\n    launcher : Launcher\n        Launcher for this provider. Possible launchers include\n        :class:`~parsl.launchers.SingleNodeLauncher` (the default),\n        :class:`~parsl.launchers.SrunLauncher`, or\n        :class:`~parsl.launchers.AprunLauncher`\n    move_files : Optional[Bool]: should files be moved? by default, Parsl will try to move files.\n    \"\"\"\n\n    @typeguard.typechecked\n    def __init__(self,\n                 partition: Optional[str] = None,\n                 account: Optional[str] = None,\n                 qos: Optional[str] = None,\n                 constraint: Optional[str] = None,\n                 channel: Channel = LocalChannel(),\n                 nodes_per_block: int = 1,\n                 cores_per_node: Optional[int] = None,\n                 mem_per_node: Optional[int] = None,\n                 init_blocks: int = 1,\n                 min_blocks: int = 0,\n                 max_blocks: int = 1,\n                 parallelism: float = 1,\n                 walltime: str = \"00:10:00\",\n                 scheduler_options: str = '',\n                 regex_job_id: str = r\"Submitted batch job (?P<id>\\S*)\",\n                 worker_init: str = '',\n                 cmd_timeout: int = 10,\n                 exclusive: bool = True,\n                 move_files: bool = True,\n                 launcher: Launcher = SingleNodeLauncher()):\n        label = 'slurm'\n        super().__init__(label,\n                         channel,\n                         nodes_per_block,\n                         init_blocks,\n                         min_blocks,\n                         max_blocks,\n                         parallelism,\n                         walltime,\n                         cmd_timeout=cmd_timeout,\n                         launcher=launcher)\n\n        self.partition = partition\n        self.cores_per_node = cores_per_node\n        self.mem_per_node = mem_per_node\n        self.exclusive = exclusive\n        self.move_files = move_files\n        self.account = account\n        self.qos = qos\n        self.constraint = constraint\n        self.scheduler_options = scheduler_options + '\\n'\n        if exclusive:\n            self.scheduler_options += \"#SBATCH --exclusive\\n\"\n        if partition:\n            self.scheduler_options += \"#SBATCH --partition={}\\n\".format(partition)\n        if account:\n            self.scheduler_options += \"#SBATCH --account={}\\n\".format(account)\n        if qos:\n            self.scheduler_options += \"#SBATCH --qos={}\\n\".format(qos)\n        if constraint:\n            self.scheduler_options += \"#SBATCH --constraint={}\\n\".format(constraint)\n\n        self.regex_job_id = regex_job_id\n        self.worker_init = worker_init + '\\n'\n\n    def _status(self):\n        '''Returns the status list for a list of job_ids\n\n        Args:\n              self\n\n        Returns:\n              [status...] : Status list of all jobs\n        '''\n        job_id_list = ','.join(\n            [jid for jid, job in self.resources.items() if not job['status'].terminal]\n        )\n        if not job_id_list:\n            logger.debug('No active jobs, skipping status update')\n            return\n\n        cmd = \"squeue --noheader --format='%i %t' --job '{0}'\".format(job_id_list)\n        logger.debug(\"Executing %s\", cmd)\n        retcode, stdout, stderr = self.execute_wait(cmd)\n        logger.debug(\"squeue returned %s %s\", stdout, stderr)\n\n        # Execute_wait failed. Do no update\n        if retcode != 0:\n            logger.warning(\"squeue failed with non-zero exit code {}\".format(retcode))\n            return\n\n        jobs_missing = set(self.resources.keys())\n        for line in stdout.split('\\n'):\n            if not line:\n                # Blank line\n                continue\n            job_id, slurm_state = line.split()\n            if slurm_state not in translate_table:\n                logger.warning(f\"Slurm status {slurm_state} is not recognized\")\n            status = translate_table.get(slurm_state, JobState.UNKNOWN)\n            logger.debug(\"Updating job {} with slurm status {} to parsl state {!s}\".format(job_id, slurm_state, status))\n            self.resources[job_id]['status'] = JobStatus(status)\n            jobs_missing.remove(job_id)\n\n        # squeue does not report on jobs that are not running. So we are filling in the\n        # blanks for missing jobs, we might lose some information about why the jobs failed.\n        for missing_job in jobs_missing:\n            logger.debug(\"Updating missing job {} to completed status\".format(missing_job))\n            self.resources[missing_job]['status'] = JobStatus(JobState.COMPLETED)\n\n    def submit(self, command: str, tasks_per_node: int, job_name=\"parsl.slurm\") -> str:\n        \"\"\"Submit the command as a slurm job.\n\n        Parameters\n        ----------\n        command : str\n            Command to be made on the remote side.\n        tasks_per_node : int\n            Command invocations to be launched per node\n        job_name : str\n            Name for the job\n        Returns\n        -------\n        job id : str\n            A string identifier for the job\n        \"\"\"\n\n        scheduler_options = self.scheduler_options\n        worker_init = self.worker_init\n        if self.mem_per_node is not None:\n            scheduler_options += '#SBATCH --mem={}g\\n'.format(self.mem_per_node)\n            worker_init += 'export PARSL_MEMORY_GB={}\\n'.format(self.mem_per_node)\n        if self.cores_per_node is not None:\n            cpus_per_task = math.floor(self.cores_per_node / tasks_per_node)\n            scheduler_options += '#SBATCH --cpus-per-task={}'.format(cpus_per_task)\n            worker_init += 'export PARSL_CORES={}\\n'.format(cpus_per_task)\n\n        job_name = \"{0}.{1}\".format(job_name, time.time())\n\n        script_path = \"{0}/{1}.submit\".format(self.script_dir, job_name)\n        script_path = os.path.abspath(script_path)\n\n        logger.debug(\"Requesting one block with {} nodes\".format(self.nodes_per_block))\n\n        job_config = {}\n        job_config[\"submit_script_dir\"] = self.channel.script_dir\n        job_config[\"nodes\"] = self.nodes_per_block\n        job_config[\"tasks_per_node\"] = tasks_per_node\n        job_config[\"walltime\"] = wtime_to_minutes(self.walltime)\n        job_config[\"scheduler_options\"] = scheduler_options\n        job_config[\"worker_init\"] = worker_init\n        job_config[\"user_script\"] = command\n\n        # Wrap the command\n        job_config[\"user_script\"] = self.launcher(command,\n                                                  tasks_per_node,\n                                                  self.nodes_per_block)\n\n        logger.debug(\"Writing submit script\")\n        self._write_submit_script(template_string, script_path, job_name, job_config)\n\n        if self.move_files:\n            logger.debug(\"moving files\")\n            channel_script_path = self.channel.push_file(script_path, self.channel.script_dir)\n        else:\n            logger.debug(\"not moving files\")\n            channel_script_path = script_path\n\n        retcode, stdout, stderr = self.execute_wait(\"sbatch {0}\".format(channel_script_path))\n\n        if retcode == 0:\n            for line in stdout.split('\\n'):\n                match = re.match(self.regex_job_id, line)\n                if match:\n                    job_id = match.group(\"id\")\n                    self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}\n                    return job_id\n            else:\n                logger.error(\"Could not read job ID from submit command standard output.\")\n                logger.error(\"Retcode:%s STDOUT:%s STDERR:%s\", retcode, stdout.strip(), stderr.strip())\n                raise SubmitException(job_name, \"Could not read job ID from submit command standard output\", stdout=stdout, stderr=stderr, retcode=retcode)\n        else:\n            logger.error(\"Submit command failed\")\n            logger.error(\"Retcode:%s STDOUT:%s STDERR:%s\", retcode, stdout.strip(), stderr.strip())\n            raise SubmitException(job_name, \"Could not read job ID from submit command standard output\", stdout=stdout, stderr=stderr, retcode=retcode)\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\n        job_id_list = ' '.join(job_ids)\n        retcode, stdout, stderr = self.execute_wait(\"scancel {0}\".format(job_id_list))\n        rets = None\n        if retcode == 0:\n            for jid in job_ids:\n                self.resources[jid]['status'] = JobStatus(JobState.CANCELLED)  # Setting state to cancelled\n            rets = [True for i in job_ids]\n        else:\n            rets = [False for i in job_ids]\n\n        return rets\n\n    @property\n    def status_polling_interval(self):\n        return 60\n", "path": "parsl/providers/slurm/slurm.py"}]}