tracing Documentation

1. Project Overview & Quickstart (tokio-rs/tracing)

File: README.md

Tracing — Structured, application-level diagnostics

Overview

tracing is a framework for instrumenting Rust programs to collect
structured, event-based diagnostic information. tracing is maintained by the
Tokio project, but does not require the tokio runtime to be used.

Branch set-up

main - Default branch, crates.io releases are done from this branch. This was previously the
v0.1.x branch.
v0.2.x - Branch containing the as-yet unreleased 0.2 version of tracing-core, tracing, and
all the other tracing crates that depend on these versions. This was previously the master
branch.

Usage

In Applications

In order to record trace events, executables have to use a Subscriber
implementation compatible with tracing. A Subscriber implements a way of
collecting trace data, such as by logging it to standard output.
tracing-subscriber's fmt module provides
a subscriber for logging traces with reasonable defaults. Additionally,
tracing-subscriber is able to consume messages emitted by log-instrumented
libraries and modules.

To use tracing-subscriber, add the following to your Cargo.toml:

toml

[dependencies]
tracing = "0.1"
tracing-subscriber = "0.3"

Then create and install a Subscriber, for example using init():

rust

use tracing::info;
use tracing_subscriber;

fn main() {
    // install global subscriber configured based on RUST_LOG envvar.
    tracing_subscriber::fmt::init();

    let number_of_yaks = 3;
    // this creates a new event, outside of any spans.
    info!(number_of_yaks, "preparing to shave yaks");

    let number_shaved = yak_shave::shave_all(number_of_yaks);
    info!(
        all_yaks_shaved = number_shaved == number_of_yaks,
        "yak shaving completed."
    );
}

Using init() calls set_global_default() so this subscriber will be used
as the default in all threads for the remainder of the duration of the
program, similar to how loggers work in the log crate.

For more control, a subscriber can be built in stages and not set globally,
but instead used to locally override the default subscriber. For example:

rust

use tracing::{info, Level};
use tracing_subscriber;

fn main() {
    let subscriber = tracing_subscriber::fmt()
        // filter spans/events with level TRACE or higher.
        .with_max_level(Level::TRACE)
        // build but do not install the subscriber.
        .finish();

    tracing::subscriber::with_default(subscriber, || {
        info!("This will be logged to stdout");
    });
    info!("This will _not_ be logged to stdout");
}

Any trace events generated outside the context of a subscriber will not be collected.

This approach allows trace data to be collected by multiple subscribers
within different contexts in the program. Note that the override only applies to the
currently executing thread; other threads will not see the change from with_default.

Once a subscriber has been set, instrumentation points may be added to the
executable using the tracing crate's macros.

In Libraries

Libraries should only rely on the tracing crate and use the provided macros
and types to collect whatever information might be useful to downstream consumers.

text

/* Detailed source-code truncated for AI context efficiency. */

toml

[dependencies]
tracing = "0.1"

Note: Libraries should NOT install a subscriber by using a method that calls
set_global_default(), as this will cause conflicts when executables try to
set the default later.

In Asynchronous Code

To trace async fns, the preferred method is using the #[instrument] attribute:

rust

use tracing::{info, instrument};
use tokio::{io::AsyncWriteExt, net::TcpStream};
use std::io;

#[instrument]
async fn write(stream: &mut TcpStream) -> io::Result {
    let result = stream.write(b"hello world\n").await;
    info!("wrote to stream; success={:?}", result.is_ok());
    result
}

Special handling is needed for the general case of code using
std::future::Future or blocks with async/await, as the
following example will not work:

rust

async {
    let _s = span.enter();
    // ...
}

The span guard _s will not exit until the future generated by the async block is complete.
Since futures and spans can be entered and exited multiple times without them completing,
the span remains entered for as long as the future exists, rather than being entered only when
it is polled, leading to very confusing and incorrect output.
For more details, see the documentation on closing spans.

This problem can be solved using the Future::instrument combinator:

rust

use tracing::Instrument;

let my_future = async {
    // ...
};

my_future
    .instrument(tracing::info_span!("my_future"))
    .await

Future::instrument attaches a span to the future, ensuring that the span's lifetime
is as long as the future's.

Under the hood, the #[instrument] macro performs the same explicit span
attachment that Future::instrument does.

Supported Rust Versions

Tracing is built against the latest stable release. The minimum supported
version is 1.65. The current Tracing version is not guaranteed to build on Rust
versions earlier than the minimum supported version.

Tracing follows the same compiler support policies as the rest of the Tokio
project. The current stable Rust compiler and the three most recent minor
versions before it will always be supported. For example, if the current stable
compiler version is 1.69, the minimum supported version will not be increased
past 1.66, three minor versions prior. Increasing the minimum supported compiler
version is not considered a semver breaking change as long as doing so complies
with this policy.

Getting Help

First, see if the answer to your question can be found in the API documentation.
If the answer is not there, there is an active community in
the Tracing Discord channel. We would be happy to try to answer your
question. Last, if that doesn't work, try opening an issue with the question.

Contributing

:balloon: Thanks for your help improving the project! We are so happy to have
you! We have a contributing guide to help you get involved in the Tracing
project.

Project layout

The tracing crate contains the primary instrumentation API, used for
instrumenting libraries and applications to emit trace data. The tracing-core
crate contains the core API primitives on which the rest of tracing is
instrumented. Authors of trace subscribers may depend on tracing-core, which
guarantees a higher level of stability.

Additionally, this repository contains several compatibility and utility
libraries built on top of tracing. Some of these crates are in a pre-release
state, and are less stable than the tracing and tracing-core crates.

The crates included as part of Tracing are:

tracing-futures: Utilities for instrumenting futures.
(crates.io|docs)
tracing-macros: Experimental macros for emitting trace events (unstable).
tracing-attributes: Procedural macro attributes for automatically
instrumenting functions. (crates.io|docs)
tracing-log: Compatibility with the log crate (unstable).
tracing-serde: A compatibility layer for serializing trace data with
serde (unstable).
tracing-subscriber: Subscriber implementations, and utilities for
implementing and composing Subscribers.
(crates.io|docs)
tracing-tower: Compatibility with the tower ecosystem (unstable).
tracing-appender: Utilities for outputting tracing data, including a file appender
and non-blocking writer. (crates.io|docs)
tracing-error: Provides SpanTrace, a type for instrumenting errors with
tracing spans. (crates.io|docs)
tracing-flame: Provides a layer for generating flame graphs based on
tracing span entry / exit events. (crates.io|docs)
tracing-journald: Provides a layer for recording events to the
Linux journald service, preserving structured data. (crates.io|docs)

In addition to this repository, here are also several third-party crates which
are not maintained by the tokio project. These include:

tracing-timing implements inter-event timing metrics on top of tracing.
It provides a subscriber that records the time elapsed between pairs of
tracing events and generates histograms.
tracing-honeycomb Provides a layer that reports traces spanning multiple machines to honeycomb.io. Backed by tracing-distributed.
tracing-distributed Provides a generic implementation of a layer that reports traces spanning multiple machines to some backend.
tracing-actix-web provides tracing integration for the actix-web web framework.
tracing-actix provides tracing integration for the actix actor
framework.
axum-insights provides tracing integration and Application insights export for the axum web framework.
tracing-gelf implements a subscriber for exporting traces in Graylog
GELF format.
tracing-coz provides integration with the coz causal profiler
(Linux-only).
tracing-bunyan-formatter provides a layer implementation that reports events and spans in bunyan format, enriched with timing information.
tide-tracing provides a tide middleware to trace all incoming requests and responses.
color-spantrace provides a formatter for rendering span traces in the
style of color-backtrace
color-eyre provides customized panic and eyre report handlers for
eyre::Report for capturing span traces and backtraces with new errors and
pretty printing them.
spandoc provides a proc macro for constructing spans from doc comments
inside of functions.
tracing-wasm provides a Subscriber/Layer implementation that reports
events and spans via browser console.log and User Timing API (window.performance).
tracing-web provides a layer implementation of level-aware logging of events
to web browsers' console.* and span events to the User Timing API (window.performance).
test-log takes care of initializing tracing for tests, based on
environment variables with an env_logger compatible syntax.
tracing-unwrap provides convenience methods to report failed unwraps on Result or Option types to a Subscriber.
diesel-tracing provides integration with diesel database connections.
tracing-tracy provides a way to collect Tracy profiles in instrumented
applications.
tracing-elastic-apm provides a layer for reporting traces to Elastic APM.
tracing-etw provides a layer for emitting Windows ETW events.
sentry-tracing provides a layer for reporting events and traces to Sentry.
tracing-forest provides a subscriber that preserves contextual coherence by
grouping together logs from the same spans during writing.
tracing-loki provides a layer for shipping logs to Grafana Loki.
tracing-logfmt provides a layer that formats events and spans into the logfmt format.
tracing-chrome provides a layer that exports trace data that can be viewed in chrome://tracing.
reqwest-tracing provides a middleware to trace reqwest HTTP requests.
tracing-cloudwatch provides a layer that sends events to AWS CloudWatch Logs.
tracing-subscriber-reload-arcswap provides a lock-free alternative to tracing_subscriber::reload::Layer using arc-swap.
clippy-tracing provides a tool to add, remove and check for tracing::instrument.

(if you're the maintainer of a tracing ecosystem crate not in this list,
please let us know!)

Note: that some of the ecosystem crates are currently unreleased and
undergoing active development. They may be less stable than tracing and
tracing-core.

External Resources

This is a list of links to blog posts, conference talks, and tutorials about
Tracing.

Blog Posts

Diagnostics with Tracing on the Tokio blog, August 2019
Production-Grade Logging in Rust Applications, November 2020
Custom Logging in Rust using tracing and tracing-subscriber, part 1 and part 2, October 2021
Instrumenting Axum projects, August 2023

Talks

Bay Area Rust Meetup talk and Q&A, March 2019
RustConf 2019 talk and slides, August 2019
Are we observable yet? @ RustyDays talk and slides, August 2020
Crabs with instruments!, September 2021
Decrusting the tracing crate by Jon Gjengset, February 2024

Help us expand this list! If you've written or spoken about Tracing, or
know of resources that aren't listed, please open a pull request adding them.

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tracing by you, shall be licensed as MIT, without any additional
terms or conditions.

File: examples/README.md

Tracing Examples

This directory contains a collection of examples that demonstrate the use of the
tracing ecosystem:

tracing:
- counters: Implements a very simple metrics system to demonstrate how
  subscribers can consume field values as typed data.
- sloggish: A demo Subscriber implementation that mimics the output of
  slog-term's Compact formatter.
tracing-attributes:
- attrs-basic: A simple example of the #[instrument] attribute.
- attrs-literal-field-names: Demonstrates using literal field names rather
  than rust tokens..
- attrs-args: An example implementing a simple recursive calculation of
  Fibonacci numbers, to demonstrate how the #[instrument] attribute can
  record function arguments.
tracing-subscriber:
- fmt: Demonstrates the use of the fmt module in tracing-subscriber,
  which provides a subscriber implementation that logs traces to the console.
  This example uses the default formatter (format::Full).
- fmt-stderr: Demonstrates overriding the output stream used by the fmt
  subscriber.
- fmt-pretty: Demonstrates the output of the format::Pretty formatter in
  tracing-subscriber::fmt.
- fmt-compact: Demonstrates the output of the format::Compact formatter in
  tracing-subscriber::fmt.
- fmt-json: Demonstrates the output of the format::Json formatter in
  tracing-subscriber::fmt.
- fmt-custom-field: Demonstrates overriding how the fmt subscriber formats
  fields on spans and events.
- fmt-custom-event: Demonstrates overriding how the fmt subscriber formats
  events.
- fmt-multiple-writers: demonstrates how fmt::Layer can write
  to multiple destinations (in this instance, stdout and a file) simultaneously.
- fmt-source-locations: demonstrates displaying source code locations
  with fmt::Layer.
- subscriber-filter: Demonstrates the tracing-subscriber::filter module,
  which provides a layer which adds configurable filtering to a subscriber
  implementation.
- tower-load: Demonstrates how dynamically reloadable filters can be used to
  debug a server under load in production.
- journald: Demonstrates how to use fmt and journald layers to output to
  both the terminal and the system journal.
- toggle-layers : Demonstrates how Layers can be wrapped with an Option allowing
  them to be dynamically toggled.
tracing-futures:
- spawny-thing: Demonstrates the use of the #[instrument] attribute macro
  asynchronous functions.
- tokio-spawny-thing.rs: Similar to spawny-thingy, but with the additional
  demonstration instrumenting concurrent tasks created with
  tokio::spawn.
- futures-proxy-server: Demonstrates the use of tracing-futures by
  implementing a simple proxy server, based on this example
  from tokio.
- async-fn: Demonstrates how asynchronous functions can be
  instrumented.
- echo: Demonstrates a tracing-instrumented variant of Tokio's echo example.
tracing-flame:
- inferno-flame: Demonstrates the use of tracing-flame to generate a flamegraph
  from spans.
tracing-tower:
- tower-client: Demonstrates the use of tracing-tower to instrument a
  simple tower HTTP/1.1 client.
- tower-server: Demonstrates the use of tracing-tower to instrument a
  simple tower HTTP/1.1 server.
tracing-serde:
- serde-yak-shave: Demonstrates the use of tracing-serde by implementing a
  subscriber that emits trace output as JSON.
tracing-log:
- hyper-echo: Demonstrates how tracing-log can be used to record
  unstructured logs from dependencies as tracing events, by instrumenting
  this example from hyper, and using tracing-log to record logs
  emitted by hyper.

File: tracing/README.md

Tracing — Structured, application-level diagnostics

tracing

Application-level tracing for Rust.

Documentation | Chat

Overview

tracing is a framework for instrumenting Rust programs to collect
structured, event-based diagnostic information.

In asynchronous systems like Tokio, interpreting traditional log messages can
often be quite challenging. Since individual tasks are multiplexed on the same
thread, associated events and log lines are intermixed making it difficult to
trace the logic flow. tracing expands upon logging-style diagnostics by
allowing libraries and applications to record structured events with additional
information about temporality and causality — unlike a log message, a span
in tracing has a beginning and end time, may be entered and exited by the
flow of execution, and may exist within a nested tree of similar spans. In
addition, tracing spans are structured, with the ability to record typed
data as well as textual messages.

The tracing crate provides the APIs necessary for instrumenting libraries
and applications to emit trace data.

Compiler support: requires rustc 1.65+

Usage

(The examples below are borrowed from the log crate's yak-shaving
example, modified to
idiomatic tracing.)

In Applications

