odem-rs-core 0.3.0

//! This module implements a scheduler for the simulation framework using an
//! event calendar based on [intrusive red–black trees].
//!
//! It is designed to schedule and manage the execution of simulation events
//! (represented as [continuations]) in a deterministic and predictable fashion.
//!
//! The scheduler leverages two separate layers for event management:
//!
//! - Time Layer: Organizes events strictly by their activation (model) [time].
//! - Rank Layer: Further orders events with identical activation times by a
//!   composite key that includes [rank], [precedence], and a global insertion
//!   counter.
//!
//! Taken together, these layers sort continuations according to:
//!
//! - Activation Time
//! - [Agent] Rank
//! - [Job] Precedence
//! - Insertion Order
//!
//! Jobs belonging to the same agent, scheduled at the same time, are grouped
//! together.
//!
//! [intrusive red–black trees]: RBTree
//! [continuations]: Continuation
//! [time]: crate::config::Time
//! [rank]: crate::config::Rank
//! [precedence]: Prec
//! [Agent]: crate::agent
//! [Job]: crate::job

use intrusive_collections::{RBTree, SinglyLinkedList};

use adapter::*;

use super::{ContTimePair, IdleOnDrop, PlanState, Scheduler};

use crate::{
	config::Config,
	continuation::{Adapter, Continuation, token},
	fsm::Brand,
	ptr::Irc,
	simulator::{Mark, Prec},
};

use core::{cell::Cell, cmp::Reverse, fmt, ops::Range};

mod adapter;

/// An event calendar for simulation scheduling based on red–black trees.
///
/// This calendar maintains two distinct red–black trees:
///
/// - Time Layer: Orders events strictly by their activation (model) time.
///   Since model time is monotonically increasing, events only transition from
///   this layer to the rank layer.
/// - Rank Layer: Orders events by a composite key of agent [rank], [mark], job
///   [precedence], and insertion order. This secondary ordering ensures that
///   events with the same activation time are executed in deterministic order.
///
/// ## Key Concepts
///
/// - Agent Rank: A user-assigned value that prioritizes [agents].
/// - Mark: An internally managed unique identifier used to group jobs from
///   the same agent together within a time slice. Although not essential for
///   determinism, it aids in interpreting trace outputs.
/// - Job Precedence: Orders [jobs] within the same agent when activation times
///   and ranks are equal.
///
/// The calendar guarantees that continuations (jobs) are sorted stably when
/// activation time, agent rank, and precedence are identical.
///
/// [rank]: crate::config::Rank
/// [mark]: Mark
/// [precedence]: Prec
/// [agents]: crate::agent
/// [jobs]: crate::job
/// [continuations]: Continuation
pub struct Calendar<C: ?Sized + Config> {
	/// Red–black tree that orders events by model time.
	///
	/// This tree holds events scheduled solely by their activation time. As
	/// model time always moves forward, continuations transition from the time
	/// layer to the rank layer.
	time_layer: RBTree<TimeKey<C>>,

	/// Red–black tree that orders events by agent rank, mark, job precedence,
	/// and insertion order.
	///
	/// This layer ensures stable ordering when events share the same
	/// activation time.
	rank_layer: RBTree<RankKey<C>>,

	/// A generator for unique agent marks.
	///
	/// Marks are used to group jobs belonging to the same agent in the same
	/// time slice. It is assumed that incrementing `Mark` by one for every
	/// agent will not overflow.
	mark: Range<Mark>,

	/// Global insertion counter for tiebreaking.
	///
	/// This counter provides a unique ordering for events scheduled at the same
	/// model time. It is assumed that this counter will never overflow.
	counter: u64,
}

impl<C: ?Sized + Config> fmt::Debug for Calendar<C> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		let mut debug = f.debug_struct("Calendar");

		debug
			.field("time_layer", &self.time_layer)
			.field("rank_layer", &self.rank_layer)
			.finish()
	}
}

