v0.1.x: prepare to release new reactor, executor, and timer (#1751)

## Motivation

The `CurrentThread` runtime's `Handle` is different than the
`CurrentThread` executor's `Handle`. This causes interoperability issues
with custom runtimes that build on the `CurrentThread` executor.

Actix-rt offers a concrete example of the issue: [`System::run_in_executor`][1]
requires a `CurrentThread` executor handle - the change in this PR
allows a `CurrentThread` runtime to be used here.

## Solution

This PR adds `fn into_inner(self)` on the runtime `Handle` that consumes
it and returns the underlying `CurrentThread` executor's `Handle`.

[1]: https://docs.rs/actix-rt/0.2.6/actix_rt/struct.System.html#method.run_in_executor

.cirrus.yml

@@ -0,0 +1,41 @@
+freebsd_instance:
+  image: freebsd-12-0-release-amd64
+
+# Test FreeBSD in a full VM on cirrus-ci.com.  Test the i686 target too, in the
+# same VM.  The binary will be built in 32-bit mode, but will execute on a
+# 64-bit kernel and in a 64-bit environment.  Our tests don't execute any of
+# the system's binaries, so the environment shouldn't matter.
+task:
+  name: FreeBSD 12.0
+  env:
+    LOOM_MAX_DURATION: 10
+  setup_script:
+    - pkg install -y curl
+    - curl https://sh.rustup.rs -sSf --output rustup.sh
+    - sh rustup.sh -y
+    - . $HOME/.cargo/env
+    - rustup target add i686-unknown-freebsd
+    - |
+      # Remove any existing patch statements
+      mv Cargo.toml Cargo.toml.bck
+      sed -n '/\[patch.crates-io\]/q;p' Cargo.toml.bck > Cargo.toml
+
+      # Patch all crates
+      cat ci/patch.toml >> Cargo.toml
+
+      # Print `Cargo.toml` for debugging
+      echo "~~~~ Cargo.toml ~~~~"
+      cat Cargo.toml
+      echo "~~~~~~~~~~~~~~~~~~~~"
+  cargo_cache:
+    folder: $HOME/.cargo/registry
+  test_script:
+    - . $HOME/.cargo/env
+    - cargo test --all
+    - cargo doc --all
+  i686_test_script:
+    - . $HOME/.cargo/env
+    - |
+      cargo test --all --exclude tokio-tls --exclude tokio-macros --target i686-unknown-freebsd
+  before_cache_script:
+    - rm -rf $HOME/.cargo/registry/index

.github/FUNDING.yml

@@ -1,3 +0,0 @@
-# These are supported funding model platforms
-
-github: [tokio-rs]

.github/ISSUE_TEMPLATE.md

@@ -0,0 +1,51 @@
+<!--
+Thank you for reporting an issue.
+
+Please fill in as much of the template below as you're able.
+-->
+
+## Version
+
+<!--
+List the versions of all `tokio` crates you are using. The easiest way to get
+this information is using `cargo-tree`.
+
+`cargo install cargo-tree`
+(see install here: https://github.com/sfackler/cargo-tree)
+
+Then:
+
+`cargo tree | grep tokio`
+-->
+
+## Platform
+
+<!---
+Output of `uname -a` (UNIX), or version and 32 or 64-bit (Windows)
+-->
+
+## Subcrates
+
+<!--
+If known, please specify the affected Tokio sub crates. Otherwise, delete this
+section.
+-->
+
+## Description
+
+<!--
+
+Enter your issue details below this comment.
+
+One way to structure the description:
+
+<short summary of the bug>
+
+I tried this code:
+
+<code sample that causes the bug>
+
+I expected to see this happen: <explanation>
+
+Instead, this happened: <explanation>
+-->

.github/ISSUE_TEMPLATE/bug_report.md

@@ -1,31 +0,0 @@
----
-name: Bug report
-about: Create a report to help us improve
-title: ''
-labels: A-tokio, C-bug
-assignees: ''
-
----
-
-**Version**
-List the versions of all `tokio` crates you are using. The easiest way to get
-this information is using `cargo tree` subcommand:
-
-`cargo tree | grep tokio`
-
-**Platform**
-The output of `uname -a` (UNIX), or version and 32 or 64-bit (Windows)
-
-**Description**
-Enter your issue details here.
-One way to structure the description:
-
-[short summary of the bug]
-
-I tried this code:
-
-[code sample that causes the bug]
-
-I expected to see this happen: [explanation]
-
-Instead, this happened: [explanation]

.github/ISSUE_TEMPLATE/config.yml

@@ -1,4 +0,0 @@
-contact_links:
-  - name: Question
-    url: https://github.com/tokio-rs/tokio/discussions
-    about: Questions about Tokio should be posted as a GitHub discussion.

.github/ISSUE_TEMPLATE/feature_request.md

@@ -1,20 +0,0 @@
----
-name: Feature request
-about: Suggest an idea for this project
-title: ''
-labels: A-tokio, C-feature-request
-assignees: ''
-
----
-
-**Is your feature request related to a problem? Please describe.**
-A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
-
-**Describe the solution you'd like**
-A clear and concise description of what you want to happen.
-
-**Describe alternatives you've considered**
-A clear and concise description of any alternative solutions or features you've considered.
-
-**Additional context**
-Add any other context or screenshots about the feature request here.

.github/PULL_REQUEST_TEMPLATE.md

@@ -5,9 +5,6 @@ the requirements below.
 Bug fixes and new features should include tests.
 
 Contributors guide: https://github.com/tokio-rs/tokio/blob/master/CONTRIBUTING.md
-
-The contributors guide includes instructions for running rustfmt and building the
-documentation, which requires special commands beyond `cargo fmt` and `cargo doc`.
 -->
 
 ## Motivation

.github/buildomat/README.md

@@ -1,20 +0,0 @@
-# Buildomat illumos CI
-
-This directory contains CI configurations for the [illumos] operating system.
-Tokio's illumos CI jobs are run using [Buildomat], a CI system developed by
-Oxide Computer, which supports illumos. See [the Buildomat README] for more
-details.
-
-## illumos-Specific CI Failures
-
-If your pull request's CI build fails on illumos, and you aren't able to easily
-reproduce the failure on other operating systems, don't worry! The
-[tokio-rs/illumos] team is responsible for maintaining Tokio's illumos support,
-and can be called on to assist contributors with illumos-specific issues. Please
-feel free to tag @tokio-rs/illumos to ask for help resolving build failures on
-illumos.
-
-[illumos]: https://www.illumos.org/
-[Buildomat]: https://github.com/oxidecomputer/buildomat
-[the Buildomat README]: https://github.com/oxidecomputer/buildomat
-[tokio-rs/illumos]: https://github.com/orgs/tokio-rs/teams/illumos

.github/buildomat/config.toml

@@ -1,8 +0,0 @@
-# Repository-level Buildomat configuration.
-# See: https://github.com/oxidecomputer/buildomat#per-repository-configuration
-
-# Enable buildomat. This one should be self-explanatory.
-enable = true
-# Allow CI runs for PRs from users outside the `tokio-rs` organization. Our
-# buildomat jobs don't touch any secrets/keys, so this should be fine.
-org_only = false

.github/dependabot.yml

@@ -1,7 +0,0 @@
-version: 2
-updates:
-  - package-ecosystem: "github-actions"
-    directory: "/"
-    schedule:
-      interval: "weekly"
-    open-pull-requests-limit: 10

.github/labeler.yml

@@ -1,42 +0,0 @@
-R-loom-blocking:
-- changed-files:
-  - any-glob-to-any-file:
-    - tokio/src/runtime/blocking/*
-    - tokio/src/runtime/blocking/**/*
-
-R-loom-sync:
-- changed-files:
-  - any-glob-to-any-file:
-    - tokio/src/sync/*
-    - tokio/src/sync/**/*
-
-R-loom-time-driver:
-- changed-files:
-  - any-glob-to-any-file:
-    - tokio/src/runtime/time/*
-    - tokio/src/runtime/time/**/*
-    - tokio/src/runtime/time_alt/*
-    - tokio/src/runtime/time_alt/**/*
-
-R-loom-current-thread:
-- changed-files:
-  - any-glob-to-any-file:
-    - tokio/src/runtime/scheduler/*
-    - tokio/src/runtime/scheduler/current_thread/*
-    - tokio/src/runtime/task/*
-    - tokio/src/runtime/task/**
-
-R-loom-multi-thread:
-- changed-files:
-  - any-glob-to-any-file:
-    - tokio/src/runtime/scheduler/*
-    - tokio/src/runtime/scheduler/multi_thread/*
-    - tokio/src/runtime/scheduler/multi_thread/**
-    - tokio/src/runtime/task/*
-    - tokio/src/runtime/task/**
-
-R-loom-util:
-- changed-files:
-  - any-glob-to-any-file:
-    - tokio-util/src/*
-    - tokio-util/src/**/*

.github/workflows/audit.yml

@@ -1,24 +0,0 @@
-name: Security Audit
-
-on:
-  push:
-    branches:
-      - master
-    paths:
-      - '**/Cargo.toml'
-  schedule:
-    - cron: '0 2 * * *' # run at 2 AM UTC
-
-permissions:
-  contents: read
-
-jobs:
-  cargo-deny:
-    permissions:
-      checks: write
-      contents: read
-      issues: write
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v7
-    - uses: EmbarkStudios/cargo-deny-action@v2

.github/workflows/labeler.yml

@@ -1,25 +0,0 @@
-name: "Pull Request Labeler"
-on:
-- pull_request_target
-
-# See .github/labeler.yml file
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
-  cancel-in-progress: true
-
-permissions:
-  contents: read
-
-jobs:
-  triage:
-    permissions:
-      contents: read  # for actions/labeler to determine modified files
-      pull-requests: write  # for actions/labeler to add labels to PRs
-    runs-on: ubuntu-latest
-    if: github.repository_owner == 'tokio-rs'
-    steps:
-    - uses: actions/labeler@v7
-      with:
-        repo-token: "${{ secrets.GITHUB_TOKEN }}"
-        sync-labels: true

.github/workflows/loom.yml

@@ -1,138 +0,0 @@
-on:
-  push:
-    branches: ["master", "tokio-*.x"]
-  pull_request:
-    types: [labeled, opened, synchronize, reopened]
-    branches: ["master", "tokio-*.x"]
-
-name: Loom
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
-  cancel-in-progress: true
-
-env:
-  RUSTFLAGS: -Dwarnings --cfg loom --cfg tokio_unstable -C debug_assertions
-  LOOM_MAX_PREEMPTIONS: 2
-  LOOM_MAX_BRANCHES: 10000
-  RUST_BACKTRACE: 1
-  # Change to specific Rust release to pin
-  rust_stable: stable
-
-permissions:
-  contents: read
-
-jobs:
-  loom-blocking:
-    name: loom tokio::runtime::spawn_blocking
-    # base_ref is null when it's not a pull request
-    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-blocking') || (github.base_ref == null))
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        # Run the blocking pool loom tests against both `spawn_blocking`
-        # queue implementations.
-        include:
-          - sharded_blocking_queue: "0"
-          - sharded_blocking_queue: "1"
-    steps:
-      - uses: actions/checkout@v7
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-            toolchain: ${{ env.rust_stable }}
-      - uses: Swatinem/rust-cache@v2
-      - name: run tests
-        run: cargo test --lib --release --features full -- --nocapture loom_blocking
-        working-directory: tokio
-        env:
-          TOKIO_UNSTABLE_SHARDED_BLOCKING_QUEUE: ${{ matrix.sharded_blocking_queue }}
-
-  loom-sync:
-    name: loom tokio::sync
-    # base_ref is null when it's not a pull request
-    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-sync') || (github.base_ref == null))
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v7
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-            toolchain: ${{ env.rust_stable }}
-      - uses: Swatinem/rust-cache@v2
-      - name: run tests
-        run: cargo test --lib --release --features full -- --nocapture sync::tests
-        working-directory: tokio
-
-  loom-time-driver:
-    name: loom time driver
-    # base_ref is null when it's not a pull request
-    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-time-driver') || (github.base_ref == null))
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v7
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-            toolchain: ${{ env.rust_stable }}
-      - uses: Swatinem/rust-cache@v2
-      - name: run tests
-        run: cargo test --lib --release --features full -- --nocapture runtime::time
-        working-directory: tokio
-
-  loom-current-thread:
-    name: loom current-thread scheduler
-    # base_ref is null when it's not a pull request
-    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-current-thread') || (github.base_ref == null))
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v7
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-            toolchain: ${{ env.rust_stable }}
-      - uses: Swatinem/rust-cache@v2
-      - name: run tests
-        run: cargo test --lib --release --features full -- --nocapture loom_current_thread
-        working-directory: tokio
-
-  loom-multi-thread:
-    name: loom multi-thread scheduler
-    # base_ref is null when it's not a pull request
-    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-multi-thread') || (github.base_ref == null))
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        include:
-          - scope: loom_multi_thread::group_a
-          - scope: loom_multi_thread::group_b
-          - scope: loom_multi_thread::group_c
-          - scope: loom_multi_thread::group_d
-    steps:
-      - uses: actions/checkout@v7
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-            toolchain: ${{ env.rust_stable }}
-      - uses: Swatinem/rust-cache@v2
-      - name: loom ${{ matrix.scope }}
-        run: cargo test --lib --release --features full -- --nocapture $SCOPE
-        working-directory: tokio
-        env:
-          SCOPE: ${{ matrix.scope }}
-
-  loom-util:
-    name: loom tokio-util
-    # base_ref is null when it's not a pull request
-    if: github.repository_owner == 'tokio-rs' && (contains(github.event.pull_request.labels.*.name, 'R-loom-util') || (github.base_ref == null))
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v7
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-            toolchain: ${{ env.rust_stable }}
-      - uses: Swatinem/rust-cache@v2
-      - name: run tests
-        run: cargo test --lib --release --features full -- --nocapture
-        working-directory: tokio-util

.github/workflows/pr-audit.yml

@@ -1,23 +0,0 @@
-name: Pull Request Security Audit
-
-on:
-  push:
-    paths:
-      - '**/Cargo.toml'
-  pull_request:
-    paths:
-      - '**/Cargo.toml'
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
-  cancel-in-progress: true
-
-permissions:
-  contents: read
-
-jobs:
-  cargo-deny:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v7
-    - uses: EmbarkStudios/cargo-deny-action@v2

.github/workflows/stress-test.yml

@@ -1,45 +0,0 @@
-name: Stress Test
-on:
-  pull_request:
-  push:
-    branches:
-      - master
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
-  cancel-in-progress: true
-
-env:
-  RUSTFLAGS: -Dwarnings
-  RUST_BACKTRACE: 1
-  # Change to specific Rust release to pin
-  rust_stable: 1.82
-
-permissions:
-  contents: read
-
-jobs:
-  stress-test:
-    name: Stress Test
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        stress-test:
-          - simple_echo_tcp
-    steps:
-      - uses: actions/checkout@v7
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: dtolnay/rust-toolchain@master
-        with:
-            toolchain: ${{ env.rust_stable }}
-      - name: Install Valgrind
-        uses: taiki-e/install-action@valgrind
-
-      - uses: Swatinem/rust-cache@v2
-      # Compiles each of the stress test examples.
-      - name: Compile stress test examples
-        run: cargo build -p stress-test --release --example ${{ matrix.stress-test }}
-
-      # Runs each of the examples using Valgrind. Detects leaks and displays them.
-      - name: Run valgrind
-        run: valgrind --error-exitcode=1 --leak-check=full --show-leak-kinds=all --fair-sched=yes ./target/release/examples/${{ matrix.stress-test }}

.github/workflows/uring-kernel-version-test.yml

@@ -1,120 +0,0 @@
-name: Uring Kernel Version Test
-
-on:
-  workflow_call:
-    inputs:
-      kernel_version:
-        description: 'Version of the Linux kernel to build'
-        required: true
-        type: string
-
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    env:
-      KERNEL_VERSION: ${{ inputs.kernel_version }}
-    steps:
-      - uses: actions/checkout@v7
-
-      - name: Install system dependencies
-        run: |
-          sudo apt-get update
-          sudo apt-get install -y \
-            build-essential bison flex libssl-dev libelf-dev \
-            qemu-system-x86 busybox-static cpio xz-utils wget
-
-      - name: Cache Linux source
-        id: cache-kernel
-        uses: actions/cache@v6
-        with:
-          path: linux-${{ env.KERNEL_VERSION }}
-          key: kernel-${{ env.KERNEL_VERSION }}
-
-      - name: Download & build Linux kernel
-        if: steps.cache-kernel.outputs.cache-hit != 'true'
-        run: |
-          MAJOR=${KERNEL_VERSION%%.*}
-          wget https://cdn.kernel.org/pub/linux/kernel/v${MAJOR}.x/linux-${KERNEL_VERSION}.tar.xz
-          tar xf linux-${KERNEL_VERSION}.tar.xz
-          cd linux-${KERNEL_VERSION}
-          make defconfig
-          make -j$(nproc)
-
-      # We are running tests also on a Kernel version that does not support io_uring
-      # to check if the fallback mechanism works
-      - name: Generate test binaries with io_uring enabled
-        run: |
-          # Build both integration (tokio/tests/) and unit (e.g., tokio/src/fs/file/tests.rs) tests with io_uring enabled
-          rustup target add x86_64-unknown-linux-musl
-          RUSTFLAGS="--cfg tokio_unstable" \
-            cargo test -p tokio --features full,io-uring \
-            --target x86_64-unknown-linux-musl --test 'fs*' --lib --no-run
-
-      - name: Prepare initramfs + tests binaries
-        run: |
-          set -e
-          rm -rf initramfs
-          mkdir -p initramfs/{bin,bin/tests,sbin,proc,sys,tmp}
-
-          # Copy test binaries into initramfs
-          for bin in target/x86_64-unknown-linux-musl/debug/deps/{fs_*,tokio-*}; do
-            if [ -f "$bin" ] && [ -x "$bin" ]; then
-              cp "$bin" initramfs/bin/tests
-            fi
-          done
-
-          # Add BusyBox & symlinks
-          cp /usr/bin/busybox initramfs/bin/
-          for cmd in sh mount uname true sleep; do ln -sf busybox initramfs/bin/$cmd; done
-          ln -sf ../bin/busybox initramfs/sbin/poweroff
-
-          # Generate init script
-          cat > initramfs/init << 'EOF'
-          #!/bin/sh
-          set -e
-          mkdir -p /dev
-          # create device nodes, as some tests require them
-          mknod /dev/null c 1 3
-          mknod /dev/zero c 1 5
-          mknod /dev/tty c 5 0
-          mount -t proc proc /proc
-          mount -t sysfs sysfs /sys
-          mkdir -p /tmp && mount -t tmpfs -o mode=1777 tmpfs /tmp
-          for f in /bin/tests/*; do RUST_BACKTRACE=1 "$f" ; done
-          EOF
-          chmod +x initramfs/init
-
-          # Pack into a CPIO archive
-          (cd initramfs && find . -print0 \
-            | cpio --null -ov --format=newc | gzip -9 > ../initramfs.cpio.gz)
-
-      - name: Run tests in QEMU
-        shell: bash
-        run: |
-          set -euo pipefail
-
-          set +e
-          qemu-system-x86_64 \
-            -kernel linux-${{ env.KERNEL_VERSION }}/arch/x86/boot/bzImage \
-            -initrd initramfs.cpio.gz \
-            -append "console=ttyS0 rootfstype=ramfs panic=1" \
-            -nographic -no-reboot -m 1024 -action panic=exit-failure 2>&1 | tee qemu-output.log
-          qemu_status=${PIPESTATUS[0]} tee_status=${PIPESTATUS[1]}
-          set -e
-
-          if [ "$qemu_status" -ne 0 ]; then
-            echo "QEMU exited with status $qemu_status"
-            exit "$qemu_status"
-          fi
-
-          if [ "$tee_status" -ne 0 ]; then
-            echo "tee exited with status $tee_status"
-            exit "$tee_status"
-          fi
-
-          if grep -q "test result: FAILED" qemu-output.log; then
-            echo "tests reported failures"
-            exit 1
-          else
-            echo "all tests passed"
-          fi

.gitignore

@@ -1,6 +1,2 @@
 target
 Cargo.lock
-
-.cargo/config.toml
-.cargo/config
-

CODE_OF_CONDUCT.md

@@ -1,7 +0,0 @@
-# Code of Conduct
-
-The Tokio project adheres to the [Rust Code of Conduct](https://www.rust-lang.org/policies/code-of-conduct). This describes the minimum behavior expected from all contributors.
-
-## Enforcement
-
-Instances of violations of the Code of Conduct can be reported by contacting the project team at [[email protected]](mailto:[email protected]).

CONTRIBUTING.md

@@ -1,6 +1,7 @@
 # Contributing to Tokio
 
-Thanks for your help improving Tokio! We are so happy to have you!
+:balloon: Thanks for your help improving the project! We are so happy to have
+you!
 
 There are opportunities to contribute to Tokio at any level. It doesn't matter if
 you are just getting started with Rust or are the most weathered expert, we can
@@ -8,47 +9,379 @@ use your help.
 
 **No contribution is too small and all contributions are valued.**
 
-See the [contributing guidelines] to get started.
+This guide will help you get started. **Do not let this guide intimidate you**.
+It should be considered a map to help you navigate the process.
 
-[contributing guidelines]: docs/contributing/README.md
+You may also get help with contributing in the [dev channel][dev], please join
+us!
 
-## Code of Conduct
+[dev]: https://gitter.im/tokio-rs/dev
+
+## Conduct
 
 The Tokio project adheres to the [Rust Code of Conduct][coc]. This describes
-the _minimum_ behavior expected from all contributors. Instances of violations of the
-Code of Conduct can be reported by contacting the project team at
-[[email protected]](mailto:[email protected]).
+the _minimum_ behavior expected from all contributors.
 
 [coc]: https://github.com/rust-lang/rust/blob/master/CODE_OF_CONDUCT.md
 
-## Need Help?
+## Contributing in Issues
 
-Reach out to us on the [Discord server] for any concern not covered in this guide.
+For any issue, there are fundamentally three ways an individual can contribute:
 
-[Discord server]: https://discord.gg/tokio
+1. By opening the issue for discussion: For instance, if you believe that you
+   have uncovered a bug in Tokio, creating a new issue in the tokio-rs/tokio
+   issue tracker is the way to report it.
 
-## LTS guarantees
+2. By helping to triage the issue: This can be done by providing
+   supporting details (a test case that demonstrates a bug), providing
+   suggestions on how to address the issue, or ensuring that the issue is tagged
+   correctly.
 
-In Tokio ≥1.0.0, each LTS release comes with the guarantee of at least one year of
-backported fixes.
+3. By helping to resolve the issue: Typically this is done either in the form of
+   demonstrating that the issue reported is not a problem after all, or more
+   often, by opening a Pull Request that changes some bit of something in
+   Tokio in a concrete and reviewable manner.
 
-The goal of these guarantees is to provide stability to the ecosystem.
+**Anybody can participate in any stage of contribution**. We urge you to
+participate in the discussion around bugs and participate in reviewing PRs.
 
-## Minimum Supported Rust Version (MSRV)
+### Asking for General Help
 
-* All Tokio ≥1.0.0 releases will support at least a 6-month old Rust
-  compiler release.
-* The MSRV will only be increased on 1.x releases.
+If you have reviewed existing documentation and still have questions or are
+having problems, you can open an issue asking for help.
 
-## Versioning Policy
+In exchange for receiving help, we ask that you contribute back a documentation
+PR that helps others avoid the problems that you encountered.
 
-With Tokio ≥1.0.0:
+### Submitting a Bug Report
 
-* Patch (1.\_.x) releases _should only_ contain bug fixes or documentation
-  changes. Besides this, these releases should not substantially change
-  runtime behavior.
-* Minor (1.x) releases may contain new functionality, MSRV increases (see
-  above), minor dependency updates, deprecations, and larger internal
-  implementation changes.
+When opening a new issue in the Tokio issue tracker, users will be presented
+with a [basic template][template] that should be filled in. If you believe that you have
+uncovered a bug, please fill out this form, following the template to the best
+of your ability. Do not worry if you cannot answer every detail, just fill in
+what you can.
 
-This is as defined by [Semantic Versioning 2.0](https://semver.org/).
+The two most important pieces of information we need in order to properly
+evaluate the report is a description of the behavior you are seeing and a simple
+test case we can use to recreate the problem on our own. If we cannot recreate
+the issue, it becomes impossible for us to fix.
+
+In order to rule out the possibility of bugs introduced by userland code, test
+cases should be limited, as much as possible, to using only Tokio APIs.
+
+See [How to create a Minimal, Complete, and Verifiable example][mcve].
+
+[mcve]: https://stackoverflow.com/help/mcve
+[template]: .github/PULL_REQUEST_TEMPLATE.md
+
+### Triaging a Bug Report
+
+Once an issue has been opened, it is not uncommon for there to be discussion
+around it. Some contributors may have differing opinions about the issue,
+including whether the behavior being seen is a bug or a feature. This discussion
+is part of the process and should be kept focused, helpful, and professional.
+
+Short, clipped responses—that provide neither additional context nor supporting
+detail—are not helpful or professional. To many, such responses are simply
+annoying and unfriendly.
+
+Contributors are encouraged to help one another make forward progress as much as
+possible, empowering one another to solve issues collaboratively. If you choose
+to comment on an issue that you feel either is not a problem that needs to be
+fixed, or if you encounter information in an issue that you feel is incorrect,
+explain why you feel that way with additional supporting context, and be willing
+to be convinced that you may be wrong. By doing so, we can often reach the
+correct outcome much faster.
+
+### Resolving a Bug Report
+
+In the majority of cases, issues are resolved by opening a Pull Request. The
+process for opening and reviewing a Pull Request is similar to that of opening
+and triaging issues, but carries with it a necessary review and approval
+workflow that ensures that the proposed changes meet the minimal quality and
+functional guidelines of the Tokio project.
+
+## Pull Requests
+
+Pull Requests are the way concrete changes are made to the code, documentation,
+and dependencies in the Tokio repository.
+
+Even tiny pull requests (e.g., one character pull request fixing a typo in API
+documentation) are greatly appreciated. Before making a large change, it is
+usually a good idea to first open an issue describing the change to solicit
+feedback and guidance. This will increase the likelihood of the PR getting
+merged.
+
+### Tests
+
+If the change being proposed alters code (as opposed to only documentation for
+example), it is either adding new functionality to Tokio or it is fixing
+existing, broken functionality. In both of these cases, the pull request should
+include one or more tests to ensure that Tokio does not regress in the future.
+There are two ways to write tests: integration tests and documentation tests
+(Tokio avoids unit tests as much as possible).
+
+#### Integration tests
+
+Integration tests go in the same crate as the code they are testing. Each sub
+crate should have a `dev-dependency` on `tokio` itself. This makes all Tokio
+utilities available to use in tests, no matter the crate being tested.
+
+The best strategy for writing a new integration test is to look at existing
+integration tests in the crate and follow the style.
+
+#### Documentation tests
+
+Ideally, every API has at least one [documentation test] that demonstrates how to
+use the API. Documentation tests are run with `cargo test --doc`. This ensures
+that the example is correct and provides additional test coverage.
+
+The trick to documentation tests is striking a balance between being succinct
+for a reader to understand and actually testing the API.
+
+Same as with integration tests, when writing a documentation test, the full
+`tokio` crate is available. This is especially useful for getting access to the
+runtime to run the example.
+
+The documentation tests will be visible from both the crate specific
+documentation **and** the `tokio` facade documentation via the re-export. The
+example should be written from the point of view of a user that is using the
+`tokio` crate. As such, the example should use the API via the facade and not by
+directly referencing the crate.
+
+The type level example for `tokio_timer::Timeout` provides a good example of a
+documentation test:
+
+```
+/// # extern crate futures;
+/// # extern crate tokio;
+/// // import the `timeout` function, usually this is done
+/// // with `use tokio::prelude::*`
+/// use tokio::prelude::FutureExt;
+/// use futures::Stream;
+/// use futures::sync::mpsc;
+/// use std::time::Duration;
+///
+/// # fn main() {
+/// let (tx, rx) = mpsc::unbounded();
+/// # tx.unbounded_send(()).unwrap();
+/// # drop(tx);
+///
+/// let process = rx.for_each(|item| {
+///     // do something with `item`
+/// # drop(item);
+/// # Ok(())
+/// });
+///
+/// # tokio::runtime::current_thread::block_on_all(
+/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds.
+/// process.timeout(Duration::from_millis(10))
+/// # ).unwrap();
+/// # }
+```
+
+Given that this is a *type* level documentation test and the primary way users
+of `tokio` will create an instance of `Timeout` is by using
+`FutureExt::timeout`, this is how the documentation test is structured.
+
+Lines that start with `/// #` are removed when the documentation is generated.
+They are only there to get the test to run. The `block_on_all` function is the
+easiest way to execute a future from a test.
+
+If this were a documentation test for the `Timeout::new` function, then the
+example would explicitly use `Timeout::new`. For example:
+
+```
+/// # extern crate futures;
+/// # extern crate tokio;
+/// use tokio::timer::Timeout;
+/// use futures::Future;
+/// use futures::sync::oneshot;
+/// use std::time::Duration;
+///
+/// # fn main() {
+/// let (tx, rx) = oneshot::channel();
+/// # tx.send(()).unwrap();
+///
+/// # tokio::runtime::current_thread::block_on_all(
+/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds.
+/// Timeout::new(rx, Duration::from_millis(10))
+/// # ).unwrap();
+/// # }
+```
+
+### Commits
+
+It is a recommended best practice to keep your changes as logically grouped as
+possible within individual commits. There is no limit to the number of commits
+any single Pull Request may have, and many contributors find it easier to review
+changes that are split across multiple commits.
+
+That said, if you have a number of commits that are "checkpoints" and don't
+represent a single logical change, please squash those together.
+
+Note that multiple commits often get squashed when they are landed (see the
+notes about [commit squashing]).
+
+#### Commit message guidelines
+
+A good commit message should describe what changed and why.
+
+1. The first line should:
+
+  * contain a short description of the change (preferably 50 characters or less,
+    and no more than 72 characters)
+  * be entirely in lowercase with the exception of proper nouns, acronyms, and
+    the words that refer to code, like function/variable names
+  * be prefixed with the name of the sub crate being changed (without the `tokio-`
+    prefix) and start with an imperative verb. If modifying `tokio` proper,
+    omit the crate prefix.
+
+  Examples:
+
+  * timer: introduce `Timeout` and deprecate `Deadline`
+  * export `Encoder`, `Decoder`, `Framed*` from tokio_codec
+
+2. Keep the second line blank.
+3. Wrap all other lines at 72 columns (except for long URLs).
+4. If your patch fixes an open issue, you can add a reference to it at the end
+   of the log. Use the `Fixes: #` prefix and the issue number. For other
+   references use `Refs: #`. `Refs` may include multiple issues, separated by a
+   comma.
+
+   Examples:
+
+   - `Fixes: #1337`
+   - `Refs: #1234`
+
+Sample complete commit message:
+
+```txt
+subcrate: explain the commit in one line
+
+Body of commit message is a few lines of text, explaining things
+in more detail, possibly giving some background about the issue
+being fixed, etc.
+
+The body of the commit message can be several paragraphs, and
+please do proper word-wrap and keep columns shorter than about
+72 characters or so. That way, `git log` will show things
+nicely even when it is indented.
+
+Fixes: #1337
+Refs: #453, #154
+```
+
+### Opening the Pull Request
+
+From within GitHub, opening a new Pull Request will present you with a
+[template] that should be filled out. Please try to do your best at filling out
+the details, but feel free to skip parts if you're not sure what to put.
+
+[template]: .github/PULL_REQUEST_TEMPLATE.md
+
+### Discuss and update
+
+You will probably get feedback or requests for changes to your Pull Request.
+This is a big part of the submission process so don't be discouraged! Some
+contributors may sign off on the Pull Request right away, others may have
+more detailed comments or feedback. This is a necessary part of the process
+in order to evaluate whether the changes are correct and necessary.
+
+**Any community member can review a PR and you might get conflicting feedback**.
+Keep an eye out for comments from code owners to provide guidance on conflicting
+feedback.
+
+**Once the PR is open, do not rebase the commits**. See [Commit Squashing] for
+more details.
+
+### Commit Squashing
+
+In most cases, **do not squash commits that you add to your Pull Request during
+the review process**. When the commits in your Pull Request land, they may be
+squashed into one commit per logical change. Metadata will be added to the
+commit message (including links to the Pull Request, links to relevant issues,
+and the names of the reviewers). The commit history of your Pull Request,
+however, will stay intact on the Pull Request page.
+
+## Reviewing Pull Requests
+
+**Any Tokio community member is welcome to review any pull request**.
+
+All Tokio contributors who choose to review and provide feedback on Pull
+Requests have a responsibility to both the project and the individual making the
+contribution. Reviews and feedback must be helpful, insightful, and geared
+towards improving the contribution as opposed to simply blocking it. If there
+are reasons why you feel the PR should not land, explain what those are. Do not
+expect to be able to block a Pull Request from advancing simply because you say
+"No" without giving an explanation. Be open to having your mind changed. Be open
+to working with the contributor to make the Pull Request better.
+
+Reviews that are dismissive or disrespectful of the contributor or any other
+reviewers are strictly counter to the Code of Conduct.
+
+When reviewing a Pull Request, the primary goals are for the codebase to improve
+and for the person submitting the request to succeed. **Even if a Pull Request
+does not land, the submitters should come away from the experience feeling like
+their effort was not wasted or unappreciated**. Every Pull Request from a new
+contributor is an opportunity to grow the community.
+
+### Review a bit at a time.
+
+Do not overwhelm new contributors.
+
+It is tempting to micro-optimize and make everything about relative performance,
+perfect grammar, or exact style matches. Do not succumb to that temptation.
+
+Focus first on the most significant aspects of the change:
+
+1. Does this change make sense for Tokio?
+2. Does this change make Tokio better, even if only incrementally?
+3. Are there clear bugs or larger scale issues that need attending to?
+4. Is the commit message readable and correct? If it contains a breaking change
+   is it clear enough?
+
+Note that only **incremental** improvement is needed to land a PR. This means
+that the PR does not need to be perfect, only better than the status quo. Follow
+up PRs may be opened to continue iterating.
+
+When changes are necessary, *request* them, do not *demand* them, and **do not
+assume that the submitter already knows how to add a test or run a benchmark**.
+
+Specific performance optimization techniques, coding styles and conventions
+change over time. The first impression you give to a new contributor never does.
+
+Nits (requests for small changes that are not essential) are fine, but try to
+avoid stalling the Pull Request. Most nits can typically be fixed by the Tokio
+Collaborator landing the Pull Request but they can also be an opportunity for
+the contributor to learn a bit more about the project.
+
+It is always good to clearly indicate nits when you comment: e.g.
+`Nit: change foo() to bar(). But this is not blocking.`
+
+If your comments were addressed but were not folded automatically after new
+commits or if they proved to be mistaken, please, [hide them][hiding-a-comment]
+with the appropriate reason to keep the conversation flow concise and relevant.
+
+### Be aware of the person behind the code
+
+Be aware that *how* you communicate requests and reviews in your feedback can
+have a significant impact on the success of the Pull Request. Yes, we may land
+a particular change that makes Tokio better, but the individual might just not
+want to have anything to do with Tokio ever again. The goal is not just having
+good code.
+
+### Abandoned or Stalled Pull Requests
+
+If a Pull Request appears to be abandoned or stalled, it is polite to first
+check with the contributor to see if they intend to continue the work before
+checking if they would mind if you took it over (especially if it just has nits
+left). When doing so, it is courteous to give the original contributor credit
+for the work they started (either by preserving their name and email address in
+the commit log, or by using an `Author: ` meta-data tag in the commit.
+
+_Adapted from the [Node.js contributing guide][node]_.
+
+[node]: https://github.com/nodejs/node/blob/master/CONTRIBUTING.md
+[hiding-a-comment]: https://help.github.com/articles/managing-disruptive-comments/#hiding-a-comment
+[documentation test]: https://doc.rust-lang.org/rustdoc/documentation-tests.html

Cargo.toml

@@ -1,39 +1,22 @@
 [workspace]
-resolver = "2"
+
 members = [
   "tokio",
-  "tokio-macros",
+  "tokio-buf",
+  "tokio-codec",
+  "tokio-current-thread",
+  "tokio-executor",
+  "tokio-fs",
+  "tokio-futures",
+  "tokio-io",
+  "tokio-reactor",
+  "tokio-signal",
+  "tokio-sync",
   "tokio-test",
-  "tokio-stream",
-  "tokio-util",
-
-  # Internal
-  "benches",
-  "examples",
-  "stress-test",
-  "tests-build",
-  "tests-integration",
+  "tokio-threadpool",
+  "tokio-timer",
+  "tokio-tcp",
+  "tokio-tls",
+  "tokio-udp",
+  "tokio-uds",
 ]
-
-[patch.crates-io]
-tokio = { path = "tokio" }
-tokio-macros = { path = "tokio-macros" }
-tokio-stream = { path = "tokio-stream" }
-tokio-test = { path = "tokio-test" }
-tokio-util = { path = "tokio-util" }
-
-[workspace.metadata.spellcheck]
-config = "spellcheck.toml"
-
-[workspace.lints.rust]
-unexpected_cfgs = { level = "warn", check-cfg = [
-  'cfg(fuzzing)',
-  'cfg(loom)',
-  'cfg(mio_unsupported_force_poll_poll)',
-  'cfg(tokio_allow_from_blocking_fd)',
-  'cfg(tokio_internal_mt_counters)',
-  'cfg(tokio_no_parking_lot)',
-  'cfg(tokio_no_tuning_tests)',
-  'cfg(tokio_unstable)',
-  'cfg(target_os, values("cygwin"))',
-] }

Cross.toml

@@ -1,5 +0,0 @@
-[build.env]
-passthrough = [
-    "RUSTFLAGS",
-    "RUST_BACKTRACE",
-]

LICENSE

@@ -1,21 +1,25 @@
-MIT License
+Copyright (c) 2019 Tokio Contributors
 
-Copyright (c) Tokio Contributors
+Permission is hereby granted, free of charge, to any
+person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the
+Software without restriction, including without
+limitation the rights to use, copy, modify, merge,
+publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software
+is furnished to do so, subject to the following
+conditions:
 
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
+The above copyright notice and this permission notice
+shall be included in all copies or substantial portions
+of the Software.
 
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.

README.md

@@ -14,22 +14,26 @@ the Rust programming language. It is:
 
 [![Crates.io][crates-badge]][crates-url]
 [![MIT licensed][mit-badge]][mit-url]
-[![Build Status][actions-badge]][actions-url]
-[![Discord chat][discord-badge]][discord-url]
+[![Build Status][azure-badge]][azure-url]
+[![Gitter chat][gitter-badge]][gitter-url]
 
 [crates-badge]: https://img.shields.io/crates/v/tokio.svg
 [crates-url]: https://crates.io/crates/tokio
 [mit-badge]: https://img.shields.io/badge/license-MIT-blue.svg
-[mit-url]: https://github.com/tokio-rs/tokio/blob/master/LICENSE
-[actions-badge]: https://github.com/tokio-rs/tokio/workflows/CI/badge.svg
-[actions-url]: https://github.com/tokio-rs/tokio/actions?query=workflow%3ACI+branch%3Amaster
-[discord-badge]: https://img.shields.io/discord/500028886025895936.svg?logo=discord&style=flat-square
-[discord-url]: https://discord.gg/tokio
+[mit-url]: LICENSE
+[azure-badge]: https://dev.azure.com/tokio-rs/Tokio/_apis/build/status/tokio-rs.tokio?branchName=master
+[azure-url]: https://dev.azure.com/tokio-rs/Tokio/_build/latest?definitionId=1&branchName=master
+[gitter-badge]: https://img.shields.io/gitter/room/tokio-rs/tokio.svg
+[gitter-url]: https://gitter.im/tokio-rs/tokio
 
 [Website](https://tokio.rs) |
-[Guides](https://tokio.rs/tokio/tutorial) |
-[API Docs](https://docs.rs/tokio/latest/tokio) |
-[Chat](https://discord.gg/tokio)
+[Guides](https://tokio.rs/docs/getting-started/hello-world/) |
+[API Docs](https://docs.rs/tokio/0.1.20/tokio) |
+[Chat](https://gitter.im/tokio-rs/tokio)
+
+The API docs for the master branch are published [here][master-dox].
+
+[master-dox]: https://tokio-rs.github.io/tokio/doc/tokio/
 
 ## Overview
 
@@ -38,86 +42,73 @@ asynchronous applications with the Rust programming language. At a high
 level, it provides a few major components:
 
 * A multithreaded, work-stealing based task [scheduler].
-* A reactor backed by the operating system's event queue (epoll, kqueue,
-  IOCP, etc.).
+* A [reactor] backed by the operating system's event queue (epoll, kqueue,
+  IOCP, etc...).
 * Asynchronous [TCP and UDP][net] sockets.
 
 These components provide the runtime components necessary for building
 an asynchronous application.
 
-[net]: https://docs.rs/tokio/latest/tokio/net/index.html
-[scheduler]: https://docs.rs/tokio/latest/tokio/runtime/index.html
+[net]: https://docs.rs/tokio/0.1.20/tokio/net/index.html
+[reactor]: https://docs.rs/tokio/0.1.20/tokio/reactor/index.html
+[scheduler]: https://docs.rs/tokio/0.1.20/tokio/runtime/index.html
 
 ## Example
 
-A basic TCP echo server with Tokio.
+A basic TCP echo server with Tokio:
 
-Make sure you enable the full features of the tokio crate on Cargo.toml:
+```rust
+extern crate tokio;
 
-```toml
-[dependencies]
-tokio = { version = "1.53.1", features = ["full"] }
-```
-Then, on your main.rs:
-
-```rust,no_run
+use tokio::prelude::*;
+use tokio::io::copy;
 use tokio::net::TcpListener;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let listener = TcpListener::bind("127.0.0.1:8080").await?;
+fn main() {
+    // Bind the server's socket.
+    let addr = "127.0.0.1:12345".parse().unwrap();
+    let listener = TcpListener::bind(&addr)
+        .expect("unable to bind TCP listener");
 
-    loop {
-        let (mut socket, _) = listener.accept().await?;
+    // Pull out a stream of sockets for incoming connections
+    let server = listener.incoming()
+        .map_err(|e| eprintln!("accept failed = {:?}", e))
+        .for_each(|sock| {
+            // Split up the reading and writing parts of the
+            // socket.
+            let (reader, writer) = sock.split();
 
-        tokio::spawn(async move {
-            let mut buf = [0; 1024];
+            // A future that echos the data and returns how
+            // many bytes were copied...
+            let bytes_copied = copy(reader, writer);
 
-            // In a loop, read data from the socket and write the data back.
-            loop {
-                let n = match socket.read(&mut buf).await {
-                    // socket closed
-                    Ok(0) => return,
-                    Ok(n) => n,
-                    Err(e) => {
-                        eprintln!("failed to read from socket; err = {:?}", e);
-                        return;
-                    }
-                };
+            // ... after which we'll print what happened.
+            let handle_conn = bytes_copied.map(|amt| {
+                println!("wrote {:?} bytes", amt)
+            }).map_err(|err| {
+                eprintln!("IO error {:?}", err)
+            });
 
-                // Write the data back
-                if let Err(e) = socket.write_all(&buf[0..n]).await {
-                    eprintln!("failed to write to socket; err = {:?}", e);
-                    return;
-                }
-            }
+            // Spawn the future as a concurrent task.
+            tokio::spawn(handle_conn)
         });
-    }
+
+    // Start the Tokio runtime
+    tokio::run(server);
 }
 ```
 
-More examples can be found [here][examples]. For a larger "real world" example, see the
-[mini-redis] repository.
-
-[examples]: https://github.com/tokio-rs/tokio/tree/master/examples
-[mini-redis]: https://github.com/tokio-rs/mini-redis/
-
-To see a list of the available feature flags that can be enabled, check our
-[docs][feature-flag-docs].
+More examples can be found [here](tokio/examples).
 
 ## Getting Help
 
 First, see if the answer to your question can be found in the [Guides] or the
 [API documentation]. If the answer is not there, there is an active community in
-the [Tokio Discord server][chat]. We would be happy to try to answer your
-question. You can also ask your question on [the discussions page][discussions].
+the [Tokio Gitter 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.
 
-[Guides]: https://tokio.rs/tokio/tutorial
-[API documentation]: https://docs.rs/tokio/latest/tokio
-[chat]: https://discord.gg/tokio
-[discussions]: https://github.com/tokio-rs/tokio/discussions
-[feature-flag-docs]: https://docs.rs/tokio/#feature-flags
+[chat]: https://gitter.im/tokio-rs/tokio
+[issue]: https://github.com/tokio-rs/tokio/issues/new
 
 ## Contributing
 
@@ -125,132 +116,91 @@ question. You can also ask your question on [the discussions page][discussions].
 you! We have a [contributing guide][guide] to help you get involved in the Tokio
 project.
 
-[guide]: https://github.com/tokio-rs/tokio/blob/master/docs/contributing/README.md
+[guide]: CONTRIBUTING.md
+
+## Project layout
+
+The `tokio` crate, found at the root, is primarily intended for use by
+application developers.  Library authors should depend on the sub crates, which
+have greater guarantees of stability.
+
+The crates included as part of Tokio are:
+
+* [`tokio-current-thread`]: Schedule the execution of futures on the current
+  thread.
+
+* [`tokio-executor`]: Task execution related traits and utilities.
+
+* [`tokio-fs`]: Filesystem (and standard in / out) APIs.
+
+* [`tokio-futures`]: Experimental `std::future::Future` and `async` / `await` support.
+
+* [`tokio-codec`]: Utilities for encoding and decoding protocol frames.
+
+* [`tokio-io`]: Asynchronous I/O related traits and utilities.
+
+* [`tokio-macros`]: Macros for usage with Tokio.
+
+* [`tokio-reactor`]: Event loop that drives I/O resources (like TCP and UDP
+  sockets).
+
+* [`tokio-tcp`]: TCP bindings for use with `tokio-io` and `tokio-reactor`.
+
+* [`tokio-threadpool`]: Schedules the execution of futures across a pool of
+  threads.
+
+* [ `tokio-timer`]: Time related APIs.
+
+* [`tokio-udp`]: UDP bindings for use with `tokio-io` and `tokio-reactor`.
+
+* [`tokio-uds`]: Unix Domain Socket bindings for use with `tokio-io` and
+  `tokio-reactor`.
+
+[`tokio-codec`]: tokio-codec
+[`tokio-current-thread`]: tokio-current-thread
+[`tokio-executor`]: tokio-executor
+[`tokio-fs`]: tokio-fs
+[`tokio-futures`]: tokio-futures
+[`tokio-io`]: tokio-io
+[`tokio-macros`]: tokio-macros
+[`tokio-reactor`]: tokio-reactor
+[`tokio-tcp`]: tokio-tcp
+[`tokio-threadpool`]: tokio-threadpool
+[`tokio-timer`]: tokio-timer
+[`tokio-udp`]: tokio-udp
+[`tokio-uds`]: tokio-uds
 
 ## Related Projects
 
 In addition to the crates in this repository, the Tokio project also maintains
 several other libraries, including:
 
-* [`axum`]: A web application framework that focuses on ergonomics and modularity.
+* [`tracing`] (formerly `tokio-trace`): A framework for application-level
+  tracing and async-aware diagnostics.
 
-* [`hyper`]: A fast and correct HTTP/1.1 and HTTP/2 implementation for Rust.
-
-* [`tonic`]: A gRPC over HTTP/2 implementation focused on high performance, interoperability, and flexibility.
-
-* [`warp`]: A super-easy, composable, web server framework for warp speeds.
-
-* [`tower`]: A library of modular and reusable components for building robust networking clients and servers.
-
-* [`tracing`] (formerly `tokio-trace`): A framework for application-level tracing and async-aware diagnostics.
-
-* [`mio`]: A low-level, cross-platform abstraction over OS I/O APIs that powers `tokio`.
+* [`mio`]: A low-level, cross-platform abstraction over OS I/O APIs that powers
+  `tokio`.
 
 * [`bytes`]: Utilities for working with bytes, including efficient byte buffers.
 
-* [`loom`]: A testing tool for concurrent Rust code.
-
-[`axum`]: https://github.com/tokio-rs/axum
-[`warp`]: https://github.com/seanmonstar/warp
-[`hyper`]: https://github.com/hyperium/hyper
-[`tonic`]: https://github.com/hyperium/tonic
-[`tower`]: https://github.com/tower-rs/tower
-[`loom`]: https://github.com/tokio-rs/loom
 [`tracing`]: https://github.com/tokio-rs/tracing
 [`mio`]: https://github.com/tokio-rs/mio
 [`bytes`]: https://github.com/tokio-rs/bytes
 
-## Changelog
-
-The Tokio repository contains multiple crates. Each crate has its own changelog.
-
- * `tokio` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio/CHANGELOG.md)
- * `tokio-util` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio-util/CHANGELOG.md)
- * `tokio-stream` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio-stream/CHANGELOG.md)
- * `tokio-macros` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio-macros/CHANGELOG.md)
- * `tokio-test` - [view changelog](https://github.com/tokio-rs/tokio/blob/master/tokio-test/CHANGELOG.md)
-
 ## Supported Rust Versions
 
-<!--
-When updating this, also update:
-- .github/workflows/ci.yml
-- CONTRIBUTING.md
-- README.md
-- tokio/README.md
-- tokio/Cargo.toml
-- tokio-util/Cargo.toml
-- tokio-test/Cargo.toml
-- tokio-stream/Cargo.toml
--->
-
-Tokio will keep a rolling MSRV (minimum supported rust version) policy of **at
-least** 6 months. When increasing the MSRV, the new Rust version must have been
-released at least six months ago. The current MSRV is 1.71.
-
-Note that the MSRV is not increased automatically, and only as part of a minor
-release. The MSRV history for past minor releases can be found below:
-
- * 1.48 to now  - Rust 1.71
- * 1.39 to 1.47 - Rust 1.70
- * 1.30 to 1.38 - Rust 1.63
- * 1.27 to 1.29 - Rust 1.56
- * 1.17 to 1.26 - Rust 1.49
- * 1.15 to 1.16 - Rust 1.46
- * 1.0 to 1.14 - Rust 1.45
-
-Note that although we try to avoid the situation where a dependency transitively
-increases the MSRV of Tokio, we do not guarantee that this does not happen.
-However, every minor release will have some set of versions of dependencies that
-works with the MSRV of that minor release.
-
-## Release schedule
-
-Tokio doesn't follow a fixed release schedule, but we typically make one minor
-release each month. We make patch releases for bugfixes as necessary.
-
-## Bug patching policy
-
-For the purposes of making patch releases with bugfixes, we have designated
-certain minor releases as LTS (long term support) releases. Whenever a bug
-warrants a patch release with a fix for the bug, it will be backported and
-released as a new patch release for each LTS minor version. Our current LTS
-releases are:
-
- * `1.47.x` - LTS release until September 2026. (MSRV 1.70)
- * `1.51.x` - LTS release until March 2027. (MSRV 1.71)
-
-Each LTS release will continue to receive backported fixes for at least a year.
-If you wish to use a fixed minor release in your project, we recommend that you
-use an LTS release.
-
-To use a fixed minor version, you can specify the version with a tilde. For
-example, to specify that you wish to use the newest `1.47.x` patch release, you
-can use the following dependency specification:
-```text
-tokio = { version = "~1.47", features = [...] }
-```
-
-### Previous LTS releases
-
- * `1.8.x` - LTS release until February 2022.
- * `1.14.x` - LTS release until June 2022.
- * `1.18.x` - LTS release until June 2023.
- * `1.20.x` - LTS release until September 2023.
- * `1.25.x` - LTS release until March 2024.
- * `1.32.x` - LTS release until September 2024.
- * `1.36.x` - LTS release until March 2025.
- * `1.38.x` - LTS release until July 2025.
- * `1.43.x` - LTS release until March 2026.
+Tokio is built against the latest stable, nightly, and beta Rust releases. The
+minimum version supported is the stable release from three months before the
+current stable release version. For example, if the latest stable Rust is 1.29,
+the minimum version supported is 1.26. The current Tokio version is not
+guaranteed to build on Rust versions earlier than the minimum supported version.
 
 ## License
 
-This project is licensed under the [MIT license].
-
-[MIT license]: https://github.com/tokio-rs/tokio/blob/master/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
+for inclusion in Tokio by you, shall be licensed as MIT, without any additional
 terms or conditions.

SECURITY.md

@@ -1,13 +0,0 @@
-## Report a security issue
-
-The Tokio project team welcomes security reports and is committed to providing prompt attention to security issues. Security issues should be reported privately via [[email protected]](mailto:[email protected]). Security issues should not be reported via the public GitHub Issue tracker.
-
-## Vulnerability coordination
-
-Remediation of security vulnerabilities is prioritized by the project team. The project team coordinates remediation with third-party project stakeholders via [GitHub Security Advisories](https://help.github.com/en/github/managing-security-vulnerabilities/about-github-security-advisories). Third-party stakeholders may include the reporter of the issue, affected direct or indirect users of Tokio, and maintainers of upstream dependencies if applicable.
-
-Downstream project maintainers and Tokio users can request participation in coordination of applicable security issues by sending your contact email address, GitHub username(s) and any other salient information to [[email protected]](mailto:[email protected]). Participation in security issue coordination processes is at the discretion of the Tokio team.
-
-## Security advisories
-
-The project team is committed to transparency in the security issue disclosure process. The Tokio team announces security issues via [project GitHub Release notes](https://github.com/tokio-rs/tokio/releases) and the [RustSec advisory database](https://github.com/RustSec/advisory-db) (i.e. `cargo-audit`).

azure-pipelines.yml

@@ -0,0 +1,103 @@
+trigger: ["master", "v0.1.x"]
+pr: ["master", "v0.1.x"]
+
+jobs:
+# Check formatting
+- template: ci/azure-rustfmt.yml
+  parameters:
+    name: rustfmt
+
+# Test top level crate
+- template: ci/azure-test-stable.yml
+  parameters:
+    name: test_tokio
+    displayName: Test tokio
+    cross: true
+    crates:
+      - tokio
+
+# Test crates that are platform specific
+- template: ci/azure-test-stable.yml
+  parameters:
+    name: test_sub_cross
+    displayName: Test sub crates -
+    cross: true
+    crates:
+      - tokio-fs
+      - tokio-reactor
+      - tokio-signal
+      - tokio-tcp
+      - tokio-tls
+      - tokio-udp
+      - tokio-uds
+
+# Test crates that are NOT platform specific
+- template: ci/azure-test-stable.yml
+  parameters:
+    name: test_linux
+    displayName: Test sub crates -
+    crates:
+      - tokio-buf
+      - tokio-codec
+      - tokio-current-thread
+      - tokio-executor
+      - tokio-io
+      - tokio-sync
+      - tokio-threadpool
+      - tokio-timer
+      - tokio-test
+
+- template: ci/azure-cargo-check.yml
+  parameters:
+    name: features
+    displayName: Check feature permtuations
+    rust: stable
+    crates:
+      tokio:
+        - codec
+        - fs
+        - io
+        - reactor
+        - rt-full
+        - tcp
+        - timer
+        - udp
+        - uds
+        - sync
+        - experimental-tracing
+      tokio-buf:
+        - util
+
+# Try cross compiling
+- template: ci/azure-cross-compile.yml
+  parameters:
+    name: cross_32bit_linux
+    target: i686-unknown-linux-gnu
+
+# This represents the minimum Rust version supported by
+# Tokio. Updating this should be done in a dedicated PR and
+# cannot be greater than two 0.x releases prior to the
+# current stable.
+#
+# Tests are not run as tests may require newer versions of
+# rust.
+- template: ci/azure-check-minrust.yml
+  parameters:
+    name: minrust
+    rust_version: 1.31.0
+
+- template: ci/azure-tsan.yml
+  parameters:
+    name: tsan
+
+- template: ci/azure-deploy-docs.yml
+  parameters:
+    dependsOn:
+      - rustfmt
+      - test_tokio
+      - test_sub_cross
+      - test_linux
+      - features
+      - cross_32bit_linux
+      - minrust
+      - tsan

benches/Cargo.toml

@@ -1,110 +0,0 @@
-[package]
-name = "benches"
-version = "0.0.0"
-publish = false
-edition = "2021"
-license = "MIT"
-
-[features]
-test-util = ["tokio/test-util"]
-
-[dependencies]
-tokio = { version = "1.5.0", path = "../tokio", features = ["full"] }
-criterion = "0.5.1"
-rand = "0.9"
-rand_chacha = "0.9"
-
-[dev-dependencies]
-tokio-util = { version = "0.7.0", path = "../tokio-util", features = ["full"] }
-tokio-stream = { version = "0.1", path = "../tokio-stream" }
-
-[target.'cfg(unix)'.dependencies]
-libc = "0.2.42"
-
-[[bench]]
-name = "spawn"
-path = "spawn.rs"
-harness = false
-
-[[bench]]
-name = "sync_broadcast"
-path = "sync_broadcast.rs"
-harness = false
-
-[[bench]]
-name = "sync_mpsc"
-path = "sync_mpsc.rs"
-harness = false
-
-[[bench]]
-name = "sync_mpsc_oneshot"
-path = "sync_mpsc_oneshot.rs"
-harness = false
-
-[[bench]]
-name = "sync_watch"
-path = "sync_watch.rs"
-harness = false
-
-[[bench]]
-name = "rt_current_thread"
-path = "rt_current_thread.rs"
-harness = false
-
-[[bench]]
-name = "rt_multi_threaded"
-path = "rt_multi_threaded.rs"
-harness = false
-
-[[bench]]
-name = "sync_notify"
-path = "sync_notify.rs"
-harness = false
-
-[[bench]]
-name = "sync_rwlock"
-path = "sync_rwlock.rs"
-harness = false
-
-[[bench]]
-name = "sync_semaphore"
-path = "sync_semaphore.rs"
-harness = false
-
-[[bench]]
-name = "signal"
-path = "signal.rs"
-harness = false
-
-[[bench]]
-name = "fs"
-path = "fs.rs"
-harness = false
-
-[[bench]]
-name = "copy"
-path = "copy.rs"
-harness = false
-
-[[bench]]
-name = "time_now"
-path = "time_now.rs"
-harness = false
-
-[[bench]]
-name = "time_timeout"
-path = "time_timeout.rs"
-harness = false
-
-[[bench]]
-name = "spawn_blocking"
-path = "spawn_blocking.rs"
-harness = false
-
-[[bench]]
-name = "remote_spawn"
-path = "remote_spawn.rs"
-harness = false
-
-[lints]
-workspace = true

benches/copy.rs

@@ -1,250 +0,0 @@
-use criterion::{criterion_group, criterion_main, Criterion};
-
-use rand::{Rng, SeedableRng};
-use rand_chacha::ChaCha20Rng;
-
-use tokio::io::{copy, repeat, AsyncRead, AsyncReadExt, AsyncWrite};
-use tokio::time::{interval, Interval, MissedTickBehavior};
-
-use std::task::Poll;
-use std::time::Duration;
-
-const KILO: usize = 1024;
-
-// Tunable parameters if you want to change this benchmark. If reader and writer
-// are matched in kilobytes per second, then this only exposes buffering to the
-// benchmark.
-const RNG_SEED: u64 = 0;
-// How much data to copy in a single benchmark run
-const SOURCE_SIZE: u64 = 256 * KILO as u64;
-// Read side provides CHUNK_SIZE every READ_SERVICE_PERIOD. If it's not called
-// frequently, it'll burst to catch up (representing OS buffers draining)
-const CHUNK_SIZE: usize = 2 * KILO;
-const READ_SERVICE_PERIOD: Duration = Duration::from_millis(1);
-// Write side buffers up to WRITE_BUFFER, and flushes to disk every
-// WRITE_SERVICE_PERIOD.
-const WRITE_BUFFER: usize = 40 * KILO;
-const WRITE_SERVICE_PERIOD: Duration = Duration::from_millis(20);
-// How likely you are to have to wait for previously written data to be flushed
-// because another writer claimed the buffer space
-const PROBABILITY_FLUSH_WAIT: f64 = 0.1;
-
-/// A slow writer that aims to simulate HDD behavior under heavy load.
-///
-/// There is a limited buffer, which is fully drained on the next write after
-/// a time limit is reached. Flush waits for the time limit to be reached
-/// and then drains the buffer.
-///
-/// At random, the HDD will stall writers while it flushes out all buffers. If
-/// this happens to you, you will be unable to write until the next time the
-/// buffer is drained.
-struct SlowHddWriter {
-    service_intervals: Interval,
-    blocking_rng: ChaCha20Rng,
-    buffer_size: usize,
-    buffer_used: usize,
-}
-
-impl SlowHddWriter {
-    fn new(service_interval: Duration, buffer_size: usize) -> Self {
-        let blocking_rng = ChaCha20Rng::seed_from_u64(RNG_SEED);
-        let mut service_intervals = interval(service_interval);
-        service_intervals.set_missed_tick_behavior(MissedTickBehavior::Delay);
-        Self {
-            service_intervals,
-            blocking_rng,
-            buffer_size,
-            buffer_used: 0,
-        }
-    }
-
-    fn service_write(
-        mut self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), std::io::Error>> {
-        // If we hit a service interval, the buffer can be cleared
-        let res = self.service_intervals.poll_tick(cx).map(|_| Ok(()));
-        if res.is_ready() {
-            self.buffer_used = 0;
-        }
-        res
-    }
-
-    fn write_bytes(
-        mut self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        writable: usize,
-    ) -> std::task::Poll<Result<usize, std::io::Error>> {
-        let service_res = self.as_mut().service_write(cx);
-
-        if service_res.is_pending() && self.blocking_rng.random_bool(PROBABILITY_FLUSH_WAIT) {
-            return Poll::Pending;
-        }
-        let available = self.buffer_size - self.buffer_used;
-
-        if available == 0 {
-            assert!(service_res.is_pending());
-            Poll::Pending
-        } else {
-            let written = available.min(writable);
-            self.buffer_used += written;
-            Poll::Ready(Ok(written))
-        }
-    }
-}
-
-impl Unpin for SlowHddWriter {}
-
-impl AsyncWrite for SlowHddWriter {
-    fn poll_write(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        buf: &[u8],
-    ) -> std::task::Poll<Result<usize, std::io::Error>> {
-        self.write_bytes(cx, buf.len())
-    }
-
-    fn poll_flush(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), std::io::Error>> {
-        self.service_write(cx)
-    }
-
-    fn poll_shutdown(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), std::io::Error>> {
-        self.service_write(cx)
-    }
-
-    fn poll_write_vectored(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        bufs: &[std::io::IoSlice<'_>],
-    ) -> std::task::Poll<Result<usize, std::io::Error>> {
-        let writable = bufs.iter().fold(0, |acc, buf| acc + buf.len());
-        self.write_bytes(cx, writable)
-    }
-
-    fn is_write_vectored(&self) -> bool {
-        true
-    }
-}
-
-/// A reader that limits the maximum chunk it'll give you back
-///
-/// Simulates something reading from a slow link - you get one chunk per call,
-/// and you are offered chunks on a schedule
-struct ChunkReader {
-    data: Vec<u8>,
-    service_intervals: Interval,
-}
-
-impl ChunkReader {
-    fn new(chunk_size: usize, service_interval: Duration) -> Self {
-        let mut service_intervals = interval(service_interval);
-        service_intervals.set_missed_tick_behavior(MissedTickBehavior::Burst);
-        let data: Vec<u8> = std::iter::repeat_n(0, chunk_size).collect();
-        Self {
-            data,
-            service_intervals,
-        }
-    }
-}
-
-impl AsyncRead for ChunkReader {
-    fn poll_read(
-        mut self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        buf: &mut tokio::io::ReadBuf<'_>,
-    ) -> Poll<std::io::Result<()>> {
-        if self.service_intervals.poll_tick(cx).is_pending() {
-            return Poll::Pending;
-        }
-        buf.put_slice(&self.data[..buf.remaining().min(self.data.len())]);
-        Poll::Ready(Ok(()))
-    }
-}
-
-fn rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_current_thread()
-        .enable_time()
-        .build()
-        .unwrap()
-}
-
-fn copy_mem_to_mem(c: &mut Criterion) {
-    let rt = rt();
-
-    c.bench_function("copy_mem_to_mem", |b| {
-        b.iter(|| {
-            let task = || async {
-                let mut source = repeat(0).take(SOURCE_SIZE);
-                let mut dest = Vec::new();
-                copy(&mut source, &mut dest).await.unwrap();
-            };
-
-            rt.block_on(task());
-        })
-    });
-}
-
-fn copy_mem_to_slow_hdd(c: &mut Criterion) {
-    let rt = rt();
-
-    c.bench_function("copy_mem_to_slow_hdd", |b| {
-        b.iter(|| {
-            let task = || async {
-                let mut source = repeat(0).take(SOURCE_SIZE);
-                let mut dest = SlowHddWriter::new(WRITE_SERVICE_PERIOD, WRITE_BUFFER);
-                copy(&mut source, &mut dest).await.unwrap();
-            };
-
-            rt.block_on(task());
-        })
-    });
-}
-
-fn copy_chunk_to_mem(c: &mut Criterion) {
-    let rt = rt();
-
-    c.bench_function("copy_chunk_to_mem", |b| {
-        b.iter(|| {
-            let task = || async {
-                let mut source =
-                    ChunkReader::new(CHUNK_SIZE, READ_SERVICE_PERIOD).take(SOURCE_SIZE);
-                let mut dest = Vec::new();
-                copy(&mut source, &mut dest).await.unwrap();
-            };
-
-            rt.block_on(task());
-        })
-    });
-}
-
-fn copy_chunk_to_slow_hdd(c: &mut Criterion) {
-    let rt = rt();
-
-    c.bench_function("copy_chunk_to_slow_hdd", |b| {
-        b.iter(|| {
-            let task = || async {
-                let mut source =
-                    ChunkReader::new(CHUNK_SIZE, READ_SERVICE_PERIOD).take(SOURCE_SIZE);
-                let mut dest = SlowHddWriter::new(WRITE_SERVICE_PERIOD, WRITE_BUFFER);
-                copy(&mut source, &mut dest).await.unwrap();
-            };
-
-            rt.block_on(task());
-        })
-    });
-}
-
-criterion_group!(
-    copy_bench,
-    copy_mem_to_mem,
-    copy_mem_to_slow_hdd,
-    copy_chunk_to_mem,
-    copy_chunk_to_slow_hdd,
-);
-criterion_main!(copy_bench);

benches/fs.rs

@@ -1,112 +0,0 @@
-#![cfg(unix)]
-
-use tokio_stream::StreamExt;
-
-use tokio::fs::File;
-use tokio::io::AsyncReadExt;
-use tokio_util::codec::{BytesCodec, FramedRead /*FramedWrite*/};
-
-use criterion::{criterion_group, criterion_main, Criterion};
-
-use std::fs::File as StdFile;
-use std::io::Read as StdRead;
-
-fn rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(2)
-        .build()
-        .unwrap()
-}
-
-const BLOCK_COUNT: usize = 1_000;
-
-const BUFFER_SIZE: usize = 4096;
-const DEV_ZERO: &str = "/dev/zero";
-
-fn async_read_codec(c: &mut Criterion) {
-    let rt = rt();
-
-    c.bench_function("async_read_codec", |b| {
-        b.iter(|| {
-            let task = || async {
-                let file = File::open(DEV_ZERO).await.unwrap();
-                let mut input_stream =
-                    FramedRead::with_capacity(file, BytesCodec::new(), BUFFER_SIZE);
-
-                for _i in 0..BLOCK_COUNT {
-                    let _bytes = input_stream.next().await.unwrap();
-                }
-            };
-
-            rt.block_on(task());
-        })
-    });
-}
-
-fn async_read_buf(c: &mut Criterion) {
-    let rt = rt();
-
-    c.bench_function("async_read_buf", |b| {
-        b.iter(|| {
-            let task = || async {
-                let mut file = File::open(DEV_ZERO).await.unwrap();
-                let mut buffer = [0u8; BUFFER_SIZE];
-
-                for _i in 0..BLOCK_COUNT {
-                    let count = file.read(&mut buffer).await.unwrap();
-                    if count == 0 {
-                        break;
-                    }
-                }
-            };
-
-            rt.block_on(task());
-        });
-    });
-}
-
-fn async_read_std_file(c: &mut Criterion) {
-    let rt = rt();
-
-    c.bench_function("async_read_std_file", |b| {
-        b.iter(|| {
-            let task = || async {
-                let mut file =
-                    tokio::task::block_in_place(|| Box::pin(StdFile::open(DEV_ZERO).unwrap()));
-
-                for _i in 0..BLOCK_COUNT {
-                    let mut buffer = [0u8; BUFFER_SIZE];
-                    let mut file_ref = file.as_mut();
-
-                    tokio::task::block_in_place(move || {
-                        file_ref.read_exact(&mut buffer).unwrap();
-                    });
-                }
-            };
-
-            rt.block_on(task());
-        });
-    });
-}
-
-fn sync_read(c: &mut Criterion) {
-    c.bench_function("sync_read", |b| {
-        b.iter(|| {
-            let mut file = StdFile::open(DEV_ZERO).unwrap();
-            let mut buffer = [0u8; BUFFER_SIZE];
-
-            for _i in 0..BLOCK_COUNT {
-                file.read_exact(&mut buffer).unwrap();
-            }
-        })
-    });
-}
-
-criterion_group!(
-    file,
-    async_read_std_file,
-    async_read_buf,
-    async_read_codec,
-    sync_read
-);
-criterion_main!(file);

