Compare commits
54 Commits
Author | SHA1 | Date |
---|---|---|
John Nunley | f1c7ae3340 | |
John Nunley | ef512cb384 | |
Jacob Rothstein | df57d9bc98 | |
James Liu | 649bdfda23 | |
John Nunley | 4b37c612f6 | |
John Nunley | 00f0b99fad | |
John Nunley | d3196999f4 | |
John Nunley | 17720b098a | |
John Nunley | b6d3a60b44 | |
John Nunley | a2c1267c85 | |
John Nunley | 00dbbbf85d | |
John Nunley | c90fd306cd | |
John Nunley | 22a9e8b305 | |
John Nunley | d5dc7a8008 | |
John Nunley | 2f3189a4b4 | |
James Liu | c7bbe489ab | |
James Liu | 7592d4188a | |
James Liu | 188f976dc3 | |
James Liu | 568a314ad9 | |
James Liu | 7ffdf5ba92 | |
Jacob Rothstein | 0baba46152 | |
dependabot[bot] | 4fbe23af69 | |
John Nunley | 6c70369102 | |
Taiki Endo | 57fcc2d991 | |
Taiki Endo | 24510a7b72 | |
John Nunley | d747bcd827 | |
John Nunley | fa117dee27 | |
John Nunley | 4b1cf40142 | |
John Nunley | 144b0576d1 | |
John Nunley | b140c46123 | |
John Nunley | 1d4769a7b5 | |
John Nunley | 6c3d45b23c | |
John Nunley | f076528d27 | |
John Nunley | c7fd967c9e | |
John Nunley | 361c5fd359 | |
John Nunley | 457cf7b888 | |
John Nunley | e1e2ab11df | |
dependabot[bot] | b91875e73b | |
John Nunley | 599c71a3f9 | |
John Nunley | 8a0832c090 | |
John Nunley | 917caad8b9 | |
John Nunley | 2cfb6e4ed0 | |
John Nunley | 4154ad2190 | |
John Nunley | 77b5b169c5 | |
John Nunley | ecddfde87a | |
Taiki Endo | ff67cb9a5f | |
Taiki Endo | 609aafb330 | |
John Nunley | a5ff8df7d9 | |
John Nunley | e19573367b | |
John Nunley | aed7279805 | |
John Nunley | 9df3dd4974 | |
John Nunley | 1a9e08ce73 | |
Yosh | 85c20eb98b | |
Taiki Endo | 8562c41062 |
|
@ -34,33 +34,32 @@ jobs:
|
|||
os: [ubuntu-latest]
|
||||
rust: [nightly, beta, stable]
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- uses: actions/checkout@v4
|
||||
- name: Install Rust
|
||||
run: rustup update ${{ matrix.rust }} && rustup default ${{ matrix.rust }}
|
||||
- run: rustup target add wasm32-unknown-unknown
|
||||
- uses: taiki-e/install-action@cargo-hack
|
||||
- run: cargo build --all --all-features --all-targets
|
||||
if: startsWith(matrix.rust, 'nightly')
|
||||
- name: Run cargo check (without dev-dependencies to catch missing feature flags)
|
||||
if: startsWith(matrix.rust, 'nightly')
|
||||
run: cargo check -Z features=dev_dep
|
||||
- run: cargo test
|
||||
- run: cargo check --all --all-features --target wasm32-unknown-unknown
|
||||
- run: cargo hack build --all --all-features --target wasm32-unknown-unknown --no-dev-deps
|
||||
|
||||
msrv:
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
matrix:
|
||||
# When updating this, the reminder to update the minimum supported
|
||||
# Rust version in Cargo.toml.
|
||||
rust: ['1.48']
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- name: Install Rust
|
||||
run: rustup update ${{ matrix.rust }} && rustup default ${{ matrix.rust }}
|
||||
- run: cargo build
|
||||
- uses: actions/checkout@v4
|
||||
- name: Install cargo-hack
|
||||
uses: taiki-e/install-action@cargo-hack
|
||||
- run: cargo hack build --rust-version
|
||||
|
||||
clippy:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- uses: actions/checkout@v4
|
||||
- name: Install Rust
|
||||
run: rustup update stable
|
||||
- run: cargo clippy --all-features --all-targets
|
||||
|
@ -68,7 +67,7 @@ jobs:
|
|||
fmt:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- uses: actions/checkout@v4
|
||||
- name: Install Rust
|
||||
run: rustup update stable
|
||||
- run: cargo fmt --all --check
|
||||
|
@ -76,7 +75,7 @@ jobs:
|
|||
miri:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- uses: actions/checkout@v4
|
||||
- name: Install Rust
|
||||
run: rustup toolchain install nightly --component miri && rustup default nightly
|
||||
- run: cargo miri test
|
||||
|
@ -85,9 +84,13 @@ jobs:
|
|||
RUSTFLAGS: ${{ env.RUSTFLAGS }} -Z randomize-layout
|
||||
|
||||
security_audit:
|
||||
permissions:
|
||||
checks: write
|
||||
contents: read
|
||||
issues: write
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- uses: actions/checkout@v4
|
||||
# https://github.com/rustsec/audit-check/issues/2
|
||||
- uses: rustsec/audit-check@master
|
||||
with:
|
||||
|
|
|
@ -13,7 +13,7 @@ jobs:
|
|||
if: github.repository_owner == 'smol-rs'
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
- uses: actions/checkout@v4
|
||||
- uses: taiki-e/create-gh-release-action@v1
|
||||
with:
|
||||
changelog: CHANGELOG.md
|
||||
|
|
58
CHANGELOG.md
58
CHANGELOG.md
|
@ -1,3 +1,61 @@
|
|||
# Version 1.11.0
|
||||
|
||||
- Re-export the `async_task::FallibleTask` primitive. (#113)
|
||||
- Support racy initialization of the executor state. This should allow the executor to be
|
||||
initialized on web targets without any issues. (#108)
|
||||
|
||||
# Version 1.10.0
|
||||
|
||||
- Add a function `spawn_batch` that allows users to spawn multiple tasks while only locking the executor once. (#92)
|
||||
|
||||
# Version 1.9.1
|
||||
|
||||
- Remove the thread-local optimization due to the bugs that it introduces. (#106)
|
||||
|
||||
# Version 1.9.0
|
||||
|
||||
- Re-introduce the thread-local task push optimization to the executor. (#93)
|
||||
- Bump `async-task` to v4.4.0. (#90)
|
||||
- Replace some unnecessary atomic operations with non-atomic operations. (#94)
|
||||
- Use weaker atomic orderings for notifications. (#95)
|
||||
- When spawning a future, avoid looking up the ID to assign to that future twice. (#96)
|
||||
|
||||
# Version 1.8.0
|
||||
|
||||
- When spawned tasks panic, the panic is caught and then surfaced in the spawned
|
||||
`Task`. Previously, the panic would be surfaced in `tick()` or `run()`. (#78)
|
||||
|
||||
# Version 1.7.2
|
||||
|
||||
- Fix compilation under WebAssembly targets (#77).
|
||||
|
||||
# Version 1.7.1
|
||||
|
||||
- Fix compilation under WebAssembly targets (#75).
|
||||
- Add a disclaimer indicating that this is a reference executor (#74).
|
||||
|
||||
# Version 1.7.0
|
||||
|
||||
- Bump `async-lock` and `futures-lite` to their latest versions. (#70)
|
||||
|
||||
# Version 1.6.0
|
||||
|
||||
- Remove the thread-local queue optimization, as it caused a number of bugs in production use cases. (#61)
|
||||
|
||||
# Version 1.5.4
|
||||
|
||||
- Fix a panic that could happen when two concurrent `run()` calls are made and the thread local task slot is left as `None`. (#55)
|
||||
|
||||
# Version 1.5.3
|
||||
|
||||
- Fix an accidental breaking change in v1.5.2, where `ex.run()` was no longer `Send`. (#50)
|
||||
- Remove the unused `memchr` dependency. (#51)
|
||||
|
||||
# Version 1.5.2
|
||||
|
||||
- Add thread-local task queue optimizations, allowing new tasks to avoid using the global queue. (#37)
|
||||
- Update `fastrand` to v2. (#45)
|
||||
|
||||
# Version 1.5.1
|
||||
|
||||
- Implement a better form of debug output for Executor and LocalExecutor. (#33)
|
||||
|
|
28
Cargo.toml
28
Cargo.toml
|
@ -3,10 +3,10 @@ name = "async-executor"
|
|||
# When publishing a new version:
|
||||
# - Update CHANGELOG.md
|
||||
# - Create "v1.x.y" git tag
|
||||
version = "1.5.1"
|
||||
authors = ["Stjepan Glavina <stjepang@gmail.com>"]
|
||||
edition = "2018"
|
||||
rust-version = "1.48"
|
||||
version = "1.11.0"
|
||||
authors = ["Stjepan Glavina <stjepang@gmail.com>", "John Nunley <dev@notgull.net>"]
|
||||
edition = "2021"
|
||||
rust-version = "1.60"
|
||||
description = "Async executor"
|
||||
license = "Apache-2.0 OR MIT"
|
||||
repository = "https://github.com/smol-rs/async-executor"
|
||||
|
@ -15,19 +15,23 @@ categories = ["asynchronous", "concurrency"]
|
|||
exclude = ["/.*"]
|
||||
|
||||
[dependencies]
|
||||
async-lock = "2.6"
|
||||
async-task = "4.0.0"
|
||||
async-task = "4.4.0"
|
||||
concurrent-queue = "2.0.0"
|
||||
fastrand = "1.3.4"
|
||||
futures-lite = "1.11.0"
|
||||
fastrand = "2.0.0"
|
||||
futures-lite = { version = "2.0.0", default-features = false }
|
||||
slab = "0.4.4"
|
||||
|
||||
[target.'cfg(target_family = "wasm")'.dependencies]
|
||||
futures-lite = { version = "2.0.0", default-features = false, features = ["std"] }
|
||||
|
||||
[dev-dependencies]
|
||||
async-channel = "1.4.1"
|
||||
async-io = "1.1.9"
|
||||
criterion = { version = "0.4.0", default-features = false, features = ["cargo_bench_support"] }
|
||||
async-channel = "2.0.0"
|
||||
async-io = "2.1.0"
|
||||
async-lock = "3.0.0"
|
||||
criterion = { version = "0.5", default-features = false, features = ["cargo_bench_support"] }
|
||||
easy-parallel = "3.1.0"
|
||||
num_cpus = "1.13.0"
|
||||
fastrand = "2.0.0"
|
||||
futures-lite = "2.0.0"
|
||||
once_cell = "1.16.0"
|
||||
|
||||
[[bench]]
|
||||
|
|
|
@ -11,6 +11,13 @@ https://docs.rs/async-executor)
|
|||
|
||||
Async executors.
|
||||
|
||||
This crate provides two reference executors that trade performance for
|
||||
functionality. They should be considered reference executors that are "good
|
||||
enough" for most use cases. For more specialized use cases, consider writing
|
||||
your own executor on top of [`async-task`].
|
||||
|
||||
[`async-task`]: https://crates.io/crates/async-task
|
||||
|
||||
## Examples
|
||||
|
||||
```rust
|
||||
|
|
|
@ -1,4 +1,5 @@
|
|||
use std::future::Future;
|
||||
use std::mem;
|
||||
use std::thread::available_parallelism;
|
||||
|
||||
use async_executor::Executor;
|
||||
use criterion::{criterion_group, criterion_main, Criterion};
|
||||
|
@ -11,7 +12,11 @@ const LIGHT_TASKS: usize = 25_000;
|
|||
static EX: Executor<'_> = Executor::new();
|
||||
|
||||
fn run(f: impl FnOnce(), multithread: bool) {
|
||||
let limit = if multithread { num_cpus::get() } else { 1 };
|
||||
let limit = if multithread {
|
||||
available_parallelism().unwrap().get()
|
||||
} else {
|
||||
1
|
||||
};
|
||||
|
||||
let (s, r) = async_channel::bounded::<()>(1);
|
||||
easy_parallel::Parallel::new()
|
||||
|
@ -47,6 +52,21 @@ fn running_benches(c: &mut Criterion) {
|
|||
);
|
||||
});
|
||||
|
||||
group.bench_function("executor::spawn_batch", |b| {
|
||||
run(
|
||||
|| {
|
||||
let mut handles = vec![];
|
||||
|
||||
b.iter(|| {
|
||||
EX.spawn_many((0..250).map(|_| future::yield_now()), &mut handles);
|
||||
});
|
||||
|
||||
handles.clear();
|
||||
},
|
||||
*multithread,
|
||||
)
|
||||
});
|
||||
|
||||
group.bench_function("executor::spawn_many_local", |b| {
|
||||
run(
|
||||
|| {
|
||||
|
@ -120,6 +140,122 @@ fn running_benches(c: &mut Criterion) {
|
|||
*multithread,
|
||||
);
|
||||
});
|
||||
|
||||
group.bench_function("executor::channels", |b| {
|
||||
run(
|
||||
|| {
|
||||
b.iter(move || {
|
||||
future::block_on(async {
|
||||
// Create channels.
|
||||
let mut tasks = Vec::new();
|
||||
let (first_send, first_recv) = async_channel::bounded(1);
|
||||
let mut current_recv = first_recv;
|
||||
|
||||
for _ in 0..TASKS {
|
||||
let (next_send, next_recv) = async_channel::bounded(1);
|
||||
let current_recv = mem::replace(&mut current_recv, next_recv);
|
||||
|
||||
tasks.push(EX.spawn(async move {
|
||||
// Send a notification on to the next task.
|
||||
for _ in 0..STEPS {
|
||||
current_recv.recv().await.unwrap();
|
||||
next_send.send(()).await.unwrap();
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for _ in 0..STEPS {
|
||||
first_send.send(()).await.unwrap();
|
||||
current_recv.recv().await.unwrap();
|
||||
}
|
||||
|
||||
for task in tasks {
|
||||
task.await;
|
||||
}
|
||||
});
|
||||
});
|
||||
},
|
||||
*multithread,
|
||||
)
|
||||
});
|
||||
|
||||
group.bench_function("executor::web_server", |b| {
|
||||
run(
|
||||
|| {
|
||||
b.iter(move || {
|
||||
future::block_on(async {
|
||||
let (db_send, db_recv) =
|
||||
async_channel::bounded::<async_channel::Sender<_>>(TASKS / 5);
|
||||
let mut db_rng = fastrand::Rng::with_seed(0x12345678);
|
||||
let mut web_rng = db_rng.fork();
|
||||
|
||||
// This task simulates a database.
|
||||
let db_task = EX.spawn(async move {
|
||||
loop {
|
||||
// Wait for a new task.
|
||||
let incoming = match db_recv.recv().await {
|
||||
Ok(incoming) => incoming,
|
||||
Err(_) => break,
|
||||
};
|
||||
|
||||
// Process the task. Maybe it takes a while.
|
||||
for _ in 0..db_rng.usize(..10) {
|
||||
future::yield_now().await;
|
||||
}
|
||||
|
||||
// Send the data back.
|
||||
incoming.send(db_rng.usize(..)).await.ok();
|
||||
}
|
||||
});
|
||||
|
||||
// This task simulates a web server waiting for new tasks.
|
||||
let server_task = EX.spawn(async move {
|
||||
for i in 0..TASKS {
|
||||
// Get a new connection.
|
||||
if web_rng.usize(..=16) == 16 {
|
||||
future::yield_now().await;
|
||||
}
|
||||
|
||||
let mut web_rng = web_rng.fork();
|
||||
let db_send = db_send.clone();
|
||||
let task = EX.spawn(async move {
|
||||
// Check if the data is cached...
|
||||
if web_rng.bool() {
|
||||
// ...it's in cache!
|
||||
future::yield_now().await;
|
||||
return;
|
||||
}
|
||||
|
||||
// Otherwise we have to make a DB call or two.
|
||||
for _ in 0..web_rng.usize(STEPS / 2..STEPS) {
|
||||
let (resp_send, resp_recv) = async_channel::bounded(1);
|
||||
db_send.send(resp_send).await.unwrap();
|
||||
criterion::black_box(resp_recv.recv().await.unwrap());
|
||||
}
|
||||
|
||||
// Send the data back...
|
||||
for _ in 0..web_rng.usize(3..16) {
|
||||
future::yield_now().await;
|
||||
}
|
||||
});
|
||||
|
||||
task.detach();
|
||||
|
||||
if i & 16 == 0 {
|
||||
future::yield_now().await;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Spawn and wait for it to stop.
|
||||
server_task.await;
|
||||
db_task.await;
|
||||
});
|
||||
})
|
||||
},
|
||||
*multithread,
|
||||
)
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -0,0 +1,95 @@
|
|||
//! An executor where you can only push a limited number of tasks.
|
||||
|
||||
use async_executor::{Executor, Task};
|
||||
use async_lock::Semaphore;
|
||||
use std::{future::Future, sync::Arc, time::Duration};
|
||||
|
||||
/// An executor where you can only push a limited number of tasks.
|
||||
struct LimitedExecutor {
|
||||
/// Inner running executor.
|
||||
executor: Executor<'static>,
|
||||
|
||||
/// Semaphore limiting the number of tasks.
|
||||
semaphore: Arc<Semaphore>,
|
||||
}
|
||||
|
||||
impl LimitedExecutor {
|
||||
fn new(max: usize) -> Self {
|
||||
Self {
|
||||
executor: Executor::new(),
|
||||
semaphore: Semaphore::new(max).into(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn a task, waiting until there is a slot available.
|
||||
async fn spawn<F: Future + Send + 'static>(&self, future: F) -> Task<F::Output>
|
||||
where
|
||||
F::Output: Send + 'static,
|
||||
{
|
||||
// Wait for a semaphore permit.
|
||||
let permit = self.semaphore.acquire_arc().await;
|
||||
|
||||
// Wrap it into a new future.
|
||||
let future = async move {
|
||||
let result = future.await;
|
||||
drop(permit);
|
||||
result
|
||||
};
|
||||
|
||||
// Spawn the task.
|
||||
self.executor.spawn(future)
|
||||
}
|
||||
|
||||
/// Run a future to completion.
|
||||
async fn run<F: Future>(&self, future: F) -> F::Output {
|
||||
self.executor.run(future).await
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
futures_lite::future::block_on(async {
|
||||
let ex = Arc::new(LimitedExecutor::new(10));
|
||||
ex.run({
|
||||
let ex = ex.clone();
|
||||
async move {
|
||||
// Spawn a bunch of tasks that wait for a while.
|
||||
for i in 0..15 {
|
||||
ex.spawn(async move {
|
||||
async_io::Timer::after(Duration::from_millis(fastrand::u64(1..3))).await;
|
||||
println!("Waiting task #{i} finished!");
|
||||
})
|
||||
.await
|
||||
.detach();
|
||||
}
|
||||
|
||||
let (start_tx, start_rx) = async_channel::bounded::<()>(1);
|
||||
let mut current_rx = start_rx;
|
||||
|
||||
// Send the first message.
|
||||
start_tx.send(()).await.unwrap();
|
||||
|
||||
// Spawn a bunch of channel tasks that wake eachother up.
|
||||
for i in 0..25 {
|
||||
let (next_tx, next_rx) = async_channel::bounded::<()>(1);
|
||||
|
||||
ex.spawn(async move {
|
||||
current_rx.recv().await.unwrap();
|
||||
println!("Channel task {i} woken up!");
|
||||
next_tx.send(()).await.unwrap();
|
||||
println!("Channel task {i} finished!");
|
||||
})
|
||||
.await
|
||||
.detach();
|
||||
|
||||
current_rx = next_rx;
|
||||
}
|
||||
|
||||
// Wait for the last task to finish.
|
||||
current_rx.recv().await.unwrap();
|
||||
|
||||
println!("All tasks finished!");
|
||||
}
|
||||
})
|
||||
.await;
|
||||
});
|
||||
}
|
|
@ -1,6 +1,5 @@
|
|||
//! An executor with task priorities.
|
||||
|
||||
use std::future::Future;
|
||||
use std::thread;
|
||||
|
||||
use async_executor::{Executor, Task};
|
||||
|
|
435
src/lib.rs
435
src/lib.rs
|
@ -1,5 +1,12 @@
|
|||
//! Async executors.
|
||||
//!
|
||||
//! This crate provides two reference executors that trade performance for
|
||||
//! functionality. They should be considered reference executors that are "good
|
||||
//! enough" for most use cases. For more specialized use cases, consider writing
|
||||
//! your own executor on top of [`async-task`].
|
||||
//!
|
||||
//! [`async-task`]: https://crates.io/crates/async-task
|
||||
//!
|
||||
//! # Examples
|
||||
//!
|
||||
//! ```
|
||||
|
@ -18,25 +25,34 @@
|
|||
//! future::block_on(ex.run(task));
|
||||
//! ```
|
||||
|
||||
#![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]
|
||||
#![warn(
|
||||
missing_docs,
|
||||
missing_debug_implementations,
|
||||
rust_2018_idioms,
|
||||
clippy::undocumented_unsafe_blocks
|
||||
)]
|
||||
#![doc(
|
||||
html_favicon_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
|
||||
)]
|
||||
#![doc(
|
||||
html_logo_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
|
||||
)]
|
||||
|
||||
use std::fmt;
|
||||
use std::future::Future;
|
||||
use std::marker::PhantomData;
|
||||
use std::panic::{RefUnwindSafe, UnwindSafe};
|
||||
use std::rc::Rc;
|
||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicPtr, Ordering};
|
||||
use std::sync::{Arc, Mutex, RwLock, TryLockError};
|
||||
use std::task::{Poll, Waker};
|
||||
|
||||
use async_lock::OnceCell;
|
||||
use async_task::Runnable;
|
||||
use async_task::{Builder, Runnable};
|
||||
use concurrent_queue::ConcurrentQueue;
|
||||
use futures_lite::{future, prelude::*};
|
||||
use slab::Slab;
|
||||
|
||||
#[doc(no_inline)]
|
||||
pub use async_task::Task;
|
||||
pub use async_task::{FallibleTask, Task};
|
||||
|
||||
/// An async executor.
|
||||
///
|
||||
|
@ -64,13 +80,15 @@ pub use async_task::Task;
|
|||
/// ```
|
||||
pub struct Executor<'a> {
|
||||
/// The executor state.
|
||||
state: OnceCell<Arc<State>>,
|
||||
state: AtomicPtr<State>,
|
||||
|
||||
/// Makes the `'a` lifetime invariant.
|
||||
_marker: PhantomData<std::cell::UnsafeCell<&'a ()>>,
|
||||
}
|
||||
|
||||
// SAFETY: Executor stores no thread local state that can be accessed via other thread.
|
||||
unsafe impl Send for Executor<'_> {}
|
||||
// SAFETY: Executor internally synchronizes all of it's operations internally.
|
||||
unsafe impl Sync for Executor<'_> {}
|
||||
|
||||
impl UnwindSafe for Executor<'_> {}
|
||||
|
@ -94,7 +112,7 @@ impl<'a> Executor<'a> {
|
|||
/// ```
|
||||
pub const fn new() -> Executor<'a> {
|
||||
Executor {
|
||||
state: OnceCell::new(),
|
||||
state: AtomicPtr::new(std::ptr::null_mut()),
|
||||
_marker: PhantomData,
|
||||
}
|
||||
}
|
||||
|
@ -137,17 +155,120 @@ impl<'a> Executor<'a> {
|
|||
pub fn spawn<T: Send + 'a>(&self, future: impl Future<Output = T> + Send + 'a) -> Task<T> {
|
||||
let mut active = self.state().active.lock().unwrap();
|
||||
|
||||
// SAFETY: `T` and the future are `Send`.
|
||||
unsafe { self.spawn_inner(future, &mut active) }
|
||||
}
|
||||
|
||||
/// Spawns many tasks onto the executor.
|
||||
///
|
||||
/// As opposed to the [`spawn`] method, this locks the executor's inner task lock once and
|
||||
/// spawns all of the tasks in one go. With large amounts of tasks this can improve
|
||||
/// contention.
|
||||
///
|
||||
/// For very large numbers of tasks the lock is occasionally dropped and re-acquired to
|
||||
/// prevent runner thread starvation. It is assumed that the iterator provided does not
|
||||
/// block; blocking iterators can lock up the internal mutex and therefore the entire
|
||||
/// executor.
|
||||
///
|
||||
/// ## Example
|
||||
///
|
||||
/// ```
|
||||
/// use async_executor::Executor;
|
||||
/// use futures_lite::{stream, prelude::*};
|
||||
/// use std::future::ready;
|
||||
///
|
||||
/// # futures_lite::future::block_on(async {
|
||||
/// let mut ex = Executor::new();
|
||||
///
|
||||
/// let futures = [
|
||||
/// ready(1),
|
||||
/// ready(2),
|
||||
/// ready(3)
|
||||
/// ];
|
||||
///
|
||||
/// // Spawn all of the futures onto the executor at once.
|
||||
/// let mut tasks = vec![];
|
||||
/// ex.spawn_many(futures, &mut tasks);
|
||||
///
|
||||
/// // Await all of them.
|
||||
/// let results = ex.run(async move {
|
||||
/// stream::iter(tasks).then(|x| x).collect::<Vec<_>>().await
|
||||
/// }).await;
|
||||
/// assert_eq!(results, [1, 2, 3]);
|
||||
/// # });
|
||||
/// ```
|
||||
///
|
||||
/// [`spawn`]: Executor::spawn
|
||||
pub fn spawn_many<T: Send + 'a, F: Future<Output = T> + Send + 'a>(
|
||||
&self,
|
||||
futures: impl IntoIterator<Item = F>,
|
||||
handles: &mut impl Extend<Task<F::Output>>,
|
||||
) {
|
||||
let mut active = Some(self.state().active.lock().unwrap());
|
||||
|
||||
// Convert the futures into tasks.
|
||||
let tasks = futures.into_iter().enumerate().map(move |(i, future)| {
|
||||
// SAFETY: `T` and the future are `Send`.
|
||||
let task = unsafe { self.spawn_inner(future, active.as_mut().unwrap()) };
|
||||
|
||||
// Yield the lock every once in a while to ease contention.
|
||||
if i.wrapping_sub(1) % 500 == 0 {
|
||||
drop(active.take());
|
||||
active = Some(self.state().active.lock().unwrap());
|
||||
}
|
||||
|
||||
task
|
||||
});
|
||||
|
||||
// Push the tasks to the user's collection.
|
||||
handles.extend(tasks);
|
||||
}
|
||||
|
||||
/// Spawn a future while holding the inner lock.
|
||||
///
|
||||
/// # Safety
|
||||
///
|
||||
/// If this is an `Executor`, `F` and `T` must be `Send`.
|
||||
unsafe fn spawn_inner<T: 'a>(
|
||||
&self,
|
||||
future: impl Future<Output = T> + 'a,
|
||||
active: &mut Slab<Waker>,
|
||||
) -> Task<T> {
|
||||
// Remove the task from the set of active tasks when the future finishes.
|
||||
let index = active.vacant_entry().key();
|
||||
let state = self.state().clone();
|
||||
let entry = active.vacant_entry();
|
||||
let index = entry.key();
|
||||
let state = self.state_as_arc();
|
||||
let future = async move {
|
||||
let _guard = CallOnDrop(move || drop(state.active.lock().unwrap().try_remove(index)));
|
||||
future.await
|
||||
};
|
||||
|
||||
// Create the task and register it in the set of active tasks.
|
||||
let (runnable, task) = unsafe { async_task::spawn_unchecked(future, self.schedule()) };
|
||||
active.insert(runnable.waker());
|
||||
//
|
||||
// SAFETY:
|
||||
//
|
||||
// If `future` is not `Send`, this must be a `LocalExecutor` as per this
|
||||
// function's unsafe precondition. Since `LocalExecutor` is `!Sync`,
|
||||
// `try_tick`, `tick` and `run` can only be called from the origin
|
||||
// thread of the `LocalExecutor`. Similarly, `spawn` can only be called
|
||||
// from the origin thread, ensuring that `future` and the executor share
|
||||
// the same origin thread. The `Runnable` can be scheduled from other
|
||||
// threads, but because of the above `Runnable` can only be called or
|
||||
// dropped on the origin thread.
|
||||
//
|
||||
// `future` is not `'static`, but we make sure that the `Runnable` does
|
||||
// not outlive `'a`. When the executor is dropped, the `active` field is
|
||||
// drained and all of the `Waker`s are woken. Then, the queue inside of
|
||||
// the `Executor` is drained of all of its runnables. This ensures that
|
||||
// runnables are dropped and this precondition is satisfied.
|
||||
//
|
||||
// `self.schedule()` is `Send`, `Sync` and `'static`, as checked below.
|
||||
// Therefore we do not need to worry about what is done with the
|
||||
// `Waker`.
|
||||
let (runnable, task) = Builder::new()
|
||||
.propagate_panic(true)
|
||||
.spawn_unchecked(|()| future, self.schedule());
|
||||
entry.insert(runnable.waker());
|
||||
|
||||
runnable.schedule();
|
||||
task
|
||||
|
@ -226,13 +347,14 @@ impl<'a> Executor<'a> {
|
|||
/// assert_eq!(res, 6);
|
||||
/// ```
|
||||
pub async fn run<T>(&self, future: impl Future<Output = T>) -> T {
|
||||
let runner = Runner::new(self.state());
|
||||
let mut runner = Runner::new(self.state());
|
||||
let mut rng = fastrand::Rng::new();
|
||||
|
||||
// A future that runs tasks forever.
|
||||
let run_forever = async {
|
||||
loop {
|
||||
for _ in 0..200 {
|
||||
let runnable = runner.runnable().await;
|
||||
let runnable = runner.runnable(&mut rng).await;
|
||||
runnable.run();
|
||||
}
|
||||
future::yield_now().await;
|
||||
|
@ -245,32 +367,82 @@ impl<'a> Executor<'a> {
|
|||
|
||||
/// Returns a function that schedules a runnable task when it gets woken up.
|
||||
fn schedule(&self) -> impl Fn(Runnable) + Send + Sync + 'static {
|
||||
let state = self.state().clone();
|
||||
let state = self.state_as_arc();
|
||||
|
||||
// TODO(stjepang): If possible, push into the current local queue and notify the ticker.
|
||||
// TODO: If possible, push into the current local queue and notify the ticker.
|
||||
move |runnable| {
|
||||
state.queue.push(runnable).unwrap();
|
||||
state.notify();
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a pointer to the inner state.
|
||||
#[inline]
|
||||
fn state_ptr(&self) -> *const State {
|
||||
#[cold]
|
||||
fn alloc_state(atomic_ptr: &AtomicPtr<State>) -> *mut State {
|
||||
let state = Arc::new(State::new());
|
||||
// TODO: Switch this to use cast_mut once the MSRV can be bumped past 1.65
|
||||
let ptr = Arc::into_raw(state) as *mut State;
|
||||
if let Err(actual) = atomic_ptr.compare_exchange(
|
||||
std::ptr::null_mut(),
|
||||
ptr,
|
||||
Ordering::AcqRel,
|
||||
Ordering::Acquire,
|
||||
) {
|
||||
// SAFETY: This was just created from Arc::into_raw.
|
||||
drop(unsafe { Arc::from_raw(ptr) });
|
||||
actual
|
||||
} else {
|
||||
ptr
|
||||
}
|
||||
}
|
||||
|
||||
let mut ptr = self.state.load(Ordering::Acquire);
|
||||
if ptr.is_null() {
|
||||
ptr = alloc_state(&self.state);
|
||||
}
|
||||
ptr
|
||||
}
|
||||
|
||||
/// Returns a reference to the inner state.
|
||||
fn state(&self) -> &Arc<State> {
|
||||
self.state.get_or_init_blocking(|| Arc::new(State::new()))
|
||||
#[inline]
|
||||
fn state(&self) -> &State {
|
||||
// SAFETY: So long as an Executor lives, it's state pointer will always be valid
|
||||
// when accessed through state_ptr.
|
||||
unsafe { &*self.state_ptr() }
|
||||
}
|
||||
|
||||
// Clones the inner state Arc
|
||||
#[inline]
|
||||
fn state_as_arc(&self) -> Arc<State> {
|
||||
// SAFETY: So long as an Executor lives, it's state pointer will always be a valid
|
||||
// Arc when accessed through state_ptr.
|
||||
let arc = unsafe { Arc::from_raw(self.state_ptr()) };
|
||||
let clone = arc.clone();
|
||||
std::mem::forget(arc);
|
||||
clone
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Executor<'_> {
|
||||
fn drop(&mut self) {
|
||||
if let Some(state) = self.state.get() {
|
||||
let mut active = state.active.lock().unwrap();
|
||||
for w in active.drain() {
|
||||
w.wake();
|
||||
}
|
||||
drop(active);
|
||||
|
||||
while state.queue.pop().is_ok() {}
|
||||
let ptr = *self.state.get_mut();
|
||||
if ptr.is_null() {
|
||||
return;
|
||||
}
|
||||
|
||||
// SAFETY: As ptr is not null, it was allocated via Arc::new and converted
|
||||
// via Arc::into_raw in state_ptr.
|
||||
let state = unsafe { Arc::from_raw(ptr) };
|
||||
|
||||
let mut active = state.active.lock().unwrap_or_else(|e| e.into_inner());
|
||||
for w in active.drain() {
|
||||
w.wake();
|
||||
}
|
||||
drop(active);
|
||||
|
||||
while state.queue.pop().is_ok() {}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -368,20 +540,70 @@ impl<'a> LocalExecutor<'a> {
|
|||
pub fn spawn<T: 'a>(&self, future: impl Future<Output = T> + 'a) -> Task<T> {
|
||||
let mut active = self.inner().state().active.lock().unwrap();
|
||||
|
||||
// Remove the task from the set of active tasks when the future finishes.
|
||||
let index = active.vacant_entry().key();
|
||||
let state = self.inner().state().clone();
|
||||
let future = async move {
|
||||
let _guard = CallOnDrop(move || drop(state.active.lock().unwrap().try_remove(index)));
|
||||
future.await
|
||||
};
|
||||
// SAFETY: This executor is not thread safe, so the future and its result
|
||||
// cannot be sent to another thread.
|
||||
unsafe { self.inner().spawn_inner(future, &mut active) }
|
||||
}
|
||||
|
||||
// Create the task and register it in the set of active tasks.
|
||||
let (runnable, task) = unsafe { async_task::spawn_unchecked(future, self.schedule()) };
|
||||
active.insert(runnable.waker());
|
||||
/// Spawns many tasks onto the executor.
|
||||
///
|
||||
/// As opposed to the [`spawn`] method, this locks the executor's inner task lock once and
|
||||
/// spawns all of the tasks in one go. With large amounts of tasks this can improve
|
||||
/// contention.
|
||||
///
|
||||
/// It is assumed that the iterator provided does not block; blocking iterators can lock up
|
||||
/// the internal mutex and therefore the entire executor. Unlike [`Executor::spawn`], the
|
||||
/// mutex is not released, as there are no other threads that can poll this executor.
|
||||
///
|
||||
/// ## Example
|
||||
///
|
||||
/// ```
|
||||
/// use async_executor::LocalExecutor;
|
||||
/// use futures_lite::{stream, prelude::*};
|
||||
/// use std::future::ready;
|
||||
///
|
||||
/// # futures_lite::future::block_on(async {
|
||||
/// let mut ex = LocalExecutor::new();
|
||||
///
|
||||
/// let futures = [
|
||||
/// ready(1),
|
||||
/// ready(2),
|
||||
/// ready(3)
|
||||
/// ];
|
||||
///
|
||||
/// // Spawn all of the futures onto the executor at once.
|
||||
/// let mut tasks = vec![];
|
||||
/// ex.spawn_many(futures, &mut tasks);
|
||||
///
|
||||
/// // Await all of them.
|
||||
/// let results = ex.run(async move {
|
||||
/// stream::iter(tasks).then(|x| x).collect::<Vec<_>>().await
|
||||
/// }).await;
|
||||
/// assert_eq!(results, [1, 2, 3]);
|
||||
/// # });
|
||||
/// ```
|
||||
///
|
||||
/// [`spawn`]: LocalExecutor::spawn
|
||||
/// [`Executor::spawn_many`]: Executor::spawn_many
|
||||
pub fn spawn_many<T: Send + 'a, F: Future<Output = T> + Send + 'a>(
|
||||
&self,
|
||||
futures: impl IntoIterator<Item = F>,
|
||||
handles: &mut impl Extend<Task<F::Output>>,
|
||||
) {
|
||||
let mut active = self.inner().state().active.lock().unwrap();
|
||||
|
||||
runnable.schedule();
|
||||
task
|
||||
// Convert all of the futures to tasks.
|
||||
let tasks = futures.into_iter().map(|future| {
|
||||
// SAFETY: This executor is not thread safe, so the future and its result
|
||||
// cannot be sent to another thread.
|
||||
unsafe { self.inner().spawn_inner(future, &mut active) }
|
||||
|
||||
// As only one thread can spawn or poll tasks at a time, there is no need
|
||||
// to release lock contention here.
|
||||
});
|
||||
|
||||
// Push them to the user's collection.
|
||||
handles.extend(tasks);
|
||||
}
|
||||
|
||||
/// Attempts to run a task if at least one is scheduled.
|
||||
|
@ -447,16 +669,6 @@ impl<'a> LocalExecutor<'a> {
|
|||
self.inner().run(future).await
|
||||
}
|
||||
|
||||
/// Returns a function that schedules a runnable task when it gets woken up.
|
||||
fn schedule(&self) -> impl Fn(Runnable) + Send + Sync + 'static {
|
||||
let state = self.inner().state().clone();
|
||||
|
||||
move |runnable| {
|
||||
state.queue.push(runnable).unwrap();
|
||||
state.notify();
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a reference to the inner executor.
|
||||
fn inner(&self) -> &Executor<'a> {
|
||||
&self.inner
|
||||
|
@ -508,7 +720,7 @@ impl State {
|
|||
fn notify(&self) {
|
||||
if self
|
||||
.notified
|
||||
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
.is_ok()
|
||||
{
|
||||
let waker = self.sleepers.lock().unwrap().notify();
|
||||
|
@ -551,9 +763,7 @@ impl Sleepers {
|
|||
fn update(&mut self, id: usize, waker: &Waker) -> bool {
|
||||
for item in &mut self.wakers {
|
||||
if item.0 == id {
|
||||
if !item.1.will_wake(waker) {
|
||||
item.1 = waker.clone();
|
||||
}
|
||||
item.1.clone_from(waker);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
@ -606,29 +816,26 @@ struct Ticker<'a> {
|
|||
/// 1) Woken.
|
||||
/// 2a) Sleeping and unnotified.
|
||||
/// 2b) Sleeping and notified.
|
||||
sleeping: AtomicUsize,
|
||||
sleeping: usize,
|
||||
}
|
||||
|
||||
impl Ticker<'_> {
|
||||
/// Creates a ticker.
|
||||
fn new(state: &State) -> Ticker<'_> {
|
||||
Ticker {
|
||||
state,
|
||||
sleeping: AtomicUsize::new(0),
|
||||
}
|
||||
Ticker { state, sleeping: 0 }
|
||||
}
|
||||
|
||||
/// Moves the ticker into sleeping and unnotified state.
|
||||
///
|
||||
/// Returns `false` if the ticker was already sleeping and unnotified.
|
||||
fn sleep(&self, waker: &Waker) -> bool {
|
||||
fn sleep(&mut self, waker: &Waker) -> bool {
|
||||
let mut sleepers = self.state.sleepers.lock().unwrap();
|
||||
|
||||
match self.sleeping.load(Ordering::SeqCst) {
|
||||
match self.sleeping {
|
||||
// Move to sleeping state.
|
||||
0 => self
|
||||
.sleeping
|
||||
.store(sleepers.insert(waker), Ordering::SeqCst),
|
||||
0 => {
|
||||
self.sleeping = sleepers.insert(waker);
|
||||
}
|
||||
|
||||
// Already sleeping, check if notified.
|
||||
id => {
|
||||
|
@ -640,31 +847,31 @@ impl Ticker<'_> {
|
|||
|
||||
self.state
|
||||
.notified
|
||||
.swap(sleepers.is_notified(), Ordering::SeqCst);
|
||||
.store(sleepers.is_notified(), Ordering::Release);
|
||||
|
||||
true
|
||||
}
|
||||
|
||||
/// Moves the ticker into woken state.
|
||||
fn wake(&self) {
|
||||
let id = self.sleeping.swap(0, Ordering::SeqCst);
|
||||
if id != 0 {
|
||||
fn wake(&mut self) {
|
||||
if self.sleeping != 0 {
|
||||
let mut sleepers = self.state.sleepers.lock().unwrap();
|
||||
sleepers.remove(id);
|
||||
sleepers.remove(self.sleeping);
|
||||
|
||||
self.state
|
||||
.notified
|
||||
.swap(sleepers.is_notified(), Ordering::SeqCst);
|
||||
.store(sleepers.is_notified(), Ordering::Release);
|
||||
}
|
||||
self.sleeping = 0;
|
||||
}
|
||||
|
||||
/// Waits for the next runnable task to run.
|
||||
async fn runnable(&self) -> Runnable {
|
||||
async fn runnable(&mut self) -> Runnable {
|
||||
self.runnable_with(|| self.state.queue.pop().ok()).await
|
||||
}
|
||||
|
||||
/// Waits for the next runnable task to run, given a function that searches for a task.
|
||||
async fn runnable_with(&self, mut search: impl FnMut() -> Option<Runnable>) -> Runnable {
|
||||
async fn runnable_with(&mut self, mut search: impl FnMut() -> Option<Runnable>) -> Runnable {
|
||||
future::poll_fn(|cx| {
|
||||
loop {
|
||||
match search() {
|
||||
|
@ -695,14 +902,13 @@ impl Ticker<'_> {
|
|||
impl Drop for Ticker<'_> {
|
||||
fn drop(&mut self) {
|
||||
// If this ticker is in sleeping state, it must be removed from the sleepers list.
|
||||
let id = self.sleeping.swap(0, Ordering::SeqCst);
|
||||
if id != 0 {
|
||||
if self.sleeping != 0 {
|
||||
let mut sleepers = self.state.sleepers.lock().unwrap();
|
||||
let notified = sleepers.remove(id);
|
||||
let notified = sleepers.remove(self.sleeping);
|
||||
|
||||
self.state
|
||||
.notified
|
||||
.swap(sleepers.is_notified(), Ordering::SeqCst);
|
||||
.store(sleepers.is_notified(), Ordering::Release);
|
||||
|
||||
// If this ticker was notified, then notify another ticker.
|
||||
if notified {
|
||||
|
@ -727,7 +933,7 @@ struct Runner<'a> {
|
|||
local: Arc<ConcurrentQueue<Runnable>>,
|
||||
|
||||
/// Bumped every time a runnable task is found.
|
||||
ticks: AtomicUsize,
|
||||
ticks: usize,
|
||||
}
|
||||
|
||||
impl Runner<'_> {
|
||||
|
@ -737,7 +943,7 @@ impl Runner<'_> {
|
|||
state,
|
||||
ticker: Ticker::new(state),
|
||||
local: Arc::new(ConcurrentQueue::bounded(512)),
|
||||
ticks: AtomicUsize::new(0),
|
||||
ticks: 0,
|
||||
};
|
||||
state
|
||||
.local_queues
|
||||
|
@ -748,7 +954,7 @@ impl Runner<'_> {
|
|||
}
|
||||
|
||||
/// Waits for the next runnable task to run.
|
||||
async fn runnable(&self) -> Runnable {
|
||||
async fn runnable(&mut self, rng: &mut fastrand::Rng) -> Runnable {
|
||||
let runnable = self
|
||||
.ticker
|
||||
.runnable_with(|| {
|
||||
|
@ -768,7 +974,7 @@ impl Runner<'_> {
|
|||
|
||||
// Pick a random starting point in the iterator list and rotate the list.
|
||||
let n = local_queues.len();
|
||||
let start = fastrand::usize(..n);
|
||||
let start = rng.usize(..n);
|
||||
let iter = local_queues
|
||||
.iter()
|
||||
.chain(local_queues.iter())
|
||||
|
@ -791,9 +997,9 @@ impl Runner<'_> {
|
|||
.await;
|
||||
|
||||
// Bump the tick counter.
|
||||
let ticks = self.ticks.fetch_add(1, Ordering::SeqCst);
|
||||
self.ticks = self.ticks.wrapping_add(1);
|
||||
|
||||
if ticks % 64 == 0 {
|
||||
if self.ticks % 64 == 0 {
|
||||
// Steal tasks from the global queue to ensure fair task scheduling.
|
||||
steal(&self.state.queue, &self.local);
|
||||
}
|
||||
|
@ -843,21 +1049,24 @@ fn steal<T>(src: &ConcurrentQueue<T>, dest: &ConcurrentQueue<T>) {
|
|||
/// Debug implementation for `Executor` and `LocalExecutor`.
|
||||
fn debug_executor(executor: &Executor<'_>, name: &str, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
// Get a reference to the state.
|
||||
let state = match executor.state.get() {
|
||||
Some(state) => state,
|
||||
None => {
|
||||
// The executor has not been initialized.
|
||||
struct Uninitialized;
|
||||
let ptr = executor.state.load(Ordering::Acquire);
|
||||
if ptr.is_null() {
|
||||
// The executor has not been initialized.
|
||||
struct Uninitialized;
|
||||
|
||||
impl fmt::Debug for Uninitialized {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.write_str("<uninitialized>")
|
||||
}
|
||||
impl fmt::Debug for Uninitialized {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.write_str("<uninitialized>")
|
||||
}
|
||||
|
||||
return f.debug_tuple(name).field(&Uninitialized).finish();
|
||||
}
|
||||
};
|
||||
|
||||
return f.debug_tuple(name).field(&Uninitialized).finish();
|
||||
}
|
||||
|
||||
// SAFETY: If the state pointer is not null, it must have been
|
||||
// allocated properly by Arc::new and converted via Arc::into_raw
|
||||
// in state_ptr.
|
||||
let state = unsafe { &*ptr };
|
||||
|
||||
/// Debug wrapper for the number of active tasks.
|
||||
struct ActiveTasks<'a>(&'a Mutex<Slab<Waker>>);
|
||||
|
@ -910,10 +1119,48 @@ fn debug_executor(executor: &Executor<'_>, name: &str, f: &mut fmt::Formatter<'_
|
|||
}
|
||||
|
||||
/// Runs a closure when dropped.
|
||||
struct CallOnDrop<F: Fn()>(F);
|
||||
struct CallOnDrop<F: FnMut()>(F);
|
||||
|
||||
impl<F: Fn()> Drop for CallOnDrop<F> {
|
||||
impl<F: FnMut()> Drop for CallOnDrop<F> {
|
||||
fn drop(&mut self) {
|
||||
(self.0)();
|
||||
}
|
||||
}
|
||||
|
||||
fn _ensure_send_and_sync() {
|
||||
use futures_lite::future::pending;
|
||||
|
||||
fn is_send<T: Send>(_: T) {}
|
||||
fn is_sync<T: Sync>(_: T) {}
|
||||
fn is_static<T: 'static>(_: T) {}
|
||||
|
||||
is_send::<Executor<'_>>(Executor::new());
|
||||
is_sync::<Executor<'_>>(Executor::new());
|
||||
|
||||
let ex = Executor::new();
|
||||
is_send(ex.run(pending::<()>()));
|
||||
is_sync(ex.run(pending::<()>()));
|
||||
is_send(ex.tick());
|
||||
is_sync(ex.tick());
|
||||
is_send(ex.schedule());
|
||||
is_sync(ex.schedule());
|
||||
is_static(ex.schedule());
|
||||
|
||||
/// ```compile_fail
|
||||
/// use async_executor::LocalExecutor;
|
||||
/// use futures_lite::future::pending;
|
||||
///
|
||||
/// fn is_send<T: Send>(_: T) {}
|
||||
/// fn is_sync<T: Sync>(_: T) {}
|
||||
///
|
||||
/// is_send::<LocalExecutor<'_>>(LocalExecutor::new());
|
||||
/// is_sync::<LocalExecutor<'_>>(LocalExecutor::new());
|
||||
///
|
||||
/// let ex = LocalExecutor::new();
|
||||
/// is_send(ex.run(pending::<()>()));
|
||||
/// is_sync(ex.run(pending::<()>()));
|
||||
/// is_send(ex.tick());
|
||||
/// is_sync(ex.tick());
|
||||
/// ```
|
||||
fn _negative_test() {}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,34 @@
|
|||
use async_executor::LocalExecutor;
|
||||
use futures_lite::future::{block_on, pending, poll_once};
|
||||
use futures_lite::pin;
|
||||
use std::cell::Cell;
|
||||
|
||||
#[test]
|
||||
fn shared_queue_slot() {
|
||||
block_on(async {
|
||||
let was_polled = Cell::new(false);
|
||||
let future = async {
|
||||
was_polled.set(true);
|
||||
pending::<()>().await;
|
||||
};
|
||||
|
||||
let ex1 = LocalExecutor::new();
|
||||
let ex2 = LocalExecutor::new();
|
||||
|
||||
// Start the futures for running forever.
|
||||
let (run1, run2) = (ex1.run(pending::<()>()), ex2.run(pending::<()>()));
|
||||
pin!(run1);
|
||||
pin!(run2);
|
||||
assert!(poll_once(run1.as_mut()).await.is_none());
|
||||
assert!(poll_once(run2.as_mut()).await.is_none());
|
||||
|
||||
// Spawn the future on executor one and then poll executor two.
|
||||
ex1.spawn(future).detach();
|
||||
assert!(poll_once(run2).await.is_none());
|
||||
assert!(!was_polled.get());
|
||||
|
||||
// Poll the first one.
|
||||
assert!(poll_once(run1).await.is_none());
|
||||
assert!(was_polled.get());
|
||||
});
|
||||
}
|
|
@ -1,3 +1,4 @@
|
|||
#[cfg(not(miri))]
|
||||
use std::mem;
|
||||
use std::panic::catch_unwind;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
@ -38,6 +39,7 @@ fn executor_cancels_everything() {
|
|||
assert_eq!(DROP.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[cfg(not(miri))]
|
||||
#[test]
|
||||
fn leaked_executor_leaks_everything() {
|
||||
static DROP: AtomicUsize = AtomicUsize::new(0);
|
||||
|
@ -119,6 +121,20 @@ fn drop_finished_task_and_then_drop_executor() {
|
|||
assert_eq!(DROP.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn iterator_panics_mid_run() {
|
||||
let ex = Executor::new();
|
||||
|
||||
let panic = std::panic::catch_unwind(|| {
|
||||
let mut handles = vec![];
|
||||
ex.spawn_many(
|
||||
(0..50).map(|i| if i == 25 { panic!() } else { future::ready(i) }),
|
||||
&mut handles,
|
||||
)
|
||||
});
|
||||
assert!(panic.is_err());
|
||||
}
|
||||
|
||||
struct CallOnDrop<F: Fn()>(F);
|
||||
|
||||
impl<F: Fn()> Drop for CallOnDrop<F> {
|
||||
|
|
|
@ -0,0 +1,99 @@
|
|||
//! Test for larger tasks.
|
||||
|
||||
use async_executor::Executor;
|
||||
use futures_lite::future::{self, block_on};
|
||||
use futures_lite::prelude::*;
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
fn do_run<Fut: Future<Output = ()>>(mut f: impl FnMut(Arc<Executor<'static>>) -> Fut) {
|
||||
// This should not run for longer than two minutes.
|
||||
#[cfg(not(miri))]
|
||||
let _stop_timeout = {
|
||||
let (stop_timeout, stopper) = async_channel::bounded::<()>(1);
|
||||
thread::spawn(move || {
|
||||
block_on(async move {
|
||||
let timeout = async {
|
||||
async_io::Timer::after(Duration::from_secs(2 * 60)).await;
|
||||
eprintln!("test timed out after 2m");
|
||||
std::process::exit(1)
|
||||
};
|
||||
|
||||
let _ = stopper.recv().or(timeout).await;
|
||||
})
|
||||
});
|
||||
stop_timeout
|
||||
};
|
||||
|
||||
let ex = Arc::new(Executor::new());
|
||||
|
||||
// Test 1: Use the `run` command.
|
||||
block_on(ex.run(f(ex.clone())));
|
||||
|
||||
// Test 2: Loop on `tick`.
|
||||
block_on(async {
|
||||
let ticker = async {
|
||||
loop {
|
||||
ex.tick().await;
|
||||
}
|
||||
};
|
||||
|
||||
f(ex.clone()).or(ticker).await
|
||||
});
|
||||
|
||||
// Test 3: Run on many threads.
|
||||
thread::scope(|scope| {
|
||||
let (_signal, shutdown) = async_channel::bounded::<()>(1);
|
||||
|
||||
for _ in 0..16 {
|
||||
let shutdown = shutdown.clone();
|
||||
let ex = &ex;
|
||||
scope.spawn(move || block_on(ex.run(shutdown.recv())));
|
||||
}
|
||||
|
||||
block_on(f(ex.clone()));
|
||||
});
|
||||
|
||||
// Test 4: Tick loop on many threads.
|
||||
thread::scope(|scope| {
|
||||
let (_signal, shutdown) = async_channel::bounded::<()>(1);
|
||||
|
||||
for _ in 0..16 {
|
||||
let shutdown = shutdown.clone();
|
||||
let ex = &ex;
|
||||
scope.spawn(move || {
|
||||
block_on(async move {
|
||||
let ticker = async {
|
||||
loop {
|
||||
ex.tick().await;
|
||||
}
|
||||
};
|
||||
|
||||
shutdown.recv().or(ticker).await
|
||||
})
|
||||
});
|
||||
}
|
||||
|
||||
block_on(f(ex.clone()));
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn smoke() {
|
||||
do_run(|ex| async move { ex.spawn(async {}).await });
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn yield_now() {
|
||||
do_run(|ex| async move { ex.spawn(future::yield_now()).await })
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timer() {
|
||||
do_run(|ex| async move {
|
||||
ex.spawn(async_io::Timer::after(Duration::from_millis(5)))
|
||||
.await;
|
||||
})
|
||||
}
|
|
@ -0,0 +1,24 @@
|
|||
use async_executor::Executor;
|
||||
use futures_lite::{future, pin};
|
||||
|
||||
#[test]
|
||||
fn two_queues() {
|
||||
future::block_on(async {
|
||||
// Create an executor with two runners.
|
||||
let ex = Executor::new();
|
||||
let (run1, run2) = (
|
||||
ex.run(future::pending::<()>()),
|
||||
ex.run(future::pending::<()>()),
|
||||
);
|
||||
let mut run1 = Box::pin(run1);
|
||||
pin!(run2);
|
||||
|
||||
// Poll them both.
|
||||
assert!(future::poll_once(run1.as_mut()).await.is_none());
|
||||
assert!(future::poll_once(run2.as_mut()).await.is_none());
|
||||
|
||||
// Drop the first one, which should leave the local queue in the `None` state.
|
||||
drop(run1);
|
||||
assert!(future::poll_once(run2.as_mut()).await.is_none());
|
||||
});
|
||||
}
|
|
@ -0,0 +1,14 @@
|
|||
use async_executor::Executor;
|
||||
use futures_lite::{future, prelude::*};
|
||||
|
||||
#[test]
|
||||
fn test_panic_propagation() {
|
||||
let ex = Executor::new();
|
||||
let task = ex.spawn(async { panic!("should be caught by the task") });
|
||||
|
||||
// Running the executor should not panic.
|
||||
assert!(ex.try_tick());
|
||||
|
||||
// Polling the task should.
|
||||
assert!(future::block_on(task.catch_unwind()).is_err());
|
||||
}
|
|
@ -0,0 +1,45 @@
|
|||
use async_executor::{Executor, LocalExecutor};
|
||||
use futures_lite::future;
|
||||
|
||||
#[cfg(not(miri))]
|
||||
const READY_COUNT: usize = 50_000;
|
||||
#[cfg(miri)]
|
||||
const READY_COUNT: usize = 505;
|
||||
|
||||
#[test]
|
||||
fn spawn_many() {
|
||||
future::block_on(async {
|
||||
let ex = Executor::new();
|
||||
|
||||
// Spawn a lot of tasks.
|
||||
let mut tasks = vec![];
|
||||
ex.spawn_many((0..READY_COUNT).map(future::ready), &mut tasks);
|
||||
|
||||
// Run all of the tasks in parallel.
|
||||
ex.run(async move {
|
||||
for (i, task) in tasks.into_iter().enumerate() {
|
||||
assert_eq!(task.await, i);
|
||||
}
|
||||
})
|
||||
.await;
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn spawn_many_local() {
|
||||
future::block_on(async {
|
||||
let ex = LocalExecutor::new();
|
||||
|
||||
// Spawn a lot of tasks.
|
||||
let mut tasks = vec![];
|
||||
ex.spawn_many((0..READY_COUNT).map(future::ready), &mut tasks);
|
||||
|
||||
// Run all of the tasks in parallel.
|
||||
ex.run(async move {
|
||||
for (i, task) in tasks.into_iter().enumerate() {
|
||||
assert_eq!(task.await, i);
|
||||
}
|
||||
})
|
||||
.await;
|
||||
});
|
||||
}
|
Loading…
Reference in New Issue