In order to record trace events, executables have to use a Subscriber
implementation compatible with tracing. A Subscriber implements a way of
collecting trace data, such as by logging it to standard output. tracing_subscriber's
fmt module provides reasonable defaults.
Additionally, tracing-subscriber is able to consume messages emitted by log-instrumented libraries and modules.

The simplest way to use a subscriber is to call the set_global_default function.

rust

use tracing::{info, Level};
use tracing_subscriber::FmtSubscriber;

fn main() {
    // a builder for `FmtSubscriber`.
    let subscriber = FmtSubscriber::builder()
        // all spans/events with a level higher than TRACE (e.g, debug, info, warn, etc.)
        // will be written to stdout.
        .with_max_level(Level::TRACE)
        // completes the builder.
        .finish();

    tracing::subscriber::set_global_default(subscriber)
        .expect("setting default subscriber failed");

    let number_of_yaks = 3;
    // this creates a new event, outside of any spans.
    info!(number_of_yaks, "preparing to shave yaks");

    let number_shaved = yak_shave::shave_all(number_of_yaks);
    info!(
        all_yaks_shaved = number_shaved == number_of_yaks,
        "yak shaving completed."
    );
}

toml

[dependencies]
tracing = "0.1"
tracing-subscriber = "0.3.0"

This subscriber will be used as the default in all threads for the remainder of the duration
of the program, similar to how loggers work in the log crate.

In addition, you can locally override the default subscriber. For example:

rust

use tracing::{info, Level};
use tracing_subscriber::FmtSubscriber;

fn main() {
    let subscriber = tracing_subscriber::FmtSubscriber::builder()
        // all spans/events with a level higher than TRACE (e.g, debug, info, warn, etc.)
        // will be written to stdout.
        .with_max_level(Level::TRACE)
        // builds the subscriber.
        .finish();

    tracing::subscriber::with_default(subscriber, || {
        info!("This will be logged to stdout");
    });
    info!("This will _not_ be logged to stdout");
}

Any trace events generated outside the context of a subscriber will not be collected.

Once a subscriber has been set, instrumentation points may be added to the
executable using the tracing crate's macros.

In Libraries

Libraries should only rely on the tracing crate and use the provided macros
and types to collect whatever information might be useful to downstream consumers.

text

/* Detailed source-code truncated for AI context efficiency. */

toml

[dependencies]
tracing = "0.1"

Note: Libraries should NOT call set_global_default(), as this will cause
conflicts when executables try to set the default later.

In Asynchronous Code

If you are instrumenting code that make use of
std::future::Future
or async/await, avoid using the Span::enter method. The following example
will not work:

rust

async {
    let _s = span.enter();
    // ...
}

rust

async {
    let _s = tracing::span!(...).entered();
    // ...
}

There are two ways to instrument asynchronous code. The first is through the
Future::instrument combinator:

rust

use tracing::Instrument;

let my_future = async {
    // ...
};

my_future
    .instrument(tracing::info_span!("my_future"))
    .await

Future::instrument attaches a span to the future, ensuring that the span's lifetime
is as long as the future's.

The second, and preferred, option is through the
#[instrument]
attribute:

rust

use tracing::{info, instrument};
use tokio::{io::AsyncWriteExt, net::TcpStream};
use std::io;

#[instrument]
async fn write(stream: &mut TcpStream) -> io::Result {
    let result = stream.write(b"hello world\n").await;
    info!("wrote to stream; success={:?}", result.is_ok());
    result
}

Under the hood, the #[instrument] macro performs the same explicit span
attachment that Future::instrument does.

Concepts

This crate provides macros for creating Spans and Events, which represent
periods of time and momentary events within the execution of a program,
respectively.

As a rule of thumb, spans should be used to represent discrete units of work
(e.g., a given request's lifetime in a server) or periods of time spent in a
given context (e.g., time spent interacting with an instance of an external
system, such as a database). In contrast, events should be used to represent
points in time within a span — a request returned with a given status code,
n new items were taken from a queue, and so on.

Spans are constructed using the span! macro, and then entered
to indicate that some code takes place within the context of that Span:

rust

use tracing::{span, Level};

// Construct a new span named "my span".
let mut span = span!(Level::INFO, "my span");
span.in_scope(|| {
    // Any trace events in this closure or code called by it will occur within
    // the span.
});
// Dropping the span will close it, indicating that it has ended.

The #[instrument] attribute macro
can reduce some of this boilerplate:

rust

use tracing::{instrument};

#[instrument]
pub fn my_function(my_arg: usize) {
    // This event will be recorded inside a span named `my_function` with the
    // field `my_arg`.
    tracing::info!("inside my_function!");
    // ...
}

The Event type represent an event that occurs instantaneously, and is
essentially a Span that cannot be entered. They are created using the event!
macro:

rust

use tracing::{event, Level};

event!(Level::INFO, "something has happened!");

Users of the log crate should note that tracing exposes a set of macros for
creating Events (trace!, debug!, info!, warn!, and error!) which may
be invoked with the same syntax as the similarly-named macros from the log
crate. Often, the process of converting a project to use tracing can begin
with a simple drop-in replacement.

Supported Rust Versions

Tracing is built against the latest stable release. The minimum supported
version is 1.42. The current Tracing version is not guaranteed to build on Rust
versions earlier than the minimum supported version.

Tracing follows the same compiler support policies as the rest of the Tokio
project. The current stable Rust compiler and the three most recent minor
versions before it will always be supported. For example, if the current stable
compiler version is 1.45, the minimum supported version will not be increased
past 1.42, three minor versions prior. Increasing the minimum supported compiler
version is not considered a semver breaking change as long as doing so complies
with this policy.

Ecosystem

In addition to tracing and tracing-core, the tokio-rs/tracing repository
contains several additional crates designed to be used with the tracing ecosystem.
This includes a collection of Subscriber implementations, as well as utility
and adapter crates to assist in writing Subscribers and instrumenting
applications.

In particular, the following crates are likely to be of interest:

tracing-futures provides a compatibility layer with the futures
crate, allowing spans to be attached to Futures, Streams, and Executors.
tracing-subscriber provides Subscriber implementations and
utilities for working with Subscribers. This includes a FmtSubscriber
FmtSubscriber for logging formatted trace data to stdout, with similar
filtering and formatting to the env_logger crate.
tracing-log provides a compatibility layer with the log crate,
allowing log messages to be recorded as tracing Events within the
trace tree. This is useful when a project using tracing have
dependencies which use log. Note that if you're using
tracing-subscriber's FmtSubscriber, you don't need to depend on
tracing-log directly.

Additionally, there are also several third-party crates which are not
maintained by the tokio project. These include:

tracing-timing implements inter-event timing metrics on top of tracing.
It provides a subscriber that records the time elapsed between pairs of
tracing events and generates histograms.
tracing-opentelemetry provides a subscriber for emitting traces to
OpenTelemetry-compatible distributed tracing systems.
tracing-honeycomb Provides a layer that reports traces spanning multiple machines to honeycomb.io. Backed by tracing-distributed.
tracing-distributed Provides a generic implementation of a layer that reports traces spanning multiple machines to some backend.
tracing-actix provides tracing integration for the actix actor
framework.
axum-insights provides tracing integration and Application insights export for the axum web framework.
tracing-gelf implements a subscriber for exporting traces in Greylog
GELF format.
tracing-coz provides integration with the coz causal profiler
(Linux-only).
test-log takes care of initializing tracing for tests, based on
environment variables with an env_logger compatible syntax.
tracing-unwrap provides convenience methods to report failed unwraps on Result or Option types to a Subscriber.
diesel-tracing provides integration with diesel database connections.
tracing-tracy provides a way to collect Tracy profiles in instrumented
applications.
tracing-elastic-apm provides a layer for reporting traces to Elastic APM.
tracing-etw provides a layer for emitting Windows ETW events.
tracing-fluent-assertions provides a fluent assertions-style testing
framework for validating the behavior of tracing spans.
sentry-tracing provides a layer for reporting events and traces to Sentry.
tracing-loki provides a layer for shipping logs to Grafana Loki.
tracing-logfmt provides a layer that formats events and spans into the logfmt format.
json-subscriber provides a layer for emitting JSON logs. The output can be customized much more than with FmtSubscriber's JSON output.

If you're the maintainer of a tracing ecosystem crate not listed above,
please let us know! We'd love to add your project to the list!

Note: that some of the ecosystem crates are currently unreleased and
undergoing active development. They may be less stable than tracing and
tracing-core.

Supported Rust Versions

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tokio by you, shall be licensed as MIT, without any additional
terms or conditions.

File: tracing-appender/README.md

Tracing — Structured, application-level diagnostics

tracing-appender

Writers for logging events and spans

Documentation | Chat

Overview

tracing is a framework for instrumenting Rust programs to
collect structured, event-based diagnostic information. tracing-appender
allows events and spans to be recorded in a non-blocking manner through a
dedicated logging thread. It also provides a RollingFileAppender
that can be used with or without the non-blocking writer.

Compiler support: requires rustc 1.63+

Usage

Add the following to your Cargo.toml:

toml

tracing-appender = "0.2"

This crate can be used in a few ways to record spans/events:

Using a RollingFileAppender to write to a log file.
This is a blocking operation.
Using any type implementing std::io::Write in a
non-blocking fashion.
Using NonBlocking and RollingFileAppender
together to write to log files in a non-blocking fashion.

Rolling File Appender

rust

fn main(){
    let file_appender = tracing_appender::rolling::hourly("/some/directory", "prefix.log");
}

This creates an hourly rotating file appender that writes to
/some/directory/prefix.log.YYYY-MM-DD-HH. [Rotation::DAILY] and
[Rotation::NEVER] are the other available options.

The file appender implements std::io::Write. To be used with
tracing_subscriber::FmtSubscriber, it must be combined
with a MakeWriter implementation to be able to record
tracing spans/event.

The rolling module's documentation provides more detail on how to use
this file appender.

Non-Blocking Writer

The example below demonstrates the construction of a non_blocking writer
with an implementation of std::io::Writer.

rust

use std::io::Error;

struct TestWriter;

impl std::io::Write for TestWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result {
        let buf_len = buf.len();
    
        println!("{:?}", buf);
        Ok(buf_len)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

fn main() {
    let (non_blocking, _guard) = tracing_appender::non_blocking(TestWriter);
    tracing_subscriber::fmt().with_writer(non_blocking).init();
}

Note: _guard is a WorkerGuard which is returned by
tracing_appender::non_blocking to ensure buffered logs are flushed to
their output in the case of abrupt terminations of a process. See
WorkerGuard module for more details.

The example below demonstrates the construction of a
tracing_appender::non_blocking writer constructed with
a std::io::Write:

rust

fn main() {
    let (non_blocking, _guard) = tracing_appender::non_blocking(std::io::stdout());
    tracing_subscriber::fmt()
        .with_writer(non_blocking)
        .init();
}

The non_blocking module's documentation provides more detail on how to
use non_blocking.

Non-Blocking Rolling File Appender

rust

fn main() {
    let file_appender = tracing_appender::rolling::hourly("/some/directory", "prefix.log");
    let (non_blocking, _guard) = tracing_appender::non_blocking(file_appender);
   tracing_subscriber::fmt()
       .with_writer(non_blocking)
       .init();
}

Supported Rust Versions

tracing-appender is built against the latest stable release. The minimum supported
version is 1.63. The current tracing-appender version is not guaranteed to build on
Rust versions earlier than the minimum supported version.

Tracing follows the same compiler support policies as the rest of the Tokio
project. The current stable Rust compiler and the three most recent minor
versions before it will always be supported. For example, if the current
stable compiler version is 1.69, the minimum supported version will not be
increased past 1.66, three minor versions prior. Increasing the minimum
supported compiler version is not considered a semver breaking change as
long as doing so complies with this policy.

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tokio by you, shall be licensed as MIT, without any additional
terms or conditions.

File: tracing-attributes/README.md

Tracing — Structured, application-level diagnostics

tracing-attributes

Macro attributes for application-level tracing.

Documentation | Chat

Overview

tracing is a framework for instrumenting Rust programs to collect
structured, event-based diagnostic information. This crate provides the
#[instrument] attribute for automatically instrumenting functions using
tracing.

Note that this macro is also re-exported by the main tracing crate.

Compiler support: requires rustc 1.65+

Usage

First, add this to your Cargo.toml:

toml

[dependencies]
tracing-attributes = "0.1.31"

This crate provides the #[instrument] attribute for instrumenting a function
with a tracing span. For example:

rust

use tracing_attributes::instrument;

#[instrument]
pub fn my_function(my_arg: usize) {
    // ...
}

Supported Rust Versions

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tokio by you, shall be licensed as MIT, without any additional
terms or conditions.

File: tracing-core/README.md

Tracing — Structured, application-level diagnostics

tracing-core

Core primitives for application-level tracing.

Documentation | Chat

Overview

tracing is a framework for instrumenting Rust programs to collect
structured, event-based diagnostic information. This crate defines the core
primitives of tracing.

The crate provides:

span::Id identifies a span within the execution of a program.
Event represents a single event within a trace.
Subscriber, the trait implemented to collect trace data.
Metadata and Callsite provide information describing spans and
events.
Field, FieldSet, Value, and ValueSet represent the
structured data attached to spans and events.
Dispatch allows spans and events to be dispatched to Subscribers.

In addition, it defines the global callsite registry and per-thread current
dispatcher which other components of the tracing system rely on.

Compiler support: requires rustc 1.65+

Usage

Application authors will typically not use this crate directly. Instead, they
will use the tracing crate, which provides a much more fully-featured
API. However, this crate's API will change very infrequently, so it may be used
when dependencies must be very stable.

Subscriber implementations may depend on tracing-core rather than tracing,
as the additional APIs provided by tracing are primarily useful for
instrumenting libraries and applications, and are generally not necessary for
Subscriber implementations.

Crate Feature Flags

The following crate feature flags are available:

std: Depend on the Rust standard library (enabled by default).

no_std users may disable this feature with default-features = false:
toml
```
[dependencies]
tracing-core = { version = "0.1.36", default-features = false }
```
Note:tracing-core's no_std support requires liballoc.

Supported Rust Versions

Tracing follows the same compiler support policies as the rest of the Tokio
project. The current stable Rust compiler and the three most recent minor
versions before it will always be supported. For example, if the current stable
compiler version is 1.69, the minimum supported version will not be increased
past 1.69, three minor versions prior. Increasing the minimum supported compiler
version is not considered a semver breaking change as long as doing so complies
with this policy.

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tokio by you, shall be licensed as MIT, without any additional
terms or conditions.

File: tracing-error/README.md

Tracing — Structured, application-level diagnostics

tracing-error

Utilities for enriching error handling with tracing diagnostic
information.

Documentation (release) | Documentation (v0.2.x) | Chat

Overview

tracing is a framework for instrumenting Rust programs to collect
scoped, structured, and async-aware diagnostics. This crate provides
integrations between tracing instrumentation and Rust error handling. It
enables enriching error types with diagnostic information from tracing
span contexts, formatting those contexts when errors are displayed, and
automatically generate tracing events when errors occur.

The crate provides the following:

SpanTrace, a captured trace of the current tracing span context
ErrorLayer, a subscriber layer which enables capturing SpanTraces

Note: This crate is currently experimental.

Compiler support: requires rustc 1.65+

Usage

tracing-error provides the SpanTrace type, which captures the current
tracing span context when it is constructed and allows it to be displayed
at a later time.

For example:

rust

use std::{fmt, error::Error};
use tracing_error::SpanTrace;

#[derive(Debug)]
pub struct MyError {
    context: SpanTrace,
    // ...
}

impl fmt::Display for MyError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        // ... format other parts of the error ...

        self.context.fmt(f)?;

        // ... format other error context information, cause chain, etc ...
        # Ok(())
    }
}

