Gap constrains + reservation fixed

CHANGELOG.md

@@ -6,13 +6,31 @@
 
 * The server scheduler now contains a safety limit for computation, configurable via `--scheduler-time-limit` (default: 5s)
 * Better scheduling policy (prefill) for heterogenous clusters
+* Workers held for a waiting high-priority task are now chosen per priority threshold, so fewer
+  workers are held back and more lower-priority work can run elsewhere
+* More precise computation of gap for low-priority tasks
+
+### Changes
+
+* HiGHs solver uses only a single thread
 
 ### Fixes
 
 * Fixed some occasional greedy backfilling in server scheduler + improvements in the reservation algorithm
 * Fixed server crash in a specific situation when an unschedulable high-priority task occurs
 * Fixed server crash caused by invalid handling of prefill
 * Fixed canceling of prefilled tasks
+* Fixed multi-node tasks that could be placed on workers from two different worker groups
+  when worker ids of the groups interleaved
+* A held worker is no longer switched to a different worker only because small tasks were placed
+  into its unusable remainder; the scheduler moves a hold only to a worker that is at least as free
+  for the waiting task, so held capacity is not repeatedly given away
+* Reservations: reservations respect priorities, and they no longer block resources
+  the waiting task could never use
+* Fixed wrong resource amounts in this computation for tasks with resource variants on workers
+  that lack some resource
+* Fixed constraining gap that could lead to priority inversion
+
 
 ## v0.26.2
 

crates/hyperqueue/src/server/autoalloc/process.rs

@@ -2502,7 +2502,10 @@ mod tests {
             let schedule_fn = self.schedule_fn.clone();
             let custom_state = self.custom_state.clone();
 
-            Box::pin(async move { (schedule_fn.get())(custom_state.clone(), worker_count).await })
+            Box::pin(async move {
+                let fut = (schedule_fn.get())(custom_state.clone(), worker_count);
+                fut.await
+            })
         }
 
         fn get_status_of_allocations(
@@ -2517,8 +2520,8 @@ mod tests {
             Box::pin(async move {
                 let mut result = Map::default();
                 for allocation_id in allocation_ids {
-                    let status =
-                        (status_fn.get())(custom_state.clone(), allocation_id.clone()).await;
+                    let fut = (status_fn.get())(custom_state.clone(), allocation_id.clone());
+                    let status = fut.await;
                     if let Some(status) = status.transpose() {
                         result.insert(allocation_id, status);
                     }
@@ -2535,7 +2538,10 @@ mod tests {
             let custom_state = self.custom_state.clone();
             let allocation_id = allocation.id.clone();
 
-            Box::pin(async move { (remove_fn.get())(custom_state.clone(), allocation_id).await })
+            Box::pin(async move {
+                let fut = (remove_fn.get())(custom_state.clone(), allocation_id);
+                fut.await
+            })
         }
     }
 

crates/hyperqueue/src/tests/server.rs

@@ -100,7 +100,9 @@ where
     // Propagate the errors
     if server_error.is_err() && test_error.is_err() {
         eprintln!("{server_error:?}");
-        test_error.expect("Test failed");
+        if let Err(error) = test_error {
+            panic!("Test failed: {error:?}");
+        }
     } else {
         test_error.expect("Test failed");
         server_error.expect("Server failed");

crates/tako/Cargo.toml

@@ -39,7 +39,7 @@ tempfile = { workspace = true }
 itertools = { workspace = true }
 
 highs = { version = "2.1", optional = true }
-microlp = { version = "0.4", optional = true }
+microlp = { version = "0.6", optional = true }
 coin_cbc = { version = "0.1", optional = true }
 
 hashbrown = { version = "0.17", features = ["serde", "inline-more"], default-features = false }

crates/tako/src/internal/scheduler/batches.rs

@@ -98,8 +98,14 @@ pub(crate) fn create_task_batches(
     let mut current: Vec<Option<_>> = iters.iter_mut().map(|it| it.next()).collect();
     let mut unique = None;
     let mut found = Vec::new();
+    // The sweep step at which each batch last consumed tasks. A batch needs a new cut only if
+    // some other batch consumed since then: otherwise the tasks of other requests above it are
+    // the same as at its previous cut, so the cut would only repeat that one with a bigger size.
+    let mut step = 0u32;
+    let mut last_step: Vec<Option<u32>> = vec![None; batches.len()];
 
     loop {
+        step += 1;
         found.clear();
         let mut highest_p = Priority::new(0);
         for (idx, c) in current.iter().enumerate() {
@@ -121,6 +127,7 @@ pub(crate) fn create_task_batches(
             let idx = found[0];
             let size = current[idx].unwrap().1;
             if unique == Some(idx) {
+                last_step[idx] = Some(step);
                 batches[idx].size += size;
                 if batches[idx].size > batches[idx].limit {
                     batches[idx].size = batches[idx].limit;
@@ -134,6 +141,14 @@ pub(crate) fn create_task_batches(
             break;
         } else {
             for idx in &found {
+                let since = last_step[*idx];
+                let changed = last_step
+                    .iter()
+                    .enumerate()
+                    .any(|(i, s)| i != *idx && s.is_some_and(|s| since.is_none_or(|l| s >= l)));
+                if !changed {
+                    continue;
+                }
                 let size = batches[*idx].size;
                 let higher_priorities: Vec<_> = batches
                     .iter_mut()
@@ -153,6 +168,7 @@ pub(crate) fn create_task_batches(
                 }
             }
             for idx in &found {
+                last_step[*idx] = Some(step);
                 batches[*idx].size += current[*idx].unwrap().1;
                 if batches[*idx].size > batches[*idx].limit {
                     batches[*idx].size = batches[*idx].limit;
@@ -249,6 +265,26 @@ fn prune_progressive<T>(vec: &mut Vec<T>, prefix_size: usize, size_limit: usize)
     vec.truncate(size_limit);
 }
 
+pub(crate) fn trim_to_unblocked(batches: &[TaskBatch]) -> Vec<TaskBatch> {
+    batches
+        .iter()
+        .filter_map(|batch| {
+            let size = batch.cuts.first().map_or(batch.size, |cut| cut.size);
+            if size == 0 {
+                return None;
+            }
+            Some(TaskBatch {
+                resource_rq_id: batch.resource_rq_id,
+                cuts: Vec::new(),
+                size,
+                limit: batch.limit,
+                limit_reached: false,
+                is_blocker: false,
+            })
+        })
+        .collect()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;