itsscb/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2025-09-25 12:00:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

wasm: support rt-multi-thread with wasm32-wasi-preview1-threads (#6510)

Browse Source
This commit is contained in:
LongYinan 2024-04-24 15:45:54 +08:00 committed by GitHub
parent a73d6bf33a
commit 9ed595767d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 127 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
.github/workflows/ci.yml vendored
View File

@ -909,16 +909,21 @@ jobs:
working-directory: tokio
wasm32-wasi:
name: wasm32-wasi
name: ${{ matrix.target }}
needs: basics
runs-on: ubuntu-latest
strategy:
matrix:
target:
- wasm32-wasi
- wasm32-wasi-preview1-threads
steps:
- uses: actions/checkout@v4
- name: Install Rust ${{ env.rust_stable }}
uses: dtolnay/rust-toolchain@stable
with:
toolchain: ${{ env.rust_stable }}
targets: wasm32-wasi
targets: ${{ matrix.target }}
# Install dependencies
- name: Install cargo-hack, wasmtime, and cargo-wasi
@ -928,28 +933,40 @@ jobs:
- uses: Swatinem/rust-cache@v2
- name: WASI test tokio full
run: cargo test -p tokio --target wasm32-wasi --features full
run: cargo test -p tokio --target ${{ matrix.target }} --features full
env:
CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
RUSTFLAGS: --cfg tokio_unstable -Dwarnings
CARGO_TARGET_WASM32_WASI_PREVIEW1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
- name: WASI test tokio-util full
run: cargo test -p tokio-util --target wasm32-wasi --features full
run: cargo test -p tokio-util --target ${{ matrix.target }} --features full
env:
CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
RUSTFLAGS: --cfg tokio_unstable -Dwarnings
CARGO_TARGET_WASM32_WASI_PREVIEW1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
- name: WASI test tokio-stream
run: cargo test -p tokio-stream --target wasm32-wasi --features time,net,io-util,sync
run: cargo test -p tokio-stream --target ${{ matrix.target }} --features time,net,io-util,sync
env:
CARGO_TARGET_WASM32_WASI_RUNNER: "wasmtime run --"
RUSTFLAGS: --cfg tokio_unstable -Dwarnings
CARGO_TARGET_WASM32_WASI_PREVIEW1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
- name: test tests-integration --features wasi-rt
# TODO: this should become: `cargo hack wasi test --each-feature`
run: cargo wasi test --test rt_yield --features wasi-rt
if: matrix.target == 'wasm32-wasi'
working-directory: tests-integration
- name: test tests-integration --features wasi-threads-rt
run: cargo test --target ${{ matrix.target }} --features wasi-threads-rt
if: matrix.target == 'wasm32-wasi-preview1-threads'
working-directory: tests-integration
env:
CARGO_TARGET_WASM32_WASI_PREVIEW1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
check-external-types:
name: check-external-types (${{ matrix.os }})
needs: basics

1
tests-integration/Cargo.toml
View File

@ -39,6 +39,7 @@ rt-process-signal = ["rt-net", "tokio/process", "tokio/signal"]
# This is an explicit feature so we can use `cargo hack` testing single features
# instead of all possible permutations.
wasi-rt = ["rt", "macros", "sync"]
wasi-threads-rt = ["wasi-rt", "rt-multi-thread"]
full = [
"macros",

6
tests-integration/tests/macros_main.rs
View File

@ -1,8 +1,4 @@
#![cfg(all(
feature = "macros",
feature = "rt-multi-thread",
not(target_os = "wasi")
))]
#![cfg(all(feature = "macros", feature = "rt-multi-thread"))]
#[tokio::main]
async fn basic_main() -> usize {

5
tests-integration/tests/macros_select.rs
View File

@ -4,7 +4,10 @@ use futures::channel::oneshot;
use futures::executor::block_on;
use std::thread;
#[cfg_attr(target_os = "wasi", ignore = "WASI: std::thread::spawn not supported")]
#[cfg_attr(
not(feature = "rt-multi-thread"),
ignore = "WASI: std::thread::spawn not supported"
)]
#[test]
fn join_with_select() {
block_on(async {

2
tokio-util/src/task/mod.rs
View File

@ -2,9 +2,7 @@
#[cfg(tokio_unstable)]
mod join_map;
#[cfg(not(target_os = "wasi"))]
mod spawn_pinned;
#[cfg(not(target_os = "wasi"))]
pub use spawn_pinned::LocalPoolHandle;
#[cfg(tokio_unstable)]

2
tokio/src/macros/cfg.rs
View File

@ -384,7 +384,7 @@ macro_rules! cfg_not_rt {
macro_rules! cfg_rt_multi_thread {
($($item:item)*) => {
$(
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))]
$item
)*

8
tokio/src/runtime/blocking/schedule.rs
View File

@ -23,9 +23,9 @@ impl BlockingSchedule {
scheduler::Handle::CurrentThread(handle) => {
handle.driver.clock.inhibit_auto_advance();
}
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
scheduler::Handle::MultiThread(_) => {}
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
scheduler::Handle::MultiThreadAlt(_) => {}
}
}
@ -45,9 +45,9 @@ impl task::Schedule for BlockingSchedule {
handle.driver.clock.allow_auto_advance();
handle.driver.unpark();
}
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
scheduler::Handle::MultiThread(_) => {}
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
scheduler::Handle::MultiThreadAlt(_) => {}
}
}

