top-k

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

@@ -260,6 +260,18 @@ class K8sConfig implements ConfigScope {
     """)
     final Duration noiseRuntimeEstimatorNoiseMagnitude
 
+    @ConfigOption
+    @Description("""
+        Max. amount of time two runtime estimates can differ to be considered equal.
+    """)
+    final Duration runtimeComparisonEpsilon
+
+    @ConfigOption
+    @Description("""
+        Number of saved top runtimes used to classify a task as critical.
+    """)
+    final int dvfsSchedulingNumTopRuntimes
+
     /* required by extension point -- do not remove */
     K8sConfig() {
         this(Collections.emptyMap())
@@ -300,7 +312,9 @@ class K8sConfig implements ConfigScope {
 
         schedulingStrategy = opts.schedulingStrategy as String ?: "Hash"
         runtimeEstimator = opts.runtimeEstimator as String ?: "LinearFit"
-        noiseRuntimeEstimatorNoiseMagnitude = opts.noiseRuntimeEstimatorNoiseMagnitude as Duration ?: new Duration(10, TimeUnit.SECONDS)
+        noiseRuntimeEstimatorNoiseMagnitude = opts.noiseRuntimeEstimatorNoiseMagnitude as Duration ?: new Duration(30, TimeUnit.SECONDS)
+        runtimeComparisonEpsilon = opts.runtimeComparisonEpsilon as Duration ?: new Duration(10, TimeUnit.SECONDS)
+        dvfsSchedulingNumTopRuntimes = opts.dvfsSchedulingTopRuntimes as int ?: 3
 
         // -- shortcut to pod image pull-policy
         if( imagePullPolicy )

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

@@ -112,7 +112,7 @@ class K8sExecutor extends Executor implements ExtensionPoint {
         K8sSchedulingStrategy strategy = null
         if (k8sConfig.schedulingStrategy == "Hash") {
             strategy = new K8sHashSchedulingStrategy()
-        } else if (k8sConfig.schedulingStrategy == "DVFS") {
+        } else if (k8sConfig.schedulingStrategy == "DVFS" || k8sConfig.schedulingStrategy == "DVFS-SPEED") {
             String[] ips = new String[nodes.length]
             for (int i = 0; i < nodes.length; i++) {
                 ips[i] = client.getPodIpAddress(K8sNodeInitDeployer.buildPodName(nodes[i]))
@@ -120,18 +120,10 @@ class K8sExecutor extends Executor implements ExtensionPoint {
             }
             strategy = new K8sDVFSSchedulingStrategy(this.runtimeEstimator,
                 new K8sDVFSClient(nodes, ips),
-                () -> getClient())
-            strategy.fullSpeedMode = false
-        } else if (k8sConfig.schedulingStrategy == "DVFS-SPEED") {
-            String[] ips = new String[nodes.length]
-            for (int i = 0; i < nodes.length; i++) {
-                ips[i] = client.getPodIpAddress(K8sNodeInitDeployer.buildPodName(nodes[i]))
-                log.info "[K8s] node ${nodes[i]} -> ${ips[i]}"
-            }
-            strategy = new K8sDVFSSchedulingStrategy(this.runtimeEstimator,
-                new K8sDVFSClient(nodes, ips),
-                () -> getClient())
-            strategy.fullSpeedMode = true
+                () -> getClient(),
+                k8sConfig.runtimeComparisonEpsilon,
+                k8sConfig.dvfsSchedulingNumTopRuntimes)
+            strategy.fullSpeedMode = k8sConfig.schedulingStrategy == "DVFS-SPEED"
         } else {
             log.error "[K8s] invalid scheduling strategy $k8sConfig.schedulingStrategy, falling back on \"Hash\""
             strategy = new K8sHashSchedulingStrategy()

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

@@ -11,7 +11,7 @@ import nextflow.k8s.K8sTaskHandler
 import nextflow.k8s.K8sTaskScheduler
 import nextflow.k8s.client.K8sClient
 import nextflow.processor.TaskRun
-import nextflow.util.ArrayTuple
+import nextflow.util.Duration
 
 /**
  * Implements a scheduling strategy utilizing dvfs to reduce the energy consumption
@@ -142,16 +142,17 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
     class SchedulingRequestComparator implements Comparator<K8sSchedulingRequest> {
         K8sRuntimeEstimator runtimeEstimator
         long currentTime
-        double avgRuntime
+        double epsilon
 
         @Override
         int compare(K8sSchedulingRequest o1, K8sSchedulingRequest o2) {
             // First, check if one of the tasks is (estimated to be) on the critical path
             double t1 = runtimeEstimator.estimate(o1.handler)
             double t2 = runtimeEstimator.estimate(o2.handler)
-            if (t1 > avgRuntime && t2 <= avgRuntime)
+
+            if (t1 > t2 + epsilon)
                 return -1
-            else if (t1 < avgRuntime && t2 > avgRuntime)
+            else if (t2 > t1 + epsilon)
                 return 1
 
             // Both are not on the critical path. Sort based on the time they spent in the queue
@@ -170,23 +171,57 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
 
     private ArrayList<WorkerNode> nodes
     private HashMap<String, WorkerNode> taskToNode
-
-    private double averageRuntime
-    private long finishedTaskCount
-
     private long globalMaxFrequency
     private long globalMinFrequency
 
     private K8sClientGetter clientGetter
 
+    private double comparisonEpsilonMillis
+
+    private double[] topRuntimes
+    private double averageRuntime
+    private double finishedTaskCount
+
     boolean fullSpeedMode
 
-    K8sDVFSSchedulingStrategy(K8sRuntimeEstimator runtimeEstimator, K8sDVFSClient dvfsClient, K8sClientGetter clientGetter) {
+    K8sDVFSSchedulingStrategy(K8sRuntimeEstimator runtimeEstimator,
+                              K8sDVFSClient dvfsClient,
+                              K8sClientGetter clientGetter,
+                              Duration runtimeComparisonEpsilon,
+                              int topRuntimeCount) {
         this.runtimeEstimator = runtimeEstimator
         this.dvfsClient = dvfsClient
         this.nodes = new ArrayList<>()
         this.taskToNode = new HashMap<>();
         this.clientGetter = clientGetter
+        this.comparisonEpsilonMillis = (double)runtimeComparisonEpsilon.toMillis()
+        this.topRuntimes = new double[topRuntimeCount]
+        for (int i = 0; i < topRuntimeCount; i++) {
+            this.topRuntimes[i] = 0.0
+        }
+        this.averageRuntime = 0.0
+        this.finishedTaskCount = 0.0
+    }
+
+    private boolean isInTopRuntimes(double rt) {
+        for (int i = 0; i < topRuntimes.size(); i++) {
+            if (rt >= topRuntimes[i])
+                return true
+        }
+        return false
+    }
+
+    private void updateTopRuntimes(double rt) {
+        for (int i = 0; i < topRuntimes.size(); i++) {
+            if (rt > topRuntimes[i]) {
+                /* Move all one down */
+                for (int j = topRuntimes.size() - 1; j > i; j--) {
+                    topRuntimes[j] = topRuntimes[j - 1];
+                }
+                topRuntimes[i] = rt
+                break
+            }
+        }
     }
 
     @Override
@@ -199,11 +234,11 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
         /* Step 1: Sort by task priority. We will attempt to schedule tasks "in order", so that the
          * highest priority tasks are assigned to nodes as soon as possible.
          *
-         * Priority is based on a) an estimation if the task is on the critical path and b) the wait time of the task.
+         * Priority is based on a) the tasks estimated runtime and b) the wait time of the task.
          */
         SchedulingRequestComparator comparator = new SchedulingRequestComparator()
         comparator.runtimeEstimator = runtimeEstimator
-        comparator.avgRuntime = averageRuntime
+        comparator.epsilon = comparisonEpsilonMillis
         comparator.currentTime = System.currentTimeMillis()
         queue.sort(comparator)
 
@@ -214,7 +249,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
              * frequency. If not, we determine a frequency (see below).
              */
             final double taskEstimation = runtimeEstimator.estimate(req.handler)
-            final boolean isCriticalPath = taskEstimation > averageRuntime
+            final boolean isCriticalPath = isInTopRuntimes(taskEstimation)
             long frequency = globalMaxFrequency
             if (!isCriticalPath && !fullSpeedMode) {
                 /* Set frequency so that we expect the runtime to be close to the mean runtime. */
@@ -280,8 +315,9 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
          * elapsed time based on that.
          */
         double runtime = (double)(task.getCompleteTimeMillis() - task.getStartTimeMillis())
-        averageRuntime = (runtime + finishedTaskCount * averageRuntime) / (finishedTaskCount + 1)
-        finishedTaskCount += 1
+        averageRuntime = (runtime + finishedTaskCount * averageRuntime) / (finishedTaskCount + 1.0)
+        finishedTaskCount += 1.0
+        updateTopRuntimes(runtime)
 
         /* Free resources allocated by this task */
         WorkerNode node = taskToNode.get(task.task.hash.toString())