ci: enable ASAN in CI (#4696)

.github/workflows/ci.yml

@@ -11,7 +11,7 @@ env:
   RUST_BACKTRACE: 1
   # Change to specific Rust release to pin
   rust_stable: stable
-  rust_nightly: nightly-2022-03-21
+  rust_nightly: nightly-2022-04-18
   rust_clippy: 1.52.0
   # When updating this, also update:
   # - README.md
@@ -207,23 +207,26 @@ jobs:
           MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-tag-raw-pointers
           PROPTEST_CASES: 10
 
-  san:
-    name: san
+  asan:
+    name: asan
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
+      - name: Install llvm
+        # Required to resolve symbols in sanitizer output
+        run: sudo apt-get install -y llvm
       - name: Install Rust ${{ env.rust_nightly }}
         uses: actions-rs/toolchain@v1
         with:
           toolchain: ${{ env.rust_nightly }}
           override: true
       - uses: Swatinem/rust-cache@v1
       - name: asan
-        run: cargo test --all-features --target x86_64-unknown-linux-gnu --lib -- --test-threads 1
-        working-directory: tokio
+        run: cargo test --workspace --all-features --target x86_64-unknown-linux-gnu --tests -- --test-threads 1
         env:
           RUSTFLAGS: -Z sanitizer=address
-          ASAN_OPTIONS: detect_leaks=0
+          # Ignore `trybuild` errors as they are irrelevant and flaky on nightly
+          TRYBUILD: overwrite
 
   cross:
     name: cross

tokio/tests/io_read.rs

@@ -8,6 +8,11 @@ use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
+mod support {
+    pub(crate) mod leaked_buffers;
+}
+use support::leaked_buffers::LeakedBuffers;
+
 #[tokio::test]
 async fn read() {
     #[derive(Default)]
@@ -37,16 +42,27 @@ async fn read() {
     assert_eq!(buf[..], b"hello world"[..]);
 }
 
-struct BadAsyncRead;
+struct BadAsyncRead {
+    leaked_buffers: LeakedBuffers,
+}
+
+impl BadAsyncRead {
+    fn new() -> Self {
+        Self {
+            leaked_buffers: LeakedBuffers::new(),
+        }
+    }
+}
 
 impl AsyncRead for BadAsyncRead {
     fn poll_read(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         _cx: &mut Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
-        *buf = ReadBuf::new(Box::leak(vec![0; buf.capacity()].into_boxed_slice()));
+        *buf = ReadBuf::new(unsafe { self.leaked_buffers.create(buf.capacity()) });
         buf.advance(buf.capacity());
+
         Poll::Ready(Ok(()))
     }
 }
@@ -55,5 +71,5 @@ impl AsyncRead for BadAsyncRead {
 #[should_panic]
 async fn read_buf_bad_async_read() {
     let mut buf = Vec::with_capacity(10);
-    BadAsyncRead.read_buf(&mut buf).await.unwrap();
+    BadAsyncRead::new().read_buf(&mut buf).await.unwrap();
 }

tokio/tests/io_take.rs

@@ -6,6 +6,11 @@ use std::task::{Context, Poll};
 use tokio::io::{self, AsyncRead, AsyncReadExt, ReadBuf};
 use tokio_test::assert_ok;
 
+mod support {
+    pub(crate) mod leaked_buffers;
+}
+use support::leaked_buffers::LeakedBuffers;
+
 #[tokio::test]
 async fn take() {
     let mut buf = [0; 6];
@@ -34,17 +39,25 @@ async fn issue_4435() {
     assert_eq!(&buf, &b"ABhell\0\0"[..]);
 }
 
-struct BadReader;
+struct BadReader {
+    leaked_buffers: LeakedBuffers,
+}
+
+impl BadReader {
+    fn new() -> Self {
+        Self {
+            leaked_buffers: LeakedBuffers::new(),
+        }
+    }
+}
 
 impl AsyncRead for BadReader {
     fn poll_read(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         _cx: &mut Context<'_>,
         read_buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
-        let vec = vec![0; 10];
-
-        let mut buf = ReadBuf::new(vec.leak());
+        let mut buf = ReadBuf::new(unsafe { self.leaked_buffers.create(10) });
         buf.put_slice(&[123; 10]);
         *read_buf = buf;
 
@@ -57,5 +70,5 @@ impl AsyncRead for BadReader {
 async fn bad_reader_fails() {
     let mut buf = Vec::with_capacity(10);
 
-    BadReader.take(10).read_buf(&mut buf).await.unwrap();
+    BadReader::new().take(10).read_buf(&mut buf).await.unwrap();
 }

tokio/tests/rt_common.rs

@@ -749,6 +749,10 @@ rt_test! {
     #[test]
     fn wake_while_rt_is_dropping() {
         use tokio::task;
+        use core::sync::atomic::{AtomicBool, Ordering};
+
+        let drop_triggered = Arc::new(AtomicBool::new(false));
+        let set_drop_triggered = drop_triggered.clone();
 
         struct OnDrop<F: FnMut()>(F);
 
@@ -764,15 +768,14 @@ rt_test! {
 
         let rt = rt();
 
-        let h1 = rt.clone();
-
         rt.spawn(async move {
             // Ensure a waker gets stored in oneshot 1.
             let _ = rx1.await;
             tx3.send(()).unwrap();
         });
 
         rt.spawn(async move {
+            let h1 = tokio::runtime::Handle::current();
             // When this task is dropped, we'll be "closing remotes".
             // We spawn a new task that owns the `tx1`, to move its Drop
             // out of here.
@@ -785,6 +788,8 @@ rt_test! {
                 h1.spawn(async move {
                     tx1.send(()).unwrap();
                 });
+                // Just a sanity check that this entire thing actually happened
+                set_drop_triggered.store(true, Ordering::Relaxed);
             });
             let _ = rx2.await;
         });
@@ -803,6 +808,9 @@ rt_test! {
 
         // Drop the rt
         drop(rt);
+
+        // Make sure that the spawn actually happened
+        assert!(drop_triggered.load(Ordering::Relaxed));
     }
 
     #[test]

tokio/tests/support/leaked_buffers.rs

@@ -0,0 +1,26 @@
+/// Can create buffers of arbitrary lifetime.
+/// Frees created buffers when dropped.
+///
+/// This struct is of course unsafe and the fact that
+/// it must outlive the created slices has to be ensured by
+/// the programmer.
+///
+/// Used at certain test scenarios as a safer version of
+/// Vec::leak, to satisfy the address sanitizer.
+pub struct LeakedBuffers {
+    leaked_vecs: Vec<Box<[u8]>>,
+}
+
+impl LeakedBuffers {
+    pub fn new() -> Self {
+        Self {
+            leaked_vecs: vec![],
+        }
+    }
+    pub unsafe fn create<'a>(&mut self, size: usize) -> &'a mut [u8] {
+        let mut new_mem = vec![0u8; size].into_boxed_slice();
+        let slice = std::slice::from_raw_parts_mut(new_mem.as_mut_ptr(), new_mem.len());
+        self.leaked_vecs.push(new_mem);
+        slice
+    }
+}