benches/latency.rs

@@ -0,0 +1,114 @@
+#![feature(test)]
+
+extern crate test;
+#[macro_use]
+extern crate futures;
+extern crate tokio;
+
+use std::io;
+use std::net::SocketAddr;
+use std::thread;
+
+use futures::sync::mpsc;
+use futures::sync::oneshot;
+use futures::{Future, Poll, Sink, Stream};
+use test::Bencher;
+use tokio::net::UdpSocket;
+
+/// UDP echo server
+struct EchoServer {
+    socket: UdpSocket,
+    buf: Vec<u8>,
+    to_send: Option<(usize, SocketAddr)>,
+}
+
+impl EchoServer {
+    fn new(s: UdpSocket) -> Self {
+        EchoServer {
+            socket: s,
+            to_send: None,
+            buf: vec![0u8; 1600],
+        }
+    }
+}
+
+impl Future for EchoServer {
+    type Item = ();
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<(), io::Error> {
+        loop {
+            if let Some(&(size, peer)) = self.to_send.as_ref() {
+                try_ready!(self.socket.poll_send_to(&self.buf[..size], &peer));
+                self.to_send = None;
+            }
+            self.to_send = Some(try_ready!(self.socket.poll_recv_from(&mut self.buf)));
+        }
+    }
+}
+
+#[bench]
+fn udp_echo_latency(b: &mut Bencher) {
+    let any_addr = "127.0.0.1:0".to_string();
+    let any_addr = any_addr.parse::<SocketAddr>().unwrap();
+
+    let (stop_c, stop_p) = oneshot::channel::<()>();
+    let (tx, rx) = oneshot::channel();
+
+    let child = thread::spawn(move || {
+        let socket = tokio::net::UdpSocket::bind(&any_addr).unwrap();
+        tx.send(socket.local_addr().unwrap()).unwrap();
+
+        let server = EchoServer::new(socket);
+        let server = server.select(stop_p.map_err(|_| panic!()));
+        let server = server.map_err(|_| ());
+        server.wait().unwrap();
+    });
+
+    let client = std::net::UdpSocket::bind(&any_addr).unwrap();
+
+    let server_addr = rx.wait().unwrap();
+    let mut buf = [0u8; 1000];
+
+    // warmup phase; for some reason initial couple of
+    // runs are much slower
+    //
+    // TODO: Describe the exact reasons; caching? branch predictor? lazy closures?
+    for _ in 0..8 {
+        client.send_to(&buf, &server_addr).unwrap();
+        let _ = client.recv_from(&mut buf).unwrap();
+    }
+
+    b.iter(|| {
+        client.send_to(&buf, &server_addr).unwrap();
+        let _ = client.recv_from(&mut buf).unwrap();
+    });
+
+    stop_c.send(()).unwrap();
+    child.join().unwrap();
+}
+
+#[bench]
+fn futures_channel_latency(b: &mut Bencher) {
+    let (mut in_tx, in_rx) = mpsc::channel(32);
+    let (out_tx, out_rx) = mpsc::channel::<_>(32);
+
+    let child = thread::spawn(|| out_tx.send_all(in_rx.then(|r| r.unwrap())).wait());
+    let mut rx_iter = out_rx.wait();
+
+    // warmup phase; for some reason initial couple of runs are much slower
+    //
+    // TODO: Describe the exact reasons; caching? branch predictor? lazy closures?
+    for _ in 0..8 {
+        in_tx.start_send(Ok(1usize)).unwrap();
+        let _ = rx_iter.next();
+    }
+
+    b.iter(|| {
+        in_tx.start_send(Ok(1usize)).unwrap();
+        let _ = rx_iter.next();
+    });
+
+    drop(in_tx);
+    child.join().unwrap().unwrap();
+}

