Basic machinery for context carryover on task spawning

Cargo.lock

@@ -4094,28 +4094,6 @@ version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
 
-[[package]]
-name = "stuck-why"
-version = "0.1.0"
-dependencies = [
- "async-fn-stream",
- "camino",
- "clap",
- "ctrlc",
- "futures",
- "itertools",
- "never",
- "orchid-api",
- "orchid-async-utils",
- "orchid-base",
- "orchid-extension",
- "orchid-host",
- "stacker",
- "substack",
- "tokio",
- "tokio-util",
-]
-
 [[package]]
 name = "subsecond"
 version = "0.7.9"

Cargo.toml

@@ -3,7 +3,6 @@ resolver = "2"
 
 members = [
   "orcx",
-  "stuck-why",
   "orchid-std",
   "orchid-host",
   "orchid-extension",

orchid-base/src/spawn.rs

@@ -1,18 +1,53 @@
-use std::any::Any;
 #[cfg(feature = "tokio")]
 use std::cell::RefCell;
 use std::marker::PhantomData;
 use std::rc::Rc;
 use std::time::Duration;
+/// TODO: scope logic
+///
+/// A list of async HOFs that you apply on a new task
+/// Total list accessible on spawn
+/// Can be extended from the same HOFs
+use std::{any::Any, cell::OnceCell};
 
 #[cfg(feature = "tokio")]
 use async_event::Event;
 use futures::future::LocalBoxFuture;
 use orchid_async_utils::debug::{Label, with_label};
-use scoped_value::{ScopeExt, WithRef};
+use scoped_value::{ScopeExt, StackExt, WithRef};
 
 use crate::on_drop;
 
+trait DynContextLayer {
+	fn apply<'a>(&'a self, inner: LocalBoxFuture<'a, ()>) -> LocalBoxFuture<'a, ()>;
+}
+impl<F: for<'a> Fn(LocalBoxFuture<'a, ()>) -> LocalBoxFuture<'a, ()>> DynContextLayer for F {
+	fn apply<'a>(&'a self, inner: LocalBoxFuture<'a, ()>) -> LocalBoxFuture<'a, ()> { self(inner) }
+}
+
+#[derive(Clone)]
+pub struct ContextLayer(Rc<dyn DynContextLayer>);
+impl ContextLayer {
+	pub fn new<F: for<'a> Fn(LocalBoxFuture<'a, ()>) -> LocalBoxFuture<'a, ()> + 'static>(
+		layer: F,
+	) -> Self {
+		Self(Rc::new(layer))
+	}
+	pub async fn apply<F: Future>(&self, fut: F) -> F::Output {
+		let mut c = OnceCell::default();
+		self.0.apply(Box::pin(async { c.set(fut.await).unwrap_or_else(|_| unreachable!()) })).await;
+		c.take().expect("context layer appears not to have executed its argument")
+	}
+}
+
+thread_local! {
+	static CONTEXT: RefCell<Vec<ContextLayer>> = const { RefCell::new(Vec::new()) };
+}
+
+pub async fn with_context<F: Future>(cl: ContextLayer, fut: F) -> F::Output {
+	CONTEXT.async_frame(cl.clone(), cl.apply(fut)).await.0
+}
+
 /// Handle for a task that gets polled independently from the codepath that
 /// spawned it. Wrapper for a [DynTaskHandle]
 pub struct TaskHandle<T>(Box<dyn DynTaskHandle>, PhantomData<T>);
@@ -20,9 +55,8 @@ impl<T: 'static> TaskHandle<T> {
 	/// Immediately stop working on the task. Unlike in Tokio's abort, this is a
 	/// guarantee
 	pub fn abort(self) { self.0.abort(); }
-	/// Stop working on the task and return the nested future. The distinction
-	/// between this and waiting until the task is complete without reparenting it
-	/// is significant for the purpose of [scoped_value] context
+	/// Wait for the task to complete without reparenting it. This is significant
+	/// for the purpose of [scoped_value] context
 	pub async fn join(self) -> T { *self.0.join().await.downcast().unwrap() }
 }
 
