odem-rs-core 0.3.0

use crate::{
	Puck,
	agent::Agent,
	config::Config,
	job::Job,
	ops::defer,
	simulator::{Prec, Sim, simulation},
};

use std::{
	boxed::Box,
	cell::Cell,
	collections::VecDeque,
	pin::{Pin, pin},
	rc::Rc,
	vec,
};

/// Simple configuration.
#[derive(Default)]
struct CustomConfig;

impl Config for CustomConfig {
	type Time = i32;
	type Rank = i32;
	type Data = ();
	type Plan = super::DefaultPlan<Self>;

	fn default_time(&self) -> Self::Time {
		0
	}

	fn default_rank(&self) -> Self::Rank {
		0
	}

	fn global_data(&self) -> &Self::Data {
		&()
	}
}

/// Wrapper around a shared counter to determine the ordering of scheduling ops.
struct Order(Cell<usize>);

impl Order {
	/// Creates a new shared counter with an initial value of 0.
	const fn new() -> Self {
		Self(Cell::new(0))
	}

	/// Returns the current value and advances the counter.
	fn next(&self) -> usize {
		let r = self.0.get();
		self.0.set(r + 1);
		r
	}
}

#[test]
fn stable_ordering() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		let shared = Order::new();

		// Create 100 agents.
		let mut pool = (0..100)
			.map(|_| Job::new(async { shared.next() }))
			.collect::<Box<_>>();

		// Activate them at the same time.
		let pucks = pool
			.iter_mut()
			.map(|job| sim.activate(unsafe { Pin::new_unchecked(job) }))
			.collect::<Box<_>>();

		// Await them in ascending order.
		for (order, puck) in pucks.into_iter().enumerate() {
			assert_eq!(
				puck.await,
				order,
				"Barring any other distinguishing factor, sorting of\
				 continuations is stable"
			);
		}
	}

	simulation(sim_main).unwrap();
}

#[test]
fn defer_passes_no_time() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		let now = sim.now();
		sim.defer().await;
		assert_eq!(now, sim.now(), "`defer` causes no model time to advance");
	}

	simulation(sim_main).unwrap();
}

#[test]
fn defer_skips_same_rank_jobs() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		let shared = Order::new();

		// Create 100 jobs.
		let mut pool = (0..100)
			.map(|_| Job::new(async { shared.next() }))
			.collect::<Box<_>>();

		// Activate them at the same time.
		for job in pool.iter_mut() {
			sim.activate(unsafe { Pin::new_unchecked(job) });
		}

		// Have them bypass the main job.
		sim.defer().await;

		assert_eq!(
			shared.next(),
			100,
			"`defer` causes all other continuations to be active first"
		);
	}

	simulation(sim_main).unwrap();
}

#[test]
fn defer_skips_lower_rank_agents() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		let shared = Rc::new(Order::new());

		// Create 100 lower-ranked agents.
		let mut pool = (0..100)
			.map(|_| {
				Agent::build()
					.with_rank(-1)
					.with_subject(shared.clone())
					.with_actions(async |shared: &Rc<Order>, _: &Sim<_>| shared.next())
					.finish()
			})
			.collect::<Box<_>>();

		// Activate them at the same time.
		for agent in pool.iter_mut() {
			sim.activate(unsafe { Pin::new_unchecked(agent) });
		}

		// Have them bypass the main job, even though they have a lower rank.
		sim.defer().await;

		assert_eq!(
			shared.next(),
			100,
			"`defer` causes all other continuations to be active first"
		);
	}

	simulation(sim_main).unwrap();
}

#[test]
fn time_ordering() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		// Create 100 jobs.
		let mut pool = (0..100)
			.map(|i| Job::new(async move { i }))
			.collect::<Box<_>>();

		// Activate them at different times.
		let pucks = pool
			.iter_mut()
			.enumerate()
			.map(|(i, job)| sim.schedule(unsafe { Pin::new_unchecked(job) }, i as i32))
			.collect::<Box<_>>();

		// Await them in ascending order and assert their equality with the
		// current model time.
		for puck in pucks {
			assert_eq!(
				puck.await,
				sim.now(),
				"Earlier continuations are processed before later ones"
			);
		}
	}

	simulation(sim_main).unwrap();
}