unsafe impl<C> Scheduler for Calendar<C>
where
	C: ?Sized + Config<Plan = Self>,
{
	type Config = C;
	type State = State<C>;

	/// Creates a new calendar instance with empty scheduling trees and reset
	/// counters.
	fn new(_: &C) -> Self {
		Self {
			time_layer: RBTree::new(TimeKey(Adapter::NEW)),
			rank_layer: RBTree::new(RankKey(Adapter::NEW)),
			mark: 1..1,
			counter: 0,
		}
	}

	/// Updates the rank for all pending jobs of a specific agent.
	///
	/// If any jobs of the agent are currently in the rank layer, this function
	/// removes them, updates the agent’s rank, and then reinserts the jobs with
	/// the new rank.
	///
	/// For all jobs of the agent currently located in the time layer, this
	/// function does nothing, since these are not sorted according to rank yet.
	fn update_rank<'brand>(&mut self, mark: Mark, rank: &Cell<C::Rank>, new_rank: C::Rank) {
		use intrusive_collections::Bound::Included;

		// Check if any jobs of the agent are present in the rank layer.
		if self.mark.contains(&mark) {
			// Collect all continuations with the specified mark.
			let mut list = SinglyLinkedList::new(Adapter::NEW);

			// Find the lower bound for the given mark and current rank.
			let mut cur = self.rank_layer.lower_bound_mut(Included(&(
				Reverse(rank.get()),
				mark,
				Prec::new(),
				0,
			)));

			// Remove all matching continuations from the rank layer.
			while let Some(cont) = cur.get() {
				if cont.brand(|inner, once| {
					let next = unsafe { inner.token(once).into_next().unwrap_unchecked() };
					inner.mark(&next).get() != mark
				}) {
					break;
				}
				list.push_front(unsafe { cur.remove().unwrap_unchecked() });
			}

			// Update the agent's rank.
			rank.set(new_rank);

			// Reinsert all removed continuations with the updated rank.
			for cont in list {
				self.rank_layer.insert(cont);
			}
		}
	}

	/// Schedules a continuation to be executed at the specified simulation
	/// time.
	fn schedule<'brand>(
		&mut self,
		cont: Irc<Continuation<'brand, C>>,
		idle: token::Idle<'brand>,
		time: C::Time,
	) -> token::Next<'brand> {
		let state = State {
			count: self.counter,
			time,
		};

		self.counter += 1;

		let next: token::Next<'_> = cont.state().transition(idle, state);
		self.time_layer.insert(IdleOnDrop::new(cont, &next));
		next
	}

	/// Defers the execution of a continuation by scheduling it at the current
	/// simulation time.
	fn defer<'brand>(
		&mut self,
		cont: Irc<Continuation<'brand, C>>,
		busy: token::Busy<'brand>,
		now: C::Time,
	) -> token::Next<'brand> {
		let idle: token::Idle<'_> = cont.state().transition(busy, ());
		self.schedule(cont, idle, now)
	}

	/// Activates a continuation at the specified simulation time.
	fn activate<'brand>(
		&mut self,
		cont: Irc<Continuation<'brand, C>>,
		idle: token::Idle<'brand>,
		now: C::Time,
	) -> token::Next<'brand> {
		let state = State {
			count: self.counter,
			time: now,
		};

		self.counter += 1;

		// Update the mark if it is stale.
		if !self.mark.contains(&cont.mark(&idle).get()) {
			cont.mark(&idle).set(self.mark.end);
			self.mark.end += 1;
		}

		let next: token::Next<'_> = cont.state().transition(idle, state);
		self.rank_layer.insert(IdleOnDrop::new(cont, &next));
		next
	}

	/// Removes a continuation from the scheduler.
	fn remove<'brand>(
		&mut self,
		cont: &Continuation<'brand, C>,
		next: token::Next<'brand>,
		_now: C::Time,
	) -> token::Idle<'brand> {
		cont.next_state(&next, |state| {
			let share = cont.branded_share(&next);

			// Attempt to locate the continuation in the rank layer.
			if let Some(result) = self
				.rank_layer
				.find_mut(&(
					Reverse(share.rank()),
					share.mark().get(),
					cont.prec(),
					state.count,
				))
				.remove()
			{
				return result;
			}

			// If not found in the rank layer, remove it from the time layer.
			unsafe {
				self.time_layer
					.cursor_mut_from_ptr(cont.detach())
					.remove()
					.unwrap_unchecked()
			}
		})
		.into_inner();

		cont.state().transition(next, ())
	}

	/// Extracts the next scheduled continuation for execution.
	///
	/// This function first attempts to remove a continuation from the rank
	/// layer. If none is available, it invalidates existing marks, extracts the
	/// first continuation from the time layer, and then transfers all
	/// continuations scheduled for the same model time into the rank layer.
	fn extract(&mut self) -> Option<ContTimePair<C>> {
		self.rank_layer
			.front_mut()
			.remove()
			.or_else(|| {
				// Invalidate all existing marks.
				self.mark.start = self.mark.end;

				// Remove the first continuation from the time layer.
				let mut cursor = self.time_layer.front_mut();
				let mut first = cursor.remove()?;

				// Retrieve the current model time, agent rank, and insertion
				// order.
				let (now, mut r1, mut s1) = first.brand(|inner, next| {
					inner.next_state(next, |state| {
						let share = inner.branded_share(next);
						let rank = share.rank();
						let mark = share.mark();

						// Update the mark
						mark.set(self.mark.end);
						self.mark.end += 1;

						(state.time, rank, state.count)
					})
				});

				// Move all continuations from the time layer scheduled for the
				// current model time into the rank layer.
				loop {
					let Some(next) = cursor.get() else {
						break;
					};

					if next.time().unwrap() != now {
						break;
					}

					let next = cursor.remove().unwrap();

					let (r2, s2) = next.brand(|inner, next| {
						inner.next_state(next, |state| {
							let share = inner.branded_share(next);
							let rank = share.rank();
							let mark = share.mark();

							if !self.mark.contains(&mark.get()) {
								mark.set(self.mark.end);
								self.mark.end += 1;
							}

							(rank, state.count)
						})
					});

					// Insert the larger key while keeping the smallest one.
					self.rank_layer.insert({
						if (Reverse(r1), first.prec(), s1) > (Reverse(r2), next.prec(), s2) {
							let tmp = first;
							(first, r1, s1) = (next, r2, s2);
							tmp
						} else {
							next
						}
					});
				}

				Some(first)
			})
			.map(|cont| ContTimePair {
				time: cont.brand(|inner, next| inner.next_state(next, |s| s.time)),
				cont: cont.into_inner(),
			})
	}
}

/// Represents the state associated with a scheduled continuation.
///
/// This state is inlined into every [`Continuation`] that is in the
/// [`Next`](crate::continuation::State::Next) state. It comprises the
/// activation time and a tiebreaking insertion counter.
pub struct State<C: ?Sized + Config> {
	/// The scheduled activation time.
	time: C::Time,
	/// A unique counter for tiebreaking between events with identical times.
	count: u64,
}

impl<C: ?Sized + Config> fmt::Debug for State<C> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.debug_struct("State")
			.field("time", &self.time)
			.field("count", &self.count)
			.finish()
	}
}

impl<C: ?Sized + Config> PlanState for State<C> {
	type Time = C::Time;

	/// Returns the activation time of the state.
	fn time(&self) -> C::Time {
		self.time
	}
}