58
tokio/src/runtime/builder.rs
View File

@ -199,9 +199,9 @@ pub(crate) type ThreadNameFn = std::sync::Arc<dyn Fn() -> String + Send + Sync +
#[derive(Clone, Copy)]
pub(crate) enum Kind {
CurrentThread,
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
MultiThread,
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
MultiThreadAlt,
}
@ -224,35 +224,33 @@ impl Builder {
Builder::new(Kind::CurrentThread, EVENT_INTERVAL)
}
cfg_not_wasi! {
/// Returns a new builder with the multi thread scheduler selected.
/// Returns a new builder with the multi thread scheduler selected.
///
/// Configuration methods can be chained on the return value.
#[cfg(feature = "rt-multi-thread")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))]
pub fn new_multi_thread() -> Builder {
// The number `61` is fairly arbitrary. I believe this value was copied from golang.
Builder::new(Kind::MultiThread, 61)
}
cfg_unstable! {
/// Returns a new builder with the alternate multi thread scheduler
/// selected.
///
/// The alternate multi threaded scheduler is an in-progress
/// candidate to replace the existing multi threaded scheduler. It
/// currently does not scale as well to 16+ processors.
///
/// This runtime flavor is currently **not considered production
/// ready**.
///
/// Configuration methods can be chained on the return value.
#[cfg(feature = "rt-multi-thread")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))]
pub fn new_multi_thread() -> Builder {
pub fn new_multi_thread_alt() -> Builder {
// The number `61` is fairly arbitrary. I believe this value was copied from golang.
Builder::new(Kind::MultiThread, 61)
}
cfg_unstable! {
/// Returns a new builder with the alternate multi thread scheduler
/// selected.
///
/// The alternate multi threaded scheduler is an in-progress
/// candidate to replace the existing multi threaded scheduler. It
/// currently does not scale as well to 16+ processors.
///
/// This runtime flavor is currently **not considered production
/// ready**.
///
/// Configuration methods can be chained on the return value.
#[cfg(feature = "rt-multi-thread")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))]
pub fn new_multi_thread_alt() -> Builder {
// The number `61` is fairly arbitrary. I believe this value was copied from golang.
Builder::new(Kind::MultiThreadAlt, 61)
}
Builder::new(Kind::MultiThreadAlt, 61)
}
}
@ -697,9 +695,9 @@ impl Builder {
pub fn build(&mut self) -> io::Result<Runtime> {
match &self.kind {
Kind::CurrentThread => self.build_current_thread_runtime(),
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
Kind::MultiThread => self.build_threaded_runtime(),
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
Kind::MultiThreadAlt => self.build_alt_threaded_runtime(),
}
}
@ -708,9 +706,9 @@ impl Builder {
driver::Cfg {
enable_pause_time: match self.kind {
Kind::CurrentThread => true,
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
Kind::MultiThread => false,
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
Kind::MultiThreadAlt => false,
},
enable_io: self.enable_io,

8
tokio/src/runtime/handle.rs
View File

@ -355,9 +355,9 @@ impl Handle {
pub fn runtime_flavor(&self) -> RuntimeFlavor {
match self.inner {
scheduler::Handle::CurrentThread(_) => RuntimeFlavor::CurrentThread,
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
scheduler::Handle::MultiThread(_) => RuntimeFlavor::MultiThread,
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
scheduler::Handle::MultiThreadAlt(_) => RuntimeFlavor::MultiThreadAlt,
}
}
@ -385,9 +385,9 @@ impl Handle {
pub fn id(&self) -> runtime::Id {
let owned_id = match &self.inner {
scheduler::Handle::CurrentThread(handle) => handle.owned_id(),
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
scheduler::Handle::MultiThread(handle) => handle.owned_id(),
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
scheduler::Handle::MultiThreadAlt(handle) => handle.owned_id(),
};
owned_id.into()

78
tokio/src/runtime/runtime.rs
View File

@ -126,11 +126,11 @@ pub(super) enum Scheduler {
CurrentThread(CurrentThread),
/// Execute tasks across multiple threads.
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
MultiThread(MultiThread),
/// Execute tasks across multiple threads.
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
MultiThreadAlt(MultiThreadAlt),
}
@ -147,40 +147,38 @@ impl Runtime {
}
}
cfg_not_wasi! {
/// Creates a new runtime instance with default configuration values.
///
/// This results in the multi threaded scheduler, I/O driver, and time driver being
/// initialized.
///
/// Most applications will not need to call this function directly. Instead,
/// they will use the [`#[tokio::main]` attribute][main]. When a more complex
/// configuration is necessary, the [runtime builder] may be used.
///
/// See [module level][mod] documentation for more details.
///
/// # Examples
///
/// Creating a new `Runtime` with default configuration values.
///
/// ```
/// use tokio::runtime::Runtime;
///
/// let rt = Runtime::new()
/// .unwrap();
///
/// // Use the runtime...
/// ```
///
/// [mod]: index.html
/// [main]: ../attr.main.html
/// [threaded scheduler]: index.html#threaded-scheduler
/// [runtime builder]: crate::runtime::Builder
#[cfg(feature = "rt-multi-thread")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))]
pub fn new() -> std::io::Result<Runtime> {
Builder::new_multi_thread().enable_all().build()
}
/// Creates a new runtime instance with default configuration values.
///
/// This results in the multi threaded scheduler, I/O driver, and time driver being
/// initialized.
///
/// Most applications will not need to call this function directly. Instead,
/// they will use the [`#[tokio::main]` attribute][main]. When a more complex
/// configuration is necessary, the [runtime builder] may be used.
///
/// See [module level][mod] documentation for more details.
///
/// # Examples
///
/// Creating a new `Runtime` with default configuration values.
///
/// ```
/// use tokio::runtime::Runtime;
///
/// let rt = Runtime::new()
/// .unwrap();
///
/// // Use the runtime...
/// ```
///
/// [mod]: index.html
/// [main]: ../attr.main.html
/// [threaded scheduler]: index.html#threaded-scheduler
/// [runtime builder]: crate::runtime::Builder
#[cfg(feature = "rt-multi-thread")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))]
pub fn new() -> std::io::Result<Runtime> {
Builder::new_multi_thread().enable_all().build()
}
/// Returns a handle to the runtime's spawner.
@ -347,9 +345,9 @@ impl Runtime {
match &self.scheduler {
Scheduler::CurrentThread(exec) => exec.block_on(&self.handle.inner, future),
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
Scheduler::MultiThread(exec) => exec.block_on(&self.handle.inner, future),
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
Scheduler::MultiThreadAlt(exec) => exec.block_on(&self.handle.inner, future),
}
}
@ -469,13 +467,13 @@ impl Drop for Runtime {
let _guard = context::try_set_current(&self.handle.inner);
current_thread.shutdown(&self.handle.inner);
}
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
Scheduler::MultiThread(multi_thread) => {
// The threaded scheduler drops its tasks on its worker threads, which is
// already in the runtime's context.
multi_thread.shutdown(&self.handle.inner);
}
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
Scheduler::MultiThreadAlt(multi_thread) => {
// The threaded scheduler drops its tasks on its worker threads, which is
// already in the runtime's context.

5
tokio/src/runtime/scheduler/inject/shared.rs
View File

@ -38,10 +38,7 @@ impl<T: 'static> Shared<T> {
}
// Kind of annoying to have to include the cfg here
#[cfg(any(
tokio_taskdump,
all(feature = "rt-multi-thread", not(target_os = "wasi"))
))]
#[cfg(any(tokio_taskdump, feature = "rt-multi-thread"))]
pub(crate) fn is_closed(&self, synced: &Synced) -> bool {
synced.is_closed
}

32
tokio/src/runtime/scheduler/mod.rs
View File

@ -32,10 +32,10 @@ pub(crate) enum Handle {
#[cfg(feature = "rt")]
CurrentThread(Arc<current_thread::Handle>),
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
MultiThread(Arc<multi_thread::Handle>),
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
MultiThreadAlt(Arc<multi_thread_alt::Handle>),
// TODO: This is to avoid triggering "dead code" warnings many other places
@ -49,10 +49,10 @@ pub(crate) enum Handle {
pub(super) enum Context {
CurrentThread(current_thread::Context),
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
MultiThread(multi_thread::Context),
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
MultiThreadAlt(multi_thread_alt::Context),
}
@ -63,10 +63,10 @@ impl Handle {
#[cfg(feature = "rt")]
Handle::CurrentThread(ref h) => &h.driver,
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
Handle::MultiThread(ref h) => &h.driver,
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
Handle::MultiThreadAlt(ref h) => &h.driver,
#[cfg(not(feature = "rt"))]
@ -89,10 +89,10 @@ cfg_rt! {
match $self {
$ty::CurrentThread($h) => $e,
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
$ty::MultiThread($h) => $e,
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
$ty::MultiThreadAlt($h) => $e,
}
}
@ -119,10 +119,10 @@ cfg_rt! {
match self {
Handle::CurrentThread(h) => current_thread::Handle::spawn(h, future, id),
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
Handle::MultiThread(h) => multi_thread::Handle::spawn(h, future, id),
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
Handle::MultiThreadAlt(h) => multi_thread_alt::Handle::spawn(h, future, id),
}
}
@ -131,10 +131,10 @@ cfg_rt! {
match *self {
Handle::CurrentThread(_) => {},
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
Handle::MultiThread(ref h) => h.shutdown(),
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
Handle::MultiThreadAlt(ref h) => h.shutdown(),
}
}
@ -146,7 +146,7 @@ cfg_rt! {
pub(crate) fn as_current_thread(&self) -> &Arc<current_thread::Handle> {
match self {
Handle::CurrentThread(handle) => handle,
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
_ => panic!("not a CurrentThread handle"),
}
}
@ -170,9 +170,9 @@ cfg_rt! {
pub(crate) fn num_workers(&self) -> usize {
match self {
Handle::CurrentThread(_) => 1,
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
Handle::MultiThread(handle) => handle.num_workers(),
#[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
Handle::MultiThreadAlt(handle) => handle.num_workers(),
}
}
@ -216,7 +216,7 @@ cfg_rt! {
pub(crate) fn expect_current_thread(&self) -> &current_thread::Context {
match self {
Context::CurrentThread(context) => context,
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
_ => panic!("expected `CurrentThread::Context`")
}
}

8
tokio/src/runtime/task/trace/mod.rs
View File

@ -195,13 +195,9 @@ pub(crate) fn trace_leaf(cx: &mut task::Context<'_>) -> Poll<()> {
if let Some(scheduler) = scheduler {
match scheduler {
scheduler::Context::CurrentThread(s) => s.defer.defer(cx.waker()),
#[cfg(all(feature = "rt-multi-thread", not(target_os = "wasi")))]
#[cfg(feature = "rt-multi-thread")]
scheduler::Context::MultiThread(s) => s.defer.defer(cx.waker()),
#[cfg(all(
tokio_unstable,
feature = "rt-multi-thread",
not(target_os = "wasi")
))]
#[cfg(all(tokio_unstable, feature = "rt-multi-thread"))]
scheduler::Context::MultiThreadAlt(_) => unimplemented!(),
}
}

6
tokio/src/task/mod.rs
View File

@ -318,10 +318,8 @@
cfg_rt! {
pub use crate::runtime::task::{JoinError, JoinHandle};
cfg_not_wasi! {
mod blocking;
pub use blocking::spawn_blocking;
}
mod blocking;
pub use blocking::spawn_blocking;
mod spawn;
pub use spawn::spawn;