Solve queueSize exceeded when using job arrays

modules/nextflow/src/main/groovy/nextflow/processor/ParallelPollingMonitor.groovy

@@ -55,7 +55,7 @@ class ParallelPollingMonitor extends TaskPollingMonitor {
 
     @Override
     protected boolean canSubmit(TaskHandler handler) {
-        return super.canSubmit(handler) && (semaphore == null || semaphore.tryAcquire())
+        return super.canSubmit(handler) && (semaphore == null || semaphore.tryAcquire(getConsumedSlots(handler)))
     }
 
     protected RateLimiter createSubmitRateLimit() {
@@ -95,7 +95,7 @@ class ParallelPollingMonitor extends TaskPollingMonitor {
 
     @Override
     boolean evict(TaskHandler handler) {
-        semaphore?.release()
+        semaphore?.release(getConsumedSlots(handler))
         return super.evict(handler)
     }
 }

modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy

@@ -198,7 +198,18 @@ class TaskPollingMonitor implements TaskMonitor {
      *      by the polling monitor
      */
     protected boolean canSubmit(TaskHandler handler) {
-        (capacity>0 ? runningQueue.size() < capacity : true) && handler.canForkProcess() && handler.isReady()
+        // Task array are added in the running queue one by one, but to submit we need to check if there is capacity for all children.(#5920)
+        int slots = getConsumedSlots(handler)
+        if( capacity > 0 && slots > capacity ) {
+            //Waning because this condition could make a workflow running until timeout.
+            log.warn1("Submission of task array ${handler.task.name} is exceeding the executor capacity (requested: $slots, capacity: $capacity)")
+            return false
+        }
+        (capacity>0 ? runningQueue.size() + slots <= capacity : true) && handler.canForkProcess() && handler.isReady()
+    }
+
+    protected static int getConsumedSlots(TaskHandler handler){
+        handler.task instanceof TaskArrayRun ? (handler.task as TaskArrayRun).getArraySize() : 1
     }
 
     /**

modules/nextflow/src/test/groovy/nextflow/processor/TaskPollingMonitorTest.groovy

@@ -20,7 +20,12 @@ package nextflow.processor
 import nextflow.Session
 import nextflow.util.Duration
 import nextflow.util.RateUnit
+import nextflow.util.ThrottlingExecutor
 import spock.lang.Specification
+import spock.lang.Unroll
+
+import java.util.concurrent.atomic.AtomicInteger
+
 /**
  *
  * @author Paolo Di Tommaso <[email protected]>
@@ -148,4 +153,37 @@ class TaskPollingMonitorTest extends Specification {
         3 * session.notifyTaskSubmit(handler)
     }
 
+
+    @Unroll
+    def 'should validate can submit method' () {
+        given:
+        def session = Mock(Session)
+        def monitor = Spy(new TaskPollingMonitor(name: 'foo', session: session, capacity : CAPACITY, pollInterval: Duration.of('1min')))
+        and:
+        def handler = Mock(TaskHandler) {
+            getTask() >> Mock(TaskRun)
+        }
+        def arrayHandler = Mock(TaskHandler) {
+            getTask() >> Mock(TaskArrayRun) {
+                getArraySize() >> ARRAY
+            }
+            canForkProcess() >> true
+            isReady() >> true
+        }
+
+        and:
+        SUBMIT.times { monitor.runningQueue.add(handler)  }
+
+        expect:
+        monitor.runningQueue.size() == SUBMIT
+        monitor.canSubmit(arrayHandler) == EXPECTED
+
+        where:
+        CAPACITY    | SUBMIT | ARRAY | EXPECTED
+        0           |   0    | 2     | true
+        2           |   0    | 4     | false
+        10          |   5    | 5     | true
+        10          |   8    | 5     | false
+    }
+
 }