Add async/await examples from tokio's test to a new "slow" test suite (rust-lang#1332)

* Make tokio tests compile with Kani

* Add expected file

* Lower unwind bound

* Raise unwind bounds for some tests

* Disable tests that hit bugs or take too long

* Disable failing tests

* Add a "slow" test suite

* Add Github action to run slow tests every night

* Fix tests and expected

* Address review comments

* Rerun CI

Co-authored-by: Fabian Zaiser <[email protected]>

Author: fzaiser

.github/workflows/slow-tests.yml

@@ -0,0 +1,30 @@
+# Copyright Kani Contributors
+# SPDX-License-Identifier: Apache-2.0 OR MIT
+name: Kani's slow tests
+on:
+  schedule:
+    - cron: "30 5 * * *" # Run this every day at 05:30 UTC
+
+env:
+  RUST_BACKTRACE: 1
+
+jobs:
+  regression:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [macos-11, ubuntu-18.04, ubuntu-20.04]
+    steps:
+      - name: Checkout Kani
+        uses: actions/checkout@v2
+
+      - name: Setup Kani Dependencies
+        uses: ./.github/actions/setup
+        with:
+          os: ${{ matrix.os }}
+
+      - name: Build Kani
+        run: cargo build
+
+      - name: Run Kani's slow tests
+        run: ./scripts/kani-slow-tests.sh

Cargo.toml

@@ -64,4 +64,5 @@ exclude = [
   "tests/cargo-kani",
   "tests/perf",
   "tests/cargo-ui",
+  "tests/slow",
 ]

scripts/kani-slow-tests.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+# Copyright Kani Contributors
+# SPDX-License-Identifier: Apache-2.0 OR MIT
+
+if [[ -z $KANI_REGRESSION_KEEP_GOING ]]; then
+  set -o errexit
+fi
+set -o pipefail
+set -o nounset
+
+SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+export PATH=$SCRIPT_DIR:$PATH
+EXTRA_X_PY_BUILD_ARGS="${EXTRA_X_PY_BUILD_ARGS:-}"
+KANI_DIR=$SCRIPT_DIR/..
+
+# This variable forces an error when there is a mismatch on the expected
+# descriptions from cbmc checks.
+# TODO: We should add a more robust mechanism to detect python unexpected behavior.
+export KANI_FAIL_ON_UNEXPECTED_DESCRIPTION="true"
+
+# Build all packages in the workspace
+cargo build
+
+# Run slow compiletests
+cargo run -p compiletest --quiet -- --suite slow --mode cargo-kani
+
+echo
+echo "Kani slow tests completed successfully."
+echo

tests/slow/tokio-proofs/Cargo.toml

@@ -0,0 +1,21 @@
+# Copyright Kani Contributors
+# SPDX-License-Identifier: Apache-2.0 OR MIT
+[package]
+name = "kani_tokio"
+version = "0.1.0"
+edition = "2021"
+
+[features]
+default = ["full"]
+full = []
+
+[dependencies]
+tokio = { version = "1.20", features = ["full"] }
+tokio-test = "0.4.0"
+tokio-stream = "0.1"
+futures = { version = "0.3.0", features = ["async-await"] }
+bytes = "1.2.1"
+tokio-util = { version = "0.7.3", features = ["io"] }
+async-stream = "0.3.3"
+# mockall = "0.11.1"
+# async-stream = "0.3"

tests/slow/tokio-proofs/expected

@@ -0,0 +1 @@
+Complete - 13 successfully verified harnesses, 0 failures, 13 total.

tests/slow/tokio-proofs/src/lib.rs

@@ -0,0 +1,7 @@
+// Copyright Kani Contributors
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+
+mod tokio;
+mod tokio_stream;
+mod tokio_test;
+mod tokio_util;

tests/slow/tokio-proofs/src/tokio/io_chain.rs

