Traditional route appears to work

Beginnings of dylib extensions, entirely untestted

orchid-base/Cargo.toml

@@ -13,7 +13,7 @@ bound = "0.6.0"
 derive_destructure = "1.0.0"
 dyn-clone = "1.0.20"
 futures = { version = "0.3.31", features = ["std"], default-features = false }
-hashbrown = "0.16.0"
+hashbrown = "0.16.1"
 itertools = "0.14.0"
 lazy_static = "1.5.0"
 never = "0.1.0"
@@ -21,13 +21,13 @@ num-traits = "0.2.19"
 orchid-api = { version = "0.1.0", path = "../orchid-api" }
 orchid-api-derive = { version = "0.1.0", path = "../orchid-api-derive" }
 orchid-api-traits = { version = "0.1.0", path = "../orchid-api-traits" }
-ordered-float = "5.0.0"
-regex = "1.11.2"
-rust-embed = "8.7.2"
+ordered-float = "5.1.0"
+regex = "1.12.2"
+rust-embed = "8.9.0"
 substack = "1.1.1"
 trait-set = "0.3.0"
 task-local = "0.1.0"
 
 [dev-dependencies]
 futures = "0.3.31"
-test_executors = "0.4.0"
+test_executors = "0.4.1"

orchid-base/src/binary.rs

@@ -0,0 +1,118 @@
+use std::pin::Pin;
+use std::rc::Rc;
+use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
+
+use orchid_api::binary::{FutureContextVT, FutureVT, OwnedWakerVT, UnitPoll};
+
+type WideBox = Box<dyn Future<Output = ()>>;
+
+static OWNED_VTABLE: RawWakerVTable = RawWakerVTable::new(
+	|data| {
+		let data = unsafe { Rc::<OwnedWakerVT>::from_raw(data as *const _) };
+		let val = RawWaker::new(Rc::into_raw(data.clone()) as *const (), &OWNED_VTABLE);
+		// Clone must create a duplicate of the Rc, so it has to be un-leaked, cloned,
+		// then leaked again.
+		let _ = Rc::into_raw(data);
+		val
+	},
+	|data| {
+		// Wake must awaken the task and then clean up the state, so the waker must be
+		// un-leaked
+		let data = unsafe { Rc::<OwnedWakerVT>::from_raw(data as *const _) };
+		(data.wake)(data.data);
+	},
+	|data| {
+		// Wake-by-ref must awaken the task while preserving the future, so the Rc is
+		// untouched
+		let data = unsafe { (data as *const OwnedWakerVT).as_ref() }.unwrap();
+		(data.wake_ref)(data.data);
+	},
+	|data| {
+		// Drop must clean up the state, so the waker must be un-leaked
+		let data = unsafe { Rc::<OwnedWakerVT>::from_raw(data as *const _) };
+		(data.drop)(data.data);
+	},
+);
+
+struct BorrowedWakerData<'a> {
+	go_around: &'a mut bool,
+	cx: FutureContextVT,
+}
+static BORROWED_VTABLE: RawWakerVTable = RawWakerVTable::new(
+	|data| {
+		let data = unsafe { (data as *mut BorrowedWakerData).as_mut() }.unwrap();
+		let owned_data = Rc::<OwnedWakerVT>::new((data.cx.waker)(data.cx.data));
+		RawWaker::new(Rc::into_raw(owned_data) as *const (), &OWNED_VTABLE)
+	},
+	|data| *unsafe { (data as *mut BorrowedWakerData).as_mut() }.unwrap().go_around = true,
+	|data| *unsafe { (data as *mut BorrowedWakerData).as_mut() }.unwrap().go_around = true,
+	|_data| {},
+);
+
+/// Convert a future to a binary-compatible format that can be sent across
+/// dynamic library boundaries
+pub fn future_to_vt<Fut: Future<Output = ()> + 'static>(fut: Fut) -> FutureVT {
+	let wide_box = Box::new(fut) as WideBox;
+	let data = Box::into_raw(Box::new(wide_box));
+	extern "C" fn drop(raw: *const ()) {
+		std::mem::drop(unsafe { Box::<WideBox>::from_raw(raw as *mut _) })
+	}
+	extern "C" fn poll(raw: *const (), cx: FutureContextVT) -> UnitPoll {
+		let mut this = unsafe { Pin::new_unchecked(&mut **(raw as *mut WideBox).as_mut().unwrap()) };
+		loop {
+			let mut go_around = false;
+			let borrowed_waker = unsafe {
+				Waker::from_raw(RawWaker::new(
+					&mut BorrowedWakerData { go_around: &mut go_around, cx } as *mut _ as *const (),
+					&BORROWED_VTABLE,
+				))
+			};
+			let mut ctx = Context::from_waker(&borrowed_waker);
+			let result = this.as_mut().poll(&mut ctx);
+			if matches!(result, Poll::Ready(())) {
+				break UnitPoll::Ready;
+			}
+			if !go_around {
+				break UnitPoll::Pending;
+			}
+		}
+	}
+	FutureVT { data: data as *const _, drop, poll }
+}
+
+struct VirtualFuture {
+	vt: FutureVT,
+}
+impl Unpin for VirtualFuture {}
+impl Future for VirtualFuture {
+	type Output = ();
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+		extern "C" fn waker(raw: *const ()) -> OwnedWakerVT {
+			let waker = unsafe { (raw as *mut Context).as_mut() }.unwrap().waker().clone();
+			let data = Box::into_raw(Box::<Waker>::new(waker)) as *const ();
+			return OwnedWakerVT { data, drop, wake, wake_ref };
+			extern "C" fn drop(raw: *const ()) {
+				std::mem::drop(unsafe { Box::<Waker>::from_raw(raw as *mut Waker) })
+			}
+			extern "C" fn wake(raw: *const ()) {
+				unsafe { Box::<Waker>::from_raw(raw as *mut Waker) }.wake();
+			}
+			extern "C" fn wake_ref(raw: *const ()) {
+				unsafe { (raw as *mut Waker).as_mut() }.unwrap().wake_by_ref();
+			}
+		}
+		let cx = FutureContextVT { data: cx as *mut Context as *const (), waker };
+		let result = (self.vt.poll)(self.vt.data, cx);
+		match result {
+			UnitPoll::Pending => Poll::Pending,
+			UnitPoll::Ready => Poll::Ready(()),
+		}
+	}
+}
+impl Drop for VirtualFuture {
+	fn drop(&mut self) { (self.vt.drop)(self.vt.data) }
+}
+
+/// Receive a future sent across dynamic library boundaries and convert it into
+/// an owned object
+pub fn vt_to_future(vt: FutureVT) -> impl Future<Output = ()> { VirtualFuture { vt } }

