ref: simplify scheduler constructor

just take the list of nodes directly.

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy

@@ -16,6 +16,7 @@
 
 package nextflow.k8s
 
+import groovy.transform.CompileDynamic
 import nextflow.k8s.strategies.K8sHashSchedulingStrategy
 
 import java.util.concurrent.TimeUnit
@@ -97,11 +98,21 @@ class K8sExecutor extends Executor implements ExtensionPoint {
         this.runtimeRecorder = new K8sRuntimeRecorder(k8sConfig.recordTaskRuntimes, k8sConfig.runtimeRecordPath)
         this.runtimeEstimator = new K8sRuntimeEstimator(k8sConfig.runtimeRecordPath)
 
-        this.taskScheduler = new K8sTaskScheduler(this, new K8sHashSchedulingStrategy())
+        this.taskScheduler = new K8sTaskScheduler(getNodeList(), new K8sHashSchedulingStrategy())
         this.schedulerThread = new Thread(this.taskScheduler)
         this.schedulerThread.start()
     }
 
+    @CompileDynamic
+    private String[] getNodeList() {
+        final resp = getClient().nodeList()
+        ArrayList<String> nodes = new ArrayList<String>()
+        for ( Map item : resp.items ) {
+            nodes.add(item.metadata.name as String)
+        }
+        return nodes.toArray()
+    }
+
     @Override
     void shutdown() {
         this.runtimeRecorder.write()

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sSchedulingStrategy.groovy

@@ -7,8 +7,15 @@ interface K8sSchedulingStrategy {
      *
      * @param scheduler the calling scheduler object
      * @param queue Pending scheduling requests
-     * @param nodes Available Kubernetes node names
      * @return A launch decision, or {@code null} when no task should be launched now
      */
-    K8sSchedulingDecision schedule(K8sTaskScheduler scheduler, List<K8sSchedulingRequest> queue, List<String> nodes)
+    K8sSchedulingDecision schedule(K8sTaskScheduler scheduler, List<K8sSchedulingRequest> queue)
+
+    /**
+     * Decides if the scheduler should immediately invoke the @ref schedule method of the strategy
+     * @param scheduler the calling scheduler object
+     * @param queue pending scheduling requests
+     * @return {@code true} if @ref schedule should immediately be called
+     */
+    boolean scheduleImmediately(K8sTaskScheduler scheduler, List<K8sSchedulingRequest> queue)
 }

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy

@@ -12,18 +12,19 @@ class K8sTaskScheduler implements Runnable {
     private final K8sExecutor executor
     private final K8sSchedulingStrategy strategy
     private final LinkedBlockingQueue<K8sSchedulingRequest> queue = new LinkedBlockingQueue<>()
-    private List<String> cachedNodes
+    private String[] nodes
 
     private synchronized boolean shouldStop
 
     private HashMap<String, ArrayList<K8sTaskHandler>> occupancy;
     private HashMap<String, String> taskToNode
 
-    K8sTaskScheduler(K8sExecutor executor, K8sSchedulingStrategy strategy) {
+    K8sTaskScheduler(String[] nodes, K8sSchedulingStrategy strategy) {
         this.executor = executor
         this.strategy = strategy
+        this.nodes = nodes
         this.occupancy = new HashMap<>()
-        for (String node : getNodes()) {
+        for (String node : nodes) {
             this.occupancy.put(node, new ArrayList<K8sTaskHandler>())
         }
         this.taskToNode = new HashMap<>()
@@ -36,8 +37,9 @@ class K8sTaskScheduler implements Runnable {
     void submit(K8sTaskHandler handler) {
         log.info "[K8s] received queued task ${handler.task.name}"
         queue.add(new K8sSchedulingRequest(handler))
-
-        /* TODO: Decide if we should invoke the scheduler immediately */
+        final pending = new ArrayList<K8sSchedulingRequest>(queue)
+        if (strategy.scheduleImmediately(this, pending))
+            drain()
     }
 
     /**
@@ -63,7 +65,7 @@ class K8sTaskScheduler implements Runnable {
     protected synchronized void drain() {
         while( true ) {
             final pending = new ArrayList<K8sSchedulingRequest>(queue)
-            final decision = strategy.schedule(this, pending, getNodes())
+            final decision = strategy.schedule(this, pending)
 
             if ( !decision )
                 return
@@ -86,6 +88,10 @@ class K8sTaskScheduler implements Runnable {
         unoccupied.keySet().toList()
     }
 
+    List<String> getNodes() {
+        return nodes.toList()
+    }
+
     /**
      * Run is the scheduler threads main function
      * */
@@ -106,20 +112,4 @@ class K8sTaskScheduler implements Runnable {
         log.info("[K8s] stopping scheduler loop")
         this.shouldStop = true
     }
-
-    protected List<String> getNodes() {
-        if ( cachedNodes == null )
-            cachedNodes = fetchNodes()
-        return cachedNodes
-    }
-
-    @CompileDynamic
-    private List<String> fetchNodes() {
-        final resp = executor.getClient().nodeList()
-        ArrayList<String> nodes = new ArrayList<String>()
-        for ( Map item : resp.items ) {
-            nodes.add(item.metadata.name as String)
-        }
-        return nodes
-    }
 }

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/strategies/K8sHashSchedulingStrategy.groovy

@@ -10,16 +10,24 @@ import nextflow.k8s.K8sTaskScheduler
 class K8sHashSchedulingStrategy implements K8sSchedulingStrategy {
 
     @Override
-    K8sSchedulingDecision schedule(K8sTaskScheduler scheduler, List<K8sSchedulingRequest> queue, List<String> nodes) {
-        if ( !queue || !nodes )
+    K8sSchedulingDecision schedule(K8sTaskScheduler scheduler, List<K8sSchedulingRequest> queue) {
+        if (!queue)
             return null
 
         final freeNodes = scheduler.freeNodes
-        if ( freeNodes.size() == 0 )
+        if (freeNodes.size() == 0)
             return null
 
         final request = queue[0]
         final index = Math.floorMod(request.task.hash.asInt(), freeNodes.size())
         return new K8sSchedulingDecision(request, freeNodes[index])
     }
+
+    @Override
+    boolean scheduleImmediately(K8sTaskScheduler scheduler, List<K8sSchedulingRequest> queue) {
+        if (!queue)
+            return false
+        final freeNodes = scheduler.freeNodes
+        return freeNodes.size() > 0
+    }
 }