Skip to content

Commit df93b6f

Browse files
danlapidclaude
andcommitted
kj-rs-tokio: a kj::EventPort driven by a per-thread tokio runtime
Second piece of the Rust I/O backend split (on top of the kj-rs bridge rework): TokioEventPort implements kj::EventPort over a per-thread tokio current_thread runtime, so a KJ event loop sleeps by parking inside tokio's scheduler -- tokio tasks (and, in the next piece, tokio I/O) run whenever KJ would otherwise block. - wait()/poll() park in LocalSet::block_on on the runtime, woken by cross-thread wake(), by KJ itself when a tokio task queues KJ work, or by the next KJ timer deadline (tokio's timer wheel; ~1 ms, the same granularity as KJ's epoll-based port). - A tokio task running inside the park may use KJ freely -- fulfill a PromiseFulfiller, arm a timer, add to a TaskSet, wake a bridged future -- and the port learns of it through KJ's own hooks rather than a convention: kj::EventLoop reports setRunnable(false) right before it calls wait() (capnproto/capnproto#2814), so the first arm during the park is a setRunnable(true) edge the port turns into a notify; and the port installs itself as the kj::TimerImpl's SleepHooks for the duration of the park, so a sooner timer armed mid-park ends the park and kj::Timer::now() reads the live clock. The one thing a task must not do is re-enter promise.wait() / waitScope.poll() on the loop thread (tokio rejects the nested block_on; the panic surfaces as a kj::Exception). - The port owns its kj::EventLoop, declared last so it is destroyed first, and cancels the LocalSet's spawned tasks in its own destructor before any member goes. Teardown order is member order rather than something a context has to remember: spawned tasks routinely own KJ promises (a bridged PromiseFuture holds an OwnPromiseNode and an armed awaiter Event; a KJ timer promise holds a TimerImpl entry), and dropping them after the loop died was a use-after-free (reproduced: ~TimerPromiseAdapter dereferencing the destroyed TimerImpl). A bare port with no TokioAsyncIoContext is safe too. One port per thread is KJ_REQUIREd on the C++ side and asserted on the Rust side. - kj_rs_tokio::spawn() enqueues !Send futures onto the loop's LocalSet, pinned to the loop thread per the bridge's single-thread waker axiom. - setupTokioAsyncIo() yields the port (loop) + wait scope only; the I/O providers over this port arrive in the kj-rs-io piece. - The capnp-cpp dependency is frozen at the head of capnproto/capnproto#2814 (lock-free Executor::isCurrent(); setRunnable(false) before EventPort::wait()), which this port relies on. Unfreeze once it merges into v2. This adds tokio's `sync` feature (Notify) to the workspace; no new crates. - Tests: a spawned task fulfilling a kj::PromiseFulfiller during a timed park and during a wait-forever park; a KJ timer armed by a task during both kinds of park, honored at its own (live-clock) deadline; a cross-thread bridged wake arriving while the loop is busy-polling rather than parked; spawned tasks holding a KJ timer / a bridged KJ promise across context teardown and across bare-port teardown (ASAN); one port per thread; a bridged future woken from a plain std::thread while the loop is parked; poll() budget under a yield-forever task; nested promise.wait() from a task surfacing as a kj::Exception; timer cancellation; already-due timers; a panicking spawned task surfacing as a kj::Exception and a detached task still completing; and two tokio-ported loops executeAsync-ing into each other concurrently (TSAN). port.rs units: wake-latch semantics, a concurrent wake() storm from multiple threads (TSAN), cross-thread Handle::spawn, poll() advancing a ready LocalSet task, !Send spawned futures, and the bare-TokioPort Drop fallback. Clean under --config=asan and --config=tsan-macos. - Object-relationship overview in tokio-event-port.h. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 4cca2aa commit df93b6f

14 files changed

Lines changed: 2128 additions & 3 deletions

File tree

build/deps/deps.jsonc

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,12 +31,17 @@
3131
"name": "tcmalloc",
3232
"type": "bazel_dep"
3333
},
34+
// Frozen at the head of capnproto/capnproto#2814 (kj/async: lock-free Executor::isCurrent();
35+
// setRunnable(false) before EventPort::wait()), which kj-rs / kj-rs-tokio rely on. Unfreeze
36+
// once it has merged into v2.
3437
{
3538
"name": "capnp-cpp",
3639
"type": "github_tarball",
3740
"owner": "capnproto",
3841
"repo": "capnproto",
3942
"branch": "v2",
43+
"freeze_commit": "49cbe3daf42d97166315acdddd9666763a1a3da3",
44+
"freeze_sha256": "6b6a9a13a47a14142caef9c07c74bd41f9c90d30e81c79c34772675219afbb9f",
4045
"extra_strip_prefix": "/c++"
4146
},
4247
// We want to avoid version skew with v8, so we use identical versions. Keep this