orchid-base/src/error.rs

@@ -128,6 +128,7 @@ impl OrcErrv {
 	pub async fn from_api<'a>(api: impl IntoIterator<Item = &'a api::OrcError>) -> Self {
 		Self(join_all(api.into_iter().map(OrcErr::from_api)).await)
 	}
+	pub fn iter(&self) -> impl Iterator<Item = OrcErr> + '_ { self.0.iter().cloned() }
 }
 impl From<OrcErr> for OrcErrv {
 	fn from(value: OrcErr) -> Self { Self(vec![value]) }

orchid-base/src/lib.rs

@@ -1,6 +1,7 @@
 pub use async_once_cell;
 use orchid_api as api;
 
+pub mod binary;
 pub mod box_cow;
 pub mod boxed_iter;
 pub mod char_filter;

orchid-base/src/logging.rs

@@ -1,47 +1,74 @@
+use std::any::Any;
+use std::cell::RefCell;
 use std::fmt::Arguments;
-use std::fs::File;
-use std::io::{Write, stderr};
+use std::io::Write;
+use std::rc::Rc;
 
-pub use api::LogStrategy;
-use itertools::Itertools;
+use futures::future::LocalBoxFuture;
 use task_local::task_local;
 
 use crate::api;
 
-#[derive(Clone)]
-pub struct Logger(api::LogStrategy);
-impl Logger {
-	pub fn new(strat: api::LogStrategy) -> Self { Self(strat) }
-	pub fn log(&self, msg: impl AsRef<str>) { writeln!(self, "{}", msg.as_ref()) }
-	pub fn strat(&self) -> api::LogStrategy { self.0.clone() }
-	pub fn is_active(&self) -> bool { !matches!(self.0, api::LogStrategy::Discard) }
-	pub fn log_buf(&self, event: impl AsRef<str>, buf: &[u8]) {
-		if std::env::var("ORCHID_LOG_BUFFERS").is_ok_and(|v| !v.is_empty()) {
-			writeln!(self, "{}: [{}]", event.as_ref(), buf.iter().map(|b| format!("{b:02x}")).join(" "))
-		}
-	}
-	pub fn write_fmt(&self, fmt: Arguments) {
-		match &self.0 {
-			api::LogStrategy::Discard => (),
-			api::LogStrategy::StdErr => {
-				stderr().write_fmt(fmt).expect("Could not write to stderr!");
-				stderr().flush().expect("Could not flush stderr")
-			},
-			api::LogStrategy::File(f) => {
-				let mut file = (File::options().write(true).create(true).truncate(true).open(f))
-					.expect("Could not open logfile");
-				file.write_fmt(fmt).expect("Could not write to logfile");
-			},
-		}
+task_local! {
+	static DEFAULT_WRITER: RefCell<Box<dyn Write>>
+}
+
+/// Set the stream used for [api::LogStrategy::Default]. If not set,
+/// [std::io::stderr] will be used.
+pub async fn with_default_stream<F: Future>(stderr: impl Write + 'static, fut: F) -> F::Output {
+	DEFAULT_WRITER.scope(RefCell::new(Box::new(stderr)), fut).await
+}
+
+pub trait LogWriter {
+	fn write_fmt<'a>(&'a self, fmt: Arguments<'a>) -> LocalBoxFuture<'a, ()>;
+}
+
+pub trait Logger: Any {
+	fn writer(&self, category: &str) -> Rc<dyn LogWriter>;
+	fn strat(&self, category: &str) -> api::LogStrategy;
+	fn is_active(&self, category: &str) -> bool {
+		!matches!(self.strat(category), api::LogStrategy::Discard)
 	}
 }
 
 task_local! {
-	static LOGGER: Logger;
+	static LOGGER: Rc<dyn Logger>;
 }
 
-pub async fn with_logger<F: Future>(logger: Logger, fut: F) -> F::Output {
-	LOGGER.scope(logger, fut).await
+pub async fn with_logger<F: Future>(logger: impl Logger + 'static, fut: F) -> F::Output {
+	LOGGER.scope(Rc::new(logger), fut).await
 }
 
-pub fn logger() -> Logger { LOGGER.try_with(|l| l.clone()).expect("Logger not set!") }
+pub fn log(category: &str) -> Rc<dyn LogWriter> {
+	LOGGER.try_with(|l| l.writer(category)).expect("Logger not set!")
+}
+
+pub fn get_logger() -> Rc<dyn Logger> { LOGGER.try_with(|l| l.clone()).expect("Logger not set!") }
+
+pub mod test {
+	use std::fmt::Arguments;
+	use std::rc::Rc;
+
+	use futures::future::LocalBoxFuture;
+
+	use crate::clone;
+	use crate::logging::{LogWriter, Logger};
+
+	#[derive(Clone)]
+	pub struct TestLogger(Rc<dyn Fn(String) -> LocalBoxFuture<'static, ()>>);
+	impl LogWriter for TestLogger {
+		fn write_fmt<'a>(&'a self, fmt: Arguments<'a>) -> LocalBoxFuture<'a, ()> {
+			(self.0)(fmt.to_string())
+		}
+	}
+	impl Logger for TestLogger {
+		fn strat(&self, _category: &str) -> orchid_api::LogStrategy { orchid_api::LogStrategy::Default }
+		fn writer(&self, _category: &str) -> std::rc::Rc<dyn LogWriter> { Rc::new(self.clone()) }
+	}
+	impl TestLogger {
+		pub fn new(f: impl AsyncFn(String) + 'static) -> Self {
+			let f = Rc::new(f);
+			Self(Rc::new(move |s| clone!(f; Box::pin(async move { f(s).await }))))
+		}
+	}
+}

orchid-base/src/reqnot.rs

@@ -1,5 +1,4 @@
 use std::cell::RefCell;
-use std::future::Future;
 use std::marker::PhantomData;
 use std::pin::{Pin, pin};
 use std::rc::Rc;
@@ -316,13 +315,25 @@ impl CommCtx {
 pub fn io_comm(
 	o: Rc<Mutex<Pin<Box<dyn AsyncWrite>>>>,
 	i: Mutex<Pin<Box<dyn AsyncRead>>>,
-	notif: impl for<'a> AsyncFn(Box<dyn MsgReader<'a> + 'a>) -> io::Result<()>,
-	req: impl for<'a> AsyncFn(Box<dyn ReqReader<'a> + 'a>) -> io::Result<Receipt<'a>>,
-) -> (impl Client + 'static, CommCtx, impl Future<Output = io::Result<()>>) {
+) -> (impl Client + 'static, CommCtx, IoCommServer) {
 	let i = Rc::new(i);
 	let (onsub, client) = IoClient::new(o.clone());
 	let (exit, onexit) = channel(1);
-	(client, CommCtx { exit }, async move {
+	(client, CommCtx { exit }, IoCommServer { o, i, onsub, onexit })
+}
+pub struct IoCommServer {
+	o: Rc<Mutex<Pin<Box<dyn AsyncWrite>>>>,
+	i: Rc<Mutex<Pin<Box<dyn AsyncRead>>>>,
+	onsub: Receiver<ReplySub>,
+	onexit: Receiver<()>,
+}
+impl IoCommServer {
+	pub async fn listen(
+		self,
+		notif: impl for<'a> AsyncFn(Box<dyn MsgReader<'a> + 'a>) -> io::Result<()>,
+		req: impl for<'a> AsyncFn(Box<dyn ReqReader<'a> + 'a>) -> io::Result<Receipt<'a>>,
+	) -> io::Result<()> {
+		let Self { o, i, onexit, onsub } = self;
 		enum Event {
 			Input(u64, IoGuard<dyn AsyncRead>),
 			Sub(ReplySub),
@@ -363,6 +374,9 @@ pub fn io_comm(
 					Err(e) => break 'body Err(e),
 					Ok(Event::Exit) => {
 						*exiting.borrow_mut() = true;
+						let mut out = o.lock().await;
+						out.as_mut().flush().await?;
+						out.as_mut().close().await?;
 						break;
 					},
 					Ok(Event::Sub(ReplySub { id, ack, cb })) => {
@@ -399,11 +413,12 @@ pub fn io_comm(
 			next?
 		}
 		Ok(())
-	})
+	}
 }
 
 #[cfg(test)]
 mod test {
+	use std::cell::RefCell;
 	use std::rc::Rc;
 
 	use futures::channel::mpsc;
@@ -414,6 +429,8 @@ mod test {
 	use test_executors::spin_on;
 	use unsync_pipe::pipe;
 
+	use crate::logging::test::TestLogger;
+	use crate::logging::with_logger;
 	use crate::reqnot::{ClientExt, MsgReaderExt, ReqReaderExt, io_comm};
 
 	#[derive(Clone, Debug, PartialEq, Coding, Hierarchy)]
@@ -422,33 +439,36 @@ mod test {
 
 	#[test]
 	fn notification() {
-		spin_on(async {
+		let logger = TestLogger::new(async |s| eprint!("{s}"));
+		spin_on(with_logger(logger, async {
 			let (in1, out2) = pipe(1024);
 			let (in2, out1) = pipe(1024);
 			let (received, mut on_receive) = mpsc::channel(2);
-			let (_, recv_ctx, run_recv) = io_comm(
-				Rc::new(Mutex::new(Box::pin(in2))),
-				Mutex::new(Box::pin(out2)),
-				async |notif| {
-					received.clone().send(notif.read::<TestNotif>().await?).await.unwrap();
-					Ok(())
+			let (_, recv_ctx, recv_srv) =
+				io_comm(Rc::new(Mutex::new(Box::pin(in2))), Mutex::new(Box::pin(out2)));
+			let (sender, ..) = io_comm(Rc::new(Mutex::new(Box::pin(in1))), Mutex::new(Box::pin(out1)));
+			join!(
+				async {
+					recv_srv
+						.listen(
+							async |notif| {
+								received.clone().send(notif.read::<TestNotif>().await?).await.unwrap();
+								Ok(())
+							},
+							async |_| panic!("Should receive notif, not request"),
+						)
+						.await
+						.unwrap()
 				},
-				async |_| panic!("Should receive notif, not request"),
+				async {
+					sender.notify(TestNotif(3)).await.unwrap();
+					assert_eq!(on_receive.next().await, Some(TestNotif(3)));
+					sender.notify(TestNotif(4)).await.unwrap();
+					assert_eq!(on_receive.next().await, Some(TestNotif(4)));
+					recv_ctx.exit().await;
+				}
 			);
-			let (sender, ..) = io_comm(
-				Rc::new(Mutex::new(Box::pin(in1))),
-				Mutex::new(Box::pin(out1)),
-				async |_| panic!("Should not receive notif"),
-				async |_| panic!("Should not receive request"),
-			);
-			join!(async { run_recv.await.unwrap() }, async {
-				sender.notify(TestNotif(3)).await.unwrap();
-				assert_eq!(on_receive.next().await, Some(TestNotif(3)));
-				sender.notify(TestNotif(4)).await.unwrap();
-				assert_eq!(on_receive.next().await, Some(TestNotif(4)));
-				recv_ctx.exit().await;
-			});
-		})
+		}))
 	}
 
 	#[derive(Clone, Debug, Coding, Hierarchy)]
@@ -460,30 +480,94 @@ mod test {
 
 	#[test]
 	fn request() {
-		spin_on(async {
+		let logger = TestLogger::new(async |s| eprint!("{s}"));
+		spin_on(with_logger(logger, async {
 			let (in1, out2) = pipe(1024);
 			let (in2, out1) = pipe(1024);
-			let (_, srv_ctx, run_srv) = io_comm(
-				Rc::new(Mutex::new(Box::pin(in2))),
-				Mutex::new(Box::pin(out2)),
-				async |_| panic!("No notifs expected"),
-				async |mut req| {
-					let val = req.read_req::<DummyRequest>().await?;
-					req.reply(&val, &(val.0 + 1)).await
+			let (_, srv_ctx, srv) =
+				io_comm(Rc::new(Mutex::new(Box::pin(in2))), Mutex::new(Box::pin(out2)));
+			let (client, client_ctx, client_srv) =
+				io_comm(Rc::new(Mutex::new(Box::pin(in1))), Mutex::new(Box::pin(out1)));
+			join!(
+				async {
+					srv
+						.listen(
+							async |_| panic!("No notifs expected"),
+							async |mut req| {
+								let val = req.read_req::<DummyRequest>().await?;
+								req.reply(&val, &(val.0 + 1)).await
+							},
+						)
+						.await
+						.unwrap()
 				},
+				async {
+					client_srv
+						.listen(
+							async |_| panic!("Not expecting ingress notif"),
+							async |_| panic!("Not expecting ingress req"),
+						)
+						.await
+						.unwrap()
+				},
+				async {
+					let response = client.request(DummyRequest(5)).await.unwrap();
+					assert_eq!(response, 6);
+					srv_ctx.exit().await;
+					client_ctx.exit().await;
+				}
 			);
-			let (client, client_ctx, run_client) = io_comm(
-				Rc::new(Mutex::new(Box::pin(in1))),
-				Mutex::new(Box::pin(out1)),
-				async |_| panic!("Not expecting ingress notif"),
-				async |_| panic!("Not expecting ingress req"),
-			);
-			join!(async { run_srv.await.unwrap() }, async { run_client.await.unwrap() }, async {
-				let response = client.request(DummyRequest(5)).await.unwrap();
-				assert_eq!(response, 6);
-				srv_ctx.exit().await;
-				client_ctx.exit().await;
-			});
-		})
+		}))
+	}
+
+	#[test]
+	fn exit() {
+		let logger = TestLogger::new(async |s| eprint!("{s}"));
+		spin_on(with_logger(logger, async {
+			let (input1, output1) = pipe(1024);
+			let (input2, output2) = pipe(1024);
+			let (reply_client, reply_context, reply_server) =
+				io_comm(Rc::new(Mutex::new(Box::pin(input1))), Mutex::new(Box::pin(output2)));
+			let (req_client, req_context, req_server) =
+				io_comm(Rc::new(Mutex::new(Box::pin(input2))), Mutex::new(Box::pin(output1)));
+			let reply_context = RefCell::new(Some(reply_context));
+			let (exit, onexit) = futures::channel::oneshot::channel::<()>();
+			join!(
+				async move {
+					reply_server
+						.listen(
+							async |hand| {
+								let _notif = hand.read::<TestNotif>().await.unwrap();
+								let context = reply_context.borrow_mut().take().unwrap();
+								context.exit().await;
+								Ok(())
+							},
+							async |mut hand| {
+								let req = hand.read_req::<DummyRequest>().await?;
+								hand.reply(&req, &(req.0 + 1)).await
+							},
+						)
+						.await
+						.unwrap();
+					exit.send(()).unwrap();
+					let _client = reply_client;
+				},
+				async move {
+					req_server
+						.listen(
+							async |_| panic!("Only the other server expected notifs"),
+							async |_| panic!("Only the other server expected requests"),
+						)
+						.await
+						.unwrap();
+					let _ctx = req_context;
+				},
+				async move {
+					req_client.request(DummyRequest(0)).await.unwrap();
+					req_client.notify(TestNotif(0)).await.unwrap();
+					onexit.await.unwrap();
+				}
+			)
+		}));
 	}
 }