@@ -46,8 +80,13 @@ pub trait SpawnSrv {
 	fn sleep_obj(&self, delay: Duration) -> LocalBoxFuture<'static, ()>;
 	/// Spawn a future off-task, but in the same thread.
 	///
-	/// The implementor should not exit while there are pending tasks to allow
-	/// external communication channels to cleanly shut down.
+	/// The implementor should not exit while there are pending tasks with
+	/// keepalive set to true to allow external communication channels to cleanly
+	/// shut down. Conversely, the caller should ensure that all keepalive tasks
+	/// have a definite join or abort point
+	///
+	/// This function does not deal with context. If you call it, you get a
+	/// "naked" task without any of your DI configuration registered in task scope
 	fn spawn_obj(
 		&self,
 		keepalive: bool,
@@ -55,31 +94,45 @@ pub trait SpawnSrv {
 	) -> Box<dyn DynTaskHandle>;
 }
 
+fn spawn_inner<F: Future + 'static>(
+	srv: &(impl SpawnSrv + ?Sized),
+	keepalive: bool,
+	spawn_lbl: &'static str,
+	fut: F,
+) -> TaskHandle<F::Output> {
+	let mut cx = CONTEXT.get();
+	let spawner_label = Label.to_string();
+	let th_obj = srv.spawn_obj(
+		keepalive,
+		Box::pin(async move {
+			// Apply
+			CONTEXT
+				.async_scope(&mut cx, async {
+					let val = CONTEXT.with_ref(|v| {
+						(v.iter().cloned().rev())
+							.fold::<LocalBoxFuture<'_, _>, _>(Box::pin(fut), |fut, cx_lvl| {
+								Box::pin(async move { cx_lvl.apply(fut).await })
+							})
+					});
+					Box::new(with_label(&format!("{spawn_lbl}[{spawner_label}]"), val).await) as _
+				})
+				.await
+		}),
+	);
+	TaskHandle(th_obj, PhantomData)
+}
+
 /// Extension functions to [SpawnSrv] to conceal all the wrapping
 pub trait SpawnExt: SpawnSrv {
 	/// Resolve after some time has elapsed
 	fn sleep(&self, delay: Duration) -> impl Future<Output = ()> + 'static { self.sleep_obj(delay) }
 	/// Start a new task on the same thread
 	fn spawn<F: Future + 'static>(&self, fut: F) -> TaskHandle<F::Output> {
-		let spawner_label = Label.to_string();
-		let th_obj = self.spawn_obj(
-			true,
-			Box::pin(
-				async move { Box::new(with_label(&format!("spawn[{spawner_label}]"), fut).await) as _ },
-			),
-		);
-		TaskHandle(th_obj, PhantomData)
+		spawn_inner(self, true, "spawn", fut)
 	}
 	/// Start a new task on the same thread
 	fn spawn_weak<F: Future + 'static>(&self, fut: F) -> TaskHandle<F::Output> {
-		let spawner_label = Label.to_string();
-		let th_obj = self.spawn_obj(
-			false,
-			Box::pin(async move {
-				Box::new(with_label(&format!("spawn_weak[{spawner_label}]"), fut).await) as _
-			}),
-		);
-		TaskHandle(th_obj, PhantomData)
+		spawn_inner(self, true, "spawn_weak", fut)
 	}
 }
 impl<T: ?Sized + SpawnSrv> SpawnExt for T {}
@@ -100,57 +153,66 @@ pub fn spawn_weak<F: Future + 'static>(fut: F) -> TaskHandle<F::Output> {
 }
 
 #[cfg(feature = "tokio")]
