diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy index 79dccb3..5de3fe8 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy +++ b/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 nodes = new ArrayList() + for ( Map item : resp.items ) { + nodes.add(item.metadata.name as String) + } + return nodes.toArray() + } + @Override void shutdown() { this.runtimeRecorder.write() diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sSchedulingStrategy.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sSchedulingStrategy.groovy index 78e19e7..822ef73 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sSchedulingStrategy.groovy +++ b/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 queue, List nodes) + K8sSchedulingDecision schedule(K8sTaskScheduler scheduler, List 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 queue) } diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy index c74f5b0..99cdbd0 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy +++ b/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 queue = new LinkedBlockingQueue<>() - private List cachedNodes + private String[] nodes private synchronized boolean shouldStop private HashMap> occupancy; private HashMap 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()) } 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(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(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 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 getNodes() { - if ( cachedNodes == null ) - cachedNodes = fetchNodes() - return cachedNodes - } - - @CompileDynamic - private List fetchNodes() { - final resp = executor.getClient().nodeList() - ArrayList nodes = new ArrayList() - for ( Map item : resp.items ) { - nodes.add(item.metadata.name as String) - } - return nodes - } } diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/strategies/K8sHashSchedulingStrategy.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/strategies/K8sHashSchedulingStrategy.groovy index 6ad559c..14769cd 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/strategies/K8sHashSchedulingStrategy.groovy +++ b/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 queue, List nodes) { - if ( !queue || !nodes ) + K8sSchedulingDecision schedule(K8sTaskScheduler scheduler, List 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 queue) { + if (!queue) + return false + final freeNodes = scheduler.freeNodes + return freeNodes.size() > 0 + } }