reduce load on SGE with qacct command

.gitignore

@@ -1,4 +1,5 @@
 *.pyc
+.*
 /src/*.egg-info
 /build
 /dist

src/toil/batchSystems/abstractGridEngineBatchSystem.py

@@ -184,7 +184,7 @@ def killJobs(self):
             while killList:
                 for jobID in list(killList):
                     batchJobID = self.getBatchSystemID(jobID)
-                    if with_retries(self.getJobExitCode, batchJobID) is not None:
+                    if with_retries(self.getJobExitCode, batchJobID, jobID) is not None:
                         logger.debug('Adding jobID %s to killedJobsQueue', jobID)
                         self.killedJobsQueue.put(jobID)
                         killList.remove(jobID)
@@ -207,7 +207,7 @@ def checkOnJobs(self):
             activity = False
             for jobID in list(self.runningJobs):
                 batchJobID = self.getBatchSystemID(jobID)
-                status = with_retries(self.getJobExitCode, batchJobID)
+                status = with_retries(self.getJobExitCode, batchJobID, jobID)
                 if status is not None:
                     activity = True
                     self.updatedJobsQueue.put((jobID, status))
@@ -285,7 +285,7 @@ def killJob(self, jobID):
             raise NotImplementedError()
 
         @abstractmethod
-        def getJobExitCode(self, batchJobID):
+        def getJobExitCode(self, batchJobID, jobID):
             """
             Returns job exit code. Implementation-specific; called by
             AbstractGridEngineWorker.checkOnJobs()

src/toil/batchSystems/gridengine.py

@@ -68,7 +68,7 @@ def submitJob(self, subLine):
             result = int(process.stdout.readline().decode('utf-8').strip())
             return result
 
-        def getJobExitCode(self, sgeJobID):
+        def getJobExitCode(self, sgeJobID, jobID):
             # the task is set as part of the job ID if using getBatchSystemID()
             job, task = (sgeJobID, None)
             if '.' in sgeJobID:
@@ -80,13 +80,19 @@ def getJobExitCode(self, sgeJobID):
                 args.extend(["-t", str(task)])
 
             logger.debug("Running %r", args)
+            iscurrtoiljob = False
             process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
             for line in process.stdout:
                 if line.startswith("failed") and int(line.split()[1]) == 1:
                     return 1
-                elif line.startswith("exit_status"):
+                if line.startswith("jobname") and line.split()[1] == "toil_job_" + str(jobID):
+                    iscurrtoiljob = True
+                if iscurrtoiljob and line.startswith("exit_status"):
                     logger.debug('Exit Status: %r', line.split()[1])
                     return int(line.split()[1])
+                #elif line.startswith("exit_status"):
+                #    logger.debug('Exit Status: %r', line.split()[1])
+                #    return int(line.split()[1])
             return None
 
         """
@@ -104,7 +110,8 @@ def prepareQsub(self, cpu, mem, jobID):
             reqline = list()
             sgeArgs = os.getenv('TOIL_GRIDENGINE_ARGS')
             if mem is not None:
-                memStr = str(old_div(mem, 1024)) + 'K'
+                memStr = str(int(math.ceil(mem/(1024*math.ceil(cpu))))) + 'K'
+                # memStr = str(old_div(mem, 1024)) + 'K'
                 if not self.boss.config.manualMemArgs:
                     # for UGE instead of SGE; see #2309
                     reqline += ['vf=' + memStr, 'h_vmem=' + memStr]
@@ -137,7 +144,7 @@ def prepareQsub(self, cpu, mem, jobID):
 
     @classmethod
     def getWaitDuration(cls):
-        return 1
+        return 15
 
     @classmethod
     def obtainSystemConstants(cls):

src/toil/batchSystems/lsf.py

@@ -94,7 +94,7 @@ def getJobExitCode(self, lsfJobID):
             process = subprocess.Popen(args, stdout=subprocess.PIPE,
                                        stderr=subprocess.STDOUT)
             started = 0
-            for line in process.stdout:
+            for line in process.stdout.readlines():
                 if "Done successfully" in line:
                     logger.debug("bjobs detected job completed for job: "
                                  "{}".format(job))
@@ -132,7 +132,7 @@ def getJobExitCode(self, lsfJobID):
             args = ["bacct", "-l", str(job)]
             process = subprocess.Popen(args, stdout=subprocess.PIPE,
                                        stderr=subprocess.STDOUT)
-            for line in process.stdout:
+            for line in process.stdout.readlines():
                 if line.find("Completed <done>") > -1:
                     logger.debug("Detected job completed for job: "
                                  "{}".format(job))
@@ -212,7 +212,7 @@ def obtainSystemConstants(cls):
 
         maxCPU = 0
         maxMEM = MemoryString("0")
-        for line in p.stdout:
+        for line in p.stdout.readlines():
                 items = line.strip().split()
                 if len(items) < num_columns:
                         RuntimeError("lshosts output has a varying number of "