-#[derive(Clone)]
-pub struct TokioSpawn {
-	notifier: Rc<Event>,
-	task_counter: Rc<RefCell<usize>>,
-}
-#[cfg(feature = "tokio")]
-impl TokioSpawn {
-	pub fn new() -> Self {
-		Self { notifier: Rc::new(Event::new()), task_counter: Rc::new(RefCell::new(0)) }
-	}
-	pub fn wait_for(&self) -> impl Future<Output = ()> {
-		self.notifier.wait_until(|| (*self.task_counter.borrow() == 0).then_some(()))
-	}
-}
+pub mod tokio {
+	use std::any::Any;
+	use std::cell::RefCell;
+	use std::rc::Rc;
+	use std::time::Duration;
 
-#[cfg(feature = "tokio")]
-impl Default for TokioSpawn {
-	fn default() -> Self { Self::new() }
-}
-#[cfg(feature = "tokio")]
-impl SpawnSrv for TokioSpawn {
-	fn sleep_obj(&self, delay: Duration) -> LocalBoxFuture<'static, ()> {
-		Box::pin(tokio::time::sleep(delay))
-	}
-	fn spawn_obj(
-		&self,
-		keepalive: bool,
-		fut: LocalBoxFuture<'static, Box<dyn Any>>,
-	) -> Box<dyn DynTaskHandle> {
-		let spawner = self.clone();
+	use async_event::Event;
+	use futures::future::LocalBoxFuture;
 
-		let on_task_drop = if keepalive {
-			*self.task_counter.borrow_mut() += 1;
-			self.notifier.notify_all();
-			Some(on_drop(move || {
-				*spawner.task_counter.borrow_mut() -= 1;
-				spawner.notifier.notify_all();
-			}))
-		} else {
-			None
-		};
-		struct TokioHandle(tokio::task::JoinHandle<Box<dyn Any>>);
-		return Box::new(TokioHandle(tokio::task::spawn_local(async move {
-			let r = fut.await;
-			std::mem::drop(on_task_drop);
-			r
-		})));
-		impl DynTaskHandle for TokioHandle {
-			fn abort(self: Box<Self>) { self.0.abort(); }
-			fn join(self: Box<Self>) -> LocalBoxFuture<'static, Box<dyn Any>> {
-				Box::pin(async { self.0.await.expect("off-task panicked") })
+	use crate::{DynTaskHandle, on_drop};
+
+	#[derive(Clone)]
+	pub struct TokioSpawn {
+		notifier: Rc<Event>,
+		task_counter: Rc<RefCell<usize>>,
+	}
+
+	impl TokioSpawn {
+		pub fn new() -> Self {
+			Self { notifier: Rc::new(Event::new()), task_counter: Rc::new(RefCell::new(0)) }
+		}
+		pub fn wait_for(&self) -> impl Future<Output = ()> {
+			self.notifier.wait_until(|| (*self.task_counter.borrow() == 0).then_some(()))
+		}
+	}
+	impl Default for TokioSpawn {
+		fn default() -> Self { Self::new() }
+	}
+	impl super::SpawnSrv for TokioSpawn {
+		fn sleep_obj(&self, delay: Duration) -> LocalBoxFuture<'static, ()> {
+			Box::pin(tokio::time::sleep(delay))
+		}
+		fn spawn_obj(
+			&self,
+			keepalive: bool,
+			fut: LocalBoxFuture<'static, Box<dyn Any>>,
+		) -> Box<dyn DynTaskHandle> {
+			let spawner = self.clone();
+
+			let on_task_drop = if keepalive {
+				*self.task_counter.borrow_mut() += 1;
+				self.notifier.notify_all();
+				Some(on_drop(move || {
+					*spawner.task_counter.borrow_mut() -= 1;
+					spawner.notifier.notify_all();
+				}))
+			} else {
+				None
+			};
+			struct TokioHandle(tokio::task::JoinHandle<Box<dyn Any>>);
+			return Box::new(TokioHandle(tokio::task::spawn_local(async move {
+				let r = fut.await;
+				std::mem::drop(on_task_drop);
+				r
+			})));
+			impl DynTaskHandle for TokioHandle {
+				fn abort(self: Box<Self>) { self.0.abort(); }
+				fn join(self: Box<Self>) -> LocalBoxFuture<'static, Box<dyn Any>> {
+					Box::pin(async { self.0.await.expect("off-task panicked") })
+				}
 			}
 		}
 	}

orchid-extension/src/tokio.rs

@@ -1,3 +1,6 @@
+#[cfg(feature = "tokio")]
+use orchid_base::tokio::TokioSpawn;
+
 #[cfg(feature = "tokio")]
 use crate::ExtensionBuilder;
 
@@ -12,8 +15,8 @@ use crate::ExtensionBuilder;
 pub async fn __tokio_entrypoint(builder: ExtensionBuilder) {
 	use std::rc::Rc;
 
+	use orchid_base::SpawnExt;
 	use orchid_base::deadlock_probe::init_spinlock_probe;
-	use orchid_base::{SpawnExt, TokioSpawn};
 	use tokio::io::{stderr, stdin, stdout};
 	use tokio::task::LocalSet;
 	use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};

orcx/src/main.rs

@@ -1,7 +1,8 @@
 use never::Never;
 use orchid_async_utils::debug::with_label;
 use orchid_base::deadlock_probe::init_spinlock_probe;
-use orchid_base::{IStr, Logger, Receipt, ReqHandle, TokioSpawn};
+use orchid_base::tokio::TokioSpawn;
+use orchid_base::{IStr, Logger, Receipt, ReqHandle};
 use orchid_extension::{self as ox, OrcReader, OrcWriter};
 use orchid_host::cmd_system::{CmdEvent, CmdRunner};
 use orchid_host::dylib::ext_dylib;

scoped-value/src/lib.rs

@@ -33,6 +33,7 @@
 //! [std::thread::LocalKey], but you can implement it for your own types via the
 //! traits [WithRef] and [WithMut]
 use std::cell::RefCell;
+use std::marker::PhantomData;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::thread::LocalKey;
@@ -125,6 +126,65 @@ impl<'a, W: WithMut<'a, T> + ?Sized, T, F: Future> Future for ScopeFuture<'a, '_
 	}
 }
 
