refactor(Echo/Scheduler): decouple queue from scheduler and optimize work-stealing

- Restructured `StealingQueue` and `Scheduler` modules to separate generic work-stealing logic from application-specific task handling, aligning with Land's architecture principle of decoupling components.
- Implemented a more efficient work-stealing algorithm using random peer iteration to reduce contention and improve task distribution efficiency.
- Updated architectural documentation in README and Deep Dive to reflect the decoupled design and priority-aware scheduling crucial for Mountain's `ActionEffect` execution.
- Revised code examples and API documentation to match the current `SchedulerBuilder` usage, ensuring consistency across the codebase.

This refactoring enhances the scheduler's performance and maintainability, directly supporting Mountain's need for reliable and efficient execution of declarative `ActionEffect` workflows in the Land editor backend.

Author: NikolaRHristov

README.md

@@ -41,31 +41,30 @@ executing complex asynchronous workflows with resilience and efficiency.
 
 ## Key Features 🔐
 
-- **Work-Stealing Scheduler:** Implements a modern work-stealing algorithm to
-  efficiently distribute tasks across a pool of worker threads, preventing
-  bottlenecks and maximizing parallelism.
+- **Work-Stealing Scheduler:** Implements a modern, priority-aware work-stealing
+  algorithm to efficiently distribute tasks across a pool of worker threads.
 - **Task Prioritization:** Supports submitting tasks with `High`, `Normal`, or
   `Low` priority, ensuring that latency-sensitive operations are handled
   immediately.
-- **Fluent Builder API:** A clean `Builder` allows for easy configuration of the
-  worker pool and other scheduler parameters.
+- **Fluent Builder API:** A clean `SchedulerBuilder` allows for easy
+  configuration of the worker pool size.
 - **Graceful Shutdown:** Provides a `Stop()` method to ensure all worker threads
   complete their current tasks and exit cleanly, preventing orphaned threads.
-- **Built for `ActionEffect`:** Serves as the ideal backend for effect systems,
-  providing the runtime engine that executes declarative, asynchronous workflows
-  defined in other parts of the application.
+- **Decoupled Architecture:** A generic `Queue` module provides the core
+  work-stealing logic, which is consumed by the application-specific
+  `Scheduler`.
 
 ---
 
 ## Core Architecture Principles 🏗️
 
-| Principle                  | Description                                                                                                                                             | Key Components Involved                               |
-| :------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------ | :---------------------------------------------------- |
-| **Performance**            | Use lock-free data structures (`crossbeam-deque`) and a work-stealing algorithm to achieve maximum throughput and low-latency task execution.           | `Queue::StealingQueue`, `Scheduler::Worker`           |
-| **Structured Concurrency** | Manage all asynchronous operations within a supervised pool of workers, providing graceful startup and shutdown, unlike fire-and-forget `tokio::spawn`. | `Scheduler::Scheduler`, `Scheduler::SchedulerBuilder` |
-| **Decoupling**             | Separate the _submission_ of a task from its _execution_. The application submits work, and the `Scheduler` handles how, when, and where it runs.       | `Scheduler::Scheduler::Submit`, `Task::Task::Task`    |
-| **Resilience**             | The scheduler's design is inherently resilient, as the failure of one task (if it panics) does not bring down the entire worker pool.                   | `Scheduler::Worker::Run` (execution loop)             |
-| **Composability**          | Provide a simple, generic `Submit` API that accepts any `Future<Output = ()>`, making it easy to integrate with any asynchronous Rust code.             | `Task::Task::Task`, `Scheduler::Scheduler::Submit`    |
+| Principle                  | Description                                                                                                                                                     | Key Components Involved                                               |
+| :------------------------- | :-------------------------------------------------------------------------------------------------------------------------------------------------------------- | :-------------------------------------------------------------------- |
+| **Performance**            | Use lock-free data structures (`crossbeam-deque`) and a high-performance work-stealing algorithm to achieve maximum throughput and low-latency task execution.  | `Queue::StealingQueue`, `Scheduler::Worker`                           |
+| **Structured Concurrency** | Manage all asynchronous operations within a supervised pool of workers, providing graceful startup and shutdown, unlike fire-and-forget `tokio::spawn`.         | `Scheduler::Scheduler`, `Scheduler::SchedulerBuilder`                 |
+| **Decoupling**             | Separate the generic **Queueing Logic** from the application-specific **Scheduler Implementation**. The scheduler uses the queue to run its tasks.              | `Queue::StealingQueue<T>`, `Scheduler::Scheduler`, `Task::Task::Task` |
+| **Resilience**             | The scheduler's design is inherently resilient; the failure of one task (if it panics) is contained within its `tokio` task and does not crash the worker pool. | `Scheduler::Worker::Run`                                              |
+| **Composability**          | Provide a simple `Submit` API that accepts any `Future<Output = ()>`, making it easy to integrate with any asynchronous Rust code.                              | `Task::Task::Task`, `Scheduler::Scheduler::Submit`                    |
 
 ---
 
