From 6b88d50456386f9757a99782d2282226f30418be Mon Sep 17 00:00:00 2001 From: eric8810 Date: Wed, 21 Oct 2026 11:24:46 +0800 Subject: [PATCH] =?UTF-8?q?fix(bridge):=20=E4=BA=8B=E4=BB=B6=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E9=92=A9=E5=AD=90=E5=9C=A8=E8=AF=BB=E5=88=B0=20events?= =?UTF-8?q?.dispatch=20=E6=97=B6=E8=AE=A1=E6=95=B0=EF=BC=8CviaHost=20?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E4=B8=8E=E4=BC=9A=E8=AF=9D=E7=9A=84=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E4=BB=BB=E5=8A=A1=E8=B5=9B=E8=B7=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 宿主读到 events.* 帧后放进会话的有序任务里执行，读线程接着读下一帧。 测试钩子原先在有序任务执行 events.dispatch 时才计数，于是运行时在 dispatch 之后发出的回复（发出方 emit 的返回）可能先被读到，套件读计数时 这次 dispatch 还没算上，emit/order-across-fast-path 偶发报 viaHost: expected true, 0 events.dispatch arrived。 现在读线程接纳事件操作时调用 Dispatch::arrived（经 peer 的 Operations、 Handler 与 Routed 转发到 Imports），钩子在这里计数；hold 仍在执行时等待。 --- crates/rutis-bridge/src/events/hub.rs | 20 +++++++++++++++- crates/rutis-bridge/src/events/testing.rs | 16 +++++++++---- crates/rutis-bridge/src/peer.rs | 11 +++++++++ crates/rutis-bridge/src/runtime/access.rs | 4 ++++ crates/rutis-bridge/src/runtime/process.rs | 6 +++++ crates/rutis-bridge/src/session/rpc.rs | 27 ++++++++++++++++++---- 6 files changed, 74 insertions(+), 10 deletions(-) diff --git a/crates/rutis-bridge/src/events/hub.rs b/crates/rutis-bridge/src/events/hub.rs index 0a739fbd1..1a8aa33c2 100644 --- a/crates/rutis-bridge/src/events/hub.rs +++ b/crates/rutis-bridge/src/events/hub.rs @@ -431,6 +431,24 @@ impl SessionEvents { } } + /// The `events.*` operation `method` was read, on the session's reader + /// (before its next frame; [`SessionEvents::invoke`] runs it later, in + /// its turn): the tap counts the dispatches here. + pub(crate) fn arrived(&self, method: &str) { + #[cfg(feature = "testing")] + if method == "events.dispatch" { + if let Some(tap) = self + .hub + .as_ref() + .and_then(|hub| hub.tap.lock().unwrap().clone()) + { + tap.received(); + } + } + #[cfg(not(feature = "testing"))] + let _ = method; + } + /// The `events.*` operations a runtime sends. pub(crate) fn invoke( self: &Arc, @@ -449,7 +467,7 @@ impl SessionEvents { .as_ref() .and_then(|hub| hub.tap.lock().unwrap().clone()) { - tap.arrived(); + tap.starting(); } self.dispatch(connection, first) } diff --git a/crates/rutis-bridge/src/events/testing.rs b/crates/rutis-bridge/src/events/testing.rs index cb31c853c..af47556db 100644 --- a/crates/rutis-bridge/src/events/testing.rs +++ b/crates/rutis-bridge/src/events/testing.rs @@ -69,7 +69,9 @@ impl Tap { Ok(()) } - /// The `events.dispatch` requests received so far. + /// The `events.dispatch` requests received so far: each is counted as + /// its session reads it, so one a runtime sent before answering a call + /// is counted once that answer is. pub fn dispatches(&self) -> u64 { self.0.dispatches.load(Ordering::SeqCst) } @@ -88,10 +90,16 @@ impl Tap { } impl TapState { - /// An `events.dispatch` arrived: counted, and held if a hold waits for - /// it (the thread blocks until the hold is complete). - pub(crate) fn arrived(&self) { + /// An `events.dispatch` was read: counted on the session's reader, + /// before its next frame, so a reply the runtime sent after it (the + /// emitter's) finds it counted. + pub(crate) fn received(&self) { self.dispatches.fetch_add(1, Ordering::SeqCst); + } + + /// An `events.dispatch` is about to run: held if a hold waits for it + /// (the thread blocks until the hold is complete). + pub(crate) fn starting(&self) { let mut hold = self.hold.lock().unwrap(); if hold.want == 0 { return; diff --git a/crates/rutis-bridge/src/peer.rs b/crates/rutis-bridge/src/peer.rs index 010ae32ea..48b76b604 100644 --- a/crates/rutis-bridge/src/peer.rs +++ b/crates/rutis-bridge/src/peer.rs @@ -26,6 +26,9 @@ use tokio::sync::watch; /// Serves one family of operations on a peer's session. pub trait Handler: Send + Sync + 'static { fn invoke(&self, peer: &Connection, target: &str, method: &str, args: Value) -> Reply; + + /// [`Dispatch::arrived`] for this family's event operations. + fn arrived(&self, _peer: &Connection, _method: &str) {} } impl Handler for F @@ -148,6 +151,14 @@ impl Operations { } impl Dispatch for Operations { + fn arrived(&self, peer: &Connection, method: &str) { + let handler = (self.handlers.lock().unwrap().get(family("", method))) + .map(|family| family.handler.clone()); + if let Some(handler) = handler { + handler.arrived(peer, method); + } + } + fn invoke(&self, peer: &Connection, target: &str, method: &str, args: Value) -> Reply { if target.is_empty() && method == "link.offers" { let announced = args diff --git a/crates/rutis-bridge/src/runtime/access.rs b/crates/rutis-bridge/src/runtime/access.rs index 269207b71..d3fd06b62 100644 --- a/crates/rutis-bridge/src/runtime/access.rs +++ b/crates/rutis-bridge/src/runtime/access.rs @@ -73,6 +73,10 @@ impl crate::Handler for Routed { fn invoke(&self, peer: &Connection, target: &str, method: &str, args: Value) -> Reply { self.0.invoke(peer, target, method, args) } + + fn arrived(&self, peer: &Connection, method: &str) { + self.0.arrived(peer, method) + } } impl RuntimeSession for PeerSession { diff --git a/crates/rutis-bridge/src/runtime/process.rs b/crates/rutis-bridge/src/runtime/process.rs index f1b625405..bb4eebe05 100644 --- a/crates/rutis-bridge/src/runtime/process.rs +++ b/crates/rutis-bridge/src/runtime/process.rs @@ -450,6 +450,12 @@ impl Imports { } } impl Dispatch for Imports { + fn arrived(&self, _connection: &Connection, method: &str) { + if let Some(events) = self.bus.get() { + events.arrived(method); + } + } + fn invoke(&self, connection: &Connection, target: &str, method: &str, args: RpcValue) -> Reply { if let Some(name) = target.strip_prefix("host:") { let (host, row) = self diff --git a/crates/rutis-bridge/src/session/rpc.rs b/crates/rutis-bridge/src/session/rpc.rs index 7cb82a37a..a8e902442 100644 --- a/crates/rutis-bridge/src/session/rpc.rs +++ b/crates/rutis-bridge/src/session/rpc.rs @@ -891,6 +891,14 @@ impl Drop for Flight { pub trait Dispatch: Send + Sync + 'static { fn invoke(&self, peer: &Connection, target: &str, method: &str, args: Value) -> Reply; + + /// The event operation `method` (an `events.*` control call) was read + /// and admitted: called on the reader, before the next frame is taken, + /// whereas [`Dispatch::invoke`] runs it later, in the session's order + /// of event operations. It must not block. The host's bus counts the + /// `events.dispatch` requests here for its test hooks (feature + /// `testing`); nothing else needs it. + fn arrived(&self, _peer: &Connection, _method: &str) {} } /// How a session identifies itself and its calls. #[derive(Clone, Debug, Default)] @@ -2292,10 +2300,14 @@ impl Connection { // table changes and emits each in its turn, the rest started in it. // Never here: a reader runs no plugin code (a listener, an // observer), which could wait for a frame only this reader reads. - let event_op = matches!( - &incoming.call, - Accepted::Invoke(target, method, _) if target.is_empty() && method.starts_with("events.") - ); + let event_op = match &incoming.call { + Accepted::Invoke(target, method, _) + if target.is_empty() && method.starts_with("events.") => + { + Some(method.clone()) + } + _ => None, + }; let inline = matches!( &incoming.call, Accepted::Invoke(target, method, args) if target.is_empty() && inline_event_op(method, args) @@ -2340,7 +2352,12 @@ impl Connection { Err(id) } }; - if let (Err(id), true) = (&delivery, event_op) { + if let Some(method) = &event_op { + // Arrived now, in frame order: before any later frame of the + // session (a reply among them) is read. + self.0.dispatch.arrived(self, method); + } + if let (Err(id), true) = (&delivery, event_op.is_some()) { self.run_ordered(id.clone(), inline); } else if let Err(id) = delivery { let peer = self.clone();