+/// Extension trait for pushing a value onto a shared vec just for the duration
+/// of a call
+pub trait StackExt<'a, T: 'a>: WithMut<'a, Vec<T>> {
+	/// Push the given value onto the stack, then pop it back off on return or
+	/// panic
+	fn sync_frame<R, F: FnOnce() -> R>(&'a self, val: T, f: F) -> (R, T) {
+		self.with_mut(|v| v.push(val));
+		let drop_trigger = StackGuard { local: Some(self), _phantom: PhantomData };
+		let ret = f();
+		let val = drop_trigger.pop();
+		(ret, val)
+	}
+	/// Swap a value into this store each time the returned future is polled, and
+	/// swap it back out when the poll returns
+	fn async_frame<Fut: Future>(
+		&'a self,
+		value: T,
+		inner: Fut,
+	) -> impl Future<Output = (Fut::Output, T)> {
+		StackFuture { key: self, value: Some(value), inner }
+	}
+}
+impl<'a, W: WithMut<'a, Vec<T>> + ?Sized, T: 'a> StackExt<'a, T> for W {}
+
+struct StackGuard<'a, W: WithMut<'a, Vec<T>> + ?Sized, T> {
+	local: Option<&'a W>,
+	_phantom: PhantomData<&'a T>,
+}
+impl<'a, W: WithMut<'a, Vec<T>> + ?Sized, T> StackGuard<'a, W, T> {
+	fn pop(mut self) -> T { self.local.take().unwrap().with_mut(|v| v.pop().unwrap()) }
+}
+impl<'a, W: WithMut<'a, Vec<T>> + ?Sized, T> Drop for StackGuard<'a, W, T> {
+	fn drop(&mut self) {
+		if let Some(w) = self.local {
+			w.with_mut(|v| v.pop());
+		}
+	}
+}
+
+struct StackFuture<'a, W: StackExt<'a, T> + ?Sized, T: 'a, F> {
+	key: &'a W,
+	value: Option<T>,
+	inner: F,
+}
+impl<'a, W: StackExt<'a, T> + ?Sized, T: 'a, F: Future> Future for StackFuture<'a, W, T, F> {
+	type Output = (F::Output, T);
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+		let Self { inner, key, value } = unsafe { Pin::get_unchecked_mut(self) };
+		let inner = unsafe { Pin::new_unchecked(inner) };
+		let init_val = value.take().expect("polled after Ready");
+		let (poll, new_val) = key.sync_frame(init_val, || inner.poll(cx));
+		if let Poll::Ready(out) = poll {
+			return Poll::Ready((out, new_val));
+		}
+		*value = Some(new_val);
+		Poll::Pending
+	}
+}
+
 #[cfg(test)]
 mod test {
 	use std::panic::{AssertUnwindSafe, catch_unwind, panic_any};

stderr

@@ -87,6 +87,7 @@ host writer blocked
 thread '<unnamed>' (220189) panicked at orchid-base/src/spawn.rs:94:48:
 cannot access a task-local storage value without setting it first
 stack backtrace:
+
    0: __rustc::rust_begin_unwind
              at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panicking.rs:689:5
    1: core::panicking::panic_fmt

stuck-why/Cargo.toml

@@ -1,28 +0,0 @@
-[package]
-name = "stuck-why"
-version = "0.1.0"
-edition = "2024"
-authors = ["Lawrence Bethlenfalvy <lbfalvy@protonmail.com>"]
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
-async-fn-stream = { version = "0.1.0", path = "../async-fn-stream" }
-camino = "1.2.4"
-clap = { version = "4.6.1", features = ["derive", "env", "cargo"] }
-ctrlc = "3.5.2"
-futures = "0.3.32"
-itertools = "0.15.0"
-never = "0.1.0"
-orchid-api = { version = "0.1.0", path = "../orchid-api" }
-orchid-async-utils = { version = "0.1.0", path = "../orchid-async-utils" }
-orchid-base = { version = "0.1.0", path = "../orchid-base" }
-orchid-extension = { version = "0.1.0", path = "../orchid-extension" }
-orchid-host = { version = "0.1.0", path = "../orchid-host", features = [
-  "tokio",
-  "orchid-extension",
-] }
-stacker = "0.1.24"
-substack = "1.1.1"
-tokio = { version = "1.52.3", features = ["full"] }
-tokio-util = { version = "0.7.18", features = ["compat"] }

stuck-why/src/main.rs

@@ -1,159 +0,0 @@
-use std::cell::RefCell;
-use std::collections::HashMap;
-use std::pin::pin;
-use std::process::ExitCode;
-use std::rc::Rc;
-use std::time::Duration;
-
-use clap::Parser;
-use futures::io;
-use itertools::Itertools;
-use orchid_async_utils::debug::{on_wake, with_label, wrap_poll};
-use orchid_base::deadlock_probe::init_spinlock_probe;
-use orchid_base::local_interner::local_interner;
-use orchid_base::{Logger, TokioSpawn, fmt, with_interner, with_logger, with_stash};
-use orchid_host::cx::Cx;
-use orchid_host::logger::LoggerImpl;
-use tokio::task::LocalSet;
-use tokio::time::Instant;
-
-/// Native interpreter for the Orchid programming language
-#[derive(Parser, Debug)]
-#[command(version, about, long_about, verbatim_doc_comment)]
-pub struct Args {
-	/// Send a log stream to a specific destination.
-	///
-	/// Supported formats:
-	/// - `--logs=msg:messaging.log`: the log channel `msg` will be routed to the
-	///   file `messaging.log`. If this is used, the following format should also
-	///   appear to specify what happens to other categories
-	/// - `--logs=orchid.log`: all unspecified channels will be routed to the file
-	///   `orchid.log`
-	/// - `--logs` only once with no value: all unspecified channels will be
-	///   routed to stderr
-	///
-	/// Some destination names have special meanings:
-	/// - `stderr` designates the platform-specific standard error output of the
-	///   interpreter process
-	/// - `off` discards received messages
-	///
-	/// Defaults for specific channels
-	/// - `warn` is routed the same as `debug`
-	/// - `error` is routed the same as `warn`
-	/// - `msg` is discarded (routed to `off`)
-	#[arg(
-		short,
-		long,
-		default_value = "off",
-		default_missing_value = "stderr",
-		next_line_help = true,
-		verbatim_doc_comment
-	)]
-	log: Vec<String>,
-}
-
-static mut STARTUP: Option<Instant> = None;
-
-fn parse_log_dest(dest: &str) -> orchid_api::LogStrategy {
-	if dest == "off" {
-		orchid_api::LogStrategy::Discard
-	} else if dest == "stderr" {
-		orchid_api::LogStrategy::Default
-	} else if let Some(path) = dest.strip_prefix('>') {
-		orchid_api::LogStrategy::File { path: path.to_string(), append: true }
-	} else {
-		orchid_api::LogStrategy::File { path: dest.to_string(), append: false }
-	}
-}
-
-fn get_logger(args: &Args) -> LoggerImpl {
-	let mut logger = LoggerImpl::default();
-	for cmd in &args.log {
-		match cmd.split_once(":") {
-			None => {
-				let strat = parse_log_dest(cmd);
-				logger.set_default(strat)
-			},
-			Some((category, dest)) => {
-				let strat = parse_log_dest(dest);
-				logger.set_category(category, strat)
-			},
-		}
-	}
-	if !logger.has_category("warn") {
-		logger.set_category("warn", logger.strat("debug"));
-	}
-	if !logger.has_category("error") {
-		logger.set_category("error", logger.strat("warn"));
-	}
-	if !logger.has_category("msg") {
-		logger.set_category("msg", orchid_api::LogStrategy::Discard);
-	}
-	logger
-}
-
-fn main() -> io::Result<ExitCode> {
-	// Use a 10MB stack for single-threaded, unoptimized operation
-	stacker::grow(10 * 1024 * 1024, || {
-		tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(wrap_poll(
-			on_wake(
-				with_label("host", async {
-					let args = Args::parse();
-					eprintln!("Orcx v{} is free software provided without warranty.", clap::crate_version!());
-					let exit_code = Rc::new(RefCell::new(ExitCode::SUCCESS));
-					let local_set = LocalSet::new();
-					let logger = get_logger(&args);
-					let logger2 = logger.clone();
-					unsafe { STARTUP = Some(Instant::now()) };
-					let spawner = Rc::new(TokioSpawn::new());
-					let cx = Cx::new(spawner.clone(), logger2);
-					let (signal_end_main, on_end_main) = futures::channel::oneshot::channel();
-					let cx1 = cx.clone();
-					local_set.spawn_local(async move {
-						let cx = &cx1;
-						init_spinlock_probe(cx.spawn(), "Orcx main");
-						with_stash(async move { cx.sleep(Duration::from_secs(20)).await }).await;
-						signal_end_main.send(()).expect("cleanup should still be waiting");
-					});
-					let cleanup = async {
-						if on_end_main.await.is_err() {
-							return;
-						}
-						tokio::time::sleep(Duration::from_secs(2)).await;
-						let mut extensions = HashMap::new();
-						let systems =
-							cx.systems.read().await.values().filter_map(|v| v.upgrade()).collect_vec();
-						let exprs = cx.exprs.iter().collect_vec();
-						for system in &systems {
-							extensions.insert(system.ext().name().clone(), system.ext().clone());
-						}
-						if extensions.is_empty() && systems.is_empty() && exprs.is_empty() {
-							return;
-						}
-						eprintln!("Shutdown is taking long. The following language constructs are still live:");
-						eprintln!("Extensions: {}", extensions.keys().join(", "));
-						for sys in &systems {
-							eprintln!("System: {:?} = {}", sys.id(), sys.ctor().name())
-						}
-						for (rc, expr) in &exprs {
-							eprintln!("{rc}x {:?} = {}", expr.id(), fmt(expr).await)
-						}
-						std::process::abort()
-					};
-					futures::future::select(
-						pin!(cleanup),
-						pin!(with_interner(local_interner(), with_logger(logger, local_set))),
-					)
-					.await;
-					let x = *exit_code.borrow();
-					Ok(x)
-				}),
-				|| eprintln!("Woke host"),
-			),
-			|cb| {
-				eprintln!("Polling host now");
-				cb();
-			},
-		))
-	})
-}