Introduce new AsyncRead / AsyncWrite ​ This PR introduces new versions of AsyncRead / AsyncWrite traits. The proposed changes aim to improve: ​

• ergonomics.
• integration of vectored operations
• working with uninitialized byte slices. ​

Overview

​ The PR changes the AsyncRead and AsyncWrite traits to accept T: Buf and T: BufMut values instead of &[u8] and &mut [u8]. Because &[u8] implements Buf and &mut [u8] implements BufMut, the same calling patterns used today are still possible. Additionally, any type that implements Buf and BufMut may be used. This includes Cursor<&[u8]>, Bytes, ... ​

Improvement in ergonomics

​ Calls to read and write accept buffers, but do not necessary use up the entirety of the buffer. Both functions return a usize representing the number of bytes read / written. Because of this, it is common to write loops such as: ​

let mut rem = &my_data[..];
​
while !rem.is_empty() {
    let n = my_socket.write(rem).await?;
    rem = &rem[n..];
}

​ The key point to notice is having to use the return value to update the position in the cursor. This is both common and error prone. The Buf / BufMut traits aim to ease this by building the cursor concept directly into the buffer. By using these traits with AsyncRead / AsyncWrite, the above loop can be simplified as: ​

let mut buf = Cursor::new(&my_data[..]);
​
while buf.has_remaining() {
    my_socket.write(&mut buf).await?;
}

​ A small reduction in code, but it removes an error prone bit of logic that must be often repeated. ​

Integration of vectored operations

​ In the AsyncRead / AsyncWrite traits provided by futures-io, vectored operations are covered using separate fns: poll_read_vectored and poll_write_vectored. These two functions have default implementations that call the non-vectored operations. ​ This has a draw back, when implementing AsyncRead / AsyncWrite, usually as a layer on top of a type such as TcpStream, the implementor must not forget to impleement these two additional functions. Otherwise, the implementation will not be able to use vectored operations even if the underlying TcpStream supports it. Secondly, it requires duplication of logic: one poll_read implementation and one poll_read_vectored implementation. It is possible to implement one in terms of the other, but this can result in sub-optimial implementations. ​ Imagine a situation where a rope data structure is being written to a socket. This structure is comprised of many smaller byte slices (perhaps thousands). To write it efficiently to the socket, avoiding copying data is preferable. To do this, the byte slices need to be loaded in an IoSlice. Since modern linux systems support a max of 1024 slices, we initialize an array of 1024 slices, iterate the rope to populate this array and call poll_write_vectored. The problem is that, as the caller, we don't know if the AsyncWrite type supports vectored operations or not, poll_write_vectored is called optimistically. However, the implementation "forgot" to proxy its function to TcpStream, so poll_read is called w/ the first entry in the IoSlice. The problem is, for each call to poll_read_vectored, we must iterate 1024 nodes in our rope to only have one chunk written at a time. ​ By using T: Buf as the argument, the decision of whether or not to use vectored operations is left up to the leaf AsyncWrite type. Intermediate layers only implement poll_write w/ T: Buf and pass it along to the inner stream. The TcpStream implementation will know that it supports vectored operations and know how many slices it can write at a time and do "the right thing". ​

Working with uninitialized byte slices

​ When passing buffers to AsyncRead, it is desirable to pass in uninitialized memory which the poll_read call will write to. This avoids the expensive step of zeroing out the memory (doing so has measurable impact at the macro level). The problem is that uninitialized memory is "unsafe", as such, care must be taken. ​ Tokio initially attempted to handle this by adding a prepare_uninitialized_buffer function. std is investigating adding a similar though improved variant of this API. However, over the years, we have learned that the prepare_uninitialized_buffer API is sub optimal for multiple reasons. ​ First, the same problem applies as vectored operations. If an implementation "forgets" to implement prepare_uninitialized_buffer then all slices must be zeroed out before passing it to poll_read, even if the implementation does "the right thing" (not read from initialized memory). In practice, most implementors end up forgetting to implement this function, resulting in memory being zeroed out. ​ Secondly, implementations of AsyncRead that should not require unsafe to implement now must add unsafe simply to avoid having memory zeroed out. ​ Switching the argument to T: BufMut solves this problem via the BufMut trait. First, BufMut provides low-level functions that return &mut [MaybeUninitialized<u8>]. Second, it provides utility functions that provide safe APIs for writing to the buffer (put_slice, put_u8, ...). Again, only the leaf AsyncRead implementations (TcpStream) must use the unsafe APIs. All layers may take advantage of uninitialized memory without the associated unsafety. ​

