Author: NikolaRHristov

Source/Queue/StealingQueue.rs

@@ -1,97 +1,183 @@
-// Defines a high-performance, priority-aware, work-stealing deque for
-// distributing tasks among scheduler workers.
+// @module StealingQueue
+// @description A generic, priority-aware, work-stealing queue implementation.
+// This module is self-contained and can be used by any scheduler or application
+// to manage and distribute tasks of any type `T`.
 
-use crossbeam_deque::{Injector, Stealer, Worker};
+#![allow(non_snake_case, non_camel_case_types)]
+
+use std::sync::Arc;
+
+use crossbeam_deque::{Injector, Stealer, Worker as WorkerDeque};
 use rand::seq::SliceRandom;
 
-use crate::Task::{Priority::Enum, Task::Struct};
+// The task must have a way to specify its priority. We define a trait for this.
+pub trait Prioritized {
+	type P: PartialEq + Eq + Copy;
 
-/// A container for a set of queues for a single priority level.
-struct PriorityQueueSet {
-	GlobalInjector:Injector<Struct>,
-	WorkerQueue:Vec<Worker<Struct>>,
-	Stealer:Vec<Stealer<Struct>>,
+	fn GetPriority(&self) -> Self::P;
 }
 
-impl PriorityQueueSet {
-	/// Creates a new set of queues for a given number of workers.
-	fn New(NumberOfWorker:usize) -> Self {
-		let WorkerQueue:Vec<Worker<Struct>> = (0..NumberOfWorker).map(|_| Worker::new_fifo()).collect();
-		Self {
-			GlobalInjector:Injector::new(),
-			Stealer:WorkerQueue.iter().map(|w| w.stealer()).collect(),
-			WorkerQueue,
-		}
-	}
+// A simple enum for the library to use. The consumer's task must map to this.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Priority {
+	High,
+
+	Normal,
+
+	Low,
 }
 
-/// A collection of worker deques that supports priority-aware work-stealing.
-///
-/// This struct holds three distinct sets of queues, one for each priority level
-/// (`High`, `Normal`, `Low`). When a worker needs a task, it always checks for
-/// higher-priority work before considering lower-priority work.
-pub(crate) struct StealingQueue {
-	High:PriorityQueueSet,
-	Normal:PriorityQueueSet,
-	Low:PriorityQueueSet,
+// The parts of the queue that can be safely shared across all threads.
+struct Shared<T> {
+	// High, Normal, Low
+	Injector:(Injector<T>, Injector<T>, Injector<T>),
+
+	// High, Normal, Low
+	Stealer:(Vec<Stealer<T>>, Vec<Stealer<T>>, Vec<Stealer<T>>),
 }
 
-impl StealingQueue {
-	/// Creates a new `StealingQueue` with a dedicated set of queues for each
-	/// priority level.
-	pub fn New(NumberOfWorker:usize) -> Self {
-		Self {
-			High:PriorityQueueSet::New(NumberOfWorker),
-			Normal:PriorityQueueSet::New(NumberOfWorker),
-			Low:PriorityQueueSet::New(NumberOfWorker),
-		}
-	}
+/// The public-facing work-stealing queue. It is generic over the task type `T`.
+/// This object is held by the task submitter.
+pub struct StealingQueue<T:Prioritized<P = Priority>> {
+	Shared:Arc<Shared<T>>,
+}
+
+/// A context object that contains everything a single worker thread needs to
+/// operate. This includes its own private, thread-local deques.
+pub struct WorkerContext<T> {
+	Id:usize,
+
+	// High, Normal, Low
+	Local:(WorkerDeque<T>, WorkerDeque<T>, WorkerDeque<T>),
+
+	Shared:Arc<Shared<T>>,
+}
+
+impl<T:Prioritized<P = Priority>> StealingQueue<T> {
+	/// Creates a new work-stealing queue system.
+	///
+	/// Returns a tuple containing:
+	/// 1. The `StealingQueue` for submitting tasks.
+	/// 2. A `Vec` of `WorkerContext`s, one for each worker to be spawned.
+	pub fn New(Count:usize) -> (Self, Vec<WorkerContext<T>>) {
+		let mut High:Vec<WorkerDeque<T>> = Vec::with_capacity(Count);
+
+		let mut Normal:Vec<WorkerDeque<T>> = Vec::with_capacity(Count);
+
+		let mut Low:Vec<WorkerDeque<T>> = Vec::with_capacity(Count);
+
+		// --- FIX: Use the documented API for creating Worker/Stealer pairs ---
+		let StealerHigh:Vec<Stealer<T>> = (0..Count)
+			.map(|_| {
+				// 1. Create the Worker.
+				let Worker = WorkerDeque::new_fifo();
+
+				// 2. Get its Stealer.
+				let Stealer = Worker.stealer();
+
+				// 3. Store the Worker part.
+				High.push(Worker);
+
+				// 4. Return the Stealer part.
+				Stealer
+			})
+			.collect();
 
-	/// Submits a task to the appropriate global injection queue based on its
-	/// priority.
-	pub fn Push(&self, Task:Struct) {
-		match Task.Priority {
-			Enum::High => self.High.GlobalInjector.push(Task),
-			Enum::Normal => self.Normal.GlobalInjector.push(Task),
-			Enum::Low => self.Low.GlobalInjector.push(Task),
+		let StealerNormal:Vec<Stealer<T>> = (0..Count)
+			.map(|_| {
+				let Worker = WorkerDeque::new_fifo();
+
+				let Stealer = Worker.stealer();
+
+				Normal.push(Worker);
+
+				Stealer
+			})
+			.collect();
+
+		let StealerLow:Vec<Stealer<T>> = (0..Count)
+			.map(|_| {
+				let Worker = WorkerDeque::new_fifo();
+
+				let Stealer = Worker.stealer();
+
+				Low.push(Worker);
+
+				Stealer
+			})
+			.collect();
+
+		let Shared = Arc::new(Shared {
+			Injector:(Injector::new(), Injector::new(), Injector::new()),
+
+			Stealer:(StealerHigh, StealerNormal, StealerLow),
+		});
+
+		let mut Context = Vec::with_capacity(Count);
+
+		for Id in 0..Count {
+			// We use remove(0) because we built the Vecs in order and need to consume them.
+			Context.push(WorkerContext {
+				Id,
+
+				Local:(High.remove(0), Normal.remove(0), Low.remove(0)),
+
+				Shared:Shared.clone(),
+			});
 		}
+
+		let Queue = Self { Shared };
+
+		(Queue, Context)
 	}
 
-	/// Attempts to find a task for a given worker, always prioritizing
-	/// `High` > `Normal` > `Low`.
-	pub fn StealForWorker(&self, WorkerId:usize) -> Option<Struct> {
-		self.FindTaskInSet(&self.High, WorkerId)
-			.or_else(|| self.FindTaskInSet(&self.Normal, WorkerId))
-			.or_else(|| self.FindTaskInSet(&self.Low, WorkerId))
+	/// Submits a new task to the queue.
+	/// This is thread-safe and can be called from anywhere.
+	pub fn Submit(&self, Task:T) {
+		match Task.GetPriority() {
+			Priority::High => self.Shared.Injector.0.push(Task),
+
+			Priority::Normal => self.Shared.Injector.1.push(Task),
+
+			Priority::Low => self.Shared.Injector.2.push(Task),
+		}
 	}
+}
 
-	/// Implements the core work-finding logic for a specific priority level.
-	/// It first checks the worker's local queue, then attempts to steal.
-	fn FindTaskInSet(&self, Set:&PriorityQueueSet, WorkerId:usize) -> Option<Struct> {
-		Set.WorkerQueue[WorkerId]
-			.pop()
-			.or_else(|| self.StealFromSetGlobalOrPeer(Set, WorkerId))
+impl<T> WorkerContext<T> {
+	/// Finds the next available task for this worker.
+	/// Implements the full priority-aware, work-stealing logic.
+	pub fn NextTask(&self) -> Option<T> {
+		// Pop from local High
+		self.Local.0.pop()
+			 // Pop from local Normal
+			.or_else(|| self.Local.1.pop())
+			 // Pop from local Low
+			.or_else(|| self.Local.2.pop())
+			 // Steal High
+			.or_else(|| self.Steal(&self.Shared.Injector.0, &self.Shared.Stealer.0, &self.Local.0))
+			 // Steal Normal
+			.or_else(|| self.Steal(&self.Shared.Injector.1, &self.Shared.Stealer.1, &self.Local.1))
+			 // Steal Low
+			.or_else(|| self.Steal(&self.Shared.Injector.2, &self.Shared.Stealer.2, &self.Local.2))
 	}
 
-	/// Attempts to steal work for a specific priority level, first from the
-	/// global queue, then from peer workers.
-	fn StealFromSetGlobalOrPeer(&self, Set:&PriorityQueueSet, WorkerId:usize) -> Option<Struct> {
-		// Try stealing from the global injector for this priority set.
-		if Set.GlobalInjector.steal_batch_and_pop(&Set.WorkerQueue[WorkerId]).is_success() {
-			return Set.WorkerQueue[WorkerId].pop();
+	fn Steal<'a>(&self, Injector:&'a Injector<T>, Stealer:&'a [Stealer<T>], Local:&'a WorkerDeque<T>) -> Option<T> {
+		if Injector.steal_batch_and_pop(Local).is_success() {
+			return Local.pop();
 		}
 
-		// Try stealing from peers for this priority set. We shuffle the indices
-		// to ensure fairness and avoid contention hotspots.
-		let mut ShuffledIndex:Vec<usize> = (0..Set.Stealer.len()).collect();
-		ShuffledIndex.shuffle(&mut rand::rng());
+		let mut Index:Vec<usize> = (0..Stealer.len()).collect();
 
-		for Index in ShuffledIndex {
-			if Index == WorkerId {
-				continue; // Don't steal from ourselves.
+		Index.shuffle(&mut rand::rng());
+
+		for i in Index {
+			if i == self.Id {
+				continue;
 			}
-			if Set.Stealer[Index].steal_batch_and_pop(&Set.WorkerQueue[WorkerId]).is_success() {
-				return Set.WorkerQueue[WorkerId].pop();
+
+			if Stealer[i].steal_batch_and_pop(Local).is_success() {
+				return Local.pop();
 			}
 		}
 

Source/Scheduler/Scheduler.rs

@@ -5,54 +5,57 @@ use std::{
 		Arc,
 		atomic::{AtomicBool, Ordering},
 	},
+	time::Duration,
 };
 
-use log::{error, info, warn};
+use log::{error, info, trace, warn};
 use tokio::task::JoinHandle;
 
-/// @module Scheduler
-/// @description Defines the main `Scheduler` that manages the worker pool, task
-/// queue, and task execution lifecycle.
-use super::Worker::Worker;
+// CHANGED: Use the new reusable library.
+use crate::Queue::StealingQueue::StealingQueue;
 use crate::{
-	Queue::StealingQueue::StealingQueue,
 	Scheduler::SchedulerBuilder::Concurrency,
 	Task::{Priority::Enum, Task::Struct},
 };
 
 /// Manages a pool of worker threads and a work-stealing queue to execute tasks
 /// efficiently. This struct is the public-facing API of the Echo scheduler.
 pub struct Scheduler {
-	/// The underlying work-stealing queue shared by all workers.
-	Queue:Arc<StealingQueue>,
+	// CHANGED: The Scheduler now holds an instance of our generic queue.
+	Queue:StealingQueue<Struct>,
 	/// Handles to the spawned worker threads, allowing for graceful shutdown.
 	WorkerHandles:Vec<JoinHandle<()>>,
 	/// An atomic flag to signal workers to shut down.
 	IsRunning:Arc<AtomicBool>,
 }
 
 impl Scheduler {
-	/// Creates and starts a new scheduler with a given configuration.
-	/// This is a crate-private function, intended to be called only by the
-	/// `SchedulerBuilder`.
-	///
-	/// @param number_of_workers - The number of worker threads to spawn.
-	/// @param _queue_configs - Configuration for named queues with concurrency
-	/// limits (future use).
 	pub(crate) fn Start(number_of_workers:usize, _queue_configs:HashMap<String, Concurrency>) -> Self {
 		info!("[Scheduler] Starting scheduler with {} worker threads.", number_of_workers);
 		let IsRunning = Arc::new(AtomicBool::new(true));
-		let Queue = Arc::new(StealingQueue::New(number_of_workers));
+
+		// 1. Create the queue system. This is now a single, clean line.
+		let (Queue, WorkerContexts) = StealingQueue::New(number_of_workers);
 
 		let mut WorkerHandles = Vec::with_capacity(number_of_workers);
 
-		for WorkerIdentifier in 0..number_of_workers {
-			let CloneQueue = Queue.clone();
+		// 2. Iterate over the contexts, giving one to each new thread.
+		for Context in WorkerContexts.into_iter() {
 			let CloneIsRunning = IsRunning.clone();
 
 			let WorkerHandle = tokio::spawn(async move {
-				let WorkerInstance = Worker::New(WorkerIdentifier, CloneQueue, CloneIsRunning);
-				WorkerInstance.Run().await;
+				// The worker logic is now simple and lives directly inside the spawned task.
+				trace!("[Worker] Starting execution loop.");
+				while CloneIsRunning.load(Ordering::Relaxed) {
+					// Use the context to find the next task.
+					if let Some(Task) = Context.NextTask() {
+						trace!("[Worker] Found task with priority {:?}. Executing.", Task.Priority);
+						Task.Future.await
+					} else {
+						tokio::time::sleep(Duration::from_millis(1)).await;
+					}
+				}
+				trace!("[Worker] Execution loop finished.");
 			});
 
 			WorkerHandles.push(WorkerHandle);
@@ -61,22 +64,14 @@ impl Scheduler {
 		Self { Queue, WorkerHandles, IsRunning }
 	}
 
-	/// Submits a new task (as a `Future`) to the scheduler's global queue.
-	/// The task will be picked up by the next available worker.
-	///
-	/// @param future_instance - The async block or function to execute.
-	/// @param task_priority - The priority of the task.
 	pub fn submit<F>(&self, future_instance:F, task_priority:Enum)
 	where
 		F: Future<Output = ()> + Send + 'static, {
 		let new_task = Struct::New(future_instance, task_priority);
-		self.Queue.Push(new_task);
+		// 3. Use the generic queue's submit method.
+		self.Queue.Submit(new_task);
 	}
 
-	/// Asynchronously shuts down the scheduler.
-	///
-	/// This signals all worker threads to stop their loops and then waits for
-	/// them to complete their current tasks and exit gracefully.
 	pub async fn ShutDown(&mut self) {
 		if !self.IsRunning.swap(false, Ordering::Relaxed) {
 			info!("[Scheduler] ShutDown already initiated.");
@@ -94,13 +89,8 @@ impl Scheduler {
 }
 
 impl Drop for Scheduler {
-	/// Ensures that the scheduler is shut down when it goes out of scope,
-	/// preventing orphaned worker threads.
 	fn drop(&mut self) {
 		if self.IsRunning.load(Ordering::Relaxed) {
-			// If the scheduler is dropped without an explicit async shutdown,
-			// we must signal the workers to stop. We cannot await the handles
-			// here, but the threads will eventually terminate.
 			warn!("[Scheduler] Scheduler dropped without explicit shutdown. Signaling workers to stop.");
 			self.IsRunning.store(false, Ordering::Relaxed);
 		}

Source/Task/Task.rs

@@ -4,6 +4,8 @@ use std::{future::Future, pin::Pin};
 /// @description Defines the `Task` struct, which is the internal unit of work
 /// for the Echo scheduler.
 use super::Priority::Enum;
+// NEW: Import traits and types from our new reusable library.
+use crate::Queue::StealingQueue::{Prioritized, Priority as ReusablePriority};
 
 /// A type alias for a boxed, send-able, pinned future that returns no value.
 /// This is the standard way to handle dynamic, async operations in Rust.
@@ -33,3 +35,17 @@ impl Struct {
 		Self { Future:Box::pin(FutureInstance), Priority:PriorityValue }
 	}
 }
+
+// NEW: Implement the `Prioritized` trait to make our `Task` compatible
+// with the generic `ReusableQueue`.
+impl Prioritized for Struct {
+	type P = ReusablePriority;
+
+	fn GetPriority(&self) -> Self::P {
+		match self.Priority {
+			Enum::High => ReusablePriority::High,
+			Enum::Normal => ReusablePriority::Normal,
+			Enum::Low => ReusablePriority::Low,
+		}
+	}
+}