benches/mio-ops.rs

@@ -0,0 +1,56 @@
+// Measure cost of different operations
+// to get a sense of performance tradeoffs
+#![feature(test)]
+
+extern crate mio;
+extern crate test;
+
+use test::Bencher;
+
+use mio::tcp::TcpListener;
+use mio::{PollOpt, Ready, Token};
+
+#[bench]
+fn mio_register_deregister(b: &mut Bencher) {
+    let addr = "127.0.0.1:0".parse().unwrap();
+    // Setup the server socket
+    let sock = TcpListener::bind(&addr).unwrap();
+    let poll = mio::Poll::new().unwrap();
+
+    const CLIENT: Token = Token(1);
+
+    b.iter(|| {
+        poll.register(&sock, CLIENT, Ready::readable(), PollOpt::edge())
+            .unwrap();
+        poll.deregister(&sock).unwrap();
+    });
+}
+
+#[bench]
+fn mio_reregister(b: &mut Bencher) {
+    let addr = "127.0.0.1:0".parse().unwrap();
+    // Setup the server socket
+    let sock = TcpListener::bind(&addr).unwrap();
+    let poll = mio::Poll::new().unwrap();
+
+    const CLIENT: Token = Token(1);
+    poll.register(&sock, CLIENT, Ready::readable(), PollOpt::edge())
+        .unwrap();
+
+    b.iter(|| {
+        poll.reregister(&sock, CLIENT, Ready::readable(), PollOpt::edge())
+            .unwrap();
+    });
+    poll.deregister(&sock).unwrap();
+}
+
+#[bench]
+fn mio_poll(b: &mut Bencher) {
+    let poll = mio::Poll::new().unwrap();
+    let timeout = std::time::Duration::new(0, 0);
+    let mut events = mio::Events::with_capacity(1024);
+
+    b.iter(|| {
+        poll.poll(&mut events, Some(timeout)).unwrap();
+    });
+}

benches/remote_spawn.rs

@@ -1,101 +0,0 @@
-//! Benchmark remote task spawning (push_remote_task) at different concurrency
-//! levels on the multi-threaded scheduler.
-//!
-//! This measures contention on the scheduler's inject queue mutex when multiple
-//! external (non-worker) threads spawn tasks into the tokio runtime simultaneously.
-//! Every rt.spawn() from an external thread unconditionally goes through
-//! push_remote_task, making this a direct measurement of inject queue contention.
-//!
-//! For each parallelism level N (1, 2, 4, 8, 16, 32, 64, capped at available parallelism):
-//! - Spawns N std::threads (external to the runtime)
-//! - Each thread spawns TOTAL_TASKS / N tasks into the runtime via rt.spawn()
-//! - All threads are synchronized with a barrier to maximize contention
-//! - Tasks are trivial no-ops to isolate the push overhead
-
-use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
-use std::sync::Barrier;
-use tokio::runtime::{self, Runtime};
-
-/// Total number of tasks spawned across all threads per iteration.
-/// Must be divisible by the largest parallelism level (64).
-const TOTAL_TASKS: usize = 12_800;
-const _: () = assert!(
-    TOTAL_TASKS.is_multiple_of(64),
-    "TOTAL_TASKS must be divisible by 64"
-);
-
-fn remote_spawn_contention(c: &mut Criterion) {
-    let parallelism_levels = parallelism_levels();
-    let mut group = c.benchmark_group("remote_spawn");
-
-    for num_threads in &parallelism_levels {
-        let num_threads = *num_threads;
-        group.bench_with_input(
-            BenchmarkId::new("threads", num_threads),
-            &num_threads,
-            |b, &num_threads| {
-                let rt = rt();
-                let tasks_per_thread = TOTAL_TASKS / num_threads;
-                let barrier = Barrier::new(num_threads);
-
-                b.iter_custom(|iters| {
-                    let mut total_duration = std::time::Duration::ZERO;
-                    for _ in 0..iters {
-                        let start = std::time::Instant::now();
-
-                        let all_handles = std::thread::scope(|s| {
-                            let handles: Vec<_> = (0..num_threads)
-                                .map(|_| {
-                                    let barrier = &barrier;
-                                    let rt = &rt;
-                                    s.spawn(move || {
-                                        let mut join_handles = Vec::with_capacity(tasks_per_thread);
-                                        barrier.wait();
-
-                                        for _ in 0..tasks_per_thread {
-                                            join_handles.push(rt.spawn(async {}));
-                                        }
-                                        join_handles
-                                    })
-                                })
-                                .collect();
-
-                            handles
-                                .into_iter()
-                                .flat_map(|h| h.join().unwrap())
-                                .collect::<Vec<_>>()
-                        });
-
-                        total_duration += start.elapsed();
-
-                        rt.block_on(async {
-                            for h in all_handles {
-                                h.await.unwrap();
-                            }
-                        });
-                    }
-                    total_duration
-                });
-            },
-        );
-    }
-
-    group.finish();
-}
-
-fn parallelism_levels() -> Vec<usize> {
-    let max_parallelism = std::thread::available_parallelism().map_or(1, |p| p.get());
-
-    [1, 2, 4, 8, 16, 32, 64]
-        .into_iter()
-        .filter(|&n| n <= max_parallelism)
-        .collect()
-}
-
-fn rt() -> Runtime {
-    runtime::Builder::new_multi_thread().build().unwrap()
-}
-
-criterion_group!(remote_spawn_benches, remote_spawn_contention);
-
-criterion_main!(remote_spawn_benches);

benches/rt_current_thread.rs

@@ -1,89 +0,0 @@
-//! Benchmark implementation details of the threaded scheduler. These benches are
-//! intended to be used as a form of regression testing and not as a general
-//! purpose benchmark demonstrating real-world performance.
-
-use tokio::runtime::{self, Runtime};
-
-use criterion::{criterion_group, criterion_main, Criterion};
-
-const NUM_SPAWN: usize = 1_000;
-
-fn rt_curr_spawn_many_local(c: &mut Criterion) {
-    let rt = rt();
-    let mut handles = Vec::with_capacity(NUM_SPAWN);
-
-    c.bench_function("spawn_many_local", |b| {
-        b.iter(|| {
-            rt.block_on(async {
-                for _ in 0..NUM_SPAWN {
-                    handles.push(tokio::spawn(async move {}));
-                }
-
-                for handle in handles.drain(..) {
-                    handle.await.unwrap();
-                }
-            });
-        })
-    });
-}
-
-fn rt_curr_spawn_many_remote_idle(c: &mut Criterion) {
-    let rt = rt();
-    let rt_handle = rt.handle();
-    let mut handles = Vec::with_capacity(NUM_SPAWN);
-
-    c.bench_function("spawn_many_remote_idle", |b| {
-        b.iter(|| {
-            for _ in 0..NUM_SPAWN {
-                handles.push(rt_handle.spawn(async {}));
-            }
-
-            rt.block_on(async {
-                for handle in handles.drain(..) {
-                    handle.await.unwrap();
-                }
-            });
-        })
-    });
-}
-
-fn rt_curr_spawn_many_remote_busy(c: &mut Criterion) {
-    let rt = rt();
-    let rt_handle = rt.handle();
-    let mut handles = Vec::with_capacity(NUM_SPAWN);
-
-    rt.spawn(async {
-        fn iter() {
-            tokio::spawn(async { iter() });
-        }
-
-        iter()
-    });
-
-    c.bench_function("spawn_many_remote_busy", |b| {
-        b.iter(|| {
-            for _ in 0..NUM_SPAWN {
-                handles.push(rt_handle.spawn(async {}));
-            }
-
-            rt.block_on(async {
-                for handle in handles.drain(..) {
-                    handle.await.unwrap();
-                }
-            });
-        })
-    });
-}
-
-fn rt() -> Runtime {
-    runtime::Builder::new_current_thread().build().unwrap()
-}
-
-criterion_group!(
-    rt_curr_scheduler,
-    rt_curr_spawn_many_local,
-    rt_curr_spawn_many_remote_idle,
-    rt_curr_spawn_many_remote_busy
-);
-
-criterion_main!(rt_curr_scheduler);

benches/rt_multi_threaded.rs

@@ -1,276 +0,0 @@
-//! Benchmark implementation details of the threaded scheduler. These benches are
-//! intended to be used as a form of regression testing and not as a general
-//! purpose benchmark demonstrating real-world performance.
-
-use tokio::runtime::{self, Runtime};
-use tokio::sync::oneshot;
-
-use std::sync::atomic::Ordering::Relaxed;
-use std::sync::atomic::{AtomicBool, AtomicUsize};
-use std::sync::{mpsc, Arc};
-use std::time::{Duration, Instant};
-
-use criterion::{criterion_group, criterion_main, Criterion};
-
-const NUM_WORKERS: usize = 4;
-const NUM_SPAWN: usize = 10_000;
-const STALL_DUR: Duration = Duration::from_micros(10);
-
-fn rt_multi_spawn_many_local(c: &mut Criterion) {
-    let rt = rt();
-
-    let (tx, rx) = mpsc::sync_channel(1000);
-    let rem = Arc::new(AtomicUsize::new(0));
-
-    c.bench_function("spawn_many_local", |b| {
-        b.iter(|| {
-            rem.store(NUM_SPAWN, Relaxed);
-
-            rt.block_on(async {
-                for _ in 0..NUM_SPAWN {
-                    let tx = tx.clone();
-                    let rem = rem.clone();
-
-                    tokio::spawn(async move {
-                        if 1 == rem.fetch_sub(1, Relaxed) {
-                            tx.send(()).unwrap();
-                        }
-                    });
-                }
-
-                rx.recv().unwrap();
-            });
-        })
-    });
-}
-
-fn rt_multi_spawn_many_remote_idle(c: &mut Criterion) {
-    let rt = rt();
-
-    let mut handles = Vec::with_capacity(NUM_SPAWN);
-
-    c.bench_function("spawn_many_remote_idle", |b| {
-        b.iter(|| {
-            for _ in 0..NUM_SPAWN {
-                handles.push(rt.spawn(async {}));
-            }
-
-            rt.block_on(async {
-                for handle in handles.drain(..) {
-                    handle.await.unwrap();
-                }
-            });
-        })
-    });
-}
-
-// The runtime is busy with tasks that consume CPU time and yield. Yielding is a
-// lower notification priority than spawning / regular notification.
-fn rt_multi_spawn_many_remote_busy1(c: &mut Criterion) {
-    let rt = rt();
-    let rt_handle = rt.handle();
-    let mut handles = Vec::with_capacity(NUM_SPAWN);
-    let flag = Arc::new(AtomicBool::new(true));
-
-    // Spawn some tasks to keep the runtimes busy
-    for _ in 0..(2 * NUM_WORKERS) {
-        let flag = flag.clone();
-        rt.spawn(async move {
-            while flag.load(Relaxed) {
-                tokio::task::yield_now().await;
-                stall();
-            }
-        });
-    }
-
-    c.bench_function("spawn_many_remote_busy1", |b| {
-        b.iter(|| {
-            for _ in 0..NUM_SPAWN {
-                handles.push(rt_handle.spawn(async {}));
-            }
-
-            rt.block_on(async {
-                for handle in handles.drain(..) {
-                    handle.await.unwrap();
-                }
-            });
-        })
-    });
-
-    flag.store(false, Relaxed);
-}
-
-// The runtime is busy with tasks that consume CPU time and spawn new high-CPU
-// tasks. Spawning goes via a higher notification priority than yielding.
-fn rt_multi_spawn_many_remote_busy2(c: &mut Criterion) {
-    const NUM_SPAWN: usize = 1_000;
-
-    let rt = rt();
-    let rt_handle = rt.handle();
-    let mut handles = Vec::with_capacity(NUM_SPAWN);
-    let flag = Arc::new(AtomicBool::new(true));
-
-    // Spawn some tasks to keep the runtimes busy
-    for _ in 0..(NUM_WORKERS) {
-        let flag = flag.clone();
-        fn iter(flag: Arc<AtomicBool>) {
-            tokio::spawn(async {
-                if flag.load(Relaxed) {
-                    stall();
-                    iter(flag);
-                }
-            });
-        }
-        rt.spawn(async {
-            iter(flag);
-        });
-    }
-
-    c.bench_function("spawn_many_remote_busy2", |b| {
-        b.iter(|| {
-            for _ in 0..NUM_SPAWN {
-                handles.push(rt_handle.spawn(async {}));
-            }
-
-            rt.block_on(async {
-                for handle in handles.drain(..) {
-                    handle.await.unwrap();
-                }
-            });
-        })
-    });
-
-    flag.store(false, Relaxed);
-}
-
-fn rt_multi_yield_many(c: &mut Criterion) {
-    const NUM_YIELD: usize = 1_000;
-    const TASKS: usize = 200;
-
-    c.bench_function("yield_many", |b| {
-        let rt = rt();
-        let (tx, rx) = mpsc::sync_channel(TASKS);
-
-        b.iter(move || {
-            for _ in 0..TASKS {
-                let tx = tx.clone();
-
-                rt.spawn(async move {
-                    for _ in 0..NUM_YIELD {
-                        tokio::task::yield_now().await;
-                    }
-
-                    tx.send(()).unwrap();
-                });
-            }
-
-            for _ in 0..TASKS {
-                rx.recv().unwrap();
-            }
-        })
-    });
-}
-
-fn rt_multi_ping_pong(c: &mut Criterion) {
-    const NUM_PINGS: usize = 1_000;
-
-    let rt = rt();
-
-    let (done_tx, done_rx) = mpsc::sync_channel(1000);
-    let rem = Arc::new(AtomicUsize::new(0));
-
-    c.bench_function("ping_pong", |b| {
-        b.iter(|| {
-            let done_tx = done_tx.clone();
-            let rem = rem.clone();
-            rem.store(NUM_PINGS, Relaxed);
-
-            rt.block_on(async {
-                tokio::spawn(async move {
-                    for _ in 0..NUM_PINGS {
-                        let rem = rem.clone();
-                        let done_tx = done_tx.clone();
-
-                        tokio::spawn(async move {
-                            let (tx1, rx1) = oneshot::channel();
-                            let (tx2, rx2) = oneshot::channel();
-
-                            tokio::spawn(async move {
-                                rx1.await.unwrap();
-                                tx2.send(()).unwrap();
-                            });
-
-                            tx1.send(()).unwrap();
-                            rx2.await.unwrap();
-
-                            if 1 == rem.fetch_sub(1, Relaxed) {
-                                done_tx.send(()).unwrap();
-                            }
-                        });
-                    }
-                });
-
-                done_rx.recv().unwrap();
-            });
-        })
-    });
-}
-
-fn rt_multi_chained_spawn(c: &mut Criterion) {
-    const ITER: usize = 1_000;
-
-    fn iter(done_tx: mpsc::SyncSender<()>, n: usize) {
-        if n == 0 {
-            done_tx.send(()).unwrap();
-        } else {
-            tokio::spawn(async move {
-                iter(done_tx, n - 1);
-            });
-        }
-    }
-
-    c.bench_function("chained_spawn", |b| {
-        let rt = rt();
-        let (done_tx, done_rx) = mpsc::sync_channel(1000);
-
-        b.iter(move || {
-            let done_tx = done_tx.clone();
-
-            rt.block_on(async {
-                tokio::spawn(async move {
-                    iter(done_tx, ITER);
-                });
-
-                done_rx.recv().unwrap();
-            });
-        })
-    });
-}
-
-fn rt() -> Runtime {
-    runtime::Builder::new_multi_thread()
-        .worker_threads(NUM_WORKERS)
-        .enable_all()
-        .build()
-        .unwrap()
-}
-
-fn stall() {
-    let now = Instant::now();
-    while now.elapsed() < STALL_DUR {
-        std::thread::yield_now();
-    }
-}
-
-criterion_group!(
-    rt_multi_scheduler,
-    rt_multi_spawn_many_local,
-    rt_multi_spawn_many_remote_idle,
-    rt_multi_spawn_many_remote_busy1,
-    rt_multi_spawn_many_remote_busy2,
-    rt_multi_ping_pong,
-    rt_multi_yield_many,
-    rt_multi_chained_spawn,
-);
-
-criterion_main!(rt_multi_scheduler);

benches/signal.rs

@@ -1,97 +0,0 @@
-//! Benchmark the delay in propagating OS signals to any listeners.
-#![cfg(unix)]
-
-use criterion::{criterion_group, criterion_main, Criterion};
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::runtime;
-use tokio::signal::unix::{signal, SignalKind};
-use tokio::sync::mpsc;
-
-struct Spinner {
-    count: usize,
-}
-
-impl Future for Spinner {
-    type Output = ();
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        if self.count > 3 {
-            Poll::Ready(())
-        } else {
-            self.count += 1;
-            cx.waker().wake_by_ref();
-            Poll::Pending
-        }
-    }
-}
-
-impl Spinner {
-    fn new() -> Self {
-        Self { count: 0 }
-    }
-}
-
-pub fn send_signal(signal: libc::c_int) {
-    use libc::{getpid, kill};
-
-    unsafe {
-        assert_eq!(kill(getpid(), signal), 0);
-    }
-}
-
-fn many_signals(c: &mut Criterion) {
-    let num_signals = 10;
-    let (tx, mut rx) = mpsc::channel(num_signals);
-
-    // Intentionally single threaded to measure delays in propagating wakes
-    let rt = runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-        .unwrap();
-
-    let spawn_signal = |kind| {
-        let tx = tx.clone();
-        rt.spawn(async move {
-            let mut signal = signal(kind).expect("failed to create signal");
-
-            while signal.recv().await.is_some() {
-                if tx.send(()).await.is_err() {
-                    break;
-                }
-            }
-        });
-    };
-
-    for _ in 0..num_signals {
-        // Pick some random signals which don't terminate the test harness
-        spawn_signal(SignalKind::child());
-        spawn_signal(SignalKind::io());
-    }
-    drop(tx);
-
-    // Turn the runtime for a while to ensure that all the spawned
-    // tasks have been polled at least once
-    rt.block_on(Spinner::new());
-
-    c.bench_function("many_signals", |b| {
-        b.iter(|| {
-            rt.block_on(async {
-                send_signal(libc::SIGCHLD);
-                for _ in 0..num_signals {
-                    rx.recv().await.expect("channel closed");
-                }
-
-                send_signal(libc::SIGIO);
-                for _ in 0..num_signals {
-                    rx.recv().await.expect("channel closed");
-                }
-            });
-        })
-    });
-}
-
-criterion_group!(signal_group, many_signals);
-
-criterion_main!(signal_group);

benches/spawn.rs

@@ -1,96 +0,0 @@
-//! Benchmark spawning a task onto the basic and threaded Tokio executors.
-//! This essentially measure the time to enqueue a task in the local and remote
-//! case.
-
-use criterion::{black_box, criterion_group, criterion_main, Criterion};
-
-async fn work() -> usize {
-    let val = 1 + 1;
-    tokio::task::yield_now().await;
-    black_box(val)
-}
-
-fn single_rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_current_thread()
-        .build()
-        .unwrap()
-}
-
-fn multi_rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(1)
-        .build()
-        .unwrap()
-}
-
-fn basic_scheduler_spawn(c: &mut Criterion) {
-    let runtime = single_rt();
-
-    c.bench_function("basic_scheduler_spawn", |b| {
-        b.iter(|| {
-            runtime.block_on(async {
-                let h = tokio::spawn(work());
-                assert_eq!(h.await.unwrap(), 2);
-            });
-        })
-    });
-}
-
-fn basic_scheduler_spawn10(c: &mut Criterion) {
-    let runtime = single_rt();
-
-    c.bench_function("basic_scheduler_spawn10", |b| {
-        b.iter(|| {
-            runtime.block_on(async {
-                let mut handles = Vec::with_capacity(10);
-                for _ in 0..10 {
-                    handles.push(tokio::spawn(work()));
-                }
-                for handle in handles {
-                    assert_eq!(handle.await.unwrap(), 2);
-                }
-            });
-        })
-    });
-}
-
-fn threaded_scheduler_spawn(c: &mut Criterion) {
-    let runtime = multi_rt();
-
-    c.bench_function("threaded_scheduler_spawn", |b| {
-        b.iter(|| {
-            runtime.block_on(async {
-                let h = tokio::spawn(work());
-                assert_eq!(h.await.unwrap(), 2);
-            });
-        })
-    });
-}
-
-fn threaded_scheduler_spawn10(c: &mut Criterion) {
-    let runtime = multi_rt();
-
-    c.bench_function("threaded_scheduler_spawn10", |b| {
-        b.iter(|| {
-            runtime.block_on(async {
-                let mut handles = Vec::with_capacity(10);
-                for _ in 0..10 {
-                    handles.push(tokio::spawn(work()));
-                }
-                for handle in handles {
-                    assert_eq!(handle.await.unwrap(), 2);
-                }
-            });
-        })
-    });
-}
-
-criterion_group!(
-    spawn,
-    basic_scheduler_spawn,
-    basic_scheduler_spawn10,
-    threaded_scheduler_spawn,
-    threaded_scheduler_spawn10,
-);
-
-criterion_main!(spawn);