Drawbacks

​ The primary drawback is genericizing the AsyncRead and AsyncWrite traits. This adds complexity. We feel that the added benefits discussed above outweighs the drawbacks, but only trying it out will validate it. ​

Relation to futures-io, std, and roadmap

​ The relationship between tokio's I/O traits and futures-io has come up a few times in the past. Tokio has historically maintained its own traits. futures-io has just been released with a simplified version of the traits. There is also talk of standardizing AsyncRead and AsyncWrite in std. Because of this, we believe that now is the perfect time to experiment with these traits. This will allow us to gain more experience before committing to traits in std. ​ The next version of Tokio will not be 1.0. This allows us to experiment with these traits and remove them for 1.0 if they do not pan out. ​ Once AsyncRead / AsyncWrite are added to std, Tokio will provide implementations for its types. Until then, tokio-util will provide a compatibility layer between Tokio types and futures-io.

First of all a disclaimer: This issue is not yet a full proposal. This serves more as a collection of things to explore, and to gather feedback on interest.

What is structured concurrency?

Structured concurrency describes programming paradigm. Concurrent tasks are structured in a fashion where there exist clean task hierarchies, and where the lifetime of all sub-tasks/child-tasks is constrained within the lifetime of their parent task.

The term was likely brought up first by Martin Sustrik in this blog post, and was a guiding idea behind the libdill library. @njsmith utilized the term in Notes on structured concurrency, or: Go statement considered harmful, and designed the python trio library around the paradigm. I highly recommend to read the blog post.

The paradigm has also been adopted by Kotlin coroutines. @elizarov gives a talk at HydraConf about structured concurrency and the evolution of Kotlins async task model, which I also highly recommend to watch. It provides some hints on things to look out for, and how APIs could look like. Kotlins documentation around coroutines is also a good resource.

Go adopted some support for structured concurrency with the errgroup package.

Benefits of structured concurrency

I again recommend to check out the linked resources, which also elaborate on this 😀

In short: Applying the structured concurrency paradigm can simplify reasoning about concurrent programs and thereby reduce errors. It is helpful at preventing resource leaks, in the same fashion as RAII allows to avoid leaks on a scope level. It might also allow for optimizations.

Examples around error reductions and simplifications

Here is one motivating example of how structured concurrency can simplify things:

We are building a building a web application A, which is intended to handle at least 1000 transactions per second. Internally each transaction requires a few concurrent interactions, which will involve reaching out to remote services. When one of those transactions fails, we need to perform certain actions. E.g. we need to call another service B for a cleanup or rollback. Without structured concurrency, we might have the idea just to do spawn(cleanup_task()) in order to do this. While this works, it has a side effect: cleanup tasks might still be running while the main webservice handler has already terminated. This sounds harmless at first, but can have surprising consequences: We obviously want our services to be resilient against overloads, so we limit the amount of concurrent requests to 2000 via an async Semaphore. This works fine for our main service handler. But what happens if lots of transactions are failing? How many cleanup tasks can run at the same point of time? The answer is unfortunately, that the number of those is effectively unbounded. Thereby our service can be overloaded through queuing up cleanup tasks - even though we protected ourself against too many concurrent API calls. This can lead to large scale outages in distributed systems.

By making sure all cleanup logic is performed inside the lifetime/scope of the main service handler, we can guarantee that the number of cleanup tasks is also bounded by our Semaphore.

Another example could be applying configuration changes at runtime: While our service is running we want to able to update it's configuration. After the configuration change is applied no transaction should still be utilizing the old configuration. What we need to do now is:

• Disable the acceptor in order to drain requests before we can apply the config change
• Wait for all ongoing transactions to complete
• Cancel transactions if they take too long to complete.
• Update the configuration
• Restart the acceptor

Without having a structured approach for concurrency, this is a lot more of a complicated problem than it sounds. Any old transaction might have spawned a subtask which might still be executing after we have updated the configuration. There is no easy way to check for the higher level code if everything has finished.

Potential for optimizations

The application of structured concurrency might allow for optimizations. E.g. we might be able to allow subtasks to borrow data inside the parent tasks scope without the need for additional heap allocations. Since the exact mechanisms are however not yet designed, the exact potential is unknown.

