Hold a worker for a queued action whose reservation only fits an idle one

nativelink-config/src/schedulers.rs

@@ -345,6 +345,26 @@ pub struct SimpleSpec {
     #[serde(default)]
     pub memory_escalation: Option<MemoryEscalationSpec>,
 
+    /// Hold a worker for a queued action that keeps finding no room. An
+    /// action whose reservation only fits an idle worker starves under
+    /// load: every pass tries it first and finds each worker partly busy,
+    /// then fills the room that opens with newer, smaller actions, so no
+    /// worker ever drains for it. With this set, once such an action has
+    /// waited `hold_after_s` and fits no busy worker until more than one
+    /// of that worker's actions finish (one finishing would make room for
+    /// an action that is only waiting its turn), the scheduler picks a
+    /// worker that would fit it idle and offers it nothing else until a
+    /// worker takes the action, it leaves the queue, or `max_hold_s`
+    /// passes. An action no connected worker would fit even idle gets no
+    /// hold. Each scheduler replica holds only workers connected to it;
+    /// replicas sharing a queue take turns per action, by a hash of the
+    /// action over the replicas whose workers would fit it as the fleet
+    /// census last read them, so one worker is held for it fleet-wide.
+    /// While a replica cannot read the census it holds on its own.
+    /// Default: unset (off)
+    #[serde(default)]
+    pub reservation_hold: Option<ReservationHoldSpec>,
+
     /// The storage backend to use for the scheduler.
     /// Default: memory
     pub experimental_backend: Option<ExperimentalSimpleSchedulerBackend>,
@@ -571,6 +591,58 @@ const fn default_memory_escalation_percent() -> u64 {
     200
 }
 
+#[derive(Deserialize, Serialize, Debug, Clone, Copy)]
+#[serde(deny_unknown_fields)]
+#[cfg_attr(feature = "dev-schema", derive(JsonSchema))]
+pub struct ReservationHoldSpec {
+    /// Seconds a queued action has to have waited, finding no room on
+    /// every pass, before a worker is held for it. Must be above zero.
+    /// Default: 30
+    #[serde(
+        default = "default_reservation_hold_after_s",
+        deserialize_with = "convert_duration_with_shellexpand"
+    )]
+    pub hold_after_s: u64,
+
+    /// Seconds a hold lasts at most, capped at a week. A hold that long
+    /// without the action fitting means the worker never reported room for
+    /// it even drained; the hold is given up so the worker is not idle for
+    /// good, and the action is not held for again until as long has
+    /// passed, twice as long for each such expiry in a row on a worker
+    /// that was idle (up to sixteen times), since then nothing kept the
+    /// action off the worker but the action itself. Must be above zero.
+    /// Default: 300
+    #[serde(
+        default = "default_reservation_max_hold_s",
+        deserialize_with = "convert_duration_with_shellexpand"
+    )]
+    pub max_hold_s: u64,
+
+    /// Most workers held at once on this scheduler replica. 0 is one tenth
+    /// of the workers connected to it, and at least one.
+    /// Default: 0
+    #[serde(default, deserialize_with = "convert_numeric_with_shellexpand")]
+    pub max_held_workers: usize,
+}
+
+impl Default for ReservationHoldSpec {
+    fn default() -> Self {
+        Self {
+            hold_after_s: default_reservation_hold_after_s(),
+            max_hold_s: default_reservation_max_hold_s(),
+            max_held_workers: 0,
+        }
+    }
+}
+
+const fn default_reservation_hold_after_s() -> u64 {
+    30
+}
+
+const fn default_reservation_max_hold_s() -> u64 {
+    300
+}
+
 #[derive(Deserialize, Serialize, Debug, Default, Clone, Copy)]
 #[serde(deny_unknown_fields)]
 #[cfg_attr(feature = "dev-schema", derive(JsonSchema))]

nativelink-scheduler/BUILD.bazel

@@ -86,6 +86,7 @@ rust_test_suite(
         "tests/property_modifier_scheduler_test.rs",
         "tests/queue_order_test.rs",
         "tests/redis_store_awaited_action_db_test.rs",
+        "tests/reservation_hold_test.rs",
         "tests/simple_scheduler_state_manager_test.rs",
         "tests/simple_scheduler_test.rs",
         "tests/store_awaited_action_db_test.rs",