We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[PBSPro] max_blocks limit is not obeyed when submitting as array-jobs 
**Describe the bug**
Parsl keeps launching new blocks beyond the `max_blocks` limit when array-jobs mode is enabled using Parsl config `scheduler_options` parameter (e.g. #PBS -J 1-10).

**To Reproduce**
Enable Parsl monitoring and checkpointing
Enable HighThroughputExecutor and PBSProProvider
Add `#PBS -J 1-10` option to `scheduler_options`
Set `max_blocks` limit to 3

**Expected behavior**
No more than 3 blocks should be launched.

**Actual behavior**
Parsl keeps on launching new blocks. Following polling log can be seen
`2020-04-23 14:41:33.575 parsl.dataflow.strategy:205 [DEBUG]  Executor htex_array_jobs has 2739 active tasks, 0/4 running/pending blocks, and 60 connected workers`

This means Parsl does not consider partially activated blocks (array-jobs with some jobs Running others in Queue status out of total array jobs in that block) when making a launch decision.
It seems that conditional check is done here [1] but I couldn't find a place where the `provisioned_blocks` variable is updated. Could you shed some light on how this is updated?

**Environment**
- OS: RHEL6.1
- Python version: 3.7 (Anaconda 4.8.2)
- Parsl version: master branch commit: a30ce173cf8593a34b81d5a9cdd646dcf63fa798

**Distributed Environment**
- PBS Pro in NSCC's ASPIRE1

[1] https://github.com/Parsl/parsl/blob/master/parsl/providers/pbspro/pbspro.py#L109

--- 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/torque/torque.py`
Content:
```
1 import logging
2 import os
3 import time
4 
5 from parsl.channels import LocalChannel
6 from parsl.launchers import AprunLauncher
7 from parsl.providers.provider_base import JobState, JobStatus
8 from parsl.providers.torque.template import template_string
9 from parsl.providers.cluster_provider import ClusterProvider
10 from parsl.utils import RepresentationMixin
11 
12 logger = logging.getLogger(__name__)
13 
14 # From the man pages for qstat for PBS/Torque systems
15 translate_table = {
16     'R': JobState.RUNNING,
17     'C': JobState.COMPLETED,  # Completed after having run
18     'E': JobState.COMPLETED,  # Exiting after having run
19     'H': JobState.HELD,  # Held
20     'Q': JobState.PENDING,  # Queued, and eligible to run
21     'W': JobState.PENDING,  # Job is waiting for it's execution time (-a option) to be reached
22     'S': JobState.HELD
23 }  # Suspended
24 
25 
26 class TorqueProvider(ClusterProvider, RepresentationMixin):
27     """Torque Execution Provider
28 
29     This provider uses sbatch to submit, squeue for status, and scancel to cancel
30     jobs. The sbatch script to be used is created from a template file in this
31     same module.
32 
33     Parameters
34     ----------
35     channel : Channel
36         Channel for accessing this provider. Possible channels include
37         :class:`~parsl.channels.LocalChannel` (the default),
38         :class:`~parsl.channels.SSHChannel`, or
39         :class:`~parsl.channels.SSHInteractiveLoginChannel`.
40     account : str
41         Account the job will be charged against.
42     queue : str
43         Torque queue to request blocks from.
44     nodes_per_block : int
45         Nodes to provision per block.
46     init_blocks : int
47         Number of blocks to provision at the start of the run. Default is 1.
48     min_blocks : int
49         Minimum number of blocks to maintain. Default is 0.
50     max_blocks : int
51         Maximum number of blocks to maintain.
52     parallelism : float
53         Ratio of provisioned task slots to active tasks. A parallelism value of 1 represents aggressive
54         scaling where as many resources as possible are used; parallelism close to 0 represents
55         the opposite situation in which as few resources as possible (i.e., min_blocks) are used.
56     walltime : str
57         Walltime requested per block in HH:MM:SS.
58     scheduler_options : str
59         String to prepend to the #PBS blocks in the submit script to the scheduler.
60     worker_init : str
61         Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.
62     launcher : Launcher
63         Launcher for this provider. Possible launchers include
64         :class:`~parsl.launchers.AprunLauncher` (the default), or
65         :class:`~parsl.launchers.SingleNodeLauncher`,
66 
67     """
68     def __init__(self,
69                  channel=LocalChannel(),
70                  account=None,
71                  queue=None,
72                  scheduler_options='',
73                  worker_init='',
74                  nodes_per_block=1,
75                  init_blocks=1,
76                  min_blocks=0,
77                  max_blocks=100,
78                  parallelism=1,
79                  launcher=AprunLauncher(),
80                  walltime="00:20:00",
81                  cmd_timeout=120):
82         label = 'torque'
83         super().__init__(label,
84                          channel,
85                          nodes_per_block,
86                          init_blocks,
87                          min_blocks,
88                          max_blocks,
89                          parallelism,
90                          walltime,
91                          launcher,
92                          cmd_timeout=cmd_timeout)
93 
94         self.account = account
95         self.queue = queue
96         self.scheduler_options = scheduler_options
97         self.worker_init = worker_init
98         self.provisioned_blocks = 0
99         self.template_string = template_string
100 
101         # Dictionary that keeps track of jobs, keyed on job_id
102         self.resources = {}
103 
104     def _status(self):
105         ''' Internal: Do not call. Returns the status list for a list of job_ids
106 
107         Args:
108               self
109 
110         Returns:
111               [status...] : Status list of all jobs
112         '''
113 
114         job_ids = list(self.resources.keys())
115         job_id_list = ' '.join(self.resources.keys())
116 
117         jobs_missing = list(self.resources.keys())
118 
119         retcode, stdout, stderr = self.execute_wait("qstat {0}".format(job_id_list))
120         for line in stdout.split('\n'):
121             parts = line.split()
122             if not parts or parts[0].upper().startswith('JOB') or parts[0].startswith('---'):
123                 continue
124             job_id = parts[0]  # likely truncated
125             for long_job_id in job_ids:
126                 if long_job_id.startswith(job_id):
127                     logger.debug('coerced job_id %s -> %s', job_id, long_job_id)
128                     job_id = long_job_id
129                     break
130             state = translate_table.get(parts[4], JobState.UNKNOWN)
131             self.resources[job_id]['status'] = JobStatus(state)
132             jobs_missing.remove(job_id)
133 
134         # squeue does not report on jobs that are not running. So we are filling in the
135         # blanks for missing jobs, we might lose some information about why the jobs failed.
136         for missing_job in jobs_missing:
137             self.resources[missing_job]['status'] = JobStatus(JobState.COMPLETED)
138 
139     def submit(self, command, tasks_per_node, job_name="parsl.torque"):
140         ''' Submits the command onto an Local Resource Manager job.
141         Submit returns an ID that corresponds to the task that was just submitted.
142 
143         If tasks_per_node <  1 : ! This is illegal. tasks_per_node should be integer
144 
145         If tasks_per_node == 1:
146              A single node is provisioned
147 
148         If tasks_per_node >  1 :
149              tasks_per_node number of nodes are provisioned.
150 
151         Args:
152              - command  :(String) Commandline invocation to be made on the remote side.
153              - tasks_per_node (int) : command invocations to be launched per node
154 
155         Kwargs:
156              - job_name (String): Name for job, must be unique
157 
158         Returns:
159              - None: At capacity, cannot provision more
160              - job_id: (string) Identifier for the job
161 
162         '''
163 
164         if self.provisioned_blocks >= self.max_blocks:
165             logger.warning("[%s] at capacity, cannot add more blocks now", self.label)
166             return None
167 
168         # Set job name
169         job_name = "parsl.{0}.{1}".format(job_name, time.time())
170 
171         # Set script path
172         script_path = "{0}/{1}.submit".format(self.script_dir, job_name)
173         script_path = os.path.abspath(script_path)
174 
175         logger.debug("Requesting nodes_per_block:%s tasks_per_node:%s", self.nodes_per_block,
176                      tasks_per_node)
177 
178         job_config = {}
179         # TODO : script_path might need to change to accommodate script dir set via channels
180         job_config["submit_script_dir"] = self.channel.script_dir
181         job_config["nodes"] = self.nodes_per_block
182         job_config["task_blocks"] = self.nodes_per_block * tasks_per_node
183         job_config["nodes_per_block"] = self.nodes_per_block
184         job_config["tasks_per_node"] = tasks_per_node
185         job_config["walltime"] = self.walltime
186         job_config["scheduler_options"] = self.scheduler_options
187         job_config["worker_init"] = self.worker_init
188         job_config["user_script"] = command
189 
190         # Wrap the command
191         job_config["user_script"] = self.launcher(command,
192                                                   tasks_per_node,
193                                                   self.nodes_per_block)
194 
195         logger.debug("Writing submit script")
196         self._write_submit_script(self.template_string, script_path, job_name, job_config)
197 
198         channel_script_path = self.channel.push_file(script_path, self.channel.script_dir)
199 
200         submit_options = ''
201         if self.queue is not None:
202             submit_options = '{0} -q {1}'.format(submit_options, self.queue)
203         if self.account is not None:
204             submit_options = '{0} -A {1}'.format(submit_options, self.account)
205 
206         launch_cmd = "qsub {0} {1}".format(submit_options, channel_script_path)
207         retcode, stdout, stderr = self.execute_wait(launch_cmd)
208 
209         job_id = None
210         if retcode == 0:
211             for line in stdout.split('\n'):
212                 if line.strip():
213                     job_id = line.strip()
214                     self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}
215         else:
216             message = "Command '{}' failed with return code {}".format(launch_cmd, retcode)
217             if (stdout is not None) and (stderr is not None):
218                 message += "\nstderr:{}\nstdout{}".format(stderr.strip(), stdout.strip())
219             logger.error(message)
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 
233         job_id_list = ' '.join(job_ids)
234         retcode, stdout, stderr = self.execute_wait("qdel {0}".format(job_id_list))
235         rets = None
236         if retcode == 0:
237             for jid in job_ids:
238                 self.resources[jid]['status'] = JobStatus(JobState.COMPLETED)  # Setting state to exiting
239             rets = [True for i in job_ids]
240         else:
241             rets = [False for i in job_ids]
242 
243         return rets
244 
245     @property
246     def status_polling_interval(self):
247         return 60
248 
249 
250 if __name__ == "__main__":
251 
252     print("None")
253
```
--- 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/torque/torque.py b/parsl/providers/torque/torque.py
--- a/parsl/providers/torque/torque.py
+++ b/parsl/providers/torque/torque.py
@@ -13,6 +13,7 @@
 
 # From the man pages for qstat for PBS/Torque systems
 translate_table = {
+    'B': JobState.RUNNING,  # This state is returned for running array jobs
     'R': JobState.RUNNING,
     'C': JobState.COMPLETED,  # Completed after having run
     'E': JobState.COMPLETED,  # Exiting after having run

{"golden_diff": "diff --git a/parsl/providers/torque/torque.py b/parsl/providers/torque/torque.py\n--- a/parsl/providers/torque/torque.py\n+++ b/parsl/providers/torque/torque.py\n@@ -13,6 +13,7 @@\n \n # From the man pages for qstat for PBS/Torque systems\n translate_table = {\n+    'B': JobState.RUNNING,  # This state is returned for running array jobs\n     'R': JobState.RUNNING,\n     'C': JobState.COMPLETED,  # Completed after having run\n     'E': JobState.COMPLETED,  # Exiting after having run\n", "issue": "[PBSPro] max_blocks limit is not obeyed when submitting as array-jobs \n**Describe the bug**\r\nParsl keeps launching new blocks beyond the `max_blocks` limit when array-jobs mode is enabled using Parsl config `scheduler_options` parameter (e.g. #PBS -J 1-10).\r\n\r\n**To Reproduce**\r\nEnable Parsl monitoring and checkpointing\r\nEnable HighThroughputExecutor and PBSProProvider\r\nAdd `#PBS -J 1-10` option to `scheduler_options`\r\nSet `max_blocks` limit to 3\r\n\r\n**Expected behavior**\r\nNo more than 3 blocks should be launched.\r\n\r\n**Actual behavior**\r\nParsl keeps on launching new blocks. Following polling log can be seen\r\n`2020-04-23 14:41:33.575 parsl.dataflow.strategy:205 [DEBUG]  Executor htex_array_jobs has 2739 active tasks, 0/4 running/pending blocks, and 60 connected workers`\r\n\r\nThis means Parsl does not consider partially activated blocks (array-jobs with some jobs Running others in Queue status out of total array jobs in that block) when making a launch decision.\r\nIt seems that conditional check is done here [1] but I couldn't find a place where the `provisioned_blocks` variable is updated. Could you shed some light on how this is updated?\r\n\r\n**Environment**\r\n- OS: RHEL6.1\r\n- Python version: 3.7 (Anaconda 4.8.2)\r\n- Parsl version: master branch commit: a30ce173cf8593a34b81d5a9cdd646dcf63fa798\r\n\r\n**Distributed Environment**\r\n- PBS Pro in NSCC's ASPIRE1\r\n\r\n[1] https://github.com/Parsl/parsl/blob/master/parsl/providers/pbspro/pbspro.py#L109\n", "before_files": [{"content": "import logging\nimport os\nimport time\n\nfrom parsl.channels import LocalChannel\nfrom parsl.launchers import AprunLauncher\nfrom parsl.providers.provider_base import JobState, JobStatus\nfrom parsl.providers.torque.template import template_string\nfrom parsl.providers.cluster_provider import ClusterProvider\nfrom parsl.utils import RepresentationMixin\n\nlogger = logging.getLogger(__name__)\n\n# From the man pages for qstat for PBS/Torque systems\ntranslate_table = {\n    'R': JobState.RUNNING,\n    'C': JobState.COMPLETED,  # Completed after having run\n    'E': JobState.COMPLETED,  # Exiting after having run\n    'H': JobState.HELD,  # Held\n    'Q': JobState.PENDING,  # Queued, and eligible to run\n    'W': JobState.PENDING,  # Job is waiting for it's execution time (-a option) to be reached\n    'S': JobState.HELD\n}  # Suspended\n\n\nclass TorqueProvider(ClusterProvider, RepresentationMixin):\n    \"\"\"Torque 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    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    account : str\n        Account the job will be charged against.\n    queue : str\n        Torque queue to request blocks from.\n    nodes_per_block : int\n        Nodes to provision per block.\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. Default is 0.\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 #PBS 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    launcher : Launcher\n        Launcher for this provider. Possible launchers include\n        :class:`~parsl.launchers.AprunLauncher` (the default), or\n        :class:`~parsl.launchers.SingleNodeLauncher`,\n\n    \"\"\"\n    def __init__(self,\n                 channel=LocalChannel(),\n                 account=None,\n                 queue=None,\n                 scheduler_options='',\n                 worker_init='',\n                 nodes_per_block=1,\n                 init_blocks=1,\n                 min_blocks=0,\n                 max_blocks=100,\n                 parallelism=1,\n                 launcher=AprunLauncher(),\n                 walltime=\"00:20:00\",\n                 cmd_timeout=120):\n        label = 'torque'\n        super().__init__(label,\n                         channel,\n                         nodes_per_block,\n                         init_blocks,\n                         min_blocks,\n                         max_blocks,\n                         parallelism,\n                         walltime,\n                         launcher,\n                         cmd_timeout=cmd_timeout)\n\n        self.account = account\n        self.queue = queue\n        self.scheduler_options = scheduler_options\n        self.worker_init = worker_init\n        self.provisioned_blocks = 0\n        self.template_string = template_string\n\n        # Dictionary that keeps track of jobs, keyed on job_id\n        self.resources = {}\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\n        job_ids = list(self.resources.keys())\n        job_id_list = ' '.join(self.resources.keys())\n\n        jobs_missing = list(self.resources.keys())\n\n        retcode, stdout, stderr = self.execute_wait(\"qstat {0}\".format(job_id_list))\n        for line in stdout.split('\\n'):\n            parts = line.split()\n            if not parts or parts[0].upper().startswith('JOB') or parts[0].startswith('---'):\n                continue\n            job_id = parts[0]  # likely truncated\n            for long_job_id in job_ids:\n                if long_job_id.startswith(job_id):\n                    logger.debug('coerced job_id %s -> %s', job_id, long_job_id)\n                    job_id = long_job_id\n                    break\n            state = translate_table.get(parts[4], JobState.UNKNOWN)\n            self.resources[job_id]['status'] = JobStatus(state)\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            self.resources[missing_job]['status'] = JobStatus(JobState.COMPLETED)\n\n    def submit(self, command, tasks_per_node, job_name=\"parsl.torque\"):\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 : ! This is illegal. tasks_per_node should be integer\n\n        If tasks_per_node == 1:\n             A single node is provisioned\n\n        If tasks_per_node >  1 :\n             tasks_per_node number of 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        if self.provisioned_blocks >= self.max_blocks:\n            logger.warning(\"[%s] at capacity, cannot add more blocks now\", self.label)\n            return None\n\n        # Set job name\n        job_name = \"parsl.{0}.{1}\".format(job_name, time.time())\n\n        # Set script path\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 nodes_per_block:%s tasks_per_node:%s\", self.nodes_per_block,\n                     tasks_per_node)\n\n        job_config = {}\n        # TODO : script_path might need to change to accommodate script dir set via channels\n        job_config[\"submit_script_dir\"] = self.channel.script_dir\n        job_config[\"nodes\"] = self.nodes_per_block\n        job_config[\"task_blocks\"] = self.nodes_per_block * tasks_per_node\n        job_config[\"nodes_per_block\"] = self.nodes_per_block\n        job_config[\"tasks_per_node\"] = tasks_per_node\n        job_config[\"walltime\"] = self.walltime\n        job_config[\"scheduler_options\"] = self.scheduler_options\n        job_config[\"worker_init\"] = self.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(self.template_string, script_path, job_name, job_config)\n\n        channel_script_path = self.channel.push_file(script_path, self.channel.script_dir)\n\n        submit_options = ''\n        if self.queue is not None:\n            submit_options = '{0} -q {1}'.format(submit_options, self.queue)\n        if self.account is not None:\n            submit_options = '{0} -A {1}'.format(submit_options, self.account)\n\n        launch_cmd = \"qsub {0} {1}\".format(submit_options, channel_script_path)\n        retcode, stdout, stderr = self.execute_wait(launch_cmd)\n\n        job_id = None\n        if retcode == 0:\n            for line in stdout.split('\\n'):\n                if line.strip():\n                    job_id = line.strip()\n                    self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}\n        else:\n            message = \"Command '{}' failed with return code {}\".format(launch_cmd, retcode)\n            if (stdout is not None) and (stderr is not None):\n                message += \"\\nstderr:{}\\nstdout{}\".format(stderr.strip(), stdout.strip())\n            logger.error(message)\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\n        job_id_list = ' '.join(job_ids)\n        retcode, stdout, stderr = self.execute_wait(\"qdel {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.COMPLETED)  # Setting state to exiting\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\n\nif __name__ == \"__main__\":\n\n    print(\"None\")\n", "path": "parsl/providers/torque/torque.py"}], "after_files": [{"content": "import logging\nimport os\nimport time\n\nfrom parsl.channels import LocalChannel\nfrom parsl.launchers import AprunLauncher\nfrom parsl.providers.provider_base import JobState, JobStatus\nfrom parsl.providers.torque.template import template_string\nfrom parsl.providers.cluster_provider import ClusterProvider\nfrom parsl.utils import RepresentationMixin\n\nlogger = logging.getLogger(__name__)\n\n# From the man pages for qstat for PBS/Torque systems\ntranslate_table = {\n    'B': JobState.RUNNING,  # This state is returned for running array jobs\n    'R': JobState.RUNNING,\n    'C': JobState.COMPLETED,  # Completed after having run\n    'E': JobState.COMPLETED,  # Exiting after having run\n    'H': JobState.HELD,  # Held\n    'Q': JobState.PENDING,  # Queued, and eligible to run\n    'W': JobState.PENDING,  # Job is waiting for it's execution time (-a option) to be reached\n    'S': JobState.HELD\n}  # Suspended\n\n\nclass TorqueProvider(ClusterProvider, RepresentationMixin):\n    \"\"\"Torque 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    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    account : str\n        Account the job will be charged against.\n    queue : str\n        Torque queue to request blocks from.\n    nodes_per_block : int\n        Nodes to provision per block.\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. Default is 0.\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 #PBS 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    launcher : Launcher\n        Launcher for this provider. Possible launchers include\n        :class:`~parsl.launchers.AprunLauncher` (the default), or\n        :class:`~parsl.launchers.SingleNodeLauncher`,\n\n    \"\"\"\n    def __init__(self,\n                 channel=LocalChannel(),\n                 account=None,\n                 queue=None,\n                 scheduler_options='',\n                 worker_init='',\n                 nodes_per_block=1,\n                 init_blocks=1,\n                 min_blocks=0,\n                 max_blocks=100,\n                 parallelism=1,\n                 launcher=AprunLauncher(),\n                 walltime=\"00:20:00\",\n                 cmd_timeout=120):\n        label = 'torque'\n        super().__init__(label,\n                         channel,\n                         nodes_per_block,\n                         init_blocks,\n                         min_blocks,\n                         max_blocks,\n                         parallelism,\n                         walltime,\n                         launcher,\n                         cmd_timeout=cmd_timeout)\n\n        self.account = account\n        self.queue = queue\n        self.scheduler_options = scheduler_options\n        self.worker_init = worker_init\n        self.provisioned_blocks = 0\n        self.template_string = template_string\n\n        # Dictionary that keeps track of jobs, keyed on job_id\n        self.resources = {}\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\n        job_ids = list(self.resources.keys())\n        job_id_list = ' '.join(self.resources.keys())\n\n        jobs_missing = list(self.resources.keys())\n\n        retcode, stdout, stderr = self.execute_wait(\"qstat {0}\".format(job_id_list))\n        for line in stdout.split('\\n'):\n            parts = line.split()\n            if not parts or parts[0].upper().startswith('JOB') or parts[0].startswith('---'):\n                continue\n            job_id = parts[0]  # likely truncated\n            for long_job_id in job_ids:\n                if long_job_id.startswith(job_id):\n                    logger.debug('coerced job_id %s -> %s', job_id, long_job_id)\n                    job_id = long_job_id\n                    break\n            state = translate_table.get(parts[4], JobState.UNKNOWN)\n            self.resources[job_id]['status'] = JobStatus(state)\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            self.resources[missing_job]['status'] = JobStatus(JobState.COMPLETED)\n\n    def submit(self, command, tasks_per_node, job_name=\"parsl.torque\"):\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 : ! This is illegal. tasks_per_node should be integer\n\n        If tasks_per_node == 1:\n             A single node is provisioned\n\n        If tasks_per_node >  1 :\n             tasks_per_node number of 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        if self.provisioned_blocks >= self.max_blocks:\n            logger.warning(\"[%s] at capacity, cannot add more blocks now\", self.label)\n            return None\n\n        # Set job name\n        job_name = \"parsl.{0}.{1}\".format(job_name, time.time())\n\n        # Set script path\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 nodes_per_block:%s tasks_per_node:%s\", self.nodes_per_block,\n                     tasks_per_node)\n\n        job_config = {}\n        # TODO : script_path might need to change to accommodate script dir set via channels\n        job_config[\"submit_script_dir\"] = self.channel.script_dir\n        job_config[\"nodes\"] = self.nodes_per_block\n        job_config[\"task_blocks\"] = self.nodes_per_block * tasks_per_node\n        job_config[\"nodes_per_block\"] = self.nodes_per_block\n        job_config[\"tasks_per_node\"] = tasks_per_node\n        job_config[\"walltime\"] = self.walltime\n        job_config[\"scheduler_options\"] = self.scheduler_options\n        job_config[\"worker_init\"] = self.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(self.template_string, script_path, job_name, job_config)\n\n        channel_script_path = self.channel.push_file(script_path, self.channel.script_dir)\n\n        submit_options = ''\n        if self.queue is not None:\n            submit_options = '{0} -q {1}'.format(submit_options, self.queue)\n        if self.account is not None:\n            submit_options = '{0} -A {1}'.format(submit_options, self.account)\n\n        launch_cmd = \"qsub {0} {1}\".format(submit_options, channel_script_path)\n        retcode, stdout, stderr = self.execute_wait(launch_cmd)\n\n        job_id = None\n        if retcode == 0:\n            for line in stdout.split('\\n'):\n                if line.strip():\n                    job_id = line.strip()\n                    self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}\n        else:\n            message = \"Command '{}' failed with return code {}\".format(launch_cmd, retcode)\n            if (stdout is not None) and (stderr is not None):\n                message += \"\\nstderr:{}\\nstdout{}\".format(stderr.strip(), stdout.strip())\n            logger.error(message)\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\n        job_id_list = ' '.join(job_ids)\n        retcode, stdout, stderr = self.execute_wait(\"qdel {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.COMPLETED)  # Setting state to exiting\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\n\nif __name__ == \"__main__\":\n\n    print(\"None\")\n", "path": "parsl/providers/torque/torque.py"}]}