@@ -0,0 +1,26 @@
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+//
+// Modifications Copyright Kani Contributors
+// See GitHub history for details.
+
+// Original copyright tokio contributors.
+// origin: tokio/tests/tokio/ at commit b2ada60e701d5c9e6644cf8fc42a100774f8e23f
+
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use tokio::io::AsyncReadExt;
+use tokio_test::assert_ok;
+
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn chain() {
+    let mut buf = Vec::new();
+    let rd1: &[u8] = b"hello ";
+    let rd2: &[u8] = b"world";
+
+    let mut rd = rd1.chain(rd2);
+    assert_ok!(rd.read_to_end(&mut buf).await);
+    assert_eq!(buf, b"hello world");
+}

tests/slow/tokio-proofs/src/tokio/io_copy.rs

@@ -0,0 +1,96 @@
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+//
+// Modifications Copyright Kani Contributors
+// See GitHub history for details.
+
+// Original copyright tokio contributors.
+// origin: tokio/tests/tokio/ at commit b2ada60e701d5c9e6644cf8fc42a100774f8e23f
+
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use bytes::BytesMut;
+use futures::ready;
+use tokio::io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
+use tokio_test::assert_ok;
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn copy() {
+    struct Rd(bool);
+
+    impl AsyncRead for Rd {
+        fn poll_read(
+            mut self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            buf: &mut ReadBuf<'_>,
+        ) -> Poll<io::Result<()>> {
+            if self.0 {
+                buf.put_slice(b"hello world");
+                self.0 = false;
+                Poll::Ready(Ok(()))
+            } else {
+                Poll::Ready(Ok(()))
+            }
+        }
+    }
+
+    let mut rd = Rd(true);
+    let mut wr = Vec::new();
+
+    let n = assert_ok!(io::copy(&mut rd, &mut wr).await);
+    assert_eq!(n, 11);
+    assert_eq!(wr, b"hello world");
+}
+
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn proxy() {
+    struct BufferedWd {
+        buf: BytesMut,
+        writer: io::DuplexStream,
+    }
+
+    impl AsyncWrite for BufferedWd {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            buf: &[u8],
+        ) -> Poll<io::Result<usize>> {
+            self.get_mut().buf.extend_from_slice(buf);
+            Poll::Ready(Ok(buf.len()))
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            let this = self.get_mut();
+
+            while !this.buf.is_empty() {
+                let n = ready!(Pin::new(&mut this.writer).poll_write(cx, &this.buf))?;
+                let _ = this.buf.split_to(n);
+            }
+
+            Pin::new(&mut this.writer).poll_flush(cx)
+        }
+
+        fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Pin::new(&mut self.writer).poll_shutdown(cx)
+        }
+    }
+
+    let (rd, wd) = io::duplex(1024);
+    let mut rd = rd.take(1024);
+    let mut wd = BufferedWd { buf: BytesMut::new(), writer: wd };
+
+    // write start bytes
+    assert_ok!(wd.write_all(&[0x42; 512]).await);
+    assert_ok!(wd.flush().await);
+
+    let n = assert_ok!(io::copy(&mut rd, &mut wd).await);
+
+    assert_eq!(n, 1024);
+}

tests/slow/tokio-proofs/src/tokio/io_lines.rs

@@ -0,0 +1,29 @@
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+//
+// Modifications Copyright Kani Contributors
+// See GitHub history for details.
+
+// Original copyright tokio contributors.
+// origin: tokio/tests/tokio/ at commit b2ada60e701d5c9e6644cf8fc42a100774f8e23f
+
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use tokio::io::AsyncBufReadExt;
+use tokio_test::assert_ok;
+
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn lines_inherent() {
+    let rd: &[u8] = b"hello\r\nworld\n\n";
+    let mut st = rd.lines();
+
+    let b = assert_ok!(st.next_line().await).unwrap();
+    assert_eq!(b, "hello");
+    let b = assert_ok!(st.next_line().await).unwrap();
+    assert_eq!(b, "world");
+    let b = assert_ok!(st.next_line().await).unwrap();
+    assert_eq!(b, "");
+    assert!(assert_ok!(st.next_line().await).is_none());
+}