#[test]
fn agent_rank_ordering() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		// Create a shared counter for the agents to increment.
		let shared = Rc::new(Order::new());

		// Create agents and have them each increment the counter, returning
		// the previous value.
		let mut pool = (0..100)
			.map(|i| {
				Agent::build()
					.with_subject(shared.clone())
					.with_actions(async |i: &Rc<Order>, _: &Sim<_>| i.next())
					.with_rank(i)
					.finish()
			})
			.collect::<Box<_>>();

		// Activate the agents.
		let pucks = pool
			.iter_mut()
			.map(|job| sim.activate(unsafe { Pin::new_unchecked(job) }))
			.collect::<Box<_>>();

		// Defer sim_main so that the agents can run.
		defer().await;

		// Assert that the returned number is anti-proportional to the
		// agent's rank. This is because higher ranks are executed before
		// lower ranks, so an agent with rank 1 should return a number 1
		// smaller than an agent with rank 0.
		for (rank, puck) in pucks.into_iter().rev().enumerate() {
			let result = puck.await;
			assert_eq!(
				rank, result,
				"Agent's jobs are sorted according to their agent's rank \
				 in descending order"
			);
		}
	}

	simulation(sim_main).unwrap();
}

#[test]
fn job_precedence_linear_ordering() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		// Create a shared counter for the jobs to increment.
		let shared = Order::new();

		// Create 100 jobs and configure their precedence intervals.
		let mut pool = (0..100)
			.map(|prec| {
				Job::build()
					.with_actions(async { shared.next() })
					.with_precedence(Prec::from(prec))
					.finish()
			})
			.collect::<Box<_>>();

		// Activate the jobs.
		let pucks = pool
			.iter_mut()
			.map(|job| sim.activate(unsafe { Pin::new_unchecked(job) }))
			.collect::<Box<_>>();

		// Defer so that the jobs can run.
		defer().await;

		// Assert that the returned number is proportional to the job's
		// precedence, indicating the correct order of activations.
		for (prec, puck) in pucks.into_iter().enumerate() {
			let result = puck.await;
			assert_eq!(
				prec, result,
				"Jobs are scheduled according to their precedence in \
				 ascending order"
			);
		}
	}

	simulation(sim_main).unwrap();
}

#[test]
fn job_precedence_nested_ordering() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		// Create a shared counter for the jobs to increment.
		let shared = Order::new();

		let mut pool1 = vec![];
		let mut pool2 = vec![];
		let mut pool3 = vec![];

		// Split the precedence interval into nested partitions.
		for prec1 in sim.active().prec().split(4).unwrap() {
			pool1.push(
				Job::build()
					.with_actions(async { shared.next() })
					.with_precedence(prec1)
					.finish(),
			);

			for prec2 in prec1.split(3).unwrap() {
				pool2.push(
					Job::build()
						.with_actions(async { shared.next() })
						.with_precedence(prec2)
						.finish(),
				);

				for prec3 in prec2.split(2).unwrap() {
					pool3.push(
						Job::build()
							.with_actions(async { shared.next() })
							.with_precedence(prec3)
							.finish(),
					);
				}
			}
		}

		// Activate the jobs.
		let mut pucks1 = pool1
			.iter_mut()
			.map(|job| sim.activate(unsafe { Pin::new_unchecked(job) }))
			.collect::<VecDeque<_>>();

		let mut pucks2 = pool2
			.iter_mut()
			.map(|job| sim.activate(unsafe { Pin::new_unchecked(job) }))
			.collect::<VecDeque<_>>();

		let mut pucks3 = pool3
			.iter_mut()
			.map(|job| sim.activate(unsafe { Pin::new_unchecked(job) }))
			.collect::<VecDeque<_>>();

		// Defer sim_main so that the jobs can run.
		defer().await;

		// Ensure that parent nodes sort before their children and earlier
		// siblings, as well as all of their child nodes sort before later
		// siblings and their child nodes.
		//
		//       0
		//   1       5
		// 2 3 4   6 7 8
		let mut i = 0;
		for _ in 0..4 {
			assert_eq!(pucks1.pop_front().unwrap().await, i);
			i += 1;

			for _ in 0..3 {
				assert_eq!(pucks2.pop_front().unwrap().await, i);
				i += 1;

				for _ in 0..2 {
					assert_eq!(pucks3.pop_front().unwrap().await, i);
					i += 1;
				}
			}
		}
	}

	simulation(sim_main).unwrap();
}