Core requirements

I think the core requirements for structured concurrency are:

• A parent task will only finish once all child tasks have finished
• When tasks are spawned, they need to be spawned in the context of a parent task. The parent needs to remember it's child tasks
• Parent tasks need to have a mechanism to cancel child tasks
• Errors in child tasks should lead the parent task to return an error as soon as possible, and all sibling tasks to get cancelled. This behavior is equivalent to the behavior of the try_join! macro in futures-rs.

Regarding the last point I am not sure whether automatic error propagation is a required point of structured concurrency and whether it can be achieved on a task level, but it definitely makes things easier.

Do we actually need to have builtin support for this?

Rusts async/await mechanism already provides structured concurrency inside a particular task: By utilizing tools like select! or join! we can run multiple child-tasks which are constrained to the same lifetime - which is the current scope. This is not possible in Go or Kotlin - which require an explicit child-task to be spawned to achieve the behavior. Therefore the benefits might be lower.

I built an examples in futures-intrusive around those mechanisms.

However the concurrency inside a certain task will not scale very well, due requiring polling of all child Futures. Therefore real concurrent tasks will be required in most applications.

On this level we have a set of tools in our toolbox that allow us to structure our current tasks manually:

• Parent tasks can wait for child tasks to join via the use of Oneshot channels or the new JoinHandles
• Parent tasks can forcefully cancel child tasks by just dropping them
• Parent tasks can gracefully cancel parents via passing a cancellation signal.

However these tools all require explicit code in order to guarantee correctness. Builtin support for structured concurrency could improve on usability and allow more developers to use good and correct defaults.

And as mentioned earlier, I think builtin support could also allow for new usages, e.g. borrowing inside child tasks or potential scheduler improvements when switching between child tasks and parent tasks.

The following posts are now mainly a braindump around how these requirements could be fulfilled and how they align with existing mechanisms.

This PR adds basic named pipes support.

Motivation

blackbeam/mysql_async#132

Solution

Implementation directly uses miow::NamedPipeBuilder to avoid making a PR for mio.

It provides two public types:

• NamedPipeServer – which is a server implementation with the familiar accept() method.
• NamedPipe – that represents client or server connection.

NamedPipeServer will hold at least one free instance of a pipe to maintain its existence. Mutex is used within accept() (only for mem::swap) to keep it shared. NamedPipeServer::new() will create the first instance with FILE_FLAG_FIRST_PIPE_INSTANCE flag to avoid named pipe instance creation race condition.

NamedPipe wraps mio::NamedPipe and provides the connect() method for client-side connections. connect() will wait for a server instance using the approach similar to a one used in .NET, namely it'll call WaitNamedPipe with default timeout using the spawn_blocking (I couldn't find a better solution). Unlike .NET, connect() won't wait if pipe doesn't exist and will error immediately.

@udoprog, regarding disconnect() – NamedPipe doesn't provide it since it won't play well with NamedPipeServer, one should simply drop an instance. Also I don't think that we should call it in poll_shutdown.

@fussybeaver, regarding security_qos_flags - implementation unconditionally adds SECURITY_IDENTIFICATION

Other thoughts

Maybe it's too high level.

Related issues

#3118

As known, due to the ownership, we could not r/w the stream at the same time.

The split in tokio seems like a faked split, because it uses mutex for r/w.

Behind the scenes, split ensures that if we both try to read and write at the same time, only one of them happen at a time.

Although we do not block on syscall, which is just the goal of async programming, but the syscall itself is locked by that mutex, i.e. when the read() syscall is processing, we could not do write() syscall. That seems very silly, with obvious performance impact. At syscall level, the read and write could be in parallel, which is meaningful for full-duplex app protocols (e.g. http 1.1 with pipe-lining and http2),

