We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Scheduler options not honored in Torque
**Describe the bug**
If a `scheduler_option` that conflicts with a default request in the standard template is made via the `TorqueProvider`, it's currently ignored. It turns out that Torque honors the first line rather than the last in case of a conflicting option. 

For eg:
```
#PBS -l nodes=1:ppn=1
#PBS -l nodes=1:ppn=16:xk   # <--- from scheduler_options, will be ignored.
```


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

Path: `parsl/providers/torque/template.py`
Content:
```
1 template_string = '''#!/bin/bash
2 
3 #PBS -S /bin/bash
4 #PBS -N ${jobname}
5 #PBS -m n
6 #PBS -l walltime=$walltime
7 #PBS -l nodes=${nodes_per_block}:ppn=${tasks_per_node}
8 #PBS -o ${submit_script_dir}/${jobname}.submit.stdout
9 #PBS -e ${submit_script_dir}/${jobname}.submit.stderr
10 ${scheduler_options}
11 
12 ${worker_init}
13 
14 export JOBNAME="${jobname}"
15 
16 ${user_script}
17 
18 '''
19
```
--- 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/template.py b/parsl/providers/torque/template.py
--- a/parsl/providers/torque/template.py
+++ b/parsl/providers/torque/template.py
@@ -1,5 +1,6 @@
 template_string = '''#!/bin/bash
 
+${scheduler_options}
 #PBS -S /bin/bash
 #PBS -N ${jobname}
 #PBS -m n
@@ -7,7 +8,6 @@
 #PBS -l nodes=${nodes_per_block}:ppn=${tasks_per_node}
 #PBS -o ${submit_script_dir}/${jobname}.submit.stdout
 #PBS -e ${submit_script_dir}/${jobname}.submit.stderr
-${scheduler_options}
 
 ${worker_init}
 
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
@@ -58,6 +58,7 @@
         Walltime requested per block in HH:MM:SS.
     scheduler_options : str
         String to prepend to the #PBS blocks in the submit script to the scheduler.
+        WARNING: scheduler_options should only be given #PBS strings, and should not have trailing newlines.
     worker_init : str
         Command to be run before starting a worker, such as 'module load Anaconda; source activate env'.
     launcher : Launcher

{"golden_diff": "diff --git a/parsl/providers/torque/template.py b/parsl/providers/torque/template.py\n--- a/parsl/providers/torque/template.py\n+++ b/parsl/providers/torque/template.py\n@@ -1,5 +1,6 @@\n template_string = '''#!/bin/bash\n \n+${scheduler_options}\n #PBS -S /bin/bash\n #PBS -N ${jobname}\n #PBS -m n\n@@ -7,7 +8,6 @@\n #PBS -l nodes=${nodes_per_block}:ppn=${tasks_per_node}\n #PBS -o ${submit_script_dir}/${jobname}.submit.stdout\n #PBS -e ${submit_script_dir}/${jobname}.submit.stderr\n-${scheduler_options}\n \n ${worker_init}\n \ndiff --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@@ -58,6 +58,7 @@\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+        WARNING: scheduler_options should only be given #PBS strings, and should not have trailing newlines.\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", "issue": "Scheduler options not honored in Torque\n**Describe the bug**\r\nIf a `scheduler_option` that conflicts with a default request in the standard template is made via the `TorqueProvider`, it's currently ignored. It turns out that Torque honors the first line rather than the last in case of a conflicting option. \r\n\r\nFor eg:\r\n```\r\n#PBS -l nodes=1:ppn=1\r\n#PBS -l nodes=1:ppn=16:xk   # <--- from scheduler_options, will be ignored.\r\n```\r\n\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    '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=1,\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"}, {"content": "template_string = '''#!/bin/bash\n\n#PBS -S /bin/bash\n#PBS -N ${jobname}\n#PBS -m n\n#PBS -l walltime=$walltime\n#PBS -l nodes=${nodes_per_block}:ppn=${tasks_per_node}\n#PBS -o ${submit_script_dir}/${jobname}.submit.stdout\n#PBS -e ${submit_script_dir}/${jobname}.submit.stderr\n${scheduler_options}\n\n${worker_init}\n\nexport JOBNAME=\"${jobname}\"\n\n${user_script}\n\n'''\n", "path": "parsl/providers/torque/template.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        WARNING: scheduler_options should only be given #PBS strings, and should not have trailing newlines.\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=1,\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"}, {"content": "template_string = '''#!/bin/bash\n\n${scheduler_options}\n#PBS -S /bin/bash\n#PBS -N ${jobname}\n#PBS -m n\n#PBS -l walltime=$walltime\n#PBS -l nodes=${nodes_per_block}:ppn=${tasks_per_node}\n#PBS -o ${submit_script_dir}/${jobname}.submit.stdout\n#PBS -e ${submit_script_dir}/${jobname}.submit.stderr\n\n${worker_init}\n\nexport JOBNAME=\"${jobname}\"\n\n${user_script}\n\n'''\n", "path": "parsl/providers/torque/template.py"}]}