Re-Sync

Author: NikolaRHristov

Source/lib.rs

@@ -0,0 +1,17 @@
+
+
+/**
+ * @module Echo Crate
+ * @description A high-performance, structured concurrency and task scheduling library for Rust,
+ * built on top of `tokio` and a work-stealing queue. Echo is designed to
+ * natively execute `Future`s, providing a robust runtime for complex, asynchronous applications.
+ */
+
+#![allow(non_snake_case, non_camel_case_types)]
+
+// --- Public API ---
+pub mod scheduler;
+pub mod task;
+
+// --- Internal Implementation ---
+mod queue;

Source/queue/StealingQueue.rs

@@ -0,0 +1,100 @@
+//! Defines a high-performance, priority-aware, work-stealing deque for
+//! distributing tasks among scheduler workers.
+
+use crossbeam_deque::{Injector, Stealer, Worker};
+use rand::seq::SliceRandom;
+
+use crate::task::{Priority, Task};
+
+/// A container for a set of queues for a single priority level.
+struct PriorityQueueSet {
+	GlobalInjector:Injector<Task>,
+	WorkerQueue:Vec<Worker<Task>>,
+	Stealer:Vec<Stealer<Task>>,
+}
+
+impl PriorityQueueSet {
+	/// Creates a new set of queues for a given number of workers.
+	fn New(NumberOfWorker:usize) -> Self {
+		let WorkerQueue:Vec<Worker<Task>> = (0..NumberOfWorker).map(|_| Worker::new_fifo()).collect();
+		Self {
+			GlobalInjector:Injector::new(),
+			Stealer:WorkerQueue.iter().map(|w| w.stealer()).collect(),
+			WorkerQueue,
+		}
+	}
+}
+
+/// 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,
+}
+
+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),
+		}
+	}
+
+	/// Submits a task to the appropriate global injection queue based on its
+	/// priority.
+	pub fn Push(&self, Task:Task) {
+		match Task.Priority {
+			Priority::High => self.High.GlobalInjector.push(Task),
+			Priority::Normal => self.Normal.GlobalInjector.push(Task),
+			Priority::Low => self.Low.GlobalInjector.push(Task),
+		}
+	}
+
+	/// Attempts to find a task for a given worker, always prioritizing
+	/// `High` > `Normal` > `Low`.
+	pub fn StealForWorker(&self, WorkerId:usize) -> Option<Task> {
+		self.FindTaskInSet(&self.High, WorkerId)
+			.or_else(|| self.FindTaskInSet(&self.Normal, WorkerId))
+			.or_else(|| self.FindTaskInSet(&self.Low, WorkerId))
+	}
+
+	/// 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<Task> {
+		Set.WorkerQueue[WorkerId]
+			.pop()
+			.or_else(|| self.StealFromSetGlobalOrPeer(Set, WorkerId))
+	}
+
+	/// 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<Task> {
+		// 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();
+		}
+
+		// 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::thread_rng());
+
+		for Index in ShuffledIndex {
+			if Index == WorkerId {
+				continue; // Don't steal from ourselves.
+			}
+			if Set.Stealer[Index].steal_batch_and_pop(&Set.WorkerQueue[WorkerId]).is_success() {
+				return Set.WorkerQueue[WorkerId].pop();
+			}
+		}
+
+		None
+	}
+}

Source/queue/mod.rs

@@ -0,0 +1,19 @@
+
+
+/**
+ * @module queue
+ * @description This module provides the high-performance, concurrent queueing
+ * implementations used by the Echo scheduler. These are internal components
+ * of the library.
+ */
+
+#![allow(non_snake_case, non_camel_case_types)]
+
+mod StealingQueue;
+
+/**
+ * Re-exports the `StealingQueue` for use within the `Echo` crate, but keeps it
+ * private from external consumers.
+ * @see StealingQueue
+ */
+pub(crate) use self::StealingQueue::StealingQueue;

Source/scheduler/Scheduler.rs

@@ -0,0 +1,104 @@
+use std::{
+	collections::HashMap,
+	future::Future,
+	sync::{
+		Arc,
+		atomic::{AtomicBool, Ordering},
+	},
+};
+
+use log::{error, info, 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;
+use crate::{
+	queue::StealingQueue,
+	scheduler::SchedulerBuilder::Concurrency,
+	task::{Priority, Task},
+};
+
+/// 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>,
+	/// 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 NumberOfWorkers - The number of worker threads to spawn.
+	/// @param QueueConfigs - Configuration for named queues with concurrency
+	/// limits (future use).
+	pub(crate) fn Start(NumberOfWorkers:usize, _QueueConfigs:HashMap<String, Concurrency>) -> Self {
+		info!("[Scheduler] Starting scheduler with {} worker threads.", NumberOfWorkers);
+		let IsRunning = Arc::new(AtomicBool::new(true));
+		let Queue = Arc::new(StealingQueue::New(NumberOfWorkers));
+
+		let mut WorkerHandles = Vec::with_capacity(NumberOfWorkers);
+
+		for WorkerId in 0..NumberOfWorkers {
+			let WorkerInstance = Worker::New(WorkerId, Queue.clone(), IsRunning.clone());
+			let WorkerHandle = tokio::spawn(async move {
+				WorkerInstance.Run().await;
+			});
+			WorkerHandles.push(WorkerHandle);
+		}
+
+		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 FutureInstance - The async block or function to execute.
+	/// @param TaskPriority - The priority of the task.
+	pub fn Submit<F>(&self, FutureInstance:F, TaskPriority:Priority)
+	where
+		F: Future<Output = ()> + Send + 'static, {
+		let NewTask = Task::New(FutureInstance, TaskPriority);
+		self.Queue.Push(NewTask);
+	}
+
+	/// 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.");
+			return;
+		}
+
+		info!("[Scheduler] Shutting down worker threads...");
+		for Handle in self.WorkerHandles.drain(..) {
+			if let Err(e) = Handle.await {
+				error!("[Scheduler] Error joining worker task during shutdown: {}", e);
+			}
+		}
+		info!("[Scheduler] All workers shut down successfully.");
+	}
+}
+
+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/scheduler/SchedulerBuilder.rs

