scheduler: fix executor allocation for non-gang policies and gang batch semantics

Three interrelated bugs prevented correct executor allocation depending
on which scheduler plugins were configured.

Bug 1 — executors never allocated without gang (DRF-only, priority-only)

  PluginManager aggregated is_ready/is_fulfilled with unwrap_or(true),
  meaning when no plugin had a batch opinion the result was true.
  AllocateAction's pre-check (`if fulfilled || ready { skip }`) and
  DispatchAction's commit condition (`if fulfilled { commit }`) therefore
  skipped or no-op'd every session, leaving tasks permanently pending.

  Fix: replace bool aggregation with Option<bool> three-value logic —
  Some(false) = veto, Some(true) = approved, None = no opinion.
  Expose bool at the public API level by tracking per-session op counts
  (no_gang_alloc_ops / no_gang_bind_ops) inside PluginManager when no
  gang plugin is loaded. is_ready/is_fulfilled return false before the
  first op and true after, so callers use uniform `if is_ready() { break; }`
  without needing to know whether gang is configured.

Bug 2 — gang only ever created 1 executor regardless of task count

  The old readiness formula `total % batch_size == 0` was true at
  every multiple, so after one allocation (total=1, batch_size=1) the
  loop broke and no further executors were created for the remaining
  pending tasks.

  Fix: introduce `incomplete_tasks` in GangState (sampled once per cycle
  as Pending + Running task count) and compute:
    needed = (incomplete_tasks / batch_size) * batch_size
    ready  = needed > 0 && total == needed
  This creates exactly enough executors to cover all incomplete tasks in
  full batches per scheduling cycle.