build/deps/gen/deps.MODULE.bazel

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,10 +27,10 @@ bazel_dep(name = "brotli", version = "1.2.0.bcr.3")
2727
# capnp-cpp
2828
http.archive(
2929
name = "capnp-cpp",
30-
sha256 = "73a3f249ec2a835abaeb8f3caa5bddde044312f2c65b6353a788028b5b7e59f2",
31-
strip_prefix = "capnproto-capnproto-7077d44/c++",
30+
sha256 = "6b6a9a13a47a14142caef9c07c74bd41f9c90d30e81c79c34772675219afbb9f",
31+
strip_prefix = "capnproto-capnproto-49cbe3d/c++",
3232
type = "tgz",
33-
url = "https://github.com/capnproto/capnproto/tarball/7077d443af81e94ea923e41b9990f697058e3b56",
33+
url = "https://github.com/capnproto/capnproto/tarball/49cbe3daf42d97166315acdddd9666763a1a3da3",
3434
)
3535
use_repo(http, "capnp-cpp")
3636

src/rust/cxx/AGENTS.md

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,9 @@ Bazel module, Cargo workspace, toolchain configuration, or external `workerd-cxx
3333
- `src/` and `include/` — cxx Rust and C++ runtimes
3434
- `syntax/`, `gen/`, and `macro/` — bridge parser and code generators
3535
- `kj-rs/` — KJ promises/futures, exceptions, ownership, refcounting, dates, and `Maybe`
36+
- `kj-rs-tokio/``TokioEventPort`: a `kj::EventPort` backed by a per-thread tokio
37+
`current_thread` runtime, plus `setupTokioAsyncIo()` (no I/O providers) and
38+
`kj_rs_tokio::spawn()`
3639
- `tests/` and `kj-rs/tests/` — Rust and C++ bridge integration tests
3740
- `tools/bazel/` — Bazel bridge-generation macro used by this component's tests
3841

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
load("@rules_rust//rust:defs.bzl", "rust_library", "rust_test")
2+
load("//:build/wd_cc_library.bzl", "wd_cc_library")
3+
load("//src/rust/cxx/tools/bazel:rust_cxx_bridge.bzl", "rust_cxx_bridge")
4+
5+
wd_cc_library(
6+
name = "kj-rs-tokio-lib",
7+
srcs = glob(["*.c++"]),
8+
hdrs = glob(["*.h"]),
9+
include_prefix = "kj-rs-tokio",
10+
linkstatic = select({
11+
"@platforms//os:windows": True,
12+
"//conditions:default": False,
13+
}),
14+
strip_include_prefix = "/src/rust/cxx/kj-rs-tokio",
15+
visibility = ["//visibility:public"],
16+
deps = [
17+
":bridge",
18+
],
19+
)
20+
21+
rust_library(
22+
name = "kj-rs-tokio",
23+
srcs = glob(["*.rs"]),
24+
compile_data = glob(["*.h"]),
25+
edition = "2024",
26+
link_deps = [
27+
":bridge",
28+
":kj-rs-tokio-lib",
29+
],
30+
target_compatible_with = select({
31+
"@//build/config:no_build": ["@platforms//:incompatible"],
32+
"//conditions:default": [],
33+
}),
34+
visibility = ["//visibility:public"],
35+
deps = [
36+
"//src/rust/cxx",
37+
"@crates_vendor//:tokio",
38+
],
39+
)
40+
41+
rust_test(
42+
name = "kj-rs-tokio_test",
43+
crate = "kj-rs-tokio",
44+
edition = "2024",
45+
target_compatible_with = select({
46+
"@//build/config:no_build": ["@platforms//:incompatible"],
47+
"//conditions:default": [],
48+
}),
49+
)
50+
51+
rust_cxx_bridge(
52+
name = "bridge",
53+
src = "ffi.rs",
54+
hdrs = glob(["*.h"]),
55+
include_prefix = "kj-rs-tokio",
56+
visibility = ["//visibility:public"],
57+
deps = [
58+
"@capnp-cpp//src/kj:kj",
59+
# kj-rs-tokio only needs the abstract async core (Promise / EventLoop / EventPort /
60+
# Timer) to build the tokio-backed EventPort; it uses neither the abstract kj streams
61+
# (:kj-async-io) nor the OS event loop (setupAsyncIo / UnixEventPort). Depending on
62+
# :kj-async-core (not the :kj-async umbrella) keeps the tokio event loop off the
63+
# concrete kj OS I/O layer (:kj-async-os), which the workerd rust-io hermeticity aspect
64+
# forbids.
65+
"@capnp-cpp//src/kj:kj-async-core",
66+
"//src/rust/cxx:core",
67+
],
68+
)

src/rust/cxx/kj-rs-tokio/ffi.rs

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
//! The `#[cxx::bridge]` FFI island for kj-rs-tokio.
2+
//!
3+
//! This is the crate's single dedicated FFI-island file (file-top `#![allow(unsafe_code)]`): the
4+
//! `#[cxx::bridge] mod bridge` is the C++ <-> Rust wire the C++ `kj_rs_tokio::TokioEventPort`
5+
//! drives (see `tokio-event-port.h`). Everything else -- `lib.rs` and the entire event-port
6+
//! business logic in `port.rs` -- is wholly-safe, compiler-proven unsafe-free under the crate-root
7+
//! `#![deny(unsafe_code)]`.
8+
#![allow(unsafe_code)]
9+
10+
use crate::port::TokioPort;
11+
use crate::port::new_tokio_port;
12+
13+
#[cxx::bridge(namespace = "kj_rs_tokio")]
14+
// FFI island: the cxx bridge macro generates the `unsafe` extern shims.
15+
// unnecessary_box_returns: returning the opaque `TokioPort` to C++ boxed is the cxx idiom. The
16+
// lint's firing is platform-dependent (it has a size threshold and `TokioPort`'s size differs by
17+
// target), so `#[expect]` would be unfulfilled on some targets.
18+
#[expect(clippy::allow_attributes)]
19+
#[allow(clippy::unnecessary_box_returns)]
20+
mod bridge {
21+
// None of these are `Result`: they cannot fail in a way C++ could handle. The panics that CAN
22+
// occur -- a second port on one thread (`TokioPort::new`), a nested `block_on` from a task
23+
// that re-entered `promise.wait()` (`wait_*`/`poll`), the LocalSet slot being gone -- are
24+
// caller-contract violations, and the in-tree cxx fork converts every panic escaping an
25+
// `extern "Rust"` fn into a `kj::Exception` thrown at the C++ call site.
26+
extern "Rust" {
27+
type TokioPort;
28+
29+
fn new_tokio_port() -> Box<TokioPort>;
30+
31+
/// Cancel every task spawned onto this thread's `LocalSet` (dropping their state now, on
32+
/// this thread). `TokioEventPort`'s destructor calls this before destroying the KJ event
33+
/// loop and timer it owns, because spawned tasks may own KJ promises. Idempotent.
34+
fn cancel_spawned_tasks(&self);
35+
36+
/// Block until `wake()` or `notify_kj_service()` is called, running tokio tasks in the
37+
/// meantime. Returns the wake latch (see `TokioPort::take_wake_latch`).
38+
fn wait_forever(&self) -> bool;
39+
40+
/// Like `wait_forever`, but additionally returns after `timeout_ns` nanoseconds. The
41+
/// C++ side computes the timeout from `kj::TimerImpl::timeoutToNextEvent()`.
42+
fn wait_timeout_ns(&self, timeout_ns: u64) -> bool;
43+
44+
/// Non-blocking: let the tokio scheduler run already-ready tasks for a bounded number of
45+
/// turns. Never sleeps. Returns the wake latch.
46+
fn poll(&self) -> bool;
47+
48+
/// Set the wake latch and unblock a concurrent `wait_*`. Callable from any thread.
49+
fn wake(&self);
50+
51+
/// Loop thread only: KJ has told the port it needs the thread back -- through
52+
/// `EventPort::setRunnable(true)` (an event was armed) or through the port's
53+
/// `TimerImpl::SleepHooks` (a sooner timer was armed while sleeping). Unblocks a
54+
/// concurrent `wait_*` without setting the wake latch; a no-op outside `wait_*`.
55+
fn notify_kj_service(&self);
56+
}
57+
}