impl Error for MyError {}

impl MyError {
    pub fn new() -> Self {
        Self {
            context: SpanTrace::capture(),
            // ... other error information ...
        }
    }
}

This crate also provides TracedError, for attaching a SpanTrace to an
existing error. The easiest way to wrap errors in TracedError is to either
use the InstrumentResult and InstrumentError traits or the From/Into
traits.

rust

use tracing_error::prelude::*;

std::fs::read_to_string("myfile.txt").in_current_span()?;

Once an error has been wrapped with with a TracedError, the SpanTrace
can be extracted one of three ways: either via TracedError's
Display/Debug implementations, or via the ExtractSpanTrace trait.

For example, here is how one might print the errors but specialize the
printing when the error is a placeholder for a wrapping SpanTrace:

rust

use std::error::Error;
use tracing_error::ExtractSpanTrace as _;

fn print_extracted_spantraces(error: &(dyn Error + 'static)) {
    let mut error = Some(error);
    let mut ind = 0;

    eprintln!("Error:");

    while let Some(err) = error {
        if let Some(spantrace) = err.span_trace() {
            eprintln!("found a spantrace:\n{}", spantrace);
        } else {
            eprintln!("{:>4}: {}", ind, err);
        }

        error = err.source();
        ind += 1;
    }
}

Whereas here, we can still display the content of the SpanTraces without
any special casing by simply printing all errors in our error chain.

rust

use std::error::Error;

fn print_naive_spantraces(error: &(dyn Error + 'static)) {
    let mut error = Some(error);
    let mut ind = 0;

    eprintln!("Error:");

    while let Some(err) = error {
        eprintln!("{:>4}: {}", ind, err);
        error = err.source();
        ind += 1;
    }
}

Applications that wish to use tracing-error-enabled errors should
construct an ErrorLayer and add it to their Subscriber in order to
enable capturing SpanTraces. For example:

rust

use tracing_error::ErrorLayer;
use tracing_subscriber::prelude::*;

fn main() {
    let subscriber = tracing_subscriber::Registry::default()
        // any number of other subscriber layers may be added before or
        // after the `ErrorLayer`...
        .with(ErrorLayer::default());

    // set the subscriber as the default for the application
    tracing::subscriber::set_global_default(subscriber);
}

Feature Flags

traced-error - Enables the TracedError type and related traits
- InstrumentResult and InstrumentError extension traits, which
  provide an in_current_span() method for bundling errors with a
  SpanTrace.
- ExtractSpanTrace extension trait, for extracting SpanTraces from
  behind dyn Error trait objects.

Supported Rust Versions

In addition to this repository, here are also several third-party crates which
are not maintained by the tokio project. These include:

color-spantrace provides a formatter for rendering SpanTrace in the style
of color-backtrace
color-eyre provides a customized version of eyre::Report for capturing
span traces and backtraces with new errors and pretty printing them in error
reports.

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tracing by you, shall be licensed as MIT, without any additional
terms or conditions.

File: tracing-flame/README.md

Tracing — Structured, application-level diagnostics

tracing-flame

A tracing Layer for generating a folded stack trace for generating flamegraphs
and flamecharts with inferno

Documentation | Chat

Overview

tracing is a framework for instrumenting Rust programs to collect
scoped, structured, and async-aware diagnostics. tracing-flame provides helpers
for consuming tracing instrumentation that can later be visualized as a
flamegraph/flamechart. Flamegraphs/flamecharts are useful for identifying performance
bottlenecks in an application. For more details, see Brendan Gregg's post
on flamegraphs.

Compiler support: requires rustc 1.65+

Usage

This crate is meant to be used in a two step process:

Capture textual representation of the spans that are entered and exited
with FlameLayer.
Feed the textual representation into inferno-flamegraph to generate the
flamegraph or flamechart.

Note: when using a buffered writer as the writer for a FlameLayer, it is necessary to
ensure that the buffer has been flushed before the data is passed into
inferno-flamegraph. For more details on how to flush the internal writer
of the FlameLayer, see the docs for FlushGuard.

Layer Setup

rust

use tracing_flame::FlameLayer;
use tracing_subscriber::{prelude::*, fmt};

fn setup_global_subscriber() -> impl Drop {
    let fmt_layer = fmt::Layer::default();

    let (flame_layer, _guard) = FlameLayer::with_file("./tracing.folded").unwrap();

    tracing_subscriber::registry()
        .with(fmt_layer)
        .with(flame_layer)
        .init().
    _guard
}

// your code here ..

As an alternative, you can provide any type that implements std::io::Write to
FlameLayer::new.

Generating the Image

To convert the textual representation of a flamegraph to a visual one, first install inferno:

console

cargo install inferno

Then, pass the file created by FlameLayer into inferno-flamegraph:

console

# flamegraph
cat tracing.folded | inferno-flamegraph > tracing-flamegraph.svg

# flamechart
cat tracing.folded | inferno-flamegraph --flamechart > tracing-flamechart.svg

Differences between `flamegraph`s and `flamechart`s

By default, inferno-flamegraph creates flamegraphs. Flamegraphs operate by
that collapsing identical stack frames and sorting them on the frame's names.

This behavior is great for multithreaded programs and long-running programs
where the same frames occur many times, for short durations, because it reduces
noise in the graph and gives the reader a better idea of the
overall time spent in each part of the application.

However, it is sometimes desirable to preserve the exact ordering of events
as they were emitted by tracing-flame, so that it is clear when each
span is entered relative to others and get an accurate visual trace of
the execution of your program. This representation is best created with a
flamechart, which does not sort or collapse identical stack frames.

Supported Rust Versions

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tracing by you, shall be licensed as MIT, without any additional
terms or conditions.

File: tracing-futures/README.md

Tracing — Structured, application-level diagnostics

tracing-futures

Utilities for instrumenting futures-based code with tracing.

Documentation | Chat

Overview

tracing is a framework for instrumenting Rust programs to collect
structured, event-based diagnostic information. This crate provides utilities
for using tracing to instrument asynchronous code written using futures and
async/await.

The crate provides the following traits:

Instrument allows a tracing span to be attached to a future, sink,
stream, or executor.
WithSubscriber allows a tracing Subscriber to be attached to a
future, sink, stream, or executor.

Compiler support: requires rustc 1.65+

Supported Rust Versions

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tracing by you, shall be licensed as MIT, without any additional
terms or conditions.

File: tracing-journald/README.md

Tracing — Structured, application-level diagnostics

tracing-journald

Support for logging tracing events natively to journald,
preserving structured information.

Overview

tracing is a framework for instrumenting Rust programs to collect
scoped, structured, and async-aware diagnostics. tracing-journald provides a
tracing_subscriber::Layer implementation for logging tracing spans
and events to systemd-journald, on Linux distributions that use
systemd.

Compiler support: requires rustc 1.65+

Supported Rust Versions

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tracing by you, shall be licensed as MIT, without any additional
terms or conditions.

2. Official Technical Reference & Guides (tokio-rs/website)

File: README.md

Tokio Website

The website for the Tokio project. Lives at https://tokio.rs.

Besides containing the content for the website, it also includes crates
containing the example code used in the tutorial. These crates can be compiled
and run.

Contributing

Thinking about contributing? Great! This should help you get the website running
locally.

Getting Started

The website is built using Next.js paired with the Bulma CSS framework.
You'll need NPM to install the required packages with:

bash

npm install

Next, start the development server:

bash

npm run dev

Then, open http://localhost:3000.

Resources

To learn more about Next.js, take a look at the following resources:

Next.js Documentation - learn about Next.js features and API.
Learn Next.js - an interactive Next.js tutorial.
Bulma documentation - learn about Bulma.

License

This project is licensed under the MIT license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Tokio by you, shall be licensed as MIT, without any additional
terms or conditions.

You can run our tests by running the commands:

text

# in doc-test (unstable APIs are needed by some tests)
RUSTFLAGS="--cfg tokio_unstable" cargo +nightly test

# in tutorial-code
cargo test --all

The doc tests verify that all code blocks are valid Rust, and the tutorial-code folder
contains the full code examples from the tutorial.

File: content/tokio/topics/bridging.md

title: "Bridging with sync code"

In most examples of using Tokio, we mark the main function with #[tokio::main]
and make the entire project asynchronous.

In some cases, you may need to run a small portion of synchronous code. For more
information on that, see spawn_blocking.

In other cases, it may be easier to structure the application as largely
synchronous, with smaller or logically distinct asynchronous portions.
For instance, a GUI application might want to run the GUI code on the
main thread and run a Tokio runtime next to it on another thread.

This page explains how you can isolate async/await to a small part of your
project.

What `#[tokio::main]` expands to

The #[tokio::main] macro is a macro that replaces your main function with a
non-async main function that starts a runtime and then calls your code. For
instance, this:

rust

#[tokio::main]
async fn main() {
    println!("Hello world");
}

is turned into this:

rust

fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("Hello world");
        })
}

by the macro. To use async/await in our own projects, we can do something
similar where we leverage the block_on method to enter the asynchronous
context where appropriate.

A synchronous interface to mini-redis

In this section, we will go through how to build a synchronous interface to
mini-redis by storing a Runtime object and using its block_on method.
In the following sections, we will discuss some alternate approaches and when
you should use each approach.

The interface that we will be wrapping is the asynchronous Client type. It
has several methods, and we will implement a blocking version of the following
methods:

To do this, we introduce a new file called src/clients/blocking_client.rs and
initialize it with a wrapper struct around the async Client type:

use tokio::net::ToSocketAddrs;
use tokio::runtime::Runtime;

pub use crate::clients::client::Message;

/// Established connection with a Redis server.
pub struct BlockingClient {
    /// The asynchronous `Client`.
    inner: crate::clients::Client,

    /// A `current_thread` runtime for executing operations on the
    /// asynchronous client in a blocking manner.
    rt: Runtime,
}

impl BlockingClient {
    pub fn connect<T: ToSocketAddrs>(addr: T) -> crate::Result {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;

        // Call the asynchronous connect method using the runtime.
        let inner = rt.block_on(crate::clients::Client::connect(addr))?;

        Ok(BlockingClient { inner, rt })
    }
}

Here, we have included the constructor function as our first example of how to
execute asynchronous methods in a non-async context. We do this using the
block_on method on the Tokio Runtime type, which executes an
asynchronous method and returns its result.

One important detail is the use of the current_thread runtime. Usually when
using Tokio, you would be using the default multi_thread runtime, which will
spawn a bunch of background threads so it can efficiently run many things at the
same time. For our use-case, we are only going to be doing one thing at the
time, so we won't gain anything by running multiple threads. This makes the
current_thread runtime a perfect fit as it doesn't spawn any threads.

The enable_all call enables the IO and timer drivers on the Tokio runtime.
If they are not enabled, the runtime is unable to perform IO or timers.

warning
Because the current_thread runtime does not spawn threads, it only operates
when block_on is called. Once block_on returns, all spawned tasks on that
runtime will freeze until you call block_on again. Use the multi_threaded
runtime if spawned tasks must keep running when not calling block_on.

Once we have this struct, most of the methods are easy to implement:

use bytes::Bytes;
use std::time::Duration;

impl BlockingClient {
    pub fn get(&mut self, key: &str) -> crate::Result<Option> {
        self.rt.block_on(self.inner.get(key))
    }

    pub fn set(&mut self, key: &str, value: Bytes) -> crate::Result<()> {
        self.rt.block_on(self.inner.set(key, value))
    }

    pub fn set_expires(
        &mut self,
        key: &str,
        value: Bytes,
        expiration: Duration,
    ) -> crate::Result<()> {
        self.rt.block_on(self.inner.set_expires(key, value, expiration))
    }

    pub fn publish(&mut self, channel: &str, message: Bytes) -> crate::Result {
        self.rt.block_on(self.inner.publish(channel, message))
    }
}

The Client::subscribe method is more interesting because it transforms the
Client into a Subscriber object. We can implement it in the following
manner:

/// A client that has entered pub/sub mode.
///
/// Once clients subscribe to a channel, they may only perform
/// pub/sub related commands. The `BlockingClient` type is
/// transitioned to a `BlockingSubscriber` type in order to
/// prevent non-pub/sub methods from being called.
pub struct BlockingSubscriber {
    /// The asynchronous `Subscriber`.
    inner: crate::clients::Subscriber,

    /// A `current_thread` runtime for executing operations on the
    /// asynchronous client in a blocking manner.
    rt: Runtime,
}

impl BlockingClient {
    pub fn subscribe(self, channels: Vec<String>) -> crate::Result {
        let subscriber = self.rt.block_on(self.inner.subscribe(channels))?;
        Ok(BlockingSubscriber {
            inner: subscriber,
            rt: self.rt,
        })
    }
}

impl BlockingSubscriber {
    pub fn get_subscribed(&self) -> &[String] {
        self.inner.get_subscribed()
    }

    pub fn next_message(&mut self) -> crate::Result<Option<Message>> {
        self.rt.block_on(self.inner.next_message())
    }

    pub fn subscribe(&mut self, channels: &[String]) -> crate::Result<()> {
        self.rt.block_on(self.inner.subscribe(channels))
    }

    pub fn unsubscribe(&mut self, channels: &[String]) -> crate::Result<()> {
        self.rt.block_on(self.inner.unsubscribe(channels))
    }
}

So, the subscribe method will first use the runtime to transform the
asynchronous Client into an asynchronous Subscriber. Then, it will store the
resulting Subscriber together with the Runtime and implement the various
methods using block_on.

Note that the asynchronous Subscriber struct has a non-async method called
get_subscribed. To handle this, we simply call it directly without involving
the runtime.

Other approaches

The above section explains the simplest way to implement a synchronous wrapper,
but it is not the only way. The approaches are:

Create a Runtime and call block_on on the async code.
Create a Runtime and spawn things on it.
Run the Runtime in a separate thread and send messages to it.

We already saw the first approach. The two other approaches are outlined below.

Spawning things on a runtime

The Runtime object has a method called spawn. When you call this method,
you create a new background task to run on the runtime. For example:

rust

use tokio::runtime::Builder;
use tokio::time::{sleep, Duration};

fn main() {
    let runtime = Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()
        .unwrap();

    let mut handles = Vec::with_capacity(10);
    for i in 0..10 {
        handles.push(runtime.spawn(my_bg_task(i)));
    }

    // Do something time-consuming while the background tasks execute.
    std::thread::sleep(Duration::from_millis(750));
    println!("Finished time-consuming task.");

    // Wait for all of them to complete.
    for handle in handles {
        // The `spawn` method returns a `JoinHandle`. A `JoinHandle` is
        // a future, so we can wait for it using `block_on`.
        runtime.block_on(handle).unwrap();
    }
}

async fn my_bg_task(i: u64) {
    // By subtracting, the tasks with larger values of i sleep for a
    // shorter duration.
    let millis = 1000 - 50 * i;
    println!("Task {} sleeping for {} ms.", i, millis);

    sleep(Duration::from_millis(millis)).await;

    println!("Task {} stopping.", i);
}

text

Task 0 sleeping for 1000 ms.
Task 1 sleeping for 950 ms.
Task 2 sleeping for 900 ms.
Task 3 sleeping for 850 ms.
Task 4 sleeping for 800 ms.
Task 5 sleeping for 750 ms.
Task 6 sleeping for 700 ms.
Task 7 sleeping for 650 ms.
Task 8 sleeping for 600 ms.
Task 9 sleeping for 550 ms.
Task 9 stopping.
Task 8 stopping.
Task 7 stopping.
Task 6 stopping.
Finished time-consuming task.
Task 5 stopping.
Task 4 stopping.
Task 3 stopping.
Task 2 stopping.
Task 1 stopping.
Task 0 stopping.

In the above example, we spawn 10 background tasks on the runtime, then wait
for all of them. As an example, this could be a good way of implementing
background network requests in a graphical application because network requests
are too time consuming to run them on the main GUI thread. Instead, you spawn
the request on a Tokio runtime running in the background, and have the task send
information back to the GUI code when the request has finished, or even
incrementally if you want a progress bar.

In this example, it is important that the runtime is configured to be a
multi_thread runtime. If you change it to be a current_thread runtime,
you will find that the time consuming task finishes before any of the background
tasks start. This is because background tasks spawned on a current_thread
runtime will only execute during calls to block_on as the runtime otherwise
doesn't have anywhere to run them.

The example waits for the spawned tasks to finish by calling block_on on the
JoinHandle returned by the call to spawn, but this isn't the only way to
do it. Here are some alternatives:

Use a message passing channel such as tokio::sync::mpsc.
Modify a shared value protected by e.g. a Mutex. This can be a good
approach for a progress bar in a GUI, where the GUI reads the shared value
every frame.

The spawn method is also available on the Handle type. The Handle type
can be cloned to get many handles to a runtime, and each Handle can be used to
spawn new tasks on the runtime.

Sending messages

The third technique is to spawn a runtime and use message passing to communicate
with it. This involves a bit more boilerplate than the other two approaches, but
it is the most flexible approach. You can find a basic example below:

rust

use tokio::runtime::Builder;
use tokio::sync::mpsc;

pub struct Task {
    name: String,
    // info that describes the task
}

async fn handle_task(task: Task) {
    println!("Got task {}", task.name);
}

#[derive(Clone)]
pub struct TaskSpawner {
    spawn: mpsc::Sender<Task>,
}

impl TaskSpawner {
    pub fn new() -> TaskSpawner {
        // Set up a channel for communicating.
        let (send, mut recv) = mpsc::channel(16);

        // Build the runtime for the new thread.
        //
        // The runtime is created before spawning the thread
        // to more cleanly forward errors if the `unwrap()`
        // panics.
        let rt = Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        std::thread::spawn(move || {
            rt.block_on(async move {
                while let Some(task) = recv.recv().await {
                    tokio::spawn(handle_task(task));
                }

                // Once all senders have gone out of scope,
                // the `.recv()` call returns None and it will
                // exit from the while loop and shut down the
                // thread.
            });
        });

        TaskSpawner {
            spawn: send,
        }
    }

    pub fn spawn_task(&self, task: Task) {
        match self.spawn.blocking_send(task) {
            Ok(()) => {},
            Err(_) => panic!("The shared runtime has shut down."),
        }
    }
}

This example could be configured in many ways. For instance, you could use a
Semaphore to limit the number of active tasks, or you could use a channel in
the opposite direction to send a response to the spawner. When you spawn a
runtime in this way, it is a type of actor.

File: content/tokio/topics/index.md

title: "Topics"

This section has self-contained articles related to various topics
that come up when writing asynchronous applications.

The currently available topics articles are:

File: content/tokio/topics/shutdown.md

title: "Graceful Shutdown"

The purpose of this page is to give an overview of how to properly implement
shutdown in asynchronous applications.

There are usually three parts to implementing graceful shutdown:

Figuring out when to shut down.
Telling every part of the program to shut down.
Waiting for other parts of the program to shut down.

The rest of this article will go through these parts. A real world
implementation of the approach described here can be found in mini-redis,
specifically the src/server.rs and
src/shutdown.rs files.

Figuring out when to shut down

This will of course depend on the application, but one very common shutdown
criteria is when the application receives a signal from the operating system.
This happens e.g. when you press ctrl+c in the terminal while the program is
running. To detect this, Tokio provides a tokio::signal::ctrl_c
function, which will sleep until such a signal is received. You might use it
like this:

use tokio::signal;

#[tokio::main]
async fn main() {
    // ... spawn application as separate task ...

    match signal::ctrl_c().await {
        Ok(()) => {},
        Err(err) => {
            eprintln!("Unable to listen for shutdown signal: {}", err);
            // we also shut down in case of error
        },
    }

    // send shutdown signal to application and wait
}

If you have multiple shutdown conditions, you can use an mpsc channel
to send the shutdown signal to one place. You can then select on
ctrl_c and the channel. For example:

use tokio::signal;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (shutdown_send, mut shutdown_recv) = mpsc::unbounded_channel();

    // ... spawn application as separate task ...
    //
    // application uses shutdown_send in case a shutdown was issued from inside
    // the application

    tokio::select! {
        _ = signal::ctrl_c() => {},
        _ = shutdown_recv.recv() => {},
    }

    // send shutdown signal to application and wait
}

Telling things to shut down

When you want to tell one or more tasks to shut down, you can use Cancellation
Tokens. These tokens allow you to notify tasks that they
should terminate themselves in response to a cancellation request, making it
easy to implement graceful shutdowns.

To share a CancellationToken between several tasks, you must clone it. This is due
to the single ownership rule that requires that each value has a single owner. When
cloning a token, you get another token that's indistinguishable from the original;
if one is cancelled, then the other is also cancelled. You can make as many clones
as you need, and when you call cancel on one of them, they're all cancelled.

Here are the steps to use CancellationToken in multiple tasks:

First, create a new CancellationToken.
Then, create a clone of the original CancellationToken by calling the clone method on the original token. This will create a new token that can be used by another task.
Pass the original or cloned token to the tasks that should respond to cancellation requests.
When you want to shut down the tasks gracefully, call the cancel method on the original or cloned token. Any task listening to the cancellation request on the original or cloned token will be notified to shut down.

Here is code snippet showcasing the above mentioned steps:

// Step 1: Create a new CancellationToken
let token = CancellationToken::new();

// Step 2: Clone the token for use in another task
let cloned_token = token.clone();

// Task 1 - Wait for token cancellation or a long time
let task1_handle = tokio::spawn(async move {
    tokio::select! {
        // Step 3: Using cloned token to listen to cancellation requests
        _ = cloned_token.cancelled() => {
            // The token was cancelled, task can shut down
        }
        _ = tokio::time::sleep(std::time::Duration::from_secs(9999)) => {
            // Long work has completed
        }
    }
});

// Task 2 - Cancel the original token after a small delay
tokio::spawn(async move {
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;

    // Step 4: Cancel the original or cloned token to notify other tasks about shutting down gracefully
    token.cancel();
});

// Wait for tasks to complete
task1_handle.await.unwrap()

With Cancellation Tokens, you don't have to shut down a task immediately when
the token is cancelled. Instead, you can run a shutdown procedure before
terminating the task, such as flushing data to a file or database, or sending
a shutdown message on a connection.

Waiting for things to finish shutting down

Once you have told other tasks to shut down, you will need to wait for them to
finish. One easy way to do so is to use a task tracker. A task tracker is a
collection of tasks. The wait method of the task tracker gives you a future
which resolves only after all of its contained futures have resolved and
the task tracker has been closed.

The following example will spawn 10 tasks, then use a task tracker to wait for
them to shut down.

use std::time::Duration;
use tokio::time::sleep;
use tokio_util::task::TaskTracker;

#[tokio::main]
async fn main() {
    let tracker = TaskTracker::new();

    for i in 0..10 {
        tracker.spawn(some_operation(i));
    }

    // Once we spawned everything, we close the tracker.
    tracker.close();

    // Wait for everything to finish.
    tracker.wait().await;

    println!("This is printed after all of the tasks.");
}

async fn some_operation(i: u64) {
    sleep(Duration::from_millis(100 * i)).await;
    println!("Task {} shutting down.", i);
}

File: content/tokio/topics/testing.md

title: "Unit Testing"

The purpose of this page is to give advice on how to write useful unit tests in
asynchronous applications.

Pausing and resuming time in tests

Sometimes, asynchronous code explicitly waits by calling tokio::time::sleep
or waiting on a tokio::time::Interval::tick. Testing behaviour based on
time (for example, an exponential backoff) can get cumbersome when the unit
test starts running very slowly. However, internally, the time-related
functionality of tokio supports pausing and resuming time. Pausing time has the
effect that any time-related future may become ready early. The condition for
the time-related future resolving early is that there are no more other futures
which may become ready. This essentially fast-forwards time when the only
future being awaited is time-related:

rust

#[tokio::test]
async fn paused_time() {
    tokio::time::pause();
    let start = std::time::Instant::now();
    tokio::time::sleep(Duration::from_millis(500)).await;
    println!("{:?}ms", start.elapsed().as_millis());
}

This code prints 0ms on a reasonable machine.

For unit tests, it is often useful to run with paused time throughout. This can
be achieved simply by setting the macro argument start_paused to true:

rust

#[tokio::test(start_paused = true)]
async fn paused_time() {
    let start = std::time::Instant::now();
    tokio::time::sleep(Duration::from_millis(500)).await;
    println!("{:?}ms", start.elapsed().as_millis());
}

Keep in mind that the start_paused attribute requires the tokio feature test-util.
See tokio::test "Configure the runtime to start with time paused" for more details.

Of course, the temporal order of future resolution is maintained, even when
using different time-related futures:

rust

#[tokio::test(start_paused = true)]
async fn interval_with_paused_time() {
    let mut interval = interval(Duration::from_millis(300));
    let _ = timeout(Duration::from_secs(1), async move {
        loop {
            interval.tick().await;
            println!("Tick!");
        }
    })
    .await;
}

This code immediately prints "Tick!" exactly 4 times.

Mocking using [`AsyncRead`] and [`AsyncWrite`]

The generic traits for reading and writing asynchronously (AsyncRead and
AsyncWrite) are implemented by, for example, sockets. They can be used for
mocking I/O performed by a socket.

Consider, for setup, this simple TCP server loop:

rust

use tokio::net::TcpListener;

#[tokio::main]
async fn main() {
    # if true { return }
    let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
    loop {
        let Ok((mut socket, _)) = listener.accept().await else {
            eprintln!("Failed to accept client");
            continue;
        };

        tokio::spawn(async move {
            let (reader, writer) = socket.split();
            // Run some client connection handler, for example:
            // handle_connection(reader, writer)
                // .await
                // .expect("Failed to handle connection");
        });
    }
}

Here, each TCP client connection is serviced by its dedicated tokio task. This
task owns a reader and a writer, which are split off of a TcpStream.

Consider now the actual client handler task, especially the where-clause of the
function signature:

rust

use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};

async fn handle_connection<Reader, Writer>(
    reader: Reader,
    mut writer: Writer,
) -> std::io::Result<()>
where
    Reader: AsyncRead + Unpin,
    Writer: AsyncWrite + Unpin,
{
    let mut line = String::new();
    let mut reader = BufReader::new(reader);

    loop {
        if let Ok(bytes_read) = reader.read_line(&mut line).await {
            if bytes_read == 0 {
                break Ok(());
            }
            writer
                .write_all(format!("Thanks for your message.\r\n").as_bytes())
                .await
                .unwrap();
        }
        line.clear();
    }
}

Essentially, the given reader and writer, which implement AsyncRead and
AsyncWrite, are serviced sequentially. For each received line, the handler
replies with "Thanks for your message.".

To unit test the client connection handler, a tokio_test::io::Builder can
be used as a mock:

rust

# use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
# 
# async fn handle_connection<Reader, Writer>(
#     reader: Reader,
#     mut writer: Writer,
# ) -> std::io::Result<()>
# where
#     Reader: AsyncRead + Unpin,
#     Writer: AsyncWrite + Unpin,
# {
#     let mut line = String::new();
#     let mut reader = BufReader::new(reader);
#
#     loop {
#         if let Ok(bytes_read) = reader.read_line(&mut line).await {
#             if bytes_read == 0 {
#                 break Ok(());
#             }
#             writer
#                 .write_all(format!("Thanks for your message.\r\n").as_bytes())
#                 .await
#                 .unwrap();
#         }
#         line.clear();
#     }
# }
#
#[tokio::test]
async fn client_handler_replies_politely() {
    let reader = tokio_test::io::Builder::new()
        .read(b"Hi there\r\n")
        .read(b"How are you doing?\r\n")
        .build();
    let writer = tokio_test::io::Builder::new()
        .write(b"Thanks for your message.\r\n")
        .write(b"Thanks for your message.\r\n")
        .build();
    let _ = handle_connection(reader, writer).await;
}

File: content/tokio/topics/tracing.md

title: Getting started with Tracing

The tracing crate is a framework for instrumenting Rust programs to collect
structured, event-based diagnostic information.

In asynchronous systems like Tokio, interpreting traditional log messages can
often be quite challenging. Since individual tasks are multiplexed on the same
thread, associated events and log lines are intermixed making it difficult to
trace the logic flow. tracing expands upon logging-style diagnostics by
allowing libraries and applications to record structured events with additional
information about temporality and causality — unlike a log message, a
Span in tracing has a beginning and end time, may be entered and exited
by the flow of execution, and may exist within a nested tree of similar spans.
For representing things that occur at a single moment in time, tracing
provides the complementary concept of events. Both Spans and Events
are structured, with the ability to record typed data as well as textual
messages.