tests/slow/tokio-proofs/src/tokio/io_mem_stream.rs

@@ -0,0 +1,146 @@
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+//
+// Modifications Copyright Kani Contributors
+// See GitHub history for details.
+
+// Original copyright tokio contributors.
+// origin: tokio/tests/tokio/ at commit b2ada60e701d5c9e6644cf8fc42a100774f8e23f
+
+#![warn(rust_2018_idioms)]
+#![cfg(feature = "full")]
+
+use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
+
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn ping_pong() {
+    let (mut a, mut b) = duplex(32);
+
+    let mut buf = [0u8; 4];
+
+    a.write_all(b"ping").await.unwrap();
+    b.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, b"ping");
+
+    b.write_all(b"pong").await.unwrap();
+    a.read_exact(&mut buf).await.unwrap();
+    assert_eq!(&buf, b"pong");
+}
+
+// Kani does not support this one yet because it uses spawn
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn across_tasks() {
+    let (mut a, mut b) = duplex(32);
+
+    let t1 = tokio::spawn(async move {
+        a.write_all(b"ping").await.unwrap();
+        let mut buf = [0u8; 4];
+        a.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf, b"pong");
+    });
+
+    let t2 = tokio::spawn(async move {
+        let mut buf = [0u8; 4];
+        b.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf, b"ping");
+        b.write_all(b"pong").await.unwrap();
+    });
+
+    t1.await.unwrap();
+    t2.await.unwrap();
+}
+
+// Kani does not support this one yet because it uses spawn
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn disconnect() {
+    let (mut a, mut b) = duplex(32);
+
+    let t1 = tokio::spawn(async move {
+        a.write_all(b"ping").await.unwrap();
+        // and dropped
+    });
+
+    let t2 = tokio::spawn(async move {
+        let mut buf = [0u8; 32];
+        let n = b.read(&mut buf).await.unwrap();
+        assert_eq!(&buf[..n], b"ping");
+
+        let n = b.read(&mut buf).await.unwrap();
+        assert_eq!(n, 0);
+    });
+
+    t1.await.unwrap();
+    t2.await.unwrap();
+}
+
+// Kani does not support this one yet because it uses spawn
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn disconnect_reader() {
+    let (a, mut b) = duplex(2);
+
+    let t1 = tokio::spawn(async move {
+        // this will block, as not all data fits into duplex
+        b.write_all(b"ping").await.unwrap_err();
+    });
+
+    let t2 = tokio::spawn(async move {
+        // here we drop the reader side, and we expect the writer in the other
+        // task to exit with an error
+        drop(a);
+    });
+
+    t2.await.unwrap();
+    t1.await.unwrap();
+}
+
+// Kani does not support this one yet because it uses spawn
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn max_write_size() {
+    let (mut a, mut b) = duplex(32);
+
+    let t1 = tokio::spawn(async move {
+        let n = a.write(&[0u8; 64]).await.unwrap();
+        assert_eq!(n, 32);
+        let n = a.write(&[0u8; 64]).await.unwrap();
+        assert_eq!(n, 4);
+    });
+
+    let mut buf = [0u8; 4];
+    b.read_exact(&mut buf).await.unwrap();
+
+    t1.await.unwrap();
+
+    // drop b only after task t1 finishes writing
+    drop(b);
+}
+
+// Kani does not support this one yet because it uses select
+#[cfg(disabled)] // because it timed out after 2h
+#[kani::proof]
+#[kani::unwind(2)]
+async fn duplex_is_cooperative() {
+    let (mut tx, mut rx) = tokio::io::duplex(1024 * 8);
+
+    tokio::select! {
+        biased;
+
+        _ = async {
+            loop {
+                let buf = [3u8; 4096];
+                tx.write_all(&buf).await.unwrap();
+                let mut buf = [0u8; 4096];
+                rx.read(&mut buf).await.unwrap();
+            }
+        } => {},
+        _ = tokio::task::yield_now() => {}
+    }
+}