mohr/nexosim
1
0
Fork 0
forked from ROMEO/nexosim
Code Pull Requests Activity

Merge pull request #47 from asynchronics/feature/tracing

Browse Source 
Feature/tracing
This commit is contained in:
Jauhien Piatlicki 2024-09-13 16:20:35 +02:00 committed by GitHub
commit ef43f99a9c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 317 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
asynchronix/Cargo.toml
View File

@ -3,6 +3,7 @@ name = "asynchronix"
# When incrementing version and releasing to crates.io:
# - Update crate version in this Cargo.toml
# - Update crate version in README.md
# - Update crate version in the lib.rs documentation of feature flags
# - Update CHANGELOG.md
# - Update if necessary copyright notice in LICENSE-MIT
# - Create a "vX.Y.Z" git tag
@ -23,6 +24,7 @@ autotests = false
[features]
# gRPC service.
grpc = ["dep:bytes", "dep:ciborium", "dep:prost", "dep:prost-types", "dep:serde", "dep:tonic", "dep:tokio", "dep:tonic"]
tracing = ["dep:tracing", "dep:tracing-subscriber"]
# DEVELOPMENT ONLY: API-unstable public exports meant for external test/benchmarking.
dev-hooks = []
@ -44,20 +46,23 @@ spin_sleep = "1"
st3 = "0.4"
tai-time = "0.3.3"
# gRPC service dependencies.
# Optional dependencies.
bytes = { version = "1", default-features = false, optional = true }
ciborium = { version = "0.2.2", optional = true }
prost = { version = "0.13", optional = true }
prost-types = { version = "0.13", optional = true }
ciborium = { version = "0.2.2", optional = true }
serde = { version = "1", optional = true }
tokio = { version = "1.0", features=["net", "rt-multi-thread"], optional = true }
tonic = { version = "0.12", default-features = false, features=["codegen", "prost", "server"], optional = true }
tracing = { version= "0.1.40", default-features = false, features=["std"], optional = true }
tracing-subscriber = { version= "0.3.18", optional = true }
[dev-dependencies]
atomic-wait = "1.1"
futures-util = "0.3"
futures-executor = "0.3"
mio = { version = "1.0", features = ["os-poll", "net"] }
tracing-subscriber = { version= "0.3.18", features=["env-filter"] }
[target.'cfg(asynchronix_loom)'.dev-dependencies]
loom = "0.5"

12
asynchronix/src/dev_hooks.rs
View File