@@ -0,0 +1,81 @@
+//! Defines the fluent builder for creating and configuring a `Scheduler`
+//! instance.
+
+use std::collections::HashMap;
+
+use log::warn;
+
+use super::Scheduler;
+
+/// An enum to define concurrency limits for named queues.
+#[derive(Debug, Clone, Copy)]
+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,
+}
+
+/// A fluent builder for creating a `Scheduler`.
+///
+/// This pattern provides a clear and readable API for configuring complex
+/// objects, such as a multi-queue, multi-threaded scheduler, before they are
+/// constructed.
+pub struct SchedulerBuilder {
+	WorkerCount:usize,
+	/// Stores the configuration for named queues.
+	QueueConfiguration:HashMap<String, Concurrency>,
+}
+
+impl SchedulerBuilder {
+	/// Creates a new `SchedulerBuilder` with default settings.
+	///
+	/// By default, the worker count is set to the number of logical CPUs on the
+	/// system, with a minimum of 2.
+	pub fn New() -> Self {
+		let DefaultWorkerCount = num_cpus::get().max(2);
+		Self { WorkerCount:DefaultWorkerCount, QueueConfiguration:HashMap::new() }
+	}
+
+	/// Sets the total number of worker threads for the scheduler's pool.
+	///
+	/// # Arguments
+	/// * `WorkerCount` - The desired number of workers. If `0`, it defaults
+	/// to the number of logical CPUs on the system.
+	pub fn WithWorkerCount(mut self, WorkerCount:usize) -> Self {
+		if WorkerCount == 0 {
+			warn!("[SchedulerBuilder] Worker count of 0 is invalid. Defaulting to number of logical CPUs.");
+			self.WorkerCount = num_cpus::get().max(2);
+		} else {
+			self.WorkerCount = WorkerCount;
+		}
+		self
+	}
+
+	/// Configures a named queue with a specific concurrency limit.
+	/// Tasks can later be submitted to this specific queue.
+	///
+	/// # Arguments
+	/// * `QueueName` - The name of the queue (e.g., "DiskIO", "Network").
+	/// * `ConcurrencyLimit` - The concurrency configuration for this queue.
+	pub fn WithQueue(mut self, QueueName:&str, ConcurrencyLimit:Concurrency) -> Self {
+		self.QueueConfiguration.insert(QueueName.to_string(), ConcurrencyLimit);
+		self
+	}
+
+	/// Builds and starts the `Scheduler` with the specified configuration.
+	/// This consumes the builder.
+	///
+	/// # Returns
+	/// A new, running `Scheduler` instance.
+	pub fn Build(self) -> Scheduler {
+		// The Scheduler's internal `Start` method will receive this
+		// configuration and set up the corresponding queues and worker logic.
+		Scheduler::Start(self.WorkerCount, self.QueueConfiguration)
+	}
+}
+
+impl Default for SchedulerBuilder {
+	/// Provides a default `SchedulerBuilder` instance.
+	fn default() -> Self { Self::New() }
+}

Source/scheduler/Worker.rs

@@ -0,0 +1,51 @@
+use std::sync::{
+	Arc,
+	atomic::{AtomicBool, Ordering},
+};
+
+use log::trace;
+use tokio::time::{Duration, sleep};
+
+/// @module Worker (Scheduler)
+/// @description Defines the `Worker` struct, which represents a single
+/// execution thread in the scheduler's pool. This is an internal component of
+/// the scheduler.
+use crate::queue::StealingQueue;
+
+/// Represents a single worker thread that continuously polls the work-stealing
+/// queue for tasks to execute.
+pub(crate) struct Worker {
+	Id:usize,
+	Queue:Arc<StealingQueue>,
+	IsRunning:Arc<AtomicBool>,
+}
+
+impl Worker {
+	pub fn New(Id:usize, Queue:Arc<StealingQueue>, IsRunning:Arc<AtomicBool>) -> Self { Self { Id, Queue, IsRunning } }
+
+	/// The main execution loop for the worker.
+	///
+	/// It continuously attempts to find a task from its local queue or by
+	/// stealing from other workers. When a task is found, its encapsulated
+	/// `Future` is awaited to completion.
+	pub async fn Run(&self) {
+		trace!("[Worker {}] Starting execution loop.", self.Id);
+		while self.IsRunning.load(Ordering::Relaxed) {
+			// Attempt to get a task from the shared queue.
+			let TaskOption = self.Queue.StealForWorker(self.Id);
+
+			if let Some(Task) = TaskOption {
+				trace!("[Worker {}] Found task with priority {:?}. Executing.", self.Id, Task.Priority);
+				// The future is executed simply by awaiting it.
+				// Any panics within the future will be caught by tokio's task system
+				// and can be handled when the JoinHandle is awaited on shutdown.
+				Task.Future.await;
+			} else {
+				// If no work is found anywhere, yield the thread to the OS to prevent
+				// a tight busy-loop from consuming 100% CPU.
+				sleep(Duration::from_millis(1)).await;
+			}
+		}
+		trace!("[Worker {}] Execution loop finished.", self.Id);
+	}
+}