From ac0dd7743c460b8d23ad0462fe1e03d78910ceda Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Thu, 28 May 2026 22:35:41 -0400 Subject: [PATCH 1/3] WIP --- Cargo.lock | 10 ++++ Cargo.toml | 1 + components/spider-scheduler/Cargo.toml | 14 +++++ components/spider-scheduler/src/core.rs | 56 ++++++++++++++++++ components/spider-scheduler/src/dispatch.rs | 28 +++++++++ components/spider-scheduler/src/error.rs | 25 ++++++++ components/spider-scheduler/src/lib.rs | 21 +++++++ .../spider-scheduler/src/storage_client.rs | 58 +++++++++++++++++++ components/spider-scheduler/src/types.rs | 35 +++++++++++ 9 files changed, 248 insertions(+) create mode 100644 components/spider-scheduler/Cargo.toml create mode 100644 components/spider-scheduler/src/core.rs create mode 100644 components/spider-scheduler/src/dispatch.rs create mode 100644 components/spider-scheduler/src/error.rs create mode 100644 components/spider-scheduler/src/lib.rs create mode 100644 components/spider-scheduler/src/storage_client.rs create mode 100644 components/spider-scheduler/src/types.rs diff --git a/Cargo.lock b/Cargo.lock index 2888d5e8f..0262e8d3f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1499,6 +1499,16 @@ dependencies = [ "tracing", ] +[[package]] +name = "spider-scheduler" +version = "0.1.0" +dependencies = [ + "async-trait", + "spider-core", + "thiserror", + "tokio-util", +] + [[package]] name = "spider-storage" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index ea9992cf3..4d8d20e8a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "components/spider-core", "components/spider-derive", "components/spider-execution-manager", + "components/spider-scheduler", "components/spider-storage", "components/spider-task-executor", "components/spider-tdl", diff --git a/components/spider-scheduler/Cargo.toml b/components/spider-scheduler/Cargo.toml new file mode 100644 index 000000000..59e9b8f7b --- /dev/null +++ b/components/spider-scheduler/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "spider-scheduler" +version = "0.1.0" +edition = "2024" + +[lib] +name = "spider_scheduler" +path = "src/lib.rs" + +[dependencies] +async-trait = "0.1.89" +spider-core = { path = "../spider-core" } +thiserror = "2.0.18" +tokio-util = "0.7.18" diff --git a/components/spider-scheduler/src/core.rs b/components/spider-scheduler/src/core.rs new file mode 100644 index 000000000..95392f49c --- /dev/null +++ b/components/spider-scheduler/src/core.rs @@ -0,0 +1,56 @@ +use std::sync::Arc; + +use async_trait::async_trait; + +use crate::{ + dispatch::DispatchSink, + error::SchedulerError, + storage_client::SchedulerStorageClient, +}; + +/// A cancellation handle used to signal a running [`SchedulerCore`] to stop. +/// +/// Cancelling the token causes [`SchedulerCore::run`] to break out of its scheduling loop and +/// return. +pub type ShutdownToken = tokio_util::sync::CancellationToken; + +/// A pluggable scheduling algorithm. +/// +/// A core owns its decision loop: it polls the inbound queue through a [`SchedulerStorageClient`], +/// applies its algorithm (reading storage as needed for placement), and writes assignments to a +/// [`DispatchSink`]. Modeling the algorithm as a trait lets different scheduling strategies share +/// the same runtime entry point. +#[async_trait] +pub trait SchedulerCore: Send { + /// The storage client the core polls and reads for placement decisions. + type Storage: SchedulerStorageClient; + + /// The dispatch sink the core writes assignments to. + type Sink: DispatchSink; + + /// Runs the scheduling loop until `shutdown` is triggered. + /// + /// The core polls the inbound queue through `storage`, applies its scheduling algorithm, and + /// writes assignments to `sink`, repeating until `shutdown` is cancelled, at which point it + /// returns. + /// + /// # Parameters + /// + /// * `storage` - The storage client used to poll the inbound queue and read state for + /// placement. + /// * `sink` - The dispatch sink that assignments are written to. + /// * `shutdown` - The token that, once cancelled, signals the loop to stop and return. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`SchedulerError`] if the scheduling loop fails irrecoverably, e.g. the storage client or + /// dispatch sink fails. + async fn run( + &mut self, + storage: Arc, + sink: Arc, + shutdown: ShutdownToken, + ) -> Result<(), SchedulerError>; +} diff --git a/components/spider-scheduler/src/dispatch.rs b/components/spider-scheduler/src/dispatch.rs new file mode 100644 index 000000000..c432d73a2 --- /dev/null +++ b/components/spider-scheduler/src/dispatch.rs @@ -0,0 +1,28 @@ +use async_trait::async_trait; + +use crate::{error::SchedulerError, types::TaskAssignment}; + +/// The write side of the dispatching queue used by the scheduler core. +/// +/// Modeled as a trait so the scheduler core can be unit-tested against a recording sink without +/// standing up the execution-manager-facing service. The production implementation is backed by a +/// bounded single-producer/multi-consumer queue. +#[async_trait] +pub trait DispatchSink: Send + Sync { + /// Enqueues a task assignment for execution managers to consume. + /// + /// Implementations backed by a bounded queue await while the queue is full, applying + /// back-pressure to the scheduler core. + /// + /// # Parameters + /// + /// * `assignment` - The task assignment to enqueue. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`SchedulerError::DispatchClosed`] if the dispatching queue is closed and can no longer + /// accept assignments. + async fn dispatch(&self, assignment: TaskAssignment) -> Result<(), SchedulerError>; +} diff --git a/components/spider-scheduler/src/error.rs b/components/spider-scheduler/src/error.rs new file mode 100644 index 000000000..8e7983d56 --- /dev/null +++ b/components/spider-scheduler/src/error.rs @@ -0,0 +1,25 @@ +use spider_core::types::id::JobId; + +/// Errors returned by [`crate::storage_client::SchedulerStorageClient`] operations. +#[derive(Debug, thiserror::Error)] +pub enum StorageClientError { + /// The inbound queue is closed and can no longer yield ready entries. + #[error("inbound queue is closed")] + InboundClosed, + + /// No job with the requested identifier exists. + #[error("job not found: {0:?}")] + JobNotFound(JobId), +} + +/// Errors returned by the scheduler runtime and its components. +#[derive(Debug, thiserror::Error)] +pub enum SchedulerError { + /// Forwarded from the storage client. + #[error(transparent)] + Storage(#[from] StorageClientError), + + /// The dispatching queue is closed and can no longer accept assignments. + #[error("dispatching queue is closed")] + DispatchClosed, +} diff --git a/components/spider-scheduler/src/lib.rs b/components/spider-scheduler/src/lib.rs new file mode 100644 index 000000000..411b26f0c --- /dev/null +++ b/components/spider-scheduler/src/lib.rs @@ -0,0 +1,21 @@ +//! Scheduler skeleton for the Spider task-execution framework. +//! +//! This crate defines the core type and trait abstractions of the scheduler: the data types +//! exchanged with storage and execution managers ([`InboundEntry`], [`TaskAssignment`]), the +//! storage and dispatch seams ([`SchedulerStorageClient`], [`DispatchSink`]), and the pluggable +//! scheduling algorithm ([`SchedulerCore`]). Concrete implementations (the dispatch queue, the +//! runtime, and scheduling algorithms) build on top of these abstractions. + +pub mod core; +pub mod dispatch; +pub mod error; +pub mod storage_client; +pub mod types; + +pub use crate::{ + core::{SchedulerCore, ShutdownToken}, + dispatch::DispatchSink, + error::{SchedulerError, StorageClientError}, + storage_client::SchedulerStorageClient, + types::{InboundEntry, TaskAssignment}, +}; diff --git a/components/spider-scheduler/src/storage_client.rs b/components/spider-scheduler/src/storage_client.rs new file mode 100644 index 000000000..5b847c639 --- /dev/null +++ b/components/spider-scheduler/src/storage_client.rs @@ -0,0 +1,58 @@ +use std::time::Duration; + +use async_trait::async_trait; +use spider_core::{job::JobState, types::id::JobId}; + +use crate::{error::StorageClientError, types::InboundEntry}; + +/// The scheduler's view of the storage layer. +/// +/// Abstracts the storage-owned inbound queue and the read-only queries a scheduling algorithm +/// needs to make placement decisions. Modeled as a trait so the scheduler runtime can be driven by +/// a real storage client in production or a mock in tests. +#[async_trait] +pub trait SchedulerStorageClient: Send + Sync { + /// Polls the storage-owned inbound (ready) queue for newly-ready tasks. + /// + /// Drains up to `max_items` ready entries across all storage lanes (regular, commit, and + /// cleanup tasks), blocking for at most `wait`. Returns an empty vector if no entry becomes + /// ready within `wait`. + /// + /// # Parameters + /// + /// * `max_items` - The maximum number of entries to return from a single poll. + /// * `wait` - The maximum duration to block waiting for ready entries. + /// + /// # Returns + /// + /// The ready entries drained from the inbound queue on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`StorageClientError::InboundClosed`] if the inbound queue is closed and can no longer + /// yield entries. + async fn poll_ready( + &self, + max_items: usize, + wait: Duration, + ) -> Result, StorageClientError>; + + /// Reads the current state of a job. + /// + /// # Parameters + /// + /// * `job_id` - The identifier of the job to query. + /// + /// # Returns + /// + /// The job's current [`JobState`] on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`StorageClientError::JobNotFound`] if no job with the given identifier exists. + async fn job_state(&self, job_id: JobId) -> Result; +} diff --git a/components/spider-scheduler/src/types.rs b/components/spider-scheduler/src/types.rs new file mode 100644 index 000000000..448a398b7 --- /dev/null +++ b/components/spider-scheduler/src/types.rs @@ -0,0 +1,35 @@ +use spider_core::types::id::{JobId, ResourceGroupId, TaskId}; + +/// A ready task drained from the storage-owned inbound queue. +/// +/// The storage client flattens storage's three ready lanes (regular, commit, and cleanup tasks) +/// into this uniform entry, resolving each to its [`TaskId`] so the scheduler core can treat every +/// ready task identically. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct InboundEntry { + /// The resource group that owns the job. + pub resource_group_id: ResourceGroupId, + + /// The job the task belongs to. + pub job_id: JobId, + + /// The ready task. + pub task_id: TaskId, +} + +/// A task placement decision written by the scheduler core to the dispatching queue. +/// +/// Assignments are intentionally lightweight: they identify the task but carry no inputs. The +/// consuming execution manager registers the task instance against storage on pull to obtain the +/// execution context (inputs, timeouts, and the TDL context). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TaskAssignment { + /// The resource group that owns the job. + pub resource_group_id: ResourceGroupId, + + /// The job the task belongs to. + pub job_id: JobId, + + /// The task to dispatch. + pub task_id: TaskId, +} From dced92eace056244a39c7dad83deefe263e8d392 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 29 May 2026 18:43:45 -0400 Subject: [PATCH 2/3] Done. --- components/spider-scheduler/src/core.rs | 45 +++++------ components/spider-scheduler/src/dispatch.rs | 28 ------- .../spider-scheduler/src/dispatch_queue.rs | 59 ++++++++++++++ components/spider-scheduler/src/error.rs | 10 ++- components/spider-scheduler/src/lib.rs | 43 +++++++--- .../spider-scheduler/src/storage_client.rs | 78 ++++++++++++++++--- components/spider-scheduler/src/types.rs | 6 +- 7 files changed, 188 insertions(+), 81 deletions(-) delete mode 100644 components/spider-scheduler/src/dispatch.rs create mode 100644 components/spider-scheduler/src/dispatch_queue.rs diff --git a/components/spider-scheduler/src/core.rs b/components/spider-scheduler/src/core.rs index 95392f49c..c6bb661c0 100644 --- a/components/spider-scheduler/src/core.rs +++ b/components/spider-scheduler/src/core.rs @@ -1,56 +1,47 @@ -use std::sync::Arc; +//! The abstract core of a Spider scheduler. use async_trait::async_trait; use crate::{ - dispatch::DispatchSink, + dispatch_queue::DispatchQueueSink, error::SchedulerError, storage_client::SchedulerStorageClient, }; -/// A cancellation handle used to signal a running [`SchedulerCore`] to stop. -/// -/// Cancelling the token causes [`SchedulerCore::run`] to break out of its scheduling loop and -/// return. -pub type ShutdownToken = tokio_util::sync::CancellationToken; - -/// A pluggable scheduling algorithm. +/// An abstracted core for a scheduling algorithm. /// /// A core owns its decision loop: it polls the inbound queue through a [`SchedulerStorageClient`], /// applies its algorithm (reading storage as needed for placement), and writes assignments to a -/// [`DispatchSink`]. Modeling the algorithm as a trait lets different scheduling strategies share -/// the same runtime entry point. +/// [`DispatchQueueSink`]. Modeling the algorithm as a trait lets different scheduling strategies +/// share the same runtime entry point. #[async_trait] pub trait SchedulerCore: Send { - /// The storage client the core polls and reads for placement decisions. - type Storage: SchedulerStorageClient; + /// The storage client used by the core to poll and read for placement decisions. + type StorageClient: SchedulerStorageClient; /// The dispatch sink the core writes assignments to. - type Sink: DispatchSink; + type Sink: DispatchQueueSink; - /// Runs the scheduling loop until `shutdown` is triggered. + /// Runs the scheduling loop until `cancellation_token` is triggered. /// - /// The core polls the inbound queue through `storage`, applies its scheduling algorithm, and - /// writes assignments to `sink`, repeating until `shutdown` is cancelled, at which point it - /// returns. + /// The core polls the inbound queue through `storage_client`, applies its scheduling algorithm, + /// and writes assignments to `sink`, repeating until `cancellation_token` is fired, at which + /// point it returns. /// /// # Parameters /// - /// * `storage` - The storage client used to poll the inbound queue and read state for + /// * `storage_client` - The storage client used to poll the inbound queue and read state for /// placement. /// * `sink` - The dispatch sink that assignments are written to. - /// * `shutdown` - The token that, once cancelled, signals the loop to stop and return. + /// * `cancellation_token` - The token to signal the scheduling loop to stop. /// /// # Errors /// - /// Returns an error if: - /// - /// * [`SchedulerError`] if the scheduling loop fails irrecoverably, e.g. the storage client or - /// dispatch sink fails. + /// Returns a [`SchedulerError`] instance indicating an irrecoverable error. async fn run( &mut self, - storage: Arc, - sink: Arc, - shutdown: ShutdownToken, + storage_client: Self::StorageClient, + sink: Self::Sink, + cancellation_token: tokio_util::sync::CancellationToken, ) -> Result<(), SchedulerError>; } diff --git a/components/spider-scheduler/src/dispatch.rs b/components/spider-scheduler/src/dispatch.rs deleted file mode 100644 index c432d73a2..000000000 --- a/components/spider-scheduler/src/dispatch.rs +++ /dev/null @@ -1,28 +0,0 @@ -use async_trait::async_trait; - -use crate::{error::SchedulerError, types::TaskAssignment}; - -/// The write side of the dispatching queue used by the scheduler core. -/// -/// Modeled as a trait so the scheduler core can be unit-tested against a recording sink without -/// standing up the execution-manager-facing service. The production implementation is backed by a -/// bounded single-producer/multi-consumer queue. -#[async_trait] -pub trait DispatchSink: Send + Sync { - /// Enqueues a task assignment for execution managers to consume. - /// - /// Implementations backed by a bounded queue await while the queue is full, applying - /// back-pressure to the scheduler core. - /// - /// # Parameters - /// - /// * `assignment` - The task assignment to enqueue. - /// - /// # Errors - /// - /// Returns an error if: - /// - /// * [`SchedulerError::DispatchClosed`] if the dispatching queue is closed and can no longer - /// accept assignments. - async fn dispatch(&self, assignment: TaskAssignment) -> Result<(), SchedulerError>; -} diff --git a/components/spider-scheduler/src/dispatch_queue.rs b/components/spider-scheduler/src/dispatch_queue.rs new file mode 100644 index 000000000..7ef57fdc1 --- /dev/null +++ b/components/spider-scheduler/src/dispatch_queue.rs @@ -0,0 +1,59 @@ +//! The dispatching queue that decouples the scheduler core's placement decisions from the +//! execution-manager-facing service. + +use async_trait::async_trait; +use spider_core::types::id::SessionId; + +use crate::{error::SchedulerError, types::TaskAssignment}; + +/// The writer side of the dispatching queue used by the scheduler core. +#[async_trait] +pub trait DispatchQueueSink: Send + Sync + Clone { + /// Enqueues a task assignment for execution managers to consume. + /// + /// # Parameters + /// + /// * `assignment` - The task assignment to enqueue. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`SchedulerError::DispatchQueueClosed`] if the dispatching queue is closed. + async fn enqueue(&self, assignment: TaskAssignment) -> Result<(), SchedulerError>; + + /// Bumps the session ID and invalidates all queued task assignments. + /// + /// # Parameters + /// + /// * `new_session_id` - The new session ID. Must be greater than the current session ID. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`SchedulerError::DispatchQueueClosed`] if the dispatching queue is closed. + /// * [`SchedulerError::InvalidSessionId`] if the new session ID is not greater than the current + /// session ID. + async fn bump_session_id(&self, new_session_id: SessionId) -> Result<(), SchedulerError>; +} + +/// The reader side of the dispatching queue, drained by the execution-manager-facing service. +#[async_trait] +pub trait DispatchQueueSource: Send + Sync + Clone { + /// Dequeues the next task assignment for an execution manager to execute. + /// + /// # Returns + /// + /// A tuple on success, containing: + /// + /// * The storage session associated with the assignment. + /// * The next task assignment ready to execute. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`SchedulerError::DispatchQueueClosed`] if the dispatching queue is closed. + async fn dequeue(&self) -> Result<(SessionId, TaskAssignment), SchedulerError>; +} diff --git a/components/spider-scheduler/src/error.rs b/components/spider-scheduler/src/error.rs index 8e7983d56..6a852c46a 100644 --- a/components/spider-scheduler/src/error.rs +++ b/components/spider-scheduler/src/error.rs @@ -1,4 +1,6 @@ -use spider_core::types::id::JobId; +//! The error types used in this crate. + +use spider_core::types::id::{JobId, SessionId}; /// Errors returned by [`crate::storage_client::SchedulerStorageClient`] operations. #[derive(Debug, thiserror::Error)] @@ -21,5 +23,9 @@ pub enum SchedulerError { /// The dispatching queue is closed and can no longer accept assignments. #[error("dispatching queue is closed")] - DispatchClosed, + DispatchQueueClosed, + + /// The session ID is invalid. + #[error("invalid session ID: {0:?}")] + InvalidSessionId(SessionId), } diff --git a/components/spider-scheduler/src/lib.rs b/components/spider-scheduler/src/lib.rs index 411b26f0c..bddd07502 100644 --- a/components/spider-scheduler/src/lib.rs +++ b/components/spider-scheduler/src/lib.rs @@ -1,20 +1,45 @@ -//! Scheduler skeleton for the Spider task-execution framework. +//! Trait and type abstractions for the Spider scheduler. //! -//! This crate defines the core type and trait abstractions of the scheduler: the data types -//! exchanged with storage and execution managers ([`InboundEntry`], [`TaskAssignment`]), the -//! storage and dispatch seams ([`SchedulerStorageClient`], [`DispatchSink`]), and the pluggable -//! scheduling algorithm ([`SchedulerCore`]). Concrete implementations (the dispatch queue, the -//! runtime, and scheduling algorithms) build on top of these abstractions. +//! The scheduler is the serial decision maker that turns ready tasks discovered by the storage +//! layer into assignments for execution managers. It owns placement and ordering policy, not +//! dependency resolution: storage decides *what* is ready, and the scheduler decides *in what +//! order* and *with what throttling* ready tasks are offered to the fleet. +//! +//! The crate defines three trait seams wired into a single pipeline — a storage client that polls +//! the ready queue, a core that makes serial decisions, and a dispatching queue that fans those +//! decisions out to execution managers: +//! +//! ```text +//! storage ── authoritative ready queue (owned by the storage layer, not this crate) +//! │ +//! │ poll_ready / poll_commit_ready / poll_cleanup_ready (SchedulerStorageClient) +//! ▼ +//! ┌───────────────────┐ +//! │ SchedulerCore │ serial loop: poll → decide → enqueue +//! └───────────────────┘ +//! │ +//! │ enqueue (DispatchQueueSink — writer side) +//! ▼ +//! ┌───────────────────┐ +//! │ dispatch queue │ bounded SPMC; a full queue back-pressures the core +//! └───────────────────┘ +//! │ +//! │ dequeue (DispatchQueueSource — reader side) +//! ▼ +//! ┌───────────────────┐ +//! │ scheduler service │ ──▶ execution managers (concurrent fan-out) +//! └───────────────────┘ +//! ``` pub mod core; -pub mod dispatch; +pub mod dispatch_queue; pub mod error; pub mod storage_client; pub mod types; pub use crate::{ - core::{SchedulerCore, ShutdownToken}, - dispatch::DispatchSink, + core::SchedulerCore, + dispatch_queue::{DispatchQueueSink, DispatchQueueSource}, error::{SchedulerError, StorageClientError}, storage_client::SchedulerStorageClient, types::{InboundEntry, TaskAssignment}, diff --git a/components/spider-scheduler/src/storage_client.rs b/components/spider-scheduler/src/storage_client.rs index 5b847c639..9f7adaf49 100644 --- a/components/spider-scheduler/src/storage_client.rs +++ b/components/spider-scheduler/src/storage_client.rs @@ -1,7 +1,12 @@ +//! The scheduler's view of the storage layer, abstracting inbound polling and placement-time reads. + use std::time::Duration; use async_trait::async_trait; -use spider_core::{job::JobState, types::id::JobId}; +use spider_core::{ + job::JobState, + types::id::{JobId, SessionId}, +}; use crate::{error::StorageClientError, types::InboundEntry}; @@ -11,33 +16,84 @@ use crate::{error::StorageClientError, types::InboundEntry}; /// needs to make placement decisions. Modeled as a trait so the scheduler runtime can be driven by /// a real storage client in production or a mock in tests. #[async_trait] -pub trait SchedulerStorageClient: Send + Sync { - /// Polls the storage-owned inbound (ready) queue for newly-ready tasks. - /// - /// Drains up to `max_items` ready entries across all storage lanes (regular, commit, and - /// cleanup tasks), blocking for at most `wait`. Returns an empty vector if no entry becomes - /// ready within `wait`. +pub trait SchedulerStorageClient: Send + Sync + Clone { + /// Polls the regular-task lane of the storage-owned inbound queue for ready tasks. /// /// # Parameters /// /// * `max_items` - The maximum number of entries to return from a single poll. - /// * `wait` - The maximum duration to block waiting for ready entries. + /// * `wait` - The maximum duration to block waiting for ready entries on the storage side. /// /// # Returns /// - /// The ready entries drained from the inbound queue on success. + /// A tuple on success, containing: + /// + /// * The storage session the poll was served under. + /// * The ready regular tasks drained from the lane. /// /// # Errors /// /// Returns an error if: /// - /// * [`StorageClientError::InboundClosed`] if the inbound queue is closed and can no longer + /// * [`StorageClientError::InboundClosed`] if the regular-task lane is closed and can no longer /// yield entries. async fn poll_ready( &self, max_items: usize, wait: Duration, - ) -> Result, StorageClientError>; + ) -> Result<(SessionId, Vec), StorageClientError>; + + /// Polls the commit-task lane of the storage-owned inbound queue for ready tasks. + /// + /// # Parameters + /// + /// * `max_items` - The maximum number of entries to return from a single poll. + /// * `wait` - The maximum duration to block waiting for ready entries on the storage side. + /// + /// # Returns + /// + /// A tuple on success, containing: + /// + /// * The storage session the poll was served under. + /// * The ready commit tasks drained from the lane. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`StorageClientError::InboundClosed`] if the commit-task lane is closed and can no longer + /// yield entries. + async fn poll_commit_ready( + &self, + max_items: usize, + wait: Duration, + ) -> Result<(SessionId, Vec), StorageClientError>; + + /// Polls the cleanup-task lane of the storage-owned inbound queue for ready tasks. + /// + /// # Parameters + /// + /// * `max_items` - The maximum number of entries to return from a single poll. + /// * `wait` - The maximum duration to block waiting for ready entries on the storage side. + /// + /// # Returns + /// + /// A tuple on success, containing: + /// + /// * The storage session the poll was served under. + /// * The ready cleanup tasks drained from the lane. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`StorageClientError::InboundClosed`] if the cleanup-task lane is closed and can no longer + /// yield entries. + async fn poll_cleanup_ready( + &self, + max_items: usize, + wait: Duration, + ) -> Result<(SessionId, Vec), StorageClientError>; /// Reads the current state of a job. /// diff --git a/components/spider-scheduler/src/types.rs b/components/spider-scheduler/src/types.rs index 448a398b7..70267b156 100644 --- a/components/spider-scheduler/src/types.rs +++ b/components/spider-scheduler/src/types.rs @@ -1,3 +1,5 @@ +//! The data types the scheduler exchanges with the storage layer and execution managers. + use spider_core::types::id::{JobId, ResourceGroupId, TaskId}; /// A ready task drained from the storage-owned inbound queue. @@ -18,10 +20,6 @@ pub struct InboundEntry { } /// A task placement decision written by the scheduler core to the dispatching queue. -/// -/// Assignments are intentionally lightweight: they identify the task but carry no inputs. The -/// consuming execution manager registers the task instance against storage on pull to obtain the -/// execution context (inputs, timeouts, and the TDL context). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct TaskAssignment { /// The resource group that owns the job. From 9c436bd9d4efbd90ddec53860fbcb58342ed6a73 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sat, 30 May 2026 14:45:35 -0400 Subject: [PATCH 3/3] Update dispatch queue's trait. --- .../spider-scheduler/src/dispatch_queue.rs | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/components/spider-scheduler/src/dispatch_queue.rs b/components/spider-scheduler/src/dispatch_queue.rs index 7ef57fdc1..2305a64d3 100644 --- a/components/spider-scheduler/src/dispatch_queue.rs +++ b/components/spider-scheduler/src/dispatch_queue.rs @@ -1,6 +1,8 @@ //! The dispatching queue that decouples the scheduler core's placement decisions from the //! execution-manager-facing service. +use std::time::Duration; + use async_trait::async_trait; use spider_core::types::id::SessionId; @@ -36,6 +38,11 @@ pub trait DispatchQueueSink: Send + Sync + Clone { /// * [`SchedulerError::InvalidSessionId`] if the new session ID is not greater than the current /// session ID. async fn bump_session_id(&self, new_session_id: SessionId) -> Result<(), SchedulerError>; + + /// # Returns + /// + /// The current size of the dispatch queue. + fn size(&self) -> usize; } /// The reader side of the dispatching queue, drained by the execution-manager-facing service. @@ -43,9 +50,14 @@ pub trait DispatchQueueSink: Send + Sync + Clone { pub trait DispatchQueueSource: Send + Sync + Clone { /// Dequeues the next task assignment for an execution manager to execute. /// + /// # Parameters + /// + /// * `wait_time` - The maximum amount of time to wait for a task assignment. + /// /// # Returns /// - /// A tuple on success, containing: + /// `None` if no task assignment is available within the specified wait time, or a tuple + /// containing: /// /// * The storage session associated with the assignment. /// * The next task assignment ready to execute. @@ -55,5 +67,8 @@ pub trait DispatchQueueSource: Send + Sync + Clone { /// Returns an error if: /// /// * [`SchedulerError::DispatchQueueClosed`] if the dispatching queue is closed. - async fn dequeue(&self) -> Result<(SessionId, TaskAssignment), SchedulerError>; + async fn dequeue( + &self, + wait_time: Duration, + ) -> Result, SchedulerError>; }