@@ -120,14 +119,15 @@ graph LR
 
 ## Project Structure Overview 🗺️
 
-The `Echo` repository is organized into a few core modules:
+The `Echo` repository is organized into a few core modules with a clear
+separation of concerns:
 
 ```
 Echo/
 └── Source/
     ├── Library.rs               # Crate root, declares all modules.
-    ├── Scheduler/               # The main public API: Scheduler and Builder.
-    ├── Queue/                   # The generic, high-performance work-stealing queue.
+    ├── Scheduler/               # The main public API: Scheduler and Builder. Consumes the Queue.
+    ├── Queue/                   # The generic, high-performance work-stealing queue library.
     └── Task/                    # The application-specific definition of a Task and its Priority.
 ```
 
@@ -162,7 +162,8 @@ used throughout the application, often via a shared context or runtime.
 
     ```rust
     // In your application's lib.rs
-    pub use Echo::Scheduler::{Scheduler, SchedulerBuilder as Builder, Priority};
+    pub use Echo::Scheduler::{Scheduler, SchedulerBuilder};
+    pub use Echo::Task::Priority::Priority;
     ```
 
 2.  **Initialize the Scheduler:** Create and start the scheduler when your
@@ -174,7 +175,7 @@ used throughout the application, often via a shared context or runtime.
     use std::sync::Arc;
 
     // Use the fluent builder to configure and build the scheduler
-    let Scheduler = Arc::new(Builder::Create().Count(8).Build());
+    let Scheduler = Arc::new(SchedulerBuilder::Create().Count(8).Build());
     ```
 
 3.  **Submit Tasks:** Use the `Scheduler` instance to submit asynchronous work
@@ -199,6 +200,7 @@ used throughout the application, often via a shared context or runtime.
 
     ```rust
     // In your application's shutdown sequence
+    // Note: Arc::get_mut requires the Arc to have only one strong reference.
     if let Some(mut Scheduler) = Arc::get_mut(&mut Scheduler) {
         Scheduler.Stop().await;
     }

Source/Library.rs

@@ -9,5 +9,7 @@
 // --- Crate Modules ---
 // Declares the main modules that constitute the library.
 pub mod Queue;
+
 pub mod Scheduler;
+
 pub mod Task;

Source/Queue/StealingQueue.rs

@@ -8,13 +8,14 @@
 
 use std::sync::Arc;
 
-use crossbeam_deque::{Injector, Stealer, Worker};
-use rand::seq::SliceRandom;
+use crossbeam_deque::{Injector, Steal, Stealer, Worker};
+use rand::Rng;
 
 /// Defines a contract for types that can be prioritized by the queue.
 pub trait Prioritized {
 	/// The type of the priority value used by the implementor.
 	type Kind: PartialEq + Eq + Copy;
+
 	/// A method to retrieve the priority of the item.
 	fn Rank(&self) -> Self::Kind;
 }
