Orchid/orchid
1
0
Fork 1
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

task_local context over context objects

Browse Source 
- interner impls logically separate from API in orchid-base (default host interner still in base for testing)
- error reporting, logging, and a variety of other features passed down via context in extension, not yet in host to maintain library-ish profile, should consider options
- no global spawn mechanic, the host has a spawn function but extensions only get a stash for enqueuing async work in sync callbacks which is then explicitly, manually, and with strict order popped and awaited
- still deadlocks nondeterministically for some ungodly reason
This commit is contained in:
Lawrence Bethlenfalvy
2026-01-01 14:54:29 +00:00
parent 06debb3636
commit 32d6237dc5
92 changed files with 2507 additions and 2223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

406
orchid-host/src/extension.rs
View File

@@ -2,7 +2,7 @@ use std::cell::RefCell;
use std::future::Future;
use std::io;
use std::num::NonZeroU64;
use std::pin::pin;
use std::pin::Pin;
use std::rc::{Rc, Weak};
use async_fn_stream::stream;
@@ -10,28 +10,33 @@ use derive_destructure::destructure;
use futures::channel::mpsc::{Sender, channel};
use futures::future::{join, join_all};
use futures::lock::Mutex;
use futures::{SinkExt, StreamExt, stream};
use hashbrown::HashMap;
use futures::{AsyncRead, AsyncWrite, AsyncWriteExt, SinkExt, StreamExt};
use hashbrown::{HashMap, HashSet};
use itertools::Itertools;
use orchid_api_traits::Request;
use orchid_base::builtin::ExtInit;
use orchid_api_traits::{Decode, Encode, Request};
use orchid_base::clone;
use orchid_base::format::{FmtCtxImpl, Format};
use orchid_base::interner::Tok;
use orchid_base::interner::{IStr, IStrv, es, ev, is, iv};
use orchid_base::location::Pos;
use orchid_base::logging::Logger;
use orchid_base::logging::logger;
use orchid_base::name::Sym;
use orchid_base::reqnot::{DynRequester, ReqNot, Requester as _};
use orchid_base::reqnot::{Client, ClientExt, MsgReaderExt, ReqHandleExt, ReqReaderExt, io_comm};
use orchid_base::stash::{stash, with_stash};
use orchid_base::tree::AtomRepr;
use crate::api;
use crate::atom::AtomHand;
use crate::ctx::Ctx;
use crate::ctx::{Ctx, JoinHandle};
use crate::dealias::{ChildError, ChildErrorKind, walk};
use crate::expr::{Expr, PathSetBuilder};
use crate::system::SystemCtor;
use crate::tree::MemberKind;
pub struct ExtPort {
pub input: Pin<Box<dyn AsyncWrite>>,
pub output: Pin<Box<dyn AsyncRead>>,
}
pub struct ReqPair<R: Request>(R, Sender<R::Response>);
/// Data held about an Extension. This is refcounted within [Extension]. It's
@@ -42,69 +47,47 @@ pub struct ReqPair<R: Request>(R, Sender<R::Response>);
pub struct ExtensionData {
name: String,
ctx: Ctx,
reqnot: ReqNot<api::HostMsgSet>,
join_ext: Option<Box<dyn JoinHandle>>,
client: Rc<dyn Client>,
systems: Vec<SystemCtor>,
logger: Logger,
next_pars: RefCell<NonZeroU64>,
exiting_snd: Sender<()>,
lex_recur: Mutex<HashMap<api::ParsId, Sender<ReqPair<api::SubLex>>>>,
strings: RefCell<HashSet<IStr>>,
string_vecs: RefCell<HashSet<IStrv>>,
}
impl Drop for ExtensionData {
fn drop(&mut self) {
let reqnot = self.reqnot.clone();
let mut exiting_snd = self.exiting_snd.clone();
(self.ctx.spawn)(Box::pin(async move {
reqnot.notify(api::HostExtNotif::Exit).await;
exiting_snd.send(()).await.unwrap()
}))
let client = self.client.clone();
let join_ext = self.join_ext.take().expect("Only called once in Drop");
stash(async move {
client.notify(api::HostExtNotif::Exit).await.unwrap();
join_ext.join().await;
})
}
}
#[derive(Clone)]
pub struct Extension(Rc<ExtensionData>);
impl Extension {
pub fn new(init: ExtInit, logger: Logger, msg_logger: Logger, ctx: Ctx) -> io::Result<Self> {
pub async fn new(mut init: ExtPort, ctx: Ctx) -> io::Result<Self> {
api::HostHeader { log_strategy: logger().strat(), msg_logs: ctx.msg_logs.strat() }
.encode(init.input.as_mut())
.await
.unwrap();
init.input.flush().await.unwrap();
let header = api::ExtensionHeader::decode(init.output.as_mut()).await.unwrap();
Ok(Self(Rc::new_cyclic(|weak: &Weak<ExtensionData>| {
let init = Rc::new(init);
let (exiting_snd, exiting_rcv) = channel::<()>(0);
(ctx.spawn)({
clone!(init, weak, ctx);
Box::pin(async move {
let rcv_stream = stream(async |mut cx| {
loop {
cx.emit(init.recv().await).await
}
});
let mut event_stream = pin!(stream::select(exiting_rcv.map(|()| None), rcv_stream));
while let Some(Some(msg)) = event_stream.next().await {
if let Some(reqnot) = weak.upgrade().map(|rc| rc.reqnot.clone()) {
let reqnot = reqnot.clone();
(ctx.spawn)(Box::pin(async move {
reqnot.receive(&msg).await;
}))
}
}
})
});
ExtensionData {
name: init.name.clone(),
exiting_snd,
ctx: ctx.clone(),
systems: (init.systems.iter().cloned())
.map(|decl| SystemCtor { decl, ext: WeakExtension(weak.clone()) })
.collect(),
logger: logger.clone(),
next_pars: RefCell::new(NonZeroU64::new(1).unwrap()),
lex_recur: Mutex::default(),
reqnot: ReqNot::new(
msg_logger,
move |sfn, _| clone!(init; Box::pin(async move { init.send(sfn).await })),
clone!(weak; move |notif, _| {
clone!(weak; Box::pin(async move {
// context not needed because exit is extension-initiated
let (client, _, future) = io_comm(
Rc::new(Mutex::new(init.input)),
Mutex::new(init.output),
clone!(weak; async move |reader| {
with_stash(async {
let this = Extension(weak.upgrade().unwrap());
let notif = reader.read::<api::ExtHostNotif>().await.unwrap();
if !matches!(notif, api::ExtHostNotif::Log(_)) {
writeln!(this.reqnot().logger(), "Host received notif {notif:?}");
writeln!(logger(), "Host received notif {notif:?}");
}
match notif {
api::ExtHostNotif::ExprNotif(api::ExprNotif::Acquire(acq)) => {
@@ -115,153 +98,191 @@ impl Extension {
if this.is_own_sys(rel.0).await {
this.0.ctx.exprs.take_expr(rel.1);
} else {
writeln!(this.reqnot().logger(), "Not our system {:?}", rel.0)
writeln!(this.0.ctx.msg_logs, "Not our system {:?}", rel.0)
}
},
api::ExtHostNotif::Log(api::Log(str)) => this.logger().log(str),
}
}))}),
{
clone!(weak, ctx);
move |hand, req| {
clone!(weak, ctx);
Box::pin(async move {
let this = Self(weak.upgrade().unwrap());
if !matches!(req, api::ExtHostReq::ExtAtomPrint(_)) {
writeln!(this.reqnot().logger(), "Host received request {req:?}");
api::ExtHostNotif::Log(api::Log(str)) => logger().log(str),
api::ExtHostNotif::Sweeped(data) => {
for i in join_all(data.strings.into_iter().map(es)).await {
this.0.strings.borrow_mut().remove(&i);
}
let i = this.ctx().i.clone();
match req {
api::ExtHostReq::Ping(ping) => hand.handle(&ping, &()).await,
api::ExtHostReq::IntReq(intreq) => match intreq {
api::IntReq::InternStr(s) => hand.handle(&s, &i.i(&*s.0).await.to_api()).await,
api::IntReq::InternStrv(v) => {
let tokens = join_all(v.0.iter().map(|m| i.ex(*m))).await;
hand.handle(&v, &i.i(&tokens).await.to_api()).await
},
api::IntReq::ExternStr(si) =>
hand.handle(&si, &Tok::<String>::from_api(si.0, &i).await.rc()).await,
api::IntReq::ExternStrv(vi) => {
let markerv = (i.ex(vi.0).await.iter()).map(|t| t.to_api()).collect_vec();
hand.handle(&vi, &markerv).await
},
for i in join_all(data.vecs.into_iter().map(ev)).await {
this.0.string_vecs.borrow_mut().remove(&i);
}
},
}
Ok(())
}).await
}),
{
clone!(weak, ctx);
async move |mut reader| {
with_stash(async {
let this = Self(weak.upgrade().unwrap());
let req = reader.read_req::<api::ExtHostReq>().await.unwrap();
let handle = reader.finish().await;
if !matches!(req, api::ExtHostReq::ExtAtomPrint(_)) {
writeln!(logger(), "Host received request {req:?}");
}
match req {
api::ExtHostReq::Ping(ping) => handle.reply(&ping, &()).await,
api::ExtHostReq::IntReq(intreq) => match intreq {
api::IntReq::InternStr(s) => {
let i = is(&s.0).await;
this.0.strings.borrow_mut().insert(i.clone());
handle.reply(&s, &i.to_api()).await
},
api::ExtHostReq::Fwd(ref fw @ api::Fwd(ref atom, ref key, ref body)) => {
let sys =
ctx.system_inst(atom.owner).await.expect("owner of live atom dropped");
let reply =
sys.reqnot().request(api::Fwded(fw.0.clone(), *key, body.clone())).await;
hand.handle(fw, &reply).await
api::IntReq::InternStrv(v) => {
let tokens = join_all(v.0.iter().map(|m| es(*m))).await;
this.0.strings.borrow_mut().extend(tokens.iter().cloned());
let i = iv(&tokens).await;
this.0.string_vecs.borrow_mut().insert(i.clone());
handle.reply(&v, &i.to_api()).await
},
api::ExtHostReq::SysFwd(ref fw @ api::SysFwd(id, ref body)) => {
let sys = ctx.system_inst(id).await.unwrap();
hand.handle(fw, &sys.request(body.clone()).await).await
api::IntReq::ExternStr(si) => {
let i = es(si.0).await;
this.0.strings.borrow_mut().insert(i.clone());
handle.reply(&si, &i.to_string()).await
},
api::ExtHostReq::SubLex(sl) => {
let (rep_in, mut rep_out) = channel(0);
{
let lex_g = this.0.lex_recur.lock().await;
let mut req_in =
lex_g.get(&sl.id).cloned().expect("Sublex for nonexistent lexid");
req_in.send(ReqPair(sl.clone(), rep_in)).await.unwrap();
api::IntReq::ExternStrv(vi) => {
let i = ev(vi.0).await;
this.0.strings.borrow_mut().extend(i.iter().cloned());
this.0.string_vecs.borrow_mut().insert(i.clone());
let markerv = i.iter().map(|t| t.to_api()).collect_vec();
handle.reply(&vi, &markerv).await
},
},
api::ExtHostReq::Fwd(ref fw @ api::Fwd(ref atom, ref key, ref body)) => {
let sys = ctx.system_inst(atom.owner).await.expect("owner of live atom dropped");
let client = sys.client();
let reply =
client.request(api::Fwded(fw.0.clone(), *key, body.clone())).await.unwrap();
handle.reply(fw, &reply).await
},
api::ExtHostReq::SysFwd(ref fw @ api::SysFwd(id, ref body)) => {
let sys = ctx.system_inst(id).await.unwrap();
handle.reply(fw, &sys.request(body.clone()).await).await
},
api::ExtHostReq::SubLex(sl) => {
let (rep_in, mut rep_out) = channel(0);
{
let lex_g = this.0.lex_recur.lock().await;
let mut req_in =
lex_g.get(&sl.id).cloned().expect("Sublex for nonexistent lexid");
req_in.send(ReqPair(sl.clone(), rep_in)).await.unwrap();
}
handle.reply(&sl, &rep_out.next().await.unwrap()).await
},
api::ExtHostReq::ExprReq(expr_req) => match expr_req {
api::ExprReq::Inspect(ins @ api::Inspect { target }) => {
let expr = ctx.exprs.get_expr(target).expect("Invalid ticket");
handle
.reply(&ins, &api::Inspected {
refcount: expr.strong_count() as u32,
location: expr.pos().to_api(),
kind: expr.to_api().await,
})
.await
},
api::ExprReq::Create(ref cre @ api::Create(ref expr)) => {
let expr = Expr::from_api(expr, PathSetBuilder::new(), ctx.clone()).await;
let expr_id = expr.id();
ctx.exprs.give_expr(expr);
handle.reply(cre, &expr_id).await
},
},
api::ExtHostReq::LsModule(ref ls @ api::LsModule(_sys, path)) => {
let reply: <api::LsModule as Request>::Response = 'reply: {
let path = ev(path).await;
let root = (ctx.root.read().await.upgrade())
.expect("LSModule called when root isn't in context");
let root_data = &*root.0.read().await;
let mut walk_ctx = (ctx.clone(), &root_data.consts);
let module =
match walk(&root_data.root, false, path.iter().cloned(), &mut walk_ctx).await
{
Ok(module) => module,
Err(ChildError { kind, .. }) =>
break 'reply Err(match kind {
ChildErrorKind::Private => panic!("Access checking was disabled"),
ChildErrorKind::Constant => api::LsModuleError::IsConstant,
ChildErrorKind::Missing => api::LsModuleError::InvalidPath,
}),
};
let mut members = std::collections::HashMap::new();
for (k, v) in &module.members {
let kind = match v.kind(ctx.clone(), &root_data.consts).await {
MemberKind::Const => api::MemberInfoKind::Constant,
MemberKind::Module(_) => api::MemberInfoKind::Module,
};
members.insert(k.to_api(), api::MemberInfo { public: v.public, kind });
}
hand.handle(&sl, &rep_out.next().await.unwrap()).await
},
api::ExtHostReq::ExprReq(expr_req) => match expr_req {
api::ExprReq::Inspect(ins @ api::Inspect { target }) => {
let expr = ctx.exprs.get_expr(target).expect("Invalid ticket");
hand
.handle(&ins, &api::Inspected {
refcount: expr.strong_count() as u32,
location: expr.pos().to_api(),
kind: expr.to_api().await,
})
.await
},
api::ExprReq::Create(ref cre @ api::Create(ref expr)) => {
let expr = Expr::from_api(expr, PathSetBuilder::new(), ctx.clone()).await;
let expr_id = expr.id();
ctx.exprs.give_expr(expr);
hand.handle(cre, &expr_id).await
},
},
api::ExtHostReq::LsModule(ref ls @ api::LsModule(_sys, path)) => {
let reply: <api::LsModule as Request>::Response = 'reply: {
let path = i.ex(path).await;
let root = (ctx.root.read().await.upgrade())
.expect("LSModule called when root isn't in context");
let root_data = &*root.0.read().await;
let mut walk_ctx = (ctx.clone(), &root_data.consts);
let module =
match walk(&root_data.root, false, path.iter().cloned(), &mut walk_ctx)
.await
{
Ok(module) => module,
Err(ChildError { kind, .. }) =>
break 'reply Err(match kind {
ChildErrorKind::Private => panic!("Access checking was disabled"),
ChildErrorKind::Constant => api::LsModuleError::IsConstant,
ChildErrorKind::Missing => api::LsModuleError::InvalidPath,
}),
};
let mut members = std::collections::HashMap::new();
for (k, v) in &module.members {
let kind = match v.kind(ctx.clone(), &root_data.consts).await {
MemberKind::Const => api::MemberInfoKind::Constant,
MemberKind::Module(_) => api::MemberInfoKind::Module,
};
members.insert(k.to_api(), api::MemberInfo { public: v.public, kind });
}
Ok(api::ModuleInfo { members })
};
hand.handle(ls, &reply).await
},
api::ExtHostReq::ResolveNames(ref rn) => {
let api::ResolveNames { constid, names, sys } = rn;
let mut resolver = {
let systems = ctx.systems.read().await;
let weak_sys = systems.get(sys).expect("ResolveNames for invalid sys");
let sys = weak_sys.upgrade().expect("ResolveNames after sys drop");
sys.name_resolver(*constid).await
};
let responses = stream(async |mut cx| {
for name in names {
cx.emit(match resolver(&ctx.i.ex(*name).await[..]).await {
Ok(abs) => Ok(abs.to_sym(&ctx.i).await.to_api()),
Err(e) => Err(e.to_api()),
})
.await
}
})
.collect()
.await;
hand.handle(rn, &responses).await
},
api::ExtHostReq::ExtAtomPrint(ref eap @ api::ExtAtomPrint(ref atom)) => {
let atom = AtomHand::from_api(atom, Pos::None, &mut ctx.clone()).await;
let unit = atom.print(&FmtCtxImpl { i: &this.ctx().i }).await;
hand.handle(eap, &unit.to_api()).await
},
}
})
}
},
),
Ok(api::ModuleInfo { members })
};
handle.reply(ls, &reply).await
},
api::ExtHostReq::ResolveNames(ref rn) => {
let api::ResolveNames { constid, names, sys } = rn;
let mut resolver = {
let systems = ctx.systems.read().await;
let weak_sys = systems.get(sys).expect("ResolveNames for invalid sys");
let sys = weak_sys.upgrade().expect("ResolveNames after sys drop");
sys.name_resolver(*constid).await
};
let responses = stream(async |mut cx| {
for name in names {
cx.emit(match resolver(&ev(*name).await[..]).await {
Ok(abs) => Ok(abs.to_sym().await.to_api()),
Err(e) => Err(e.to_api()),
})
.await
}
})
.collect()
.await;
handle.reply(rn, &responses).await
},
api::ExtHostReq::ExtAtomPrint(ref eap @ api::ExtAtomPrint(ref atom)) => {
let atom = AtomHand::from_api(atom, Pos::None, &mut ctx.clone()).await;
let unit = atom.print(&FmtCtxImpl::default()).await;
handle.reply(eap, &unit.to_api()).await
},
}
})
.await
}
},
);
let join_ext = ctx.spawn(async {
future.await.unwrap();
// extension exited successfully
});
ExtensionData {
name: header.name.clone(),
ctx: ctx.clone(),
systems: (header.systems.iter().cloned())
.map(|decl| SystemCtor { decl, ext: WeakExtension(weak.clone()) })
.collect(),
join_ext: Some(join_ext),
next_pars: RefCell::new(NonZeroU64::new(1).unwrap()),
lex_recur: Mutex::default(),
client: Rc::new(client),
strings: RefCell::default(),
string_vecs: RefCell::default(),
}
})))
}
pub fn name(&self) -> &String { &self.0.name }
#[must_use]
pub fn reqnot(&self) -> &ReqNot<api::HostMsgSet> { &self.0.reqnot }
pub fn client(&self) -> &dyn Client { &*self.0.client }
#[must_use]
pub fn ctx(&self) -> &Ctx { &self.0.ctx }
#[must_use]
pub fn logger(&self) -> &Logger { &self.0.logger }
pub fn system_ctors(&self) -> impl Iterator<Item = &SystemCtor> { self.0.systems.iter() }
#[must_use]
pub async fn is_own_sys(&self, id: api::SysId) -> bool {
let Some(sys) = self.ctx().system_inst(id).await else {
writeln!(self.logger(), "Invalid system ID {id:?}");
writeln!(logger(), "Invalid system ID {id:?}");
return false;
};
Rc::ptr_eq(&self.0, &sys.ext().0)
@@ -274,7 +295,7 @@ impl Extension {
}
pub(crate) async fn lex_req<F: Future<Output = Option<api::SubLexed>>>(
&self,
source: Tok<String>,
source: IStr,
src: Sym,
pos: u32,
sys: api::SysId,
@@ -287,9 +308,10 @@ impl Extension {
self.0.lex_recur.lock().await.insert(id, req_in); // lex_recur released
let (ret, ()) = join(
async {
let res = (self.reqnot())
let res = (self.client())
.request(api::LexExpr { id, pos, sys, src: src.to_api(), text: source.to_api() })
.await;
.await
.unwrap();
// collect sender to unblock recursion handler branch before returning
self.0.lex_recur.lock().await.remove(&id);
res
@@ -306,10 +328,10 @@ impl Extension {
}
pub fn system_drop(&self, id: api::SysId) {
let rc = self.clone();
(self.ctx().spawn)(Box::pin(async move {
rc.reqnot().request(api::SystemDrop(id)).await;
let _ = self.ctx().spawn(with_stash(async move {
rc.client().request(api::SystemDrop(id)).await.unwrap();
rc.ctx().systems.write().await.remove(&id);
}))
}));
}
#[must_use]
pub fn downgrade(&self) -> WeakExtension { WeakExtension(Rc::downgrade(&self.0)) }