stream|tokio: Port proptest fuzz harnesses to use cargo-fuzz

This change ports fuzz tests from the black-box fuzzing framework,
proptest-rs over to use the grey-box fuzzing framework cargo-fuzz.

Refs: #5391

CONTRIBUTING.md

@@ -187,7 +187,7 @@ LOOM_MAX_PREEMPTIONS=1 RUSTFLAGS="--cfg loom" \
 
 You can run miri tests with
 ```
-MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" PROPTEST_CASES=10 \
+MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" \
     cargo +nightly miri test --features full --lib
 ```
 
@@ -209,6 +209,31 @@ 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 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

tokio-stream/Cargo.toml

@@ -38,9 +38,6 @@ parking_lot = "0.12.0"
 tokio-test = { path = "../tokio-test" }
 futures = { version = "0.3", default-features = false }
 
-[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
-proptest = "1"
-
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]

tokio-stream/fuzz/.gitignore

@@ -0,0 +1,4 @@
+target
+corpus
+artifacts
+coverage

tokio-stream/fuzz/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "tokio-stream-fuzz"
+version = "0.0.0"
+publish = false
+edition = "2018"
+
+[package.metadata]
+cargo-fuzz = true
+
+[dependencies]
+libfuzzer-sys = "0.4"
+tokio-test = { path = "../../tokio-test" }
+
+[dependencies.tokio-stream]
+path = ".."
+
+
+# Prevent this from interfering with workspaces
+[workspace]
+members = ["."]
+
+[profile.release]
+debug = 1
+
+[[bin]]
+name = "fuzz_stream_map"
+path = "fuzz_targets/fuzz_stream_map.rs"
+test = false
+doc = false

tokio-stream/fuzz/fuzz_targets/fuzz_stream_map.rs

@@ -0,0 +1,80 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+use std::pin::Pin;
+
+use tokio_stream::{self as stream, pending, Stream, StreamExt, StreamMap};
+use tokio_test::{assert_ok, assert_pending, assert_ready, task};
+
+macro_rules! assert_ready_some {
+    ($($t:tt)*) => {
+        match assert_ready!($($t)*) {
+            Some(v) => v,
+            None => panic!("expected `Some`, got `None`"),
+        }
+    };
+}
+
+macro_rules! assert_ready_none {
+    ($($t:tt)*) => {
+        match assert_ready!($($t)*) {
+            None => {}
+            Some(v) => panic!("expected `None`, got `Some({:?})`", v),
+        }
+    };
+}
+
+fn pin_box<T: Stream<Item = U> + 'static, U>(s: T) -> Pin<Box<dyn Stream<Item = U>>> {
+    Box::pin(s)
+}
+
+fuzz_target!(|data: &[u8]| {
+    use std::task::{Context, Poll};
+
+    struct DidPoll<T> {
+        did_poll: bool,
+        inner: T,
+    }
+
+    impl<T: Stream + Unpin> Stream for DidPoll<T> {
+        type Item = T::Item;
+
+        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T::Item>> {
+            self.did_poll = true;
+            Pin::new(&mut self.inner).poll_next(cx)
+        }
+    }
+
+    for _ in 0..10 {
+        let mut map = task::spawn(StreamMap::new());
+        let mut expect = 0;
+
+        for (i, is_empty) in data.iter().map(|x| *x != 0).enumerate() {
+            let inner = if is_empty {
+                pin_box(stream::empty::<()>())
+            } else {
+                expect += 1;
+                pin_box(stream::pending::<()>())
+            };
+
+            let stream = DidPoll {
+                did_poll: false,
+                inner,
+            };
+
+            map.insert(i, stream);
+        }
+
+        if expect == 0 {
+            assert_ready_none!(map.poll_next());
+        } else {
+            assert_pending!(map.poll_next());
+
+            assert_eq!(expect, map.values().count());
+
+            for stream in map.values() {
+                assert!(stream.did_poll);
+            }
+        }
+    }
+});

tokio-stream/tests/stream_stream_map.rs

@@ -325,63 +325,6 @@ fn one_ready_many_none() {
     }
 }
 
