Allow storing the intrusive linked-list pointers in an arbitrary
location in the node. This is in preparation for using the linked list
in the scheduler.
In order to make using the intrusive linked list more flexible, a trait
is introduced to abstract mapping an entry to raw pointers and the next
/ prev pointers. This also pushes more unsafety onto the user.
If an entry is inserted in the queue before the next deadline, the
DelayQueue needs to update the Delay tracking the next time to poll.
If there is an existing Delay, reset that rather than replacing it as if
it's already been polled the task will be waiting for a notification
before it will poll again, and dropping the Delay means that that
notification will never be performed.
This is a workaround for the fact that the doc(cfg) from outer cfg_*
macros doesn't get applied correctly. Its included in the rt-threaded
branch only, which is what is used for doc.rs via all-features.
The Incoming types currently don't take ownership of the listener, but
in most cases, users who want to use the Listener as a stream will only
want to use the stream from that point on. So, implement Stream directly
on the Listener types.
## Motivation
#2174
On epoll platforms, the read end of a pipe closing is signaled to the write end
through the `EPOLLERR` event [[1](http://man7.org/linux/man-pages/man2/epoll_ctl.2.html)]. If readiness is not registered for this
event, it will silently pass through `epoll_wait` calls.
Additionally, this specific case that `EPOLLERR` is triggered leaves the write
end of the pipe (parent process) waiting for a wakeup that never occurs.
## Solution
Similar to the `HUP` event on Unix platforms, errors are now always masked
through registrations so that both read and write ends of a connection are made
aware of errors.
In cases where pipes are used and the read end closes, write ends that are
waiting for a wakeup are properly notified and try to write again. This allows
a client to observe `BrokenPipe` and go through the proper cleanup and/or
restablishment of connection.
Closes#2174
Signed-off-by: Kevin Leimkuhler <[email protected]>
The link to the contributing guide in the tokio sub crate was
referencing a non-existent file. This updates the link to reference
the repo root's CONTRIBUTING.md file.
Fixes: #2266
Currently, the documentation for `tokio::sync::RwLock` states that it
has an unspecified priority policy dependent on the operating system.
This is incorrect: Tokio's `RwLock` is fairly queued. The incorrect
documentation appears to have been copied from the `std::sync::RwLock`
docs, for which this *is* the case.
This commit corrects the documentation to describe the actual priority
policy.
Signed-off-by: Eliza Weisman <[email protected]>
The added derived traits mirror the traits of std::time::Instant. As
tokio::time::Instant is just a wrapper around std::time::Instant, it
should also derive all the traits std::time::Instant derives.
See discussion in #2222. This wake/notify call has been there in one
form or another since the very early days of tokio. Currently though, it
is not clear that it is needed; the contract for polling is that you
must keep polling until you get `Pending`, so doing a wakeup when we are
about to return `Ready` is premature.
While the module documentation explains that a timed out future (as
created through the tokio::time::timeout function) is canceled, the
function's actual documentation doesn't mention that at all. This change
adds this relevant information to the documentation.
`StreamMap` is similar to `StreamExt::merge` in that it combines source
streams into a single merged stream that yields values in the order that
they arrive from the source streams. However, `StreamMap` has a lot more
flexibility in usage patterns.
`StreamMap` can:
- Merge an arbitrary number of streams.
- Track which source stream the value was received from.
- Handle inserting and removing streams from the set of managed streams
at any point during iteration.
All source streams held by `StreamMap` are indexed using a key. This key
is included with the value when a source stream yields a value. The key
is also used to remove the stream from the `StreamMap` before the stream
has completed streaming.
Because the `StreamMap` API moves streams during runtime, both streams
and keys must be `Unpin`. In order to insert a `!Unpin` stream into a
`StreamMap`, use `pin!` to pin the stream to the stack or `Box::pin` to
pin the stream in the heap.
This reduces the `mem::size_of::<watch::Receiver>()` from 4 words to 2.
- The `id` is now the pointer of the `Arc<WatchInner>`.
- The `ver` is moved into the `WatchInner`.
The previous implementation would perform a load that might be part of a
data race. The value read would be used only when race did not occur.
This would be well defined in a memory model where a load that is a part
of race merely returns an undefined value, the Rust memory model on the
other hand defines it to be undefined behaviour.
Perform read conditionally to avoid data race.
Covered by existing loom tests after changing casualty check to be
immediate rather than deferred.
Fixes: #2087
The `macros` feature flag was ommitted despite the fact that these
macros require the feature flag to function. The macros are now scoped
by the `macros` feature flag.
This is *not* a breaking change due to the fact that the macros were
broken without the `macros` feature flag in the first place.
* util: add futures-io/tokio::io compatibility layer
This PR adds a compatibility layer with conversions between the
`tokio::io` and `futures-io` versions of the `AsyncRead` and
`AsyncWrite` traits.
I initially opened this PR against `tokio-compat`, but we decided that
a compatibility layer for current versions of the `tokio` and
`futures-io` crates (rather than for compatibility with legacy code)
ought to go in `tokio-util` instead. See:
https://github.com/tokio-rs/tokio-compat/pull/2#issuecomment-551310953
This is based on code originally written by @Nemo157 as part of the
`futures-tokio-compat` crate, and is contributed on behalf of the
original author:
https://github.com/Nemo157/futures-tokio-compat/issues/2#issuecomment-544118866Closestokio-rs/tokio-compat#2
Co-authored-by: Wim Looman <[email protected]>
Signed-off-by: Eliza Weisman <[email protected]>
When using #[tokio::main] on a function with generics, the generics are
skipped. Simply using #vis #sig instead of #vis fn #name(#inputs) #ret
fixes the problem.
Fixes#2176
Provides a `try_join!` macro that supports concurrently driving multiple
`Result` futures on the same task and await the completion of all the
futures as `Ok` or the **first** `Err` future.
Adds `is_cancelled()` and `is_panic()` methods to `JoinError`, as well as
`into_panic()` and `try_into_panic()` methods which, when applicable, returns
the payload of the panic.
`parking_lot` provides synchronization primitives that tend to be
more efficient than the ones in `std`. However, depending on
`parking_lot` pulls in a number of dependencies resulting
in additional compilation time.
Adding *optional* support for `parking_lot` allows the end user
to opt-in when the trade offs make sense for their case.
Used for stack pinning and based on `pin_mut!` from the pin-util crate.
Pinning is used often when working with stream operators and the select!
macro. Given the small size of `pin!` it makes more sense to include a
version than re-export one from a separate crate or require the user to
depend on `pin-util` themselves.
Provides a `select!` macro for concurrently waiting on multiple async
expressions. The macro has similar goals and syntax as the one provided
by the `futures` crate, but differs significantly in implementation.
First, this implementation does not require special traits to be
implemented on futures or streams (i.e., no `FuseFuture`). A design goal
is to be able to pass a "plain" async fn result into the select! macro.
Even without `FuseFuture`, this `select!` implementation is able to
handle all cases the `futures::select!` macro can handle. It does this
by supporting pre-poll conditions on branches and result pattern
matching. For pre-conditions, each branch is able to include a condition
that disables the branch if it evaluates to false. This allows the user
to guard futures that have already been polled, preventing double
polling. Pattern matching can be used to disable streams that complete.
A second big difference is the macro is implemented almost entirely as a
declarative macro. The biggest advantage to using this strategy is that
the user will not need to alter the rustc recursion limit except in the
most extreme cases.
The resulting future also tends to be smaller in many cases.
Make sure the tail mutex is acquired when `condvar` is notified,
otherwise the wakeup may be lost and the sender could be left waiting.
Use `notify_all()` instead of `notify_one()` to ensure that the correct
sender is woken. Finally, only do any of this when there are no more
readers left.
Additionally, calling `send()` is buggy and may cause a panic when
the slot has another pending send.
The `&mut self` requirements for `TcpStream` methods ensure that there are at
most two tasks using the stream--one for reading and one for writing.
`TcpStream::split` allows two separate tasks to hold a reference to a single
`TcpStream`. `TcpStream::{peek,poll_peek}` only poll for read readiness, and
therefore are safe to use with a `ReadHalf`.
Instead of duplicating `TcpStream::poll_peek`, a private method is now used by
both `poll_peek` methods that uses the fact that only a `&TcpStream` is
required.
Closes#2136
The Tokio runtime provides a "shell" runtime when `rt-core` is not
available. This shell runtime is enough to support `#[tokio::main`] and
`#[tokio::test].
A previous change disabled these two attr macros when `rt-core` was not
selected. This patch fixes this by re-enabling the `main` and `test`
attr macros without `rt-core` and adds some integration tests to prevent
future regressions.
`AsyncRead` is safe to implement but can be implemented so that it
reports that it read more bytes than it actually did. `poll_read_buf` on
the other head implicitly trusts that the returned length is actually
correct which makes it possible to advance the buffer past what has
actually been initialized.
An alternative fix could be to avoid the panic and instead advance by
`n.min(b.len())`
Loom currently does not compile on windows due to a
transitive dependency on `generator`. The `generator`
crate builds have started to fail on windows CI. Loom
is not run under windows, however, so removing the
loom dependency on windows is sufficient to fix CI.
Refs: https://github.com/Xudong-Huang/generator-rs/issues/19
* Add a method to test if split streams come from the same stream.
The exposed stream ID can also be used as key in associative containers.
* Document the fact that split stream IDs can dangle.
* runtime: add Handle::try_current
Makes it possible to get a Handle only if a Runtime has been started, without panicing if that isn't the case
* Use an error instead
This PR introduces a new pattern for task-local storage. It allows for storage
and retrieval of data in an asynchronous context. It does so using a new pattern
based on past experience.
A quick example:
```rust
tokio::task_local! {
static FOO: u32;
}
FOO.scope(1, async move {
some_async_fn().await;
assert_eq!(FOO.get(), 1);
}).await;
```
## Background of task-local storage
The goal for task-local storage is to be able to provide some ambiant context in
an asynchronous context. One primary use case is for distributed tracing style
systems where a request identifier is made available during the context of a
request / response exchange. In a synchronous context, thread-local storage
would be used for this. However, with asynchronous Rust, logic is run in a
"task", which is decoupled from an underlying thread. A task may run on many
threads and many tasks may be multiplexed on a single thread. This hints at the
need for task-local storage.
### Early attempt
Futures 0.1 included a [task-local storage][01] strategy. This was based around
using the "runtime task" (more on this later) as the scope. When a task was
spawned with `tokio::spawn`, a task-local map would be created and assigned
with that task. Any task-local value that was stored would be stored in this
map. Whenever the runtime polled the task, it would set the task context
enabling access to find the value.
There are two main problems with this strategy which ultimetly lead to the
removal of runtime task-local storage:
1) In asynchronous Rust, a "task" is not a clear-cut thing.
2) The implementation did not leverage the significant optimizations that the
compiler provides for thread-local storage.
### What is a "task"?
With synchronous Rust, a "thread" is a clear concept: the construct you get with
`thread::spawn`. With asynchronous Rust, there is no strict definition of a
"task". A task is most commonly the construct you get when calling
`tokio::spawn`. The construct obtained with `tokio::spawn` will be referred to
as the "runtime task". However, it is also possible to multiplex asynchronous
logic within the context of a runtime task. APIs such as
[`task::LocalSet`][local-set] , [`FuturesUnordered`][futures-unordered],
[`select!`][select], and [`join!`][join] provide the ability to embed a mini
scheduler within a single runtime task.
Revisiting the primary use case, setting a request identifier for the duration
of a request response exchange, here is a scenario in which using the "runtime
task" as the scope for task-local storage would fail:
```rust
task_local!(static REQUEST_ID: Cell<u64> = Cell::new(0));
let request1 = get_request().await;
let request2 = get_request().await;
let (response1, response2) = join!{
async {
REQUEST_ID.with(|cell| cell.set(request1.identifier()));
process(request1)
},
async {
REQUEST_ID.with(|cell| cell.set(request2.identifier()));
process(request2)
},
};
```
`join!` multiplexes the execution of both branches on the same runtime task.
Given this, if `REQUEST_ID` is scoped by the runtime task, the request ID would
leak across the request / response exchange processing.
This is not a theoretical problem, but was hit repeatedly in practice. For
example, Hyper's HTTP/2.0 implementation multiplexes many request / response
exchanges on the same runtime task.
### Compiler thread-local optimizations
A second smaller problem with the original task-local storage strategy is that
it required re-implementing "thread-local storage" like constructs but without
being able to get the compiler to help optimize. A discussion of how the
compiler optimizes thread-local storage is out of scope for this PR description,
but suffice to say a task-local storage implementation should be able to
leverage thread-locals as much as possible.
## A new task-local strategy
Introduced in this PR is a new strategy for dealing with task-local storage.
Instead of using the runtime task as the thread-local scope, the proposed
task-local API allows the user to define any arbitrary scope. This solves the
problem of binding task-locals to the runtime task:
```rust
tokio::task_local!(static FOO: u32);
FOO.scope(1, async move {
some_async_fn().await;
assert_eq!(FOO.get(), 1);
}).await;
```
The `scope` function establishes a task-local scope for the `FOO` variable. It
takes a value to initialize `FOO` with and an async block. The `FOO` task-local
is then available for the duration of the provided block. `scope` returns a new
future that must then be awaited on.
`tokio::task_local` will define a new thread-local. The future returned from
`scope` will set this thread-local at the start of `poll` and unset it at the
end of `poll`. `FOO.get` is a simple thread-local access with no special logic.
This strategy solves both problems. Task-locals can be scoped at any level and
can leverage thread-local compiler optimizations.
Going back to the previous example:
```rust
task_local! {
static REQUEST_ID: u64;
}
let request1 = get_request().await;
let request2 = get_request().await;
let (response1, response2) = join!{
async {
let identifier = request1.identifier();
REQUEST_ID.scope(identifier, async {
process(request1).await
}).await
},
async {
let identifier = request2.identifier();
REQUEST_ID.scope(identifier, async {
process(request2).await
}).await
},
};
```
There is no longer a problem with request identifiers leaking.
## Disadvantages
The primary disadvantage of this strategy is that the "set and forget" pattern
with thread-locals is not possible.
```rust
thread_local! {
static FOO: Cell<usize> = Cell::new(0);
}
thread::spawn(|| {
FOO.with(|cell| cell.set(123));
do_work();
});
```
In this example, `FOO` is set at the start of the thread and automatically
cleared when the thread terminates. While this is nice in some cases, it only
really logically makes sense because the scope of a "thread" is clear (the
thread).
A similar pattern can be done with the proposed stratgy but would require an
explicit setting of the scope at the root of `tokio::spawn`. Additionally, one
should only do this if the runtime task is the appropriate scope for the
specific task-local variable.
Another disadvantage is that this new method does not support lazy initialization
but requires an explicit `LocalKey::scope` call to set the task-local value. In
this case since task-local's are different from thread-locals it is fine.
[01]: https://docs.rs/futures/0.1.29/futures/task/struct.LocalKey.html
[local-set]: #
[futures-unordered]: https://docs.rs/futures/0.3.1/futures/stream/struct.FuturesUnordered.html
[select]: https://docs.rs/futures/0.3.1/futures/macro.select.html
[join]: https://docs.rs/futures/0.3.1/futures/macro.join.html
Provides an asynchronous equivalent to `Iterator::collect()`. A sealed
`FromStream` trait is added. Stabilization is pending Rust supporting
`async` trait fns.
Provides an equivalent to stream `select()` from futures-rs. `merge`
best describes the operation (vs. `select`). `futures-rs` named the
operation "select" for historical reasons and did not rename it back to
`merge` in 0.3. The operation is most commonly named `merge` else where
as well (e.g. ReactiveX).
With the rt-threaded feature flag we create a threaded scheduler by
default. The documentation had a copy-and-paste error from the section
about the basic scheduler.
Previously acquire operations reading a value written by a successful
CAS in `drop_join_handle_fast` did not synchronize with it. The CAS
wasn't guaranteed to happen before the task deallocation, and so
created a data race between the two.
Use release success ordering to ensure synchronization.
Previously, when the threaded scheduler was in the shutdown process, it
would hold a lock while dropping in-flight tasks. If those tasks
included a drop handler that attempted to wake a second task, the wake
operation would attempt to acquire a lock held by the scheduler. This
results in a deadlock.
Dropping the lock before dropping tasks resolves the problem.
Fixes#2046
Previously, if an IO event was received during the runtime shutdown
process, it was possible to enter a deadlock. This was due to the
scheduler shutdown logic not expecting tasks to get scheduled once the
worker was in the shutdown process.
This patch fixes the deadlock by checking the queues for new tasks after
each call to park. If a new task is received, it is forcefully shutdown.
Fixes#2061
* io: Fix the Seek adapter and add a tested example.
If the first 'AsyncRead::start_seek' call returns Ready,
'AsyncRead::poll_complete' will be called.
Previously, a start_seek that immediately returned 'Ready' would cause
the Seek adapter to return 'Pending' without registering a Waker.
* fs: Do not return write errors from methods on AsyncSeek.
Write errors should only be returned on subsequent writes or on flush.
Also copy the last_write_err assert from 'poll_read' to both
'start_seek' and 'poll_complete' for consistency.
* Links are added where missing and examples are improved.
* Improve `stdin`, `stdout`, and `stderr` documentation by going
into more details regarding what can go wrong in concurrent
situations and provide examples for `stdout` and `stderr`.
Tweak context to remove more fns and usage of `Option`. Remove
`ThreadContext` struct as it is reduced to just `Handle`. Avoid passing
around individual driver handles and instead limit to the
`runtime::Handle` struct.
Currently, the only way to run a `tokio::task::LocalSet` is to call its
`block_on` method with a `&mut Runtime`, like
```rust
let mut rt = tokio::runtime::Runtime::new();
let local = tokio::task::LocalSet::new();
local.block_on(&mut rt, async {
// whatever...
});
```
Unfortunately, this means that `LocalSet` doesn't work with the
`#[tokio::main]` and `#[tokio::test]` macros, since the `main`
function is _already_ inside of a call to `block_on`.
**Solution**
This branch adds a `LocalSet::run` method, which takes a future and
returns a new future that runs that future on the `LocalSet`. This
is analogous to `LocalSet::block_on`, except that it can be called in
an async context.
Additionally, this branch implements `Future` for `LocalSet`. Awaiting
a `LocalSet` will run all spawned local futures until they complete.
This allows code like
```rust
#[tokio::main]
async fn main() {
let local = tokio::task::LocalSet::new();
local.spawn_local(async {
// ...
});
local.spawn_local(async {
// ...
tokio::task::spawn_local(...);
// ...
});
local.await;
}
```
The `LocalSet` docs have been updated to show the usage with
`#[tokio::main]` rather than with manually created runtimes, where
applicable.
Closes#1906Closes#1908Fixes#2057
Adds `Handle::current()` for accessing a handle to the runtime
associated with the current thread. This handle can then be
passed to other threads in order to spawn or perform other
runtime related tasks.
The `Waker::will_wake` compares both a data pointer and a vtable to
decide if wakers are equivalent. To avoid false negatives during
comparison, use the same vtable for a waker stored in `WakerRef`.
This patch improves the behavior of frozen time (a testing utility made
available with the `test-util` feature flag). Instead of of requiring
`time::advance` to be called in order to advance the value returned by
`Instant::now`, calls to `time::Driver::park_timeout` will use the
provided duration to advance the time.
This is the desired behavior as the timeout is used to indicate when the
next scheduled delay needs to be fired.
* tokio: remove documentation stating `Receiver` is clone-able.
The documentation for `broadcast` stated that both `Sender` and
`Receiver` are clonable. This isn't the case: `Receiver`s cannot be
cloned (and shouldn't be cloned).
In addition, mention that `Receiver` is `Sync`, and mention that both
`Receiver` and `Sender` are `Send`.
Fixes: #2032
* Clarify that Sender and Receiver are only Send and Sync if T is Send or Sync.
Provides a `RwLock` based on a semaphore. The semaphore is initialized
with 32 permits. A read acquires a single permit and a write acquires all 32
permits. This ensures that reads (up to 32) may happen concurrently and
writes happen exclusively.
Extend internal semaphore to support batch operations. With this PR,
consumers of the semaphore are able to atomically request more than one
permit. This is useful for implementing a RwLock.
Previously, thread-locals used by the various drivers were situated
with the driver code. This resulted in state being spread out and many
thread-locals being required to run a runtime.
This PR coalesces the thread-locals into a single struct.
Storing a `Runtime` value in a thread-local resulted in a panic due to
the inability to access the parker.
This fixes the bug by skipping parking if it fails. In general, there
isn't much that we can do besides not parking.
Fixes#593
Nested spawn_blocking calls would result in a panic due to the necessary
context not being setup. This patch sets the blocking pool context from
within a blocking pool.
Fixes#1982
`ToSocketAddrs` is a sealed trait pending changes in Rust that will allow
defining async trait fns. Until then, `net::lookup_host` is provided as a way
to convert a `T: ToSocketAddrs` into `SocketAddr`s.
Introduces `StreamExt` trait. This trait will be used to add utility functions
to make working with streams easier. This patch includes two functions:
* `next`: a future returning the item in the stream.
* `map`: transform each item in the stream.
`num_threads` is deprecated. Instead, `core_threads` and `max_threads` are
introduced. `core_threads` specifies the number of "always on" threads used
for the async task executor and `max_threads` specifies the maximum number
of threads that the runtime may spawn.
Adds a broadcast channel implementation. A broadcast channel is a
multi-producer, multi-consumer channel where each consumer receives a
clone of every value sent. This is useful for implementing pub / sub
style patterns.
Implemented as a ring buffer, a Vec of the specified capacity is
allocated on initialization of the channel. Values are pushed into
slots.
When the channel is full, a send overwrites the oldest value. Receivers
detect this and return an error on the next call to receive. This
prevents unbounded buffering and does not make the channel vulnerable to
the slowest consumer.
Closes: #1585
The blocking task queue was not explicitly drained as part of the
blocking pool shutdown logic. It was originally assumed that the
contents of the queue would be dropped when the blocking pool structure
is dropped. However, tasks must be explicitly shutdown, so we must drain
the queue can call `shutdown` on each task.
Fixes#1970, #1946
Calls to tasks should not be nested. Currently, while a task is being
executed and the runtime is shutting down, a call to wake() can result
in the wake target to be dropped. This, in turn, results in the drop
handler being called.
If the user holds a ref cell borrow, a mutex guard, or any such value,
dropping the task inline can result in a deadlock.
The fix is to permit tasks to be scheduled during the shutdown process
and dropping the tasks once they are popped from the queue.
Fixes#1929, #1886
Update the rotted thread_pool benchmarks. These benchmarks are not the
greatest, but as of now it is all we have for micro benchmarks.
Adds a little yielding in the parker as it helps a bit.
Currently, a `LocalSet` does not notify the `LocalFuture` again at the
end of a tick. This means that if we didn't poll every task in the run
queue during that tick (e.g. there are more than 61 tasks enqueued),
those tasks will not be polled.
This commit fixes this issue by changing `local::Scheduler::tick` to
return whether or not the local future needs to be notified again, and
waking the task if so.
Fixes#1899Fixes#1900
Signed-off-by: Eliza Weisman <[email protected]>
## Motivation
#1800 removed the lazy binding of `Delay`s to timers. With the removal of the
logic required for that, `HandlePriv` is no longer needed. This PR removes the
use of `HandlePriv`.
A `TODO` was also removed that would panic if when registering a new `Delay`
the current timer handle was full. That has been fixed to now immediately
transition that `Delay` to an error state that can be handled in a similar way
to other error states.
Signed-off-by: Kevin Leimkuhler <[email protected]>
## Motivation
There's currently an issue in `task::LocalSet` where dropping the local
set can result in an infinite loop if a task running in the local set is
notified from outside the local set (e.g. by a timer). This was reported
in issue #1885.
This issue exists because the `Drop` impl for `task::local::Scheduler`
does not drain the queue of tasks notified externally, the way the basic
scheduler does. Instead, only the local queue is drained, leaving some
tasks in place. Since these tasks are never removed, the loop that
continues trying to cancel tasks until the owned task list is totally
empty continues infinitely.
I think this issue was due to the `Drop` impl being written before a
remote queue was added to the local scheduler, and the need to close the
remote queue as well was overlooked.
## Solution
This branch solves the problem by clearing the local scheduler's remote
queue as well as the local one.
I've added a test that reproduces the behavior. The test fails on master
and passes after this change.
In addition, this branch factors out the common task queue logic in the
basic scheduler runtime and the `LocalSet` struct in `tokio::task`. This
is because as more work was done on the `LocalSet`, it has gotten closer
and closer to the basic scheduler in behavior, and factoring out the
shared code reduces the risk of errors caused by `LocalSet` not doing
something that the basic scheduler does. The queues are now encapsulated
by a `MpscQueues` struct in `tokio::task::queue` (crate-public). As a
follow-up, I'd also like to look into changing this type to use the same
remote queue type as the threadpool (a linked list).
In particular, I noticed the basic scheduler has a flag that indicates
the remote queue has been closed, which is set when dropping the
scheduler. This prevents tasks from being added after the scheduler has
started shutting down, stopping a potential task leak. Rather than
duplicating this code in `LocalSet`, I thought it was probably better to
factor it out into a shared type.
There are a few cases where there are small differences in behavior,
though, so there is still a need for separate types implemented _using_
the new `MpscQueues` struct. However, it should cover most of the
identical code.
Note that this diff is rather large, due to the refactoring. However, the
actual fix for the infinite loop is very simple. It can be reviewed on its own
by looking at commit 4f46ac6. The refactor is in a separate commit, with
the SHA 90b5b1f.
Fixes#1885
Signed-off-by: Eliza Weisman <[email protected]>
## Motivation
Currently, `tokio::task::LocalSet`'s `block_on` method requires the
future to live for the 'static lifetime. However, this bound is not
required — the future is wrapped in a `LocalFuture`, and then passed
into `Runtime::block_on`, which does _not_ require a `'static` future.
This came up while updating `tokio-compat` to work with version 0.2. To
mimic the behavior of `tokio` 0.1's `current_thread::Runtime::run`, we
want to be able to have a runtime block on the `recv` future from an
mpsc channel indicating when the runtime is idle. To support `!Send`
futures, as the old `current_thread::Runtime` did, we must do so inside
of a `LocalSet`. However, with the current bounds, we cannot await an
`mpsc::Receiver`'s `recv` future inside the `LocalSet::block_on` call.
## Solution
This branch removes the unnecessary `'static` bound.
Signed-off-by: Eliza Weisman <[email protected]>
Adds `read_buf` and `write_buf` which work with `T: BufMut` and `T: Buf`
respectively. This adds an easy API for using the buffer traits provided
by `bytes.
`oneshot::Receiver::try_recv` does not provide any information as to the
reason **why** receiving failed. The two cases are that the channel is
empty or that the channel closed.
`TryRecvError` is changed to be an enum of those two cases. This is
backwards compatible as `TryRecvError` was an opaque struct.
This also expands on `oneshot` API documentation, adding details and
examples.
Closes#1872
Provide convenience methods for encoding and decoding big-endian numbers
on top of asynchronous I/O streams. Only primitive types are provided
(24 and 48 bit numbers are omitted).
In general, using these methods won't be the fastest way to do
encoding/decoding with asynchronous byte streams, but they help to get
simple things working fast.
The "global executor" thread-local is to track where to spawn new tasks,
**not** which scheduler is active on the current thread. This fixes a
bug with scheduling tasks on the basic_scheduler by tracking the
currently active basic_scheduler with a dedicated thread-local variable.
Fixes: #1851
In version 0.1 there was File::into_std method that destructured
tokio_fs::File into std::fs:File. That method was lacking in
version 0.2.
Fixes: #1852
It turns out that the `Scheduler::release` method on `LocalSet`'s
`Scheduler` *is* called, when the `Scheduler` is dropped with tasks
still running. Currently, that method is `unreachable!`, which means
that dropping a `LocalSet` with tasks running will panic.
This commit fixes the panic, by pushing released tasks to
`pending_drop`. This is the same as `BasicScheduler`.
Fixes#1842
This fixes the API docs for both `TcpListener::incoming` and
`UnixListener::incoming`. The function now takes `&mut self` instead of
`self`. Adds an example for both function.
## Motivation
In earlier versions of `tokio`, the `current_thread::Runtime` type could
be used to run `!Send` futures. However, PR #1716 merged the
current-thread and threadpool runtimes into a single type, which can no
longer run `!Send` futures. There is still a need in some cases to
support futures that don't implement `Send`, and the `tokio-compat`
crate requires this in order to provide APIs that existed in `tokio`
0.1.
## Solution
This branch implements the API described by @carllerche in
https://github.com/tokio-rs/tokio/pull/1716#issuecomment-549496309. It
adds a new `LocalSet` type and `spawn_local` function to `tokio::task`.
The `LocalSet` type is used to group together a set of tasks which must
run on the same thread and don't implement `Send`. These are available
when a new "rt-util" feature flag is enabled.
Currently, the local task set is run by passing it a reference to a
`Runtime` and a future to `block_on`. In the future, we may also want
to investigate allowing spawned futures to construct their own local
task sets, which would be executed on the worker that the future is
executing on.
In order to implement the new API, I've made some internal changes to
the `task` module and `Schedule` trait to support scheduling both `Send`
and `!Send` futures.
Signed-off-by: Eliza Weisman <[email protected]>
This provides the ability to get the raw OS handle for a `File`. The
`Into*` variant cannot be provided as `File` needs to maintain ownership
of the `File`. The actual handle may have been moved to a background
thread.
This updates the tokio `Command` and `Child` behavior to match that of
the stdlib: spawned processes will *not* be automatically killed when
the handle is dropped
Unlike the stdlib, any dropped (unix) processes may be reaped by tokio
behind-the-scenes after they exit and if new processes are awaited,
which mitigates the risks of piling up unreaped zombie unix processes
A `Command::kill_on_drop` method is added to allow the caller to
control whether the spawned child should be killed when the handle is
dropped. By default, this value is `false`.
The `Child::forget` method has been removed, as it is superseded by
`Command::kill_on_drop`
Changes the set of `default` feature flags to `[]`. By default, only
core traits are included without specifying feature flags. This makes it
easier for users to pick the components they need.
For convenience, a `full` feature flag is included that includes all
components.
Tests are configured to require the `full` feature. Testing individual
feature flags will need to be moved to a separate crate.
Closes#1791
This directory was deleted when `cargo hack` was introduced, however
there were some tests that were still useful (macro failure output).
Also, additional build tests will be added over time.
Annotates types in `tokio::io` module with their required feature flag.
This annotation is included in generated documentation.
Notes:
* The annotation must be on the type or function itself. Annotating just
the re-export is not sufficient.
* The annotation must be **inside** the `pin_project!` macro or it is
lost.
* runtime: cleanup and add config options
This patch finishes the cleanup as part of the transition to Tokio 0.2.
A number of changes were made to take advantage of having all Tokio
types in a single crate. Also, fixes using Tokio types from
`spawn_blocking`.
* Many threads, one resource driver
Previously, in the threaded scheduler, a resource driver (mio::Poll /
timer combo) was created per thread. This was more or less fine, except
it required balancing across the available drivers. When using a
resource driver from **outside** of the thread pool, balancing is
tricky. The change was original done to avoid having a dedicated driver
thread.
Now, instead of creating many resource drivers, a single resource driver
is used. Each scheduler thread will attempt to "lock" the resource
driver before parking on it. If the resource driver is already locked,
the thread uses a condition variable to park. Contention should remain
low as, under load, the scheduler avoids using the drivers.
* Add configuration options to enable I/O / time
New configuration options are added to `runtime::Builder` to allow
enabling I/O and time drivers on a runtime instance basis. This is
useful when wanting to create lightweight runtime instances to execute
compute only tasks.
* Bug fixes
The condition variable parker is updated to the same algorithm used in
`std`. This is motivated by some potential deadlock cases discovered by
`loom`.
The basic scheduler is fixed to fairly schedule tasks. `push_front` was
accidentally used instead of `push_back`.
I/O, time, and spawning now work from within `spawn_blocking` closures.
* Misc cleanup
The threaded scheduler is no longer generic over `P :Park`. Instead, it
is hard coded to a specific parker. Tests, including loom tests, are
updated to use `Runtime` directly. This provides greater coverage.
The `blocking` module is moved back into `runtime` as all usage is
within `runtime` itself.
## Motivation
Tokio's crate-level docs are currently pretty sparse, and in some cases
reference old names for APIs. Before 0.2 is released, they could use a
fresh coat of paint.
## Solution
This branch reworks and expands the `lib.rs` docs. In particular, I've
added a new "A Tour of Tokio" section, inspired by the [standard
library's similarly-named section][std]. This section lists all of
`tokio`'s public modules, and summarizes their major APIs. It also lists
the feature flags necessary to enable those APIs.
[std]: https://doc.rust-lang.org/std/index.html#a-tour-of-the-rust-standard-library
Signed-off-by: Eliza Weisman <[email protected]>
## Motivation
The `tokio::runtime` module's docs need to be updated to
track recent changes.
## Solution
This branch updates and expands the `runtime` docs.
Signed-off-by: Eliza Weisman <[email protected]>
## Motivation
The `tokio::io` module's docs are fairly sparse and not particularly up
to date. They ought to be improved before release.
## Solution
This branch adds new module-level docs to `tokio::io`. The new docs are
largely inspired by `std::io`'s documentation, and highlight the
similarities and differences between `tokio::io` and `std::io`.
Signed-off-by: Eliza Weisman <[email protected]>
## Motivation
The new `tokio::task` module is pretty lacking in API docs.
## Solution
This branch adds new API docs to the `task` module, including:
* Module-level docs with a summary of the differences between
tasks and threads
* Examples of how to use the `task` APIs in the module-level docs
* More docs for `yield_now`
* More docs and examples for `JoinHandle`, based on the
`std::thread::JoinHandle` API docs.
This branch contains commits cherry-picked from #1794
Signed-off-by: Eliza Weisman <[email protected]>
## Motivation
Similar to #1666, it is no longer necessary to lazily register delays with the
executions default timer. All delays are expected to be created from within a
runtime, and should panic if not done so.
## Solution
`tokio::time` now assumes there to be a `CURRENT_TIMER` set when creating a
delay; this can be assumed if called within a tokio runtime. If there is no
current timer, the application will panic with a "no current timer" message.
## Follow-up
Similar to #1666, `HandlePriv` can probably be removed, but this mainly prepares
for 0.2 API changes. Because it is not in the public API, this can be done in a
following change.
Signed-off-by: Kevin Leimkuhler <[email protected]>
The I/O driver is made private and moved to `tokio::io::driver`. `Registration` is
moved to `tokio::io::Registration` and `PollEvented` is moved to `tokio::io::PollEvented`.
Additionally, the concurrent slab used by the I/O driver is cleaned up and extracted to
`tokio::util::slab`, allowing it to eventually be used by other types.
When checking if a worker needs to be unparked, the SeqCst load does not
provide the necessary synchronization to ensure the scheduled task is
visible to the searching worker. The `load` is switched to
`fetch_add(0)` which does establish the necessary synchronization.
Adding unit tests catching this bug will require a fix to loom and will
be done at a later time. The bug fix has been validated with manual
testing.
Fixes#1768
Removes dependencies between Tokio feature flags. For example, `process`
should not depend on `sync` simply because it uses the `mpsc` channel.
Instead, feature flags represent **public** APIs that become available
with the feature enabled. When the feature is not enabled, the
functionality is removed. If another Tokio component requires the
functionality, it is stays as `pub(crate)`.
The threaded scheduler is now exposed under `rt-threaded`. This feature
flag only enables the threaded scheduler and does not include I/O,
networking, or time. Those features must be explictly enabled.
A `full` feature flag is added that enables all features.
`stdin`, `stdout`, `stderr` are exposed under `io-std`.
Macros are used to scope code by feature flag.
There are limitations with `async/await` (no GAT) requiring the value to
be cloned on receive. The `poll` based API is not currently exposed.
This makes the `Clone` requirement explicit.
This exposes `std{in, out, err}` under io feature by moving
`fs::blocking` module into `io::blocking`.
As `fs` feature depends on `io-trait` feature, `fs` implementations can
always access `io` module.
It no longer supports executing !Send futures. The use case for
It is wanting a “light” runtime. There will be “local” task execution
using a different strategy coming later.
This patch also renames `thread_pool` -> `threaded_scheduler`, but
only in public APIs for now.
In an effort to reach API stability, the `tokio` crate is shedding its
_public_ dependencies on crates that are either a) do not provide a
stable (1.0+) release with longevity guarantees or b) match the `tokio`
release cadence. Of course, implementing `std` traits fits the
requirements.
The on exception, for now, is the `Stream` trait found in `futures_core`.
It is expected that this trait will not change much and be moved into `std.
Since Tokio is not yet going reaching 1.0, I feel that it is acceptable to maintain
a dependency on this trait given how foundational it is.
Since the `Stream` implementation is optional, types that are logically
streams provide `async fn next_*` functions to obtain the next value.
Avoiding the `next()` name prevents fn conflicts with `StreamExt::next()`.
Additionally, some misc cleanup is also done:
- `tokio::io::io` -> `tokio::io::util`.
- `delay` -> `delay_until`.
- `Timeout::new` -> `timeout(...)`.
- `signal::ctrl_c()` returns a future instead of a stream.
- `{tcp,unix}::Incoming` is removed (due to lack of `Stream` trait).
- `time::Throttle` is removed (due to lack of `Stream` trait).
- Fix: `mpsc::UnboundedSender::send(&self)` (no more conflict with `Sink` fns).
This patch started as an effort to make `time::Timer` private. However, in an
effort to get the build compiling again, more and more changes were made. This
probably should have been broken up, but here we are. I will attempt to
summarize the changes here.
* Feature flags are reorganized to make clearer. `net-driver` becomes
`io-driver`. `rt-current-thread` becomes `rt-core`.
* The `Runtime` can be created without any executor. This replaces `enter`. It
also allows creating I/O / time drivers that are standalone.
* `tokio::timer` is renamed to `tokio::time`. This brings it in line with `std`.
* `tokio::timer::Timer` is renamed to `Driver` and made private.
* The `clock` module is removed. Instead, an `Instant` type is provided. This
type defaults to calling `std::time::Instant`. A `test-util` feature flag can
be used to enable hooking into time.
* The `blocking` module is moved to the top level and is cleaned up.
* The `task` module is moved to the top level.
* The thread-pool's in-place blocking implementation is cleaned up.
* `runtime::Spawner` is renamed to `runtime::Handle` and can be used to "enter"
a runtime context.
It is a more common use case that is to simply cause a delay for an amount of time.
I think it is more appropriate to show off `delay_for` in the example rather than `delay` that is useful only for less common use cases.
Fixes a bug in the thread-pool executor related to shutdown
concurrent with a task that is self-notifying. A `loom` test is
added to validate the fix.
Additionally, in anticipation of the `thread_pool` module being
switched to private, tests are updated to use `Runtime` directly
instead of `thread_pool`. Those tests that cannot be updated
are switched to unit tests.
The `Executor` trait is sub-optimal as it forces a `Box<dyn Future>` to
spawn. Instead, `tokio::spawn` delegates to the specific runtime
implementation set for the current execution context.
`TypedExecutor`, while useful, has seen limited adoption. As such, it is
removed from `tokio` proper. Moving it to `tokio-util` is a possibility
that can be explored as follow up work.
Simplify Tokio's runtime construct by combining both Runtime variants
into a single type. The execution style can be controlled by a
configuration setting on `Builder`.
The implication of this change is that there is no longer any way to
spawn `!Send` futures. This, however, is a temporary limitation. A
different strategy will be employed for supporting `!Send` futures.
Included in this patch is a rework of `task::JoinHandle` to support
using this type from both the thread-pool and current-thread executors.
The compat crate is moved to https://github.com/tokio-rs/tokio-compat.
This allows pinning it to specific revisions of the Tokio git
repository. The master branch is intended to go through significant
churn and it will be easier to update the compat layer in batches.
## Motivation
The `futures` crate's [`compat` module][futures-compat] provides
interoperability between `futures` 0.1 and `std::future` _future types_
(e.g. implementing `std::future::Future` for a type that implements the
`futures` 0.1 `Future` trait). However, this on its own is insufficient
to run code written against `tokio` 0.1 on a `tokio` 0.2 runtime, if
that code also relies on `tokio`'s runtime services. If legacy tasks are
executed that rely on `tokio::timer`, perform IO using `tokio`'s
reactor, or call `tokio::spawn`, those API calls will fail unless there
is also a runtime compatibility layer.
## Solution
As proposed in #1549, this branch introduces a new `tokio-compat` crate,
with implementations of the thread pool and current-thread runtimes that
are capable of running both tokio 0.1 and tokio 0.2 tasks. The compat
runtime creates a background thread that runs a `tokio` 0.1 timer and
reactor, and sets itself as the `tokio` 0.1 executor as well as the
default 0.2 executor. This allows 0.1 futures that use 0.1 timer,
reactor, and executor APIs may run alongside `std::future` tasks on the
0.2 runtime.
### Examples
Spawning both `tokio` 0.1 and `tokio` 0.2 futures:
```rust
use futures_01::future::lazy;
tokio_compat::run(lazy(|| {
// spawn a `futures` 0.1 future using the `spawn` function from the
// `tokio` 0.1 crate:
tokio_01::spawn(lazy(|| {
println!("hello from tokio 0.1!");
Ok(())
}));
// spawn an `async` block future on the same runtime using `tokio`
// 0.2's `spawn`:
tokio_02::spawn(async {
println!("hello from tokio 0.2!");
});
Ok(())
}))
```
Futures on the compat runtime can use `timer` APIs from both 0.1 and 0.2
versions of `tokio`:
```rust
use std::time::{Duration, Instant};
use futures_01::future::lazy;
use tokio_compat::prelude::*;
tokio_compat::run_03(async {
// Wait for a `tokio` 0.1 `Delay`...
let when = Instant::now() + Duration::from_millis(10);
tokio_01::timer::Delay::new(when)
// convert the delay future into a `std::future` that we can `await`.
.compat()
.await
.expect("tokio 0.1 timer should work!");
println!("10 ms have elapsed");
// Wait for a `tokio` 0.2 `Delay`...
let when = Instant::now() + Duration::from_millis(20);
tokio_02::timer::delay(when).await;
println!("20 ms have elapsed");
});
```
## Future Work
This is just an initial implementation of a `tokio-compat` crate; there
are more compatibility layers we'll want to provide before that crate is
complete. For example, we should also provide compatibility between
`tokio` 0.2's `AsyncRead` and `AsyncWrite` traits and the `futures` 0.1
and `futures` 0.3 versions of those traits. In #1549, @carllerche also
suggests that the `compat` crate provide reimplementations of APIs that
were removed from `tokio` 0.2 proper, such as the `tcp::Incoming`
future.
Additionally, there is likely extra work required to get the
`tokio-threadpool` 0.1 `blocking` APIs to work on the compat runtime.
This will be addressed in a follow-up PR.
Fixes: #1605Fixes: #1552
Refs: #1549
[futures-compat]: https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures/compat/index.html
It is hard to maintain features list manually, so use cargo-hack's
`--each-feature` flag. And cargo-hack provides a workaround for an issue
that dev-dependencies leaking into normal build (`--no-dev-deps` flag),
so removed own ci tool.
Also, compared to running tests on all features, there is not much
advantage in running tests on each feature, so only the default features
and all features are tested.
If the behavior changes depending on the feature, we need to test it as
another job in CI.
I need this to implement SMTP pipelining checks. I mostly need to
flush my send buffer when the read buffer is empty before waiting for
the next command.
A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).
The sync implementation is now provided by the main `tokio` crate.
Functionality can be opted out of by using the various net related
feature flags.
A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).
The executor implementation is now provided by the main `tokio` crate.
Functionality can be opted out of by using the various net related
feature flags.
## Motivation
The `tokio_net::driver` module currently stores the state associated
with scheduled IO resources in a `Slab` implementation from the `slab`
crate. Because inserting items into and removing items from `slab::Slab`
requires mutable access, the slab must be placed within a `RwLock`. This
has the potential to be a performance bottleneck especially in the context of
the work-stealing scheduler where tasks and the reactor are often located on
the same thread.
`tokio-net` currently reimplements the `ShardedRwLock` type from
`crossbeam` on top of `parking_lot`'s `RwLock` in an attempt to squeeze
as much performance as possible out of the read-write lock around the
slab. This introduces several dependencies that are not used elsewhere.
## Solution
This branch replaces the `RwLock<Slab>` with a lock-free sharded slab
implementation.
The sharded slab is based on the concept of _free list sharding_
described by Leijen, Zorn, and de Moura in [_Mimalloc: Free List
Sharding in Action_][mimalloc], which describes the implementation of a
concurrent memory allocator. In this approach, the slab is sharded so
that each thread has its own thread-local list of slab _pages_. Objects
are always inserted into the local slab of the thread where the
insertion is performed. Therefore, the insert operation needs not be
synchronized.
However, since objects can be _removed_ from the slab by threads other
than the one on which they were inserted, removal operations can still
occur concurrently. Therefore, Leijen et al. introduce a concept of
_local_ and _global_ free lists. When an object is removed on the same
thread it was originally inserted on, it is placed on the local free
list; if it is removed on another thread, it goes on the global free
list for the heap of the thread from which it originated. To find a free
slot to insert into, the local free list is used first; if it is empty,
the entire global free list is popped onto the local free list. Since
the local free list is only ever accessed by the thread it belongs to,
it does not require synchronization at all, and because the global free
list is popped from infrequently, the cost of synchronization has a
reduced impact. A majority of insertions can occur without any
synchronization at all; and removals only require synchronization when
an object has left its parent thread.
The sharded slab was initially implemented in a separate crate (soon to
be released), vendored in-tree to decrease `tokio-net`'s dependencies.
Some code from the original implementation was removed or simplified,
since it is only necessary to support `tokio-net`'s use case, rather
than to provide a fully generic implementation.
[mimalloc]: https://www.microsoft.com/en-us/research/uploads/prod/2019/06/mimalloc-tr-v1.pdf
## Performance
These graphs were produced by out-of-tree `criterion` benchmarks of the
sharded slab implementation.
The first shows the results of a benchmark where an increasing number of
items are inserted and then removed into a slab concurrently by five
threads. It compares the performance of the sharded slab implementation
with a `RwLock<slab::Slab>`:
<img width="1124" alt="Screen Shot 2019-10-01 at 5 09 49 PM" src="https://user-images.githubusercontent.com/2796466/66078398-cd6c9f80-e516-11e9-9923-0ed6292e8498.png">
The second graph shows the results of a benchmark where an increasing
number of items are inserted and then removed by a _single_ thread. It
compares the performance of the sharded slab implementation with an
`RwLock<slab::Slab>` and a `mut slab::Slab`.
<img width="925" alt="Screen Shot 2019-10-01 at 5 13 45 PM" src="https://user-images.githubusercontent.com/2796466/66078469-f0974f00-e516-11e9-95b5-f65f0aa7e494.png">
Note that while the `mut slab::Slab` (i.e. no read-write lock) is
(unsurprisingly) faster than the sharded slab in the single-threaded
benchmark, the sharded slab outperforms the un-contended
`RwLock<slab::Slab>`. This case, where the lock is uncontended and only
accessed from a single thread, represents the best case for the current
use of `slab` in `tokio-net`, since the lock cannot be conditionally
removed in the single-threaded case.
These benchmarks demonstrate that, while the sharded approach introduces
a small constant-factor overhead, it offers significantly better
performance across concurrent accesses.
## Notes
This branch removes the following dependencies `tokio-net`:
- `parking_lot`
- `num_cpus`
- `crossbeam_util`
- `slab`
This branch adds the following dev-dependencies:
- `proptest`
- `loom`
Note that these dev dependencies were used to implement tests for the
sharded-slab crate out-of-tree, and were necessary in order to vendor
the existing tests. Alternatively, since the implementation is tested
externally, we _could_ remove these tests in order to avoid picking up
dev-dependencies. However, this means that we should try to ensure that
`tokio-net`'s vendored implementation doesn't diverge significantly from
upstream's, since it would be missing a majority of its tests.
Signed-off-by: Eliza Weisman <[email protected]>
This adds an extra spawned task during the thread-pool shutdown loom
test. This results in additional cases being tested, primarily tasks
being stolen.
A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).
The `io` implementation is now provided by the main `tokio` crate.
Functionality can be opted out of by using the various net related
feature flags.
A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).
The `net` implementation is now provided by the main `tokio` crate.
Functionality can be opted out of by using the various net related
feature flags.
Previously, support for `blocking` was done through a static `POOL` that
would spawn threads on demand. While this made the pool accessible at
all times, it made it hard to configure, and it was impossible to keep
multiple blocking pools.
This patch changes `blocking` to instead use a "default" global like the
ones used for timers, executors, and the like. There is now
`blocking::with_pool`, which is used by both thread-pool workers and the
current-thread runtime to ensure that a pool is available to tasks.
This patch also changes `ThreadPool` to spawn its worker threads on the
blocking pool rather than as free-standing threads. This is in
preparation for the coming in-place blocking work.
One downside of this change is that thread names are no longer
"semantic". All threads are named by the pool name, and individual
threads are not (currently) given names with numerical suffixes like
before.
Historically, logging has been added haphazardly. Here, we entirely
remove logging as none of it is particularly useful. In the future, we
will add tracing back in order to expose useful data to the user of
Tokio.
Related to #1318, Tokio APIs that are "less stable" are moved into a new
`tokio-util` crate. This crate will mirror `tokio` and provide
additional APIs that may require a greater rate of breaking changes.
As examples require `tokio-util`, they are moved into a separate
crate (`examples`). This has the added advantage of being able to avoid
example only dependencies in the `tokio` crate.
A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).
The `timer` implementation is now provided by the main `tokio` crate.
The `timer` functionality may still be excluded from the build by
skipping the `timer` feature flag.
## Motivation
The `tokio_net` resources can be created outside of a runtime due to how tokio
has been used with futures to date. For example, this allows a `TcpStream` to be
created, and later passed into a runtime:
```
let stream = TcpStream::connect(...).and_then(|socket| {
// do something
});
tokio::run(stream);
```
In order to support this functionality, the reactor was lazily bound to the
resource on the first call to `poll_read_ready`/`poll_write_ready`. This
required a lot of additional complexity in the binding logic to support.
With the tokio 0.2 common case, this is no longer necessary and can be removed.
All resources are expected to be created from within a runtime, and should panic
if not done so.
Closes#1168
## Solution
The `tokio_net` crate now assumes there to be a `CURRENT_REACTOR` set on the
worker thread creating a resource; this can be assumed if called within a tokio
runtime. If there is no current reactor, the application will panic with a "no
current reactor" message.
With this assumption, all the unsafe and atomics have been removed from
`tokio_net::driver::Registration` as it is no longer needed.
There is no longer any reason to pass in handles to the family of `from_std` methods on `net` resources. `Handle::current` has therefore a more restricted private use where it is only used in `driver::Registration::new`.
Signed-off-by: Kevin Leimkuhler <[email protected]>
A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).
The `fs` implementation is now provided by the main `tokio` crate. The
`fs` functionality may still be excluded from the build by skipping the
`fs` feature flag.
This patch is a ground up rewrite of the existing work-stealing thread
pool. The goal is to reduce overhead while simplifying code when
possible.
At a high level, the following architectural changes were made:
- The local run queues were switched for bounded circle buffer queues.
- Reduce cross-thread synchronization.
- Refactor task constructs to use a single allocation and always include
a join handle (#887).
- Simplify logic around putting workers to sleep and waking them up.
**Local run queues**
Move away from crossbeam's implementation of the Chase-Lev deque. This
implementation included unnecessary overhead as it supported
capabilities that are not needed for the work-stealing thread pool.
Instead, a fixed size circle buffer is used for the local queue. When
the local queue is full, half of the tasks contained in it are moved to
the global run queue.
**Reduce cross-thread synchronization**
This is done via many small improvements. Primarily, an upper bound is
placed on the number of concurrent stealers. Limiting the number of
stealers results in lower contention. Secondly, the rate at which
workers are notified and woken up is throttled. This also reduces
contention by preventing many threads from racing to steal work.
**Refactor task structure**
Now that Tokio is able to target a rust version that supports
`std::alloc` as well as `std::task`, the pool is able to optimize how
the task structure is laid out. Now, a single allocation per task is
required and a join handle is always provided enabling the spawner to
retrieve the result of the task (#887).
**Simplifying logic**
When possible, complexity is reduced in the implementation. This is done
by using locks and other simpler constructs in cold paths. The set of
sleeping workers is now represented as a `Mutex<VecDeque<usize>>`.
Instead of optimizing access to this structure, we reduce the amount the
pool must access this structure.
Secondly, we have (temporarily) removed `threadpool::blocking`. This
capability will come back later, but the original implementation was way
more complicated than necessary.
**Results**
The thread pool benchmarks have improved significantly:
Old thread pool:
```
test chained_spawn ... bench: 2,019,796 ns/iter (+/- 302,168)
test ping_pong ... bench: 1,279,948 ns/iter (+/- 154,365)
test spawn_many ... bench: 10,283,608 ns/iter (+/- 1,284,275)
test yield_many ... bench: 21,450,748 ns/iter (+/- 1,201,337)
```
New thread pool:
```
test chained_spawn ... bench: 147,943 ns/iter (+/- 6,673)
test ping_pong ... bench: 537,744 ns/iter (+/- 20,928)
test spawn_many ... bench: 7,454,898 ns/iter (+/- 283,449)
test yield_many ... bench: 16,771,113 ns/iter (+/- 733,424)
```
Real-world benchmarks improve significantly as well. This is testing the hyper hello
world server using: `wrk -t1 -c50 -d10`:
Old scheduler:
```
Running 10s test @ http://127.0.0.1:3000
1 threads and 50 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 371.53us 99.05us 1.97ms 60.53%
Req/Sec 114.61k 8.45k 133.85k 67.00%
1139307 requests in 10.00s, 95.61MB read
Requests/sec: 113923.19
Transfer/sec: 9.56MB
```
New scheduler:
```
Running 10s test @ http://127.0.0.1:3000
1 threads and 50 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 275.05us 69.81us 1.09ms 73.57%
Req/Sec 153.17k 10.68k 171.51k 71.00%
1522671 requests in 10.00s, 127.79MB read
Requests/sec: 152258.70
Transfer/sec: 12.78MB
```
As discussed in #1620, the attribute names for `#[tokio::main]` and
`#[tokio::test]` aren't great. Specifically, they both use
`single_thread` and `multi_thread`, as opposed to names that match the
runtime names: `current_thread` and `threadpool`. This PR changes the
former to the latter.
Fixes#1627.
`is_terminated` must return `true` until the future has been polled at least once to make sure that the associated block in select is called even after the delay has elapsed.
You use `Delay` in a `select!` by [fusing it](https://docs.rs/futures-preview/0.3.0-alpha.19/futures/future/trait.FutureExt.html#method.fuse):
```rust
let delay = tokio::timer::delay(/* ... */);
let delay = delay.fuse();
select! {
_ = delay => {
/* work here */
}
}
```
When polling the task, the current waker is saved to the oneshot state.
When the handle is migrated to a new task and polled again, the waker
must be swaped from the old waker to the new waker. In some cases, there
is a potential for the old waker to leak.
This bug was caught by loom with the recently added memory leak
detection.
Use a counter to count notifications. This protects against spurious
wakeups by pthreads and other libraries. The state transitions now
track num_idle precisely.
The standard library's `io` module has small utilities such as `repeat`,
`empty`, and `sink`, which return `Read` and `Write` implementations.
These can come in handy in some circiumstances. `tokio::io` has no
equivalents that implement `AsyncRead`/`AsyncWrite`.
This commit adds `repeat`, `empty`, and `sink` helpers to `tokio::io`.
In the past, it was not possible to choose to use the multi-threaded
tokio `Runtime` in tests, which meant that any test that transitively
used `executor::threadpool::blocking` would fail with
```
'blocking' annotation used from outside the context of a thread pool
```
This patch adds a runtime annotation attribute to `#[tokio::test]` just
like `#[tokio::main]` has, which lets users opt in to the threadpool
runtime over `current_thread` (the default).
The algorithm backing `AtomicWaker` effectively uses a spin lock backed
by notifying & yielding the current task. This adds a `spin_lock_hint`
annotation to cover this case.
While, in practice, the omission of `spin_lock_hint` would not cause
problems, there are platforms that do not handle spin locks very well
and could enter a deadlock in pathological cases.
- Adds a minimum `rt-current-thread` optional feature that exports
`tokio::runtime::current_thread`.
- Adds a `macros` optional feature to enable the `#[tokio::main]` and
`#[tokio::test]` attributes.
- Adjusts `#[tokio::main]` macro to select a runtime "automatically" if
a specific strategy isn't specified. Allows using the macro with only
the rt-current-thread feature.
* Removes most pin-projection related unsafe code.
* Removes manual Unpin implementations.
As references always implement Unpin, there is no need to implement
Unpin manually.
* Adds tests to check that Unpin requirement does not change accidentally
because changing Unpin requirements will be breaking changes.
`BufWriter` and `BufReader` did not previously forward the "opposite" trait (`AsyncRead` for `BufWriter` and `AsyncWrite` for `BufReader`). This meant that there was no way to have both directions buffered at once. This patch fixes that, and introduces a convenience type + constructor for this double-wrapped construct.
This adds `Barrier` to `tokio-sync`, which is an asynchronous alternative to [`std::sync::Barrier`](https://doc.rust-lang.org/std/sync/struct.Barrier.html). It is a synchronization primitive that allows multiple futures to "rendezvous" at certain points in their execution.
Currently, when threads in the blocking pool shutdown due to being idle
the counter tracking threads is not decremented. This prevents new threads
from being spawned to replace the shutdown threads.
This renames `Lock` to `Mutex`, and brings the API more in line with `std::sync::Mutex`.
In partcular, locking now only takes `&self`, with the expectation that you place the `Mutex` in an `Arc` (or something similar) to share it between threads.
Fixes#1544.
Part of #1210.
Bring back `split` utility as a free fn instead of a method on
`AsyncRead`. This utility wraps the `stream` in an `Arc` and uses mutual
exclusion to ensure correct access.
Additionally, the specialized `split_mut` fn on TcpStream and UdsStream
is promoted to `split`.
The Tokio project adheres to the [Rust Code of Conduct](https://www.rust-lang.org/policies/code-of-conduct). This describes the minimum behavior expected from all contributors.
## Enforcement
Instances of violations of the Code of Conduct can be reported by contacting the project team at [[email protected]](mailto:[email protected]).
This guide will help you get started. **Do not let this guide intimidate you**.
It should be considered a map to help you navigate the process.
You may also get help with contributing in the [dev channel][dev], please join
The [dev channel][dev] is available for any concerns not covered in this guide, please join
us!
[dev]: https://gitter.im/tokio-rs/dev
[dev]: https://discord.gg/6yGkFeN
## Conduct
The Tokio project adheres to the [Rust Code of Conduct][coc]. This describes
the _minimum_ behavior expected from all contributors.
the _minimum_ behavior expected from all contributors. Instances of violations of the Code of Conduct can be reported by contacting the project team at [[email protected]](mailto:[email protected]).
_NOTE_: Tokio's [`master`](https://github.com/tokio-rs/tokio) branch is currently in the process of moving to [`std::future::Future`](https://doc.rust-lang.org/std/future/trait.Future.html), for `v0.1.x` based tokio releases please check out the [`v0.1.x`](https://github.com/tokio-rs/tokio/tree/v0.1.x) branch.
A runtime for writing reliable, asynchronous, and slim applications with
the Rust programming language. It is:
@@ -17,7 +15,7 @@ the Rust programming language. It is:
// In this case, CPU pool completes the benchmark faster, but this is due to how
// CpuPool currently behaves, starving other futures. This completes the
// benchmark quickly but results in poor runtime characteristics for a thread
// pool.
//
// See rust-lang-nursery/futures-rs#617
//
modcpupool{
usefutures::future::{self,Executor};
usefutures::{task,Async};
usefutures_cpupool::*;
usenum_cpus;
usestd::sync::atomic::AtomicUsize;
usestd::sync::atomic::Ordering::SeqCst;
usestd::sync::{mpsc,Arc};
#[bench]
fnspawn_many(b: &muttest::Bencher){
letpool=CpuPool::new(num_cpus::get());
let(tx,rx)=mpsc::sync_channel(10);
letrem=Arc::new(AtomicUsize::new(0));
b.iter(move||{
rem.store(super::NUM_SPAWN,SeqCst);
for_in0..super::NUM_SPAWN{
lettx=tx.clone();
letrem=rem.clone();
pool.execute(future::lazy(move||{
if1==rem.fetch_sub(1,SeqCst){
tx.send(()).unwrap();
}
Ok(())
}))
.ok()
.unwrap();
}
let_=rx.recv().unwrap();
});
}
#[bench]
fnyield_many(b: &muttest::Bencher){
letpool=CpuPool::new(num_cpus::get());
lettasks=super::TASKS_PER_CPU*num_cpus::get();
let(tx,rx)=mpsc::sync_channel(tasks);
b.iter(move||{
for_in0..tasks{
letmutrem=super::NUM_YIELD;
lettx=tx.clone();
pool.execute(future::poll_fn(move||{
rem-=1;
ifrem==0{
tx.send(()).unwrap();
Ok(Async::Ready(()))
}else{
// Notify the current task
task::current().notify();
// Not ready
Ok(Async::NotReady)
}
}))
.ok()
.unwrap();
}
for_in0..tasks{
let_=rx.recv().unwrap();
}
});
}
}
Some files were not shown because too many files have changed in this diff
Show More
Reference in New Issue
Block a user
Blocking a user prevents them from interacting with repositories, such as opening or commenting on pull requests or issues. Learn more about blocking a user.
Carl Lerche