@ -15,7 +15,17 @@ impl Executor {
///
/// The maximum number of threads is set with the `pool_size` parameter.
pub fn new(pool_size: usize) -> Self {
Self(executor::Executor::new_multi_threaded(pool_size))
let dummy_context = crate::executor::SimulationContext {
#[cfg(feature = "tracing")]
time_reader: crate::util::sync_cell::SyncCell::new(
crate::time::TearableAtomicTime::new(crate::time::MonotonicTime::EPOCH),
)
.reader(),
};
Self(executor::Executor::new_multi_threaded(
pool_size,
dummy_context,
))
}
/// Spawns a task which output will never be retrieved.

43
asynchronix/src/executor.rs
View File

@ -7,11 +7,23 @@ mod task;
use std::future::Future;
use std::sync::atomic::AtomicUsize;
use crate::macros::scoped_thread_local::scoped_thread_local;
#[cfg(feature = "tracing")]
use crate::time::AtomicTimeReader;
use task::Promise;
/// Unique identifier for executor instances.
static NEXT_EXECUTOR_ID: AtomicUsize = AtomicUsize::new(0);
/// Context common to all executor types.
#[derive(Clone)]
pub(crate) struct SimulationContext {
#[cfg(feature = "tracing")]
pub(crate) time_reader: AtomicTimeReader,
}
scoped_thread_local!(pub(crate) static SIMULATION_CONTEXT: SimulationContext);
/// A single-threaded or multi-threaded `async` executor.
#[derive(Debug)]
pub(crate) enum Executor {
@ -21,8 +33,8 @@ pub(crate) enum Executor {
impl Executor {
/// Creates an executor that runs futures on the current thread.
pub(crate) fn new_single_threaded() -> Self {
Self::StExecutor(st_executor::Executor::new())
pub(crate) fn new_single_threaded(simulation_context: SimulationContext) -> Self {
Self::StExecutor(st_executor::Executor::new(simulation_context))
}
/// Creates an executor that runs futures on a thread pool.
@ -33,8 +45,11 @@ impl Executor {
///
/// This will panic if the specified number of threads is zero or is more
/// than `usize::BITS`.
pub(crate) fn new_multi_threaded(num_threads: usize) -> Self {
Self::MtExecutor(mt_executor::Executor::new(num_threads))
pub(crate) fn new_multi_threaded(
num_threads: usize,
simulation_context: SimulationContext,
) -> Self {
Self::MtExecutor(mt_executor::Executor::new(num_threads, simulation_context))
}
/// Spawns a task which output will never be retrieved.
@ -88,6 +103,16 @@ mod tests {
use super::*;
fn dummy_simulation_context() -> SimulationContext {
SimulationContext {
#[cfg(feature = "tracing")]
time_reader: crate::util::sync_cell::SyncCell::new(
crate::time::TearableAtomicTime::new(crate::time::MonotonicTime::EPOCH),
)
.reader(),
}
}
/// An object that runs an arbitrary closure when dropped.
struct RunOnDrop<F: FnOnce()> {
drop_fn: Option<F>,
@ -208,25 +233,25 @@ mod tests {
#[test]
fn executor_deadlock_st() {
executor_deadlock(Executor::new_single_threaded());
executor_deadlock(Executor::new_single_threaded(dummy_simulation_context()));
}
#[test]
fn executor_deadlock_mt() {
executor_deadlock(Executor::new_multi_threaded(3));
executor_deadlock(Executor::new_multi_threaded(3, dummy_simulation_context()));
}
#[test]
fn executor_deadlock_mt_one_worker() {
executor_deadlock(Executor::new_multi_threaded(1));
executor_deadlock(Executor::new_multi_threaded(1, dummy_simulation_context()));
}
#[test]
fn executor_drop_cycle_st() {
executor_drop_cycle(Executor::new_single_threaded());
executor_drop_cycle(Executor::new_single_threaded(dummy_simulation_context()));
}
#[test]
fn executor_drop_cycle_mt() {
executor_drop_cycle(Executor::new_multi_threaded(3));
executor_drop_cycle(Executor::new_multi_threaded(3, dummy_simulation_context()));
}
}

17
asynchronix/src/executor/mt_executor.rs
View File

@ -56,11 +56,10 @@ use std::time::{Duration, Instant};
use crossbeam_utils::sync::{Parker, Unparker};
use slab::Slab;
use super::task::{self, CancelToken, Promise, Runnable};
use super::{SimulationContext, NEXT_EXECUTOR_ID, SIMULATION_CONTEXT};
use crate::macros::scoped_thread_local::scoped_thread_local;
use crate::util::rng::Rng;
use super::task::{self, CancelToken, Promise, Runnable};
use super::NEXT_EXECUTOR_ID;
use pool_manager::PoolManager;
const BUCKET_SIZE: usize = 128;
@ -95,7 +94,7 @@ impl Executor {
///
/// This will panic if the specified number of threads is zero or is more
/// than `usize::BITS`.
pub(crate) fn new(num_threads: usize) -> Self {
pub(crate) fn new(num_threads: usize, simulation_context: SimulationContext) -> Self {
let parker = Parker::new();
let unparker = parker.unparker().clone();
@ -141,11 +140,15 @@ impl Executor {
.spawn({
let context = context.clone();
let active_tasks = active_tasks.clone();
let simulation_context = simulation_context.clone();
move || {
let worker = Worker::new(local_queue, context);
ACTIVE_TASKS.set(&active_tasks, || {
LOCAL_WORKER
.set(&worker, || run_local_worker(&worker, id, worker_parker))
SIMULATION_CONTEXT.set(&simulation_context, || {
ACTIVE_TASKS.set(&active_tasks, || {
LOCAL_WORKER.set(&worker, || {
run_local_worker(&worker, id, worker_parker)
})
})
});
}
})

28
asynchronix/src/executor/st_executor.rs
View File

@ -8,6 +8,7 @@ use slab::Slab;
use super::task::{self, CancelToken, Promise, Runnable};
use super::NEXT_EXECUTOR_ID;
use crate::executor::{SimulationContext, SIMULATION_CONTEXT};
use crate::macros::scoped_thread_local::scoped_thread_local;
const QUEUE_MIN_CAPACITY: usize = 32;
@ -21,11 +22,13 @@ pub(crate) struct Executor {
context: ExecutorContext,
/// List of tasks that have not completed yet.
active_tasks: RefCell<Slab<CancelToken>>,
/// Read-only handle to the simulation time.
simulation_context: SimulationContext,
}
impl Executor {
/// Creates an executor that runs futures on the current thread.
pub(crate) fn new() -> Self {
pub(crate) fn new(simulation_context: SimulationContext) -> Self {
// Each executor instance has a unique ID inherited by tasks to ensure
// that tasks are scheduled on their parent executor.
let executor_id = NEXT_EXECUTOR_ID.fetch_add(1, Ordering::Relaxed);
@ -40,6 +43,7 @@ impl Executor {
Self {
context,
active_tasks,
simulation_context,
}
}
@ -102,14 +106,16 @@ impl Executor {
/// Execute spawned tasks, blocking until all futures have completed or
/// until the executor reaches a deadlock.
pub(crate) fn run(&mut self) {
ACTIVE_TASKS.set(&self.active_tasks, || {
EXECUTOR_CONTEXT.set(&self.context, || loop {
let task = match self.context.queue.borrow_mut().pop() {
Some(task) => task,
None => break,
};
SIMULATION_CONTEXT.set(&self.simulation_context, || {
ACTIVE_TASKS.set(&self.active_tasks, || {
EXECUTOR_CONTEXT.set(&self.context, || loop {
let task = match self.context.queue.borrow_mut().pop() {
Some(task) => task,
None => break,
};
task.run();
task.run();
})
})
});
}
@ -225,9 +231,9 @@ impl<T: Future> Drop for CancellableFuture<T> {
///
/// # Panics
///
/// This function will panic if called from called outside from the executor
/// work thread or from another executor instance than the one the task for this
/// `Runnable` was spawned on.
/// This function will panic if called from outside the executor worker thread
/// or from another executor instance than the one the task for this `Runnable`
/// was spawned on.
fn schedule_task(task: Runnable, executor_id: usize) {
EXECUTOR_CONTEXT
.map(|context| {

28
asynchronix/src/lib.rs
View File

@ -379,6 +379,22 @@
//! [pony]: https://www.ponylang.io/
//!
//!
//! # Feature flags
//!
//! ## Tracing support
//!
//! The `tracing` feature flag provides support for the
//! [`tracing`](https://docs.rs/tracing/latest/tracing/) crate and can be
//! activated in `Cargo.toml` with:
//!
//! ```toml
//! [dependencies]
//! asynchronix = { version = "0.3", features = ["tracing"] }
//! ```
//!
//! See the [`tracing`] module for more information.
//!
//!
//! # Other resources
//!
//! ## Other examples
@ -411,17 +427,21 @@
pub(crate) mod channel;
pub(crate) mod executor;
#[cfg(feature = "grpc")]
pub mod grpc;
mod loom_exports;
pub(crate) mod macros;
pub mod model;
pub mod ports;
#[cfg(feature = "grpc")]
pub mod registry;
pub mod simulation;
pub mod time;
pub(crate) mod util;
#[cfg(feature = "grpc")]
pub mod grpc;
#[cfg(feature = "grpc")]
pub mod registry;
#[cfg(feature = "tracing")]
pub mod tracing;
#[cfg(feature = "dev-hooks")]
pub mod dev_hooks;

21
asynchronix/src/simulation.rs
View File

@ -146,10 +146,9 @@ use recycle_box::{coerce_box, RecycleBox};
use crate::executor::Executor;
use crate::model::{Context, Model, SetupContext};
use crate::ports::{InputFn, ReplierFn};
use crate::time::{Clock, MonotonicTime, TearableAtomicTime};
use crate::time::{AtomicTime, Clock, MonotonicTime};
use crate::util::seq_futures::SeqFuture;
use crate::util::slot;
use crate::util::sync_cell::SyncCell;
/// Simulation environment.
///
@ -192,7 +191,7 @@ use crate::util::sync_cell::SyncCell;
pub struct Simulation {
executor: Executor,
scheduler_queue: Arc<Mutex<SchedulerQueue>>,
time: SyncCell<TearableAtomicTime>,
time: AtomicTime,
clock: Box<dyn Clock>,
}
@ -201,7 +200,7 @@ impl Simulation {
pub(crate) fn new(
executor: Executor,
scheduler_queue: Arc<Mutex<SchedulerQueue>>,
time: SyncCell<TearableAtomicTime>,
time: AtomicTime,
clock: Box<dyn Clock + 'static>,
) -> Self {
Self {
@ -257,7 +256,7 @@ impl Simulation {
Ok(())
}
/// Returns scheduler.
/// Returns a scheduler handle.
pub fn scheduler(&self) -> Scheduler {
Scheduler::new(self.scheduler_queue.clone(), self.time.reader())
}
@ -488,14 +487,22 @@ pub(crate) fn add_model<M: Model>(
scheduler: Scheduler,
executor: &Executor,
) {
#[cfg(feature = "tracing")]
let span = tracing::span!(target: env!("CARGO_PKG_NAME"), tracing::Level::INFO, "model", name);
let context = Context::new(name, LocalScheduler::new(scheduler, mailbox.address()));
let setup_context = SetupContext::new(&mailbox, &context, executor);
model.setup(&setup_context);
let mut receiver = mailbox.0;
executor.spawn_and_forget(async move {
let fut = async move {
let mut model = model.init(&context).await.0;
while receiver.recv(&mut model, &context).await.is_ok() {}
});
};
#[cfg(feature = "tracing")]
let fut = tracing::Instrument::instrument(fut, span);
executor.spawn_and_forget(fut);
}

10
asynchronix/src/simulation/scheduler.rs
View File

@ -18,22 +18,18 @@ use crate::executor::Executor;
use crate::model::Model;
use crate::ports::InputFn;
use crate::simulation::Address;
use crate::time::{MonotonicTime, TearableAtomicTime};
use crate::time::{AtomicTimeReader, MonotonicTime};
use crate::util::priority_queue::PriorityQueue;
use crate::util::sync_cell::SyncCellReader;
/// Scheduler.
#[derive(Clone)]
pub struct Scheduler {
scheduler_queue: Arc<Mutex<SchedulerQueue>>,
time: SyncCellReader<TearableAtomicTime>,
time: AtomicTimeReader,
}
impl Scheduler {
pub(crate) fn new(
scheduler_queue: Arc<Mutex<SchedulerQueue>>,
time: SyncCellReader<TearableAtomicTime>,
) -> Self {
pub(crate) fn new(scheduler_queue: Arc<Mutex<SchedulerQueue>>, time: AtomicTimeReader) -> Self {
Self {
scheduler_queue,
time,

17
asynchronix/src/simulation/sim_init.rs
View File

@ -1,10 +1,10 @@
use std::fmt;
use std::sync::{Arc, Mutex};
use crate::executor::Executor;
use crate::executor::{Executor, SimulationContext};
use crate::model::Model;
use crate::time::{AtomicTime, MonotonicTime, TearableAtomicTime};
use crate::time::{Clock, NoClock};
use crate::time::{MonotonicTime, TearableAtomicTime};
use crate::util::priority_queue::PriorityQueue;
use crate::util::sync_cell::SyncCell;
@ -14,7 +14,7 @@ use super::{add_model, Mailbox, Scheduler, SchedulerQueue, Simulation};
pub struct SimInit {
executor: Executor,
scheduler_queue: Arc<Mutex<SchedulerQueue>>,
time: SyncCell<TearableAtomicTime>,
time: AtomicTime,
clock: Box<dyn Clock + 'static>,
}
@ -32,17 +32,22 @@ impl SimInit {
/// be between 1 and `usize::BITS` (inclusive).
pub fn with_num_threads(num_threads: usize) -> Self {
let num_threads = num_threads.clamp(1, usize::BITS as usize);
let time = SyncCell::new(TearableAtomicTime::new(MonotonicTime::EPOCH));
let simulation_context = SimulationContext {
#[cfg(feature = "tracing")]
time_reader: time.reader(),
};
let executor = if num_threads == 1 {
Executor::new_single_threaded()
Executor::new_single_threaded(simulation_context)
} else {
Executor::new_multi_threaded(num_threads)
Executor::new_multi_threaded(num_threads, simulation_context)
};
Self {
executor,
scheduler_queue: Arc::new(Mutex::new(PriorityQueue::new())),
time: SyncCell::new(TearableAtomicTime::new(MonotonicTime::EPOCH)),
time,
clock: Box::new(NoClock::new()),
}
}

3
asynchronix/src/time.rs
View File

@ -52,3 +52,6 @@ pub use tai_time::MonotonicTime;
pub use clock::{AutoSystemClock, Clock, NoClock, SyncStatus, SystemClock};
pub(crate) use monotonic_time::TearableAtomicTime;
pub(crate) type AtomicTime = crate::util::sync_cell::SyncCell<TearableAtomicTime>;
pub(crate) type AtomicTimeReader = crate::util::sync_cell::SyncCellReader<TearableAtomicTime>;

183
asynchronix/src/tracing.rs Normal file
View File

@ -0,0 +1,183 @@
//! Support for structured logging.
//!
//! # Overview
//!
//! When the `tracing` feature is activated, each tracing event or span emitted
//! by a model is wrapped in a [`tracing::Span`] with the following metadata:
//!
//! - name: `model`,
//! - target: `asynchronix`,
//! - verbosity level: [`Level::INFO`](tracing::Level::INFO),
//! - a unique field called `name`, associated to the model name provided in
//! [`SimInit::add_model`](crate::simulation::SimInit::add_model).
//!
//! The emission of `model` spans can be readily used for [event
//! filtering](#event-filtering-examples), using for instance the
//! [`tracing_subscriber::fmt`][mod@tracing_subscriber::fmt] subscriber. By
//! default, however, this subscriber will timestamp events with the wall clock
//! time. Because it is often desirable to log events using the simulation time
//! instead of (or on top of) the wall clock time, this module provides a custom
//! [`SimulationTime`] timer compatible with
//! [`tracing_subscriber::fmt`][mod@tracing_subscriber::fmt].
//!
//!
//! # Configuration
//!
//! Using the `tracing-subscriber` crate, simulation events can be logged to
//! standard output by placing the following call anywhere before
//! [`SimInit::init`](crate::simulation::SimInit::init):
//!
//! ```
//! tracing_subscriber::fmt::init();
//! ```
//!
//! Logging from a model is then a simple matter of using the `tracing` macros,
//! for instance:
//!
//! ```
//! use tracing::warn;
//!
//! pub struct MyModel { /* ... */ }
//!
//! impl MyModel {
//! pub fn some_input_port(&mut self, _some_parameter: i32) {
//! // ...
//! warn!("something happened inside the simulation");
//! // ...
//! }
//! }
//! ```
//!
//! However, this will stamp events with the system time rather than the
//! simulation time. To use simulation time instead, a dedicated timer can be
//! configured:
//!
//! ```
//! use asynchronix::tracing::SimulationTime;
//!
//! tracing_subscriber::fmt()
//! .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
//! .with_timer(SimulationTime::with_system_timer())
//! .init();
//! ```
//!
//! This timer will automatically revert to system time stamping for tracing
//! events generated outside of simulation models, e.g.:
//!
//! ```text
//! [2001-02-03 04:05:06.789012345] WARN model{name="my_model"}: my_simulation: something happened inside the simulation
//! 2024-09-10T14:39:24.670921Z INFO my_simulation: something happened outside the simulation
//! ```
//!
//! Alternatively, `SimulationTime::with_system_timer_always()` can be used to
//! always prepend the system time even for simulation events:
//!
//! ```text
//! 2024-09-10T14:39:22.124945Z [2001-02-03 04:05:06.789012345] WARN model{name="my_model"}: my_simulation: something happened inside the simulation
//! 2024-09-10T14:39:24.670921Z INFO my_simulation: something happened outside the simulation
//! ```
//!
//!
//! # Event filtering examples
//!
//! Note that event filtering based on the `RUST_LOG` environment variable
//! requires the `env-filter` feature of the
//! [`tracing-subscriber`][tracing_subscriber] crate.
//!
//! The following `RUST_LOG` directive could be used to only let warnings and
//! errors pass through but still see model span information (which is emitted
//! as info):
//!
//! ```text
//! $ RUST_LOG="warn,[model]=info" cargo run --release my_simulation
//! [2001-01-01 00:00:06.000000000] WARN model{name="kettle"}: my_simulation: water is boiling
//! [2001-01-01 00:01:36.000000000] WARN model{name="timer"}: my_simulation: ring ring
//! [2001-01-01 00:01:36.000000000] WARN model{name="kettle"}: my_simulation: water is ready
//! ```
//!
//! In order to see warnings or errors for the `kettle` model only, this
//! directive could be modified as follows:
//!
//! ```text
//! $ RUST_LOG="[model{name=kettle}]=warn" cargo run --release my_simulation
//! [2001-01-01 00:00:06.000000000] WARN model{name="kettle"}: my_simulation: water is boiling
//! [2001-01-01 00:01:36.000000000] WARN model{name="kettle"}: my_simulation: water is ready
//! ```
//!
//! If the `model` span name collides with that of spans defined outside
//! `asynchronix`, the above filters can be made more specific using
//! `asynchronix[model]` instead of just `[model]`.
//!
//!
//! # Customization
//!
//! The [`tracing-subscriber`][tracing_subscriber] crate allows for
//! customization such as logging to files or formatting logs with JSON.
//!
//! Further customization is possible by implementing a
//! [`tracing_subscriber::layer::Layer`] or a dedicated [`tracing::Subscriber`].
use std::fmt;
use tracing_subscriber::fmt::format::Writer;
use tracing_subscriber::fmt::time::{FormatTime, SystemTime};
use crate::executor::SIMULATION_CONTEXT;
/// A timer that can be used in conjunction with the
/// [`tracing-subscriber`][tracing_subscriber] crate to log events using the
/// simulation time instead of (or on top of) the wall clock time.
///
/// See the [module-level documentation][crate::tracing] for more details.
#[derive(Default, Debug)]
pub struct SimulationTime<const VERBOSE: bool, T> {
sys_timer: T,
}
impl SimulationTime<false, SystemTime> {
/// Constructs a new simulation timer which falls back to the [`SystemTime`]
/// timer for events generated outside the simulator.
pub fn with_system_timer() -> Self {
Self::default()
}
}
impl SimulationTime<true, SystemTime> {
/// Constructs a new simulation timer which prepends a [`SystemTime`]
/// timestamp to all tracing events, as well as a simulation timestamp for
/// simulation events.
pub fn with_system_timer_always() -> Self {
Self::default()
}
}
impl<T: FormatTime> SimulationTime<false, T> {
/// Constructs a new simulation timer which falls back to the provided
/// timer for tracing events generated outside the simulator.
pub fn with_custom_timer(sys_timer: T) -> Self {
Self { sys_timer }
}
}
impl<T: FormatTime> SimulationTime<true, T> {
/// Constructs a new simulation timer which prepends a timestamp generated
/// with the provided timer to all tracing events, as well as a simulation
/// timestamp for simulation events.
pub fn with_custom_timer_always(sys_timer: T) -> Self {
Self { sys_timer }
}
}
impl<const VERBOSE: bool, T: FormatTime> FormatTime for SimulationTime<VERBOSE, T> {
fn format_time(&self, w: &mut Writer<'_>) -> fmt::Result {
SIMULATION_CONTEXT
.map(|ctx| {
if VERBOSE {
self.sys_timer.format_time(w)?;
w.write_char(' ')?;
}
write!(w, "[{:.9}]", ctx.time_reader.try_read().unwrap())
})
.unwrap_or_else(|| self.sys_timer.format_time(w))
}
}