-#[cfg(not(target_os = "wasi"))]
-proptest::proptest! {
-    #[test]
-    fn fuzz_pending_complete_mix(kinds: Vec<bool>) {
-        use std::task::{Context, Poll};
-
-        struct DidPoll<T> {
-            did_poll: bool,
-            inner: T,
-        }
-
-        impl<T: Stream + Unpin> Stream for DidPoll<T> {
-            type Item = T::Item;
-
-            fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>)
-                -> Poll<Option<T::Item>>
-            {
-                self.did_poll = true;
-                Pin::new(&mut self.inner).poll_next(cx)
-            }
-        }
-
-        for _ in 0..10 {
-            let mut map = task::spawn(StreamMap::new());
-            let mut expect = 0;
-
-            for (i, &is_empty) in kinds.iter().enumerate() {
-                let inner = if is_empty {
-                    pin_box(stream::empty::<()>())
-                } else {
-                    expect += 1;
-                    pin_box(stream::pending::<()>())
-                };
-
-                let stream = DidPoll {
-                    did_poll: false,
-                    inner,
-                };
-
-                map.insert(i, stream);
-            }
-
-            if expect == 0 {
-                assert_ready_none!(map.poll_next());
-            } else {
-                assert_pending!(map.poll_next());
-
-                assert_eq!(expect, map.values().count());
-
-                for stream in map.values() {
-                    assert!(stream.did_poll);
-                }
-            }
-        }
-    }
-}
-
 fn pin_box<T: Stream<Item = U> + 'static, U>(s: T) -> Pin<Box<dyn Stream<Item = U>>> {
     Box::pin(s)
 }

tokio/Cargo.toml

@@ -147,7 +147,6 @@ tempfile = "3.1.0"
 async-stream = "0.3"
 
 [target.'cfg(not(any(target_arch = "wasm32", target_arch = "wasm64")))'.dev-dependencies]
-proptest = "1"
 socket2 = "0.4"
 
 [target.'cfg(not(all(any(target_arch = "wasm32", target_arch = "wasm64"), target_os = "unknown")))'.dev-dependencies]

tokio/fuzz/.gitignore

@@ -0,0 +1,4 @@
+target
+corpus
+artifacts
+coverage

tokio/fuzz/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "tokio-fuzz"
+version = "0.0.0"
+publish = false
+edition = "2018"
+
+[package.metadata]
+cargo-fuzz = true
+
+[dependencies]
+libfuzzer-sys = "0.4"
+
+[dependencies.tokio]
+path = ".."
+features = ["fs","net","process","rt","sync","signal","time"]
+
+
+# Prevent this from interfering with workspaces
+[workspace]
+members = ["."]
+
+[profile.release]
+debug = 1
+
+[[bin]]
+name = "fuzz_linked_list"
+path = "fuzz_targets/fuzz_linked_list.rs"
+test = false
+doc = false

tokio/fuzz/fuzz_targets/fuzz_linked_list.rs

@@ -0,0 +1,7 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+
+fuzz_target!(|data: &[u8]| {
+    tokio::fuzz::fuzz_linked_list(data);
+});

tokio/src/fuzz.rs

@@ -0,0 +1 @@
+pub use crate::util::linked_list::tests::fuzz_linked_list;

tokio/src/lib.rs

@@ -631,3 +631,6 @@ cfg_macros! {
 #[cfg(feature = "io-util")]
 #[cfg(test)]
 fn is_unpin<T: Unpin>() {}
+
+#[cfg(fuzzing)]
+pub mod fuzz;

tokio/src/util/linked_list.rs

@@ -352,9 +352,9 @@ impl<T> fmt::Debug for Pointers<T> {
     }
 }
 
-#[cfg(test)]
+#[cfg(any(test, fuzzing))]
 #[cfg(not(loom))]
-mod tests {
+pub(crate) mod tests {
     use super::*;
 
     use std::pin::Pin;
@@ -623,31 +623,21 @@ mod tests {
         }
     }
 
-    #[cfg(not(tokio_wasm))]
-    proptest::proptest! {
-        #[test]
-        fn fuzz_linked_list(ops: Vec<usize>) {
-            run_fuzz(ops);
-        }
-    }
-
-    #[cfg(not(tokio_wasm))]
-    fn run_fuzz(ops: Vec<usize>) {
-        use std::collections::VecDeque;
-
-        #[derive(Debug)]
+    #[cfg(fuzzing)]
+    pub fn fuzz_linked_list(ops: &[u8]) {
         enum Op {
             Push,
             Pop,
             Remove(usize),
         }
+        use std::collections::VecDeque;
 
         let ops = ops
             .iter()
-            .map(|i| match i % 3 {
+            .map(|i| match i % 3u8 {
                 0 => Op::Push,
                 1 => Op::Pop,
-                2 => Op::Remove(i / 3),
+                2 => Op::Remove((i / 3u8) as usize),
                 _ => unreachable!(),
             })
             .collect::<Vec<_>>();