benches/spawn_blocking.rs

@@ -1,71 +0,0 @@
-//! Benchmark spawn_blocking at different concurrency levels on the multi-threaded scheduler.
-//!
-//! For each parallelism level N (1, 2, 4, 8, 16, 32, 64, capped at available parallelism):
-//! - Spawns N regular async tasks
-//! - Each task spawns M batches of B spawn_blocking tasks (no-ops)
-//! - Each batch is awaited to completion before starting the next
-
-use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
-use tokio::runtime::{self, Runtime};
-use tokio::task::JoinSet;
-
-/// Number of batches per task
-const NUM_BATCHES: usize = 100;
-/// Number of spawn_blocking calls per batch
-const BATCH_SIZE: usize = 16;
-
-fn spawn_blocking_concurrency(c: &mut Criterion) {
-    let max_parallelism = std::thread::available_parallelism().map_or(1, |p| p.get());
-
-    let parallelism_levels: Vec<usize> = [1, 2, 4, 8, 16, 32, 64]
-        .into_iter()
-        .filter(|&n| n <= max_parallelism)
-        .collect();
-
-    let mut group = c.benchmark_group("spawn_blocking");
-
-    for num_tasks in parallelism_levels {
-        group.bench_with_input(
-            BenchmarkId::new("concurrency", num_tasks),
-            &num_tasks,
-            |b, &num_tasks| {
-                let rt = rt();
-
-                b.iter(|| {
-                    rt.block_on(async {
-                        let mut tasks = JoinSet::new();
-
-                        for _ in 0..num_tasks {
-                            tasks.spawn(async {
-                                for _ in 0..NUM_BATCHES {
-                                    let mut batch = JoinSet::new();
-
-                                    for _ in 0..BATCH_SIZE {
-                                        batch.spawn_blocking(|| black_box(0));
-                                    }
-
-                                    batch.join_all().await;
-                                }
-                            });
-                        }
-
-                        tasks.join_all().await;
-                    });
-                });
-            },
-        );
-    }
-
-    group.finish();
-}
-
-fn rt() -> Runtime {
-    runtime::Builder::new_multi_thread()
-        .enable_all()
-        .build()
-        .unwrap()
-}
-
-criterion_group!(spawn_blocking_benches, spawn_blocking_concurrency);
-
-criterion_main!(spawn_blocking_benches);

benches/sync_broadcast.rs

@@ -1,82 +0,0 @@
-use rand::{Rng, RngCore, SeedableRng};
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
-use tokio::sync::{broadcast, Notify};
-
-use criterion::measurement::WallTime;
-use criterion::{black_box, criterion_group, criterion_main, BenchmarkGroup, Criterion};
-
-fn rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(6)
-        .build()
-        .unwrap()
-}
-
-fn do_work(rng: &mut impl RngCore) -> u32 {
-    use std::fmt::Write;
-    let mut message = String::new();
-    for i in 1..=10 {
-        let _ = write!(&mut message, " {i}={}", rng.random::<f64>());
-    }
-    message
-        .as_bytes()
-        .iter()
-        .map(|&c| c as u32)
-        .fold(0, u32::wrapping_add)
-}
-
-fn contention_impl<const N_TASKS: usize>(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    let (tx, _rx) = broadcast::channel::<usize>(1000);
-    let wg = Arc::new((AtomicUsize::new(0), Notify::new()));
-
-    for n in 0..N_TASKS {
-        let wg = wg.clone();
-        let mut rx = tx.subscribe();
-        let mut rng = rand::rngs::StdRng::seed_from_u64(n as u64);
-        rt.spawn(async move {
-            while (rx.recv().await).is_ok() {
-                let r = do_work(&mut rng);
-                let _ = black_box(r);
-                if wg.0.fetch_sub(1, Ordering::Relaxed) == 1 {
-                    wg.1.notify_one();
-                }
-            }
-        });
-    }
-
-    const N_ITERS: usize = 100;
-
-    g.bench_function(N_TASKS.to_string(), |b| {
-        b.iter(|| {
-            rt.block_on({
-                let wg = wg.clone();
-                let tx = tx.clone();
-                async move {
-                    for i in 0..N_ITERS {
-                        assert_eq!(wg.0.fetch_add(N_TASKS, Ordering::Relaxed), 0);
-                        tx.send(i).unwrap();
-                        while wg.0.load(Ordering::Relaxed) > 0 {
-                            wg.1.notified().await;
-                        }
-                    }
-                }
-            })
-        })
-    });
-}
-
-fn bench_contention(c: &mut Criterion) {
-    let mut group = c.benchmark_group("contention");
-    contention_impl::<10>(&mut group);
-    contention_impl::<100>(&mut group);
-    contention_impl::<500>(&mut group);
-    contention_impl::<1000>(&mut group);
-    group.finish();
-}
-
-criterion_group!(contention, bench_contention);
-
-criterion_main!(contention);

benches/sync_mpsc.rs

@@ -1,399 +0,0 @@
-use tokio::sync::mpsc;
-
-use criterion::measurement::WallTime;
-use criterion::{black_box, criterion_group, criterion_main, BenchmarkGroup, Criterion};
-
-#[derive(Debug, Copy, Clone)]
-struct Medium(#[allow(dead_code)] [usize; 64]);
-impl Default for Medium {
-    fn default() -> Self {
-        Medium([0; 64])
-    }
-}
-
-#[derive(Debug, Copy, Clone)]
-struct Large(#[allow(dead_code)] [Medium; 64]);
-impl Default for Large {
-    fn default() -> Self {
-        Large([Medium::default(); 64])
-    }
-}
-
-#[cfg(target_pointer_width = "64")]
-const ARRAY_CAP: usize = 32;
-
-#[cfg(not(target_pointer_width = "64"))]
-const ARRAY_CAP: usize = 16;
-
-const LIST_CAP: usize = ARRAY_CAP + 1;
-const ROUNDTRIP_ITERS: usize = 1_000;
-
-fn rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(6)
-        .build()
-        .unwrap()
-}
-
-fn create_medium<const SIZE: usize>(g: &mut BenchmarkGroup<WallTime>) {
-    g.bench_function(SIZE.to_string(), |b| {
-        b.iter(|| {
-            black_box(&mpsc::channel::<Medium>(SIZE));
-        })
-    });
-}
-
-fn create_data<T, const SIZE: usize>(g: &mut BenchmarkGroup<WallTime>, prefix: &str) {
-    g.bench_function(format!("{prefix}_{SIZE}"), |b| {
-        b.iter(|| {
-            black_box(mpsc::channel::<T>(SIZE));
-        })
-    });
-}
-
-fn send_data<T: Default, const SIZE: usize>(g: &mut BenchmarkGroup<WallTime>, prefix: &str) {
-    let rt = rt();
-
-    g.bench_function(format!("{prefix}_{SIZE}"), |b| {
-        b.iter(|| {
-            let (tx, mut rx) = mpsc::channel::<T>(SIZE);
-
-            let _ = rt.block_on(tx.send(T::default()));
-
-            rt.block_on(rx.recv()).unwrap();
-        })
-    });
-}
-
-fn roundtrip_try_send_recv_data<T: Default, const SIZE: usize>(
-    g: &mut BenchmarkGroup<WallTime>,
-    prefix: &str,
-) {
-    let (tx, mut rx) = mpsc::channel::<T>(SIZE);
-
-    g.bench_function(format!("{prefix}_{SIZE}"), |b| {
-        b.iter(|| {
-            for _ in 0..ROUNDTRIP_ITERS {
-                tx.try_send(T::default()).unwrap();
-                black_box(rx.try_recv().unwrap());
-            }
-        })
-    });
-}
-
-fn contention_bounded(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("bounded", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::channel::<usize>(1_000_000);
-
-                for _ in 0..5 {
-                    let tx = tx.clone();
-                    tokio::spawn(async move {
-                        for i in 0..1000 {
-                            tx.send(i).await.unwrap();
-                        }
-                    });
-                }
-
-                for _ in 0..1_000 * 5 {
-                    let _ = rx.recv().await;
-                }
-            })
-        })
-    });
-}
-
-fn contention_bounded_recv_many(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("bounded_recv_many", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::channel::<usize>(1_000_000);
-
-                for _ in 0..5 {
-                    let tx = tx.clone();
-                    tokio::spawn(async move {
-                        for i in 0..1000 {
-                            tx.send(i).await.unwrap();
-                        }
-                    });
-                }
-
-                let mut buffer = Vec::<usize>::with_capacity(5_000);
-                let mut total = 0;
-                while total < 1_000 * 5 {
-                    total += rx.recv_many(&mut buffer, 5_000).await;
-                }
-            })
-        })
-    });
-}
-
-fn contention_bounded_full(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("bounded_full", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::channel::<usize>(100);
-
-                for _ in 0..5 {
-                    let tx = tx.clone();
-                    tokio::spawn(async move {
-                        for i in 0..1000 {
-                            tx.send(i).await.unwrap();
-                        }
-                    });
-                }
-
-                for _ in 0..1_000 * 5 {
-                    let _ = rx.recv().await;
-                }
-            })
-        })
-    });
-}
-
-fn contention_bounded_full_recv_many(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("bounded_full_recv_many", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::channel::<usize>(100);
-
-                for _ in 0..5 {
-                    let tx = tx.clone();
-                    tokio::spawn(async move {
-                        for i in 0..1000 {
-                            tx.send(i).await.unwrap();
-                        }
-                    });
-                }
-
-                let mut buffer = Vec::<usize>::with_capacity(5_000);
-                let mut total = 0;
-                while total < 1_000 * 5 {
-                    total += rx.recv_many(&mut buffer, 5_000).await;
-                }
-            })
-        })
-    });
-}
-
-fn contention_unbounded(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("unbounded", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::unbounded_channel::<usize>();
-
-                for _ in 0..5 {
-                    let tx = tx.clone();
-                    tokio::spawn(async move {
-                        for i in 0..1000 {
-                            tx.send(i).unwrap();
-                        }
-                    });
-                }
-
-                for _ in 0..1_000 * 5 {
-                    let _ = rx.recv().await;
-                }
-            })
-        })
-    });
-}
-
-fn contention_unbounded_recv_many(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("unbounded_recv_many", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::unbounded_channel::<usize>();
-
-                for _ in 0..5 {
-                    let tx = tx.clone();
-                    tokio::spawn(async move {
-                        for i in 0..1000 {
-                            tx.send(i).unwrap();
-                        }
-                    });
-                }
-
-                let mut buffer = Vec::<usize>::with_capacity(5_000);
-                let mut total = 0;
-                while total < 1_000 * 5 {
-                    total += rx.recv_many(&mut buffer, 5_000).await;
-                }
-            })
-        })
-    });
-}
-
-fn uncontented_bounded(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("bounded", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::channel::<usize>(1_000_000);
-
-                for i in 0..5000 {
-                    tx.send(i).await.unwrap();
-                }
-
-                for _ in 0..5_000 {
-                    let _ = rx.recv().await;
-                }
-            })
-        })
-    });
-}
-
-fn uncontented_bounded_recv_many(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("bounded_recv_many", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::channel::<usize>(1_000_000);
-
-                for i in 0..5000 {
-                    tx.send(i).await.unwrap();
-                }
-
-                let mut buffer = Vec::<usize>::with_capacity(5_000);
-                let mut total = 0;
-                while total < 1_000 * 5 {
-                    total += rx.recv_many(&mut buffer, 5_000).await;
-                }
-            })
-        })
-    });
-}
-
-fn uncontented_unbounded(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("unbounded", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::unbounded_channel::<usize>();
-
-                for i in 0..5000 {
-                    tx.send(i).unwrap();
-                }
-
-                for _ in 0..5_000 {
-                    let _ = rx.recv().await;
-                }
-            })
-        })
-    });
-}
-
-fn uncontented_unbounded_recv_many(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-
-    g.bench_function("unbounded_recv_many", |b| {
-        b.iter(|| {
-            rt.block_on(async move {
-                let (tx, mut rx) = mpsc::unbounded_channel::<usize>();
-
-                for i in 0..5000 {
-                    tx.send(i).unwrap();
-                }
-
-                let mut buffer = Vec::<usize>::with_capacity(5_000);
-                let mut total = 0;
-                while total < 1_000 * 5 {
-                    total += rx.recv_many(&mut buffer, 5_000).await;
-                }
-            })
-        })
-    });
-}
-
-fn bench_create_medium(c: &mut Criterion) {
-    let mut group = c.benchmark_group("create_medium");
-    create_medium::<1>(&mut group);
-    create_medium::<100>(&mut group);
-    create_medium::<100_000>(&mut group);
-    group.finish();
-}
-
-fn bench_create_small(c: &mut Criterion) {
-    let mut group = c.benchmark_group("create_small");
-    create_data::<Medium, 1>(&mut group, "medium");
-    create_data::<Medium, 10>(&mut group, "medium");
-    create_data::<Medium, ARRAY_CAP>(&mut group, "medium");
-    create_data::<Medium, LIST_CAP>(&mut group, "medium");
-    create_data::<Large, 1>(&mut group, "large");
-    create_data::<Large, 10>(&mut group, "large");
-    create_data::<Large, ARRAY_CAP>(&mut group, "large");
-    create_data::<Large, LIST_CAP>(&mut group, "large");
-    group.finish();
-}
-
-fn bench_send(c: &mut Criterion) {
-    let mut group = c.benchmark_group("send");
-    send_data::<Medium, 1>(&mut group, "medium");
-    send_data::<Medium, 10>(&mut group, "medium");
-    send_data::<Medium, 1000>(&mut group, "medium");
-    send_data::<Large, 1>(&mut group, "large");
-    send_data::<Large, 10>(&mut group, "large");
-    send_data::<Large, 1000>(&mut group, "large");
-    group.finish();
-}
-
-fn bench_roundtrip_try_send_recv(c: &mut Criterion) {
-    let mut group = c.benchmark_group("roundtrip_try_send_recv");
-    roundtrip_try_send_recv_data::<Medium, 1>(&mut group, "medium");
-    roundtrip_try_send_recv_data::<Medium, 10>(&mut group, "medium");
-    roundtrip_try_send_recv_data::<Medium, ARRAY_CAP>(&mut group, "medium");
-    roundtrip_try_send_recv_data::<Medium, LIST_CAP>(&mut group, "medium");
-    group.finish();
-}
-
-fn bench_contention(c: &mut Criterion) {
-    let mut group = c.benchmark_group("contention");
-    contention_bounded(&mut group);
-    contention_bounded_recv_many(&mut group);
-    contention_bounded_full(&mut group);
-    contention_bounded_full_recv_many(&mut group);
-    contention_unbounded(&mut group);
-    contention_unbounded_recv_many(&mut group);
-    group.finish();
-}
-
-fn bench_uncontented(c: &mut Criterion) {
-    let mut group = c.benchmark_group("uncontented");
-    uncontented_bounded(&mut group);
-    uncontented_bounded_recv_many(&mut group);
-    uncontented_unbounded(&mut group);
-    uncontented_unbounded_recv_many(&mut group);
-    group.finish();
-}
-
-criterion_group!(create, bench_create_medium);
-criterion_group!(create_small, bench_create_small);
-criterion_group!(send, bench_send);
-criterion_group!(roundtrip_try_send_recv, bench_roundtrip_try_send_recv);
-criterion_group!(contention, bench_contention);
-criterion_group!(uncontented, bench_uncontented);
-
-criterion_main!(
-    create,
-    create_small,
-    send,
-    roundtrip_try_send_recv,
-    contention,
-    uncontented
-);

benches/sync_mpsc_oneshot.rs

@@ -1,56 +0,0 @@
-use tokio::{
-    runtime::Runtime,
-    sync::{mpsc, oneshot},
-};
-
-use criterion::{criterion_group, criterion_main, Criterion};
-
-fn request_reply_current_thread(c: &mut Criterion) {
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .build()
-        .unwrap();
-
-    request_reply(c, rt);
-}
-
-fn request_reply_multi_threaded(c: &mut Criterion) {
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(1)
-        .build()
-        .unwrap();
-
-    request_reply(c, rt);
-}
-
-fn request_reply(b: &mut Criterion, rt: Runtime) {
-    let tx = rt.block_on(async move {
-        let (tx, mut rx) = mpsc::channel::<oneshot::Sender<()>>(10);
-        tokio::spawn(async move {
-            while let Some(reply) = rx.recv().await {
-                reply.send(()).unwrap();
-            }
-        });
-        tx
-    });
-
-    b.bench_function("request_reply", |b| {
-        b.iter(|| {
-            let task_tx = tx.clone();
-            rt.block_on(async move {
-                for _ in 0..1_000 {
-                    let (o_tx, o_rx) = oneshot::channel();
-                    task_tx.send(o_tx).await.unwrap();
-                    let _ = o_rx.await;
-                }
-            })
-        })
-    });
-}
-
-criterion_group!(
-    sync_mpsc_oneshot_group,
-    request_reply_current_thread,
-    request_reply_multi_threaded,
-);
-
-criterion_main!(sync_mpsc_oneshot_group);

benches/sync_notify.rs

@@ -1,104 +0,0 @@
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
-
-use tokio::sync::Notify;
-
-use criterion::measurement::WallTime;
-use criterion::{criterion_group, criterion_main, BenchmarkGroup, Criterion};
-
-fn rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(6)
-        .build()
-        .unwrap()
-}
-
-fn notify_waiters<const N_WAITERS: usize>(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-    let notify = Arc::new(Notify::new());
-    let counter = Arc::new(AtomicUsize::new(0));
-    for _ in 0..N_WAITERS {
-        rt.spawn({
-            let notify = notify.clone();
-            let counter = counter.clone();
-            async move {
-                loop {
-                    notify.notified().await;
-                    counter.fetch_add(1, Ordering::Relaxed);
-                }
-            }
-        });
-    }
-
-    const N_ITERS: usize = 500;
-    g.bench_function(N_WAITERS.to_string(), |b| {
-        b.iter(|| {
-            counter.store(0, Ordering::Relaxed);
-            loop {
-                notify.notify_waiters();
-                if counter.load(Ordering::Relaxed) >= N_ITERS {
-                    break;
-                }
-            }
-        })
-    });
-}
-
-fn notify_one<const N_WAITERS: usize>(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-    let notify = Arc::new(Notify::new());
-    let counter = Arc::new(AtomicUsize::new(0));
-    for _ in 0..N_WAITERS {
-        rt.spawn({
-            let notify = notify.clone();
-            let counter = counter.clone();
-            async move {
-                loop {
-                    notify.notified().await;
-                    counter.fetch_add(1, Ordering::Relaxed);
-                }
-            }
-        });
-    }
-
-    const N_ITERS: usize = 500;
-    g.bench_function(N_WAITERS.to_string(), |b| {
-        b.iter(|| {
-            counter.store(0, Ordering::Relaxed);
-            loop {
-                notify.notify_one();
-                if counter.load(Ordering::Relaxed) >= N_ITERS {
-                    break;
-                }
-            }
-        })
-    });
-}
-
-fn bench_notify_one(c: &mut Criterion) {
-    let mut group = c.benchmark_group("notify_one");
-    notify_one::<10>(&mut group);
-    notify_one::<50>(&mut group);
-    notify_one::<100>(&mut group);
-    notify_one::<200>(&mut group);
-    notify_one::<500>(&mut group);
-    group.finish();
-}
-
-fn bench_notify_waiters(c: &mut Criterion) {
-    let mut group = c.benchmark_group("notify_waiters");
-    notify_waiters::<10>(&mut group);
-    notify_waiters::<50>(&mut group);
-    notify_waiters::<100>(&mut group);
-    notify_waiters::<200>(&mut group);
-    notify_waiters::<500>(&mut group);
-    group.finish();
-}
-
-criterion_group!(
-    notify_waiters_simple,
-    bench_notify_one,
-    bench_notify_waiters
-);
-
-criterion_main!(notify_waiters_simple);

benches/sync_rwlock.rs

@@ -1,163 +0,0 @@
-use std::sync::Arc;
-use tokio::{sync::RwLock, task};
-
-use criterion::measurement::WallTime;
-use criterion::{black_box, criterion_group, criterion_main, BenchmarkGroup, Criterion};
-
-fn read_uncontended(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(6)
-        .build()
-        .unwrap();
-
-    let lock = Arc::new(RwLock::new(()));
-    g.bench_function("read", |b| {
-        b.iter(|| {
-            let lock = lock.clone();
-            rt.block_on(async move {
-                for _ in 0..6 {
-                    let read = lock.read().await;
-                    let _read = black_box(read);
-                }
-            })
-        })
-    });
-}
-
-fn read_concurrent_uncontended_multi(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(6)
-        .build()
-        .unwrap();
-
-    async fn task(lock: Arc<RwLock<()>>) {
-        let read = lock.read().await;
-        let _read = black_box(read);
-    }
-
-    let lock = Arc::new(RwLock::new(()));
-    g.bench_function("read_concurrent_multi", |b| {
-        b.iter(|| {
-            let lock = lock.clone();
-            rt.block_on(async move {
-                let j = tokio::try_join! {
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone()))
-                };
-                j.unwrap();
-            })
-        })
-    });
-}
-
-fn read_concurrent_uncontended(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .build()
-        .unwrap();
-
-    async fn task(lock: Arc<RwLock<()>>) {
-        let read = lock.read().await;
-        let _read = black_box(read);
-    }
-
-    let lock = Arc::new(RwLock::new(()));
-    g.bench_function("read_concurrent", |b| {
-        b.iter(|| {
-            let lock = lock.clone();
-            rt.block_on(async move {
-                tokio::join! {
-                    task(lock.clone()),
-                    task(lock.clone()),
-                    task(lock.clone()),
-                    task(lock.clone()),
-                    task(lock.clone()),
-                    task(lock.clone())
-                };
-            })
-        })
-    });
-}
-
-fn read_concurrent_contended_multi(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(6)
-        .build()
-        .unwrap();
-
-    async fn task(lock: Arc<RwLock<()>>) {
-        let read = lock.read().await;
-        let _read = black_box(read);
-    }
-
-    let lock = Arc::new(RwLock::new(()));
-    g.bench_function("read_concurrent_multi", |b| {
-        b.iter(|| {
-            let lock = lock.clone();
-            rt.block_on(async move {
-                let write = lock.write().await;
-                let j = tokio::try_join! {
-                    async move { drop(write); Ok(()) },
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone())),
-                    task::spawn(task(lock.clone())),
-                };
-                j.unwrap();
-            })
-        })
-    });
-}
-
-fn read_concurrent_contended(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .build()
-        .unwrap();
-
-    async fn task(lock: Arc<RwLock<()>>) {
-        let read = lock.read().await;
-        let _read = black_box(read);
-    }
-
-    let lock = Arc::new(RwLock::new(()));
-    g.bench_function("read_concurrent", |b| {
-        b.iter(|| {
-            let lock = lock.clone();
-            rt.block_on(async move {
-                let write = lock.write().await;
-                tokio::join! {
-                    async move { drop(write) },
-                    task(lock.clone()),
-                    task(lock.clone()),
-                    task(lock.clone()),
-                    task(lock.clone()),
-                    task(lock.clone()),
-                };
-            })
-        })
-    });
-}
-
-fn bench_contention(c: &mut Criterion) {
-    let mut group = c.benchmark_group("contention");
-    read_concurrent_contended(&mut group);
-    read_concurrent_contended_multi(&mut group);
-    group.finish();
-}
-
-fn bench_uncontented(c: &mut Criterion) {
-    let mut group = c.benchmark_group("uncontented");
-    read_uncontended(&mut group);
-    read_concurrent_uncontended(&mut group);
-    read_concurrent_uncontended_multi(&mut group);
-    group.finish();
-}
-
-criterion_group!(contention, bench_contention);
-criterion_group!(uncontented, bench_uncontented);
-
-criterion_main!(contention, uncontented);

benches/sync_semaphore.rs

@@ -1,147 +0,0 @@
-use std::sync::Arc;
-use tokio::runtime::Runtime;
-use tokio::{sync::Semaphore, task};
-
-use criterion::measurement::WallTime;
-use criterion::{criterion_group, criterion_main, BenchmarkGroup, Criterion};
-
-fn single_rt() -> Runtime {
-    tokio::runtime::Builder::new_current_thread()
-        .build()
-        .unwrap()
-}
-
-fn multi_rt() -> Runtime {
-    tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(6)
-        .build()
-        .unwrap()
-}
-
-fn uncontended(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = multi_rt();
-
-    let s = Arc::new(Semaphore::new(10));
-    g.bench_function("multi", |b| {
-        b.iter(|| {
-            let s = s.clone();
-            rt.block_on(async move {
-                for _ in 0..6 {
-                    let permit = s.acquire().await;
-                    drop(permit);
-                }
-            })
-        })
-    });
-}
-
-async fn task(s: Arc<Semaphore>) {
-    let permit = s.acquire().await;
-    drop(permit);
-}
-
-fn uncontended_concurrent_multi(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = multi_rt();
-
-    let s = Arc::new(Semaphore::new(10));
-    g.bench_function("concurrent_multi", |b| {
-        b.iter(|| {
-            let s = s.clone();
-            rt.block_on(async move {
-                let j = tokio::try_join! {
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone()))
-                };
-                j.unwrap();
-            })
-        })
-    });
-}
-
-fn uncontended_concurrent_single(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = single_rt();
-
-    let s = Arc::new(Semaphore::new(10));
-    g.bench_function("concurrent_single", |b| {
-        b.iter(|| {
-            let s = s.clone();
-            rt.block_on(async move {
-                tokio::join! {
-                    task(s.clone()),
-                    task(s.clone()),
-                    task(s.clone()),
-                    task(s.clone()),
-                    task(s.clone()),
-                    task(s.clone())
-                };
-            })
-        })
-    });
-}
-
-fn contended_concurrent_multi(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = multi_rt();
-
-    let s = Arc::new(Semaphore::new(5));
-    g.bench_function("concurrent_multi", |b| {
-        b.iter(|| {
-            let s = s.clone();
-            rt.block_on(async move {
-                let j = tokio::try_join! {
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone())),
-                    task::spawn(task(s.clone()))
-                };
-                j.unwrap();
-            })
-        })
-    });
-}
-
-fn contended_concurrent_single(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = single_rt();
-
-    let s = Arc::new(Semaphore::new(5));
-    g.bench_function("concurrent_single", |b| {
-        b.iter(|| {
-            let s = s.clone();
-            rt.block_on(async move {
-                tokio::join! {
-                    task(s.clone()),
-                    task(s.clone()),
-                    task(s.clone()),
-                    task(s.clone()),
-                    task(s.clone()),
-                    task(s.clone())
-                };
-            })
-        })
-    });
-}
-
-fn bench_contention(c: &mut Criterion) {
-    let mut group = c.benchmark_group("contention");
-    contended_concurrent_multi(&mut group);
-    contended_concurrent_single(&mut group);
-    group.finish();
-}
-
-fn bench_uncontented(c: &mut Criterion) {
-    let mut group = c.benchmark_group("uncontented");
-    uncontended(&mut group);
-    uncontended_concurrent_multi(&mut group);
-    uncontended_concurrent_single(&mut group);
-    group.finish();
-}
-
-criterion_group!(contention, bench_contention);
-criterion_group!(uncontented, bench_uncontented);
-
-criterion_main!(contention, uncontented);

benches/sync_watch.rs

@@ -1,85 +0,0 @@
-use rand::prelude::*;
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::Arc;
-use tokio::sync::{watch, Notify};
-
-use criterion::measurement::WallTime;
-use criterion::{black_box, criterion_group, criterion_main, BenchmarkGroup, Criterion};
-
-fn rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(6)
-        .build()
-        .unwrap()
-}
-
-fn do_work(rng: &mut impl RngCore) -> u32 {
-    use std::fmt::Write;
-    let mut message = String::new();
-    for i in 1..=10 {
-        let _ = write!(&mut message, " {i}={}", rng.random::<f64>());
-    }
-    message
-        .as_bytes()
-        .iter()
-        .map(|&c| c as u32)
-        .fold(0, u32::wrapping_add)
-}
-
-fn contention_resubscribe<const N_TASKS: usize>(g: &mut BenchmarkGroup<WallTime>) {
-    let rt = rt();
-    let (snd, _) = watch::channel(0i32);
-    let snd = Arc::new(snd);
-    let wg = Arc::new((AtomicU64::new(0), Notify::new()));
-    for n in 0..N_TASKS {
-        let mut rcv = snd.subscribe();
-        let wg = wg.clone();
-        let mut rng = rand::rngs::StdRng::seed_from_u64(n as u64);
-        rt.spawn(async move {
-            while rcv.changed().await.is_ok() {
-                let _ = *rcv.borrow(); // contend on rwlock
-                let r = do_work(&mut rng);
-                let _ = black_box(r);
-                if wg.0.fetch_sub(1, Ordering::Release) == 1 {
-                    wg.1.notify_one();
-                }
-            }
-        });
-    }
-
-    const N_ITERS: usize = 100;
-    g.bench_function(N_TASKS.to_string(), |b| {
-        b.iter(|| {
-            rt.block_on({
-                let snd = snd.clone();
-                let wg = wg.clone();
-                async move {
-                    tokio::spawn(async move {
-                        for _ in 0..N_ITERS {
-                            assert_eq!(wg.0.fetch_add(N_TASKS as u64, Ordering::Relaxed), 0);
-                            let _ = snd.send(black_box(42));
-                            while wg.0.load(Ordering::Acquire) > 0 {
-                                wg.1.notified().await;
-                            }
-                        }
-                    })
-                    .await
-                    .unwrap();
-                }
-            });
-        })
-    });
-}
-
-fn bench_contention_resubscribe(c: &mut Criterion) {
-    let mut group = c.benchmark_group("contention_resubscribe");
-    contention_resubscribe::<10>(&mut group);
-    contention_resubscribe::<100>(&mut group);
-    contention_resubscribe::<500>(&mut group);
-    contention_resubscribe::<1000>(&mut group);
-    group.finish();
-}
-
-criterion_group!(contention, bench_contention_resubscribe);
-
-criterion_main!(contention);

benches/tcp.rs