Bug 3 — duplicate executor creation when Dispatch and Allocate ran in the
  same cycle for a gang session

  DispatchAction lacked a pre-session guard: a session whose executor
  was still in Binding state (counted in Gang's `allocated`) received
  additional bindings every cycle.

  Fix: check is_fulfilled before entering the bind loop; skip the session
  if it is already satisfied.

Additional changes:

- Remove "drf" from DEFAULT_POLICIES; DRF remains opt-in via
  cluster.policies. Default stack is now priority + gang + shim.
- Allow listing "shim" in cluster.policies without error (log warning,
  ignore); shim is always-on.
- Regression tests added for all affected configurations:
    · drf-only and priority-only: executor is allocated
    · gang batch_size=1: one executor per incomplete task per cycle
    · gang batch_size=2: exactly two executors per cycle
    · no-gang: at most one executor per cycle
    · dispatch without gang: idle executor is bound
    · plugin aggregation: None/Some(false)/Some(true) edge cases

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Zhe Jin

common/src/ctx.rs

@@ -27,9 +27,18 @@ const DEFAULT_FLAME_CONF: &str = "flame-cluster.yaml";
 const DEFAULT_CONTEXT_NAME: &str = "flame";
 const DEFAULT_FLAME_ENDPOINT: &str = "http://127.0.0.1:8080";
 /// Default policies to enable when none specified in config.
-/// Available configurable policies: "priority", "drf", "gang"
-/// Note: "shim" plugin is always enabled (required for executor matching)
-pub const DEFAULT_POLICIES: &[&str] = &["priority", "drf", "gang"];
+///
+/// Effective default scheduler stack: `priority + gang + shim`
+///   - "gang" is listed here so batch-size semantics and allocation guards are active
+///     by default.  Users who do not need gang scheduling may omit it; scheduling
+///     will still work but allocates one executor per cycle with no batch constraint.
+///   - "shim" is always-on (non-configurable) and runs regardless of this list.
+///
+/// Optional configurable policies (add to cluster.policies in your config to enable):
+///   - "drf" — Dominant Resource Fairness; enables fair multi-resource sharing across sessions.
+///
+/// Note: listing "shim" in the config is harmless — a warning is logged and it is ignored.
+pub const DEFAULT_POLICIES: &[&str] = &["priority", "gang"];
 const DEFAULT_STORAGE: &str = "sqlite://flame.db";
 const DEFAULT_MAX_EXECUTORS_PER_NODE: u32 = 128;
 pub const DEFAULT_SESSION_RETRY_LIMITS: u32 = 5;

session_manager/src/scheduler/actions/allocate.rs

@@ -118,8 +118,12 @@ impl Action for AllocateAction {
             // (see `Context`): Gang's counters include binds Dispatch committed via
             // `Statement` in this same `Context`. If those already satisfy gang scheduling, do not
             // create more executors here.
-            let fulfilled = ctx.is_fulfilled(&ssn)?;
-            let ready = ctx.is_ready(&ssn)?;
+            //
+            // `is_ready` / `is_fulfilled` return true once the batch is satisfied (gang) or
+            // after the first op this cycle (no gang). Skip only when one of them is already
+            // true before we begin.
+            let fulfilled = ctx.is_fulfilled(&ssn);
+            let ready = ctx.is_ready(&ssn);
             if fulfilled || ready {
                 tracing::debug!(
                     "Skip allocate resources for session <{}>: is_fulfilled={}, is_ready={}",
@@ -132,28 +136,34 @@ impl Action for AllocateAction {
 
             let mut stmt = Statement::new(ss.clone(), ctx.plugins.clone(), ctx.controller.clone());
 
-            let pipelineable = void_executors
+            let pipelineable: Vec<_> = void_executors
                 .values()
                 .chain(unbinding_executors.values())
-                .filter(|e| ctx.is_available(e, &ssn).unwrap_or(false));
+                .filter(|e| ctx.is_available(e, &ssn).unwrap_or(false))
+                .cloned()
+                .collect();
 
-            for exec in pipelineable {
+            for exec in &pipelineable {
                 stmt.pipeline(exec, &ssn)?;
-                if ctx.is_ready(&ssn)? {
+                if ctx.is_ready(&ssn) {
                     break;
                 }
             }
 
-            for node in nodes.iter() {
-                if ctx.is_ready(&ssn)? {
+            'node_loop: for node in nodes.iter() {
+                if ctx.is_ready(&ssn) {
                     break;
                 }
-                while ctx.is_allocatable(node, &ssn)? && !ctx.is_ready(&ssn)? {
+                while ctx.is_allocatable(node, &ssn)? {
                     stmt.allocate(node, &ssn)?;
+                    if ctx.is_ready(&ssn) {
+                        break 'node_loop;
+                    }
                 }
             }
 
-            if ctx.is_ready(&ssn)? {
+            let ready_after = ctx.is_ready(&ssn);
+            if ready_after && !stmt.is_empty() {
                 let op_count = stmt.len();
                 let pipelined_ids = stmt.commit().await?;
                 for id in &pipelined_ids {
@@ -162,7 +172,13 @@ impl Action for AllocateAction {
                 }
                 tracing::debug!("Committed {} op(s) for session <{}>", op_count, ssn.id);
                 nodes.sort_by(|a, b| node_order_fn.cmp(a, b));
-                open_ssns.push(ssn.clone());
+                // Re-queue when the batch is complete so remaining pipeline slots can be
+                // filled in the same cycle (gang with multiple tasks). With no gang,
+                // `ready_after` is also true after 1 op, but the pre-check above will
+                // immediately skip the session on re-entry, bounding allocation to 1 op.
+                if ready_after {
+                    open_ssns.push(ssn.clone());
+                }
             } else if !stmt.is_empty() {
                 tracing::debug!(
                     "Discarding incomplete batch for session <{}>: not enough resources",

session_manager/src/scheduler/actions/dispatch.rs

@@ -65,6 +65,21 @@ impl Action for DispatchAction {
                 continue;
             }
 
+            // Skip sessions that already have enough executors committed (or in-flight from a
+            // previous cycle).  Without this guard a session whose executor is still running
+            // `on_session_enter` (Binding state, counted in Gang's `allocated`) would receive
+            // additional bindings every scheduling cycle.
+            //
+            // `is_fulfilled` returns true once the batch is satisfied (gang) or after the
+            // first bind this cycle (no gang), so this guard works uniformly for both cases.
+            if ctx.is_fulfilled(&ssn) {
+                tracing::debug!(
+                    "Session <{}> is already fulfilled (existing executors satisfy batch_size), skip dispatch.",
+                    ssn.id
+                );
+                continue;
+            }
+
             tracing::debug!(
                 "Session <{}> is underused, start to allocate resources.",
                 &ssn.id
@@ -75,13 +90,13 @@ impl Action for DispatchAction {
             for (_, exec) in idle_executors.iter() {
                 if ctx.is_available(exec, &ssn)? {
                     stmt.bind(exec, &ssn)?;
-                    if ctx.is_fulfilled(&ssn)? {
+                    if ctx.is_fulfilled(&ssn) {
                         break;
                     }
                 }
             }
 
-            if ctx.is_fulfilled(&ssn)? {
+            if ctx.is_fulfilled(&ssn) && !stmt.is_empty() {
                 tracing::debug!("Bind executor for session <{}>.", ssn.id);
                 let bound_ids = stmt.commit().await?;
                 for id in &bound_ids {
@@ -97,7 +112,7 @@ impl Action for DispatchAction {
                 continue;
             } else if !stmt.is_empty() {
                 tracing::debug!(
-                    "Discarding unfulfilled binding for session <{}>: no available idle executors",
+                    "Discarding unfulfilled binding for session <{}>: gang batch incomplete",
                     ssn.id
                 );
                 stmt.discard()?;

session_manager/src/scheduler/ctx.rs

@@ -77,13 +77,21 @@ impl Context {
     /// Allocation-side batch readiness (e.g. Gang: pipelined + allocated executors form full
     /// batches). Reflects in-memory plugin state, including `Statement` ops earlier in this
     /// same cycle (typically pipeline/allocate before commit).
-    pub fn is_ready(&self, ssn: &SessionInfoPtr) -> Result<bool, FlameError> {
+    ///
+    /// Returns `true` when the batch is complete (gang) or after the first pipeline/allocate
+    /// op in a no-gang cycle.  Returns `false` before any op has been attempted, or when gang
+    /// reports the batch is still incomplete.
+    pub fn is_ready(&self, ssn: &SessionInfoPtr) -> bool {
         self.plugins.is_ready(ssn)
     }
 
     /// Binding-side batch readiness (e.g. Gang: bound + on-session executors form full batches).
-    /// After Dispatch commits binds, this can be true so Allocate skips provisioning.
-    pub fn is_fulfilled(&self, ssn: &SessionInfoPtr) -> Result<bool, FlameError> {
+    /// After Dispatch commits binds, this can be `true` so Allocate skips provisioning.
+    ///
+    /// Returns `true` when the batch is fulfilled (gang) or after the first bind op in a
+    /// no-gang cycle.  Returns `false` before any bind has been attempted, or when gang
+    /// reports the batch is still incomplete.
+    pub fn is_fulfilled(&self, ssn: &SessionInfoPtr) -> bool {
         self.plugins.is_fulfilled(ssn)
     }