Do not update the job status when it's finished

Always refresh the work directory when job finished is received.
MemVerge · Sep 8, 2023 · dbf06b5 · dbf06b5
1 parent 7bc44f3
commit dbf06b5
Show file tree

Hide file tree

Showing 6 changed files with 35 additions and 50 deletions.
diff --git a/plugins/nf-float/src/main/com/memverge/nextflow/FloatGridExecutor.groovy b/plugins/nf-float/src/main/com/memverge/nextflow/FloatGridExecutor.groovy
@@ -167,6 +167,7 @@ class FloatGridExecutor extends AbstractGridExecutor {
 
             if (res.succeeded) {
                 job = FloatJob.parse(res.out)
+                job = floatJobs.updateJob(job)
             }
         } catch (Exception e) {
             log.warn "[float] failed to retrieve job status $nfJobID, float: ${job.floatJobID}", e
@@ -532,15 +533,16 @@ class FloatGridExecutor extends AbstractGridExecutor {
         if (!job) {
             return FloatStatus.UNKNOWN
         }
-        log.debug "[float] task id: ${task.id}, nf-job-id: $job.nfJobID, " +
+        log.info "[float] task id: ${task.id}, nf-job-id: $job.nfJobID, " +
                 "float-job-id: $job.floatJobID, float status: $job.status"
-        if (job.finished) {
-            floatJobs.refreshWorkDir(job.nfJobID)
-            task.exitStatus = job.rcCode
-        }
         return job.status
     }
 
+    Integer getJobRC(TaskId id) {
+        def job = getJob(id)
+        return job.rcCode
+    }
+
     static private Map<FloatStatus, QueueStatus> STATUS_MAP = new HashMap<>()
 
     static {

diff --git a/plugins/nf-float/src/main/com/memverge/nextflow/FloatJob.groovy b/plugins/nf-float/src/main/com/memverge/nextflow/FloatJob.groovy
@@ -126,8 +126,8 @@ class FloatJob {
         return status ? status.isFinished() : false
     }
 
-    static Map<String, FloatJob> parseJobMap(String input) {
-        Map<String, FloatJob> ret = new HashMap<>()
+    static List<FloatJob> parseJobMap(String input) {
+        List<FloatJob> ret = []
         try {
             def parser = new JsonSlurper()
             def obj = parser.parseText(input)
@@ -142,7 +142,7 @@ class FloatJob {
                     job.floatJobID = floatJobID
                     job.status = FloatStatus.of(status)
                     job.rc = i.rc as String
-                    ret[nfJobID] = job
+                    ret.add(job)
                 }
             }
         } catch (Exception e) {

diff --git a/plugins/nf-float/src/main/com/memverge/nextflow/FloatJobs.groovy b/plugins/nf-float/src/main/com/memverge/nextflow/FloatJobs.groovy
@@ -15,14 +15,12 @@
  */
 package com.memverge.nextflow
 
-import groovy.transform.WithReadLock
-import groovy.transform.WithWriteLock
+
 import groovy.util.logging.Slf4j
 import nextflow.file.FileHelper
 import nextflow.processor.TaskId
 import org.apache.commons.lang.RandomStringUtils
 
-import java.nio.file.NoSuchFileException
 import java.nio.file.Path
 import java.util.concurrent.ConcurrentHashMap
 
@@ -59,7 +57,6 @@ class FloatJobs {
         return floatJobID2oc.getOrDefault(floatJobID, ocs[0])
     }
 
-    @WithReadLock
     Map<String, FloatJob> getNfJobID2job() {
         return nfJobID2FloatJob
     }
@@ -73,12 +70,18 @@ class FloatJobs {
         return updateOcStatus(ocs[0], text)
     }
 
-    FloatStatus getJobStatus(String nfJobID) {
-        FloatJob job = nfJobID2FloatJob.get(nfJobID)
-        if (job == null) {
-            return FloatStatus.UNKNOWN
+    FloatJob updateJob(FloatJob job) {
+        FloatJob existingJob = nfJobID2job.get(job.nfJobID)
+        if (existingJob != null && existingJob.finished) {
+            // job already finished, no need to update
+            job = existingJob
+        } else {
+            nfJobID2job.put(job.nfJobID, job)
+        }
+        if (job.finished) {
+            refreshWorkDir(job.nfJobID)
         }
-        return job.status
+        return job
     }
 
     def refreshWorkDir(String nfJobID) {
@@ -89,37 +92,15 @@ class FloatJobs {
         }
     }
 
-    @WithWriteLock
     def updateOcStatus(String oc, String text) {
-        def stMap = FloatJob.parseJobMap(text)
-        stMap.each { nfJobID, job ->
+        def jobs = FloatJob.parseJobMap(text)
+        jobs.each {job ->
             if (!job.nfJobID || !job.status) {
                 return
             }
             floatJobID2oc.put(job.floatJobID, oc)
-            def currentSt = getJobStatus(nfJobID)
-            def workDir = nfJobID2workDir.get(job.nfJobID)
-            if (workDir && job.finished) {
-                refreshWorkDir(job.nfJobID)
-                def files = ['.command.out', '.command.err', '.exitcode']
-                if (!currentSt.finished && job.finished) {
-                    for (filename in files) {
-                        def name = workDir.resolve(filename)
-                        try {
-                            !FileHelper.checkIfExists(name, [checkIfExists: true])
-                        } catch (NoSuchFileException ex) {
-                            log.info "[float] job $nfJobID completed " +
-                                    "but file not found: ${ex.message}"
-                            job.status = currentSt
-                            return
-                        }
-                    }
-                    log.debug "[float] found $files in: $workDir"
-                }
-            }
-
+            updateJob(job)
         }
-        nfJobID2FloatJob += stMap
         log.debug "[float] update op-center $oc job status"
         return nfJobID2FloatJob
     }

diff --git a/plugins/nf-float/src/main/com/memverge/nextflow/FloatTaskHandler.groovy b/plugins/nf-float/src/main/com/memverge/nextflow/FloatTaskHandler.groovy
@@ -74,16 +74,18 @@ class FloatTaskHandler extends GridTaskHandler {
         final FloatStatus st =  floatExecutor.getJobStatus(task)
         if (st.finished) {
             status = COMPLETED
+            task.exitStatus = readExitStatus()
             if (task.exitStatus == null) {
-                task.exitStatus = readExitStatus()
+                task.exitStatus = floatExecutor.getJobRC(task.id)
             }
-            // both exit status and job rc code are empty
             if (task.exitStatus == null) {
                 if (st.isError()) {
                     task.exitStatus = 1
                 } else {
                     task.exitStatus = 0
                 }
+                log.info "both .exitcode and rc are empty for ${task.id}," +
+                        "set exit to ${task.exitStatus}"
             }
             task.stdout = outputFile
             task.stderr = errorFile

diff --git a/plugins/nf-float/src/test/com/memverge/nextflow/FloatGridExecutorTest.groovy b/plugins/nf-float/src/test/com/memverge/nextflow/FloatGridExecutorTest.groovy
@@ -535,7 +535,7 @@ class FloatGridExecutorTest extends FloatBaseTest {
         res['tJob-5'] == qs.ERROR
         res['tJob-6'] == qs.ERROR
         res['tJob-7'] == qs.ERROR
-        res['tJob-8'] == qs.UNKNOWN
+        res['tJob-8'] == qs.DONE
         res['tJob-9'] == qs.RUNNING
         res['tJob-10'] == qs.RUNNING
         res['tJob-11'] == qs.RUNNING

diff --git a/plugins/nf-float/src/test/com/memverge/nextflow/FloatJobTest.groovy b/plugins/nf-float/src/test/com/memverge/nextflow/FloatJobTest.groovy
@@ -148,9 +148,9 @@ class FloatJobTest extends Specification {
         ]"""
 
         when:
-        def stMap = FloatJob.parseJobMap(out)
-        def st1 = stMap['job-1']
-        def st2 = stMap['job-3']
+        def jobs = FloatJob.parseJobMap(out)
+        def st1 = jobs.get(0)
+        def st2 = jobs.get(1)
 
         then:
         st1.status == FloatStatus.ERROR
@@ -168,9 +168,9 @@ class FloatJobTest extends Specification {
         final out = """No jobs"""
 
         when:
-        def stMap = FloatJob.parseJobMap(out)
+        def jobs = FloatJob.parseJobMap(out)
 
         then:
-        stMap.size() == 0
+        jobs.size() == 0
     }
 }