src/rust/cxx/kj-rs-tokio/lib.rs

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
//! Rust half of the tokio-backed KJ event loop foundation.
2+
//!
3+
//! This crate owns a per-thread tokio `current_thread` runtime and exposes the primitives the
4+
//! C++ `kj_rs_tokio::TokioEventPort` (see `tokio-event-port.h` for the full contract) needs to
5+
//! implement `kj::EventPort`: parking the thread in `Runtime::block_on` (which drives the
6+
//! whole tokio scheduler while C++ is "blocked"), a bounded non-blocking `poll`, and the
7+
//! cross-thread `wake()` latch that `kj::Executor` and
8+
//! `kj::newPromiseAndCrossThreadFulfiller` depend on, and the `notify_kj_service` entry point KJ
9+
//! reaches (via `setRunnable(true)` / the port's `TimerImpl::SleepHooks`) to hand the thread
10+
//! back whenever a tokio task has queued KJ work.
11+
12+
// Safety & panic enforcement walls. Test code exempted.
13+
//
14+
// `unsafe` is quarantined into a single named FFI island: the crate root denies `unsafe_code`, so
15+
// the entire event-port business logic (`TokioPort`, the `wait`/`poll`/`wake` machinery) is
16+
// *compiler-proven* to contain no hand-written unsafe. The one island that opts back in
17+
// via `#![allow(unsafe_code)]` is `ffi.rs`: the `#[cxx::bridge]` wire.
18+
#![deny(unsafe_op_in_unsafe_fn)]
19+
#![deny(unsafe_code)]
20+
#![deny(clippy::undocumented_unsafe_blocks)]
21+
#![deny(
22+
clippy::unwrap_used,
23+
clippy::expect_used,
24+
clippy::panic,
25+
clippy::unreachable,
26+
clippy::todo,
27+
clippy::unimplemented
28+
)]
29+
#![cfg_attr(
30+
test,
31+
allow(
32+
clippy::unwrap_used,
33+
clippy::expect_used,
34+
clippy::panic,
35+
clippy::unreachable,
36+
clippy::todo,
37+
clippy::unimplemented
38+
)
39+
)]
40+
41+
pub use port::TokioPort;
42+
pub use port::current_handle;
43+
pub use port::spawn;
44+
45+
mod ffi;
46+
mod port;

0 commit comments

Comments
 (0)