We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add account option to slurm provider
See Parsl/libsubmit#68

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

{"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@@ -43,6 +43,9 @@\n     ----------\n     partition : str\n         Slurm partition to request blocks from. If none, no partition slurm directive will be specified.\n+    account : str\n+        Slurm account to which to charge resources used by the job. If none, the job will use the\n+        user's default account.\n     channel : Channel\n         Channel for accessing this provider. Possible channels include\n         :class:`~parsl.channels.LocalChannel` (the default),\n@@ -77,12 +80,13 @@\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+    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],\n+                 account: Optional[str] = None,\n                  channel: Channel = LocalChannel(),\n                  nodes_per_block: int = 1,\n                  cores_per_node: Optional[int] = None,\n@@ -115,11 +119,14 @@\n         self.mem_per_node = mem_per_node\n         self.exclusive = exclusive\n         self.move_files = move_files\n+        self.account = account\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         self.worker_init = worker_init + '\\n'\n \n     def _status(self):\n", "issue": "Add account option to slurm provider\nSee Parsl/libsubmit#68\n", "before_files": [{"content": "import os\nimport math\nimport time\nimport logging\nimport typeguard\n\nfrom typing import Optional\n\nfrom parsl.channels import LocalChannel\nfrom parsl.channels.base import Channel\nfrom parsl.launchers import SingleNodeLauncher\nfrom parsl.launchers.launchers import Launcher\nfrom parsl.providers.cluster_provider import ClusterProvider\nfrom parsl.providers.provider_base import JobState, JobStatus\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\n}  # (special exit state\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 none, no partition slurm directive will be specified.\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    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],\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                 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.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        self.worker_init = worker_init + '\\n'\n\n    def _status(self):\n        ''' Internal: Do not call. 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(self.resources.keys())\n        cmd = \"squeue --job {0}\".format(job_id_list)\n        logger.debug(\"Executing sqeueue\")\n        retcode, stdout, stderr = self.execute_wait(cmd)\n        logger.debug(\"sqeueue returned\")\n\n        # Execute_wait failed. Do no update\n        if retcode != 0:\n            logger.warning(\"squeue failed with non-zero exit code {} - see https://github.com/Parsl/parsl/issues/1588\".format(retcode))\n            return\n\n        jobs_missing = list(self.resources.keys())\n        for line in stdout.split('\\n'):\n            parts = line.split()\n            if parts and parts[0] != 'JOBID':\n                job_id = parts[0]\n                status = translate_table.get(parts[4], JobState.UNKNOWN)\n                logger.debug(\"Updating job {} with slurm status {} to parsl status {}\".format(job_id, parts[4], 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, tasks_per_node, job_name=\"parsl.slurm\"):\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 (must be unique).\n        Returns\n        -------\n        None or str\n            If at capacity, returns None; otherwise, 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        job_id = None\n        if retcode == 0:\n            for line in stdout.split('\\n'):\n                if line.startswith(\"Submitted batch job\"):\n                    job_id = line.split(\"Submitted batch job\")[1].strip()\n                    self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}\n        else:\n            print(\"Submission of command to scale_out failed\")\n            logger.error(\"Retcode:%s STDOUT:%s STDERR:%s\", retcode, stdout.strip(), stderr.strip())\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\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 typeguard\n\nfrom typing import Optional\n\nfrom parsl.channels import LocalChannel\nfrom parsl.channels.base import Channel\nfrom parsl.launchers import SingleNodeLauncher\nfrom parsl.launchers.launchers import Launcher\nfrom parsl.providers.cluster_provider import ClusterProvider\nfrom parsl.providers.provider_base import JobState, JobStatus\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\n}  # (special exit state\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 none, no partition slurm directive will be specified.\n    account : str\n        Slurm account to which to charge resources used by the job. If none, the job will use the\n        user's default account.\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    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],\n                 account: 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                 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.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        self.worker_init = worker_init + '\\n'\n\n    def _status(self):\n        ''' Internal: Do not call. 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(self.resources.keys())\n        cmd = \"squeue --job {0}\".format(job_id_list)\n        logger.debug(\"Executing sqeueue\")\n        retcode, stdout, stderr = self.execute_wait(cmd)\n        logger.debug(\"sqeueue returned\")\n\n        # Execute_wait failed. Do no update\n        if retcode != 0:\n            logger.warning(\"squeue failed with non-zero exit code {} - see https://github.com/Parsl/parsl/issues/1588\".format(retcode))\n            return\n\n        jobs_missing = list(self.resources.keys())\n        for line in stdout.split('\\n'):\n            parts = line.split()\n            if parts and parts[0] != 'JOBID':\n                job_id = parts[0]\n                status = translate_table.get(parts[4], JobState.UNKNOWN)\n                logger.debug(\"Updating job {} with slurm status {} to parsl status {}\".format(job_id, parts[4], 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, tasks_per_node, job_name=\"parsl.slurm\"):\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 (must be unique).\n        Returns\n        -------\n        None or str\n            If at capacity, returns None; otherwise, 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        job_id = None\n        if retcode == 0:\n            for line in stdout.split('\\n'):\n                if line.startswith(\"Submitted batch job\"):\n                    job_id = line.split(\"Submitted batch job\")[1].strip()\n                    self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}\n        else:\n            print(\"Submission of command to scale_out failed\")\n            logger.error(\"Retcode:%s STDOUT:%s STDERR:%s\", retcode, stdout.strip(), stderr.strip())\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\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"}]}