We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add missing abstract method in LSF batchsystem

Add missing abstract method in LSF batchsystem


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/toil/batchSystems/lsf.py`
Content:
```
1 #Copyright (C) 2013 by Thomas Keane (tk2@sanger.ac.uk)
2 #
3 #Permission is hereby granted, free of charge, to any person obtaining a copy
4 #of this software and associated documentation files (the "Software"), to deal
5 #in the Software without restriction, including without limitation the rights
6 #to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
7 #copies of the Software, and to permit persons to whom the Software is
8 #furnished to do so, subject to the following conditions:
9 #
10 #The above copyright notice and this permission notice shall be included in
11 #all copies or substantial portions of the Software.
12 #
13 #THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14 #IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15 #FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16 #AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17 #LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
18 #OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
19 #THE SOFTWARE.
20 from __future__ import absolute_import
21 import logging
22 import subprocess
23 import time
24 from threading import Thread
25 from datetime import date
26 
27 # Python 3 compatibility imports
28 from six.moves.queue import Empty, Queue
29 
30 from toil.batchSystems import MemoryString
31 from toil.batchSystems.abstractBatchSystem import BatchSystemSupport
32 
33 logger = logging.getLogger( __name__ )
34 
35 
36 
37 def prepareBsub(cpu, mem):
38     mem = '' if mem is None else '-R "select[type==X86_64 && mem > ' + str(int(mem/ 1000000)) + '] rusage[mem=' + str(int(mem/ 1000000)) + ']" -M' + str(int(mem/ 1000000)) + '000'
39     cpu = '' if cpu is None else '-n ' + str(int(cpu))
40     bsubline = ["bsub", mem, cpu,"-cwd", ".", "-o", "/dev/null", "-e", "/dev/null"]
41     return bsubline
42 
43 def bsub(bsubline):
44     process = subprocess.Popen(" ".join(bsubline), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)
45     liney = process.stdout.readline()
46     logger.debug("BSUB: " + liney)
47     result = int(liney.strip().split()[1].strip('<>'))
48     logger.debug("Got the job id: %s" % (str(result)))
49     return result
50 
51 def getjobexitcode(lsfJobID):
52         job, task = lsfJobID
53 
54         #first try bjobs to find out job state
55         args = ["bjobs", "-l", str(job)]
56         logger.debug("Checking job exit code for job via bjobs: " + str(job))
57         process = subprocess.Popen(" ".join(args), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)
58         started = 0
59         for line in process.stdout:
60             if line.find("Done successfully") > -1:
61                 logger.debug("bjobs detected job completed for job: " + str(job))
62                 return 0
63             elif line.find("Completed <exit>") > -1:
64                 logger.debug("bjobs detected job failed for job: " + str(job))
65                 return 1
66             elif line.find("New job is waiting for scheduling") > -1:
67                 logger.debug("bjobs detected job pending scheduling for job: " + str(job))
68                 return None
69             elif line.find("PENDING REASONS") > -1:
70                 logger.debug("bjobs detected job pending for job: " + str(job))
71                 return None
72             elif line.find("Started on ") > -1:
73                 started = 1
74 
75         if started == 1:
76             logger.debug("bjobs detected job started but not completed: " + str(job))
77             return None
78 
79         #if not found in bjobs, then try bacct (slower than bjobs)
80         logger.debug("bjobs failed to detect job - trying bacct: " + str(job))
81 
82         args = ["bacct", "-l", str(job)]
83         logger.debug("Checking job exit code for job via bacct:" + str(job))
84         process = subprocess.Popen(" ".join(args), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)
85         for line in process.stdout:
86             if line.find("Completed <done>") > -1:
87                 logger.debug("Detected job completed for job: " + str(job))
88                 return 0
89             elif line.find("Completed <exit>") > -1:
90                 logger.debug("Detected job failed for job: " + str(job))
91                 return 1
92         logger.debug("Cant determine exit code for job or job still running: " + str(job))
93         return None
94 
95 class Worker(Thread):
96     def __init__(self, newJobsQueue, updatedJobsQueue, boss):
97         Thread.__init__(self)
98         self.newJobsQueue = newJobsQueue
99         self.updatedJobsQueue = updatedJobsQueue
100         self.currentjobs = list()
101         self.runningjobs = set()
102         self.boss = boss
103 
104     def run(self):
105         while True:
106             # Load new job ids:
107             while not self.newJobsQueue.empty():
108                 self.currentjobs.append(self.newJobsQueue.get())
109 
110             # Launch jobs as necessary:
111             while len(self.currentjobs) > 0:
112                 jobID, bsubline = self.currentjobs.pop()
113                 lsfJobID = bsub(bsubline)
114                 self.boss.jobIDs[(lsfJobID, None)] = jobID
115                 self.boss.lsfJobIDs[jobID] = (lsfJobID, None)
116                 self.runningjobs.add((lsfJobID, None))
117 
118             # Test known job list
119             for lsfJobID in list(self.runningjobs):
120                 exit = getjobexitcode(lsfJobID)
121                 if exit is not None:
122                     self.updatedJobsQueue.put((lsfJobID, exit))
123                     self.runningjobs.remove(lsfJobID)
124 
125             time.sleep(10)
126 
127 class LSFBatchSystem(BatchSystemSupport):
128     """
129     The interface for running jobs on lsf, runs all the jobs you give it as they come in,
130     but in parallel.
131     """
132     @classmethod
133     def supportsWorkerCleanup(cls):
134         return False
135 
136     @classmethod
137     def supportsHotDeployment(cls):
138         return False
139 
140     def __init__(self, config, maxCores, maxMemory, maxDisk):
141         super(LSFBatchSystem, self).__init__(config, maxCores, maxMemory, maxDisk)
142         self.lsfResultsFile = self._getResultsFileName(config.jobStore)
143         #Reset the job queue and results (initially, we do this again once we've killed the jobs)
144         self.lsfResultsFileHandle = open(self.lsfResultsFile, 'w')
145         self.lsfResultsFileHandle.close() #We lose any previous state in this file, and ensure the files existence
146         self.currentjobs = set()
147         self.obtainSystemConstants()
148         self.jobIDs = dict()
149         self.lsfJobIDs = dict()
150         self.nextJobID = 0
151 
152         self.newJobsQueue = Queue()
153         self.updatedJobsQueue = Queue()
154         self.worker = Worker(self.newJobsQueue, self.updatedJobsQueue, self)
155         self.worker.setDaemon(True)
156         self.worker.start()
157 
158     def __des__(self):
159         #Closes the file handle associated with the results file.
160         self.lsfResultsFileHandle.close() #Close the results file, cos were done.
161 
162     def issueBatchJob(self, jobNode):
163         jobID = self.nextJobID
164         self.nextJobID += 1
165         self.currentjobs.add(jobID)
166         bsubline = prepareBsub(jobNode.cores, jobNode.memory) + [jobNode.command]
167         self.newJobsQueue.put((jobID, bsubline))
168         logger.debug("Issued the job command: %s with job id: %s " % (jobNode.command, str(jobID)))
169         return jobID
170 
171     def getLsfID(self, jobID):
172         if not jobID in self.lsfJobIDs:
173              RuntimeError("Unknown jobID, could not be converted")
174 
175         (job,task) = self.lsfJobIDs[jobID]
176         if task is None:
177              return str(job)
178         else:
179              return str(job) + "." + str(task)
180 
181     def killBatchJobs(self, jobIDs):
182         """Kills the given job IDs.
183         """
184         for jobID in jobIDs:
185             logger.debug("DEL: " + str(self.getLsfID(jobID)))
186             self.currentjobs.remove(jobID)
187             process = subprocess.Popen(["bkill", self.getLsfID(jobID)])
188             del self.jobIDs[self.lsfJobIDs[jobID]]
189             del self.lsfJobIDs[jobID]
190 
191         toKill = set(jobIDs)
192         while len(toKill) > 0:
193             for jobID in list(toKill):
194                 if getjobexitcode(self.lsfJobIDs[jobID]) is not None:
195                     toKill.remove(jobID)
196 
197             if len(toKill) > 0:
198                 logger.warn("Tried to kill some jobs, but something happened and they are still going, "
199                              "so I'll try again")
200                 time.sleep(5)
201 
202     def getIssuedBatchJobIDs(self):
203         """A list of jobs (as jobIDs) currently issued (may be running, or maybe 
204         just waiting).
205         """
206         return self.currentjobs
207 
208     def getRunningBatchJobIDs(self):
209         """Gets a map of jobs (as jobIDs) currently running (not just waiting) 
210         and a how long they have been running for (in seconds).
211         """
212         times = {}
213         currentjobs = set(self.lsfJobIDs[x] for x in self.getIssuedBatchJobIDs())
214         process = subprocess.Popen(["bjobs"], stdout = subprocess.PIPE)
215 
216         for curline in process.stdout:
217             items = curline.strip().split()
218             if (len(items) > 9 and (items[0]) in currentjobs) and items[2] == 'RUN':
219                 jobstart = "/".join(items[7:9]) + '/' + str(date.today().year)
220                 jobstart = jobstart + ' ' + items[9]
221                 jobstart = time.mktime(time.strptime(jobstart,"%b/%d/%Y %H:%M"))
222                 jobstart = time.mktime(time.strptime(jobstart,"%m/%d/%Y %H:%M:%S"))
223                 times[self.jobIDs[(items[0])]] = time.time() - jobstart
224         return times
225 
226     def getUpdatedBatchJob(self, maxWait):
227         try:
228             sgeJobID, retcode = self.updatedJobsQueue.get(timeout=maxWait)
229             self.updatedJobsQueue.task_done()
230             jobID, retcode = (self.jobIDs[sgeJobID], retcode)
231             self.currentjobs -= {self.jobIDs[sgeJobID]}
232         except Empty:
233             pass
234         else:
235             return jobID, retcode, None
236 
237     def getWaitDuration(self):
238         """We give parasol a second to catch its breath (in seconds)
239         """
240         #return 0.0
241         return 15
242 
243     @classmethod
244     def getRescueBatchJobFrequency(cls):
245         """Parasol leaks jobs, but rescuing jobs involves calls to parasol list jobs and pstat2,
246         making it expensive. We allow this every 10 minutes..
247         """
248         return 1800
249 
250     def obtainSystemConstants(self):
251         p = subprocess.Popen(["lshosts"], stdout = subprocess.PIPE, stderr = subprocess.STDOUT)
252 
253         line = p.stdout.readline()
254         items = line.strip().split()
255         num_columns = len(items)
256         cpu_index = None
257         mem_index = None
258         for i in range(num_columns):
259                 if items[i] == 'ncpus':
260                         cpu_index = i
261                 elif items[i] == 'maxmem':
262                         mem_index = i
263 
264         if cpu_index is None or mem_index is None:
265                 RuntimeError("lshosts command does not return ncpus or maxmem columns")
266 
267         p.stdout.readline()
268 
269         self.maxCPU = 0
270         self.maxMEM = MemoryString("0")
271         for line in p.stdout:
272                 items = line.strip().split()
273                 if len(items) < num_columns:
274                         RuntimeError("lshosts output has a varying number of columns")
275                 if items[cpu_index] != '-' and items[cpu_index] > self.maxCPU:
276                         self.maxCPU = items[cpu_index]
277                 if items[mem_index] != '-' and MemoryString(items[mem_index]) > self.maxMEM:
278                         self.maxMEM = MemoryString(items[mem_index])
279 
280         if self.maxCPU is 0 or self.maxMEM is 0:
281                 RuntimeError("lshosts returns null ncpus or maxmem info")
282         logger.debug("Got the maxCPU: %s" % (self.maxMEM))
283
```
--- 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/src/toil/batchSystems/lsf.py b/src/toil/batchSystems/lsf.py
--- a/src/toil/batchSystems/lsf.py
+++ b/src/toil/batchSystems/lsf.py
@@ -137,6 +137,9 @@
     def supportsHotDeployment(cls):
         return False
 
+    def shutdown(self):
+        pass
+
     def __init__(self, config, maxCores, maxMemory, maxDisk):
         super(LSFBatchSystem, self).__init__(config, maxCores, maxMemory, maxDisk)
         self.lsfResultsFile = self._getResultsFileName(config.jobStore)

{"golden_diff": "diff --git a/src/toil/batchSystems/lsf.py b/src/toil/batchSystems/lsf.py\n--- a/src/toil/batchSystems/lsf.py\n+++ b/src/toil/batchSystems/lsf.py\n@@ -137,6 +137,9 @@\n     def supportsHotDeployment(cls):\n         return False\n \n+    def shutdown(self):\n+        pass\n+\n     def __init__(self, config, maxCores, maxMemory, maxDisk):\n         super(LSFBatchSystem, self).__init__(config, maxCores, maxMemory, maxDisk)\n         self.lsfResultsFile = self._getResultsFileName(config.jobStore)\n", "issue": "Add missing abstract method in LSF batchsystem\n\nAdd missing abstract method in LSF batchsystem\n\n", "before_files": [{"content": "#Copyright (C) 2013 by Thomas Keane (tk2@sanger.ac.uk)\n#\n#Permission is hereby granted, free of charge, to any person obtaining a copy\n#of this software and associated documentation files (the \"Software\"), to deal\n#in the Software without restriction, including without limitation the rights\n#to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n#copies of the Software, and to permit persons to whom the Software is\n#furnished to do so, subject to the following conditions:\n#\n#The above copyright notice and this permission notice shall be included in\n#all copies or substantial portions of the Software.\n#\n#THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n#IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n#FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n#AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n#LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n#OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN\n#THE SOFTWARE.\nfrom __future__ import absolute_import\nimport logging\nimport subprocess\nimport time\nfrom threading import Thread\nfrom datetime import date\n\n# Python 3 compatibility imports\nfrom six.moves.queue import Empty, Queue\n\nfrom toil.batchSystems import MemoryString\nfrom toil.batchSystems.abstractBatchSystem import BatchSystemSupport\n\nlogger = logging.getLogger( __name__ )\n\n\n\ndef prepareBsub(cpu, mem):\n    mem = '' if mem is None else '-R \"select[type==X86_64 && mem > ' + str(int(mem/ 1000000)) + '] rusage[mem=' + str(int(mem/ 1000000)) + ']\" -M' + str(int(mem/ 1000000)) + '000'\n    cpu = '' if cpu is None else '-n ' + str(int(cpu))\n    bsubline = [\"bsub\", mem, cpu,\"-cwd\", \".\", \"-o\", \"/dev/null\", \"-e\", \"/dev/null\"]\n    return bsubline\n\ndef bsub(bsubline):\n    process = subprocess.Popen(\" \".join(bsubline), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)\n    liney = process.stdout.readline()\n    logger.debug(\"BSUB: \" + liney)\n    result = int(liney.strip().split()[1].strip('<>'))\n    logger.debug(\"Got the job id: %s\" % (str(result)))\n    return result\n\ndef getjobexitcode(lsfJobID):\n        job, task = lsfJobID\n\n        #first try bjobs to find out job state\n        args = [\"bjobs\", \"-l\", str(job)]\n        logger.debug(\"Checking job exit code for job via bjobs: \" + str(job))\n        process = subprocess.Popen(\" \".join(args), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)\n        started = 0\n        for line in process.stdout:\n            if line.find(\"Done successfully\") > -1:\n                logger.debug(\"bjobs detected job completed for job: \" + str(job))\n                return 0\n            elif line.find(\"Completed <exit>\") > -1:\n                logger.debug(\"bjobs detected job failed for job: \" + str(job))\n                return 1\n            elif line.find(\"New job is waiting for scheduling\") > -1:\n                logger.debug(\"bjobs detected job pending scheduling for job: \" + str(job))\n                return None\n            elif line.find(\"PENDING REASONS\") > -1:\n                logger.debug(\"bjobs detected job pending for job: \" + str(job))\n                return None\n            elif line.find(\"Started on \") > -1:\n                started = 1\n\n        if started == 1:\n            logger.debug(\"bjobs detected job started but not completed: \" + str(job))\n            return None\n\n        #if not found in bjobs, then try bacct (slower than bjobs)\n        logger.debug(\"bjobs failed to detect job - trying bacct: \" + str(job))\n\n        args = [\"bacct\", \"-l\", str(job)]\n        logger.debug(\"Checking job exit code for job via bacct:\" + str(job))\n        process = subprocess.Popen(\" \".join(args), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)\n        for line in process.stdout:\n            if line.find(\"Completed <done>\") > -1:\n                logger.debug(\"Detected job completed for job: \" + str(job))\n                return 0\n            elif line.find(\"Completed <exit>\") > -1:\n                logger.debug(\"Detected job failed for job: \" + str(job))\n                return 1\n        logger.debug(\"Cant determine exit code for job or job still running: \" + str(job))\n        return None\n\nclass Worker(Thread):\n    def __init__(self, newJobsQueue, updatedJobsQueue, boss):\n        Thread.__init__(self)\n        self.newJobsQueue = newJobsQueue\n        self.updatedJobsQueue = updatedJobsQueue\n        self.currentjobs = list()\n        self.runningjobs = set()\n        self.boss = boss\n\n    def run(self):\n        while True:\n            # Load new job ids:\n            while not self.newJobsQueue.empty():\n                self.currentjobs.append(self.newJobsQueue.get())\n\n            # Launch jobs as necessary:\n            while len(self.currentjobs) > 0:\n                jobID, bsubline = self.currentjobs.pop()\n                lsfJobID = bsub(bsubline)\n                self.boss.jobIDs[(lsfJobID, None)] = jobID\n                self.boss.lsfJobIDs[jobID] = (lsfJobID, None)\n                self.runningjobs.add((lsfJobID, None))\n\n            # Test known job list\n            for lsfJobID in list(self.runningjobs):\n                exit = getjobexitcode(lsfJobID)\n                if exit is not None:\n                    self.updatedJobsQueue.put((lsfJobID, exit))\n                    self.runningjobs.remove(lsfJobID)\n\n            time.sleep(10)\n\nclass LSFBatchSystem(BatchSystemSupport):\n    \"\"\"\n    The interface for running jobs on lsf, runs all the jobs you give it as they come in,\n    but in parallel.\n    \"\"\"\n    @classmethod\n    def supportsWorkerCleanup(cls):\n        return False\n\n    @classmethod\n    def supportsHotDeployment(cls):\n        return False\n\n    def __init__(self, config, maxCores, maxMemory, maxDisk):\n        super(LSFBatchSystem, self).__init__(config, maxCores, maxMemory, maxDisk)\n        self.lsfResultsFile = self._getResultsFileName(config.jobStore)\n        #Reset the job queue and results (initially, we do this again once we've killed the jobs)\n        self.lsfResultsFileHandle = open(self.lsfResultsFile, 'w')\n        self.lsfResultsFileHandle.close() #We lose any previous state in this file, and ensure the files existence\n        self.currentjobs = set()\n        self.obtainSystemConstants()\n        self.jobIDs = dict()\n        self.lsfJobIDs = dict()\n        self.nextJobID = 0\n\n        self.newJobsQueue = Queue()\n        self.updatedJobsQueue = Queue()\n        self.worker = Worker(self.newJobsQueue, self.updatedJobsQueue, self)\n        self.worker.setDaemon(True)\n        self.worker.start()\n\n    def __des__(self):\n        #Closes the file handle associated with the results file.\n        self.lsfResultsFileHandle.close() #Close the results file, cos were done.\n\n    def issueBatchJob(self, jobNode):\n        jobID = self.nextJobID\n        self.nextJobID += 1\n        self.currentjobs.add(jobID)\n        bsubline = prepareBsub(jobNode.cores, jobNode.memory) + [jobNode.command]\n        self.newJobsQueue.put((jobID, bsubline))\n        logger.debug(\"Issued the job command: %s with job id: %s \" % (jobNode.command, str(jobID)))\n        return jobID\n\n    def getLsfID(self, jobID):\n        if not jobID in self.lsfJobIDs:\n             RuntimeError(\"Unknown jobID, could not be converted\")\n\n        (job,task) = self.lsfJobIDs[jobID]\n        if task is None:\n             return str(job)\n        else:\n             return str(job) + \".\" + str(task)\n\n    def killBatchJobs(self, jobIDs):\n        \"\"\"Kills the given job IDs.\n        \"\"\"\n        for jobID in jobIDs:\n            logger.debug(\"DEL: \" + str(self.getLsfID(jobID)))\n            self.currentjobs.remove(jobID)\n            process = subprocess.Popen([\"bkill\", self.getLsfID(jobID)])\n            del self.jobIDs[self.lsfJobIDs[jobID]]\n            del self.lsfJobIDs[jobID]\n\n        toKill = set(jobIDs)\n        while len(toKill) > 0:\n            for jobID in list(toKill):\n                if getjobexitcode(self.lsfJobIDs[jobID]) is not None:\n                    toKill.remove(jobID)\n\n            if len(toKill) > 0:\n                logger.warn(\"Tried to kill some jobs, but something happened and they are still going, \"\n                             \"so I'll try again\")\n                time.sleep(5)\n\n    def getIssuedBatchJobIDs(self):\n        \"\"\"A list of jobs (as jobIDs) currently issued (may be running, or maybe \n        just waiting).\n        \"\"\"\n        return self.currentjobs\n\n    def getRunningBatchJobIDs(self):\n        \"\"\"Gets a map of jobs (as jobIDs) currently running (not just waiting) \n        and a how long they have been running for (in seconds).\n        \"\"\"\n        times = {}\n        currentjobs = set(self.lsfJobIDs[x] for x in self.getIssuedBatchJobIDs())\n        process = subprocess.Popen([\"bjobs\"], stdout = subprocess.PIPE)\n\n        for curline in process.stdout:\n            items = curline.strip().split()\n            if (len(items) > 9 and (items[0]) in currentjobs) and items[2] == 'RUN':\n                jobstart = \"/\".join(items[7:9]) + '/' + str(date.today().year)\n                jobstart = jobstart + ' ' + items[9]\n                jobstart = time.mktime(time.strptime(jobstart,\"%b/%d/%Y %H:%M\"))\n                jobstart = time.mktime(time.strptime(jobstart,\"%m/%d/%Y %H:%M:%S\"))\n                times[self.jobIDs[(items[0])]] = time.time() - jobstart\n        return times\n\n    def getUpdatedBatchJob(self, maxWait):\n        try:\n            sgeJobID, retcode = self.updatedJobsQueue.get(timeout=maxWait)\n            self.updatedJobsQueue.task_done()\n            jobID, retcode = (self.jobIDs[sgeJobID], retcode)\n            self.currentjobs -= {self.jobIDs[sgeJobID]}\n        except Empty:\n            pass\n        else:\n            return jobID, retcode, None\n\n    def getWaitDuration(self):\n        \"\"\"We give parasol a second to catch its breath (in seconds)\n        \"\"\"\n        #return 0.0\n        return 15\n\n    @classmethod\n    def getRescueBatchJobFrequency(cls):\n        \"\"\"Parasol leaks jobs, but rescuing jobs involves calls to parasol list jobs and pstat2,\n        making it expensive. We allow this every 10 minutes..\n        \"\"\"\n        return 1800\n\n    def obtainSystemConstants(self):\n        p = subprocess.Popen([\"lshosts\"], stdout = subprocess.PIPE, stderr = subprocess.STDOUT)\n\n        line = p.stdout.readline()\n        items = line.strip().split()\n        num_columns = len(items)\n        cpu_index = None\n        mem_index = None\n        for i in range(num_columns):\n                if items[i] == 'ncpus':\n                        cpu_index = i\n                elif items[i] == 'maxmem':\n                        mem_index = i\n\n        if cpu_index is None or mem_index is None:\n                RuntimeError(\"lshosts command does not return ncpus or maxmem columns\")\n\n        p.stdout.readline()\n\n        self.maxCPU = 0\n        self.maxMEM = MemoryString(\"0\")\n        for line in p.stdout:\n                items = line.strip().split()\n                if len(items) < num_columns:\n                        RuntimeError(\"lshosts output has a varying number of columns\")\n                if items[cpu_index] != '-' and items[cpu_index] > self.maxCPU:\n                        self.maxCPU = items[cpu_index]\n                if items[mem_index] != '-' and MemoryString(items[mem_index]) > self.maxMEM:\n                        self.maxMEM = MemoryString(items[mem_index])\n\n        if self.maxCPU is 0 or self.maxMEM is 0:\n                RuntimeError(\"lshosts returns null ncpus or maxmem info\")\n        logger.debug(\"Got the maxCPU: %s\" % (self.maxMEM))\n", "path": "src/toil/batchSystems/lsf.py"}], "after_files": [{"content": "#Copyright (C) 2013 by Thomas Keane (tk2@sanger.ac.uk)\n#\n#Permission is hereby granted, free of charge, to any person obtaining a copy\n#of this software and associated documentation files (the \"Software\"), to deal\n#in the Software without restriction, including without limitation the rights\n#to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n#copies of the Software, and to permit persons to whom the Software is\n#furnished to do so, subject to the following conditions:\n#\n#The above copyright notice and this permission notice shall be included in\n#all copies or substantial portions of the Software.\n#\n#THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n#IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n#FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n#AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n#LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n#OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN\n#THE SOFTWARE.\nfrom __future__ import absolute_import\nimport logging\nimport subprocess\nimport time\nfrom threading import Thread\nfrom datetime import date\n\n# Python 3 compatibility imports\nfrom six.moves.queue import Empty, Queue\n\nfrom toil.batchSystems import MemoryString\nfrom toil.batchSystems.abstractBatchSystem import BatchSystemSupport\n\nlogger = logging.getLogger( __name__ )\n\n\n\ndef prepareBsub(cpu, mem):\n    mem = '' if mem is None else '-R \"select[type==X86_64 && mem > ' + str(int(mem/ 1000000)) + '] rusage[mem=' + str(int(mem/ 1000000)) + ']\" -M' + str(int(mem/ 1000000)) + '000'\n    cpu = '' if cpu is None else '-n ' + str(int(cpu))\n    bsubline = [\"bsub\", mem, cpu,\"-cwd\", \".\", \"-o\", \"/dev/null\", \"-e\", \"/dev/null\"]\n    return bsubline\n\ndef bsub(bsubline):\n    process = subprocess.Popen(\" \".join(bsubline), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)\n    liney = process.stdout.readline()\n    logger.debug(\"BSUB: \" + liney)\n    result = int(liney.strip().split()[1].strip('<>'))\n    logger.debug(\"Got the job id: %s\" % (str(result)))\n    return result\n\ndef getjobexitcode(lsfJobID):\n        job, task = lsfJobID\n\n        #first try bjobs to find out job state\n        args = [\"bjobs\", \"-l\", str(job)]\n        logger.debug(\"Checking job exit code for job via bjobs: \" + str(job))\n        process = subprocess.Popen(\" \".join(args), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)\n        started = 0\n        for line in process.stdout:\n            if line.find(\"Done successfully\") > -1:\n                logger.debug(\"bjobs detected job completed for job: \" + str(job))\n                return 0\n            elif line.find(\"Completed <exit>\") > -1:\n                logger.debug(\"bjobs detected job failed for job: \" + str(job))\n                return 1\n            elif line.find(\"New job is waiting for scheduling\") > -1:\n                logger.debug(\"bjobs detected job pending scheduling for job: \" + str(job))\n                return None\n            elif line.find(\"PENDING REASONS\") > -1:\n                logger.debug(\"bjobs detected job pending for job: \" + str(job))\n                return None\n            elif line.find(\"Started on \") > -1:\n                started = 1\n\n        if started == 1:\n            logger.debug(\"bjobs detected job started but not completed: \" + str(job))\n            return None\n\n        #if not found in bjobs, then try bacct (slower than bjobs)\n        logger.debug(\"bjobs failed to detect job - trying bacct: \" + str(job))\n\n        args = [\"bacct\", \"-l\", str(job)]\n        logger.debug(\"Checking job exit code for job via bacct:\" + str(job))\n        process = subprocess.Popen(\" \".join(args), shell=True, stdout = subprocess.PIPE, stderr = subprocess.STDOUT)\n        for line in process.stdout:\n            if line.find(\"Completed <done>\") > -1:\n                logger.debug(\"Detected job completed for job: \" + str(job))\n                return 0\n            elif line.find(\"Completed <exit>\") > -1:\n                logger.debug(\"Detected job failed for job: \" + str(job))\n                return 1\n        logger.debug(\"Cant determine exit code for job or job still running: \" + str(job))\n        return None\n\nclass Worker(Thread):\n    def __init__(self, newJobsQueue, updatedJobsQueue, boss):\n        Thread.__init__(self)\n        self.newJobsQueue = newJobsQueue\n        self.updatedJobsQueue = updatedJobsQueue\n        self.currentjobs = list()\n        self.runningjobs = set()\n        self.boss = boss\n\n    def run(self):\n        while True:\n            # Load new job ids:\n            while not self.newJobsQueue.empty():\n                self.currentjobs.append(self.newJobsQueue.get())\n\n            # Launch jobs as necessary:\n            while len(self.currentjobs) > 0:\n                jobID, bsubline = self.currentjobs.pop()\n                lsfJobID = bsub(bsubline)\n                self.boss.jobIDs[(lsfJobID, None)] = jobID\n                self.boss.lsfJobIDs[jobID] = (lsfJobID, None)\n                self.runningjobs.add((lsfJobID, None))\n\n            # Test known job list\n            for lsfJobID in list(self.runningjobs):\n                exit = getjobexitcode(lsfJobID)\n                if exit is not None:\n                    self.updatedJobsQueue.put((lsfJobID, exit))\n                    self.runningjobs.remove(lsfJobID)\n\n            time.sleep(10)\n\nclass LSFBatchSystem(BatchSystemSupport):\n    \"\"\"\n    The interface for running jobs on lsf, runs all the jobs you give it as they come in,\n    but in parallel.\n    \"\"\"\n    @classmethod\n    def supportsWorkerCleanup(cls):\n        return False\n\n    @classmethod\n    def supportsHotDeployment(cls):\n        return False\n\n    def shutdown(self):\n        pass\n\n    def __init__(self, config, maxCores, maxMemory, maxDisk):\n        super(LSFBatchSystem, self).__init__(config, maxCores, maxMemory, maxDisk)\n        self.lsfResultsFile = self._getResultsFileName(config.jobStore)\n        #Reset the job queue and results (initially, we do this again once we've killed the jobs)\n        self.lsfResultsFileHandle = open(self.lsfResultsFile, 'w')\n        self.lsfResultsFileHandle.close() #We lose any previous state in this file, and ensure the files existence\n        self.currentjobs = set()\n        self.obtainSystemConstants()\n        self.jobIDs = dict()\n        self.lsfJobIDs = dict()\n        self.nextJobID = 0\n\n        self.newJobsQueue = Queue()\n        self.updatedJobsQueue = Queue()\n        self.worker = Worker(self.newJobsQueue, self.updatedJobsQueue, self)\n        self.worker.setDaemon(True)\n        self.worker.start()\n\n    def __des__(self):\n        #Closes the file handle associated with the results file.\n        self.lsfResultsFileHandle.close() #Close the results file, cos were done.\n\n    def issueBatchJob(self, jobNode):\n        jobID = self.nextJobID\n        self.nextJobID += 1\n        self.currentjobs.add(jobID)\n        bsubline = prepareBsub(jobNode.cores, jobNode.memory) + [jobNode.command]\n        self.newJobsQueue.put((jobID, bsubline))\n        logger.debug(\"Issued the job command: %s with job id: %s \" % (jobNode.command, str(jobID)))\n        return jobID\n\n    def getLsfID(self, jobID):\n        if not jobID in self.lsfJobIDs:\n             RuntimeError(\"Unknown jobID, could not be converted\")\n\n        (job,task) = self.lsfJobIDs[jobID]\n        if task is None:\n             return str(job)\n        else:\n             return str(job) + \".\" + str(task)\n\n    def killBatchJobs(self, jobIDs):\n        \"\"\"Kills the given job IDs.\n        \"\"\"\n        for jobID in jobIDs:\n            logger.debug(\"DEL: \" + str(self.getLsfID(jobID)))\n            self.currentjobs.remove(jobID)\n            process = subprocess.Popen([\"bkill\", self.getLsfID(jobID)])\n            del self.jobIDs[self.lsfJobIDs[jobID]]\n            del self.lsfJobIDs[jobID]\n\n        toKill = set(jobIDs)\n        while len(toKill) > 0:\n            for jobID in list(toKill):\n                if getjobexitcode(self.lsfJobIDs[jobID]) is not None:\n                    toKill.remove(jobID)\n\n            if len(toKill) > 0:\n                logger.warn(\"Tried to kill some jobs, but something happened and they are still going, \"\n                             \"so I'll try again\")\n                time.sleep(5)\n\n    def getIssuedBatchJobIDs(self):\n        \"\"\"A list of jobs (as jobIDs) currently issued (may be running, or maybe \n        just waiting).\n        \"\"\"\n        return self.currentjobs\n\n    def getRunningBatchJobIDs(self):\n        \"\"\"Gets a map of jobs (as jobIDs) currently running (not just waiting) \n        and a how long they have been running for (in seconds).\n        \"\"\"\n        times = {}\n        currentjobs = set(self.lsfJobIDs[x] for x in self.getIssuedBatchJobIDs())\n        process = subprocess.Popen([\"bjobs\"], stdout = subprocess.PIPE)\n\n        for curline in process.stdout:\n            items = curline.strip().split()\n            if (len(items) > 9 and (items[0]) in currentjobs) and items[2] == 'RUN':\n                jobstart = \"/\".join(items[7:9]) + '/' + str(date.today().year)\n                jobstart = jobstart + ' ' + items[9]\n                jobstart = time.mktime(time.strptime(jobstart,\"%b/%d/%Y %H:%M\"))\n                jobstart = time.mktime(time.strptime(jobstart,\"%m/%d/%Y %H:%M:%S\"))\n                times[self.jobIDs[(items[0])]] = time.time() - jobstart\n        return times\n\n    def getUpdatedBatchJob(self, maxWait):\n        try:\n            sgeJobID, retcode = self.updatedJobsQueue.get(timeout=maxWait)\n            self.updatedJobsQueue.task_done()\n            jobID, retcode = (self.jobIDs[sgeJobID], retcode)\n            self.currentjobs -= {self.jobIDs[sgeJobID]}\n        except Empty:\n            pass\n        else:\n            return jobID, retcode, None\n\n    def getWaitDuration(self):\n        \"\"\"We give parasol a second to catch its breath (in seconds)\n        \"\"\"\n        #return 0.0\n        return 15\n\n    @classmethod\n    def getRescueBatchJobFrequency(cls):\n        \"\"\"Parasol leaks jobs, but rescuing jobs involves calls to parasol list jobs and pstat2,\n        making it expensive. We allow this every 10 minutes..\n        \"\"\"\n        return 1800\n\n    def obtainSystemConstants(self):\n        p = subprocess.Popen([\"lshosts\"], stdout = subprocess.PIPE, stderr = subprocess.STDOUT)\n\n        line = p.stdout.readline()\n        items = line.strip().split()\n        num_columns = len(items)\n        cpu_index = None\n        mem_index = None\n        for i in range(num_columns):\n                if items[i] == 'ncpus':\n                        cpu_index = i\n                elif items[i] == 'maxmem':\n                        mem_index = i\n\n        if cpu_index is None or mem_index is None:\n                RuntimeError(\"lshosts command does not return ncpus or maxmem columns\")\n\n        p.stdout.readline()\n\n        self.maxCPU = 0\n        self.maxMEM = MemoryString(\"0\")\n        for line in p.stdout:\n                items = line.strip().split()\n                if len(items) < num_columns:\n                        RuntimeError(\"lshosts output has a varying number of columns\")\n                if items[cpu_index] != '-' and items[cpu_index] > self.maxCPU:\n                        self.maxCPU = items[cpu_index]\n                if items[mem_index] != '-' and MemoryString(items[mem_index]) > self.maxMEM:\n                        self.maxMEM = MemoryString(items[mem_index])\n\n        if self.maxCPU is 0 or self.maxMEM is 0:\n                RuntimeError(\"lshosts returns null ncpus or maxmem info\")\n        logger.debug(\"Got the maxCPU: %s\" % (self.maxMEM))\n", "path": "src/toil/batchSystems/lsf.py"}]}