@@ -0,0 +1,260 @@
+#![feature(test)]
+
+extern crate futures;
+extern crate tokio;
+
+#[macro_use]
+extern crate tokio_io;
+
+pub extern crate test;
+
+mod prelude {
+    pub use futures::*;
+    pub use tokio::net::{TcpListener, TcpStream};
+    pub use tokio::reactor::Reactor;
+    pub use tokio_io::io::read_to_end;
+
+    pub use std::io::{self, Read, Write};
+    pub use std::thread;
+    pub use std::time::Duration;
+    pub use test::{self, Bencher};
+}
+
+mod connect_churn {
+    use prelude::*;
+
+    const NUM: usize = 300;
+    const CONCURRENT: usize = 8;
+
+    #[bench]
+    fn one_thread(b: &mut Bencher) {
+        let addr = "127.0.0.1:0".parse().unwrap();
+
+        b.iter(move || {
+            let listener = TcpListener::bind(&addr).unwrap();
+            let addr = listener.local_addr().unwrap();
+
+            // Spawn a single future that accepts & drops connections
+            let serve_incomings = listener
+                .incoming()
+                .map_err(|e| panic!("server err: {:?}", e))
+                .for_each(|_| Ok(()));
+
+            let connects = stream::iter_result((0..NUM).map(|_| {
+                Ok(TcpStream::connect(&addr).and_then(|sock| {
+                    sock.set_linger(Some(Duration::from_secs(0))).unwrap();
+                    read_to_end(sock, vec![])
+                }))
+            }));
+
+            let connects_concurrent = connects
+                .buffer_unordered(CONCURRENT)
+                .map_err(|e| panic!("client err: {:?}", e))
+                .for_each(|_| Ok(()));
+
+            serve_incomings
+                .select(connects_concurrent)
+                .map(|_| ())
+                .map_err(|_| ())
+                .wait()
+                .unwrap();
+        });
+    }
+
+    fn n_workers(n: usize, b: &mut Bencher) {
+        let (shutdown_tx, shutdown_rx) = sync::oneshot::channel();
+        let (addr_tx, addr_rx) = sync::oneshot::channel();
+
+        // Spawn reactor thread
+        let server_thread = thread::spawn(move || {
+            // Bind the TCP listener
+            let listener = TcpListener::bind(&"127.0.0.1:0".parse().unwrap()).unwrap();
+
+            // Get the address being listened on.
+            let addr = listener.local_addr().unwrap();
+
+            // Send the remote & address back to the main thread
+            addr_tx.send(addr).unwrap();
+
+            // Spawn a single future that accepts & drops connections
+            let serve_incomings = listener
+                .incoming()
+                .map_err(|e| panic!("server err: {:?}", e))
+                .for_each(|_| Ok(()));
+
+            // Run server
+            serve_incomings
+                .select(shutdown_rx)
+                .map(|_| ())
+                .map_err(|_| ())
+                .wait()
+                .unwrap();
+        });
+
+        // Get the bind addr of the server
+        let addr = addr_rx.wait().unwrap();
+
+        b.iter(move || {
+            use std::sync::{Arc, Barrier};
+
+            // Create a barrier to coordinate threads
+            let barrier = Arc::new(Barrier::new(n + 1));
+
+            // Spawn worker threads
+            let threads: Vec<_> = (0..n)
+                .map(|_| {
+                    let barrier = barrier.clone();
+                    let addr = addr.clone();
+
+                    thread::spawn(move || {
+                        let connects = stream::iter_result((0..(NUM / n)).map(|_| {
+                            Ok(TcpStream::connect(&addr)
+                                .map_err(|e| panic!("connect err: {:?}", e))
+                                .and_then(|sock| {
+                                    sock.set_linger(Some(Duration::from_secs(0))).unwrap();
+                                    read_to_end(sock, vec![])
+                                }))
+                        }));
+
+                        barrier.wait();
+
+                        connects
+                            .buffer_unordered(CONCURRENT)
+                            .map_err(|e| panic!("client err: {:?}", e))
+                            .for_each(|_| Ok(()))
+                            .wait()
+                            .unwrap();
+                    })
+                })
+                .collect();
+
+            barrier.wait();
+
+            for th in threads {
+                th.join().unwrap();
+            }
+        });
+
+        // Shutdown the server
+        shutdown_tx.send(()).unwrap();
+        server_thread.join().unwrap();
+    }
+
+    #[bench]
+    fn two_threads(b: &mut Bencher) {
+        n_workers(1, b);
+    }
+
+    #[bench]
+    fn multi_threads(b: &mut Bencher) {
+        n_workers(4, b);
+    }
+}
+
+mod transfer {
+    use prelude::*;
+    use std::{cmp, mem};
+
+    const MB: usize = 3 * 1024 * 1024;
+
+    struct Drain {
+        sock: TcpStream,
+        chunk: usize,
+    }
+
+    impl Future for Drain {
+        type Item = ();
+        type Error = io::Error;
+
+        fn poll(&mut self) -> Poll<(), io::Error> {
+            let mut buf: [u8; 1024] = unsafe { mem::uninitialized() };
+
+            loop {
+                match try_nb!(self.sock.read(&mut buf[..self.chunk])) {
+                    0 => return Ok(Async::Ready(())),
+                    _ => {}
+                }
+            }
+        }
+    }
+
+    struct Transfer {
+        sock: TcpStream,
+        rem: usize,
+        chunk: usize,
+    }
+
+    impl Future for Transfer {
+        type Item = ();
+        type Error = io::Error;
+
+        fn poll(&mut self) -> Poll<(), io::Error> {
+            while self.rem > 0 {
+                let len = cmp::min(self.rem, self.chunk);
+                let buf = &DATA[..len];
+
+                let n = try_nb!(self.sock.write(&buf));
+                self.rem -= n;
+            }
+
+            Ok(Async::Ready(()))
+        }
+    }
+
+    static DATA: [u8; 1024] = [0; 1024];
+
+    fn one_thread(b: &mut Bencher, read_size: usize, write_size: usize) {
+        let addr = "127.0.0.1:0".parse().unwrap();
+
+        b.iter(move || {
+            let listener = TcpListener::bind(&addr).unwrap();
+            let addr = listener.local_addr().unwrap();
+
+            // Spawn a single future that accepts 1 connection, Drain it and drops
+            let server = listener
+                .incoming()
+                .into_future() // take the first connection
+                .map_err(|(e, _other_incomings)| e)
+                .map(|(connection, _other_incomings)| connection.unwrap())
+                .and_then(|sock| {
+                    sock.set_linger(Some(Duration::from_secs(0))).unwrap();
+                    let drain = Drain {
+                        sock: sock,
+                        chunk: read_size,
+                    };
+                    drain
+                        .map(|_| ())
+                        .map_err(|e| panic!("server error: {:?}", e))
+                })
+                .map_err(|e| panic!("server err: {:?}", e));
+
+            let client = TcpStream::connect(&addr)
+                .and_then(move |sock| Transfer {
+                    sock: sock,
+                    rem: MB,
+                    chunk: write_size,
+                })
+                .map_err(|e| panic!("client err: {:?}", e));
+
+            server.join(client).wait().unwrap();
+        });
+    }
+
+    mod small_chunks {
+        use prelude::*;
+
+        #[bench]
+        fn one_thread(b: &mut Bencher) {
+            super::one_thread(b, 32, 32);
+        }
+    }
+
+    mod big_chunks {
+        use prelude::*;
+
+        #[bench]
+        fn one_thread(b: &mut Bencher) {
+            super::one_thread(b, 1_024, 1_024);
+        }
+    }
+}

benches/time_now.rs

@@ -1,24 +0,0 @@
-//! Benchmark spawning a task onto the basic and threaded Tokio executors.
-//! This essentially measure the time to enqueue a task in the local and remote
-//! case.
-
-use criterion::{black_box, criterion_group, criterion_main, Criterion};
-
-fn time_now_current_thread(c: &mut Criterion) {
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .enable_time()
-        .build()
-        .unwrap();
-
-    c.bench_function("time_now_current_thread", |b| {
-        b.iter(|| {
-            rt.block_on(async {
-                black_box(tokio::time::Instant::now());
-            })
-        })
-    });
-}
-
-criterion_group!(time_now, time_now_current_thread);
-
-criterion_main!(time_now);

benches/time_timeout.rs

@@ -1,109 +0,0 @@
-use std::time::{Duration, Instant};
-
-use criterion::{black_box, criterion_group, criterion_main, Criterion};
-use tokio::{
-    runtime::Runtime,
-    time::{sleep, timeout},
-};
-
-// a very quick async task, but might timeout
-async fn quick_job() -> usize {
-    1
-}
-
-fn build_run_time(workers: usize) -> Runtime {
-    if workers == 1 {
-        tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap()
-    } else {
-        tokio::runtime::Builder::new_multi_thread()
-            .enable_all()
-            .worker_threads(workers)
-            .build()
-            .unwrap()
-    }
-}
-
-fn single_thread_scheduler_timeout(c: &mut Criterion) {
-    do_timeout_test(c, 1, "single_thread_timeout");
-}
-
-fn multi_thread_scheduler_timeout(c: &mut Criterion) {
-    do_timeout_test(c, 8, "multi_thread_timeout-8");
-}
-
-fn do_timeout_test(c: &mut Criterion, workers: usize, name: &str) {
-    let runtime = build_run_time(workers);
-    c.bench_function(name, |b| {
-        b.iter_custom(|iters| {
-            let start = Instant::now();
-            runtime.block_on(async {
-                black_box(spawn_timeout_job(iters as usize, workers)).await;
-            });
-            start.elapsed()
-        })
-    });
-}
-
-async fn spawn_timeout_job(iters: usize, procs: usize) {
-    let mut handles = Vec::with_capacity(procs);
-    for _ in 0..procs {
-        handles.push(tokio::spawn(async move {
-            for _ in 0..iters / procs {
-                let h = timeout(Duration::from_secs(1), quick_job());
-                assert_eq!(black_box(h.await.unwrap()), 1);
-            }
-        }));
-    }
-    for handle in handles {
-        handle.await.unwrap();
-    }
-}
-
-fn single_thread_scheduler_sleep(c: &mut Criterion) {
-    do_sleep_test(c, 1, "single_thread_sleep");
-}
-
-fn multi_thread_scheduler_sleep(c: &mut Criterion) {
-    do_sleep_test(c, 8, "multi_thread_sleep-8");
-}
-
-fn do_sleep_test(c: &mut Criterion, workers: usize, name: &str) {
-    let runtime = build_run_time(workers);
-
-    c.bench_function(name, |b| {
-        b.iter_custom(|iters| {
-            let start = Instant::now();
-            runtime.block_on(async {
-                black_box(spawn_sleep_job(iters as usize, workers)).await;
-            });
-            start.elapsed()
-        })
-    });
-}
-
-async fn spawn_sleep_job(iters: usize, procs: usize) {
-    let mut handles = Vec::with_capacity(procs);
-    for _ in 0..procs {
-        handles.push(tokio::spawn(async move {
-            for _ in 0..iters / procs {
-                let _h = black_box(sleep(Duration::from_secs(1)));
-            }
-        }));
-    }
-    for handle in handles {
-        handle.await.unwrap();
-    }
-}
-
-criterion_group!(
-    timeout_benchmark,
-    single_thread_scheduler_timeout,
-    multi_thread_scheduler_timeout,
-    single_thread_scheduler_sleep,
-    multi_thread_scheduler_sleep
-);
-
-criterion_main!(timeout_benchmark);

ci/azure-cargo-check.yml

@@ -0,0 +1,29 @@
+parameters:
+  noDefaultFeatures: '--no-default-features'
+
+jobs:
+- job: ${{ parameters.name }}
+  displayName: ${{ parameters.displayName }}
+  pool:
+    vmImage: ubuntu-16.04
+  steps:
+  - template: azure-install-rust.yml
+    parameters:
+      rust_version: ${{ parameters.rust }}
+
+  - template: azure-is-release.yml
+
+  - ${{ each crate in parameters.crates }}:
+    - ${{ each feature in crate.value }}:
+      - script: cargo check ${{ parameters.noDefaultFeatures }} --features ${{ feature }}
+        displayName: Check `${{ crate.key }}`, features = ${{ feature }}
+        workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }}
+        condition: and(succeeded(), not(variables['isRelease']))
+
+  - template: azure-patch-crates.yml
+
+  - ${{ each crate in parameters.crates }}:
+    - ${{ each feature in crate.value }}:
+      - script: cargo check ${{ parameters.noDefaultFeatures }} --features ${{ feature }}
+        displayName: Check `${{ crate.key }}`, features = ${{ feature }}
+        workingDirectory: $(Build.SourcesDirectory)/${{ crate.key }}

ci/azure-check-minrust.yml

@@ -0,0 +1,14 @@
+jobs:
+- job: ${{ parameters.name }}
+  displayName: Min supported Rust version
+  pool:
+    vmImage: ubuntu-16.04
+  steps:
+  - template: azure-install-rust.yml
+    parameters:
+      rust_version: ${{ parameters.rust_version }}
+
+  - template: azure-patch-crates.yml
+
+  - script: cargo check --all
+    displayName: cargo check --all

ci/azure-cross-compile.yml

@@ -0,0 +1,27 @@
+jobs:
+- job: ${{ parameters.name }}
+  displayName: ${{ parameters.displayName }}
+  pool:
+    vmImage: ubuntu-16.04
+  steps:
+  - template: azure-install-rust.yml
+    parameters:
+      rust_version: stable
+
+  - script: sudo apt-get update
+    displayName: "apt-get update"
+
+  - script: sudo apt-get install gcc-multilib
+    displayName: "Install gcc-multilib"
+
+  - script: rustup target add ${{ parameters.target }}
+    displayName: "Add target"
+
+  # Always patch
+  - template: azure-patch-crates.yml
+
+  - script: cargo check --all --exclude tokio-tls --target ${{ parameters.target }}
+    displayName: Check source
+
+  - script: cargo check --tests --all --exclude tokio-tls --target ${{ parameters.target }}
+    displayName: Check tests

ci/azure-deploy-docs.yml

@@ -0,0 +1,38 @@
+parameters:
+  dependsOn: []
+
+jobs:
+- job: documentation
+  displayName: 'Deploy API Documentation'
+  condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/master'))
+  pool:
+    vmImage: 'Ubuntu 16.04'
+  dependsOn:
+    - ${{ parameters.dependsOn }}
+  steps:
+  - template: azure-install-rust.yml
+    parameters:
+      rust_version: stable
+  - script: |
+      cargo doc --all --no-deps
+      cp -R target/doc '$(Build.BinariesDirectory)'
+    displayName: 'Generate Documentation'
+  - script: |
+      set -e
+
+      git --version
+      ls -la
+      git init
+      git config user.name 'Deployment Bot (from Azure Pipelines)'
+      git config user.email '[email protected]'
+      git config --global credential.helper 'store --file ~/.my-credentials'
+      printf "protocol=https\nhost=github.com\nusername=carllerche\npassword=%s\n\n" "$GITHUB_TOKEN" | git credential-store --file ~/.my-credentials store
+      git remote add origin https://github.com/tokio-rs/tokio
+      git checkout -b gh-pages
+      git add .
+      git commit -m 'Deploy Tokio API documentation'
+      git push -f origin gh-pages
+    env:
+      GITHUB_TOKEN: $(githubPersonalToken)
+    workingDirectory: '$(Build.BinariesDirectory)'
+    displayName: 'Deploy Documentation'

ci/azure-install-rust.yml

@@ -0,0 +1,33 @@
+steps:
+  # Linux and macOS.
+  - script: |
+      set -e
+      curl https://sh.rustup.rs -sSf | sh -s -- -y --default-toolchain none
+      export PATH=$PATH:$HOME/.cargo/bin
+      rustup toolchain install $RUSTUP_TOOLCHAIN
+      rustup default $RUSTUP_TOOLCHAIN
+      echo "##vso[task.setvariable variable=PATH;]$PATH:$HOME/.cargo/bin"
+    env:
+      RUSTUP_TOOLCHAIN: ${{parameters.rust_version}}
+    displayName: "Install rust (*nix)"
+    condition: not(eq(variables['Agent.OS'], 'Windows_NT'))
+
+  # Windows.
+  - script: |
+      echo "windows"
+      curl -sSf -o rustup-init.exe https://win.rustup.rs
+      rustup-init.exe -y --default-toolchain none
+      set PATH=%PATH%;%USERPROFILE%\.cargo\bin
+      rustup toolchain install %RUSTUP_TOOLCHAIN%
+      rustup default %RUSTUP_TOOLCHAIN%
+      echo "##vso[task.setvariable variable=PATH;]%PATH%;%USERPROFILE%\.cargo\bin"
+    env:
+      RUSTUP_TOOLCHAIN: ${{parameters.rust_version}}
+    displayName: Install rust (windows)
+    condition: eq(variables['Agent.OS'], 'Windows_NT')
+
+  # All platforms.
+  - script: |
+        rustc -Vv
+        cargo -V
+    displayName: Query rust and cargo versions

ci/azure-is-release.yml

@@ -0,0 +1,9 @@
+steps:
+  - bash: |
+      set -e
+
+      if git log --no-merges -1 --format='%B' | grep -qF '[ci-release]'; then
+        echo "##vso[task.setvariable variable=isRelease]true"
+      fi
+    failOnStderr: true
+    displayName: Check if release commit

ci/azure-patch-crates.yml

@@ -0,0 +1,16 @@
+steps:
+  - script: |
+      set -e
+
+      # Remove any existing patch statements
+      mv Cargo.toml Cargo.toml.bck
+      sed -n '/\[patch.crates-io\]/q;p' Cargo.toml.bck > Cargo.toml
+
+      # Patch all crates
+      cat ci/patch.toml >> Cargo.toml
+
+      # Print `Cargo.toml` for debugging
+      echo "~~~~ Cargo.toml ~~~~"
+      cat Cargo.toml
+      echo "~~~~~~~~~~~~~~~~~~~~"
+    displayName: Patch Cargo.toml

ci/azure-rustfmt.yml

@@ -0,0 +1,16 @@
+jobs:
+# Check formatting
+- job: ${{ parameters.name }}
+  displayName: Check rustfmt
+  pool:
+    vmImage: ubuntu-16.04
+  steps:
+    - template: azure-install-rust.yml
+      parameters:
+        rust_version: stable
+    - script: |
+        rustup component add rustfmt
+      displayName: Install rustfmt
+    - script: |
+        cargo fmt --all -- --check
+      displayName: Check formatting

ci/azure-test-stable.yml

@@ -0,0 +1,41 @@
+jobs:
+- job: ${{ parameters.name }}
+  displayName: ${{ parameters.displayName }}
+  strategy:
+    matrix:
+      Linux:
+        vmImage: ubuntu-16.04
+
+      ${{ if parameters.cross }}:
+        MacOS:
+          vmImage: macOS-10.13
+        Windows:
+          vmImage: vs2017-win2016
+  pool:
+    vmImage: $(vmImage)
+
+  steps:
+  - template: azure-install-rust.yml
+    parameters:
+      rust_version: stable
+
+  - template: azure-is-release.yml
+
+  - ${{ each crate in parameters.crates }}:
+    - script: cargo test
+      env:
+        LOOM_MAX_DURATION: 10
+        CI: 'True'
+      displayName: cargo test -p ${{ crate }}
+      workingDirectory: $(Build.SourcesDirectory)/${{ crate }}
+      condition: and(succeeded(), ne(variables['isRelease'], 'true'))
+
+  - template: azure-patch-crates.yml
+
+  - ${{ each crate in parameters.crates }}:
+    - script: cargo test
+      env:
+        LOOM_MAX_DURATION: 10
+        CI: 'True'
+      displayName: cargo test -p ${{ crate }} (PATCHED)
+      workingDirectory: $(Build.SourcesDirectory)/${{ crate }}

ci/azure-tsan.yml

@@ -0,0 +1,36 @@
+jobs:
+- job: ${{ parameters.name }}
+  displayName: TSAN
+  strategy:
+    matrix:
+      Timer:
+        cmd: cargo test -p tokio-timer --test hammer
+      Threadpool:
+        cmd: cargo test -p tokio-threadpool --tests
+  pool:
+    vmImage: ubuntu-16.04
+  steps:
+  - template: azure-install-rust.yml
+    parameters:
+      rust_version: nightly-2019-07-17
+
+  - template: azure-patch-crates.yml
+  - script: |
+        set -e
+
+        # Make sure the benchmarks compile
+        export ASAN_OPTIONS="detect_odr_violation=0 detect_leaks=0"
+        export TSAN_OPTIONS="suppressions=`pwd`/ci/tsan"
+        export RUST_BACKTRACE=1
+
+        # Run address sanitizer
+        RUSTFLAGS="-Z sanitizer=address" \
+        $(cmd) --target x86_64-unknown-linux-gnu
+
+        # Run thread sanitizer
+        RUSTFLAGS="-Z sanitizer=thread" \
+        $(cmd) --target x86_64-unknown-linux-gnu
+    displayName: TSAN / MSAN
+    env:
+      TSAN: yes
+

ci/patch.toml

@@ -0,0 +1,20 @@
+# Patch dependencies to run all tests against versions of the crate in the
+# repository.
+[patch.crates-io]
+tokio = { path = "tokio" }
+tokio-buf = { path = "tokio-buf" }
+tokio-codec = { path = "tokio-codec" }
+tokio-current-thread = { path = "tokio-current-thread" }
+tokio-executor = { path = "tokio-executor" }
+tokio-fs = { path = "tokio-fs" }
+tokio-futures = { path = "tokio-futures" }
+tokio-io = { path = "tokio-io" }
+tokio-reactor = { path = "tokio-reactor" }
+tokio-signal = { path = "tokio-signal" }
+tokio-sync = { path = "tokio-sync" }
+tokio-threadpool = { path = "tokio-threadpool" }
+tokio-timer = { path = "tokio-timer" }
+tokio-tcp = { path = "tokio-tcp" }
+tokio-tls = { path = "tokio-tls" }
+tokio-udp = { path = "tokio-udp" }
+tokio-uds = { path = "tokio-uds" }

ci/tsan

@@ -0,0 +1,44 @@
+# TSAN suppressions file for Tokio
+
+# TSAN does not understand fences and `Arc::drop` is implemented using a fence.
+# This causes many false positives.
+race:Arc*drop
+race:Weak*drop
+
+# `std` mpsc is not used in any Tokio code base. This race is triggered by some
+# rust runtime logic.
+race:std*mpsc_queue
+
+# Probably more fences in std.
+race:__call_tls_dtors
+
+# The epoch-based GC uses fences.
+race:crossbeam_epoch
+
+# Push and steal operations in crossbeam-deque may cause data races, but such
+# data races are safe. If a data race happens, the value read by `steal` is
+# forgotten and the steal operation is then retried.
+race:crossbeam_deque*push
+race:crossbeam_deque*steal
+
+# This filters out expected data race in the Treiber stack implementations.
+# Treiber stacks are inherently racy. The pop operation will attempt to access
+# the "next" pointer on the node it is attempting to pop. However, at this
+# point it has not gained ownership of the node and another thread might beat
+# it and take ownership of the node first (touching the next pointer). The
+# original pop operation will fail due to the ABA guard, but tsan still picks
+# up the access on the next pointer.
+race:Backup::next_sleeper
+race:Backup::set_next_sleeper
+race:WorkerEntry::set_next_sleeper
+
+# This ignores a false positive caused by `thread::park()`/`thread::unpark()`.
+# See: https://github.com/rust-lang/rust/pull/54806#issuecomment-436193353
+race:pthread_cond_destroy
+
+# Recent rand dependency updates and seeding changes have introduced
+# lazy_static's and other racy code. See:
+# https://github.com/tokio-rs/tokio/pull/1358#issuecomment-516172383
+race:RandomState*::build_hasher
+race:lazy_static::
+race:c2_chacha::guts

deny.toml

@@ -1,21 +0,0 @@
-# https://embarkstudios.github.io/cargo-deny/cli/init.html
-
-[graph]
-all-features = true
-
-[licenses]
-allow = [
-    "MIT",
-    "Apache-2.0",
-]
-exceptions = [
-    { allow = ["Unicode-3.0"], crate = "unicode-ident" },
-]
-
-[bans]
-multiple-versions = "allow"
-wildcards = "deny"
-
-[sources]
-unknown-registry = "deny"
-unknown-git = "deny"

docs/contributing/README.md