You can use tracing to:

emit distributed traces to an OpenTelemetry collector
debug your application with Tokio Console
log to stdout, a log file or journald
profile where your application is spending time

Setup

To begin, add tracing and tracing-subscriber as dependencies:

toml

[dependencies]
tracing = "0.1"
tracing-subscriber = "0.3"

The tracing crate provides the API we will use to emit traces. The
tracing-subscriber crate provides some basic utilities for forwarding those
traces to external listeners (e.g., stdout).

Subscribing to Traces

If you are authoring an executable (as opposed to a library), you will need to
register a tracing subscriber. Subscribers are types that process traces
emitted by your application and its dependencies, and can perform tasks
such as computing metrics, monitoring for errors, and re-emitting traces to the
outside world (e.g., journald, stdout, or an open-telemetry daemon).

In most circumstances, you should register your tracing subscriber as early as
possible in your main function. For instance, the FmtSubscriber type
provided by tracing-subscriber prints formatted traces and events to
stdout, can be registered like so:

rust

# mod mini_redis {
#       pub type Error = Box<dyn std::error::Error + Send + Sync>;
#       pub type Result<T> = std::result::Result<T, Error>;
# }
#[tokio::main]
pub async fn main() -> mini_redis::Result<()> {
    // construct a subscriber that prints formatted traces to stdout
    let subscriber = tracing_subscriber::FmtSubscriber::new();
    // use that subscriber to process traces emitted after this point
    tracing::subscriber::set_global_default(subscriber)?;
# /*
    ...
# */ Ok(())
}

If you run your application now, you may see some trace events emitted by Tokio,
but you will need to modify your own application to emit traces to get the most
out of tracing.

Subscriber Configuration

In the above example, we've configured FmtSubscriber with its default
configuration. However, tracing-subscriber also provides a number of ways to
configure the FmtSubscriber's behavior, such as customizing the output
format, including additional information (such as thread IDs or source code
locations) in the logs, and writing the logs to somewhere other than stdout.

For example:

rust

// Start configuring a `fmt` subscriber
let subscriber = tracing_subscriber::fmt()
    // Use a more compact, abbreviated log format
    .compact()
    // Display source code file paths
    .with_file(true)
    // Display source code line numbers
    .with_line_number(true)
    // Display the thread ID an event was recorded on
    .with_thread_ids(true)
    // Don't display the event's target (module path)
    .with_target(false)
    // Build the subscriber
    .finish();

For details on the available configuration options, see the
tracing_subscriber::fmt documentation.

In addition to the FmtSubscriber type from tracing-subscriber, other
Subscribers can implement their own ways of recording tracing data. This
includes alternative output formats, analysis and aggregation, and integration
with other systems such as distributed tracing or log aggregation services. A
number of crates provide additional Subscriber implementations that may be of
interest. See here for an (incomplete) list of additional
Subscriber implementations.

Finally, in some cases, it may be useful to combine multiple different ways of
recording traces together to build a single Subscriber that implements
multiple behaviors. For this purpose, the tracing-subscriber crate provides a
Layer trait that represents a component that may be composed together with
other Layers to form a Subscriber. See here for details on using
Layers.

Emitting Spans and Events

The easiest way to emit spans is with the instrument proc-macro annotation
provided by tracing, which re-writes the bodies of functions to emit spans
each time they are invoked; e.g.:

rust

#[tracing::instrument]
fn trace_me(a: u32, b: u32) -> u32 {
    a + b
}

Each invocation of trace_me will emit a tracing Span that:

has a verbosity level of info (the "middle ground" verbosity),
is named trace_me,
has fields a and b, whose values are the arguments of trace_me

The instrument attribute is highly configurable; e.g., to trace
the method in mini-redis-server that handles each connection:

rust

# struct Handler {
#     connection: tokio::net::TcpStream,
# }
# pub type Error = Box<dyn std::error::Error + Send + Sync>;
# pub type Result<T> = std::result::Result<T, Error>;
use tracing::instrument;

impl Handler {
    /// Process a single connection.
    #[instrument(
        name = "Handler::run",
        skip(self),
        fields(
            // `%` serializes the peer IP addr with `Display`
            peer_addr = %self.connection.peer_addr().unwrap()
        ),
    )]
    async fn run(&mut self) -> mini_redis::Result<()> {
# /*
        ...
# */ Ok::<_, _>(())
    }
}

mini-redis-server will now emit a tracing Span for each incoming
connection that:

has a verbosity level of info (the "middle ground" verbosity),
is named Handler::run,
has some structured data associated with it.
- fields(...) indicates that emitted span should include
  the fmt::Display representation of the connection's SocketAddr in a field
  called peer_addr.
- skip(self) indicates that emitted span should not record Handler's debug representation.

You can also construct a Span manually by invoking the span! macro, or
any of its leveled shorthands (error_span!, warn_span!, info_span!,
debug_span!, trace_span!).

To emit events, invoke the event! macro, or any of its leveled shorthands
(error!, warn!, info!, debug!, trace!). For instance, to
log that a client sent a malformed command:

rust

# type Error = Box<dyn std::error::Error + Send + Sync>;
# type Result<T> = std::result::Result<T, Error>;
# struct Command;
# impl Command {
#     fn from_frame<T>(frame: T) -> Result<()> {
#         Result::Ok(())
#     }
#     fn from_error<T>(err: T) {}
# }
# let frame = ();
// Convert the redis frame into a command struct. This returns an
// error if the frame is not a valid redis command.
let cmd = match Command::from_frame(frame) {
    Ok(cmd) => cmd,
    Err(cause) => {
        // The frame was malformed and could not be parsed. This is
        // probably indicative of an issue with the client (as opposed
        // to our server), so we (1) emit a warning...
        //
        // The syntax here is a shorthand provided by the `tracing`
        // crate. It can be thought of as similar to:
        //      tracing::warn! {
        //          cause = format!("{}", cause),
        //          "failed to parse command from frame"
        //      };
        // `tracing` provides structured logging, so information is
        // "logged" as key-value pairs.
        tracing::warn! {
            %cause,
            "failed to parse command from frame"
        };
        // ...and (2) respond to the client with the error:
        Command::from_error(cause)
    }
};

If you run your application, you now will see events decorated with their span's
context emitted for each incoming connection that it processes.

File: content/tokio/topics/tracing-next-steps.md

title: "Next steps with Tracing"

Tokio-console

tokio-console is an htop-like utility that enables you to see a real-time view
of an application’s spans and events. It can also represent "resources" that the
Tokio runtime has created, such as Tasks. It's essential for understanding
performance issues during the development process.

For instance, to use tokio-console in the mini-redis project,
you need to enable the tracing feature for the Tokio package:

toml

# Update the tokio import in your Cargo.toml
tokio = { version = "1", features = ["full", "tracing"] }

Note: The full feature doesn't enable tracing.

You'll also need to add a dependency on the console-subscriber package. This
crate provides a Subscriber implementation that will replace the one currently
used by mini-redis:

toml

# Add this to the dependencies section of your Cargo.toml
console-subscriber = "0.1.5"

Finally, in src/bin/server.rs, replace the call to tracing_subscriber with
the call to console-subscriber:

Replace this:

rust

# use std::error::Error;
tracing_subscriber::fmt::try_init()?;
# Ok::<(), Box<dyn Error + Send + Sync + 'static>>(())

...with this:

rust

console_subscriber::init();

This will enable the console_subscriber, which means that any instrumentation
relevant to tokio-console will be recorded. Logging to stdout will still
happen (based on the value of the RUST_LOG environment variable).

Now we should be ready to start up mini-redis again, this time using the
tokio_unstable flag (which is needed to enable tracing):

RUSTFLAGS="--cfg tokio_unstable" cargo run --bin mini-redis-server

The tokio_unstable flag allows us to make use of additional APIs provided by
Tokio which do not currently have a guarantee of stability (in other words,
breaking changes are allowed for these APIs).

All that is remaining is to run the console itself in another terminal. The
easiest way to do that would be to install it from crates.io:

cargo install --locked tokio-console

and then run it with:

tokio-console

The initial view you will see is for tokio Tasks that are currently running.
Example: tokio-console Taskview

It can also show Tasks for a period of time after they have completed (the color
for these will be grey). You can generate some traces by running the mini-redis
hello world example (this is available in the mini-redis
repository):

cargo run --example hello_world

If you press r, you can switch to the Resources view. This displays
semaphores, mutexes, and other constructs that are being used by the Tokio
runtime. Example: tokio-console Resourceview

Whenever you need to introspect the Tokio runtime to understand the performance
of you application better, you can make use of tokio-console to view what is
happening in real time, helping you to spot deadlocks and other issues.

To learn more about how to use tokio-console, please visit its documentation
page.

Integrating with OpenTelemetry

OpenTelemetry (OTel) means multiple things; for
one, it's an open specification, defining a data model for traces and metrics
that can handle the needs of most users. It is also a set of language-specific
SDKs, providing instrumentation so that traces and metrics can be emitted from
an application. Thirdly, there is the OpenTelemetry Collector, a binary that
runs alongside your application to collect the traces and metrics, ultimately
pushing those out to a telemetry vendor, such as DataDog, Honeycomb, or AWS
X-Ray. It can also send data to tools such as Prometheus instead.

The opentelemetry crate is what
provides the OpenTelemetry SDK for Rust, and is what we will be using for this
tutorial.

In this tutorial, we will be setting up mini-redis to send data to
Jaeger, which is a UI for visualizing traces.

To run an instance of Jaeger, you can use Docker:

docker run -d -p6831:6831/udp -p6832:6832/udp -p16686:16686 -p14268:14268 jaegertracing/all-in-one:latest

You can visit the Jaeger page by going to http://localhost:16686.
It will look like this:

We'll come back to this page once we have some trace data generated and sent.

To set up mini-redis, we'll first need to add a few dependencies. Update your
Cargo.toml with the following:

toml

# Implements the types defined in the Otel spec
opentelemetry = "0.17.0"
# Integration between the tracing crate and the opentelemetry crate
tracing-opentelemetry = "0.17.2" 
# Allows you to export data to Jaeger
opentelemetry-jaeger = "0.16.0"

Now, in src/bin/server.rs, add the following imports:

rust

use opentelemetry::global;
use tracing_subscriber::{
    fmt, layer::SubscriberExt, util::SubscriberInitExt,
};

We will look at what each of these do in a moment.

The next step is to replace the call to tracing_subscriber with the OTel
setup.

Replace this:

rust

# use std::error::Error;
tracing_subscriber::fmt::try_init()?;
# Ok::<(), Box<dyn Error + Send + Sync + 'static>>(())

...with this:

rust

# use std::error::Error;
# use opentelemetry::global;
# use tracing_subscriber::{
#     fmt, layer::SubscriberExt, util::SubscriberInitExt,
# };
// Allows you to pass along context (i.e., trace IDs) across services
global::set_text_map_propagator(opentelemetry_jaeger::Propagator::new());
// Sets up the machinery needed to export data to Jaeger
// There are other OTel crates that provide pipelines for the vendors
// mentioned earlier.
let tracer = opentelemetry_jaeger::new_pipeline()
    .with_service_name("mini-redis")
    .install_simple()?;

// Create a tracing layer with the configured tracer
let opentelemetry = tracing_opentelemetry::layer().with_tracer(tracer);

// The SubscriberExt and SubscriberInitExt traits are needed to extend the
// Registry to accept `opentelemetry (the OpenTelemetryLayer type).
tracing_subscriber::registry()
    .with(opentelemetry)
    // Continue logging to stdout
    .with(fmt::Layer::default())
    .try_init()?;
# Ok::<(), Box<dyn Error + Send + Sync + 'static>>(())

Now you should be able to start up mini-redis:

cargo run --bin mini-redis-server

In another terminal, run the hello world example (this is available in the
mini-redis repository):

cargo run --example hello_world

Now, refresh the Jaeger UI that we had open, and on the main Search page, find
"mini-redis" as one of the options in the Service dropdown.

Select that option, and click the "Find Traces" button. This should show the
request we just made from running the example.

Clicking on the trace should show you a detailed view of the spans that were
emitted during the handling of the hello world example.

That's it for now! You can explore this further by sending more requests, adding
additional instrumentation for mini-redis, or setting up OTel with a telemetry
vendor (instead of the Jaeger instance we are running locally). For this last
one, you might need to pull in an additional crate (for example, for sending
data to the OTel Collector, you'll need the opentelemetry-otlp crate). There
are many examples available in the opentelemetry-rust
repository.

Note: The mini-redis repo already contains a full example of OpenTelemetry with
AWS X-Ray, details of which can be found in the
README, as well as
the
Cargo.toml
and
src/bin/server.rs
files.

File: content/tokio/tutorial/async.md

title: "Async in depth"

At this point, we have completed a fairly comprehensive tour of asynchronous
Rust and Tokio. Now we will dig deeper into Rust's asynchronous runtime model.
At the very beginning of the tutorial, we hinted that asynchronous Rust takes a
unique approach. Now, we explain what that means.

Futures

As a quick review, let's take a very basic asynchronous function. This is
nothing new compared to what the tutorial has covered so far.

rust

use tokio::net::TcpStream;

async fn my_async_fn() {
    println!("hello from async");
    let _socket = TcpStream::connect("127.0.0.1:3000").await.unwrap();
    println!("async TCP operation complete");
}

We call the function and it returns some value. We call .await on that value.

rust

# async fn my_async_fn() {}
#[tokio::main]
async fn main() {
    let what_is_this = my_async_fn();
    // Nothing has been printed yet.
    what_is_this.await;
    // Text has been printed and socket has been
    // established and closed.
}

The value returned by my_async_fn() is a future. A future is a value that
implements the std::future::Future trait provided by the standard
library. They are values that contain the in-progress asynchronous computation.

The std::future::Future trait definition is:

rust

use std::pin::Pin;
use std::task::{Context, Poll};

pub trait Future {
    type Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context)
        -> Poll<Self::Output>;
}

The associated type Output is the type that the future produces once
it completes. The Pin type is how Rust is able to support borrows in
async functions. See the standard library documentation for more
details.

Unlike how futures are implemented in other languages, a Rust future does not
represent a computation happening in the background, rather the Rust future
is the computation itself. The owner of the future is responsible for
advancing the computation by polling the future. This is done by calling
Future::poll.

Implementing `Future`

Let's implement a very simple future. This future will:

Wait until a specific instant in time.
Output some text to STDOUT.
Yield a string.

rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