#[test]
fn agent_mark_ordering() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		let shared = Rc::new(Order::new());

		// Create 4 agents.
		let mut pool = (0..4)
			.map(|i| {
				Agent::new((
					(i, shared.clone()),
					async |(i, shared): &(usize, Rc<Order>), sim: &Sim<_>| {
						// Create 2 jobs for each agent.
						let mut pool = (0..3)
							.map(|_| {
								Job::new(async {
									// Make sure that the jobs are scheduled first and run later.
									sim.advance(1).await;
									shared.next()
								})
							})
							.collect::<Box<_>>();

						// Activate them at the same time.
						let mut pucks = vec![];

						for job in pool.iter_mut() {
							pucks.push(sim.activate(unsafe { Pin::new_unchecked(job) }));

							// Leave time for other agents to activate their jobs.
							// This is necessary to cause the scheduler
							// to interleave the jobs if it sorts them stably only.
							sim.defer().await;
						}

						// Ensure that the jobs are executed contiguously, even if
						// the initially activated agents ordering interleaves.
						for (j, puck) in pucks.into_iter().enumerate() {
							let order = puck.await;
							assert_eq!(
								order,
								i * 3 + j,
								"All jobs of agent A should be sorted before the \
								 jobs of agent B, if there is at least one job \
								 of A sorted before all jobs of B."
							);
						}
					},
				))
			})
			.collect::<Box<_>>();

		// Activate them at the same time.
		for agent in pool.iter_mut() {
			sim.activate(unsafe { Pin::new_unchecked(agent) });
		}

		// Give them time to execute.
		sim.advance(2).await;
	}

	simulation(sim_main).unwrap();
}

#[test]
fn agent_rank_update() {
	async fn sim_main(sim: &Sim<CustomConfig>) {
		let shared = Rc::new(Order::new());

		// Create two agents, one with a low rank, one with a higher rank.
		let smith = pin!(
			Agent::build()
				.with_rank(0)
				.with_subject(shared.clone())
				.with_actions(async |shared: &Rc<Order>, sim: &Sim<CustomConfig>| {
					// Create two jobs and schedule them immediately.
					let j1 = pin!(Job::new(async { shared.next() }));

					// Start the job while we have the low rank.
					let p1 = sim.activate(j1);

					// Promote this agent.
					sim.update_rank(1);

					// Since this agent has been promoted, this job should have
					// executed first.
					assert_eq!(p1.await, 0);
				})
				.finish()
		);

		let brown = pin!(
			Agent::build()
				.with_rank(1)
				.with_subject(shared.clone())
				.with_actions(async |shared: &Rc<Order>, sim: &Sim<CustomConfig>| {
					// Create two jobs and schedule them at different times.
					let j1 = pin!(Job::new(async { shared.next() }));

					let p1 = sim.activate(j1);

					// Demote this agent.
					sim.update_rank(0);

					// Since this agent has been demoted, this job should have
					// executed later.
					assert_eq!(p1.await, 1);
				})
				.finish()
		);

		// Activate the agents and give them time to terminate.
		sim.activate(smith);
		sim.activate(brown);

		sim.advance(2).await;
	}

	simulation(sim_main).unwrap();
}