@@ -1,48 +0,0 @@
-# Contributing
-
-This guide will help you get started. **Do not let this guide intimidate you**.
-It should be considered a map to help you navigate the process.
-
-## Quick start
-
-If you are unsure where to begin, use the following guides:
-
-- Want to report or triage a bug? Start with [Contributing in Issues](contributing-in-issues.md).
-- Looking for something to work on? Check the open [`issues`](https://github.com/tokio-rs/tokio/issues)
-(some might be in progress or still under discussion, so leave a comment before starting the work).
-- Planning to submit a PR? Read [Pull Requests](pull-requests.md) for the full workflow and required checks.
-- Want to understand what the labels on issues mean? See [Keeping track of issues and PRs](keeping-track-of-issues-and-prs.md).
-- Interested in code review? See [Reviewing Pull Requests](reviewing-pull-requests.md).
-
-## Table of Contents
-
-- [Contributing in Issues](contributing-in-issues.md)
-    - [Asking for General Help](contributing-in-issues.md#asking-for-general-help)
-    - [Submitting a Bug Report](contributing-in-issues.md#submitting-a-bug-report)
-    - [Triaging a Bug Report](contributing-in-issues.md#triaging-a-bug-report)
-    - [Resolving a Bug Report](contributing-in-issues.md#resolving-a-bug-report)
-- [Pull Requests](pull-requests.md)
-    - [Step-by-Step Contribution Workflow](pull-requests.md#step-by-step-contribution-workflow)
-    - [Cargo Commands](pull-requests.md#cargo-commands)
-    - [Performing spellcheck on Tokio codebase](pull-requests.md#performing-spellcheck-on-tokio-codebase)
-    - [Tests](pull-requests.md#tests)
-        - [Integration tests](pull-requests.md#integration-tests)
-        - [Fuzz tests](pull-requests.md#fuzz-tests)
-        - [Documentation tests](pull-requests.md#documentation-tests)
-    - [Benchmarks](pull-requests.md#benchmarks)
-    - [Commits](pull-requests.md#commits)
-        - [Commit message guidelines](pull-requests.md#commit-message-guidelines)
-    - [Opening the Pull Request](pull-requests.md#opening-the-pull-request)
-    - [Discuss and update](pull-requests.md#discuss-and-update)
-    - [Commit Squashing](pull-requests.md#commit-squashing)
-- [Reviewing Pull Requests](reviewing-pull-requests.md)
-    - [Review a bit at a time](reviewing-pull-requests.md#review-a-bit-at-a-time)
-    - [Be aware of the person behind the code](reviewing-pull-requests.md#be-aware-of-the-person-behind-the-code)
-    - [Abandoned or Stalled Pull Requests](reviewing-pull-requests.md#abandoned-or-stalled-pull-requests)
-- [How to specify crates dependencies versions](how-to-specify-crates-dependencies-versions.md)
-- [Keeping track of issues and PRs](keeping-track-of-issues-and-prs.md)
-    - [Area](keeping-track-of-issues-and-prs.md#area)
-    - [Category](keeping-track-of-issues-and-prs.md#category)
-    - [Calls for participation](keeping-track-of-issues-and-prs.md#calls-for-participation)
-    - [Module](keeping-track-of-issues-and-prs.md#module)
-    - [Topic](keeping-track-of-issues-and-prs.md#topic)

docs/contributing/contributing-in-issues.md

@@ -1,79 +0,0 @@
-## Contributing in Issues
-
-For any issue, there are fundamentally three ways an individual can contribute:
-
-1. By opening the issue for discussion: For instance, if you believe that you
-   have discovered a bug in Tokio, creating a new issue in [the tokio-rs/tokio
-   issue tracker][issue] is the way to report it.
-
-2. By helping to triage the issue: This can be done by providing
-   supporting details (a test case that demonstrates a bug), providing
-   suggestions on how to address the issue, or ensuring that the issue is tagged
-   correctly.
-
-3. By helping to resolve the issue: Typically this is done either in the form of
-   demonstrating that the issue reported is not a problem after all, or more
-   often, by opening a Pull Request that changes some bit of something in
-   Tokio in a concrete and reviewable manner.
-
-[issue]: https://github.com/tokio-rs/tokio/issues
-
-**Anybody can participate in any stage of contribution**. We urge you to
-participate in the discussion around bugs and participate in reviewing PRs.
-
-### Asking for General Help
-
-If you have reviewed existing documentation and still have questions or are
-having problems, you can [open a discussion] asking for help.
-
-In exchange for receiving help, we ask that you contribute back a documentation
-PR that helps others avoid the problems that you encountered.
-
-[open a discussion]: https://github.com/tokio-rs/tokio/discussions/new/choose
-
-### Submitting a Bug Report
-
-When opening a new issue in the Tokio issue tracker, you will be presented
-with a basic template that should be filled in. If you believe that you have
-uncovered a bug, please fill out this form, following the template to the best
-of your ability. Do not worry if you cannot answer every detail, just fill in
-what you can.
-
-The two most important pieces of information we need in order to properly
-evaluate the report is a description of the behavior you are seeing and a simple
-test case we can use to recreate the problem on our own. If we cannot recreate
-the issue, it becomes impossible for us to fix.
-
-In order to rule out the possibility of bugs introduced by userland code, test
-cases should be limited, as much as possible, to using only Tokio APIs.
-
-See [How to create a Minimal, Complete, and Verifiable example][mcve].
-
-[mcve]: https://stackoverflow.com/help/mcve
-
-### Triaging a Bug Report
-
-Once an issue has been opened, it is not uncommon for there to be discussion
-around it. Some contributors may have differing opinions about the issue,
-including whether the behavior being seen is a bug or a feature. This discussion
-is part of the process and should be kept focused, helpful, and professional.
-
-Short, clipped responses—that provide neither additional context nor supporting
-detail—are not helpful or professional. To many, such responses are simply
-annoying and unfriendly.
-
-Contributors are encouraged to help one another make forward progress as much as
-possible, empowering one another to solve issues collaboratively. If you choose
-to comment on an issue that you feel either is not a problem that needs to be
-fixed, or if you encounter information in an issue that you feel is incorrect,
-explain why you feel that way with additional supporting context, and be willing
-to be convinced that you may be wrong. By doing so, we can often reach the
-correct outcome much faster.
-
-### Resolving a Bug Report
-
-In the majority of cases, issues are resolved by opening a Pull Request. The
-process for opening and reviewing a Pull Request is similar to that of opening
-and triaging issues, but carries with it a necessary review and approval
-workflow that ensures that the proposed changes meet the minimal quality and
-functional guidelines of the Tokio project.

docs/contributing/how-to-specify-crates-dependencies-versions.md

@@ -1,32 +0,0 @@
-# How to specify crates dependencies versions
-
-Each crate (e.g., `tokio-util`, `tokio-stream`, etc.) should specify dependencies
-according to the following rules:
-
-1. The listed version should be the oldest version that the crate works with
-(e.g., if `tokio-util` works with `tokio` version `1.44` but not `1.43`, then
-`tokio-util` should specify version `1.44` for its `tokio` dependency).
-We don't require users to use the latest version unnecessarily.
-2. When a crate starts using a newer feature in a dependency, the version
-should be bumped to the version that introduced it.
-3. If a crate depends on an unreleased feature in a dependency, it may use
-`path =` dependency to specify this. Since path dependencies must be removed
-during the release of the crate, this ensures that it can't be released until
-the dependency has a new version.
-
-Consider the following example from `tokio-stream`:
-
-```toml
-[dependencies]
-futures-core = { version = "0.3.0" }
-pin-project-lite = "0.2.11"
-tokio = { version = "1.38.0", path = "../tokio", features = ["sync"] }
-```
-
-In this case, local development of `tokio-stream` uses the local version
-of `tokio` via the `path` dependency. This means that it's currently not
-possible to release `tokio-stream`, and `tokio` should be released first.
-Once a new version of `tokio` is released (in this example the `1.38.0`),
-the path dependency should be removed.
-As mentioned before, this version should only be bumped when adding a new
-feature in the crate that relies on a newer version.

docs/contributing/keeping-track-of-issues-and-prs.md

@@ -1,91 +0,0 @@
-## Keeping track of issues and PRs
-
-The Tokio GitHub repository has a lot of issues and PRs to keep track of. This
-section explains the meaning of various labels, as well as our [GitHub
-project][project]. The section is primarily targeted at maintainers.  Most
-contributors aren't able to set these labels.
-
-### Area
-
-The area label describes the crates relevant to this issue or PR.
-
-- **A-ci** This issue concerns our GitHub Actions setup.
-- **A-tokio** This issue concerns the main Tokio crate.
-- **A-readme** This issue is related to documentation such as README.md.
-- **A-benches** This issue concerns the benchmarks.
-- **A-examples** This issue concerns the examples.
-- **A-tokio-test** The issue concerns the `tokio-test` crate.
-- **A-tokio-util** This issue concerns the `tokio-util` crate.
-- **A-tokio-macros** This issue concerns the `tokio-macros` crate. Should only
-  be used for the procedural macros, and not `join!` or `select!`.
-- **A-tokio-stream** This issue concerns the `tokio-stream` crate.
-
-### Category
-
-- **C-bug** This is a bug-report. Bug-fix PRs use `C-enhancement` instead.
-- **C-enhancement** This is a PR that adds a new features.
-- **C-maintenance** This is an issue or PR about stuff such as documentation,
-  GitHub Actions or code quality.
-- **C-feature-request** This is a feature request. Implementations of feature
-  requests use `C-enhancement` instead.
-- **C-feature-accepted** If you submit a PR for this feature request, we won't
-  close it with the reason "we don't want this". Issues with this label should
-  also have the `C-feature-request` label.
-- **C-musing** Stuff like tracking issues or roadmaps. "musings about a better
-  world"
-- **C-proposal** A proposal of some kind, and a request for comments.
-- **C-question** A user question. Large overlap with GitHub discussions.
-- **C-request** A non-feature request, e.g. "please add deprecation notices to
-  `-alpha.*` versions of crates"
-
-### Calls for participation
-
-- **E-help-wanted** Stuff where we want help. Often seen together with `C-bug`
-  or `C-feature-accepted`.
-- **E-easy** This is easy, ranging from quick documentation fixes to stuff you
-  can do after reading the tutorial on our website.
-- **E-medium** This is not `E-easy` or `E-hard`.
-- **E-hard** This either involves very tricky code, is something we don't know
-  how to solve, or is challenging for some other reason.
-- **E-needs-mvce** This bug is missing a minimal complete and verifiable
-  example.
-
-The "E-" prefix is the same as used in the Rust compiler repository. Some
-issues are missing a difficulty rating, but feel free to ask on our Discord
-server if you want to know how challenging an issue likely is.
-
-### Module
-
-The module label provides a more fine grained categorization than **Area**.
-
-- **M-blocking** Things relevant to `spawn_blocking`, `block_in_place`.
-- **M-codec** The `tokio_util::codec` module.
-- **M-compat** The `tokio_util::compat` module.
-- **M-coop** Things relevant to coop.
-- **M-fs** The `tokio::fs` module.
-- **M-io** The `tokio::io` module.
-- **M-macros** Issues about any kind of macro.
-- **M-metrics** Things relevant to `tokio::runtime::metrics`.
-- **M-net** The `tokio::net` module.
-- **M-process** The `tokio::process` module.
-- **M-runtime** The `tokio::runtime` module.
-- **M-signal** The `tokio::signal` module.
-- **M-sync** The `tokio::sync` module.
-- **M-task** The `tokio::task` module.
-- **M-time** The `tokio::time` module.
-- **M-tracing** Tracing support in Tokio.
-- **M-taskdump** Things relevant to taskdump.
-
-### Topic
-
-Some extra information.
-
-- **T-docs** This is about documentation.
-- **T-io-uring** This is about io-uring (Linux).
-- **T-performance** This is about performance.
-- **T-v0.1.x** This is about old Tokio.
-- **T-wasm** This is about Web Assembly.
-
-Any label not listed here is not in active use.
-
-[project]: https://github.com/orgs/tokio-rs/projects/1

docs/contributing/pull-requests.md

@@ -1,409 +0,0 @@
-## Pull Requests
-
-Pull Requests are the way concrete changes are made to the code, documentation,
-and dependencies in the Tokio repository.
-
-Even tiny pull requests (e.g., one-character pull request fixing a typo in API
-documentation) are greatly appreciated.
-
-> [!NOTE]
-> Before making a large change, it is usually a good idea to first open an
-> issue describing the change to solicit feedback and guidance.
-> This will increase the likelihood of the PR getting merged.
-
-### Step-by-Step Contribution Workflow
-
-Once you've found an issue you'd like to work on, follow these steps before opening a Pull Request. This workflow provides a chronological overview of the contribution process and points to the relevant documentation where additional detail is available.
-
-#### 1. Fork and Clone the Repository
-
-Fork the repository to your GitHub account and clone your fork locally.
-
-If you've previously cloned the repository, ensure your local copy is up to date before creating a new branch.
-
-```bash
-git checkout master
-git pull upstream master
-```
-
----
-
-#### 2. Create a Feature Branch
-
-Never commit directly to your local `master` branch. Instead, create a descriptive feature branch for each change you work on.
-
-```bash
-git checkout -b my-feature
-```
-
-Using a dedicated branch keeps your default branch clean and makes it easier to update your Pull Request during code review.
-
----
-
-#### 3. Implement Your Changes
-
-Make the code or documentation changes needed. Check your work with `git diff` to confirm the changes look correct before moving on.
-
-If your changes introduce new functionality or modify existing behavior, consider whether additional tests or documentation should also be added.
-
----
-
-#### 4. Verify Your Changes
-
-Before opening a Pull Request, run the project's verification steps. Refer to the sections below for details on when each command should be used.
-
----
-
-#### 5. Commit and Push Your Changes
-
-Commit your changes following the [project's commit message guidelines](#commit-message-guidelines).
-
-```bash
-git add .
-git commit -m "module: describe your change"
-git push origin my-feature
-```
-
----
-
-#### 6. Open a Pull Request
-
-Open a Pull Request from your feature branch to the main repository. See [Opening the Pull Request](#opening-the-pull-request) for what to include and how the review process works.
-
-### Cargo Commands
-
-Due to the extensive use of features in Tokio, you will often need to add extra
-arguments to many common cargo commands. This section lists some commonly needed
-commands.
-
-Some commands just need the `--all-features` argument:
-
-```
-cargo build --all-features
-cargo check --all-features
-cargo test --all-features
-```
-
-> [!NOTE]
-> There are some features that are not available in some systems (e.g., `io-uring`
-> which is Linux only). In that case you cannot use the `--all-features` and you
-> should specify only the features that are supported (e.g., `--features=full`).
-
-Ideally, you should use the same version of clippy as the one used in CI
-(defined by `env.rust_clippy` in [ci.yml][ci.yml]), because newer versions
-might have new lints:
-
-[ci.yml]: ../../.github/workflows/ci.yml
-
-<!--
-When updating this, also update:
-- .github/workflows/ci.yml
-- README.md
-- tokio/README.md
-- tokio/Cargo.toml
-- tokio-util/Cargo.toml
-- tokio-test/Cargo.toml
-- tokio-stream/Cargo.toml
--->
-
-```
-cargo +1.88 clippy --all --tests --all-features
-```
-
-When building documentation, a simple `cargo doc` is not sufficient. To produce
-documentation equivalent to what will be produced in docs.rs's builds of Tokio's
-docs, please use:
-
-```
-RUSTDOCFLAGS="--cfg docsrs --cfg tokio_unstable" RUSTFLAGS="--cfg docsrs --cfg tokio_unstable" cargo +nightly doc --all-features [--open]
-```
-
-This turns on indicators to display the Cargo features required for
-conditionally compiled APIs in Tokio, and it enables documentation of unstable
-Tokio features. Notice that it is necessary to pass cfg flags to both RustDoc
-*and* rustc.
-
-There is a more concise way to build docs.rs-equivalent docs by using [`cargo
-docs-rs`], which reads the above documentation flags out of Tokio's Cargo.toml
-as docs.rs itself does.
-
-[`cargo docs-rs`]: https://github.com/dtolnay/cargo-docs-rs
-
-```
-cargo install --locked cargo-docs-rs
-cargo +nightly docs-rs [--open]
-```
-
-The `cargo fmt` command does not work on the Tokio codebase. You can use the
-command below instead:
-
-```
-# Mac or Linux
-rustfmt --check --edition 2021 $(git ls-files '*.rs')
-
-# Powershell
-Get-ChildItem . -Filter "*.rs" -Recurse | foreach { rustfmt --check --edition 2021 $_.FullName }
-```
-The `--check` argument prints the things that need to be fixed. If you remove
-it, `rustfmt` will update your files locally instead.
-
-You can run loom tests with
-```
-cd tokio # tokio crate in workspace
-LOOM_MAX_PREEMPTIONS=1 LOOM_MAX_BRANCHES=10000 RUSTFLAGS="--cfg loom -C debug_assertions" \
-    cargo test --lib --release --features full -- --test-threads=1 --nocapture
-```
-Additionally, you can also add `--cfg tokio_unstable` to the `RUSTFLAGS` environment variable to
-run loom tests that test unstable features.
-
-You can run miri tests with
-```
-MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-strict-provenance" \
-    cargo +nightly miri test --features full --lib --tests
-```
-
-### Performing spellcheck on Tokio codebase
-
-You can perform a spell-check on the Tokio codebase. For details of how to use the spellcheck tool, feel free to visit
-https://github.com/drahnr/cargo-spellcheck
-```
-# First install the spell-check plugin
-cargo install --locked cargo-spellcheck
-
-# Then run the cargo spell check command
-cargo spellcheck check
-```
-
-If the command rejects a word, you should backtick the rejected word if it's code related. If not, the
-rejected word should be put into `spellcheck.dic` file.
-
-Note that when you add a word into the file, you should also update the first line which tells the spellcheck tool
-the total number of words included in the file
-
-### Tests
-
-If the change being proposed alters code (as opposed to only documentation for
-example), it is either adding new functionality to Tokio or it is fixing
-existing, broken functionality. In both of these cases, the pull request should
-include one or more tests to ensure that Tokio does not regress in the future.
-There are two ways to write tests: [integration tests][integration-tests]
-and [documentation tests][documentation-tests].
-(Tokio avoids [unit tests][unit-tests] as much as possible).
-
-Tokio uses [conditional compilation attributes][conditional-compilation]
-throughout the codebase, to modify rustc's behavior. Code marked with such
-attributes can be enabled using `RUSTFLAGS` and `RUSTDOCFLAGS` environment
-variables. One of the most prevalent flags passed in these variables is
-the `--cfg` option. To run tests in a particular file, check first what
-options #![cfg] declaration defines for that file.
-
-For instance, to run a test marked with the 'tokio_unstable' cfg option,
-you must pass this flag to the compiler when running the test.
-```
-$ RUSTFLAGS="--cfg tokio_unstable" cargo test -p tokio --all-features --test rt_metrics
-```
-
-#### Integration tests
-
-Integration tests go in the same crate as the code they are testing. Each sub
-crate should have a `dev-dependency` on `tokio` itself. This makes all Tokio
-utilities available to use in tests, no matter the crate being tested.
-
-The best strategy for writing a new integration test is to look at existing
-integration tests in the crate and follow the style.
-
-#### Fuzz tests
-
-Some of our crates include a set of fuzz tests, this will be marked by a
-directory `fuzz`. It is a good idea to run fuzz tests after each change.
-To get started with fuzz testing you'll need to install
-[cargo-fuzz](https://github.com/rust-fuzz/cargo-fuzz).
-
-`cargo install --locked cargo-fuzz`
-
-To list the available fuzzing harnesses you can run;
-
-```bash
-$ cd tokio
-$ cargo fuzz list
-fuzz_linked_list
-```
-
-Running a fuzz test is as simple as;
-
-`cargo fuzz run fuzz_linked_list`
-
-**NOTE**: Keep in mind that by default when running a fuzz test the fuzz
-harness will run forever and will only exit if you `ctrl-c` or it finds
-a bug.
-
-#### Documentation tests
-
-Ideally, every API has at least one [documentation test] that demonstrates how to
-use the API. Documentation tests are run with `cargo test --doc`. This ensures
-that the example is correct and provides additional test coverage.
-
-The trick to documentation tests is striking a balance between being succinct
-for a reader to understand and actually testing the API.
-
-Same as with integration tests, when writing a documentation test, the full
-`tokio` crate is available. This is especially useful for getting access to the
-runtime to run the example.
-
-The documentation tests will be visible from both the crate-specific
-documentation **and** the `tokio` facade documentation via the re-export. The
-example should be written from the point of view of a user that is using the
-`tokio` crate. As such, the example should use the API via the facade and not by
-directly referencing the crate.
-
-The type level example for `tokio::time::timeout` provides a good example of a
-documentation test:
-
-```
-/// Create a new `Timeout` set to expire in 10 milliseconds.
-///
-/// ```rust
-/// use tokio::time::timeout;
-/// use tokio::sync::oneshot;
-///
-/// use std::time::Duration;
-///
-/// # async fn dox() {
-/// let (tx, rx) = oneshot::channel();
-/// # tx.send(()).unwrap();
-///
-/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds.
-/// if let Err(_) = timeout(Duration::from_millis(10), rx).await {
-///     println!("did not receive value within 10 ms");
-/// }
-/// # }
-/// ```
-```
-
-Lines that start with `/// #` are removed when the documentation is generated.
-
-### Benchmarks
-
-You can run benchmarks locally for the changes you've made to the tokio codebase.
-Tokio currently uses [Criterion](https://github.com/bheisler/criterion.rs) as its benchmarking tool. To run a benchmark
-against the changes you have made, for example, you can run;
-
-```bash
-cd benches
-
-# Run all benchmarks.
-cargo bench
-
-# Run all tests in the `benches/fs.rs` file
-cargo bench --bench fs
-
-# Run the `async_read_buf` benchmark in `benches/fs.rs` specifically.
-cargo bench async_read_buf
-
-# After running benches, you can check the statistics under `tokio/target/criterion/`
-```
-
-You can also refer to [Criterion] docs for additional options and details.
-
-[Criterion]: https://docs.rs/criterion/latest/criterion/
-
-### Commits
-
-It is a recommended best practice to keep your changes as logically grouped as
-possible within individual commits. There is no limit to the number of commits
-any single Pull Request may have, and many contributors find it easier to review
-changes that are split across multiple commits.
-
-That said, if you have a number of commits that are "checkpoints" and don't
-represent a single logical change, please squash those together.
-
-Note that multiple commits often get squashed when they are landed (see the
-notes about [commit squashing](#commit-squashing)).
-
-#### Commit message guidelines
-
-A good commit message should describe what changed and why.
-
-1. The first line should:
-
-* contain a short description of the change (preferably 50 characters or less,
-  and no more than 72 characters)
-* be entirely in lowercase with the exception of proper nouns, acronyms, and
-  the words that refer to code, like function/variable names
-* start with an imperative verb
-* not have a period at the end
-* be prefixed with the name of the module being changed; usually this is the
-  same as the M-* label on the PR
-
-Examples:
-
-* time: introduce `Timeout` and deprecate `Deadline`
-* codec: export `Encoder`, `Decoder`, `Framed*`
-* ci: fix the FreeBSD ci configuration
-
-2. Keep the second line blank.
-3. Wrap all other lines at 72 columns (except for long URLs).
-4. If your patch fixes an open issue, you can add a reference to it at the end
-   of the log. Use the `Fixes: #` prefix and the issue number. For other
-   references use `Refs: #`. `Refs` may include multiple issues, separated by a
-   comma.
-
-   Examples:
-
-    - `Fixes: #1337`
-    - `Refs: #1234`
-
-Sample complete commit message:
-
-```txt
-module: explain the commit in one line
-
-Body of commit message is a few lines of text, explaining things
-in more detail, possibly giving some background about the issue
-being fixed, etc.
-
-The body of the commit message can be several paragraphs, and
-please do proper word-wrap and keep columns shorter than about
-72 characters or so. That way, `git log` will show things
-nicely even when it is indented.
-
-Fixes: #1337
-Refs: #453, #154
-```
-
-### Opening the Pull Request
-
-From within GitHub, opening a new Pull Request will present you with a
-[template] that should be filled out. Please try to do your best at filling out
-the details, but feel free to skip parts if you're not sure what to put.
-
-[template]: ../../.github/PULL_REQUEST_TEMPLATE.md
-
-### Discuss and update
-
-You will probably get feedback or requests for changes to your Pull Request.
-This is a big part of the submission process so don't be discouraged! Some
-contributors may sign off on the Pull Request right away, others may have
-more detailed comments or feedback. This is a necessary part of the process
-in order to evaluate whether the changes are correct and necessary.
-
-**Any community member can review a PR and you might get conflicting feedback**.
-Keep an eye out for comments from code owners to provide guidance on conflicting
-feedback.
-
-**Once the PR is open, do not rebase the commits**. See [Commit Squashing](#commit-squashing) for
-more details.
-
-### Commit Squashing
-
-In most cases, **do not squash commits that you add to your Pull Request during
-the review process**. When the commits in your Pull Request land, they may be
-squashed into one commit per logical change. Metadata will be added to the
-commit message (including links to the Pull Request, links to relevant issues,
-and the names of the reviewers). The commit history of your Pull Request,
-however, will stay intact on the Pull Request page.
-
-[integration-tests]: https://doc.rust-lang.org/rust-by-example/testing/integration_testing.html
-[unit-tests]: https://doc.rust-lang.org/rust-by-example/testing/unit_testing.html
-[documentation-tests]: https://doc.rust-lang.org/rust-by-example/testing/doc_testing.html
-[conditional-compilation]: https://doc.rust-lang.org/reference/conditional-compilation.html

docs/contributing/reviewing-pull-requests.md

@@ -1,80 +0,0 @@
-## Reviewing Pull Requests
-
-**Any Tokio community member is welcome to review any pull request**.
-
-All Tokio contributors who choose to review and provide feedback on Pull
-Requests have a responsibility to both the project and the individual making the
-contribution. Reviews and feedback must be helpful, insightful, and geared
-towards improving the contribution as opposed to simply blocking it. If there
-are reasons why you feel the PR should not land, explain what those are. Do not
-expect to be able to block a Pull Request from advancing simply because you say
-"No" without giving an explanation. Be open to having your mind changed. Be open
-to working with the contributor to make the Pull Request better.
-
-Reviews that are dismissive or disrespectful of the contributor or any other
-reviewers are strictly counter to the Code of Conduct.
-
-When reviewing a Pull Request, the primary goals are for the codebase to improve
-and for the person submitting the request to succeed. **Even if a Pull Request
-does not land, the submitters should come away from the experience feeling like
-their effort was not wasted or unappreciated**. Every Pull Request from a new
-contributor is an opportunity to grow the community.
-
-### Review a bit at a time
-
-Do not overwhelm new contributors.
-
-It is tempting to micro-optimize and make everything about relative performance,
-perfect grammar, or exact style matches. Do not succumb to that temptation.
-
-Focus first on the most significant aspects of the change:
-
-1. Does this change make sense for Tokio?
-2. Does this change make Tokio better, even if only incrementally?
-3. Are there clear bugs or larger scale issues that need attending to?
-4. Is the commit message readable and correct? If it contains a breaking change
-   is it clear enough?
-
-Note that only **incremental** improvement is needed to land a PR. This means
-that the PR does not need to be perfect, only better than the status quo. Follow
-up PRs may be opened to continue iterating.
-
-When changes are necessary, *request* them, do not *demand* them, and **do not
-assume that the submitter already knows how to add a test or run a benchmark**.
-
-Specific performance optimization techniques, coding styles and conventions
-change over time. The first impression you give to a new contributor never does.
-
-Nits (requests for small changes that are not essential) are fine, but try to
-avoid stalling the Pull Request. Most nits can typically be fixed by the Tokio
-Collaborator landing the Pull Request but they can also be an opportunity for
-the contributor to learn a bit more about the project.
-
-It is always good to clearly indicate nits when you comment: e.g.
-`Nit: change foo() to bar(). But this is not blocking.`
-
-If your comments were addressed but were not folded automatically after new
-commits or if they proved to be mistaken, please, [hide them][hiding-a-comment]
-with the appropriate reason to keep the conversation flow concise and relevant.
-
-### Be aware of the person behind the code
-
-Be aware that *how* you communicate requests and reviews in your feedback can
-have a significant impact on the success of the Pull Request. Yes, we may land
-a particular change that makes Tokio better, but the individual might just not
-want to have anything to do with Tokio ever again. The goal is not just having
-good code.
-
-### Abandoned or Stalled Pull Requests
-
-If a Pull Request appears to be abandoned or stalled, it is polite to first
-check with the contributor to see if they intend to continue the work before
-checking if they would mind if you took it over (especially if it just has nits
-left). When doing so, it is courteous to give the original contributor credit
-for the work they started (either by preserving their name and email address in
-the commit log, or by using an `Author: ` meta-data tag in the commit.
-
-_Adapted from the [Node.js contributing guide][node]_.
-
-[node]: https://github.com/nodejs/node/blob/master/CONTRIBUTING.md
-[hiding-a-comment]: https://help.github.com/articles/managing-disruptive-comments/#hiding-a-comment

examples/Cargo.toml

@@ -1,117 +0,0 @@
-[package]
-name = "examples"
-version = "0.0.0"
-publish = false
-edition = "2021"
-license = "MIT"
-
-# If you copy one of the examples into a new project, you should be using
-# [dependencies] instead, and delete the **path**.
-[dev-dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["full", "tracing"] }
-tokio-util = { version = "0.7.0", path = "../tokio-util", features = ["full"] }
-tokio-stream = { version = "0.1", path = "../tokio-stream" }
-
-tracing = "0.1"
-tracing-subscriber = { version = "0.3.1", default-features = false, features = ["fmt", "ansi", "env-filter", "tracing-log"] }
-bytes = "1.0.0"
-futures = { version = "0.3.0", features = ["thread-pool"]}
-http = "1"
-serde = "1.0"
-serde_derive = "1.0"
-serde_json = "1.0"
-httparse = "1.0"
-httpdate = "1.0"
-once_cell = "1.5.2"
-
-[target.'cfg(target_os = "linux")'.dev-dependencies]
-libc = "0.2"
-
-[target.'cfg(all(tokio_unstable, target_os = "linux"))'.dev-dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["full", "tracing", "taskdump"] }
-
-[target.'cfg(windows)'.dev-dependencies.windows-sys]
-version = "0.61"
-
-[[example]]
-name = "chat"
-path = "chat.rs"
-
-[[example]]
-name = "connect-tcp"
-path = "connect-tcp.rs"
-
-[[example]]
-name = "connect-udp"
-path = "connect-udp.rs"
-
-[[example]]
-name = "echo-tcp"
-path = "echo-tcp.rs"
-
-[[example]]
-name = "graceful-shutdown"
-path = "graceful-shutdown.rs"
-
-[[example]]
-name = "echo-udp"
-path = "echo-udp.rs"
-
-[[example]]
-name = "hello_world"
-path = "hello_world.rs"
-
-[[example]]
-name = "print_each_packet"
-path = "print_each_packet.rs"
-
-[[example]]
-name = "proxy"
-path = "proxy.rs"
-
-[[example]]
-name = "tinydb"
-path = "tinydb.rs"
-
-[[example]]
-name = "udp-client"
-path = "udp-client.rs"
-
-[[example]]
-name = "udp-codec"
-path = "udp-codec.rs"
-
-[[example]]
-name = "tinyhttp"
-path = "tinyhttp.rs"
-
-[[example]]
-name = "custom-executor"
-path = "custom-executor.rs"
-
-[[example]]
-name = "custom-executor-tokio-context"
-path = "custom-executor-tokio-context.rs"
-
-[[example]]
-name = "named-pipe"
-path = "named-pipe.rs"
-
-[[example]]
-name = "named-pipe-ready"
-path = "named-pipe-ready.rs"
-
-[[example]]
-name = "named-pipe-multi-client"
-path = "named-pipe-multi-client.rs"
-
-[[example]]
-name = "dump"
-path = "dump.rs"
-
-[[example]]
-name = "prewarm-fd-table"
-path = "prewarm-fd-table.rs"
-
-[lints]
-workspace = true

examples/README.md

@@ -1,23 +0,0 @@
-## Examples of how to use Tokio
-
-This directory contains a number of examples showcasing various capabilities of
-the `tokio` crate.
-
-All examples can be executed with:
-
-```
-cargo run --example $name
-```
-
-A good starting point for the examples would be [`hello_world`](hello_world.rs)
-and [`echo-tcp`](echo-tcp.rs). Additionally [the tokio website][tokioweb] contains
-additional guides for some of the examples.
-
-For a larger "real world" example, see the [`mini-redis`][redis] repository.
-
-If you've got an example you'd like to see here, please feel free to open an
-issue. Otherwise if you've got an example you'd like to add, please feel free
-to make a PR!
-
-[tokioweb]: https://tokio.rs/tokio/tutorial
-[redis]: https://github.com/tokio-rs/mini-redis

examples/chat.rs

@@ -1,256 +0,0 @@
-//! A chat server that broadcasts a message to all connections.
-//!
-//! This example is explicitly more verbose than it has to be. This is to
-//! illustrate more concepts.
-//!
-//! A chat server for telnet clients. After a telnet client connects, the first
-//! line should contain the client's name. After that, all lines sent by a
-//! client are broadcasted to all other connected clients.
-//!
-//! Because the client is telnet, lines are delimited by "\r\n".
-//!
-//! You can test this out by running:
-//!
-//!     cargo run --example chat
-//!
-//! And then in another terminal run:
-//!
-//!     telnet localhost 6142
-//!
-//! You can run the `telnet` command in any number of additional windows.
-//!
-//! You can run the second command in multiple windows and then chat between the
-//! two, seeing the messages from the other client as they're received. For all
-//! connected clients they'll all join the same room and see everyone else's
-//! messages.
-
-#![warn(rust_2018_idioms)]
-
-use tokio::net::{TcpListener, TcpStream};
-use tokio::sync::{mpsc, Mutex};
-use tokio_stream::StreamExt;
-use tokio_util::codec::{Framed, LinesCodec};
-
-use futures::SinkExt;
-use std::collections::HashMap;
-use std::env;
-use std::error::Error;
-use std::io;
-use std::net::SocketAddr;
-use std::sync::Arc;
-
-const DEFAULT_ADDR: &str = "127.0.0.1:6142";
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    use tracing_subscriber::{fmt::format::FmtSpan, EnvFilter};
-    // Configure a `tracing` subscriber that logs traces emitted by the chat
-    // server.
-    tracing_subscriber::fmt()
-        // Filter what traces are displayed based on the RUST_LOG environment
-        // variable.
-        //
-        // Traces emitted by the example code will always be displayed. You
-        // can set `RUST_LOG=tokio=trace` to enable additional traces emitted by
-        // Tokio itself.
-        .with_env_filter(EnvFilter::from_default_env().add_directive("chat=info".parse()?))
-        // Log events when `tracing` spans are created, entered, exited, or
-        // closed. When Tokio's internal tracing support is enabled (as
-        // described above), this can be used to track the lifecycle of spawned
-        // tasks on the Tokio runtime.
-        .with_span_events(FmtSpan::FULL)
-        // Set this subscriber as the default, to collect all traces emitted by
-        // the program.
-        .init();
-
-    // Create the shared state. This is how all the peers communicate.
-    //
-    // The server task will hold a handle to this. For every new client, the
-    // `state` handle is cloned and passed into the task that processes the
-    // client connection.
-    let state = Arc::new(Mutex::new(Shared::new()));
-
-    let addr = env::args()
-        .nth(1)
-        .unwrap_or_else(|| DEFAULT_ADDR.to_string());
-
-    // Bind a TCP listener to the socket address.
-    //
-    // Note that this is the Tokio TcpListener, which is fully async.
-    let listener = TcpListener::bind(&addr).await?;
-
-    tracing::info!("server running on {addr}");
-
-    loop {
-        // Asynchronously wait for an inbound TcpStream.
-        let (stream, addr) = listener.accept().await?;
-
-        // Clone a handle to the `Shared` state for the new connection.
-        let state = Arc::clone(&state);
-
-        // Spawn our handler to be run asynchronously.
-        tokio::spawn(async move {
-            tracing::debug!("accepted connection from {addr}");
-            if let Err(e) = process(state, stream, addr).await {
-                tracing::warn!("Connection from {addr} failed: {e:?}");
-            }
-        });
-    }
-}
-
-/// Shorthand for the transmit half of the message channel.
-type Tx = mpsc::UnboundedSender<String>;
-
-/// Shorthand for the receive half of the message channel.
-type Rx = mpsc::UnboundedReceiver<String>;
-
-/// Data that is shared between all peers in the chat server.
-///
-/// This is the set of `Tx` handles for all connected clients. Whenever a
-/// message is received from a client, it is broadcasted to all peers by
-/// iterating over the `peers` entries and sending a copy of the message on each
-/// `Tx`.
-struct Shared {
-    peers: HashMap<SocketAddr, Tx>,
-}
-
-/// The state for each connected client.
-struct Peer {
-    /// The TCP socket wrapped with the `Lines` codec, defined below.
-    ///
-    /// This handles sending and receiving data on the socket. When using
-    /// `Lines`, we can work at the line level instead of having to manage the
-    /// raw byte operations.
-    lines: Framed<TcpStream, LinesCodec>,
-
-    /// Receive half of the message channel.
-    ///
-    /// This is used to receive messages from peers. When a message is received
-    /// off of this `Rx`, it will be written to the socket.
-    rx: Rx,
-}
-
-impl Shared {
-    /// Create a new, empty, instance of `Shared`.
-    fn new() -> Self {
-        Shared {
-            peers: HashMap::new(),
-        }
-    }
-
-    /// Send a `LineCodec` encoded message to every peer, except
-    /// for the sender.
-    ///
-    /// This function also cleans up disconnected peers automatically.
-    async fn broadcast(&mut self, sender: SocketAddr, message: &str) {
-        let mut failed_peers = Vec::new();
-        let message = message.to_string(); // Clone once for all sends
-
-        for (addr, tx) in self.peers.iter() {
-            if *addr != sender && tx.send(message.clone()).is_err() {
-                // Receiver has been dropped, mark for removal
-                failed_peers.push(*addr);
-            }
-        }
-
-        // Clean up disconnected peers
-        for addr in failed_peers {
-            self.peers.remove(&addr);
-            tracing::debug!("Removed disconnected peer: {addr}");
-        }
-    }
-}
-
-impl Peer {
-    /// Create a new instance of `Peer`.
-    async fn new(
-        state: Arc<Mutex<Shared>>,
-        lines: Framed<TcpStream, LinesCodec>,
-    ) -> io::Result<Peer> {
-        // Get the client socket address
-        let addr = lines.get_ref().peer_addr()?;
-
-        // Create a channel for this peer
-        let (tx, rx) = mpsc::unbounded_channel();
-
-        // Add an entry for this `Peer` in the shared state map.
-        state.lock().await.peers.insert(addr, tx);
-
-        Ok(Peer { lines, rx })
-    }
-}
-
-/// Process an individual chat client
-async fn process(
-    state: Arc<Mutex<Shared>>,
-    stream: TcpStream,
-    addr: SocketAddr,
-) -> Result<(), Box<dyn Error>> {
-    let mut lines = Framed::new(stream, LinesCodec::new());
-
-    // Send a prompt to the client to enter their username.
-    lines.send("Please enter your username:").await?;
-
-    // Read the first line from the `LineCodec` stream to get the username.
-    let Some(Ok(username)) = lines.next().await else {
-        // We didn't get a line so we return early here.
-        tracing::error!("Failed to get username from {addr}. Client disconnected.");
-        return Ok(());
-    };
-
-    // Register our peer with state which internally sets up some channels.
-    let mut peer = Peer::new(state.clone(), lines).await?;
-
-    // A client has connected, let's let everyone know.
-    {
-        let mut state = state.lock().await;
-        let msg = format!("{username} has joined the chat");
-        tracing::info!("{msg}");
-        state.broadcast(addr, &msg).await;
-    }
-
-    // Process incoming messages until our stream is exhausted by a disconnect.
-    loop {
-        tokio::select! {
-            // A message was received from a peer. Send it to the current user.
-            Some(msg) = peer.rx.recv() => {
-                if let Err(e) = peer.lines.send(&msg).await {
-                    tracing::error!("Failed to send message to {username}: {e:?}");
-                    break;
-                }
-            }
-            result = peer.lines.next() => match result {
-                // A message was received from the current user, we should
-                // broadcast this message to the other users.
-                Some(Ok(msg)) => {
-                    let mut state = state.lock().await;
-                    let msg = format!("{username}: {msg}");
-
-                    state.broadcast(addr, &msg).await;
-                }
-                // An error occurred.
-                Some(Err(e)) => {
-                    tracing::error!(
-                        "an error occurred while processing messages for {username}; error = {e:?}"
-                    );
-                    break;
-                }
-                // The stream has been exhausted.
-                None => break,
-            },
-        }
-    }
-
-    // If this section is reached it means that the client was disconnected!
-    // Let's let everyone still connected know about it.
-    {
-        let mut state = state.lock().await;
-        state.peers.remove(&addr);
-
-        let msg = format!("{username} has left the chat");
-        tracing::info!("{msg}");
-        state.broadcast(addr, &msg).await;
-    }
-
-    Ok(())
-}

examples/connect-tcp.rs

@@ -1,73 +0,0 @@
-//! An example of hooking up stdin/stdout to a TCP stream.
-//!
-//! This example will connect to a socket address specified in the argument list
-//! and then forward all data read on stdin to the server, printing out all data
-//! received on stdout. Each line entered on stdin will be translated to a TCP
-//! packet which is then sent to the remote address.
-//!
-//! Note that this is not currently optimized for performance, especially
-//! around buffer management. Rather it's intended to show an example of
-//! working with a client.
-//!
-//! This example can be quite useful when interacting with the other examples in
-//! this repository! Many of them recommend running this as a simple "hook up
-//! stdin/stdout to a server" to get up and running.
-
-#![warn(rust_2018_idioms)]
-
-use tokio::io::{stdin, stdout};
-use tokio::net::TcpStream;
-use tokio_util::codec::{BytesCodec, FramedRead, FramedWrite};
-
-use bytes::Bytes;
-use futures::{future, Sink, SinkExt, Stream, StreamExt};
-use std::env;
-use std::error::Error;
-use std::net::SocketAddr;
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    // Parse what address we're going to connect to
-    let args = env::args().skip(1).collect::<Vec<_>>();
-    let addr = args
-        .first()
-        .ok_or("this program requires at least one argument")?;
-    let addr = addr.parse::<SocketAddr>()?;
-
-    let stdin = FramedRead::new(stdin(), BytesCodec::new());
-    let stdin = stdin.map(|i| i.map(|bytes| bytes.freeze()));
-    let stdout = FramedWrite::new(stdout(), BytesCodec::new());
-
-    connect(&addr, stdin, stdout).await?;
-
-    Ok(())
-}
-
-pub async fn connect(
-    addr: &SocketAddr,
-    mut stdin: impl Stream<Item = Result<Bytes, std::io::Error>> + Unpin,
-    mut stdout: impl Sink<Bytes, Error = std::io::Error> + Unpin,
-) -> Result<(), Box<dyn Error>> {
-    let mut stream = TcpStream::connect(addr).await?;
-    let (r, w) = stream.split();
-    let mut sink = FramedWrite::new(w, BytesCodec::new());
-    // filter map Result<BytesMut, Error> stream into just a Bytes stream to match stdout Sink
-    // on the event of an Error, log the error and end the stream
-    let mut stream = FramedRead::new(r, BytesCodec::new())
-        .filter_map(|i| match i {
-            //BytesMut into Bytes
-            Ok(i) => future::ready(Some(i.freeze())),
-            Err(e) => {
-                eprintln!("failed to read from socket; error={e}");
-                future::ready(None)
-            }
-        })
-        .map(Ok);
-
-    tokio::select! {
-        r = sink.send_all(&mut stdin) => r?,
-        r = stdout.send_all(&mut stream) => r?,
-    }
-
-    Ok(())
-}

examples/connect-udp.rs

@@ -1,94 +0,0 @@
-//! An example of hooking up stdin/stdout to a UDP stream.
-//!
-//! This example will connect to a socket address specified in the argument list
-//! and then forward all data read on stdin to the server, printing out all data
-//! received on stdout. Each line entered on stdin will be translated to a UDP
-//! packet which is then sent to the remote address.
-//!
-//! Note that this is not currently optimized for performance, especially
-//! around buffer management. Rather it's intended to show an example of
-//! working with a client.
-//!
-//! This example can be quite useful when interacting with the other examples in
-//! this repository! Many of them recommend running this as a simple "hook up
-//! stdin/stdout to a server" to get up and running.
-
-#![warn(rust_2018_idioms)]
-
-use tokio::io::{stdin, stdout};
-use tokio::net::UdpSocket;
-use tokio_util::codec::{BytesCodec, FramedRead, FramedWrite};
-
-use bytes::Bytes;
-use futures::{Sink, SinkExt, Stream, StreamExt};
-use std::env;
-use std::error::Error;
-use std::net::SocketAddr;
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    // Parse what address we're going to connect to
-    let args = env::args().skip(1).collect::<Vec<_>>();
-    let addr = args
-        .first()
-        .ok_or("this program requires at least one argument")?;
-    let addr = addr.parse::<SocketAddr>()?;
-
-    let stdin = FramedRead::new(stdin(), BytesCodec::new());
-    let stdin = stdin.map(|i| i.map(|bytes| bytes.freeze()));
-    let stdout = FramedWrite::new(stdout(), BytesCodec::new());
-
-    connect(&addr, stdin, stdout).await?;
-
-    Ok(())
-}
-
-pub async fn connect(
-    addr: &SocketAddr,
-    stdin: impl Stream<Item = Result<Bytes, std::io::Error>> + Unpin,
-    stdout: impl Sink<Bytes, Error = std::io::Error> + Unpin,
-) -> Result<(), Box<dyn Error>> {
-    // We'll bind our UDP socket to a local IP/port, but for now we
-    // basically let the OS pick both of those.
-    let bind_addr = if addr.ip().is_ipv4() {
-        "0.0.0.0:0"
-    } else {
-        "[::]:0"
-    };
-
-    let socket = UdpSocket::bind(&bind_addr).await?;
-    socket.connect(addr).await?;
-
-    tokio::select! {
-        r = send(stdin, &socket) => r?,
-        r = recv(stdout, &socket) => r?,
-    }
-
-    Ok(())
-}
-
-async fn send(
-    mut stdin: impl Stream<Item = Result<Bytes, std::io::Error>> + Unpin,
-    writer: &UdpSocket,
-) -> Result<(), std::io::Error> {
-    while let Some(item) = stdin.next().await {
-        let buf = item?;
-        writer.send(&buf[..]).await?;
-    }
-
-    Ok(())
-}
-
-async fn recv(
-    mut stdout: impl Sink<Bytes, Error = std::io::Error> + Unpin,
-    reader: &UdpSocket,
-) -> Result<(), std::io::Error> {
-    loop {
-        let mut buf = vec![0; 1024];
-        let n = reader.recv(&mut buf[..]).await?;
-
-        if n > 0 {
-            stdout.send(Bytes::copy_from_slice(&buf[..n])).await?;
-        }
-    }
-}

examples/custom-executor-tokio-context.rs

@@ -1,32 +0,0 @@
-// This example shows how to use the tokio runtime with any other executor
-//
-//It takes advantage from RuntimeExt which provides the extension to customize your
-//runtime.
-
-use tokio::net::TcpListener;
-use tokio::runtime::Builder;
-use tokio::sync::oneshot;
-use tokio_util::context::RuntimeExt;
-
-fn main() {
-    let (tx, rx) = oneshot::channel();
-    let rt1 = Builder::new_multi_thread()
-        .worker_threads(1)
-        // no timer!
-        .build()
-        .unwrap();
-    let rt2 = Builder::new_multi_thread()
-        .worker_threads(1)
-        .enable_all()
-        .build()
-        .unwrap();
-
-    // Without the `HandleExt.wrap()` there would be a panic because there is
-    // no timer running, since it would be referencing runtime r1.
-    rt1.block_on(rt2.wrap(async move {
-        let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
-        println!("addr: {:?}", listener.local_addr());
-        tx.send(()).unwrap();
-    }));
-    futures::executor::block_on(rx).unwrap();
-}

examples/custom-executor.rs

@@ -1,56 +0,0 @@
-// This example shows how to use the tokio runtime with any other executor
-//
-// The main components are a spawn fn that will wrap futures in a special future
-// that will always enter the tokio context on poll. This only spawns one extra thread
-// to manage and run the tokio drivers in the background.
-
-use tokio::net::TcpListener;
-use tokio::sync::oneshot;
-
-fn main() {
-    let (tx, rx) = oneshot::channel();
-
-    my_custom_runtime::spawn(async move {
-        let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
-
-        println!("addr: {:?}", listener.local_addr());
-
-        tx.send(()).unwrap();
-    });
-
-    futures::executor::block_on(rx).unwrap();
-}
-
-mod my_custom_runtime {
-    use once_cell::sync::Lazy;
-    use std::future::Future;
-    use tokio_util::context::TokioContext;
-
-    pub fn spawn(f: impl Future<Output = ()> + Send + 'static) {
-        EXECUTOR.spawn(f);
-    }
-
-    struct ThreadPool {
-        inner: futures::executor::ThreadPool,
-        rt: tokio::runtime::Runtime,
-    }
-
-    static EXECUTOR: Lazy<ThreadPool> = Lazy::new(|| {
-        // Spawn tokio runtime on a single background thread
-        // enabling IO and timers.
-        let rt = tokio::runtime::Builder::new_multi_thread()
-            .enable_all()
-            .build()
-            .unwrap();
-        let inner = futures::executor::ThreadPool::builder().create().unwrap();
-
-        ThreadPool { inner, rt }
-    });
-
-    impl ThreadPool {
-        fn spawn(&self, f: impl Future<Output = ()> + Send + 'static) {
-            let handle = self.rt.handle().clone();
-            self.inner.spawn_ok(TokioContext::new(f, handle));
-        }
-    }
-}

examples/dump.rs

@@ -1,99 +0,0 @@
-//! This example demonstrates tokio's experimental task dumping functionality.
-//! This application deadlocks. Input CTRL+C to display traces of each task, or
-//! input CTRL+C twice within 1 second to quit.
-
-#[cfg(all(
-    tokio_unstable,
-    target_os = "linux",
-    any(
-        target_arch = "aarch64",
-        target_arch = "x86",
-        target_arch = "x86_64",
-        target_arch = "s390x"
-    )
-))]
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    use std::sync::Arc;
-    use tokio::sync::Barrier;
-
-    #[inline(never)]
-    async fn a(barrier: Arc<Barrier>) {
-        b(barrier).await
-    }
-
-    #[inline(never)]
-    async fn b(barrier: Arc<Barrier>) {
-        c(barrier).await
-    }
-
-    #[inline(never)]
-    async fn c(barrier: Arc<Barrier>) {
-        barrier.wait().await;
-    }
-
-    // Prints a task dump upon receipt of CTRL+C, or returns if CTRL+C is
-    // inputted twice within a second.
-    async fn dump_or_quit() {
-        use tokio::time::{timeout, Duration, Instant};
-        let handle = tokio::runtime::Handle::current();
-        let mut last_signal: Option<Instant> = None;
-        // wait for CTRL+C
-        while let Ok(_) = tokio::signal::ctrl_c().await {
-            // exit if a CTRL+C is inputted twice within 1 second
-            if let Some(time_since_last_signal) = last_signal.map(|i| i.elapsed()) {
-                if time_since_last_signal < Duration::from_secs(1) {
-                    return;
-                }
-            }
-            last_signal = Some(Instant::now());
-
-            // capture a dump, and print each trace
-            println!("{:-<80}", "");
-            if let Ok(dump) = timeout(Duration::from_secs(2), handle.dump()).await {
-                for task in dump.tasks().iter() {
-                    let id = task.id();
-                    let trace = task.trace();
-                    println!("TASK {id}:");
-                    println!("{trace}\n");
-                }
-            } else {
-                println!("Task dumping timed out. Use a native debugger (like gdb) to debug the deadlock.");
-            }
-            println!("{:-<80}", "");
-            println!("Input CTRL+C twice within 1 second to exit.");
-        }
-    }
-
-    println!("This program has a deadlock.");
-    println!("Input CTRL+C to print a task dump.");
-    println!("Input CTRL+C twice within 1 second to exit.");
-
-    // oops! this barrier waits for one more task than will ever come.
-    let barrier = Arc::new(Barrier::new(3));
-
-    let task_1 = tokio::spawn(a(barrier.clone()));
-    let task_2 = tokio::spawn(a(barrier));
-
-    tokio::select!(
-        _ = dump_or_quit() => {},
-        _ = task_1 => {},
-        _ = task_2 => {},
-    );
-
-    Ok(())
-}
-
-#[cfg(not(all(
-    tokio_unstable,
-    target_os = "linux",
-    any(
-        target_arch = "aarch64",
-        target_arch = "x86",
-        target_arch = "x86_64",
-        target_arch = "s390x"
-    )
-)))]
-fn main() {
-    println!("task dumps are not available")
-}

examples/echo-tcp.rs

@@ -1,85 +0,0 @@
-//! A "hello world" echo server with Tokio
-//!
-//! This server will create a TCP listener, accept connections in a loop, and
-//! write back everything that's read off of each TCP connection.
-//!
-//! Because the Tokio runtime uses a thread pool, each TCP connection is
-//! processed concurrently with all other TCP connections across multiple
-//! threads.
-//!
-//! To see this server in action, you can run this in one terminal:
-//!
-//!     cargo run --example echo-tcp
-//!
-//! and in another terminal you can run:
-//!
-//!     cargo run --example connect-tcp 127.0.0.1:8080
-//!
-//! Each line you type in to the `connect-tcp` terminal should be echo'd back to
-//! you! If you open up multiple terminals running the `connect-tcp` example you
-//! should be able to see them all make progress simultaneously.
-
-#![warn(rust_2018_idioms)]
-
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::TcpListener;
-
-use std::env;
-use std::error::Error;
-
-const DEFAULT_ADDR: &str = "127.0.0.1:8080";
-const BUFFER_SIZE: usize = 4096;
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    // Allow passing an address to listen on as the first argument of this
-    // program, but otherwise we'll just set up our TCP listener on
-    // 127.0.0.1:8080 for connections.
-    let addr = env::args()
-        .nth(1)
-        .unwrap_or_else(|| DEFAULT_ADDR.to_string());
-
-    // Next up we create a TCP listener which will listen for incoming
-    // connections. This TCP listener is bound to the address we determined
-    // above and must be associated with an event loop.
-    let listener = TcpListener::bind(&addr).await?;
-    println!("Listening on: {addr}");
-
-    loop {
-        // Asynchronously wait for an inbound socket.
-        let (mut socket, addr) = listener.accept().await?;
-
-        // And this is where much of the magic of this server happens. We
-        // crucially want all clients to make progress concurrently, rather than
-        // blocking one on completion of another. To achieve this we use the
-        // `tokio::spawn` function to execute the work in the background.
-        //
-        // Essentially here we're executing a new task to run concurrently,
-        // which will allow all of our clients to be processed concurrently.
-
-        tokio::spawn(async move {
-            let mut buf = vec![0; BUFFER_SIZE];
-
-            // In a loop, read data from the socket and write the data back.
-            loop {
-                match socket.read(&mut buf).await {
-                    Ok(0) => {
-                        // Connection closed by peer
-                        return;
-                    }
-                    Ok(n) => {
-                        // Write the data back. If writing fails, log the error and exit.
-                        if let Err(e) = socket.write_all(&buf[0..n]).await {
-                            eprintln!("Failed to write to socket {}: {}", addr, e);
-                            return;
-                        }
-                    }
-                    Err(e) => {
-                        eprintln!("Failed to read from socket {}: {}", addr, e);
-                        return;
-                    }
-                }
-            }
-        });
-    }
-}

examples/echo-udp.rs

@@ -1,92 +0,0 @@
-//! An UDP echo server that just sends back everything that it receives.
-//!
-//! If you're on Unix you can test this out by in one terminal executing:
-//!
-//!     cargo run --example echo-udp
-//!
-//! and in another terminal you can run:
-//!
-//!     cargo run --example connect-udp 127.0.0.1:8080
-//!
-//! Each line you type in to the `connect-udp` terminal should be echo'd back to you!
-//!
-//! # Binding to all interfaces
-//!
-//! By default this example binds to `127.0.0.1` so it is only reachable
-//! from the local machine.
-//!
-//! To listen on all interfaces instead:
-//!
-//! ```sh
-//! cargo run --example echo-udp -- 0.0.0.0:8080
-//! ```
-//!
-//! Binding to `0.0.0.0` exposes the server on all network interfaces.
-//! Only do this in trusted network environments.
-//!
-//! On multi-homed systems, a UDP socket bound to a wildcard address
-//! (`0.0.0.0` or `::`) cannot always send replies from the same local IP
-//! that received the packet. Replies may therefore originate from a
-//! different address than the client targeted. See [this Cloudflare blog
-//! post][udp-blog] for more details.
-//!
-//! [udp-blog]: https://blog.cloudflare.com/everything-you-ever-wanted-to-know-about-udp-sockets-but-were-afraid-to-ask-part-1
-
-#![warn(rust_2018_idioms)]
-
-use std::error::Error;
-use std::net::SocketAddr;
-use std::{env, io};
-use tokio::net::UdpSocket;
-
-struct Server {
-    socket: UdpSocket,
-    buf: Vec<u8>,
-    to_send: Option<(usize, SocketAddr)>,
-}
-
-impl Server {
-    async fn run(self) -> Result<(), io::Error> {
-        let Server {
-            socket,
-            mut buf,
-            mut to_send,
-        } = self;
-
-        loop {
-            // First we check to see if there's a message we need to echo back.
-            // If so then we try to send it back to the original source, waiting
-            // until it's writable and we're able to do so.
-            if let Some((size, peer)) = to_send {
-                let amt = socket.send_to(&buf[..size], &peer).await?;
-
-                println!("Echoed {amt}/{size} bytes to {peer}");
-            }
-
-            // If we're here then `to_send` is `None`, so we take a look for the
-            // next message we're going to echo back.
-            to_send = Some(socket.recv_from(&mut buf).await?);
-        }
-    }
-}
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    let addr = env::args()
-        .nth(1)
-        .unwrap_or_else(|| "127.0.0.1:8080".to_string());
-
-    let socket = UdpSocket::bind(&addr).await?;
-    println!("Listening on: {}", socket.local_addr()?);
-
-    let server = Server {
-        socket,
-        buf: vec![0; 1024],
-        to_send: None,
-    };
-
-    // This starts the server task.
-    server.run().await?;
-
-    Ok(())
-}

examples/graceful-shutdown.rs

@@ -1,120 +0,0 @@
-//! Graceful shutdown example.
-//!
-//! This example follows the same approach described in the
-//! [Graceful Shutdown tutorial](https://tokio.rs/tokio/topics/shutdown):
-//!
-//! - A [`CancellationToken`] tells tasks to stop accepting new work.
-//! - A [`TaskTracker`] waits for in-flight work to complete.
-//!
-//! It runs a TCP echo server on `127.0.0.1:6142`. When Ctrl+C is
-//! pressed, the server stops accepting connections and waits for all
-//! active connections to finish before exiting.
-//!
-//! Start the server:
-//!
-//!     cargo run --example graceful-shutdown
-//!
-//! Then connect with:
-//!
-//!     nc 127.0.0.1 6142
-//!
-//! Press Ctrl+C on the server to trigger a graceful shutdown.
-
-#![warn(rust_2018_idioms)]
-
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
-use tokio::net::{TcpListener, TcpStream};
-use tokio::time::{self, Duration};
-use tokio_util::sync::CancellationToken;
-use tokio_util::task::TaskTracker;
-
-use std::error::Error;
-use std::net::SocketAddr;
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    let listener = TcpListener::bind("127.0.0.1:6142").await?;
-    println!("listening on 127.0.0.1:6142");
-
-    let token = CancellationToken::new();
-    let tracker = TaskTracker::new();
-
-    loop {
-        tokio::select! {
-            result = listener.accept() => {
-                let (socket, addr) = match result {
-                    Ok(conn) => conn,
-                    Err(e) => {
-                        // Transient errors (e.g. fd exhaustion) are recoverable,
-                        // so we log and continue. A production server might add a
-                        // backoff or break on fatal errors to avoid a busy loop.
-                        eprintln!("failed to accept: {e}");
-                        continue;
-                    }
-                };
-                println!("accepted connection from {addr}");
-
-                let token = token.clone();
-                tracker.spawn(handle_connection(socket, addr, token));
-            }
-
-            _ = tokio::signal::ctrl_c() => {
-                println!("\nshutdown signal received, waiting for connections to finish");
-                break;
-            }
-        }
-    }
-
-    // Signal all tasks to stop and wait for them to complete.
-    token.cancel();
-    tracker.close();
-    tracker.wait().await;
-
-    println!("shutdown complete");
-    Ok(())
-}
-
-async fn handle_connection(mut socket: TcpStream, addr: SocketAddr, token: CancellationToken) {
-    tokio::select! {
-        _ = echo(&mut socket) => {}
-        _ = token.cancelled() => {
-            notify_shutdown(&mut socket).await;
-        }
-    }
-
-    println!("connection from {addr} closed");
-}
-
-/// Reads lines from the client and writes them back.
-///
-/// Called for every accepted connection. Runs until the client disconnects
-/// or a read/write error occurs.
-async fn echo(socket: &mut TcpStream) {
-    let (reader, mut writer) = socket.split();
-    let mut reader = BufReader::new(reader);
-    let mut line = String::new();
-
-    loop {
-        match reader.read_line(&mut line).await {
-            Ok(0) | Err(_) => return,
-            Ok(_) => {
-                if writer.write_all(line.as_bytes()).await.is_err() {
-                    return;
-                }
-                line.clear();
-            }
-        }
-    }
-}
-
-/// Sends a shutdown notice to the client before closing the connection.
-///
-/// Called when the cancellation token fires. Uses a timeout so that a
-/// slow or unresponsive client cannot hold up the server shutdown.
-async fn notify_shutdown(socket: &mut TcpStream) {
-    let _ = time::timeout(
-        Duration::from_secs(1),
-        socket.write_all(b"server shutting down\n"),
-    )
-    .await;
-}

examples/hello_world.rs

@@ -1,31 +0,0 @@
-//! A simple client that opens a TCP stream, writes "hello world\n", and closes
-//! the connection.
-//!
-//! To start a server that this client can talk to on port 6142, you can use this command:
-//!
-//!     ncat -l 6142
-//!
-//! And then in another terminal run:
-//!
-//!     cargo run --example hello_world
-
-#![warn(rust_2018_idioms)]
-
-use tokio::io::AsyncWriteExt;
-use tokio::net::TcpStream;
-
-use std::error::Error;
-
-#[tokio::main]
-pub async fn main() -> Result<(), Box<dyn Error>> {
-    // Open a TCP stream to the socket address.
-    //
-    // Note that this is the Tokio TcpStream, which is fully async.
-    let mut stream = TcpStream::connect("127.0.0.1:6142").await?;
-    println!("created stream");
-
-    let result = stream.write_all(b"hello world\n").await;
-    println!("wrote to stream; success={:?}", result.is_ok());
-
-    Ok(())
-}

examples/named-pipe-multi-client.rs

@@ -1,98 +0,0 @@
-use std::io;
-
-#[cfg(windows)]
-async fn windows_main() -> io::Result<()> {
-    use std::time::Duration;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
-    use tokio::time;
-    use windows_sys::Win32::Foundation::ERROR_PIPE_BUSY;
-
-    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-multi-client";
-    const N: usize = 10;
-
-    // The first server needs to be constructed early so that clients can
-    // be correctly connected. Otherwise a waiting client will error.
-    //
-    // Here we also make use of `first_pipe_instance`, which will ensure
-    // that there are no other servers up and running already.
-    let mut server = ServerOptions::new()
-        .first_pipe_instance(true)
-        .create(PIPE_NAME)?;
-
-    let server = tokio::spawn(async move {
-        // Artificial workload.
-        time::sleep(Duration::from_secs(1)).await;
-
-        for _ in 0..N {
-            // Wait for client to connect.
-            server.connect().await?;
-            let mut inner = server;
-
-            // Construct the next server to be connected before sending the one
-            // we already have of onto a task. This ensures that the server
-            // isn't closed (after it's done in the task) before a new one is
-            // available. Otherwise the client might error with
-            // `io::ErrorKind::NotFound`.
-            server = ServerOptions::new().create(PIPE_NAME)?;
-
-            let _ = tokio::spawn(async move {
-                let mut buf = vec![0u8; 4];
-                inner.read_exact(&mut buf).await?;
-                inner.write_all(b"pong").await?;
-                Ok::<_, io::Error>(())
-            });
-        }
-
-        Ok::<_, io::Error>(())
-    });
-
-    let mut clients = Vec::new();
-
-    for _ in 0..N {
-        clients.push(tokio::spawn(async move {
-            // This showcases a generic connect loop.
-            //
-            // We immediately try to create a client, if it's not found or
-            // the pipe is busy we use the specialized wait function on the
-            // client builder.
-            let mut client = loop {
-                match ClientOptions::new().open(PIPE_NAME) {
-                    Ok(client) => break client,
-                    Err(e) if e.raw_os_error() == Some(ERROR_PIPE_BUSY as i32) => (),
-                    Err(e) => return Err(e),
-                }
-
-                time::sleep(Duration::from_millis(5)).await;
-            };
-
-            let mut buf = [0u8; 4];
-            client.write_all(b"ping").await?;
-            client.read_exact(&mut buf).await?;
-            Ok::<_, io::Error>(buf)
-        }));
-    }
-
-    for client in clients {
-        let result = client.await?;
-        assert_eq!(&result?[..], b"pong");
-    }
-
-    server.await??;
-    Ok(())
-}
-
-#[tokio::main]
-async fn main() -> io::Result<()> {
-    #[cfg(windows)]
-    {
-        windows_main().await?;
-    }
-
-    #[cfg(not(windows))]
-    {
-        println!("Named pipes are only supported on Windows!");
-    }
-
-    Ok(())
-}

examples/named-pipe-ready.rs

@@ -1,161 +0,0 @@
-use std::io;
-
-#[cfg(windows)]
-async fn windows_main() -> io::Result<()> {
-    use tokio::io::Interest;
-    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
-
-    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-single-client";
-
-    let server = ServerOptions::new().create(PIPE_NAME)?;
-
-    let server = tokio::spawn(async move {
-        // Note: we wait for a client to connect.
-        server.connect().await?;
-
-        let buf = {
-            let mut read_buf = [0u8; 5];
-            let mut read_buf_cursor = 0;
-
-            loop {
-                server.readable().await?;
-
-                let buf = &mut read_buf[read_buf_cursor..];
-
-                match server.try_read(buf) {
-                    Ok(n) => {
-                        read_buf_cursor += n;
-
-                        if read_buf_cursor == read_buf.len() {
-                            break;
-                        }
-                    }
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue;
-                    }
-                    Err(e) => {
-                        return Err(e);
-                    }
-                }
-            }
-
-            read_buf
-        };
-
-        {
-            let write_buf = b"pong\n";
-            let mut write_buf_cursor = 0;
-
-            loop {
-                let buf = &write_buf[write_buf_cursor..];
-
-                if buf.is_empty() {
-                    break;
-                }
-
-                server.writable().await?;
-
-                match server.try_write(buf) {
-                    Ok(n) => {
-                        write_buf_cursor += n;
-                    }
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue;
-                    }
-                    Err(e) => {
-                        return Err(e);
-                    }
-                }
-            }
-        }
-
-        Ok::<_, io::Error>(buf)
-    });
-
-    let client = tokio::spawn(async move {
-        // There's no need to use a connect loop here, since we know that the
-        // server is already up - `open` was called before spawning any of the
-        // tasks.
-        let client = ClientOptions::new().open(PIPE_NAME)?;
-
-        let mut read_buf = [0u8; 5];
-        let mut read_buf_cursor = 0;
-        let write_buf = b"ping\n";
-        let mut write_buf_cursor = 0;
-
-        loop {
-            let mut interest = Interest::READABLE;
-            if write_buf_cursor < write_buf.len() {
-                interest |= Interest::WRITABLE;
-            }
-
-            let ready = client.ready(interest).await?;
-
-            if ready.is_readable() {
-                let buf = &mut read_buf[read_buf_cursor..];
-
-                match client.try_read(buf) {
-                    Ok(n) => {
-                        read_buf_cursor += n;
-
-                        if read_buf_cursor == read_buf.len() {
-                            break;
-                        }
-                    }
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue;
-                    }
-                    Err(e) => {
-                        return Err(e);
-                    }
-                }
-            }
-
-            if ready.is_writable() {
-                let buf = &write_buf[write_buf_cursor..];
-
-                if buf.is_empty() {
-                    continue;
-                }
-
-                match client.try_write(buf) {
-                    Ok(n) => {
-                        write_buf_cursor += n;
-                    }
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue;
-                    }
-                    Err(e) => {
-                        return Err(e);
-                    }
-                }
-            }
-        }
-
-        let buf = String::from_utf8_lossy(&read_buf).into_owned();
-
-        Ok::<_, io::Error>(buf)
-    });
-
-    let (server, client) = tokio::try_join!(server, client)?;
-
-    assert_eq!(server?, *b"ping\n");
-    assert_eq!(client?, "pong\n");
-
-    Ok(())
-}
-
-#[tokio::main]
-async fn main() -> io::Result<()> {
-    #[cfg(windows)]
-    {
-        windows_main().await?;
-    }
-
-    #[cfg(not(windows))]
-    {
-        println!("Named pipes are only supported on Windows!");
-    }
-
-    Ok(())
-}

examples/named-pipe.rs

@@ -1,60 +0,0 @@
-use std::io;
-
-#[cfg(windows)]
-async fn windows_main() -> io::Result<()> {
-    use tokio::io::AsyncWriteExt;
-    use tokio::io::{AsyncBufReadExt, BufReader};
-    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
-
-    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-single-client";
-
-    let server = ServerOptions::new().create(PIPE_NAME)?;
-
-    let server = tokio::spawn(async move {
-        // Note: we wait for a client to connect.
-        server.connect().await?;
-
-        let mut server = BufReader::new(server);
-
-        let mut buf = String::new();
-        server.read_line(&mut buf).await?;
-        server.write_all(b"pong\n").await?;
-        Ok::<_, io::Error>(buf)
-    });
-
-    let client = tokio::spawn(async move {
-        // There's no need to use a connect loop here, since we know that the
-        // server is already up - `open` was called before spawning any of the
-        // tasks.
-        let client = ClientOptions::new().open(PIPE_NAME)?;
-
-        let mut client = BufReader::new(client);
-
-        let mut buf = String::new();
-        client.write_all(b"ping\n").await?;
-        client.read_line(&mut buf).await?;
-        Ok::<_, io::Error>(buf)
-    });
-
-    let (server, client) = tokio::try_join!(server, client)?;
-
-    assert_eq!(server?, "ping\n");
-    assert_eq!(client?, "pong\n");
-
-    Ok(())
-}
-
-#[tokio::main]
-async fn main() -> io::Result<()> {
-    #[cfg(windows)]
-    {
-        windows_main().await?;
-    }
-
-    #[cfg(not(windows))]
-    {
-        println!("Named pipes are only supported on Windows!");
-    }
-
-    Ok(())
-}

examples/prewarm-fd-table.rs

@@ -1,86 +0,0 @@
-//! Demonstrates pre-warming the Linux file descriptor table to avoid latency
-//! spikes caused by file descriptor table growth in multi-threaded processes.
-//!
-//! On Linux, the kernel's FD table is grown lazily and protected by RCU
-//! synchronization. In multi-threaded processes, when a syscall like `socket()`
-//! triggers a table resize, the calling thread blocks until all RCU readers
-//! quiesce. This can cause stalls of tens of milliseconds on tokio worker threads,
-//! blocking the entire event loop (not just one task).
-//!
-//! The workaround is to force the kernel to expand the FD table once per process
-//! (before any runtime starts), by duplicating an FD to a high slot and then
-//! closing it. The kernel never shrinks the FD table during a process's lifetime,
-//! so the capacity persists.
-//!
-//! This is most relevant for services that open many connections concurrently
-//! (e.g. HTTP servers, connection pools). The pre-warm target should be at least
-//! your expected peak FD count, and must not exceed `RLIMIT_NOFILE`.
-//!
-//! See: <https://github.com/tokio-rs/tokio/issues/7970>
-//!
-//! Usage:
-//!
-//!     cargo run --example prewarm-fd-table
-
-#![warn(rust_2018_idioms)]
-
-/// Pre-warms the FD table using `fcntl(F_DUPFD_CLOEXEC)` to duplicate an FD
-/// into a high slot, expanding the table in a single syscall. `F_DUPFD_CLOEXEC`
-/// allocates the lowest available FD >= `target`, so it never clobbers an
-/// existing FD.
-#[cfg(target_os = "linux")]
-fn prewarm_fd_table(target: i32) -> std::io::Result<()> {
-    use std::os::unix::io::{FromRawFd, OwnedFd};
-
-    let dev_null = std::fs::File::open("/dev/null")?;
-    let raw = unsafe {
-        libc::fcntl(
-            std::os::unix::io::AsRawFd::as_raw_fd(&dev_null),
-            libc::F_DUPFD_CLOEXEC,
-            target,
-        )
-    };
-    if raw < 0 {
-        return Err(std::io::Error::last_os_error());
-    }
-
-    // Close both FDs. The table capacity persists.
-    let _owned = unsafe { OwnedFd::from_raw_fd(raw) };
-    drop(dev_null);
-
-    Ok(())
-}
-
-/// Fully safe alternative using only stdlib. Requires O(n) syscalls instead of
-/// one, but avoids `unsafe` entirely.
-#[cfg(target_os = "linux")]
-#[allow(dead_code)]
-fn prewarm_fd_table_safe(target: i32) -> std::io::Result<()> {
-    let f = std::fs::File::open("/dev/null")?;
-    let _fds: Vec<_> = (0..target)
-        .map(|_| f.try_clone())
-        .collect::<Result<_, _>>()?;
-    Ok(())
-}
-
-fn main() {
-    #[cfg(target_os = "linux")]
-    {
-        const FD_TARGET: i32 = 10_000;
-
-        println!("Pre-warming FD table to {FD_TARGET} entries...");
-        if let Err(e) = prewarm_fd_table(FD_TARGET) {
-            eprintln!("Warning: failed to pre-warm FD table: {e}");
-        } else {
-            println!("FD table pre-warmed successfully.");
-        }
-    }
-
-    // Build the runtime *after* pre-warming.
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .build()
-        .unwrap();
-
-    rt.block_on(async {});
-}

examples/print_each_packet.rs

@@ -1,106 +0,0 @@
-//! A "print-each-packet" server with Tokio
-//!
-//! This server will create a TCP listener, accept connections in a loop, and
-//! put down in the stdout everything that's read off of each TCP connection.
-//!
-//! Because the Tokio runtime uses a thread pool, each TCP connection is
-//! processed concurrently with all other TCP connections across multiple
-//! threads.
-//!
-//! To see this server in action, you can run this in one terminal:
-//!
-//!     cargo run --example print\_each\_packet
-//!
-//! and in another terminal you can run:
-//!
-//!     cargo run --example connect-tcp 127.0.0.1:8080
-//!
-//! Each line you type in to the `connect-tcp` terminal should be written to terminal!
-//!
-//! Minimal js example:
-//!
-//! ```js
-//! var net = require("net");
-//!
-//! var listenPort = 8080;
-//!
-//! var server = net.createServer(function (socket) {
-//!     socket.on("data", function (bytes) {
-//!         console.log("bytes", bytes);
-//!     });
-//!
-//!     socket.on("end", function() {
-//!         console.log("Socket received FIN packet and closed connection");
-//!     });
-//!     socket.on("error", function (error) {
-//!         console.log("Socket closed with error", error);
-//!     });
-//!
-//!     socket.on("close", function (with_error) {
-//!         if (with_error) {
-//!             console.log("Socket closed with result: Err(SomeError)");
-//!         } else {
-//!             console.log("Socket closed with result: Ok(())");
-//!         }
-//!     });
-//!
-//! });
-//!
-//! server.listen(listenPort);
-//!
-//! console.log("Listening on:", listenPort);
-//! ```
-//!
-
-#![warn(rust_2018_idioms)]
-
-use tokio::net::TcpListener;
-use tokio_stream::StreamExt;
-use tokio_util::codec::{BytesCodec, Decoder};
-
-use std::env;
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    // Allow passing an address to listen on as the first argument of this
-    // program, but otherwise we'll just set up our TCP listener on
-    // 127.0.0.1:8080 for connections.
-    let addr = env::args()
-        .nth(1)
-        .unwrap_or_else(|| "127.0.0.1:8080".to_string());
-
-    // Next up we create a TCP listener which will listen for incoming
-    // connections. This TCP listener is bound to the address we determined
-    // above and must be associated with an event loop, so we pass in a handle
-    // to our event loop. After the socket's created we inform that we're ready
-    // to go and start accepting connections.
-    let listener = TcpListener::bind(&addr).await?;
-    println!("Listening on: {addr}");
-
-    loop {
-        // Asynchronously wait for an inbound socket.
-        let (socket, _) = listener.accept().await?;
-
-        // And this is where much of the magic of this server happens. We
-        // crucially want all clients to make progress concurrently, rather than
-        // blocking one on completion of another. To achieve this we use the
-        // `tokio::spawn` function to execute the work in the background.
-        //
-        // Essentially here we're executing a new task to run concurrently,
-        // which will allow all of our clients to be processed concurrently.
-        tokio::spawn(async move {
-            // We're parsing each socket with the `BytesCodec` included in `tokio::codec`.
-            let mut framed = BytesCodec::new().framed(socket);
-
-            // We loop while there are messages coming from the Stream `framed`.
-            // The stream will return None once the client disconnects.
-            while let Some(message) = framed.next().await {
-                match message {
-                    Ok(bytes) => println!("bytes: {bytes:?}"),
-                    Err(err) => println!("Socket closed with error: {err:?}"),
-                }
-            }
-            println!("Socket received FIN packet and closed connection");
-        });
-    }
-}

examples/proxy.rs

@@ -1,64 +0,0 @@
-//! A proxy that forwards data to another server and forwards that server's
-//! responses back to clients.
-//!
-//! Because the Tokio runtime uses a thread pool, each TCP connection is
-//! processed concurrently with all other TCP connections across multiple
-//! threads.
-//!
-//! You can showcase this by running this in one terminal:
-//!
-//!     cargo run --example proxy
-//!
-//! This in another terminal
-//!
-//!     cargo run --example echo-tcp
-//!
-//! And finally this in another terminal
-//!
-//!     cargo run --example connect-tcp 127.0.0.1:8081
-//!
-//! This final terminal will connect to our proxy, which will in turn connect to
-//! the echo server, and you'll be able to see data flowing between them.
-
-#![warn(rust_2018_idioms)]
-
-use tokio::io::copy_bidirectional;
-use tokio::net::{TcpListener, TcpStream};
-
-use std::env;
-use std::error::Error;
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    let listen_addr = env::args()
-        .nth(1)
-        .unwrap_or_else(|| "127.0.0.1:8081".to_string());
-    let server_addr = env::args()
-        .nth(2)
-        .unwrap_or_else(|| "127.0.0.1:8080".to_string());
-
-    println!("Listening on: {listen_addr}");
-    println!("Proxying to: {server_addr}");
-
-    let listener = TcpListener::bind(listen_addr).await?;
-
-    while let Ok((mut inbound, _)) = listener.accept().await {
-        let server_addr = server_addr.clone();
-
-        tokio::spawn(async move {
-            let mut outbound = match TcpStream::connect(server_addr).await {
-                Ok(outbound) => outbound,
-                Err(e) => {
-                    println!("Failed to connect; error={e}");
-                    return;
-                }
-            };
-
-            if let Err(e) = copy_bidirectional(&mut inbound, &mut outbound).await {
-                println!("Failed to transfer; error={e}");
-            }
-        });
-    }
-
-    Ok(())
-}

examples/tinydb.rs

@@ -1,224 +0,0 @@
-//! A "tiny database" and accompanying protocol
-//!
-//! This example shows the usage of shared state amongst all connected clients,
-//! namely a database of key/value pairs. Each connected client can send a
-//! series of GET/SET commands to query the current value of a key or set the
-//! value of a key.
-//!
-//! This example has a simple protocol you can use to interact with the server.
-//! To run, first run this in one terminal window:
-//!
-//!     cargo run --example tinydb
-//!
-//! and next in another windows run:
-//!
-//!     cargo run --example connect-tcp 127.0.0.1:8080
-//!
-//! In the `connect-tcp` window you can type in commands where when you hit enter
-//! you'll get a response from the server for that command. An example session
-//! is:
-//!
-//!
-//!     $ cargo run --example connect-tcp 127.0.0.1:8080
-//!     GET foo
-//!     foo = bar
-//!     GET FOOBAR
-//!     error: no key FOOBAR
-//!     SET FOOBAR my awesome string
-//!     set FOOBAR = `my awesome string`, previous: None
-//!     SET foo tokio
-//!     set foo = `tokio`, previous: Some("bar")
-//!     GET foo
-//!     foo = tokio
-//!
-//! Namely you can issue two forms of commands:
-//!
-//! * `GET $key` - this will fetch the value of `$key` from the database and
-//!   return it. The server's database is initially populated with the key `foo`
-//!   set to the value `bar`
-//! * `SET $key $value` - this will set the value of `$key` to `$value`,
-//!   returning the previous value, if any.
-
-#![warn(rust_2018_idioms)]
-
-use tokio::net::TcpListener;
-use tokio_stream::StreamExt;
-use tokio_util::codec::{Framed, LinesCodec};
-
-use futures::SinkExt;
-use std::collections::HashMap;
-use std::env;
-use std::error::Error;
-use std::sync::{Arc, Mutex};
-
-/// The in-memory database shared amongst all clients.
-///
-/// This database will be shared via `Arc`, so to mutate the internal map we're
-/// going to use a `Mutex` for interior mutability.
-struct Database {
-    map: Mutex<HashMap<String, String>>,
-}
-
-/// Possible requests our clients can send us
-enum Request {
-    Get { key: String },
-    Set { key: String, value: String },
-}
-
-/// Responses to the `Request` commands above
-enum Response {
-    Value {
-        key: String,
-        value: String,
-    },
-    Set {
-        key: String,
-        value: String,
-        previous: Option<String>,
-    },
-    Error {
-        msg: String,
-    },
-}
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    // Parse the address we're going to run this server on
-    // and set up our TCP listener to accept connections.
-    let addr = env::args()
-        .nth(1)
-        .unwrap_or_else(|| "127.0.0.1:8080".to_string());
-
-    let listener = TcpListener::bind(&addr).await?;
-    println!("Listening on: {addr}");
-
-    // Create the shared state of this server that will be shared amongst all
-    // clients. We populate the initial database and then create the `Database`
-    // structure. Note the usage of `Arc` here which will be used to ensure that
-    // each independently spawned client will have a reference to the in-memory
-    // database.
-    let mut initial_db = HashMap::new();
-    initial_db.insert("foo".to_string(), "bar".to_string());
-    let db = Arc::new(Database {
-        map: Mutex::new(initial_db),
-    });
-
-    loop {
-        match listener.accept().await {
-            Ok((socket, _)) => {
-                // After getting a new connection first we see a clone of the database
-                // being created, which is creating a new reference for this connected
-                // client to use.
-                let db = db.clone();
-
-                // Like with other small servers, we'll `spawn` this client to ensure it
-                // runs concurrently with all other clients. The `move` keyword is used
-                // here to move ownership of our db handle into the async closure.
-                tokio::spawn(async move {
-                    // Since our protocol is line-based we use `tokio_codecs`'s `LineCodec`
-                    // to convert our stream of bytes, `socket`, into a `Stream` of lines
-                    // as well as convert our line based responses into a stream of bytes.
-                    let mut lines = Framed::new(socket, LinesCodec::new());
-
-                    // Here for every line we get back from the `Framed` decoder,
-                    // we parse the request, and if it's valid we generate a response
-                    // based on the values in the database.
-                    while let Some(result) = lines.next().await {
-                        match result {
-                            Ok(line) => {
-                                let response = handle_request(&line, &db);
-
-                                let response = response.serialize();
-
-                                if let Err(e) = lines.send(response.as_str()).await {
-                                    println!("error on sending response; error = {e:?}");
-                                }
-                            }
-                            Err(e) => {
-                                println!("error on decoding from socket; error = {e:?}");
-                            }
-                        }
-                    }
-
-                    // The connection will be closed at this point as `lines.next()` has returned `None`.
-                });
-            }
-            Err(e) => println!("error accepting socket; error = {e:?}"),
-        }
-    }
-}
-
-fn handle_request(line: &str, db: &Arc<Database>) -> Response {
-    let request = match Request::parse(line) {
-        Ok(req) => req,
-        Err(e) => return Response::Error { msg: e },
-    };
-
-    let mut db = db.map.lock().unwrap();
-    match request {
-        Request::Get { key } => match db.get(&key) {
-            Some(value) => Response::Value {
-                key,
-                value: value.clone(),
-            },
-            None => Response::Error {
-                msg: format!("no key {key}"),
-            },
-        },
-        Request::Set { key, value } => {
-            let previous = db.insert(key.clone(), value.clone());
-            Response::Set {
-                key,
-                value,
-                previous,
-            }
-        }
-    }
-}
-
-impl Request {
-    fn parse(input: &str) -> Result<Request, String> {
-        let mut parts = input.splitn(3, ' ');
-        match parts.next() {
-            Some("GET") => {
-                let key = parts.next().ok_or("GET must be followed by a key")?;
-                if parts.next().is_some() {
-                    return Err("GET's key must not be followed by anything".into());
-                }
-                Ok(Request::Get {
-                    key: key.to_string(),
-                })
-            }
-            Some("SET") => {
-                let key = match parts.next() {
-                    Some(key) => key,
-                    None => return Err("SET must be followed by a key".into()),
-                };
-                let value = match parts.next() {
-                    Some(value) => value,
-                    None => return Err("SET needs a value".into()),
-                };
-                Ok(Request::Set {
-                    key: key.to_string(),
-                    value: value.to_string(),
-                })
-            }
-            Some(cmd) => Err(format!("unknown command: {cmd}")),
-            None => Err("empty input".into()),
-        }
-    }
-}
-
-impl Response {
-    fn serialize(&self) -> String {
-        match *self {
-            Response::Value { ref key, ref value } => format!("{key} = {value}"),
-            Response::Set {
-                ref key,
-                ref value,
-                ref previous,
-            } => format!("set {key} = `{value}`, previous: {previous:?}"),
-            Response::Error { ref msg } => format!("error: {msg}"),
-        }
-    }
-}

examples/udp-codec.rs

@@ -1,80 +0,0 @@
-//! This example leverages `BytesCodec` to create a UDP client and server which
-//! speak a custom protocol.
-//!
-//! Here we're using the codec from `tokio-codec` to convert a UDP socket to a stream of
-//! client messages. These messages are then processed and returned back as a
-//! new message with a new destination. Overall, we then use this to construct a
-//! "ping pong" pair where two sockets are sending messages back and forth.
-
-#![warn(rust_2018_idioms)]
-
-use tokio::net::UdpSocket;
-use tokio::{io, time};
-use tokio_stream::StreamExt;
-use tokio_util::codec::BytesCodec;
-use tokio_util::udp::UdpFramed;
-
-use bytes::Bytes;
-use futures::{FutureExt, SinkExt};
-use std::env;
-use std::error::Error;
-use std::net::SocketAddr;
-use std::time::Duration;
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    let addr = env::args()
-        .nth(1)
-        .unwrap_or_else(|| "127.0.0.1:0".to_string());
-
-    // Bind both our sockets and then figure out what ports we got.
-    let a = UdpSocket::bind(&addr).await?;
-    let b = UdpSocket::bind(&addr).await?;
-
-    let b_addr = b.local_addr()?;
-
-    let mut a = UdpFramed::new(a, BytesCodec::new());
-    let mut b = UdpFramed::new(b, BytesCodec::new());
-
-    // Start off by sending a ping from a to b, afterwards we just print out
-    // what they send us and continually send pings
-    let a = ping(&mut a, b_addr);
-
-    // The second client we have will receive the pings from `a` and then send
-    // back pongs.
-    let b = pong(&mut b);
-
-    // Run both futures simultaneously of `a` and `b` sending messages back and forth.
-    match tokio::try_join!(a, b) {
-        Err(e) => println!("an error occurred; error = {e:?}"),
-        _ => println!("done!"),
-    }
-
-    Ok(())
-}
-
-async fn ping(socket: &mut UdpFramed<BytesCodec>, b_addr: SocketAddr) -> Result<(), io::Error> {
-    socket.send((Bytes::from(&b"PING"[..]), b_addr)).await?;
-
-    for _ in 0..4usize {
-        let (bytes, addr) = socket.next().map(|e| e.unwrap()).await?;
-
-        println!("[a] recv: {}", String::from_utf8_lossy(&bytes));
-
-        socket.send((Bytes::from(&b"PING"[..]), addr)).await?;
-    }
-
-    Ok(())
-}
-
-async fn pong(socket: &mut UdpFramed<BytesCodec>) -> Result<(), io::Error> {
-    let timeout = Duration::from_millis(200);
-
-    while let Ok(Some(Ok((bytes, addr)))) = time::timeout(timeout, socket.next()).await {
-        println!("[b] recv: {}", String::from_utf8_lossy(&bytes));
-
-        socket.send((Bytes::from(&b"PONG"[..]), addr)).await?;
-    }
-
-    Ok(())
-}

netlify.toml

@@ -1,19 +0,0 @@
-[build]
-  command = """
-    rustup install nightly --profile minimal && cargo doc --no-deps --all-features
-    """
-  publish = "target/doc"
-
-[build.environment]
-  RUSTDOCFLAGS="""
-    --cfg docsrs \
-    --cfg tokio_unstable \
-    """
-  RUSTFLAGS="""
-    --cfg docsrs \
-    --cfg tokio_unstable
-    """
-
-[[redirects]]
-  from = "/"
-  to = "/tokio"

spellcheck.dic

@@ -1,328 +0,0 @@
-327
-&
-+
-<
-=
->
-\
-~
-—
-0o777
-0s
-0xA
-0xD
-100ms
-100ns
-10ms
-10μs
-~12
-120s
-12.5%
-±1m
-±1ms
-1ms
-1s
-25%
-250ms
-2x
-~4
-443
-450ms
-50ms
-8MB
-ABI
-accessors
-adaptor
-adaptors
-Adaptors
-AIO
-ambient
-amongst
-api
-APIs
-async
-atomics
-awaitable
-backend
-backpressure
-backtrace
-backtraces
-backtracing
-binded
-bitfield
-bitfields
-Blockingly
-boolean
-broadcasted
-cancelled
-cancelling
-Cancelling
-CLI
-cloneable
-codebase
-codec
-codecs
-combinator
-combinators
-condvar
-config
-Config
-connectionless
-coroutines
-cpu
-cpus
-cqe
-CQE
-cqe's
-customizable
-Customizable
-Cygwin
-datagram
-Datagram
-datagrams
-deallocate
-deallocated
-Deallocates
-debuginfo
-decrement
-decrementing
-demangled
-dequeued
-dereferenced
-derefs
-deregister
-deregistered
-deregistering
-deregisters
-Deregisters
-deregistration
-descriptor's
-destructor
-destructors
-destructure
-Destructures
-Dev
-dns
-DNS
-DoS
-dwOpenMode
-endian
-enqueue
-enqueued
-EntryInner
-enum
-eof
-Erlang's
-errored
-EWMA
-expirations
-fcntl
-fd
-fd's
-FIFOs
-filename
-filesystem
-filesystems
-fn
-fns
-FreeBSD
-frontend
-fs
-functionalities
-getters
-GID
-goroutines
-Growable
-gzip
-H2
-hashmaps
-HashMaps
-hashsets
-HdrHistogram
-ICMP
-ie
-iff
-Illumos
-impl
-implementers
-implementor
-implementors
-incrementing
-inlining
-interoperate
-invariants
-Invariants
-io
-IOCP
-iOS
-IOs
-io_uring
-IP
-IPv4
-IPv6
-iteratively
-Kotlin's
-latencies
-Lauck
-libc
-lifecycle
-lifo
-LLVM
-lookups
-macOS
-MacOS
-Marsaglia's
-metadata
-mio
-Mio
-mio's
-miri
-misconfigured
-mock's
-monomorphization
-mpmc
-mpsc
-multi
-Multi
-multicast
-Multithreaded
-mut
-mutex
-Mutex
-Nagle
-namespace
-NetBSD
-nonblocking
-nondecreasing
-noop
-ntasks
-NTO
-NUMA
-ok
-oneshot
-opcode
-OpenBSD
-ORed
-os
-parker
-parsers
-peekable
-PGID
-PID
-plaintext
-poller
-POSIX
-proxied
-qos
-RAII
-RCU
-reallocations
-recv's
-Redox
-refactors
-refcount
-refcounting
-repo
-repr
-representable
-reqwest
-reregistering
-resize
-resized
-RMW
-RNG
-runtime
-runtime's
-runtimes
-rwlock
-rx
-scalability
-scheduler's
-semver
-setpgid
-sharded
-signalled
-signalling
-SmallCrush
-Solaris
-spawner
-spawners
-Splitter
-spmc
-spsc
-SQE
-src
-stabilised
-startup
-stateful
-stderr
-stdin
-stdout
-stealable
-stealer
-stealers
-struct
-structs
-subfield
-suboptimal
-subprocess
-superset
-symlink
-symlinks
-sys
-syscall
-syscalls
-tcp
-TCP
-TestU01
-threadpool
-timestamp
-timestamps
-TLS
-TOCTOU
-TODO
-tokio
-Tokio
-tokio's
-Tokio's
-tuple
-Tuple
-tvOS
-tx
-udp
-UDP
-UID
-unhandled
-unix
-unlink
-unpark
-Unpark
-unparked
-unparking
-unparks
-Unparks
-unreceived
-unsafety
-unsets
-Unsets
-unsynchronized
-untrusted
-uring
-usecases
-Valgrind
-Varghese
-vec
-versa
-versioned
-versioning
-visionOS
-vtable
-waker
-wakers
-Wakers
-wakeup
-wakeups
-WASI
-watchOS
-workstealing
-ZST

spellcheck.toml

@@ -1,13 +0,0 @@
-dev_comments = false
-skip_readme = false
-
-[Hunspell]
-lang = "en_US"
-search_dirs = ["."]
-extra_dictionaries = ["spellcheck.dic"]
-skip_os_lookups = true
-use_builtin = true
-
-[Hunspell.quirks]
-allow_concatenation = true
-

stress-test/Cargo.toml

@@ -1,18 +0,0 @@
-[package]
-name = "stress-test"
-version = "0.1.0"
-authors = ["Tokio Contributors <[email protected]>"]
-edition = "2021"
-license = "MIT"
-publish = false
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
-tokio = { version = "1.0.0", path = "../tokio/", features = ["full"] }
-
-[dev-dependencies]
-rand = "0.9"
-
-[lints]
-workspace = true

stress-test/examples/simple_echo_tcp.rs

@@ -1,58 +0,0 @@
-//! Simple TCP echo server to check memory leaks using Valgrind.
-use std::{thread::sleep, time::Duration};
-
-use tokio::{
-    io::{AsyncReadExt, AsyncWriteExt},
-    net::{TcpListener, TcpSocket},
-    runtime::Builder,
-    sync::oneshot,
-};
-
-const TCP_ENDPOINT: &str = "127.0.0.1:8080";
-const NUM_MSGS: usize = 100;
-const MSG_SIZE: usize = 1024;
-
-fn main() {
-    let rt = Builder::new_multi_thread().enable_io().build().unwrap();
-    let rt2 = Builder::new_multi_thread().enable_io().build().unwrap();
-
-    rt.spawn(async {
-        let listener = TcpListener::bind(TCP_ENDPOINT).await.unwrap();
-        let (mut socket, _) = listener.accept().await.unwrap();
-        let (mut rd, mut wr) = socket.split();
-        while tokio::io::copy(&mut rd, &mut wr).await.is_ok() {}
-    });
-
-    // wait a bit so that the listener binds.
-    sleep(Duration::from_millis(100));
-
-    // create a channel to let the main thread know that all the messages were sent and received.
-    let (tx, mut rx) = oneshot::channel();
-
-    rt2.spawn(async {
-        let addr = TCP_ENDPOINT.parse().unwrap();
-        let socket = TcpSocket::new_v4().unwrap();
-        let mut stream = socket.connect(addr).await.unwrap();
-
-        let mut buff = [0; MSG_SIZE];
-        for _ in 0..NUM_MSGS {
-            let one_mega_random_bytes: Vec<u8> =
-                (0..MSG_SIZE).map(|_| rand::random::<u8>()).collect();
-            stream
-                .write_all(one_mega_random_bytes.as_slice())
-                .await
-                .unwrap();
-            let _ = stream.read(&mut buff).await.unwrap();
-        }
-        tx.send(()).unwrap();
-    });
-
-    loop {
-        // check that we're done.
-        match rx.try_recv() {
-            Err(oneshot::error::TryRecvError::Empty) => (),
-            Err(oneshot::error::TryRecvError::Closed) => panic!("channel got closed..."),
-            Ok(()) => break,
-        }
-    }
-}

target-specs/README.md

@@ -1,9 +0,0 @@
-This is used for the `no-atomic-u64-test` ci check that verifies that Tokio
-works even if the `AtomicU64` type is missing.
-
-When increasing the nightly compiler version, you may need to regenerate this
-target using the following command:
-```
-rustc +nightly -Z unstable-options --print target-spec-json --target i686-unknown-linux-gnu | grep -v 'is-builtin' | sed 's/"max-atomic-width": 64/"max-atomic-width": 32/' > target-specs/i686-unknown-linux-gnu.json
-```
-

target-specs/i686-unknown-linux-gnu.json

@@ -1,48 +0,0 @@
-{
-  "arch": "x86",
-  "cpu": "pentium4",
-  "crt-objects-fallback": "false",
-  "crt-static-respected": true,
-  "data-layout": "e-m:e-p:32:32-p270:32:32-p271:32:32-p272:64:64-i128:128-f64:32:64-f80:32-n8:16:32-S128",
-  "default-uwtable": true,
-  "dynamic-linking": true,
-  "env": "gnu",
-  "has-rpath": true,
-  "has-thread-local": true,
-  "linker-flavor": "gnu-cc",
-  "llvm-target": "i686-unknown-linux-gnu",
-  "max-atomic-width": 32,
-  "metadata": {
-    "description": "32-bit Linux (kernel 3.2, glibc 2.17+)",
-    "host_tools": true,
-    "std": true,
-    "tier": 1
-  },
-  "os": "linux",
-  "position-independent-executables": true,
-  "pre-link-args": {
-    "gnu-cc": [
-      "-m32"
-    ],
-    "gnu-lld-cc": [
-      "-m32"
-    ]
-  },
-  "relro-level": "full",
-  "rustc-abi": "x86-sse2",
-  "stack-probes": {
-    "kind": "inline"
-  },
-  "supported-sanitizers": [
-    "address"
-  ],
-  "supported-split-debuginfo": [
-    "packed",
-    "unpacked",
-    "off"
-  ],
-  "target-family": [
-    "unix"
-  ],
-  "target-pointer-width": 32
-}

tests-build/Cargo.toml

@@ -1,20 +0,0 @@
-[package]
-name = "tests-build"
-version = "0.1.0"
-authors = ["Tokio Contributors <[email protected]>"]
-edition = "2021"
-license = "MIT"
-publish = false
-
-[features]
-full = ["tokio/full"]
-rt = ["tokio/rt", "tokio/macros"]
-
-[dependencies]
-tokio = { version = "1.0.0", path = "../tokio", optional = true }
-
-[dev-dependencies]
-trybuild = "1.0"
-
-[lints]
-workspace = true

tests-build/README.md

@@ -1,10 +0,0 @@
-Tests the various combination of feature flags. This is broken out to a separate
-crate to work around limitations with cargo features.
-
-To run all of the tests in this directory, run the following commands:
-```
-cargo test --features full
-cargo test --features rt
-```
-If any of the tests fail, you can pass `TRYBUILD=overwrite` to the `cargo test`
-command that failed to have it regenerate the test output.

tests-build/src/lib.rs

@@ -1,2 +0,0 @@
-#[cfg(feature = "tokio")]
-pub use tokio;

tests-build/tests/fail/macros_core_no_default.rs

@@ -1,6 +0,0 @@
-use tests_build::tokio;
-
-#[tokio::main]
-async fn my_fn() {}
-
-fn main() {}

tests-build/tests/fail/macros_core_no_default.stderr

@@ -1,7 +0,0 @@
-error: The default runtime flavor is `multi_thread`, but the `rt-multi-thread` feature is disabled.
- --> $DIR/macros_core_no_default.rs:3:1
-  |
-3 | #[tokio::main]
-  | ^^^^^^^^^^^^^^
-  |
-  = note: this error originates in the attribute macro `tokio::main` (in Nightly builds, run with -Z macro-backtrace for more info)

tests-build/tests/fail/macros_dead_code.rs

@@ -1,8 +0,0 @@
-#![deny(dead_code)]
-
-use tests_build::tokio;
-
-#[tokio::main]
-async fn f() {}
-
-fn main() {}