This reverts commit fe2b997.
We are avoiding adding poll_read_buf to tokio itself for now. The patch is
reverted now in order to not block the v0.3.2 release (#3059).
Currently when `RwLockWriteGuard::downgrade` the `MAX_READS - 1`
permits are added to the semaphore. When `RwLockWriteGuard::drop`
gets invoked however another `MAX_READS` permits are added. This
results in releasing more permits that were actually aquired when
downgrading a write to a read lock. This is why we need to call
`mem::forget` on the `RwLockWriteGuard` in order to avoid
invoking the destructor.
Fixes: #2941
* io: Add AsyncFd
This adds AsyncFd, a unix-only structure to allow for read/writability states
to be monitored for arbitrary file descriptors.
Issue: #2728
* driver: fix shutdown notification unreliability
Previously, there was a race window in which an IO driver shutting down could
fail to notify ScheduledIo instances of this state; in particular, notification
of outstanding ScheduledIo registrations was driven by `Driver::drop`, but
registrations bypass `Driver` and go directly to a `Weak<Inner>`. The `Driver`
holds the `Arc<Inner>` keeping `Inner` alive, but it's possible that a new
handle could be registered (or a new readiness future created for an existing
handle) after the `Driver::drop` handler runs and prior to `Inner` being
dropped.
This change fixes this in two parts: First, notification of outstanding
ScheduledIo handles is pushed down into the drop method of `Inner` instead,
and, second, we add state to ScheduledIo to ensure that we remember that the IO
driver we're bound to has shut down after the initial shutdown notification, so
that subsequent readiness future registrations can immediately return (instead
of potentially blocking indefinitely).
Fixes: #2924
The `Receiver` handle maintains a position in the broadcast channel for
itself. Cloning implies copying the state of the value. Intuitively,
cloning a `broadcast::Receiver` would return a new receiver with an
identical position. However, the current implementation returns a new
`Receiver` positioned at the tail of the channel.
This behavior subtlety is why `new_subscriber()` is used to create
`Receiver` handles. An alternate API should consider the position issue.
Refs: #2933
This combines the `dns` and `net` feature flags. Previously, `dns` was
included as part of `net`. Given that is is rare that one would want
`dns` without `net`, DNS is now entirely gated w/ `net`.
The `parking_lot` feature is included as part of `full`.
Some misc docs are tweaked to reflect feature flag changes.
Changes inherent methods to take `&self` instead of `&mut self`. This
brings the API in line with `std`.
This patch is implemented by using a `tokio::sync::Mutex` to guard the
internal `File` state. This is not an ideal implementation strategy
doesn't make a big impact compared to having to dispatch operations to a
background thread followed by a blocking syscall.
In the future, the implementation can be improved as we explore async
file-system APIs provided by the operating-system (iocp / io_uring).
Closes#2927
Uses the infrastructure added by #2828 to enable switching
`TcpListener::accept` to use `&self`.
This also switches `poll_accept` to use `&self`. While doing introduces
a hazard, `poll_*` style functions are considered low-level. Most users
will use the `async fn` variants which are more misuse-resistant.
TcpListener::incoming() is temporarily removed as it has the same
problem as `TcpSocket::by_ref()` and will be implemented later.
This enables the caller to configure the socket and to explicitly bind
the socket before converting it to a `TcpStream` or `TcpListener`.
Closes: #2902
* remove unnecessary wheel::Poll
the timer wheel uses the `wheel::Poll` struct as input when
advancing the timer to the next time step. the `Poll` struct
contains an instant representing the time step to advance to
and also contains an optional and mutable reference to an
`Expiration` struct. from what I can tell, the latter field
is only used in the context of polling the wheel and does not
need to be exposed outside of that method. without the
expiration field the `Poll` struct is nothing more than a
wrapper around the instant being polled. this change removes
the `Poll` struct and updates integration points accordingly.
* remove Stack trait in favor of concrete Stack implementation
* remove timer Registration struct
This change is intended to do the minimum to unblock 0.3; as such, for now, we
duplicate the internal `time::wheel` structures in tokio-util, rather than trying
to refactor things at this stage.
Co-authored-by: Bryan Donlan <[email protected]>
In the `readiness` future, before inserting a waiter into the list, the current socket readiness is eagerly checked. However, it would return as a `ReadyEvent` the entire socket readiness, instead of just the interest desired from `readiness(interest)`. This would result in the later call to `clear_readiness(event)` removing all of it.
Closes#2886
Adding closed future, makes it possible to select over closed and some other
work, so that the task is woken when the channel is closed and can proactively
cancel itself.
Added a mpsc::Sender::closed future that will become ready when the receiver
is closed.
As tokio does not rely on poisoning, we can
avoid always unwrapping when locking by handling
the `PoisonError` in the Mutex shim.
Signed-off-by: Zahari Dichev <[email protected]>
Updates the mpsc channel to use the intrusive waker based sempahore.
This enables using `Sender` with `&self`.
Instead of using `Sender::poll_ready` to ensure capacity and updating
the `Sender` state, `async fn Sender::reserve()` is added. This function
returns a `Permit` value representing the reserved capacity.
Fixes: #2637
Refs: #2718 (intrusive waiters)
These functions have object safety issues. It also has been decided to
avoid vectored operations on the I/O traits. A later PR will bring back
vectored operations on specific types that support them.
Refs: #2879, #2716
As we go into 0.3 we no longer need to support older versions of Rust where
IoSlice did not implement Copy and Clone, so we can more easily initialize the
IoSlice array in net::tcp::stream.
Co-authored-by: Bryan Donlan <[email protected]>
This change will still internally compile any `signal` resources
required when `process` is enabled on unix systems, but it will not
publicly turn on the cargo feature
This refactors I/O registration in a few ways:
- Cleans up the cached readiness in `PollEvented`. This cache used to
be helpful when readiness was a linked list of `*mut Node`s in
`Registration`. Previous refactors have turned `Registration` into just
an `AtomicUsize` holding the current readiness, so the cache is just
extra work and complexity. Gone.
- Polling the `Registration` for readiness now gives a `ReadyEvent`,
which includes the driver tick. This event must be passed back into
`clear_readiness`, so that the readiness is only cleared from `Registration`
if the tick hasn't changed. Previously, it was possible to clear the
readiness even though another thread had *just* polled the driver and
found the socket ready again.
- Registration now also contains an `async fn readiness`, which stores
wakers in an instrusive linked list. This allows an unbounded number
of tasks to register for readiness (previously, only 1 per direction (read
and write)). By using the intrusive linked list, there is no concern of
leaking the storage of the wakers, since they are stored inside the `async fn`
and released when the future is dropped.
- Registration retains a `poll_readiness(Direction)` method, to support
`AsyncRead` and `AsyncWrite`. They aren't able to use `async fn`s, and
so there are 2 reserved slots for those methods.
- IO types where it makes sense to have multiple tasks waiting on them
now take advantage of this new `async fn readiness`, such as `UdpSocket`
and `UnixDatagram`.
Additionally, this makes the `io-driver` "feature" internal-only (no longer
documented, not part of public API), and adds a second internal-only
feature, `io-readiness`, to group together linked list part of registration
that is only used by some of the IO types.
After a bit of discussion, changing stream-based transports (like
`TcpStream`) to have `async fn read(&self)` is punted, since that
is likely too easy of a footgun to activate.
Refs: #2779, #2728
*In `watch::Receiver::changed` `Notified` was polled
for the first time to ensure the waiter is registered while
assuming that the first poll will always return `Pending`.
It is the case however that another instance of `Notified`
is dropped without receiving its notification, this "orphaned"
notification can be used to satisfy another waiter without
even registering it. This commit accounts for that scenario.
When the mpsc channel receiver closes the channel, receiving should
return `None` once all in-progress sends have completed. When a sender
reserves capacity, this prevents the receiver from fully shutting down.
Previously, when the sender, after reserving capacity, dropped without
sending a message, the receiver was not notified. This results in
blocking the shutdown process until all sender handles drop.
This patch adds a receiver notification when the channel is both closed
and all outstanding sends have completed.
When the mpsc channel receiver closes the channel, receiving should
return `None` once all in-progress sends have completed. When a sender
reserves capacity, this prevents the receiver from fully shutting down.
Previously, when the sender, after reserving capacity, dropped without
sending a message, the receiver was not notified. This results in
blocking the shutdown process until all sender handles drop.
This patch adds a receiver notification when the channel is both closed
and all outstanding sends have completed.
* Add const constructors to `RwLock`, `Notify`, and `Semaphore`.
Referring to the types in `tokio::sync`.
Also add `const` to `new` for the remaining atomic integers in `src/loom` and `UnsafeCell`.
Builds upon previous work in #2790Closes#2756
Decouples getting the latest `watch` value from receiving the change
notification. The `Receiver` async method becomes
`Receiver::changed()`. The latest value is obtained from
`Receiver::borrow()`.
The implementation is updated to use `Notify`. This requires adding
`Notify::notify_waiters`. This method is generally useful but is kept
private for now.
Previous docs look like they were based on the docs for
`insert_at`. Changed names of variables referred to and the
explanation of when the value will be returned and under what
condition it will be immediately available to make sense for a
Duration argument instead of an Instant.
* This changes the `Child::kill` to be an async method which awaits the
child after sending a kill signal. This avoids leaving zombie
processes on Unix platforms if the caller forgets to await the child
after the kill completes
* A `start_kill` method was also added on `Child` which only sends the
kill signal to the child process. This allows for kill signals to be
sent even outside of async contexts.
* add Child::try_wait to mirror the std API
* replace Future impl on Child with `.wait()` method to bring our
APIs closer to those in std and it allow us to
internally fuse the future so that repeated calls to `wait` result in
the same value (similar to std) without forcing the caller to fuse the
outer future
* Also change `Child::id` to return an Option result to avoid
allowing the caller to accidentally use the pid on Unix systems after
the child has been reaped
* Also remove deprecated Child methods
* sync: Implement map methods of parking_lot fame
Generally, this mimics the way `MappedRwLock*Guard`s are implemented in
`parking_lot`. By storing a raw pointer in the guards themselves
referencing the mapped data and maintaining type invariants through
`PhantomData`. I didn't try to think too much about this, so if someone
has objections I'd love to hear them.
I've also dropped the internal use of `ReleasingPermit`, since it made
the guards unecessarily large. The number of permits that need to be
released are already known by the guards themselves, and is instead
governed directly in the relevant `Drop` impls. This has the benefit of
making the guards as small as possible, for the non-mapped variants this
means a single reference is enough.
`fmt::Debug` impls have been adjusted to behave exactly like the
delegating impls in `parking_lot`. `fmt::Display` impls have been added
for all guard types which behave the same. This does change the format
of debug impls, for which I'm not sure if we provide any guarantees.
Generally, this mimics the way `MappedRwLock*Guard`s are implemented in
`parking_lot`. By storing a raw pointer in the guards themselves
referencing the mapped data and maintaining type invariants through
`PhantomData`. I didn't try to think too much about this, so if someone
has objections I'd love to hear them.
I've also dropped the internal use of `ReleasingPermit`, since it made
the guards unecessarily large. The number of permits that need to be
released are already known by the guards themselves, and is instead
governed directly in the relevant `Drop` impls. This has the benefit of
making the guards as small as possible, for the non-mapped variants this
means a single reference is enough.
`fmt::Debug` impls have been adjusted to behave exactly like the
delegating impls in `parking_lot`. `fmt::Display` impls have been added
for all guard types which behave the same. This does change the format
of debug impls, for which I'm not sure if we provide any guarantees.
* sync: move CancellationToken to tokio-util
The `CancellationToken` utility is only available with the
`tokio_unstable` flag. This was done as the API is not final, but it
adds friction for users.
This patch moves `CancellationToken` to tokio-util where it is generally
available. The tokio-util crate does not have any constraints on
breaking change releases.
* fix clippy
* clippy again
* net: adding examples for UnixDatagram
Adding examples to documentation for UnixDatagram
* net: document named UnixDatagrams persistence
Add documentation to indicate that named UnixDatagrams 'leak'
socket files after execution.
* net: rustfmt issue in UnixDatagram
Fixing rustfmt issue in UnixDatagram
* net: adding examples for UnixDatagram
Fixes: #2686
Refs: #1679
Refs: #1111
The I/O driver uses a slab to store per-resource state. Doing this
provides two benefits. First, allocating state is streamlined. Second,
resources may be safely indexed using a `usize` type. The `usize` is
used passed to the OS's selector when registering for receiving events.
The original slab implementation used a `Vec` backed by `RwLock`. This
primarily caused contention when reading state. This implementation also
only **grew** the slab capacity but never shrank. In #1625, the slab was
rewritten to use a lock-free strategy. The lock contention was removed
but this implementation was still grow-only.
This change adds the ability to release memory. Similar to the previous
implementation, it structures the slab to use a vector of pages. This
enables growing the slab without having to move any previous entries. It
also adds the ability to release pages. This is done by introducing a
lock when allocating/releasing slab entries. This does not impact
benchmarks, primarily due to the existing implementation not being
"done" and also having a lock around allocating and releasing.
A `Slab::compact()` function is added. Pages are iterated. When a page
is found with no slots in use, the page is freed. The `compact()`
function is called occasionally by the I/O driver.
Fixes#2505
JoinHandle of threads created by the pool are now tracked and properly joined at
shutdown. If the thread does not return within the timeout, then it's not joined and
left to the OS for cleanup.
Also, break a cycle between wakers held by the timer and the runtime.
Fixes#2641, #2535
Previously, we would fail to reset the coop budget in this case, making
it so that `coop::poll_proceed` would perpetually yield `Poll::Pending`
in nested executers even when run in `block_in_place`.
This is also a further improvement on #2645.
The new implementation changes the behavior such that set_len is called
after poll_read. The motivation of this change is that it makes it much
more obvious that a rouge panic won't give the caller access to a vector
containing exposed uninitialized memory. The new implementation also
makes sure to not zero memory twice.
Additionally, it makes the various implementations more consistent with
each other regarding the naming of variables, and whether we store how many
bytes we have read, or how many were in the container originally.
Fixes: #2544
* Add Unit Test demonstrating the issue
This test demonstrates a panic that occurs when the user inserts an
item with an instant in the past, and then tries to reset the timeout
using the returned key
* Guard reset_at against removals of expired items
Trying to remove an already expired Timer Wheel entry (called by
DelayQueue.reset()) causes panics in some cases as described in (#2573)
This prevents this panic by removing the item from the expired queue and
not the wheel in these cases
Fixes: #2473
The documentation build failed with errors such as
error: `[read]` public documentation for `take` links to a private item
--> tokio/src/io/util/async_read_ext.rs:1078:9
|
1078 | / /// Creates an adaptor which reads at most `limit` bytes from it.
1079 | | ///
1080 | | /// This function returns a new instance of `AsyncRead` which will read
1081 | | /// at most `limit` bytes, after which it will always return EOF
... |
1103 | | /// }
1104 | | /// ```
| |_______________^
|
note: the lint level is defined here
--> tokio/src/lib.rs:13:9
|
13 | #![deny(intra_doc_link_resolution_failure)]
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
= note: the link appears in this line:
bytes read and future calls to [`read()`][read] may succeed.
`duplex` returns a pair of connected `DuplexStream`s.
`DuplexStream` is a bidirectional type that can be used to simulate IO,
but over an in-process piece of memory.
Dropping a runtime normally involves waiting for any outstanding blocking tasks
to complete. When this drop happens in an asynchronous context, we previously
would issue a cryptic panic due to trying to block in an asynchronous context.
This change improves the panic message, and adds a `shutdown_blocking()` function
which can be used to shutdown a runtime without blocking at all, as an out for
cases where this really is necessary.
Co-authored-by: Bryan Donlan <[email protected]>
Co-authored-by: Alice Ryhl <[email protected]>
For some yet unknown reason using the default on a wrapped `Bufreader<TcpStream>`
causes the hyper server to sometimes fail to send the entire body in the
response.
This fixes that problem for us and ensures that hyper has a chance to
use vectored IO (making it a good change regardless of the mentioned
bug)
A fast path in block_on_place was failing to call exit() in the case where we
were in a block_on call.
Fixes: #2639
Co-authored-by: Bryan Donlan <[email protected]>
## Motivation
When debugging asynchronous systems, it can be very valuable to inspect
what tasks are currently active (see #2510). The [`tracing` crate] and
related libraries provide an interface for Rust libraries and
applications to emit and consume structured, contextual, and async-aware
diagnostic information. Because this diagnostic information is
structured and machine-readable, it is a better fit for the
task-tracking use case than textual logging — `tracing` spans can be
consumed to generate metrics ranging from a simple counter of active
tasks to histograms of poll durations, idle durations, and total task
lifetimes. This information is potentially valuable to both Tokio users
*and* to maintainers.
Additionally, `tracing` is maintained by the Tokio project and is
becoming widely adopted by other libraries in the "Tokio stack", such as
[`hyper`], [`h2`], and [`tonic`] and in [other] [parts] of the broader Rust
ecosystem. Therefore, it is suitable for use in Tokio itself.
[`tracing` crate]: https://github.com/tokio-rs/tracing
[`hyper`]: https://github.com/hyperium/hyper/pull/2204
[`h2`]: https://github.com/hyperium/h2/pull/475
[`tonic`]: https://github.com/hyperium/tonic/blob/570c606397e47406ec148fe1763586e87a8f5298/tonic/Cargo.toml#L48
[other]: https://github.com/rust-lang/chalk/pull/525
[parts]: https://github.com/rust-lang/compiler-team/issues/331
## Solution
This PR is an MVP for instrumenting Tokio with `tracing` spans. When the
"tracing" optional dependency is enabled, every spawned future will be
instrumented with a `tracing` span.
The generated spans are at the `TRACE` verbosity level, and have the
target "tokio::task", which may be used by consumers to filter whether
they should be recorded. They include fields for the type name of the
spawned future and for what kind of task the span corresponds to (a
standard `spawn`ed task, a local task spawned by `spawn_local`, or a
`blocking` task spawned by `spawn_blocking`). Because `tracing` has
separate concepts of "opening/closing" and "entering/exiting" a span, we
enter these spans every time the spawned task is polled. This allows
collecting data such as:
- the total lifetime of the task from `spawn` to `drop`
- the number of times the task was polled before it completed
- the duration of each individual time that the span was polled (and
therefore, aggregated metrics like histograms or averages of poll
durations)
- the total time a span was actively being polled, and the total time
it was alive but **not** being polled
- the time between when the task was `spawn`ed and the first poll
As an example, here is the output of a version of the `chat` example
instrumented with `tracing`:
And, with multiple connections actually sending messages:
I haven't added any `tracing` spans in the example, only converted the
existing `println!`s to `tracing::info` and `tracing::error` for
consistency. The span durations in the above output are generated by
`tracing-subscriber`. Of course, a Tokio-specific subscriber could
generate even more detailed statistics, but that's follow-up work once
basic tracing support has been added.
Note that the `Instrumented` type from `tracing-futures`, which attaches
a `tracing` span to a future, was reimplemented inside of Tokio to avoid
a dependency on that crate. `tracing-futures` has a feature flag that
enables an optional dependency on Tokio, and I believe that if another
crate in a dependency graph enables that feature while Tokio's `tracing`
support is also enabled, it would create a circular dependency that
Cargo wouldn't be able to handle. Also, it avoids a dependency for a
very small amount of code that is unlikely to ever change.
There is, of course, room for plenty of future work here. This might
include:
- instrumenting other parts of `tokio`, such as I/O resources and
channels (possibly via waker instrumentation)
- instrumenting the threadpool so that the state of worker threads
can be inspected
- writing `tracing-subscriber` `Layer`s to collect and display
Tokio-specific data from these traces
- using `track_caller` (when it's stable) to record _where_ a task
was `spawn`ed from
However, this is intended as an MVP to get us started on that path.
Signed-off-by: Eliza Weisman <[email protected]>
Previously, dropping the Write handle would issue a `shutdown(Both)`. However,
shutting down the read half is not portable and not the correct action to take.
This changes the behavior of OwnedWriteHalf to only perform a `shutdown(Write)`
on drop.
* doc: Update the docs of "pause" to state that time will still advance
This was changed in #2059. This had me extremely confused for some time
as my timeouts fired immediately, without the wrapped future that were
waiting on IO to actually run long enough.
I am not sure about the exact wording here but this had me very confused
for some time. Deprecating "pause" and giving it a more accurate name
may be a good idea as well.
```rust
async fn timeout_advances() {
time::pause();
timeout(ms(1), async {
// Change to 1 and the this future resolve, 2 or
// more and the timeout resolves
for _ in 0..2 {
tokio::task::yield_now().await
}
})
.await
.unwrap();
}
```
* Update tokio/src/time/clock.rs
Co-authored-by: Alice Ryhl <[email protected]>
Co-authored-by: Alice Ryhl <[email protected]>
This paragraph from `std::io::Read::read_ext()` applies to
*implementors* of `Read`. Since `AsyncReadExt` can't and shouldn't be
implemented outside of this crate, this documentation is unnecessary.
This took me a bit to catch on to because I didn't really think there was any reason to investigate the individual documentation of each half. As someone dealing with TCP streams directly for first time (without previous experience from other languages) this caught me by surprise
This patch updates the coop logic so that the budget is only decremented
if a future makes progress (that is, if it returns `Ready`). This is
realized by restoring the budget to its former value after
`poll_proceed` _unless_ the caller indicates that it made progress.
The thinking here is that we always want tasks to make progress when we
poll them. With the way things were, if a task polled 128 resources that
could make no progress, and just returned `Pending`, then a 129th
resource that _could_ make progress would not be polled. Worse yet, this
could manifest as a deadlock, if the first 128 resources were all
_waiting_ for the 129th resource, since it would _never_ be polled.
The downside of this change is that `Pending` resources now do not take
up any part of the budget, even though they _do_ take up time on the
executor. If a task is particularly aggressive (or unoptimized), and
polls a large number of resources that cannot make progress whenever it
is polled, then coop will allow it to run potentially much longer before
yielding than it could before. The impact of this should be relatively
contained though, because tasks that behaved in this way in the past
probably ignored `Pending` _anyway_, so whether a resource returned
`Pending` due to coop or due to lack of progress may not make a
difference to it.
Trait OpenOptionsExt is now implemented for fs::OpenOption.
In order to access the underlying std::fs::OpenOptions wrapped in
tokio's OpenOption, an as_inner_mut method was added to OpenOption,
only visible to the parent module.
Fixes: #2366
The initial idea was to implement a thin wrapper around an internally
held `std::fs::DirBuilder` instance. This, however, didn't work due to
`std::fs::DirBuilder` not having a Copy/Clone traits implemented, which
are necessary for constructing an instance to move-capture it into a
closure.
Instead, we mirror `std::fs::DirBuilder` configuration by storing the
`recursive` and (unix-only) `mode` parameters locally, which are then
used to construct an `std::fs::DirBuilder` instance on-the-fly.
This commit also mirrors the (unix-only) DirBuilderExt trait from std.
Fixes: #2369
Adds a method to retrieve a mutable reference to the Framed stream's read buffer.
This makes it possible to e.g. externally clear the buffer to prevent the codec from
parsing stale data.
Previously, in the broadcast channel, receiver wakers were passed to the
sender via an atomic stack with allocated nodes. When a message was
sent, the stack was drained. This caused a problem when many receivers
pushed a waiter node then dropped. The waiter node remained indefinitely
in cases where no values were sent.
This patch switches broadcast to use the intrusive linked-list waiter
strategy used by `Notify` and `Semaphore.
I was building the docs and got the following documentation warning:
warning: unknown attribute `norun`. Did you mean `no_run`?
--> tokio/src/time/throttle.rs:13:1
|
13 | / /// Slows down a stream by enforcing a delay between items.
14 | | /// They will be produced not more often than the specified interval.
15 | | ///
16 | | /// # Example
... |
31 | | /// # }
32 | | /// ```
| |_______^
|
= help: the code block will either not be tested if not marked as a rust one or will be run (which you might not want)
Framed was designed to encapsulate both AsyncRead and AsyncWrite so
that it could wrap two-way connections. It used Fuse to manage the pinned
io object between the FramedWrite and FramedRead structs.
I replaced the Fuse struct by isolating the state used in reading and
writing, and making the code generic over that instead. This means
the FramedImpl struct now has a parameter for the state, and contains
the logic for both directions. The Framed* structs are now simply
wrappers around this type
Hopefully removing the `Pin` handling made things easier to
understand, too.
`tokio::test` and related macros now use the absolute path
`::core::prelude::v1::test` to refer to the built-in `test` macro.
This absolute path was introduced in rust-lang/rust#62086.
Although very similar in many regards, illumos and Solaris have been
diverging since the end of OpenSolaris. With the addition of illumos as
a Rust target, it must be wired into the same interfaces which it was
consuming when running under the 'solaris' target.
Add additional methods to allow PollEvented to be created with an appropriate
mio::Ready state, so that it can be properly registered with the reactor.
Fixes#2413
In some cases, when a call to `block_in_place` completes, the runtime is
reinstated on the thread. In this case, the task budget must also be set
in order to avoid starving other tasks on the worker.
As a first step towards structured concurrency, this change adds a
CancellationToken for graceful cancellation of tasks.
The task can be awaited by an arbitrary amount of tasks due to the usage
of an intrusive list.
The token can be cloned. In addition to this child tokens can be derived.
When the parent token gets cancelled, all child tokens will also get
cancelled.
## Motivation
Currently, an issue exists where a `LocalSet` has a single cooperative
task budget that's shared across all futures spawned on the `LocalSet`
_and_ by any future passed to `LocalSet::run_until` or
`LocalSet::block_on`. Because these methods will poll the `run_until`
future before polling spawned tasks, it is possible for that task to
_always_ deterministically starve the entire `LocalSet` so that no local
tasks can proceed. When the completion of that future _itself_ depends
on other tasks on the `LocalSet`, this will then result in a deadlock,
as in issue #2460.
A detailed description of why this is the case, taken from [this
comment][1]:
`LocalSet` wraps each time a local task is run in `budget`:
https://github.com/tokio-rs/tokio/blob/947045b9445f15fb9314ba0892efa2251076ae73/tokio/src/task/local.rs#L406
This is identical to what tokio's other schedulers do when running
tasks, and in theory should give each task its own budget every time
it's polled.
_However_, `LocalSet` is different from other schedulers. Unlike the
runtime schedulers, a `LocalSet` is itself a future that's run on
another scheduler, in `block_on`. `block_on` _also_ sets a budget:
https://github.com/tokio-rs/tokio/blob/947045b9445f15fb9314ba0892efa2251076ae73/tokio/src/runtime/basic_scheduler.rs#L131
The docs for `budget` state that:
https://github.com/tokio-rs/tokio/blob/947045b9445f15fb9314ba0892efa2251076ae73/tokio/src/coop.rs#L73
This means that inside of a `LocalSet`, the calls to `budget` are
no-ops. Instead, each future polled by the `LocalSet` is subtracting
from a single global budget.
`LocalSet`'s `RunUntil` future polls the provided future before polling
any other tasks spawned on the local set:
https://github.com/tokio-rs/tokio/blob/947045b9445f15fb9314ba0892efa2251076ae73/tokio/src/task/local.rs#L525-L535
In this case, the provided future is `JoinAll`. Unfortunately, every
time a `JoinAll` is polled, it polls _every_ joined future that has not
yet completed. When the number of futures in the `JoinAll` is >= 128,
this means that the `JoinAll` immediately exhausts the task budget. This
would, in theory, be a _good_ thing --- if the `JoinAll` had a huge
number of `JoinHandle`s in it and none of them are ready, it would limit
the time we spend polling those join handles.
However, because the `LocalSet` _actually_ has a single shared task
budget, this means polling the `JoinAll` _always_ exhausts the entire
budget. There is now no budget remaining to poll any other tasks spawned
on the `LocalSet`, and they are never able to complete.
[1]: https://github.com/tokio-rs/tokio/issues/2460#issuecomment-621403122
## Solution
This branch solves this issue by resetting the task budget when polling
a `LocalSet`. I've added a new function to `coop` for resetting the task
budget to `UNCONSTRAINED` for the duration of a closure, and thus
allowing the `budget` calls in `LocalSet` to _actually_ create a new
budget for each spawned local task. Additionally, I've changed
`LocalSet` to _also_ ensure that a separate task budget is applied to
any future passed to `block_on`/`run_until`.
Additionally, I've added a test reproducing the issue described in
#2460. This test fails prior to this change, and passes after it.
Fixes#2460
Signed-off-by: Eliza Weisman <[email protected]>
Do not export the `scoped_thread_local` macro outside of the Tokio
crate. This is not considered a breaking change as the macro never
worked if used from outside of the crate due to the generated code
referencing crate-private types.
This PR adds a new `OwnedMutexGuard` type and `lock_owned` and
`try_lock_owned` methods for `Arc<Mutex<T>>`. This is pretty much the
same as the similar APIs added in #2421.
I've also corrected some existing documentation that incorrectly
implied that the existing `lock` method cloned an internal `Arc` — I
think this may be a holdover from `tokio` 0.1's `Lock` type?
Signed-off-by: Eliza Weisman <[email protected]>
If a Delay has been polled, then the task that polled it may be waiting
for a notification. If the delay gets reset to a time in the past, then
it immediately becomes elapsed, so it should notify the relevant task.
Previously, the function picking the default number of threads for the
threaded runtime did not factor in `max_threads`. Instead, it only used
the value returned by `num_cpus`. However, if `num_cpus` returns a value
greater than `max_threads`, then the function would panic.
This patch fixes the function by limiting the default number of threads
by `max_threads`.
Fixes#2452
Broadcast uses a ring buffer to store values sent to the channel. In order to
deal with slow receivers, the oldest values are overwritten with new values
once the buffer wraps. A receiver should be able to calculate how many values
it has missed.
Additionally, when the broadcast closes, a final value of `None` is sent to
the channel. If the buffer has wrapped, this value overwrites the oldest
value.
This is an issue mainly in a single capacity broadcast when a value is sent
and then the sender is dropped. The original value is immediately overwritten
with `None` meaning that receivers assume they have lagged behind.
**Solution**
A value of `None` is no longer sent to the channel when the final sender has
been dropped. This solves the single capacity broadcast case by completely
removing the behavior of overwriting values when the channel is closed.
Now, when the final sender is dropped a closed bit is set on the next slot
that the channel is supposed to send to.
In the case of a fast receiver, if it finds a slot where the closed bit is
set, it knows the channel is closed without locking the tail.
In the case of a slow receiver, it must first find out if it has missed any
values. This is similar to before, but must be able to account for channel
closure.
If the channel is not closed, the oldest value may be located at index `n`. If
the channel is closed, the oldest value is located at index `n - 1`.
Knowing the index where the oldest value is located, a receiver can calculate
how many values it may have missed and starts to catch up.
Closes#2425
The link to tokio::main was relative to tokio_macros crate in the source
directory. This is why it worked in local build of documentation and not
in doc.rs.
Refs: #1473
This enables `block_in_place` to be used in more contexts. Specifically,
it allows you to block whenever you are off the tokio runtime (like if
you are not using tokio, are in a `spawn_blocking` closure, etc.), and
in the threaded scheduler's `block_on`. Blocking in `LocalSet` and the
basic scheduler's` block_on` is still disallowed.
Fixes#2327.
Fixes#2393.
Included changes
- all simple references like `<type>.<name>.html` for these types
- enum
- fn
- struct
- trait
- type
- simple references for methods, like struct.DelayQueue.html#method.poll
Refs: #1473
Some reasons to prefer logical count as the default:
- Chips reporting many logical CPUs vs physical, such as via
hyperthreading, probably know better than us about the workload the CPUs
can handle.
- The logical count (`num_cpus::get()`) takes into consideration
schedular affinity, and cgroups CPU quota, in case the user wants to
limit the amount of CPUs a process can use.
Closes#2269
Fixes a couple bugs in the work-stealing queue introduced as
part of #2315. First, the cursor needs to be able to represent more
values than the size of the buffer. This is to be able to track if
`tail` is ahead of `head` or if they are identical. This bug resulted in
the "overflow" path being taken before the buffer was full.
The second bug can happen when a queue is being stolen from concurrently
with stealing into. In this case, it is possible for buffer slots to be
overwritten before they are released by the stealer. This is harder to
happen in practice due to the first bug preventing the queue from
filling up 100%, but could still happen. It triggered an assertion in
`steal_into`. This bug slipped through due to a bug in loom not
correctly catching the case. The loom bug is fixed as part of
tokio-rs/loom#119.
Fixes: #2382
Previously, the `Mutex::lock`, `RwLock::{read, write}`, and
`Semaphore::acquire` futures in `tokio::sync` implemented `Send + Sync`
automatically. This was by virtue of being implemented using a `poll_fn`
that only closed over `Send + Sync` types. However, this broke in
PR #2325, which rewrote those types using the new `batch_semaphore`.
Now, they await an `Acquire` future, which contains a `Waiter`, which
internally contains an `UnsafeCell`, and thus does not implement `Sync`.
Since removing previously implemented traits breaks existing code, this
inadvertantly caused a breaking change. There were tests ensuring that
the `Mutex`, `RwLock`, and `Semaphore` types themselves were `Send +
Sync`, but no tests that the _futures they return_ implemented those
traits.
I've fixed this by adding an explicit impl of `Sync` for the
`batch_semaphore::Acquire` future. Since the `Waiter` type held by this
struct is only accessed when borrowed mutably, it is safe for it to
implement `Sync`.
Additionally, I've added to the bounds checks for the effected
`tokio::sync` types to ensure that returned futures continue to
implement `Send + Sync` in the future.
The new queue uses `u8` to track offsets. Cursors are expected to wrap.
An operation was performed with `+` instead of `wrapping_add`. This was
not _obviously_ issue before as it is difficult to wrap a `usize` on
64bit platforms, but wrapping a `u8` is trivial.
The fix is to use `wrapping_add` instead of `+`. A new test is added
that catches the issue.
Fixes#2361
The work-stealing scheduler includes an optimization where each worker
includes a single slot to store the **last** scheduled task. Tasks in
scheduler's LIFO slot are executed next. This speeds up and reduces
latency with message passing patterns.
Previously, this optimization was susceptible to starving other tasks in
certain cases. If two tasks ping-ping between each other without ever
yielding, the worker would never execute other tasks.
An early PR (#2160) introduced a form of pre-emption. Each task is
allocated a per-poll operation budget. Tokio resources will return ready
until the budget is depleted, at which point, Tokio resources will
always return `Pending`.
This patch leverages the operation budget to limit the LIFO scheduler
optimization. When executing tasks from the LIFO slot, the budget is
**not** reset. Once the budget goes to zero, the task in the LIFO slot
is pushed to the back of the queue.
## Motivation
When cancelling futures which are waiting to acquire semaphore permits,
there is a possible dangling pointer if notified futures are dropped
after the notified wakers have been split into a separate list. Because
these futures' wait queue nodes are no longer in the main list guarded
by the lock, their `Drop` impls will complete immediately, and they may
be dropped while still in the list of tasks to notify.
## Solution
This branch fixes this by popping from the wait list inside the lock.
The wakers of popped nodes are temporarily stored in a stack array,
so that they can be notified after the lock is released. Since the
size of the stack array is fixed, we may in some cases have to loop
multiple times, acquiring and releasing the lock, until all permits
have been released. This may also have the possible side advantage of
preventing a thread releasing a very large number of permits from
starving other threads that need to enqueue waiters.
I've also added a loom test that can reliably reproduce a segfault
on master, but passes on this branch (after a lot of iterations).
Signed-off-by: Eliza Weisman <[email protected]>
This reverts commit 826fc21abf.
The code was intentional. Holding the lock while notifying is
unnecessary. Also change the code to use `drop` so clippy doesn't
confuse people against their will.
Fixes a test from a PR that was written before the recent loom upgrade.
A change in the details how loom executes models resulted in the test to
start failing. The fix is to reduce the number of iterations performed
by the test.
This commit improves the memory ordering in the implementation of
Inner's increment function. The former code did a sequentially
consistent load of self.num, then entered a loop with a sequentially
consistent compare and swap on the same, bailing out with and Err only
if the loaded value was MAX_TIMEOUTS. The use of SeqCst means that all
threads must observe all relevant memory operations in the same order,
implying synchronization between all CPUs.
This commit adjusts the implementation in two key ways. First, the
initial load of self.num is now down with Relaxed ordering. If two
threads entered this code simultaneously, formerly, tokio required
that one proceed before the other, negating their parallelism. Now,
either thread may proceed without coordination. Second, the SeqCst
compare_and_swap is changed to a Release, Relaxed
compare_exchange_weak. The first memory ordering referrs to success:
if the value is swapped the load of that value for comparison will be
Relaxed and the store will be Release. The second memory ordering
referrs to failure: if the value is not swapped the load is
Relaxed. The _weak variant may spuriously fail but will generate
better code.
These changes mean that it is possible for more loops to be taken per
call than strictly necessary but with greater parallelism available on
this operation, improved energy consumption as CPUs don't have to
coordinate as much.
When the queue was polled and yielded an index from the wheel, the delay
until the next item was never updated. As a result, when one item was
yielded from `poll_idx` the following insert erronously updated the
delay to the instant of the inserted item.
Fixes: #1700
Loom is having a big refresh to improve performance and tighten up the
concurrency model. This diff tracks those changes.
Included in the changes is the removal of `CausalCell` deferred checks.
This is due to it technically being undefined behavior in the C++11
memory model. To address this, the work-stealing queue is updated to
avoid needing this behavior. This is done by limiting the queue to have
one concurrent stealer.
The resume function was breaking the guarantee that Instants should
never be less than any previously measured Instants when created.
Altered the pause and resume function such that they will not break this
guarantee. After resume, the time should continue from where it left
off.
Created test to prove that the advanced function still works as
expected.
Added additional tests for the pause/advance/resume functions.
## Motivation
Many of Tokio's synchronization primitives (`RwLock`, `Mutex`,
`Semaphore`, and the bounded MPSC channel) are based on the internal
semaphore implementation, called `semaphore_ll`. This semaphore type
provides a lower-level internal API for the semaphore implementation
than the public `Semaphore` type, and supports "batch" operations, where
waiters may acquire more than one permit at a time, and batches of
permits may be released back to the semaphore.
Currently, `semaphore_ll` uses an atomic singly-linked list for the
waiter queue. The linked list implementation is specific to the
semaphore. This implementation therefore requires a heap allocation for
every waiter in the queue. These allocations are owned by the semaphore,
rather than by the task awaiting permits from the semaphore. Critically,
they are only _deallocated_ when permits are released back to the
semaphore, at which point it dequeues as many waiters from the front of
the queue as can be satisfied with the released permits. If a task
attempts to acquire permits from the semaphore and is cancelled (such as
by timing out), their waiter nodes remain in the list until they are
dequeued while releasing permits. In cases where large numbers of tasks
are cancelled while waiting for permits, this results in extremely high
memory use for the semaphore (see #2237).
## Solution
@Matthias247 has proposed that Tokio adopt the approach used in his
`futures-intrusive` crate: using an _intrusive_ linked list to store the
wakers of tasks waiting on a synchronization primitive. In an intrusive
list, each list node is stored as part of the entry that node
represents, rather than in a heap allocation that owns the entry.
Because futures must be pinned in order to be polled, the necessary
invariant of such a list --- that entries may not move while in the list
--- may be upheld by making the waiter node `!Unpin`. In this approach,
the waiter node can be stored inline in the future, rather than
requiring separate heap allocation, and cancelled futures may remove
their nodes from the list.
This branch adds a new semaphore implementation that uses the intrusive
list added to Tokio in #2210. The implementation is essentially a hybrid
of the old `semaphore_ll` and the semaphore used in `futures-intrusive`:
while a `Mutex` around the wait list is necessary, since the intrusive
list is not thread-safe, the permit state is stored outside of the mutex
and updated atomically.
The mutex is acquired only when accessing the wait list — if a task
can acquire sufficient permits without waiting, it does not need to
acquire the lock. When releasing permits, we iterate over the wait
list from the end of the queue until we run out of permits to release,
and split off all the nodes that received enough permits to wake up
into a separate list. Then, we can drain the new list and notify those
wakers *after* releasing the lock. Because the split operation only
modifies the pointers on the head node of the split-off list and the
new tail node of the old list, it is O(1) and does not require an
allocation to return a variable length number of waiters to notify.
Because of the intrusive list invariants, the API provided by the new
`batch_semaphore` is somewhat different than that of `semaphore_ll`. In
particular, the `Permit` type has been removed. This type was primarily
intended allow the reuse of a wait list node allocated on the heap.
Since the intrusive list means we can avoid heap-allocating waiters,
this is no longer necessary. Instead, acquiring permits is done by
polling an `Acquire` future returned by the `Semaphore` type. The use of
a future here ensures that the waiter node is always pinned while
waiting to acquire permits, and that a reference to the semaphore is
available to remove the waiter if the future is cancelled.
Unfortunately, the current implementation of the bounded MPSC requires a
`poll_acquire` operation, and has methods that call it while outside of
a pinned context. Therefore, I've left the old `semaphore_ll`
implementation in place to be used by the bounded MPSC, and updated the
`Mutex`, `RwLock`, and `Semaphore` APIs to use the new implementation.
Hopefully, a subsequent change can update the bounded MPSC to use the
new semaphore as well.
Fixes#2237
Signed-off-by: Eliza Weisman <[email protected]>
Since the original shell runtime was implemented, utilities have been
added to encapsulate `unsafe`. The shell runtime is now able to use
those utilities and not include its own `unsafe` code.
tokio-util uses tokio::stream::StreamExt, which was not introduced until
tokio v0.2.5. The current dependency specification is incorrect, and
breaks with cargo update -Z minimal-versions.
This PR was prompted by having encountered a few cases of people not noticing that Runtime::handle can be cloned, and therefore not realizing it could be moved to another thread.
A single call to `poll` on a top-level task may potentially do a lot of
work before it returns `Poll::Pending`. If a task runs for a long period
of time without yielding back to the executor, it can starve other tasks
waiting on that executor to execute them, or drive underlying resources.
See for example rust-lang/futures-rs#2047, rust-lang/futures-rs#1957,
and rust-lang/futures-rs#869. Since Rust does not have a runtime, it is
difficult to forcibly preempt a long-running task.
Consider a future like this one:
```rust
use tokio::stream::StreamExt;
async fn drop_all<I: Stream>(input: I) {
while let Some(_) = input.next().await {}
}
```
It may look harmless, but consider what happens under heavy load if the
input stream is _always_ ready. If we spawn `drop_all`, the task will
never yield, and will starve other tasks and resources on the same
executor.
This patch adds a `coop` module that provides an opt-in mechanism for
futures to cooperate with the executor to avoid starvation. This
alleviates the problem above:
```
use tokio::stream::StreamExt;
async fn drop_all<I: Stream>(input: I) {
while let Some(_) = input.next().await {
tokio::coop::proceed().await;
}
}
```
The call to [`proceed`] will coordinate with the executor to make sure
that every so often control is yielded back to the executor so it can
run other tasks.
The implementation uses a thread-local counter that simply counts how
many "cooperation points" we have passed since the task was first
polled. Once the "budget" has been spent, any subsequent points will
return `Poll::Pending`, eventually making the top-level task yield. When
it finally does yield, the executor resets the budget before
running the next task.
The budget per task poll is currently hard-coded to 128. Eventually, we
may want to make it dynamic as more cooperation points are added. The
number 128 was chosen more or less arbitrarily to balance the cost of
yielding unnecessarily against the time an executor may be "held up".
At the moment, all the tokio leaf futures ("resources") call into coop,
but external futures have no way of doing so. We probably want to
continue limiting coop points to leaf futures in the future, but may
want to also enable third-party leaf futures to cooperate to benefit the
ecosystem as a whole. This is reflected in the methods marked as `pub`
in `mod coop` (even though the module is only `pub(crate)`). We will
likely also eventually want to expose `coop::limit`, which enables
sub-executors and manual `impl Future` blocks to avoid one sub-task
spending all of their poll budget.
Benchmarks (see tokio-rs/tokio#2160) suggest that the overhead of `coop`
is marginal.
A refactor of the scheduler internals focusing on simplifying and
reducing unsafety. There are no fundamental logic changes.
* The state transitions of the core task component are refined and
reduced.
* `basic_scheduler` has most unsafety removed.
* `local_set` has most unsafety removed.
* `threaded_scheduler` limits most unsafety to its queue implementation.
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 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]).
**NOTE**: Tokio's [`master`](https://github.com/tokio-rs/tokio) is currently undergoing heavy development. This branch and the alpha releases will see API breaking changes and there are currently significant performance regressions that still need to be fixed before the final release. Use the [`v0.1.x`](https://github.com/tokio-rs/tokio/tree/v0.1.x) branch for stable releases.
A runtime for writing reliable, asynchronous, and slim applications with
the Rust programming language. It is:
@@ -17,21 +15,22 @@ the Rust programming language. It is:
The Tokio project team welcomes security reports and is committed to providing prompt attention to security issues. Security issues should be reported privately via [[email protected]](mailto:[email protected]). Security issues should not be reported via the public Github Issue tracker.
## Vulnerability coordination
Remediation of security vulnerabilities is prioritized by the project team. The project team coordinates remediation with third-party project stakeholders via [Github Security Advisories](https://help.github.com/en/github/managing-security-vulnerabilities/about-github-security-advisories). Third-party stakeholders may include the reporter of the issue, affected direct or indirect users of Tokio, and maintainers of upstream dependencies if applicable.
Downstream project maintainers and Tokio users can request participation in coordination of applicable security issues by sending your contact email address, Github username(s) and any other salient information to [[email protected]](mailto:[email protected]). Participation in security issue coordination processes is at the discretion of the Tokio team.
## Security advisories
The project team is committed to transparency in the security issue disclosure process. The Tokio team announces security issues via [project Github Release notes](https://github.com/tokio-rs/tokio/releases) and the [RustSec advisory database](https://github.com/RustSec/advisory-db) (i.e. `cargo-audit`).
/// Returns `true` if the error was caused by the operation timing out.
pubfnis_timeout(&self)-> bool{
self.timeout
}
}
implFrom<EnterError>forRunTimeoutError{
fnfrom(_: EnterError)-> Self{
RunTimeoutError::new(false)
}
}
// ===== impl BlockError =====
impl<T>BlockError<T>{
/// Returns the error yielded by the future being blocked on
pubfninto_inner(self)-> Option<T>{
self.inner
}
}
impl<T>From<EnterError>forBlockError<T>{
fnfrom(_: EnterError)-> Self{
BlockError{inner: None}
}
}
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