Could we do a real split? Like we could clone mio::net::TcpStream (https://docs.rs/mio/0.6.18/mio/net/struct.TcpStream.html#method.try_clone), but sharing the registration and other stuff? That way, we do have real zero cost abstraction.

Hey, Please provide wasm support to this library. Networking is one of the important feature to work on Rust+WASM codes. Not sure is there any tracker for this. Happy to contribute if any.

Thanks

It has been reported (https://github.com/rust-lang-nursery/futures-rs/issues/1170) that there have been lost wake ups.

Downgrading to threadpool 0.1.4 reportedly solves the problem. #459 is the only meaningful change.

Motivation

Ref: https://github.com/tokio-rs/tokio/issues/4296 https://github.com/tokio-rs/tokio/issues/4296#issuecomment-986005787

In some cases, one could time successfully awaited calls to write_all (or write) with a shutdown of the runtime, and have the write not even be attempted. This can be a bit surprising.

The purpose of this PR is to find a way (if possible) to fix that. There would be no guarantee that the write actually succeeds (any OS error could be hit at the time the write actually gets executed), but at least it would be attempted.

Solution

I have found a sequence of events that leads to spawn_blocking tasks being "ignored". I've written a note about it in a comment. I'm not sure if it's intentional that we won't try draining the queue of blocking tasks before shutting down. Couldn't we tweak the shutdown logic to execute all tasks that were scheduled before the call to shutdown?
If users are concerned about shutting down the runtime taking a long time because of blocking tasks, they can call https://docs.rs/tokio/latest/tokio/runtime/struct.Runtime.html#method.shutdown_timeout or shutdown_background.

The documentation in https://docs.rs/tokio/latest/tokio/runtime/struct.Runtime.html#shutdown says:

The current thread will block until the shut down operation has completed. -Drain any scheduled work queues.

So it should already be expected that shutting down a runtime could block to some extent?

Do you think it would make sense to change the shutdown logic to execute all pending tasks? If so, I can figure out how to do the code change.

Part of #2471, extends the work of #2445. Both this PR and #2445 together close #2471.

This PR introduces the MappedMutexGuard type and adds a map associated function to MutexGuard. The work here is largely based on #2445, so thanks @udoprog for the great PR. This was very easy to implement using your work as a reference.

MappedMutexGuard works almost exactly the same as parking_lot::MappedMutexGuard, but adapted to the internals of tokio::sync::Mutex. The MappedMutexGuard type stores a reference to the semaphore from the original Mutex as well as a raw pointer *mut T that is the result of calling the function passed to map. The Mutex does not hold a mutable reference to its data (it uses an UnsafeCell), so I do not believe that it is possible to accidentally run into any aliasing issues.

I added documentation based on the work in #2445 and the documentation in parking_lot/lock_api. There are doctests in both map and try_map that are almost exactly the same as the ones in #2445 for RwLockWriteGuard. I generated the docs with cargo doc and everything looks great. :tada:

As per #1318, Tokio has been merged into a single crate and components are split by feature flag. Now, with all features enabled, the tokio crate is quite heavy.

Regardless of the direction, two things will happen:

• documentation: the required feature flag for all components will be documented similar to how Tonic does this (see transport module here: https://docs.rs/tonic/0.1.0-alpha.6/tonic/).
• meta feature: A full feature flag will be provided that enables all Tokio feature flags.

The question is, should default include all features, no features, or some features.

Default on

One of the main drawbacks mentioned in #1318 was that, when features are enabled by default, libraries will accidentally depend on more features than necessary. Doing so will force these features to be enabled by consumers of the library. Also, the end user can accidentally use features that were enabled by the dependency. When the dependency changes its feature flags, the application breaks as the required features are no longer available.

Default off

An alternative would be to default to no feature flags enabled by default. In this case, depending on tokio will only enable core traits (AsyncRead, AsyncWrite, ToSocketAddrs) and an empty Runtime type that doesn't do much when used. Getting started guides, examples, the README would instruct users to depend on tokio as:

tokio = { version = "0.2.0", features = ["full"] }

Libraries will be instructed to pick only the features they require. The primary drawback of this strategy is that it adds a bump to the getting started flow.

Default some

A middle ground would be to define a subset of features that should be enabled by default. It is unclear how to pick the features to enable by default as different Tokio users use significantly different feature sets. Because the choice is arbitrary, the end user will have no way to intuit if a feature is enabled by default or not.

There has been frustration among Tokio users regarding the number of crates pulled in when depending on Tokio. Here is an opportunity to discuss an alternative strategy. By doing this RFC, users who are happy with the current situation may express this.

Summary

Do not maintain tokio-* sub crates, instead all Tokio code will exist in a single tokio crate and components are enabled or disabled using feature flags.

For example, depending on only the timer functionality could be done with:

tokio = { version = "0.2.0", default-features = false, features = [ "timer" ] }

By default, tokio would have the same components enabled as it does today.

Motivation

Maintaining a large number of crates comes with an increased maintainership burden. Maintaining correct dependencies between crates is complex. Users feel that large number of dependencies == bloat. Additional rational can be found here.

Details

Tokio must maintain semver stability of its core APIs. This includes traits as well as some types, such as TcpStream. Tokio would like to be able to release breaking changes to less fundamental APIs without having to break the entire Tokio ecosystem.

Currently, Tokio achieves this goal by breaking up all the various components into individual crates. Doing this allows less stable components to release breaking changes without touching stable components. However, this strategy has drawbacks (see Motivation section).

In this proposal, all Tokio components would be moved into a single crate. Each component would have an associated feature flag, similar to how Tokio does it today.

Not much would change for application developers, they would still just depend on tokio and enable / disable feature flags as needed. Library developers would no longer depend on sub crates. Instead, they would depend on tokio and only pull in the features that they need.

Type stability

Core types can maintain stability between breaking semver releases. For example, if the TcpStream type does not change between Tokio version 0.2 and Tokio version 0.3, then the following steps would be taken to release 0.3:

• Release tokio 0.3
• Update tokio 0.2 to depend on tokio 0.3.
• Replace the implementation of TcpStream in 0.3 by re-exporting the implementation from 0.3.
• Release a new patch version for 0.2 including the re-exported TcpStream type from 0.3.

By doing this, TcpStream from 0.2 and 0.3 are the same type.

Drawbacks

• The breaking change release process becomes more complicated as all untouched types must be re-exported in the old version.
• If a user does not update the patch 0.2 patch release in the above scenario, they can end up with both 0.2 and 0.3.

Alternatives

Continue to release new crates for each component.

This introduces a couple of new types:

• tokio::net::windows::NamedPipe
• tokio::net::windows::NamedPipeBuilder
• tokio::net::windows::NamedPipeClientBuilder
• tokio::net::windows::PipeMode

It is based on a suggestion I did in #3388, which is to try and push the exported types to the minimum low level of abstraction necessary to enable their use. This is also a working proposal for #3511.

It is also based on the API proposed by @Darksonn here. With some removals and changes.

In particular I try to closely as possible:

• Provide as close to 1:1 mapping between the exported functions and their corresponding system calls, each documented with a link to msdn.
• Where it's supported and makes sense, make those functions async. WaitNamedPipe unfortunately only provides a blocking API, so it uses the asyncify approach that tokio::fs does and blocks in a worker thread (see comment).

This provides two builders, because named pipe clients and servers are constructed differently. Clients through CreateFile, and servers (the one creating the named pipe) through CreateNamedPipe. I decided to call the one creating the named pipe NamedPipeBuilder, all though naming is preliminary.

For how to use, see the included documentation and examples.

Motivation

Providing named pipes has been stuck for a while, because the current proposal implements a high level model. This PR ports the part from it into a lower level NamedPipe type which can either be used directly or to provide higher level APIs like the one suggested in #3388. Once this lands, it can be done external to the project where it can be more easily experimented with, like tokio-util.

Solution

This tries to implement the bare minimum async wrapping necessary to use named pipes while providing a minimum level of ergonomics and sanity checking.

Note that even if we don't end up going with exporting low level APIs like this, hopefully this can be used as a cleaner internal API to build something else.

Builders taking self

I modified the builders to take self at one point instead of &mut self, because the following pattern seems more common when building named pipes:

let server_builder = NamedPipeBuilder::new("\\\\.pipe\\mypipe")
    .first_pipe_instance(true);

// somewhere else

loop {
    let pipe = server_builder.create()?;
    pipe.connect().await?;
}

And if it returned &mut Self, it wouldn't be as convenient to use the functional style. But feel free to provide your input.

Is your feature request related to a problem? Please describe.

Currently, when developers compile their tokio applications to Wasm, the applications are going to fail at runtime due to Wasm's lack of support for standard networking APIs. We would like to implement a feature flag for tokio to support the wasm32-wasi compiler target -- specifically for the WasmEdge Runtime.

Describe the solution you'd like

We would like to start an effort to merge our fork upstream to tokio's main repo.

Describe alternatives you've considered

Please see the background and our research below.

Additional context

One of the emerging use cases of WebAssembly (Wasm) is a lightweight and secure alternative to Linux containers. Leading container management tools, such as Docker, containerd, Podman / crun, and Kubernetes, have all supported Wasm containers, and in particular WasmEdge, for this reason.

However, to run containerized microservices and serverless functions in Wasm, we need advanced networking sockets in the Wasm runtime. By “advanced”, I mean non-blocking and async sockets. Ideally, we would have a Wasm compiler target in tokio so that when tokio applications are compiled for the wasm32-wasi target, it will utilize Wasm socket APIs. Yet, the official Wasm standards have been slow to incorporate sockets.

To support users and customers who wish to use tokio and Rust networking apps in Wasm today, the WasmEdge community decided to create its own libc-like socket API in Wasm. WasmEdge is the default Wasm runtime distributed with Docker Desktop and Fedora / Red Hat EPEL. It's networking sockets API can reach a wide developer audience.

We forked tokio to support the WasmEdge socket API when compiling to wasm32-wasi. The approach has gained some tractions:

Example 1: This is a popular microservice template for an HTTP service backed by a MySQL server. Both the HTTP server (using forked hyper) and MySQL client (using forked mysql_async) in the WasmEdge app are based on the tokio fork.

https://github.com/second-state/microservice-rust-mysql

Example 2: This is a port of the popular Dapr SDK to Wasm. It utilizes (a forked) request on top of the forked tokio to perform HTTP API calls.

https://github.com/second-state/dapr-sdk-wasi https://github.com/second-state/dapr-wasm

Forking tokio is only a temporary solution to demonstrate demands for this type of application. Now we would like to merge the fork back into tokio so that downstream crates that depend on tokio could also benefit.

Specially, we would like to propose the following:

• Add a feature flag wasmedge
• When the feature is turned on AND the compiler target is wasm32-wasi, tokio will use WasmEdge sockets.

Please let me know your thoughts and happy new year!

Motivation

The documentation of steal_count does not match what is actually counted.

Solution

Fix the documentation of steal_count. Additionally, add steal_operations as a metric to track what steal_count was originally documented to count. Having both metrics allows us to see some effects of different stealing policies on work-stealing.

Closes #5281

Version

$ cargo tree | grep tokio
benches v0.0.0 (/home/ivan/tmp/envoy/tokio/benches)
└── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio)
    └── tokio-macros v1.8.2 (proc-macro) (/home/ivan/tmp/envoy/tokio/tokio-macros)
        └── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
    ├── tokio-stream v0.1.11 (/home/ivan/tmp/envoy/tokio/tokio-stream)
    │   └── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
    │   ├── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
    │   └── tokio-test v0.4.2 (/home/ivan/tmp/envoy/tokio/tokio-test)
    │       ├── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
    │       └── tokio-stream v0.1.11 (/home/ivan/tmp/envoy/tokio/tokio-stream) (*)
    │       └── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
    └── tokio-test v0.4.2 (/home/ivan/tmp/envoy/tokio/tokio-test) (*)
├── tokio-stream v0.1.11 (/home/ivan/tmp/envoy/tokio/tokio-stream) (*)
└── tokio-util v0.7.4 (/home/ivan/tmp/envoy/tokio/tokio-util)
    ├── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
    ├── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
    ├── tokio-stream v0.1.11 (/home/ivan/tmp/envoy/tokio/tokio-stream) (*)
    └── tokio-test v0.4.2 (/home/ivan/tmp/envoy/tokio/tokio-test) (*)
examples v0.0.0 (/home/ivan/tmp/envoy/tokio/examples)
├── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
├── tokio-stream v0.1.11 (/home/ivan/tmp/envoy/tokio/tokio-stream) (*)
├── tokio-util v0.7.4 (/home/ivan/tmp/envoy/tokio/tokio-util) (*)
stress-test v0.1.0 (/home/ivan/tmp/envoy/tokio/stress-test)
└── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
tests-build v0.1.0 (/home/ivan/tmp/envoy/tokio/tests-build)
tests-integration v0.1.0 (/home/ivan/tmp/envoy/tokio/tests-integration)
└── tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
tokio v1.23.0 (/home/ivan/tmp/envoy/tokio/tokio) (*)
tokio-macros v1.8.2 (proc-macro) (/home/ivan/tmp/envoy/tokio/tokio-macros) (*)
tokio-stream v0.1.11 (/home/ivan/tmp/envoy/tokio/tokio-stream) (*)
tokio-test v0.4.2 (/home/ivan/tmp/envoy/tokio/tokio-test) (*)
tokio-util v0.7.4 (/home/ivan/tmp/envoy/tokio/tokio-util) (*)

Platform

$ uname -a
Linux ivan-ThinkPad-T470 5.4.0-135-generic #152-Ubuntu SMP Wed Nov 23 20:19:22 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux

Description tinyhttp example panics against HTTP requests

running as :

$ RUST_BACKTRACE=1 cargo run --example tinyhttp
    Finished dev [unoptimized + debuginfo] target(s) in 0.08s
     Running `target/debug/examples/tinyhttp`
Listening on: 127.0.0.1:8080
thread 'tokio-runtime-worker' panicked at 'attempt to subtract with overflow', examples/tinyhttp.rs:172:29
stack backtrace:
   0: rust_begin_unwind
             at /rustc/897e37553bba8b42751c67658967889d11ecd120/library/std/src/panicking.rs:584:5
   1: core::panicking::panic_fmt
             at /rustc/897e37553bba8b42751c67658967889d11ecd120/library/core/src/panicking.rs:142:14
   2: core::panicking::panic
             at /rustc/897e37553bba8b42751c67658967889d11ecd120/library/core/src/panicking.rs:48:5
   3: <tinyhttp::Http as tokio_util::codec::decoder::Decoder>::decode::{{closure}}
             at ./examples/tinyhttp.rs:172:29
   4: <tinyhttp::Http as tokio_util::codec::decoder::Decoder>::decode
             at ./examples/tinyhttp.rs:184:17
   5: <tokio_util::codec::framed_impl::FramedImpl<T,U,R> as futures_core::stream::Stream>::poll_next
             at ./tokio-util/src/codec/framed_impl.rs:203:38
   6: <tokio_util::codec::framed::Framed<T,U> as futures_core::stream::Stream>::poll_next
             at ./tokio-util/src/codec/framed.rs:301:9
   7: <&mut S as futures_core::stream::Stream>::poll_next
             at /home/ivan/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-core-0.3.25/src/stream.rs:104:9
   8: <tokio_stream::stream_ext::next::Next<St> as core::future::future::Future>::poll
             at ./tokio-stream/src/stream_ext/next.rs:42:9
   9: tinyhttp::process::{{closure}}
             at ./examples/tinyhttp.rs:49:47
...

curl request:

$ curl -v localhost:8080/plaintext
*   Trying 127.0.0.1:8080...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET /plaintext HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.68.0
> Accept: */*
> 
* Empty reply from server
* Connection #0 to host localhost left intact
curl: (52) Empty reply from server

Version

$ cargo tree | grep tokio | sed -e 's/.*\(tokio.*v[^ ]*\).*/\1/' | sort | uniq
tokio-io-timeout v1.2.0
tokio-macros v1.8.2
tokio-serial v5.4.4
tokio-stream v0.1.11
tokio-util v0.7.4
tokio v1.23.0

Platform

$ uname -a
Linux rk-dell 5.15.0-56-generic #62-Ubuntu SMP Tue Nov 22 19:54:14 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux

Description When reading from a serial port using AsyncFd in O_NONBLOCK mode, an extra read() call is performed:

read(16, "1", 3013)                     = 1
read(16, 0x55cc4763f34c, 3012)          = -1 EAGAIN (Resource temporarily unavailable)
epoll_wait(3, [{events=EPOLLIN|EPOLLOUT, data={u32=2, u64=2}}], 1024, -1) = 1

One can assume that if the buffer didn't fill up, the following read won't succeed, so it should proceed directly to epoll_wait instead of attempting the read and having it fail. Note that on some serial devices, they will never return more than a smaller buffer (say 64) so I would encourage this optimization to only take place when a single character is read.

This was fixed for non-AsyncFd reads here: tokio-rs PR-4970 tokio-rs PR-4840

Currently, we provide the UnixStream type for reading from a unix domain socket on Linux, but Linux also provides pipes. Domain sockets and pipes are not the same (see here). Furthermore, opening a named pipe with UnixStream::connect will fail with an error. Currently the only way to asynchronously read from a pipe is to use AsyncFd.

I would like to see a type in tokio::net for use with fifo pipes.