struct Delay {
    when: Instant,
}

impl Future for Delay {
    type Output = &'static str;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<&'static str>
    {
        if Instant::now() >= self.when {
            println!("Hello world");
            Poll::Ready("done")
        } else {
            // Ignore this line for now.
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

#[tokio::main]
async fn main() {
    let when = Instant::now() + Duration::from_millis(10);
    let future = Delay { when };

    let out = future.await;
    assert_eq!(out, "done");
}

Async fn as a Future

In the main function, we instantiate the future and call .await on it. From
async functions, we may call .await on any value that implements Future. In
turn, calling an async function returns an anonymous type that implements
Future. In the case of async fn main(), the generated future is roughly:

rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

enum MainFuture {
    // Initialized, never polled
    State0,
    // Waiting on `Delay`, i.e. the `future.await` line.
    State1(Delay),
    // The future has completed.
    Terminated,
}
# struct Delay { when: Instant };
# impl Future for Delay {
#     type Output = &'static str;
#     fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<&'static str> {
#         unimplemented!();
#     }
# }

impl Future for MainFuture {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<()>
    {
        use MainFuture::*;

        loop {
            match *self {
                State0 => {
                    let when = Instant::now() +
                        Duration::from_millis(10);
                    let future = Delay { when };
                    *self = State1(future);
                }
                State1(ref mut my_future) => {
                    match Pin::new(my_future).poll(cx) {
                        Poll::Ready(out) => {
                            assert_eq!(out, "done");
                            *self = Terminated;
                            return Poll::Ready(());
                        }
                        Poll::Pending => {
                            return Poll::Pending;
                        }
                    }
                }
                Terminated => {
                    panic!("future polled after completion")
                }
            }
        }
    }
}

Rust futures are state machines. Here, MainFuture is represented as an
enum of the future's possible states. The future starts in the State0 state.
When poll is invoked, the future attempts to advance its internal state as
much as possible. If the future is able to complete, Poll::Ready is returned
containing the output of the asynchronous computation.

If the future is not able to complete, usually due to resources it is
waiting on not being ready, then Poll::Pending is returned. Receiving
Poll::Pending indicates to the caller that the future will complete at a later
time and the caller should invoke poll again later.

We also see that futures are composed of other futures. Calling poll on the
outer future results in calling the inner future's poll function.

Executors

Asynchronous Rust functions return futures. Futures must have poll called on
them to advance their state. Futures are composed of other futures. So, the
question is, what calls poll on the very most outer future?

Recall from earlier, to run asynchronous functions, they must either be
passed to tokio::spawn or be the main function annotated with
#[tokio::main]. This results in submitting the generated outer future to the
Tokio executor. The executor is responsible for calling Future::poll on the
outer future, driving the asynchronous computation to completion.

Mini Tokio

To better understand how this all fits together, let's implement our own minimal
version of Tokio! The full code can be found here.

rust

use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use futures::task;

fn main() {
    let mut mini_tokio = MiniTokio::new();

    mini_tokio.spawn(async {
        let when = Instant::now() + Duration::from_millis(10);
        let future = Delay { when };

        let out = future.await;
        assert_eq!(out, "done");
    });

    mini_tokio.run();
}
# struct Delay { when: Instant }
# impl Future for Delay {
#     type Output = &'static str;
#     fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<&'static str> {
#         Poll::Ready("done")
#     }
# }

struct MiniTokio {
    tasks: VecDeque<Task>,
}

type Task = Pin + Send>>;

impl MiniTokio {
    fn new() -> MiniTokio {
        MiniTokio {
            tasks: VecDeque::new(),
        }
    }

    /// Spawn a future onto the mini-tokio instance.
    fn spawn<F>(&mut self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        self.tasks.push_back(Box::pin(future));
    }

    fn run(&mut self) {
        let waker = task::noop_waker();
        let mut cx = Context::from_waker(&waker);

        while let Some(mut task) = self.tasks.pop_front() {
            if task.as_mut().poll(&mut cx).is_pending() {
                self.tasks.push_back(task);
            }
        }
    }
}

This runs the async block. A Delay instance is created with the requested
delay and is awaited on. However, our implementation so far has a major flaw.
Our executor never goes to sleep. The executor continuously loops all
spawned futures and polls them. Most of the time, the futures will not be ready
to perform more work and will return Poll::Pending again. The process will
burn CPU cycles and generally not be very efficient.

Ideally, we want mini-tokio to only poll futures when the future is able to make
progress. This happens when a resource that the task is blocked on becomes ready
to perform the requested operation. If the task wants to read data from a TCP
socket, then we only want to poll the task when the TCP socket has received
data. In our case, the task is blocked on the given Instant being reached.
Ideally, mini-tokio would only poll the task once that instant in time has
passed.

To achieve this, when a resource is polled, and the resource is not ready,
the resource will send a notification once it transitions into a ready state.

Wakers

Wakers are the missing piece. This is the system by which a resource is able to
notify the waiting task that the resource has become ready to continue some
operation.

Let's look at the Future::poll definition again:

rust,compile_fail

fn poll(self: Pin<&mut Self>, cx: &mut Context)
    -> Poll<Self::Output>;

The Context argument to poll has a waker() method. This method returns a
Waker bound to the current task. The Waker has a wake() method. Calling
this method signals to the executor that the associated task should be scheduled
for execution. Resources call wake() when they transition to a ready state to
notify the executor that polling the task will be able to make progress.

Updating `Delay`

We can update Delay to use wakers:

rust

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use std::thread;

struct Delay {
    when: Instant,
}

impl Future for Delay {
    type Output = &'static str;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<&'static str>
    {
        if Instant::now() >= self.when {
            println!("Hello world");
            Poll::Ready("done")
        } else {
            // Get a handle to the waker for the current task
            let waker = cx.waker().clone();
            let when = self.when;

            // Spawn a timer thread.
            thread::spawn(move || {
                let now = Instant::now();

                if now < when {
                    thread::sleep(when - now);
                }

                waker.wake();
            });

            Poll::Pending
        }
    }
}

Now, once the requested duration has elapsed, the calling task is notified and
the executor can ensure the task is scheduled again. The next step is to update
mini-tokio to listen for wake notifications.

There are still a few remaining issues with our Delay implementation. We will
fix them later.

warning
When a future returns Poll::Pending, it must ensure that the waker is
signalled at some point. Forgetting to do this results in the task hanging
indefinitely.

Forgetting to wake a task after returning Poll::Pending is a common
source of bugs.

Recall the first iteration of Delay. Here was the future implementation:

rust

# use std::future::Future;
# use std::pin::Pin;
# use std::task::{Context, Poll};
# use std::time::Instant;
# struct Delay { when: Instant }
impl Future for Delay {
    type Output = &'static str;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<&'static str>
    {
        if Instant::now() >= self.when {
            println!("Hello world");
            Poll::Ready("done")
        } else {
            // Ignore this line for now.
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

Before returning Poll::Pending, we called cx.waker().wake_by_ref(). This is
to satisfy the future contract. By returning Poll::Pending, we are responsible
for signalling the waker. Because we didn't implement the timer thread yet, we
signalled the waker inline. Doing so will result in the future being immediately
re-scheduled, executed again, and probably not be ready to complete.

Notice that you are allowed to signal the waker more often than necessary. In
this particular case, we signal the waker even though we are not ready to
continue the operation at all. There is nothing wrong with this besides some
wasted CPU cycles. However, this particular implementation will result in a busy
loop.

Updating Mini Tokio

The next step is updating Mini Tokio to receive waker notifications. We want the
executor to only run tasks when they are woken, and to do this, Mini Tokio will
provide its own waker. When the waker is invoked, its associated task is queued
to be executed. Mini-Tokio passes this waker to the future when it polls the
future.

The updated Mini Tokio will use a channel to store scheduled tasks. Channels
allow tasks to be queued for execution from any thread. Wakers must be Send
and Sync.

info
The Send and Sync traits are marker traits related to concurrency
provided by Rust. Types that can be sent to a different thread are
Send. Most types are Send, but something like Rc is not. Types
that can be concurrently accessed through immutable references are
Sync. A type can be Send but not Sync — a good example is
Cell, which can be modified through an immutable reference, and
is thus not safe to access concurrently.

For more details, see the related chapter in the Rust book.

Update the MiniTokio struct.

rust

use std::sync::mpsc;
use std::sync::Arc;

struct MiniTokio {
    scheduled: mpsc::Receiver<Arc<Task>>,
    sender: mpsc::Sender<Arc<Task>>,
}

struct Task {
    // This will be filled in soon.
}

Wakers are Sync and can be cloned. When wake is called, the task must be
scheduled for execution. To implement this, we have a channel. When the wake()
is called on the waker, the task is pushed into the send half of the channel.
Our Task structure will implement the wake logic. To do this, it needs to
contain both the spawned future and the channel send half. We place the future
in a TaskFuture struct alongside a Poll enum to keep track of the latest
Future::poll() result, which is needed to handle spurious wake-ups. More
details are given in the implementation of the poll() method in TaskFuture.

rust

# use std::future::Future;
# use std::pin::Pin;
# use std::sync::mpsc;
# use std::task::Poll;
use std::sync::{Arc, Mutex};

/// A structure holding a future and the result of
/// the latest call to its `poll` method.
struct TaskFuture {
    future: Pin + Send>>,
    poll: Poll<()>,
}

struct Task {
    // The `Mutex` is to make `Task` implement `Sync`. Only
    // one thread accesses `task_future` at any given time.
    // The `Mutex` is not required for correctness. Real Tokio
    // does not use a mutex here, but real Tokio has
    // more lines of code than can fit in a single tutorial
    // page.
    task_future: Mutex<TaskFuture>,
    executor: mpsc::Sender<Arc<Task>>,
}

impl Task {
    fn schedule(self: &Arc<Self>) {
        self.executor.send(self.clone());
    }
}

To schedule the task, the Arc is cloned and sent through the channel. Now, we
need to hook our schedule function with std::task::Waker. The
standard library provides a low-level API to do this using manual vtable
construction. This strategy provides maximum flexibility to
implementors, but requires a bunch of unsafe boilerplate code. Instead of using
RawWakerVTable directly, we will use the ArcWake utility
provided by the futures crate. This allows us to implement a simple trait to
expose our Task struct as a waker.

Add the following dependency to your Cargo.toml to pull in futures.

toml

futures = "0.3"

Then implement futures::task::ArcWake.

rust

use futures::task::{self, ArcWake};
use std::sync::Arc;
# struct Task {}
# impl Task {
#     fn schedule(self: &Arc<Self>) {}
# }
impl ArcWake for Task {
    fn wake_by_ref(arc_self: &Arc<Self>) {
        arc_self.schedule();
    }
}

When the timer thread above calls waker.wake(), the task is pushed into the
channel. Next, we implement receiving and executing the tasks in the
MiniTokio::run() function.

text

/* Detailed source-code truncated for AI context efficiency. */

Multiple things are happening here. First, MiniTokio::run() is implemented.
The function runs in a loop receiving scheduled tasks from the channel.
As tasks are pushed into the channel when they are woken, these tasks are able
to make progress when executed.

Additionally, the MiniTokio::new() and MiniTokio::spawn() functions are
adjusted to use a channel rather than a VecDeque. When new tasks are spawned,
they are given a clone of the sender-part of the channel, which the task can
use to schedule itself on the runtime.

The Task::poll() function creates the waker using the ArcWake utility from
the futures crate. The waker is used to create a task::Context. That
task::Context is passed to poll.

Summary

We have now seen an end-to-end example of how asynchronous Rust works. Rust's
async/await feature is backed by traits. This allows third-party crates, like
Tokio, to provide the execution details.

Asynchronous Rust operations are lazy and require a caller to poll them.
Wakers are passed to futures to link a future to the task calling it.
When a resource is not ready to complete an operation, Poll::Pending is
returned and the task's waker is recorded.
When the resource becomes ready, the task's waker is notified.
The executor receives the notification and schedules the task to execute.
The task is polled again, this time the resource is ready and the task makes
progress.

A few loose ends

Recall when we were implementing the Delay future, we said there were a few
more things to fix. Rust's asynchronous model allows a single future to migrate
across tasks while it executes. Consider the following:

rust

use futures::future::poll_fn;
use std::future::Future;
use std::pin::Pin;
# use std::task::{Context, Poll};
# use std::time::{Duration, Instant};
# struct Delay { when: Instant }
# impl Future for Delay {
#   type Output = ();
#   fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> {
#       Poll::Pending
#   }
# }

#[tokio::main]
async fn main() {
    let when = Instant::now() + Duration::from_millis(10);
    let mut delay = Some(Delay { when });

    poll_fn(move |cx| {
        let mut delay = delay.take().unwrap();
        let res = Pin::new(&mut delay).poll(cx);
        assert!(res.is_pending());
        tokio::spawn(async move {
            delay.await;
        });

        Poll::Ready(())
    }).await;
}

The poll_fn function creates a Future instance using a closure. The snippet
above creates a Delay instance, polls it once, then sends the Delay instance
to a new task where it is awaited. In this example, Delay::poll is called more
than once with different Waker instances. When this happens, you must make
sure to call wake on the Waker passed to the most recent call to poll.

When implementing a future, it is critical to assume that each call to poll
could supply a different Waker instance. The poll function must update any
previously recorded waker with the new one.

Our earlier implementation of Delay spawned a new thread every time it was
polled. This is fine, but can be very inefficient if it is polled too often
(e.g. if you select! over that future and some other future, both are polled
whenever either has an event). One approach to this is to remember whether you
have already spawned a thread, and only spawn a new thread if you haven't
already spawned one. However if you do this, you must ensure that the thread's
Waker is updated on later calls to poll, as you are otherwise not waking the
most recent Waker.

To fix our earlier implementation, we could do something like this:

text

/* Detailed source-code truncated for AI context efficiency. */

It is a bit involved, but the idea is, on each call to poll, the future checks
if the supplied waker matches the previously recorded waker. If the two wakers
match, then there is nothing else to do. If they do not match, then the recorded
waker must be updated.

`Notify` utility

We demonstrated how a Delay future could be implemented by hand using wakers.
Wakers are the foundation of how asynchronous Rust works. Usually, it is not
necessary to drop down to that level. For example, in the case of Delay, we
could implement it entirely with async/await by using the
tokio::sync::Notify utility. This utility provides a basic task
notification mechanism. It handles the details of wakers, including making sure
that the recorded waker matches the current task.

Using Notify, we can implement a delay function using
async/await like this:

rust

use tokio::sync::Notify;
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::thread;

async fn delay(dur: Duration) {
    let when = Instant::now() + dur;
    let notify = Arc::new(Notify::new());
    let notify_clone = notify.clone();

    thread::spawn(move || {
        let now = Instant::now();

        if now < when {
            thread::sleep(when - now);
        }

        notify_clone.notify_one();
    });


    notify.notified().await;
}

File: content/tokio/tutorial/channels.md

title: "Channels"

Now that we have learned a bit about concurrency with Tokio, let's apply this on
the client side. Put the server code we wrote before into an explicit binary
file:

bash

$ mkdir src/bin
$ mv src/main.rs src/bin/server.rs

and create a new binary file that will contain the client code:

bash

$ touch src/bin/client.rs

In this file you will write this page's code. Whenever you want to run it,
you will have to launch the server first in a separate terminal window:

bash

$ cargo run --bin server

And then the client, separately:

bash

$ cargo run --bin client

That being said, let's code!

Say we want to run two concurrent Redis commands. We can spawn
one task per command. Then the two commands would happen concurrently.

At first, we might try something like:

rust,compile_fail

use mini_redis::client;

#[tokio::main]
async fn main() {
    // Establish a connection to the server
    let mut client = client::connect("127.0.0.1:6379").await.unwrap();

    // Spawn two tasks, one gets a key, the other sets a key
    let t1 = tokio::spawn(async {
        let res = client.get("foo").await;
    });

    let t2 = tokio::spawn(async {
        client.set("foo", "bar".into()).await;
    });

    t1.await.unwrap();
    t2.await.unwrap();
}

This does not compile because both tasks need to access the client somehow.
As Client does not implement Copy, it will not compile without some code to
facilitate this sharing. Additionally, Client::set takes &mut self, which
means that exclusive access is required to call it. We could open a connection
per task, but that is not ideal. We cannot use std::sync::Mutex as .await
would need to be called with the lock held. We could use tokio::sync::Mutex,
but that would only allow a single in-flight request. If the client implements
pipelining (in short, sending many commands without waiting for each prior
command's response), an async mutex results in underutilizing
the connection.

Message passing

The answer is to use message passing. The pattern involves spawning a dedicated
task to manage the client resource. Any task that wishes to issue a request
sends a message to the client task. The client task issues the request on
behalf of the sender, and the response is sent back to the sender.

Using this strategy, a single connection is established. The task managing the
client is able to get exclusive access in order to call get and set.
Additionally, the channel works as a buffer. Operations may be sent to the
client task while the client task is busy. Once the client task is
available to process new requests, it pulls the next request from the channel.
This can result in better throughput, and be extended to support connection
pooling.

Tokio's channel primitives

Tokio provides a number of channels, each serving a different purpose.

mpsc: multi-producer, single-consumer channel. Many values can be sent.
oneshot: single-producer, single consumer channel. A single value can be sent.
broadcast: multi-producer, multi-consumer. Many values can be sent. Each
receiver sees every value.
watch: multi-producer, multi-consumer. Many values can be sent, but no
history is kept. Receivers only see the most recent value.

If you need a multi-producer multi-consumer channel where only one consumer sees
each message, you can use the async-channel crate. There are also channels
for use outside of asynchronous Rust, such as std::sync::mpsc and
crossbeam::channel. These channels wait for messages by blocking the
thread, which is not allowed in asynchronous code.

In this section, we will use mpsc and oneshot. The other types of message
passing channels are explored in later sections. The full code from this section
is found here.

Define the message type

In most cases, when using message passing, the task receiving the messages
responds to more than one command. In our case, the task will respond to GET and
SET commands. To model this, we first define a Command enum and include a
variant for each command type.

rust

use bytes::Bytes;

#[derive(Debug)]
enum Command {
    Get {
        key: String,
    },
    Set {
        key: String,
        val: Bytes,
    }
}

Create the channel

In the main function, an mpsc channel is created.

rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // Create a new channel with a capacity of at most 32.
    let (tx, mut rx) = mpsc::channel(32);
# tx.send(()).await.unwrap();

    // ... Rest comes here
}

The mpsc channel is used to send commands to the task managing the redis
connection. The multi-producer capability allows messages to be sent from many
tasks. Creating the channel returns two values, a sender and a receiver. The two
handles are used separately. They may be moved to different tasks.

The channel is created with a capacity of 32. If messages are sent faster than
they are received, the channel will store them. Once the 32 messages are stored
in the channel, calling send(...).await will go to sleep until a message has
been removed by the receiver.

Sending from multiple tasks is done by cloning the Sender. For example:

rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);
    let tx2 = tx.clone();

    tokio::spawn(async move {
        tx.send("sending from first handle").await.unwrap();
    });

    tokio::spawn(async move {
        tx2.send("sending from second handle").await.unwrap();
    });

    while let Some(message) = rx.recv().await {
        println!("GOT = {}", message);
    }
}

Both messages are sent to the single Receiver handle. It is not possible to
clone the receiver of an mpsc channel.

When every Sender has gone out of scope or has otherwise been dropped, it is
no longer possible to send more messages into the channel. At this point, the
recv call on the Receiver will return None, which means that all senders
are gone and the channel is closed.

In our case of a task that manages the Redis connection, it knows that it
can close the Redis connection once the channel is closed, as the connection
will not be used again.

Spawn manager task

Next, spawn a task that processes messages from the channel. First, a client
connection is established to Redis. Then, received commands are issued via the
Redis connection.

rust

use mini_redis::client;
# enum Command {
#    Get { key: String },
#    Set { key: String, val: bytes::Bytes }
# }
# async fn dox() {
# let (_, mut rx) = tokio::sync::mpsc::channel(10);
// The `move` keyword is used to **move** ownership of `rx` into the task.
let manager = tokio::spawn(async move {
    // Establish a connection to the server
    let mut client = client::connect("127.0.0.1:6379").await.unwrap();

    // Start receiving messages
    while let Some(cmd) = rx.recv().await {
        use Command::*;

        match cmd {
            Get { key } => {
                client.get(&key).await;
            }
            Set { key, val } => {
                client.set(&key, val).await;
            }
        }
    }
});
# }

Now, update the two tasks to send commands over the channel instead of issuing
them directly on the Redis connection.

rust

# #[derive(Debug)]
# enum Command {
#    Get { key: String },
#    Set { key: String, val: bytes::Bytes }
# }
# async fn dox() {
# let (mut tx, _) = tokio::sync::mpsc::channel(10);
// The `Sender` handles are moved into the tasks. As there are two
// tasks, we need a second `Sender`.
let tx2 = tx.clone();

// Spawn two tasks, one gets a key, the other sets a key
let t1 = tokio::spawn(async move {
    let cmd = Command::Get {
        key: "foo".to_string(),
    };

    tx.send(cmd).await.unwrap();
});

let t2 = tokio::spawn(async move {
    let cmd = Command::Set {
        key: "foo".to_string(),
        val: "bar".into(),
    };

    tx2.send(cmd).await.unwrap();
});
# }

At the bottom of the main function, we .await the join handles to ensure the
commands fully complete before the process exits.

rust

# type Jh = tokio::task::JoinHandle<()>;
# async fn dox(t1: Jh, t2: Jh, manager: Jh) {
t1.await.unwrap();
t2.await.unwrap();
manager.await.unwrap();
# }

Receive responses

The final step is to receive the response back from the manager task. The GET
command needs to get the value and the SET command needs to know if the
operation completed successfully.

To pass the response, a oneshot channel is used. The oneshot channel is a
single-producer, single-consumer channel optimized for sending a single value.
In our case, the single value is the response.

Similar to mpsc, oneshot::channel() returns a sender and receiver handle.

rust

use tokio::sync::oneshot;

# async fn dox() {
let (tx, rx) = oneshot::channel();
# tx.send(()).unwrap();
# }

Unlike mpsc, no capacity is specified as the capacity is always one.
Additionally, neither handle can be cloned.

To receive responses from the manager task, before sending a command, a oneshot
channel is created. The Sender half of the channel is included in the command
to the manager task. The receive half is used to receive the response.

First, update Command to include the Sender. For convenience, a type alias
is used to reference the Sender.

rust

use tokio::sync::oneshot;
use bytes::Bytes;

/// Multiple different commands are multiplexed over a single channel.
#[derive(Debug)]
enum Command {
    Get {
        key: String,
        resp: Responder<Option>,
    },
    Set {
        key: String,
        val: Bytes,
        resp: Responder<()>,
    },
}

/// Provided by the requester and used by the manager task to send
/// the command response back to the requester.
type Responder<T> = oneshot::Sender<mini_redis::Result<T>>;

Now, update the tasks issuing the commands to include the oneshot::Sender.

rust

# use tokio::sync::{oneshot, mpsc};
# use bytes::Bytes;
# #[derive(Debug)]
# enum Command {
#     Get { key: String, resp: Responder<Option> },
#     Set { key: String, val: Bytes, resp: Responder<()> },
# }
# type Responder<T> = oneshot::Sender<mini_redis::Result<T>>;
# fn dox() {
# let (mut tx, mut rx) = mpsc::channel(10);
# let mut tx2 = tx.clone();
let t1 = tokio::spawn(async move {
    let (resp_tx, resp_rx) = oneshot::channel();
    let cmd = Command::Get {
        key: "foo".to_string(),
        resp: resp_tx,
    };

    // Send the GET request
    tx.send(cmd).await.unwrap();

    // Await the response
    let res = resp_rx.await;
    println!("GOT = {:?}", res);
});

let t2 = tokio::spawn(async move {
    let (resp_tx, resp_rx) = oneshot::channel();
    let cmd = Command::Set {
        key: "foo".to_string(),
        val: "bar".into(),
        resp: resp_tx,
    };

    // Send the SET request
    tx2.send(cmd).await.unwrap();

    // Await the response
    let res = resp_rx.await;
    println!("GOT = {:?}", res);
});
# }

Finally, update the manager task to send the response over the oneshot channel.

rust

# use tokio::sync::{oneshot, mpsc};
# use bytes::Bytes;
# #[derive(Debug)]
# enum Command {
#     Get { key: String, resp: Responder<Option> },
#     Set { key: String, val: Bytes, resp: Responder<()> },
# }
# type Responder<T> = oneshot::Sender<mini_redis::Result<T>>;
# async fn dox(mut client: mini_redis::client::Client) {
# let (_, mut rx) = mpsc::channel::<Command>(10);
while let Some(cmd) = rx.recv().await {
    match cmd {
        Command::Get { key, resp } => {
            let res = client.get(&key).await;
            // Ignore errors
            let _ = resp.send(res);
        }
        Command::Set { key, val, resp } => {
            let res = client.set(&key, val).await;
            // Ignore errors
            let _ = resp.send(res);
        }
    }
}
# }

Calling send on oneshot::Sender completes immediately and does not
require an .await. This is because send on a oneshot channel will always
fail or succeed immediately without any form of waiting.

Sending a value on a oneshot channel returns Err when the receiver half has
dropped. This indicates the receiver is no longer interested in the response. In
our scenario, the receiver cancelling interest is an acceptable event. The Err
returned by resp.send(...) does not need to be handled.

You can find the entire code here.

Backpressure and bounded channels

Whenever concurrency or queuing is introduced, it is important to ensure that the
queing is bounded and the system will gracefully handle the load. Unbounded queues
will eventually fill up all available memory and cause the system to fail in
unpredictable ways.

Tokio takes care to avoid implicit queuing. A big part of this is the fact that
async operations are lazy. Consider the following:

rust

# fn async_op() {}
# fn dox() {
loop {
    async_op();
}
# }
# fn main() {}

If the asynchronous operation runs eagerly, the loop will repeatedly queue a new
async_op to run without ensuring the previous operation completed. This
results in implicit unbounded queuing. Callback based systems and eager
future based systems are particularly susceptible to this.

However, with Tokio and asynchronous Rust, the above snippet will not result
in async_op running at all. This is because .await is never called. If the
snippet is updated to use .await, then the loop waits for the operation to
complete before starting over.

rust

# async fn async_op() {}
# async fn dox() {
loop {
    // Will not repeat until `async_op` completes
    async_op().await;
}
# }
# fn main() {}

Concurrency and queuing must be explicitly introduced. Ways to do this include:

tokio::spawn
select!
join!
mpsc::channel

When doing so, take care to ensure the total amount of concurrency is bounded. For
example, when writing a TCP accept loop, ensure that the total number of open
sockets is bounded. When using mpsc::channel, pick a manageable channel
capacity. Specific bound values will be application specific.

Taking care and picking good bounds is a big part of writing reliable Tokio applications.

File: content/tokio/tutorial/framing.md

title: "Framing"

We will now apply what we just learned about I/O and implement the Mini-Redis
framing layer. Framing is the process of taking a byte stream and converting it
to a stream of frames. A frame is a unit of data transmitted between two peers.
The Redis protocol frame is defined as follows:

rust

use bytes::Bytes;

enum Frame {
    Simple(String),
    Error(String),
    Integer(u64),
    Bulk(Bytes),
    Null,
    Array(Vec<Frame>),
}

Note how the frame only consists of data without any semantics. The command
parsing and implementation happen at a higher level.

For HTTP, a frame might look like:

rust

# use bytes::Bytes;
# type Method = ();
# type Uri = ();
# type Version = ();
# type HeaderMap = ();
# type StatusCode = ();
enum HttpFrame {
    RequestHead {
        method: Method,
        uri: Uri,
        version: Version,
        headers: HeaderMap,
    },
    ResponseHead {
        status: StatusCode,
        version: Version,
        headers: HeaderMap,
    },
    BodyChunk {
        chunk: Bytes,
    },
}

To implement framing for Mini-Redis, we will implement a Connection struct
that wraps a TcpStream and reads/writes mini_redis::Frame values.

rust

use tokio::net::TcpStream;
use mini_redis::{Frame, Result};

struct Connection {
    stream: TcpStream,
    // ... other fields here
}

impl Connection {
    /// Read a frame from the connection.
    /// 
    /// Returns `None` if EOF is reached
    pub async fn read_frame(&mut self)
        -> Result<Option<Frame>>
    {
        // implementation here
# unimplemented!();
    }

    /// Write a frame to the connection.
    pub async fn write_frame(&mut self, frame: &Frame)
        -> Result<()>
    {
        // implementation here
# unimplemented!();
    }
}

You can find the details of the Redis wire protocol here. The full
Connection code is found here.

Buffered reads

The read_frame method waits for an entire frame to be received before
returning. A single call to TcpStream::read() may return an arbitrary amount
of data. It could contain an entire frame, a partial frame, or multiple frames.
If a partial frame is received, the data is buffered and more data is read from
the socket. If multiple frames are received, the first frame is returned and
the rest of the data is buffered until the next call to read_frame.

If you haven't already, create a new file called connection.rs.

bash

touch src/connection.rs

To implement this, Connection needs a read buffer field. Data is read from the
socket into the read buffer. When a frame is parsed, the corresponding data is
removed from the buffer.

We will use BytesMut as the buffer type. This is a mutable version of
Bytes.

rust

use bytes::BytesMut;
use tokio::net::TcpStream;

pub struct Connection {
    stream: TcpStream,
    buffer: BytesMut,
}

impl Connection {
    pub fn new(stream: TcpStream) -> Connection {
        Connection {
            stream,
            // Allocate the buffer with 4kb of capacity.
            buffer: BytesMut::with_capacity(4096),
        }
    }
}

Next, we implement the read_frame() method.

rust

use tokio::io::AsyncReadExt;
use bytes::Buf;
use mini_redis::Result;

# struct Connection {
#   stream: tokio::net::TcpStream,
#   buffer: bytes::BytesMut,
# }
# struct Frame {}
# impl Connection {
pub async fn read_frame(&mut self)
    -> Result<Option<Frame>>
{
    loop {
        // Attempt to parse a frame from the buffered data. If
        // enough data has been buffered, the frame is
        // returned.
        if let Some(frame) = self.parse_frame()? {
            return Ok(Some(frame));
        }

        // There is not enough buffered data to read a frame.
        // Attempt to read more data from the socket.
        //
        // On success, the number of bytes is returned. `0`
        // indicates "end of stream".
        if 0 == self.stream.read_buf(&mut self.buffer).await? {
            // The remote closed the connection. For this to be
            // a clean shutdown, there should be no data in the
            // read buffer. If there is, this means that the
            // peer closed the socket while sending a frame.
            if self.buffer.is_empty() {
                return Ok(None);
            } else {
                return Err("connection reset by peer".into());
            }
        }
    }
}
# fn parse_frame(&self) -> Result<Option<Frame>> { unimplemented!() }
# }

Let's break this down. The read_frame method operates in a loop. First,
self.parse_frame() is called. This will attempt to parse a redis frame from
self.buffer. If there is enough data to parse a frame, the frame is returned
to the caller of read_frame().Otherwise, we attempt to read more data from the
socket into the buffer. After reading more data, parse_frame() is called
again. This time, if enough data has been received, parsing may succeed.

When reading from the stream, a return value of 0 indicates that no more data
will be received from the peer. If the read buffer still has data in it, this
indicates a partial frame has been received and the connection is being
terminated abruptly. This is an error condition and Err is returned.

The `Buf` trait

When reading from the stream, read_buf is called. This version of the read
function takes a value implementing BufMut from the bytes crate.

First, consider how we would implement the same read loop using read().
Vec could be used instead of BytesMut.

rust

use tokio::net::TcpStream;

pub struct Connection {
    stream: TcpStream,
    buffer: Vec,
    cursor: usize,
}

impl Connection {
    pub fn new(stream: TcpStream) -> Connection {
        Connection {
            stream,
            // Allocate the buffer with 4kb of capacity.
            buffer: vec![0; 4096],
            cursor: 0,
        }
    }
}

And the read_frame() function on Connection:

rust

use mini_redis::{Frame, Result};

# use tokio::io::AsyncReadExt;
# pub struct Connection {
#     stream: tokio::net::TcpStream,
#     buffer: Vec,
#     cursor: usize,
# }
# impl Connection {
pub async fn read_frame(&mut self)
    -> Result<Option<Frame>>
{
    loop {
        if let Some(frame) = self.parse_frame()? {
            return Ok(Some(frame));
        }

        // Ensure the buffer has capacity
        if self.buffer.len() == self.cursor {
            // Grow the buffer
            self.buffer.resize(self.cursor * 2, 0);
        }

        // Read into the buffer, tracking the number
        // of bytes read
        let n = self.stream.read(
            &mut self.buffer[self.cursor..]).await?;

        if 0 == n {
            if self.cursor == 0 {
                return Ok(None);
            } else {
                return Err("connection reset by peer".into());
            }
        } else {
            // Update our cursor
            self.cursor += n;
        }
    }
}
# fn parse_frame(&mut self) -> Result<Option<Frame>> { unimplemented!() }
# }

When working with byte arrays and read, we must also maintain a cursor
tracking how much data has been buffered. We must make sure to pass the empty
portion of the buffer to read(). Otherwise, we would overwrite buffered data.
If our buffer gets filled up, we must grow the buffer in order to keep reading.
In parse_frame() (not included), we would need to parse data contained by
self.buffer[..self.cursor].

Because pairing a byte array with a cursor is very common, the bytes crate
provides an abstraction representing a byte array and cursor. The Buf trait is
implemented by types from which data can be read. The BufMut trait is
implemented by types into which data can be written. When passing a T: BufMut
to read_buf(), the buffer's internal cursor is automatically updated by
read_buf. Because of this, in our version of read_frame, we do not need to
manage our own cursor.

Additionally, when using Vec, the buffer must be initialized. vec![0; 4096] allocates an array of 4096 bytes and writes zero to every entry. When
resizing the buffer, the new capacity must also be initialized with zeros. The
initialization process is not free. When working with BytesMut and BufMut,
capacity is uninitialized. The BytesMut abstraction prevents us from
reading the uninitialized memory. This lets us avoid the initialization step.

Parsing

Now, let's look at the parse_frame() function. Parsing is done in two steps.

Ensure a full frame is buffered and find the end index of the frame.
Parse the frame.

The mini-redis crate provides us with a function for both of these steps:

We will also reuse the Buf abstraction to help. A Buf is passed into
Frame::check. As the check function iterates the passed in buffer, the
internal cursor will be advanced. When check returns, the buffer's internal
cursor points to the end of the frame.

For the Buf type, we will use std::io::Cursor<&[u8]>.

rust

use mini_redis::{Frame, Result};
use mini_redis::frame::Error::Incomplete;
use bytes::Buf;
use std::io::Cursor;

# pub struct Connection {
#     stream: tokio::net::TcpStream,
#     buffer: bytes::BytesMut,
# }
# impl Connection {
fn parse_frame(&mut self)
    -> Result<Option<Frame>>
{
    // Create the `T: Buf` type.
    let mut buf = Cursor::new(&self.buffer[..]);

    // Check whether a full frame is available
    match Frame::check(&mut buf) {
        Ok(_) => {
            // Get the byte length of the frame
            let len = buf.position() as usize;

            // Reset the internal cursor for the
            // call to `parse`.
            buf.set_position(0);

            // Parse the frame
            let frame = Frame::parse(&mut buf)?;

            // Discard the frame from the buffer
            self.buffer.advance(len);

            // Return the frame to the caller.
            Ok(Some(frame))
        }
        // Not enough data has been buffered
        Err(Incomplete) => Ok(None),
        // An error was encountered
        Err(e) => Err(e.into()),
    }
}
# }

The full Frame::check function can be found here. We will
not cover it in its entirety.

The relevant thing to note is that Buf's "byte iterator" style APIs are used.
These fetch data and advance the internal cursor. For example, to parse a frame,
the first byte is checked to determine the type of the frame. The function used
is Buf::get_u8. This fetches the byte at the current cursor's position and
advances the cursor by one.

There are more useful methods on the Buf trait. Check the API docs
for more details.

Buffered writes

The other half of the framing API is the write_frame(frame) function. This
function writes an entire frame to the socket. In order to minimize write
syscalls, writes will be buffered. A write buffer is maintained and frames are
encoded to this buffer before being written to the socket. However, unlike
read_frame(), the entire frame is not always buffered to a byte array before
writing to the socket.

Consider a bulk stream frame. The value being written is Frame::Bulk(Bytes).
The wire format of a bulk frame is a frame head, which consists of the $
character followed by the data length in bytes. The majority of the frame is the
contents of the Bytes value. If the data is large, copying it to an
intermediate buffer would be costly.

To implement buffered writes, we will use the BufWriter struct.
This struct is initialized with a T: AsyncWrite and implements AsyncWrite
itself. When write is called on BufWriter, the write does not go directly to
the inner writer, but to a buffer. When the buffer is full, the contents are
flushed to the inner writer and the inner buffer is cleared. There are also
optimizations that allow bypassing the buffer in certain cases.

We will not attempt a full implementation of write_frame() as part of the
tutorial. See the full implementation here.

First, the Connection struct is updated:

rust

use tokio::io::BufWriter;
use tokio::net::TcpStream;
use bytes::BytesMut;

pub struct Connection {
    stream: BufWriter<TcpStream>,
    buffer: BytesMut,
}

impl Connection {
    pub fn new(stream: TcpStream) -> Connection {
        Connection {
            stream: BufWriter::new(stream),
            buffer: BytesMut::with_capacity(4096),
        }
    }
}

Next, write_frame() is implemented.

rust

use tokio::io::{self, AsyncWriteExt};
use mini_redis::Frame;

# struct Connection {
#   stream: tokio::io::BufWriter<tokio::net::TcpStream>,
#   buffer: bytes::BytesMut,
# }
# impl Connection {
async fn write_frame(&mut self, frame: &Frame)
    -> io::Result<()>
{
    match frame {
        Frame::Simple(val) => {
            self.stream.write_u8(b'+').await?;
            self.stream.write_all(val.as_bytes()).await?;
            self.stream.write_all(b"\r\n").await?;
        }
        Frame::Error(val) => {
            self.stream.write_u8(b'-').await?;
            self.stream.write_all(val.as_bytes()).await?;
            self.stream.write_all(b"\r\n").await?;
        }
        Frame::Integer(val) => {
            self.stream.write_u8(b':').await?;
            self.write_decimal(*val).await?;
        }
        Frame::Null => {
            self.stream.write_all(b"$-1\r\n").await?;
        }
        Frame::Bulk(val) => {
            let len = val.len();

            self.stream.write_u8(b'$').await?;
            self.write_decimal(len as u64).await?;
            self.stream.write_all(val).await?;
            self.stream.write_all(b"\r\n").await?;
        }
        Frame::Array(_val) => unimplemented!(),
    }

    self.stream.flush().await;

    Ok(())
}
# async fn write_decimal(&mut self, val: u64) -> io::Result<()> { unimplemented!() }
# }

The functions used here are provided by AsyncWriteExt. They are available on
TcpStream as well, but it would not be advisable to issue single byte writes
without the intermediate buffer.

write_u8 writes a single byte to the writer.
write_all writes the entire slice to the writer.
write_decimal is implemented by mini-redis.

The function ends with a call to self.stream.flush().await. Because
BufWriter stores writes in an intermediate buffer, calls to write do not
guarantee that the data is written to the socket. Before returning, we want the
frame to be written to the socket. The call to flush() writes any data pending
in the buffer to the socket.

Another alternative would be to not call flush() in write_frame().
Instead, provide a flush() function on Connection. This would allow the
caller to write queue multiple small frames in the write buffer then write them
all to the socket with one write syscall. Doing this complicates the
Connection API. Simplicity is one of Mini-Redis' goals, so we decided to
include the flush().await call in fn write_frame().

1. Project Overview & Quickstart (tokio-rs/tracing) Copy Chapter

File: README.md Copy Chapter

Overview Copy Chapter

Branch set-up Copy Section

Usage Copy Chapter

In Applications Copy Section

In Libraries Copy Section

In Asynchronous Code Copy Section

Supported Rust Versions Copy Chapter

Getting Help Copy Chapter

Contributing Copy Chapter

Project layout Copy Chapter

Related Crates Copy Chapter

External Resources Copy Chapter

Blog Posts

Talks

License Copy Chapter

Contribution Copy Section

File: examples/README.md Copy Chapter

Tracing Examples

File: tracing/README.md Copy Chapter

tracing

Overview Copy Chapter

Usage Copy Chapter

In Applications Copy Section

In Libraries Copy Section

In Asynchronous Code Copy Section

Concepts Copy Section

Supported Rust Versions Copy Chapter

Ecosystem Copy Chapter

Related Crates Copy Section

Supported Rust Versions Copy Chapter

License Copy Chapter

Contribution Copy Section

File: tracing-appender/README.md Copy Chapter

tracing-appender

Overview Copy Chapter

Usage Copy Chapter

Rolling File Appender Copy Chapter

Non-Blocking Writer Copy Chapter

Non-Blocking Rolling File Appender Copy Chapter

Supported Rust Versions Copy Chapter

License Copy Chapter

Contribution Copy Section

File: tracing-attributes/README.md Copy Chapter

tracing-attributes

Overview Copy Chapter

Usage Copy Chapter

Supported Rust Versions Copy Chapter

License Copy Chapter

Contribution Copy Section

File: tracing-core/README.md Copy Chapter

tracing-core

Overview Copy Chapter

Usage Copy Chapter

Crate Feature Flags Copy Section

Supported Rust Versions Copy Chapter

License Copy Chapter

Contribution Copy Section

File: tracing-error/README.md Copy Chapter

tracing-error

Overview

Usage Copy Chapter

Feature Flags Copy Chapter

Supported Rust Versions Copy Chapter

Related Crates Copy Chapter

License Copy Chapter

Contribution Copy Section

File: tracing-flame/README.md Copy Chapter

tracing-flame

Overview

Usage Copy Chapter

Layer Setup Copy Chapter

Generating the Image Copy Chapter

Differences between `flamegraph`s and `flamechart`s Copy Chapter

Supported Rust Versions Copy Chapter

License Copy Chapter

Contribution Copy Section

File: tracing-futures/README.md Copy Chapter

tracing-futures

1. Project Overview & Quickstart (tokio-rs/tracing)

File: README.md

Overview

Branch set-up

Usage

In Applications

In Libraries

In Asynchronous Code

Supported Rust Versions

Getting Help

Contributing

Project layout

Related Crates

External Resources

License

Contribution

File: examples/README.md

File: tracing/README.md

Overview

Usage

In Applications

In Libraries

In Asynchronous Code

Concepts

Supported Rust Versions

Ecosystem

Related Crates

Supported Rust Versions

License

Contribution

File: tracing-appender/README.md

Overview

Usage

Rolling File Appender

Non-Blocking Writer

Non-Blocking Rolling File Appender

Supported Rust Versions

License

Contribution

File: tracing-attributes/README.md

Overview

Usage

Supported Rust Versions

License

Contribution

File: tracing-core/README.md

Overview

Usage

Crate Feature Flags

Supported Rust Versions

License

Contribution

File: tracing-error/README.md

Usage

Feature Flags

Supported Rust Versions

Related Crates

License

Contribution

File: tracing-flame/README.md

Usage

Layer Setup

Generating the Image

Differences between `flamegraph`s and `flamechart`s

Supported Rust Versions

License

Contribution

File: tracing-futures/README.md

Overview

Supported Rust Versions

License

Contribution

File: tracing-journald/README.md

Overview

Supported Rust Versions

License

Contribution

2. Official Technical Reference & Guides (tokio-rs/website)

File: README.md

Contributing