@@ -23,7 +24,9 @@ pub trait Prioritized {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum Priority {
 	High,
+
 	Normal,
+
 	Low,
 }
 
@@ -34,6 +37,7 @@ pub enum Priority {
 pub struct Share<T> {
 	/// Global, multi-producer queues for each priority.
 	pub Injector:(Injector<T>, Injector<T>, Injector<T>),
+
 	/// Share handles for stealing tasks from each worker's queue.
 	pub Stealer:(Vec<Stealer<T>>, Vec<Stealer<T>>, Vec<Stealer<T>>),
 }
@@ -54,8 +58,10 @@ pub struct StealingQueue<T:Prioritized<Kind = Priority>> {
 pub struct Context<T> {
 	/// A unique identifier for the worker, used to avoid self-stealing.
 	pub Identifier:usize,
+
 	/// Thread-local work queues for each priority level.
 	pub Local:(Worker<T>, Worker<T>, Worker<T>),
+
 	/// A reference to the shared components of the entire queue system.
 	pub Share:Arc<Share<T>>,
 }
@@ -71,57 +77,73 @@ impl<T:Prioritized<Kind = Priority>> StealingQueue<T> {
 	/// 2. A `Vec` of `Context` objects, one for each worker thread to own.
 	pub fn Create(Count:usize) -> (Self, Vec<Context<T>>) {
 		let mut High:Vec<Worker<T>> = Vec::with_capacity(Count);
+
 		let mut Normal:Vec<Worker<T>> = Vec::with_capacity(Count);
+
 		let mut Low:Vec<Worker<T>> = Vec::with_capacity(Count);
 
 		// For each priority level, create a thread-local worker queue and its
 		// corresponding shared stealer.
 		let StealerHigh:Vec<Stealer<T>> = (0..Count)
 			.map(|_| {
 				let Worker = Worker::new_fifo();
+
 				let Stealer = Worker.stealer();
+
 				High.push(Worker);
+
 				Stealer
 			})
 			.collect();
 
 		let StealerNormal:Vec<Stealer<T>> = (0..Count)
 			.map(|_| {
 				let Worker = Worker::new_fifo();
+
 				let Stealer = Worker.stealer();
+
 				Normal.push(Worker);
+
 				Stealer
 			})
 			.collect();
 
 		let StealerLow:Vec<Stealer<T>> = (0..Count)
 			.map(|_| {
 				let Worker = Worker::new_fifo();
+
 				let Stealer = Worker.stealer();
+
 				Low.push(Worker);
+
 				Stealer
 			})
 			.collect();
 
 		// Bundle all shared components into an Arc for safe sharing.
 		let Share = Arc::new(Share {
 			Injector:(Injector::new(), Injector::new(), Injector::new()),
+
 			Stealer:(StealerHigh, StealerNormal, StealerLow),
 		});
 
 		// Create a unique context for each worker, giving it ownership of its
 		// local queues and a reference to the shared components.
-		let mut Contexts = Vec::with_capacity(Count);
+		let mut Context = Vec::with_capacity(Count);
+
 		for Identifier in 0..Count {
-			Contexts.push(Context {
+			Context.push(Context {
 				Identifier,
+
 				Local:(High.remove(0), Normal.remove(0), Low.remove(0)),
+
 				Share:Share.clone(),
 			});
 		}
 
 		let Queue = Self { Share };
-		(Queue, Contexts)
+
+		(Queue, Context)
 	}
 
 	/// Submits a new task to the appropriate global queue based on its
@@ -130,7 +152,9 @@ impl<T:Prioritized<Kind = Priority>> StealingQueue<T> {
 	pub fn Submit(&self, Task:T) {
 		match Task.Rank() {
 			Priority::High => self.Share.Injector.0.push(Task),
+
 			Priority::Normal => self.Share.Injector.1.push(Task),
+
 			Priority::Low => self.Share.Injector.2.push(Task),
 		}
 	}
@@ -157,22 +181,38 @@ impl<T> Context<T> {
 	/// It first tries to steal a batch from the global injector queue. If that
 	/// fails, it attempts to steal from a randomly chosen peer worker to ensure
 	/// fair distribution and avoid contention hotspots.
-	pub fn Steal<'a>(&self, Injector:&'a Injector<T>, Stealers:&'a [Stealer<T>], Local:&'a Worker<T>) -> Option<T> {
-		if Injector.steal_batch_and_pop(Local).is_success() {
-			return Local.pop();
+	pub fn Steal<'a>(&self, Injector:&'a Injector<T>, Stealer:&'a [Stealer<T>], Local:&'a Worker<T>) -> Option<T> {
+		// First, try to steal a batch from the global injector.
+		// `steal_batch_and_pop` is efficient: it moves a batch into our local
+		// queue and returns one task immediately if successful.
+		if let Steal::Success(Task) = Injector.steal_batch_and_pop(Local) {
+			return Some(Task);
+		}
+
+		// If the global queue is empty, try stealing from peers.
+		let Count = Stealer.len();
+
+		if Count <= 1 {
+			return None;
 		}
 
-		let mut Indices:Vec<usize> = (0..Stealers.len()).collect();
-		Indices.shuffle(&mut rand::rng());
+		// Allocation-free random iteration: pick a random starting point.
+		let Start = rand::rng().random_range(0..Count);
 
-		for Index in Indices {
+		// Iterate through all peers starting from the random offset.
+		for i in 0..Count {
+			let Index = (Start + i) % Count;
+
+			// Don't steal from ourselves.
 			if Index == self.Identifier {
 				continue;
 			}
-			if Stealers[Index].steal_batch_and_pop(Local).is_success() {
-				return Local.pop();
+
+			if let Steal::Success(Task) = Stealer[Index].steal_batch_and_pop(Local) {
+				return Some(Task);
 			}
 		}
+
 		None
 	}
 }

Source/Scheduler/Scheduler.rs

@@ -25,8 +25,10 @@ use crate::{
 pub struct Scheduler {
 	/// The underlying work-stealing queue system used for task submission.
 	Queue:StealingQueue<Task>,
+
 	/// Handles to the spawned worker threads, used for graceful shutdown.
 	Handle:Vec<JoinHandle<()>>,
+
 	/// An atomic flag to signal all workers to shut down.
 	Running:Arc<AtomicBool>,
 }
@@ -38,15 +40,18 @@ impl Scheduler {
 	/// `SchedulerBuilder`'s `Build` method.
 	pub fn Create(Count:usize, _Configuration:HashMap<String, Concurrency>) -> Self {
 		info!("[Scheduler] Create with {} workers.", Count);
+
 		let Running = Arc::new(AtomicBool::new(true));
 
 		// Create the entire queue system and retrieve the contexts for each worker.
 		let (Queue, Contexts) = StealingQueue::<Task>::Create(Count);
 
 		let mut Handle = Vec::with_capacity(Count);
+
 		// Spawn an asynchronous task for each worker.
 		for Context in Contexts.into_iter() {
 			let Running = Running.clone();
+
 			Handle.push(tokio::spawn(async move {
 				// Each task creates and runs a worker, consuming its context.
 				Worker::Create(Context, Running).Run().await;
@@ -73,15 +78,18 @@ impl Scheduler {
 	pub async fn Stop(&mut self) {
 		if !self.Running.swap(false, Ordering::Relaxed) {
 			info!("[Scheduler] Stop already initiated.");
+
 			return;
 		}
 
 		info!("[Scheduler] Stopping worker threads...");
+
 		for Handle in self.Handle.drain(..) {
 			if let Err(Error) = Handle.await {
 				error!("[Scheduler] Error joining worker: {}", Error);
 			}
 		}
+
 		info!("[Scheduler] All workers stopped.");
 	}
 }
@@ -94,6 +102,7 @@ impl Drop for Scheduler {
 	fn drop(&mut self) {
 		if self.Running.load(Ordering::Relaxed) {
 			warn!("[Scheduler] Dropped without explicit stop. Signaling workers.");
+
 			self.Running.store(false, Ordering::Relaxed);
 		}
 	}

Source/Scheduler/SchedulerBuilder.rs

@@ -13,6 +13,7 @@ use crate::Scheduler::Scheduler::Scheduler;
 pub enum Concurrency {
 	/// Specifies a maximum number of concurrent tasks for a queue.
 	Limit(usize),
+
 	/// Allows an unlimited number of concurrent tasks for a queue.
 	Unlimited,
 }
@@ -25,6 +26,7 @@ pub enum Concurrency {
 pub struct SchedulerBuilder {
 	/// The number of worker threads to be spawned in the scheduler's pool.
 	Count:usize,
+
 	/// Configuration for named queues with concurrency limits. (For future use)
 	Configuration:HashMap<String, Concurrency>,
 }
@@ -36,6 +38,7 @@ impl SchedulerBuilder {
 	/// system, with a minimum of two workers to ensure work-stealing is viable.
 	pub fn Create() -> Self {
 		let Default = num_cpus::get().max(2);
+
 		Self { Count:Default, Configuration:HashMap::new() }
 	}
 
@@ -45,17 +48,20 @@ impl SchedulerBuilder {
 	pub fn Count(mut self, Count:usize) -> Self {
 		if Count == 0 {
 			warn!("[Builder] Worker count of 0 is invalid. Defaulting to logical CPUs.");
+
 			self.Count = num_cpus::get().max(2);
 		} else {
 			self.Count = Count;
 		}
+
 		self
 	}
 
 	/// Configures a named queue with a specific concurrency limit. (For future
 	/// use)
 	pub fn Queue(mut self, Name:&str, Limit:Concurrency) -> Self {
 		self.Configuration.insert(Name.to_string(), Limit);
+
 		self
 	}