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

## File: README.md

![Tracing — Structured, application-level diagnostics][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]
[![Discord chat][discord-badge]][discord-url]

[crates-badge]: https://img.shields.io/crates/v/tracing.svg
[crates-url]: https://crates.io/crates/tracing
[docs-badge]: https://docs.rs/tracing/badge.svg
[docs-url]: https://docs.rs/tracing
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://tracing.rs
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: LICENSE
[actions-badge]: https://github.com/tokio-rs/tracing/workflows/CI/badge.svg
[actions-url]: https://github.com/tokio-rs/tracing/actions?query=workflow%3ACI
[discord-badge]: https://img.shields.io/discord/500028886025895936?logo=discord&label=discord&logoColor=white
[discord-url]: https://discord.gg/EeF3cQw

[Website](https://tokio.rs) |
[Chat](https://discord.gg/EeF3cQw)

## 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`](https://github.com/tokio-rs/tracing/tree/main) - Default branch, crates.io releases are done from this branch. This was previously the
  `v0.1.x` branch.
- [`v0.2.x`](https://github.com/tokio-rs/tracing/tree/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`][tracing-subscriber-docs]'s [`fmt` module][fmt] 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.

[tracing-subscriber-docs]: https://docs.rs/tracing-subscriber/
[fmt]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/index.html
[`set_global_default`]: https://docs.rs/tracing/latest/tracing/subscriber/fn.set_global_default.html


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.

[`tracing-subscriber`]: https://docs.rs/tracing-subscriber/
[fmt]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/index.html
[`init()`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/fn.init.html
[`set_global_default()`]: https://docs.rs/tracing/latest/tracing/subscriber/fn.set_global_default.html

### 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.

```
/* 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 fn`s, the preferred method is using the [`#[instrument]`][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`][std-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][closing].

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]`][instrument] macro performs the same explicit span
attachment that `Future::instrument` does.

[std-future]: https://doc.rust-lang.org/stable/std/future/trait.Future.html
[closing]: https://docs.rs/tracing/latest/tracing/span/index.html#closing-spans
[`Future::instrument`]: https://docs.rs/tracing/latest/tracing/trait.Instrument.html#method.instrument
[instrument]: https://docs.rs/tracing/latest/tracing/attr.instrument.html

## 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][chat]. We would be happy to try to answer your
question. Last, if that doesn't work, try opening an [issue] with the question.

[chat]: https://discord.gg/EeF3cQw
[issue]: https://github.com/tokio-rs/tracing/issues/new

## Contributing

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

[guide]: CONTRIBUTING.md

## 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][fut-crates]|[docs][fut-docs])

* [`tracing-macros`]: Experimental macros for emitting trace events (unstable).

* [`tracing-attributes`]: Procedural macro attributes for automatically
    instrumenting functions. ([crates.io][attr-crates]|[docs][attr-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 `Subscriber`s.
  ([crates.io][sub-crates]|[docs][sub-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][app-crates]|[docs][app-docs])

* [`tracing-error`]: Provides `SpanTrace`, a type for instrumenting errors with
  tracing spans. ([crates.io][err-crates]|[docs][err-docs])

* [`tracing-flame`]: Provides a layer for generating flame graphs based on
  tracing span entry / exit events. ([crates.io][flame-crates]|[docs][flame-docs])

* [`tracing-journald`]: Provides a layer for recording events to the
  Linux `journald` service, preserving structured data. ([crates.io][jour-crates]|[docs][jour-docs])

[`tracing`]: tracing
[`tracing-core`]: tracing-core
[`tracing-futures`]: tracing-futures
[`tracing-macros`]: tracing-macros
[`tracing-attributes`]: tracing-attributes
[`tracing-log`]: tracing-log
[`tracing-serde`]: tracing-serde
[`tracing-subscriber`]: tracing-subscriber
[`tracing-tower`]: tracing-tower
[`tracing-appender`]: tracing-appender
[`tracing-error`]: tracing-error
[`tracing-flame`]: tracing-flame
[`tracing-journald`]: tracing-journald

[fut-crates]: https://crates.io/crates/tracing-futures
[fut-docs]: https://docs.rs/tracing-futures

[attr-crates]: https://crates.io/crates/tracing-attributes
[attr-docs]: https://docs.rs/tracing-attributes

[sub-crates]: https://crates.io/crates/tracing-subscriber
[sub-docs]: https://docs.rs/tracing-subscriber

[otel-crates]: https://crates.io/crates/tracing-opentelemetry
[otel-docs]: https://docs.rs/tracing-opentelemetry
[OpenTelemetry]: https://opentelemetry.io/

[app-crates]: https://crates.io/crates/tracing-appender
[app-docs]: https://docs.rs/tracing-appender

[err-crates]: https://crates.io/crates/tracing-error
[err-docs]: https://docs.rs/tracing-error

[flame-crates]: https://crates.io/crates/tracing-flame
[flame-docs]: https://docs.rs/tracing-flame

[jour-crates]: https://crates.io/crates/tracing-journald
[jour-docs]: https://docs.rs/tracing-journald

## Related Crates

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!)

[`tracing-timing`]: https://crates.io/crates/tracing-timing
[`tracing-honeycomb`]: https://crates.io/crates/tracing-honeycomb
[`tracing-distributed`]: https://crates.io/crates/tracing-distributed
[honeycomb.io]: https://www.honeycomb.io/
[`tracing-actix`]: https://crates.io/crates/tracing-actix
[`tracing-actix-web`]: https://crates.io/crates/tracing-actix-web
[`axum-insights`]: https://crates.io/crates/axum-insights
[`tracing-gelf`]: https://crates.io/crates/tracing-gelf
[`tracing-coz`]: https://crates.io/crates/tracing-coz
[coz]: https://github.com/plasma-umass/coz
[`tracing-bunyan-formatter`]: https://crates.io/crates/tracing-bunyan-formatter
[`tide-tracing`]: https://crates.io/crates/tide-tracing
[tide]: https://crates.io/crates/tide
[bunyan]: https://github.com/trentm/node-bunyan
[`color-spantrace`]: https://docs.rs/color-spantrace
[`color-eyre`]: https://docs.rs/color-eyre
[`spandoc`]: https://docs.rs/spandoc
[`tracing-wasm`]: https://docs.rs/tracing-wasm
[`tracing-web`]: https://crates.io/crates/tracing-web
[`test-log`]: https://crates.io/crates/test-log
[User Timing API (`window.performance`)]: https://developer.mozilla.org/en-US/docs/Web/API/User_Timing_API
[`tracing-unwrap`]: https://docs.rs/tracing-unwrap
[`diesel`]: https://crates.io/crates/diesel
[`diesel-tracing`]: https://crates.io/crates/diesel-tracing
[`tracing-tracy`]: https://crates.io/crates/tracing-tracy
[Tracy]: https://github.com/wolfpld/tracy
[`tracing-elastic-apm`]: https://crates.io/crates/tracing-elastic-apm
[Elastic APM]: https://www.elastic.co/apm
[`tracing-etw`]: https://github.com/microsoft/rust_win_etw/tree/main/win_etw_tracing
[ETW]: https://docs.microsoft.com/en-us/windows/win32/etw/about-event-tracing
[`sentry-tracing`]: https://crates.io/crates/sentry-tracing
[Sentry]: https://sentry.io/welcome/
[`tracing-forest`]: https://crates.io/crates/tracing-forest
[`tracing-loki`]: https://crates.io/crates/tracing-loki
[Grafana Loki]: https://grafana.com/oss/loki/
[`tracing-logfmt`]: https://crates.io/crates/tracing-logfmt
[`tracing-chrome`]: https://crates.io/crates/tracing-chrome
[`reqwest-tracing`]: https://crates.io/crates/reqwest-tracing
[`reqwest`]: https://crates.io/crates/reqwest
[`tracing-cloudwatch`]: https://crates.io/crates/tracing-cloudwatch
[`tracing-subscriber-reload-arcswap`]: https://crates.io/crates/tracing-subscriber-reload-arcswap
[`clippy-tracing`]: https://crates.io/crates/clippy-tracing

**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][tokio-blog-2019-08] on the Tokio blog, August 2019
* [Production-Grade Logging in Rust Applications][production-logging-2020], November 2020
* [Custom Logging in Rust using `tracing` and `tracing-subscriber`, part 1][custom-logging-part-1] and [part 2][custom-logging-part-2], October 2021
* [Instrumenting Axum projects][detsys-blog-2023-08], August 2023

[tokio-blog-2019-08]: https://tokio.rs/blog/2019-08-tracing/
[detsys-blog-2023-08]: https://determinate.systems/posts/instrumenting-axum

#### Talks

* [Bay Area Rust Meetup talk and Q&A][bay-rust-2019-03], March 2019
* [RustConf 2019 talk][rust-conf-2019-08-video] and [slides][rust-conf-2019-08-slides], August 2019
* [Are we observable yet? @ RustyDays talk][rusty-days-2020-08-video] and [slides][rusty-days-2020-08-slides], August 2020
* [Crabs with instruments!][tremorcon-2021-09], September 2021
* [Decrusting the tracing crate by Jon Gjengset][decrusting-tracing-2024-02], February 2024

[bay-rust-2019-03]: https://www.youtube.com/watch?v=j_kXRg3zlec
[rust-conf-2019-08-video]: https://www.youtube.com/watch?v=JjItsfqFIdo
[rust-conf-2019-08-slides]: https://www.elizas.website/slides/rustconf-8-2019.pdf
[rusty-days-2020-08-video]: https://youtu.be/HtKnLiFwHJM
[rusty-days-2020-08-slides]: https://docs.google.com/presentation/d/1zrxJs7fJgQ29bKfnAll1bYTo9cYZxsCZUwDDtyp5Fak/edit?usp=sharing
[production-logging-2020]: https://medium.com/better-programming/production-grade-logging-in-rust-applications-2c7fffd108a6
[custom-logging-part-1]: https://burgers.io/custom-logging-in-rust-using-tracing
[custom-logging-part-2]: https://burgers.io/custom-logging-in-rust-using-tracing-part-2
[tremorcon-2021-09]: https://www.youtube.com/watch?v=ZC7fyqshun8
[decrusting-tracing-2024-02]: https://www.youtube.com/watch?v=21rtHinFA40

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](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`].
  + `fmt-compact`: Demonstrates the output of the [`format::Compact`] formatter in
    [`tracing-subscriber::fmt`][`fmt`].
  + `fmt-json`: Demonstrates the output of the [`format::Json`] formatter in
    [`tracing-subscriber::fmt`][`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][tasks] created with 
    `tokio::spawn`.
  + `futures-proxy-server`: Demonstrates the use of `tracing-futures` by
    implementing a simple proxy server, based on [this example][tokio-proxy]
    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][echo] from `hyper`, and using `tracing-log` to record logs
    emitted by `hyper`.

[tasks]: (https://docs.rs/tokio/0.2.21/tokio/task/index.html)
[tokio-proxy]: https://github.com/tokio-rs/tokio/blob/v0.1.x/tokio/examples/proxy.rs
[echo]: https://github.com/hyperium/hyper/blob/0.12.x/examples/echo.rs
[`fmt`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/index.html
[`format::Full`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/format/struct.Full.html
[`format::Pretty`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/format/struct.Pretty.html
[`format::Compact`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/format/struct.Compact.html
[`format::Json`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/format/struct.Json.html
[`fmt::Layer`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/struct.Layer.html

---

## File: tracing/README.md

![Tracing — Structured, application-level diagnostics][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

# tracing

Application-level tracing for Rust.

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]
[![Discord chat][discord-badge]][discord-url]

[Documentation][docs-url] | [Chat][discord-url]

[crates-badge]: https://img.shields.io/crates/v/tracing.svg
[crates-url]: https://crates.io/crates/tracing/0.1.44
[docs-badge]: https://docs.rs/tracing/badge.svg
[docs-url]: https://docs.rs/tracing/0.1.44
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://tracing.rs/tracing
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: LICENSE
[actions-badge]: https://github.com/tokio-rs/tracing/workflows/CI/badge.svg
[actions-url]:https://github.com/tokio-rs/tracing/actions?query=workflow%3ACI
[discord-badge]: https://img.shields.io/discord/500028886025895936?logo=discord&label=discord&logoColor=white
[discord-url]: https://discord.gg/EeF3cQw

## 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+][msrv]*

[msrv]: #supported-rust-versions

## Usage

(The examples below are borrowed from the `log` crate's yak-shaving
[example](https://docs.rs/log/0.4.10/log/index.html#examples), 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`](https://docs.rs/tracing-subscriber/)'s
[`fmt` module](https://docs.rs/tracing-subscriber/0.3/tracing_subscriber/fmt/index.html) 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");
}
```

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.

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.

```
/* 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`](https://doc.rust-lang.org/stable/std/future/trait.Future.html)
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();
    // ...
}
```

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](https://tracing.rs/tracing/span/index.html#closing-spans).

There are two ways to instrument asynchronous code. The first is through the
[`Future::instrument`](https://docs.rs/tracing/latest/tracing/trait.Instrument.html#method.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]`](https://docs.rs/tracing/0.1.44/tracing/attr.instrument.html)
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 `Span`s and `Event`s, 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.

`Span`s 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]`](https://docs.rs/tracing/0.1.44/tracing/attr.instrument.html) 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 `Event`s (`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

### Related Crates

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 `Subscriber`s 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 `Future`s, `Stream`s, and `Executor`s.
- [`tracing-subscriber`] provides `Subscriber` implementations and
  utilities for working with `Subscriber`s. 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` `Event`s 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!

[`tracing-timing`]: https://crates.io/crates/tracing-timing
[`tracing-opentelemetry`]: https://crates.io/crates/tracing-opentelemetry
[OpenTelemetry]: https://opentelemetry.io/
[`tracing-honeycomb`]: https://crates.io/crates/tracing-honeycomb
[`tracing-distributed`]: https://crates.io/crates/tracing-distributed
[honeycomb.io]: https://www.honeycomb.io/
[`tracing-actix`]: https://crates.io/crates/tracing-actix
[`axum-insights`]: https://crates.io/crates/axum-insights
[`tracing-gelf`]: https://crates.io/crates/tracing-gelf
[`tracing-coz`]: https://crates.io/crates/tracing-coz
[coz]: https://github.com/plasma-umass/coz
[`test-log`]: https://crates.io/crates/test-log
[`tracing-unwrap`]: https://docs.rs/tracing-unwrap
[`diesel`]: https://crates.io/crates/diesel
[`diesel-tracing`]: https://crates.io/crates/diesel-tracing
[`tracing-tracy`]: https://crates.io/crates/tracing-tracy
[Tracy]: https://github.com/wolfpld/tracy
[`tracing-elastic-apm`]: https://crates.io/crates/tracing-elastic-apm
[Elastic APM]: https://www.elastic.co/apm
[`tracing-etw`]: https://github.com/microsoft/tracing-etw
[ETW]: https://docs.microsoft.com/en-us/windows/win32/etw/about-event-tracing
[`tracing-fluent-assertions`]: https://crates.io/crates/tracing-fluent-assertions
[`sentry-tracing`]: https://crates.io/crates/sentry-tracing
[Sentry]: https://sentry.io/welcome/
[`tracing-loki`]: https://crates.io/crates/tracing-loki
[Grafana Loki]: https://grafana.com/oss/loki/
[`tracing-logfmt`]: https://crates.io/crates/tracing-logfmt
[`json-subscriber`]: https://crates.io/crates/json-subscriber

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

[`log`]: https://docs.rs/log/0.4.6/log/
[`tokio-rs/tracing`]: https://github.com/tokio-rs/tracing
[`tracing-futures`]: https://github.com/tokio-rs/tracing/tree/main/tracing-futures
[`tracing-subscriber`]: https://github.com/tokio-rs/tracing/tree/main/tracing-subscriber
[`tracing-log`]: https://github.com/tokio-rs/tracing/tree/main/tracing-log
[`env_logger`]: https://crates.io/crates/env_logger
[`FmtSubscriber`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/struct.Subscriber.html
[`examples`]: https://github.com/tokio-rs/tracing/tree/main/examples

## 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.

## License

This project is licensed under the [MIT license](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][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

# tracing-appender

Writers for logging events and spans

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]
[![Discord chat][discord-badge]][discord-url]

[Documentation][docs-url] | [Chat][discord-url]

[crates-badge]: https://img.shields.io/crates/v/tracing-appender.svg
[crates-url]: https://crates.io/crates/tracing-appender/0.2.5
[docs-badge]: https://docs.rs/tracing-appender/badge.svg
[docs-url]: https://docs.rs/tracing-appender/0.2.5
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://docs.rs/tracing-appender/
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: ../LICENSE
[actions-badge]: https://github.com/tokio-rs/tracing/workflows/CI/badge.svg
[actions-url]:https://github.com/tokio-rs/tracing/actions?query=workflow%3ACI
[discord-badge]: https://img.shields.io/discord/500028886025895936?logo=discord&label=discord&logoColor=white
[discord-url]: https://discord.gg/EeF3cQw

## Overview

[`tracing`][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`][file_appender] 
that can be used with _or_ without the non-blocking writer.

*Compiler support: [requires `rustc` 1.63+][msrv]*

[msrv]: #supported-rust-versions

## 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`][file_appender] to write to a log file. 
 This is a blocking operation.
 - Using *any* type implementing [`std::io::Write`][write] in a 
 non-blocking fashion.
 - Using [`NonBlocking`][non_blocking] and [`RollingFileAppender`][file_appender] 
 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`][write]. To be used with 
[`tracing_subscriber::FmtSubscriber`][fmt_subscriber], it must be combined 
with a [`MakeWriter`][make_writer] 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`][write].

```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`][guard] 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`][guard] module for more details.

The example below demonstrates the construction of a 
[`tracing_appender::non_blocking`][non_blocking] writer constructed with 
a [`std::io::Write`][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();
}
```

[tracing]: https://docs.rs/tracing/latest/tracing/
[make_writer]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/trait.MakeWriter.html
[write]: https://doc.rust-lang.org/std/io/trait.Write.html
[non_blocking]: https://docs.rs/tracing-appender/latest/tracing_appender/non_blocking/index.html
[rolling]: https://docs.rs/tracing-appender/latest/tracing_appender/rolling/index.html
[guard]: https://docs.rs/tracing-appender/latest/tracing_appender/non_blocking/struct.WorkerGuard.html
[file_appender]: https://docs.rs/tracing-appender/latest/tracing_appender/rolling/struct.RollingFileAppender.html
[fmt_subscriber]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/struct.Subscriber.html

## 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](../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][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

# tracing-attributes

Macro attributes for application-level tracing.

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]
[![Discord chat][discord-badge]][discord-url]

[Documentation][docs-url] | [Chat][discord-url]

[crates-badge]: https://img.shields.io/crates/v/tracing-attributes.svg
[crates-url]: https://crates.io/crates/tracing-attributes/0.1.31
[docs-badge]: https://docs.rs/tracing-attributes/badge.svg
[docs-url]: https://docs.rs/tracing-attributes/0.1.31
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://tracing.rs/tracing_attributes
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: LICENSE
[actions-badge]: https://github.com/tokio-rs/tracing/workflows/CI/badge.svg
[actions-url]:https://github.com/tokio-rs/tracing/actions?query=workflow%3ACI
[discord-badge]: https://img.shields.io/discord/500028886025895936?logo=discord&label=discord&logoColor=white
[discord-url]: https://discord.gg/EeF3cQw

## 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+][msrv]*

[msrv]: #supported-rust-versions

## 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) {
    // ...
}
```

[`tracing`]: https://crates.io/crates/tracing
[span]: https://docs.rs/tracing/latest/tracing/span/index.html

## 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.

## License

This project is licensed under the [MIT license](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][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

# tracing-core

Core primitives for application-level tracing.

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]
[![Discord chat][discord-badge]][discord-url]

[Documentation][docs-url] | [Chat][discord-url]

[crates-badge]: https://img.shields.io/crates/v/tracing-core.svg
[crates-url]: https://crates.io/crates/tracing-core/0.1.36
[docs-badge]: https://docs.rs/tracing-core/badge.svg
[docs-url]: https://docs.rs/tracing-core/0.1.36
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://tracing.rs/tracing_core
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: LICENSE
[actions-badge]: https://github.com/tokio-rs/tracing/workflows/CI/badge.svg
[actions-url]:https://github.com/tokio-rs/tracing/actions?query=workflow%3ACI
[discord-badge]: https://img.shields.io/discord/500028886025895936?logo=discord&label=discord&logoColor=white
[discord-url]: https://discord.gg/EeF3cQw

## 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 `Subscriber`s.

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+][msrv]*

[msrv]: #supported-rust-versions

## 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`.

[`tracing`]: ../tracing
[`span::Id`]: https://docs.rs/tracing-core/0.1.36/tracing_core/span/struct.Id.html
[`Event`]: https://docs.rs/tracing-core/0.1.36/tracing_core/event/struct.Event.html
[`Subscriber`]: https://docs.rs/tracing-core/0.1.36/tracing_core/subscriber/trait.Subscriber.html
[`Metadata`]: https://docs.rs/tracing-core/0.1.36/tracing_core/metadata/struct.Metadata.html
[`Callsite`]: https://docs.rs/tracing-core/0.1.36/tracing_core/callsite/trait.Callsite.html
[`Field`]: https://docs.rs/tracing-core/0.1.36/tracing_core/field/struct.Field.html
[`FieldSet`]: https://docs.rs/tracing-core/0.1.36/tracing_core/field/struct.FieldSet.html
[`Value`]: https://docs.rs/tracing-core/0.1.36/tracing_core/field/trait.Value.html
[`ValueSet`]: https://docs.rs/tracing-core/0.1.36/tracing_core/field/struct.ValueSet.html
[`Dispatch`]: https://docs.rs/tracing-core/0.1.36/tracing_core/dispatcher/struct.Dispatch.html

## 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.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](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][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

# tracing-error

Utilities for enriching error handling with [`tracing`] diagnostic
information.

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]
[![Discord chat][discord-badge]][discord-url]
![maintenance status][maint-badge]

[Documentation (release)][docs-url] | [Documentation (v0.2.x)][docs-v0.2.x-url] | [Chat][discord-url]

[crates-badge]: https://img.shields.io/crates/v/tracing-error.svg
[crates-url]: https://crates.io/crates/tracing-error/0.2.1
[docs-badge]: https://docs.rs/tracing-error/badge.svg
[docs-url]: https://docs.rs/tracing-error/0.2.1/tracing_error
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://tracing.rs/tracing_error
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: LICENSE
[actions-badge]: https://github.com/tokio-rs/tracing/workflows/CI/badge.svg
[actions-url]:https://github.com/tokio-rs/tracing/actions?query=workflow%3ACI
[discord-badge]: https://img.shields.io/discord/500028886025895936?logo=discord&label=discord&logoColor=white
[discord-url]: https://discord.gg/EeF3cQw
[maint-badge]: https://img.shields.io/badge/maintenance-experimental-blue.svg

# 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 `SpanTrace`s

**Note**: This crate is currently experimental.

*Compiler support: [requires `rustc` 1.65+][msrv]*

[msrv]: #supported-rust-versions

## 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 [`SpanTrace`]s. 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 `SpanTrace`s from
    behind `dyn Error` trait objects.

## 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.

## Related Crates

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.

[`color-spantrace`]: https://github.com/yaahc/color-spantrace
[`color-eyre`]: https://github.com/yaahc/color-eyre

## License

This project is licensed under the [MIT license](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.

[`SpanTrace`]: https://docs.rs/tracing-error/*/tracing_error/struct.SpanTrace.html
[`ErrorLayer`]: https://docs.rs/tracing-error/*/tracing_error/struct.ErrorLayer.html
[`TracedError`]: https://docs.rs/tracing-error/*/tracing_error/struct.TracedError.html
[`InstrumentResult`]: https://docs.rs/tracing-error/*/tracing_error/trait.InstrumentResult.html
[`InstrumentError`]: https://docs.rs/tracing-error/*/tracing_error/trait.InstrumentError.html
[`ExtractSpanTrace`]: https://docs.rs/tracing-error/*/tracing_error/trait.ExtractSpanTrace.html
[`in_current_span()`]: https://docs.rs/tracing-error/*/tracing_error/trait.InstrumentResult.html#tymethod.in_current_span
[span]: https://docs.rs/tracing/latest/tracing/span/index.html
[events]: https://docs.rs/tracing/latest/tracing/struct.Event.html
[`Subscriber`]: https://docs.rs/tracing/latest/tracing/trait.Subscriber.html
[subscriber layer]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/layer/trait.Layer.html
[`tracing`]: https://docs.rs/tracing
[`std::error::Error`]: https://doc.rust-lang.org/stable/std/error/trait.Error.html
[`SpanTrace`]: https://docs.rs/tracing-error/0.2.1/tracing_error/struct.SpanTrace.html
[`ErrorLayer`]: https://docs.rs/tracing-error/0.2.1/tracing_error/struct.ErrorLayer.html

---

## File: tracing-flame/README.md

![Tracing — Structured, application-level diagnostics][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

# tracing-flame

A [tracing] [`Layer`][`FlameLayer`] for generating a folded stack trace for generating flamegraphs
and flamecharts with [`inferno`]

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]
[![Discord chat][discord-badge]][discord-url]
![maintenance status][maint-badge]

[Documentation][docs-url] | [Chat][discord-url]

# 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+][msrv]*

[msrv]: #supported-rust-versions
[post]: http://www.brendangregg.com/flamegraphs.html

## Usage

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

1. Capture textual representation of the spans that are entered and exited
   with [`FlameLayer`].
2. 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

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.

## License

This project is licensed under the [MIT license](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.

[`inferno`]: https://docs.rs/inferno
[`FlameLayer`]: https://docs.rs/tracing-flame/*/tracing_flame/struct.FlameLayer.html
[`FlushGuard`]: https://docs.rs/tracing-flame/*/tracing_flame/struct.FlushGuard.html
[`inferno-flamegraph`]: https://docs.rs/inferno/0.9.5/inferno/index.html#producing-a-flame-graph
[`tracing`]: https://github.com/tokio-rs/tracing/tree/main/tracing
[crates-badge]: https://img.shields.io/crates/v/tracing-flame.svg
[crates-url]: https://crates.io/crates/tracing-flame
[docs-badge]: https://docs.rs/tracing-flame/badge.svg
[docs-url]: https://docs.rs/tracing-flame/0.2.6
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://tracing.rs/tracing_flame
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: LICENSE
[actions-badge]: https://github.com/tokio-rs/tracing/workflows/CI/badge.svg
[actions-url]:https://github.com/tokio-rs/tracing/actions?query=workflow%3ACI
[discord-badge]: https://img.shields.io/discord/500028886025895936?logo=discord&label=discord&logoColor=white
[discord-url]: https://discord.gg/EeF3cQw
[maint-badge]: https://img.shields.io/badge/maintenance-experimental-blue.svg

---

## File: tracing-futures/README.md

![Tracing — Structured, application-level diagnostics][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

# tracing-futures

Utilities for instrumenting futures-based code with [`tracing`].

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
[![Build Status][actions-badge]][actions-url]
[![Discord chat][discord-badge]][discord-url]
![maintenance status][maint-badge]

[Documentation][docs-url] | [Chat][discord-url]

[crates-badge]: https://img.shields.io/crates/v/tracing-futures.svg
[crates-url]: https://crates.io/crates/tracing-futures/0.2.5
[docs-badge]: https://docs.rs/tracing-futures/badge.svg
[docs-url]: https://docs.rs/tracing-futures/0.2.5/tracing_futures
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://tracing.rs/tracing_futures
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: LICENSE
[actions-badge]: https://github.com/tokio-rs/tracing/workflows/CI/badge.svg
[actions-url]:https://github.com/tokio-rs/tracing/actions?query=workflow%3ACI
[discord-badge]: https://img.shields.io/discord/500028886025895936?logo=discord&label=discord&logoColor=white
[discord-url]: https://discord.gg/EeF3cQw
[maint-badge]: https://img.shields.io/badge/maintenance-actively--developed-brightgreen.svg

## 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.

[`Instrument`]: https://docs.rs/tracing-futures/0.2.5/tracing_futures/trait.Instrument.html
[`WithSubscriber`]: https://docs.rs/tracing-futures/0.2.5/tracing_futures/trait.WithSubscriber.html
[span]: https://docs.rs/tracing/latest/tracing/span/index.html
[`Subscriber`]: https://docs.rs/tracing/latest/tracing/subscriber/index.html
[`tracing`]: https://crates.io/crates/tracing

*Compiler support: [requires `rustc` 1.65+][msrv]*

[msrv]: #supported-rust-versions

## 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.

## License

This project is licensed under the [MIT license](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][splash]

[splash]: https://raw.githubusercontent.com/tokio-rs/tracing/main/assets/splash.svg

# tracing-journald

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

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![Documentation (v0.2.x)][docs-v0.2.x-badge]][docs-v0.2.x-url]
[![MIT licensed][mit-badge]][mit-url]
![maintenance status][maint-badge]

[crates-badge]: https://img.shields.io/crates/v/tracing-journald.svg
[crates-url]: https://crates.io/crates/tracing-journald/0.3.2
[docs-badge]: https://docs.rs/tracing-journald/badge.svg
[docs-url]: https://docs.rs/tracing-journald/0.3.2
[docs-v0.2.x-badge]: https://img.shields.io/badge/docs-v0.2.x-blue
[docs-v0.2.x-url]: https://tracing.rs/tracing_journald
[mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
[mit-url]: LICENSE
[maint-badge]: https://img.shields.io/badge/maintenance-experimental-blue.svg

## Overview

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

*Compiler support: [requires `rustc` 1.65+][msrv]*

[msrv]: #supported-rust-versions
[`tracing`]: https://crates.io/crates/tracing
[layer]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/layer/trait.Layer.html
[journald]: https://www.freedesktop.org/software/systemd/man/systemd-journald.service.html

## 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.

## License

This project is licensed under the [MIT license](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.

* [hello-tokio](tutorial-code/hello-tokio/src/main.rs)
* [spawning](tutorial-code/spawning/src/main.rs)
* [shared-state](tutorial-code/shared-state/src/main.rs)
* [channels](tutorial-code/channels/src/main.rs)
* [io](tutorial-code/io)
    * [echo-server-copy](tutorial-code/io/src/echo-server-copy.rs)
    * [echo-server](tutorial-code/io/src/echo-server.rs)
* [mini-tokio](tutorial-code/mini-tokio/src/main.rs)

## 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](http://localhost:3000).

[Next.js]: https://nextjs.org/
[Bulma]: https://bulma.io/

### Resources

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

- [Next.js Documentation](https://nextjs.org/docs) - learn about Next.js features and API.
- [Learn Next.js](https://nextjs.org/learn) - an interactive Next.js tutorial.
- [Bulma documentation](https://bulma.io/documentation/) - learn about Bulma.

## License

This project is licensed under the [MIT license](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:
```
# 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:

 * [`Client::get`]
 * [`Client::set`]
 * [`Client::set_expires`]
 * [`Client::publish`]
 * [`Client::subscribe`]

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:
```rs
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:
```rs
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:
```rs
/// 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].

[`Runtime`]: https://docs.rs/tokio/1/tokio/runtime/struct.Runtime.html
[`block_on`]: https://docs.rs/tokio/1/tokio/runtime/struct.Runtime.html#method.block_on
[`spawn`]: https://docs.rs/tokio/1/tokio/runtime/struct.Runtime.html#method.spawn
[`spawn_blocking`]: https://docs.rs/tokio/1/tokio/task/fn.spawn_blocking.html
[`multi_thread`]: https://docs.rs/tokio/1/tokio/runtime/struct.Builder.html#method.new_multi_thread
[`current_thread`]: https://docs.rs/tokio/1/tokio/runtime/struct.Builder.html#method.new_current_thread
[`enable_all`]: https://docs.rs/tokio/1/tokio/runtime/struct.Builder.html#method.enable_all
[`JoinHandle`]: https://docs.rs/tokio/1/tokio/task/struct.JoinHandle.html
[`tokio::sync::mpsc`]: https://docs.rs/tokio/1/tokio/sync/mpsc/index.html
[`Handle`]: https://docs.rs/tokio/1/tokio/runtime/struct.Handle.html
[`Semaphore`]: https://docs.rs/tokio/1/tokio/sync/struct.Semaphore.html
[`Client`]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html
[`Client::get`]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.get
[`Client::set`]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.set
[`Client::set_expires`]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.set_expires
[`Client::publish`]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.publish
[`Client::subscribe`]: https://docs.rs/mini-redis/0.4/mini_redis/client/struct.Client.html#method.subscribe
[actor]: https://ryhl.io/blog/actors-with-tokio/

---

## 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:

 * [Bridging with sync code](/tokio/topics/bridging)
 * [Graceful shutdown](/tokio/topics/shutdown)
 * [Getting started with Tracing](/tokio/topics/tracing)
 * [Next steps with Tracing](/tokio/topics/tracing-next-steps)
 * [Testing](/tokio/topics/testing)

---

## 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`][server.rs] and
[`src/shutdown.rs`][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`][ctrl_c]
function, which will sleep until such a signal is received. You might use it
like this:
```rs
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][mpsc]
to send the shutdown signal to one place. You can then [select] on
[`ctrl_c`][ctrl_c] and the channel. For example:
```rs
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][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:

1. First, create a new `CancellationToken`.
2. 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.
3. Pass the original or cloned token to the tasks that should respond to cancellation requests.
4. 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:

```rs
// 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.

```rs
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);
}
```

[ctrl_c]: https://docs.rs/tokio/1/tokio/signal/fn.ctrl_c.html
[task tracker]: https://docs.rs/tokio-util/latest/tokio_util/task/task_tracker
[`wait`]: https://docs.rs/tokio-util/latest/tokio_util/task/task_tracker/struct.TaskTracker.html#method.wait
[select]: https://docs.rs/tokio/1/tokio/macro.select.html
[cancellation-tokens]: https://docs.rs/tokio-util/latest/tokio_util/sync/struct.CancellationToken.html
[shutdown.rs]: https://github.com/tokio-rs/mini-redis/blob/master/src/shutdown.rs
[server.rs]: https://github.com/tokio-rs/mini-redis/blob/master/src/server.rs
[mini-redis]: https://github.com/tokio-rs/mini-redis/
[mpsc]: https://docs.rs/tokio/1/tokio/sync/mpsc/index.html

---

## 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"](https://docs.rs/tokio/latest/tokio/attr.test.html#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.

[`tokio::time::Interval::tick`]: https://docs.rs/tokio/1/tokio/time/struct.Interval.html#method.tick
[`tokio::time::sleep`]: https://docs.rs/tokio/1/tokio/time/fn.sleep.html

## 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;
}
```

[`AsyncRead`]: https://docs.rs/tokio/latest/tokio/io/trait.AsyncRead.html
[`AsyncWrite`]: https://docs.rs/tokio/latest/tokio/io/trait.AsyncWrite.html
[`split`]: https://docs.rs/tokio/latest/tokio/net/struct.TcpStream.html#method.split
[`TcpStream`]: https://docs.rs/tokio/latest/tokio/net/struct.TcpStream.html
[`tokio_test::io::Builder`]: https://docs.rs/tokio-test/latest/tokio_test/io/struct.Builder.html

---

## 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 [`Span`]s and [`Event`]s
are *structured*, with the ability to record typed data as well as textual
messages.

[`Span`]: https://docs.rs/tracing/latest/tracing/#spans
[`Event`]: https://docs.rs/tracing/latest/tracing/#events

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

[`tracing`]: https://docs.rs/tracing
[`tracing-subscriber`]: https://docs.rs/tracing-subscriber
[OpenTelemetry]: https://docs.rs/tracing-opentelemetry
[Tokio Console]: https://docs.rs/console-subscriber
[`stdout`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/index.html
[a log file]: https://docs.rs/tracing-appender/latest/tracing_appender/
[`journald`]: https://docs.rs/tracing-journald/latest/tracing_journald/
[profile]: https://docs.rs/tracing-timing/latest/tracing_timing/

# 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).

[*subscriber*]: https://docs.rs/tracing/latest/tracing/#subscribers

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(())
}
```

[`FmtSubscriber`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/index.html

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][fmt-cfg].

In addition to the [`FmtSubscriber`] type from [`tracing-subscriber`], other
`Subscriber`s 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][related-crates] 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 `Layer`s to form a `Subscriber`. See [here][`Layer`] for details on using
`Layer`s.

[fmt-cfg]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/fmt/index.html#configuration
[related-crates]: https://docs.rs/tracing/latest/tracing/index.html#related-crates
[`Layer`]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/layer/index.html

# 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:

1. has a verbosity [level] of `info` (the "middle ground" verbosity),
2. is named `trace_me`,
3. has fields `a` and `b`, whose values are the arguments of `trace_me`

[`instrument`]: https://docs.rs/tracing/latest/tracing/attr.instrument.html

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

[`mini-redis-server`]: ../tutorial/setup#mini-redis

```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:

1. has a verbosity [level] of `info` (the "middle ground" verbosity),
2. is named `Handler::run`,
3. 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.

[level]: https://docs.rs/tracing/latest/tracing/struct.Level.html

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!`]).

[`span!`]: https://docs.rs/tracing/*/tracing/macro.span.html
[`error_span!`]: https://docs.rs/tracing/*/tracing/macro.error_span.html
[`warn_span!`]: https://docs.rs/tracing/*/tracing/macro.warn_span.html
[`info_span!`]: https://docs.rs/tracing/*/tracing/macro.info_span.html
[`debug_span!`]: https://docs.rs/tracing/*/tracing/macro.debug_span.html
[`trace_span!`]: https://docs.rs/tracing/*/tracing/macro.trace_span.html

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)
    }
};
```

[`event!`]: https://docs.rs/tracing/*/tracing/macro.event.html
[`error!`]: https://docs.rs/tracing/*/tracing/macro.error.html
[`warn!`]: https://docs.rs/tracing/*/tracing/macro.warn.html
[`info!`]: https://docs.rs/tracing/*/tracing/macro.info.html
[`debug!`]: https://docs.rs/tracing/*/tracing/macro.debug.html
[`trace!`]: https://docs.rs/tracing/*/tracing/macro.trace.html

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`](https://github.com/tokio-rs/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](https://github.com/tokio-rs/mini-redis),
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):

```sh
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:

```sh
cargo install --locked tokio-console
```

and then run it with:

```sh
tokio-console
```

The initial view you will see is for tokio Tasks that are currently running.
Example: ![tokio-console Task
view](https://raw.githubusercontent.com/tokio-rs/console/main/assets/tasks_list.png)

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](https://github.com/tokio-rs/mini-redis)):

```sh
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 Resource
view](https://raw.githubusercontent.com/tokio-rs/console/main/assets/resources.png)

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](https://docs.rs/tokio-console/latest/tokio_console/#using-the-console).

# Integrating with OpenTelemetry

[OpenTelemetry](https://opentelemetry.io/) (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](https://crates.io/crates/opentelemetry) 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](https://www.jaegertracing.io/), which is a UI for visualizing traces.

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

```sh
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:

```sh
cargo run --bin mini-redis-server
```

In another terminal, run the hello world example (this is available in the
[mini-redis repository](https://github.com/tokio-rs/mini-redis)):

```sh
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](https://github.com/open-telemetry/opentelemetry-rust/tree/main/examples).

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`](https://github.com/tokio-rs/mini-redis#aws-x-ray-example), as well as
the
[`Cargo.toml`](https://github.com/tokio-rs/mini-redis/blob/24d9d9f466d9078c46477bf5c2d68416553b9872/Cargo.toml#L35-L41)
and
[`src/bin/server.rs`](https://github.com/tokio-rs/mini-redis/blob/24d9d9f466d9078c46477bf5c2d68416553b9872/src/bin/server.rs#L59-L94)
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] trait provided by the standard
library. They are values that contain the in-progress asynchronous computation.

The [`std::future::Future`][trait] 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][assoc] `Output` is the type that the future produces once
it completes. The [`Pin`][pin] type is how Rust is able to support borrows in
`async` functions. See the [standard library][pin] 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:

1. Wait until a specific instant in time.
2. Output some text to STDOUT.
3. 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][mini-tokio].

```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][ch].

[`Rc`]: https://doc.rust-lang.org/std/rc/struct.Rc.html
[`Cell`]: https://doc.rust-lang.org/std/cell/struct.Cell.html
[ch]: https://doc.rust-lang.org/book/ch16-04-extensible-concurrency-sync-and-send.html

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`][`Waker`]. The
standard library provides a low-level API to do this using [manual vtable
construction][vtable]. This strategy provides maximum flexibility to
implementors, but requires a bunch of unsafe boilerplate code. Instead of using
[`RawWakerVTable`][vtable] 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`][`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.

```
/* 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:

```
/* 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`][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`][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;
}
```

[assoc]: https://doc.rust-lang.org/book/ch19-03-advanced-traits.html#specifying-placeholder-types-in-trait-definitions-with-associated-types
[trait]: https://doc.rust-lang.org/std/future/trait.Future.html
[pin]: https://doc.rust-lang.org/std/pin/index.html
[`Waker`]: https://doc.rust-lang.org/std/task/struct.Waker.html
[mini-tokio]: https://github.com/tokio-rs/website/blob/master/tutorial-code/mini-tokio/src/main.rs
[vtable]: https://doc.rust-lang.org/std/task/struct.RawWakerVTable.html
[`ArcWake`]: https://docs.rs/futures/0.3/futures/task/trait.ArcWake.html
[`futures`]: https://docs.rs/futures/
[notify]: https://docs.rs/tokio/1/tokio/sync/struct.Notify.html

---

## 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.

[pipelining]: https://redis.io/topics/pipelining

# 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][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][full].

[channels]: https://docs.rs/tokio/1/tokio/sync/index.html
[mpsc]: https://docs.rs/tokio/1/tokio/sync/mpsc/index.html
[oneshot]: https://docs.rs/tokio/1/tokio/sync/oneshot/index.html
[broadcast]: https://docs.rs/tokio/1/tokio/sync/broadcast/index.html
[watch]: https://docs.rs/tokio/1/tokio/sync/watch/index.html
[`async-channel`]: https://docs.rs/async-channel/
[`std::sync::mpsc`]: https://doc.rust-lang.org/stable/std/sync/mpsc/index.html
[`crossbeam::channel`]: https://docs.rs/crossbeam/latest/crossbeam/channel/index.html

# 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][full].

# 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.

[full]: https://github.com/tokio-rs/website/blob/master/tutorial-code/channels/src/main.rs

---

## 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][proto]. The full
`Connection` code is found [here][full].

[proto]: https://redis.io/topics/protocol
[full]: https://github.com/tokio-rs/mini-redis/blob/tutorial/src/connection.rs

# 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`][BytesMutStruct] as the buffer type. This is a mutable version of
[`Bytes`][BytesStruct].

```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.

[BytesMutStruct]: https://docs.rs/bytes/1/bytes/struct.BytesMut.html
[BytesStruct]: https://docs.rs/bytes/1/bytes/struct.Bytes.html

## 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.

[`BufMut`]: https://docs.rs/bytes/1/bytes/buf/trait.BufMut.html
[`bytes`]: https://docs.rs/bytes/

# Parsing

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

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

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

1. [`Frame::check`](https://docs.rs/mini-redis/0.4/mini_redis/frame/enum.Frame.html#method.check)
2. [`Frame::parse`](https://docs.rs/mini-redis/0.4/mini_redis/frame/enum.Frame.html#method.parse)

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]>`][`Cursor`].

```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`][check] function can be found [here][check]. 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][`Buf`]
for more details.

[check]: https://github.com/tokio-rs/mini-redis/blob/tutorial/src/frame.rs#L65-L103
[`Buf::get_u8`]: https://docs.rs/bytes/1/bytes/buf/trait.Buf.html#method.get_u8
[`Buf`]: https://docs.rs/bytes/1/bytes/buf/trait.Buf.html
[`Cursor`]: https://doc.rust-lang.org/stable/std/io/struct.Cursor.html

# 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][buf-writer].
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][write-frame].

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()`.

[buf-writer]: https://docs.rs/tokio/1/tokio/io/struct.BufWriter.html
[write-frame]: https://github.com/tokio-rs/mini-redis/blob/tutorial/src/connection.rs#L159-L184
[`AsyncWriteExt`]: https://docs.rs/tokio/1/tokio/io/trait.AsyncWriteExt.html
[`write_u8`]: https://docs.rs/tokio/1/tokio/io/trait.AsyncWriteExt.html#method.write_u8
[`write_all`]: https://docs.rs/tokio/1/tokio/io/trait.AsyncWriteExt.html#method.write_all
[`write_decimal`]: https://github.com/tokio-rs/mini-redis/blob/tutorial/src/connection.rs#L225-L238