use core::future::Future;
use std::fmt;
use std::io;
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use kovan_channel::flavors::RecvDeadline;
use super::blocking::Pool;
use super::context;
use super::sched::Sched;
use super::shard::{Shard, Shared};
use super::task::{self, ScheduleInfo, Task};
use super::workers::Workers;
mod builder;
mod cores;
mod enter;
mod fallible;
pub(crate) mod io_knobs;
mod knobs;
mod panic;
mod placement;
mod stats;
pub use self::builder::Builder;
pub use self::enter::{EnterGuard, RuntimeId, TryCurrentError, worker_index};
pub(crate) use self::fallible::{try_current, try_spawn, try_spawn_blocking, try_spawn_local};
pub use super::rng::RngSeed;
pub use super::tuning::{TaskMeta, UnhandledPanic};
pub(crate) struct Inner {
id: u64,
pub(crate) workers: Workers,
pub(crate) pool: Arc<Pool>,
local_margin: usize,
#[cfg(ring)]
rings: bool,
pub(crate) clock: crate::time::clock::Clock,
guests: OnceLock<kovan_map::HashMap<usize, Arc<Shared>>>,
task_hooks: Option<Arc<super::tuning::TaskHooks>>,
shutdown_on_panic: bool,
panicked: std::sync::atomic::AtomicBool,
on_panic: crate::sync::Notify,
rng_seed: Option<RngSeed>,
lingering: super::linger::Lingering,
pub(crate) io: io_knobs::IoKnobs,
}
pub struct Runtime {
inner: Arc<Inner>,
exits: kovan_channel::unbounded::Receiver<()>,
}
#[derive(Clone)]
pub struct RuntimeHandle {
inner: Arc<Inner>,
}
impl Runtime {
pub fn builder() -> Builder {
Builder::new()
}
pub fn new() -> io::Result<Self> {
Builder::new().build()
}
pub fn handle(&self) -> RuntimeHandle {
RuntimeHandle {
inner: Arc::clone(&self.inner),
}
}
pub fn block_on<F: Future>(&self, future: F) -> F::Output {
block_on_with(&self.inner, future)
}
pub fn spawn<F>(&self, future: F) -> Task<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
spawn_with(&self.inner, future)
}
pub fn spawn_blocking<F, T>(&self, f: F) -> Task<T>
where
F: FnOnce() -> T + Send + 'static,
T: Send + 'static,
{
self.inner.pool.spawn(f)
}
pub fn shutdown_timeout(mut self, timeout: Duration) {
let deadline = std::time::Instant::now() + timeout;
self.stop_and_join(Some(timeout));
self.finish_lingering(Some(deadline));
}
fn finish_lingering(&self, deadline: Option<std::time::Instant>) {
if self.inner.lingering.is_over() {
return;
}
if !self.inner.lingers_done() {
let linger = std::time::Instant::now() + self.inner.io.linger;
let deadline = deadline.map_or(linger, |d| d.min(linger));
let _ = block_on_here(
&self.inner,
crate::time::timeout_at(deadline, self.inner.run_lingers()),
);
}
self.inner.lingering.end();
}
fn stop_and_join(&mut self, timeout: Option<Duration>) {
self.inner.lingering.close();
self.inner.workers.stop_all();
let threads = self.inner.workers.take_threads();
if let Some(timeout) = timeout {
let deadline = std::time::Instant::now() + timeout;
for thread in &threads {
if thread.is_finished() {
continue;
}
if !matches!(self.exits.recv_deadline(deadline), RecvDeadline::Msg(())) {
break;
}
}
for thread in threads {
if thread.is_finished() {
let _ = thread.join();
}
}
} else {
for thread in threads {
let _ = thread.join();
}
}
}
}
#[cfg(test)]
impl Runtime {
pub(crate) fn inner(&self) -> &Arc<Inner> {
&self.inner
}
}
impl Drop for Runtime {
fn drop(&mut self) {
self.stop_and_join(None);
self.finish_lingering(None);
}
}
impl fmt::Debug for Runtime {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Runtime")
.field("worker_threads", &self.inner.workers.running())
.finish_non_exhaustive()
}
}
impl RuntimeHandle {
pub fn block_on<F: Future>(&self, future: F) -> F::Output {
block_on_with(&self.inner, future)
}
pub fn spawn<F>(&self, future: F) -> Task<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
spawn_with(&self.inner, future)
}
pub fn spawn_blocking<F, T>(&self, f: F) -> Task<T>
where
F: FnOnce() -> T + Send + 'static,
T: Send + 'static,
{
self.inner.pool.spawn(f)
}
pub fn current() -> Self {
Self { inner: current() }
}
pub(crate) fn inner(&self) -> &Arc<Inner> {
&self.inner
}
}
impl fmt::Debug for RuntimeHandle {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("RuntimeHandle")
.field("worker_threads", &self.inner.workers.running())
.finish_non_exhaustive()
}
}
static DEFAULT: OnceLock<Runtime> = OnceLock::new();
impl Inner {
pub(crate) fn guests(&self) -> &kovan_map::HashMap<usize, Arc<Shared>> {
self.guests
.get_or_init(|| kovan_map::HashMap::with_capacity(builder::GUESTS_CAPACITY))
}
pub(crate) fn guests_made(&self) -> Option<&kovan_map::HashMap<usize, Arc<Shared>>> {
self.guests.get()
}
pub(crate) fn id(&self) -> u64 {
self.id
}
pub(crate) fn lingering(&self) -> &super::linger::Lingering {
&self.lingering
}
}
#[expect(
clippy::panic,
reason = "no caller to hand the error to; documented under `# Panics`"
)]
pub(crate) fn default_runtime() -> &'static Runtime {
fallible::try_default()
.unwrap_or_else(|err| panic!("the default runtime could not start: {err}"))
}
pub(crate) fn current() -> Arc<Inner> {
context::runtime().unwrap_or_else(|| Arc::clone(&default_runtime().inner))
}
pub(crate) fn with_clock<R>(f: impl FnOnce(Option<&crate::time::clock::Clock>) -> R) -> R {
match context::runtime() {
Some(inner) => f(Some(&inner.clock)),
None => f(DEFAULT.get().map(|rt| &rt.inner.clock)),
}
}
fn running() -> Option<Arc<Inner>> {
context::runtime().or_else(|| DEFAULT.get().map(|rt| Arc::clone(&rt.inner)))
}
pub(crate) fn running_inner() -> Option<Arc<Inner>> {
running()
}
pub(crate) fn spawn_linger(
inner: &Arc<Inner>,
linger: super::linger::Guarded,
here: bool,
) -> Result<(), super::linger::Guarded> {
if let Some(shard) = inner.workers.place(inner.id, inner.local_margin) {
spawn_on(inner, shard, linger).detach();
return Ok(());
}
let running = context::current(|current| current.is_some());
if here && (running || context::own_shard_if_made().is_some()) {
let _runtime = context::set_runtime(inner);
spawn_here(linger, false).detach();
return Ok(());
}
Err(linger)
}
pub(crate) fn running_service_shard() -> Option<Arc<Shared>> {
let inner = running()?;
inner
.workers
.place(inner.id, inner.local_margin)
.map(Arc::clone)
}
#[cfg(ring)]
pub(crate) fn service_rings() -> bool {
let inner = current();
if inner.workers.running() > 0 {
inner.rings
} else {
super::backend::chosen().is_ok_and(|b| b.rings())
}
}
pub(crate) fn service_shard() -> Arc<Shared> {
let inner = current();
match inner.workers.place(inner.id, inner.local_margin) {
Some(shard) => Arc::clone(shard),
None => Arc::clone(&own_shard().shared),
}
}
#[expect(
clippy::panic,
reason = "no caller to hand the error to; documented under `# Panics`"
)]
fn own_shard() -> std::rc::Rc<Shard> {
match context::own_shard() {
Ok(shard) => shard,
Err(err) => panic!("this thread's shard could not start: {err}"),
}
}
#[inline(always)]
fn spawn_with<F>(inner: &Inner, future: F) -> Task<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
let Some(shard) = inner.workers.place(inner.id, inner.local_margin) else {
return spawn_here(future, false);
};
spawn_on(inner, shard, future)
}
#[inline(always)]
fn spawn_on<F>(inner: &Inner, shard: &Arc<Shared>, future: F) -> Task<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
let hooks = inner.task_hooks.as_ref();
let (runnable, task) = task::spawn(
future,
Sched::counted(Arc::clone(shard), None, hooks.cloned()),
);
spawned(hooks.map(|hooks| &**hooks), &task);
if let Some(runnable) = super::sched::queue_here(shard, runnable, ScheduleInfo::new(false)) {
runnable.schedule();
}
task
}
#[inline(always)]
fn spawned<T>(hooks: Option<&super::tuning::TaskHooks>, task: &Task<T>) {
if let Some(hooks) = hooks {
spawned_with(hooks, task.id());
}
}
#[cold]
#[inline(never)]
fn spawned_with(hooks: &super::tuning::TaskHooks, id: task::Id) {
crate::task::name_now(id);
super::tuning::call(hooks.spawn.as_ref(), id);
}
fn context_hooks() -> Option<Arc<super::tuning::TaskHooks>> {
if !super::tuning::ANY_TASK_HOOKS.load(std::sync::atomic::Ordering::Acquire) {
return None;
}
context::runtime().and_then(|inner| inner.task_hooks.clone())
}
fn spawn_here<F>(future: F, bound: bool) -> Task<F::Output>
where
F: Future + 'static,
F::Output: 'static,
{
let place = |shard: &Shard| {
(
Arc::clone(&shard.shared),
bound.then(|| context::home_for(shard)),
)
};
let (shared, home) =
context::current(|current| current.map(place)).unwrap_or_else(|| place(&own_shard()));
let hooks = context_hooks();
let (runnable, task, bound_to) = if let Some(home) = home {
let at = Arc::as_ptr(&home);
let (runnable, task) = unsafe {
task::spawn_unchecked(future, Sched::counted(shared, Some(home), hooks.clone()))
};
(runnable, task, Some(at))
} else {
let (runnable, task) =
unsafe { task::spawn_unchecked(future, Sched::counted(shared, None, hooks.clone())) };
(runnable, task, None)
};
spawned(hooks.as_deref(), &task);
let left = match bound_to {
Some(home) => {
let home = unsafe { &*home };
super::sched::queue_home_here(home, runnable, ScheduleInfo::new(false))
}
None => Some(runnable),
};
if let Some(runnable) = left {
runnable.schedule();
}
task
}
fn block_on_with<F: Future>(inner: &Arc<Inner>, future: F) -> F::Output {
if inner.shutdown_on_panic {
return block_on_here(inner, panic::watched(inner, future));
}
block_on_here(inner, future)
}
fn block_on_here<F: Future>(inner: &Arc<Inner>, future: F) -> F::Output {
let _runtime = context::set_runtime(inner);
let mut future = core::pin::pin!(future);
if let Some(output) =
context::current(|current| current.map(|shard| shard.block_on(future.as_mut())))
{
return output;
}
let shard = own_shard();
let _current = context::enter(&shard);
let _guest = super::paused::Guest::enter(inner, &shard.shared);
let _seeded = inner
.rng_seed
.map(|seed| super::rng::Reseeded::new(seed, 0));
#[cfg(port)]
shard.take_port_back();
shard.block_on_then_leave(future.as_mut())
}
pub fn block_on<F: Future>(future: F) -> F::Output {
block_on_with(¤t(), future)
}
#[inline]
pub fn spawn<F>(future: F) -> Task<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
let spawned = context::with_runtime(|inner| match inner {
Some(inner) if inner.task_hooks.is_none() => Ok(spawn_with(inner, future)),
_ => Err(future),
});
spawned.unwrap_or_else(spawn_current)
}
#[inline(never)]
fn spawn_current<F>(future: F) -> Task<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
spawn_with(¤t(), future)
}
pub fn spawn_local<F>(future: F) -> Task<F::Output>
where
F: Future + 'static,
F::Output: 'static,
{
spawn_here(future, true)
}
pub fn spawn_blocking<F, T>(f: F) -> Task<T>
where
F: FnOnce() -> T + Send + 'static,
T: Send + 'static,
{
current().pool.spawn(f)
}
#[cfg(unix)]
pub(crate) fn spawn_service<F>(future: F)
where
F: Future<Output = ()> + Send + 'static,
{
let shared = context::current(|current| current.map(|s| Arc::clone(&s.shared)))
.unwrap_or_else(service_shard);
let (runnable, task) = unsafe {
task::spawn_unchecked(future, Sched::uncounted(shared))
};
runnable.schedule();
task.detach();
}
#[cfg(port)]
pub(crate) fn spawn_blocking_inside(f: impl FnOnce() + Send + 'static) -> bool {
match blocking_pool_inside() {
Some(pool) => {
pool.spawn(f).detach();
true
}
None => false,
}
}
#[cfg(completion)]
pub(crate) fn blocking_pool_inside() -> Option<Arc<super::blocking::Pool>> {
context::runtime().map(|inner| Arc::clone(&inner.pool))
}
pub(crate) fn allowed_cores() -> Vec<usize> {
#[cfg(target_os = "linux")]
{
current_mask().map_or_else(Vec::new, |set| {
(0..libc::CPU_SETSIZE as usize)
.filter(|&cpu| unsafe { libc::CPU_ISSET(cpu, &set) })
.collect()
})
}
#[cfg(windows)]
{
use crate::syscore::windows::{GetCurrentProcess, GetProcessAffinityMask};
let (mut process, mut system) = (0usize, 0usize);
if unsafe { GetProcessAffinityMask(GetCurrentProcess(), &mut process, &mut system) } != 0 {
return (0..usize::BITS as usize)
.filter(|&cpu| process & (1 << cpu) != 0)
.collect();
}
Vec::new()
}
#[cfg(not(any(target_os = "linux", windows)))]
{
Vec::new()
}
}
#[cfg(target_os = "linux")]
pub(crate) type CpuMask = libc::cpu_set_t;
#[cfg(windows)]
pub(crate) type CpuMask = usize;
#[cfg(not(any(target_os = "linux", windows)))]
pub(crate) type CpuMask = ();
pub(crate) fn current_mask() -> Option<CpuMask> {
#[cfg(target_os = "linux")]
{
unsafe {
let mut set: libc::cpu_set_t = core::mem::zeroed();
(libc::sched_getaffinity(0, size_of::<libc::cpu_set_t>(), &mut set) == 0).then_some(set)
}
}
#[cfg(not(target_os = "linux"))]
None
}
pub(crate) fn apply_mask(mask: &CpuMask) {
#[cfg(target_os = "linux")]
let _ = unsafe { libc::sched_setaffinity(0, size_of::<libc::cpu_set_t>(), mask) };
#[cfg(windows)]
{
use crate::syscore::windows::{GetCurrentThread, SetThreadAffinityMask};
unsafe { SetThreadAffinityMask(GetCurrentThread(), *mask) };
}
#[cfg(not(any(target_os = "linux", windows)))]
let _ = mask;
}
pub(crate) fn pin_to(core: usize) {
#[cfg(target_os = "linux")]
unsafe {
let mut set: libc::cpu_set_t = core::mem::zeroed();
libc::CPU_SET(core, &mut set);
let _ = libc::sched_setaffinity(0, size_of::<libc::cpu_set_t>(), &set);
}
#[cfg(windows)]
if core < usize::BITS as usize {
use crate::syscore::windows::{GetCurrentThread, SetThreadAffinityMask};
unsafe { SetThreadAffinityMask(GetCurrentThread(), 1 << core) };
}
#[cfg(not(any(target_os = "linux", windows)))]
let _ = core;
}