test(s2n-quic-dc): disable tracing by default for tests with large output (#2571)

Author: camshaft

dc/s2n-quic-dc/src/path/secret/map/cleaner.rs

@@ -60,6 +60,12 @@ impl Cleaner {
         C: 'static + time::Clock + Send + Sync,
         S: event::Subscriber,
     {
+        // check to see if we're in a simulation before spawning a thread
+        #[cfg(any(test, feature = "testing"))]
+        if bach::is_active() {
+            return;
+        }
+
         let state = Arc::downgrade(&state);
         let handle = std::thread::Builder::new()
             .name("dc_quic::cleaner".into())

dc/s2n-quic-dc/src/stream/send/tests.rs

@@ -270,9 +270,15 @@ macro_rules! negative_suite {
 
 macro_rules! current_thread {
     ($name:ident, $protocol:expr, $buffer_len:expr, $iterations:expr, $features:expr) => {
-        #[tokio::test]
-        async fn $name() {
-            run($protocol, $buffer_len, $iterations, $features).await;
+        #[test]
+        fn $name() {
+            crate::testing::without_tracing(|| {
+                let rt = tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .unwrap();
+                rt.block_on(run($protocol, $buffer_len, $iterations, $features));
+            });
         }
     };
 }
@@ -281,6 +287,8 @@ macro_rules! multi_thread {
     ($name:ident, $protocol:expr, $buffer_len:expr, $iterations:expr, $features:expr) => {
         #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
         async fn $name() {
+            // Since these run different threads, it's not super easy to use the `without_tracing` function.
+            // Hopefully that will be ok for the CI with limited memory.
             run($protocol, $buffer_len, $iterations, $features).await;
         }
     };
@@ -290,12 +298,14 @@ macro_rules! sim {
     ($name:ident, $protocol:expr, $buffer_len:expr, $iterations:expr, $features:expr) => {
         #[test]
         fn $name() {
-            crate::testing::sim(|| {
-                async {
-                    run($protocol, $buffer_len, $iterations, $features).await;
-                }
-                .primary()
-                .spawn();
+            crate::testing::without_tracing(|| {
+                crate::testing::sim(|| {
+                    async {
+                        run($protocol, $buffer_len, $iterations, $features).await;
+                    }
+                    .primary()
+                    .spawn();
+                });
             });
         }
     };

dc/s2n-quic-dc/src/stream/testing.rs

@@ -34,8 +34,7 @@ pub type Stream = application::Stream<Subscriber>;
 pub type Writer = send::application::Writer<Subscriber>;
 pub type Reader = recv::application::Reader<Subscriber>;
 
-// TODO enable this once ready
-const DEFAULT_POOLED: bool = false;
+const DEFAULT_POOLED: bool = true;
 
 // limit the number of threads used in testing to reduce costs of harnesses
 const TEST_THREADS: usize = 2;
@@ -398,7 +397,7 @@ pub mod server {
     impl Default for Builder {
         fn default() -> Self {
             Self {
-                backlog: 16,
+                backlog: 4096,
                 flavor: accept::Flavor::default(),
                 protocol: Protocol::Tcp,
                 map_capacity: 16,

dc/s2n-quic-dc/src/stream/tests/request_response.rs

@@ -6,7 +6,7 @@ use crate::{
         socket::Protocol,
         testing::{self, Stream, MAX_DATAGRAM_SIZE},
     },
-    testing::{ext::*, sim, sleep, spawn, task::spawn_named, timeout},
+    testing::{ext::*, sim, sleep, spawn, task::spawn_named, timeout, without_tracing},
 };
 use bolero::{produce, TypeGenerator, ValueGenerator as _};
 use core::time::Duration;
@@ -186,9 +186,9 @@ async fn check_read(
         let offset = data.offset();
         let capacity = data.buffered_len();
         let mut data = data.with_write_limit(capacity.min(max_read_len));
-        tracing::info!(offset, capacity = data.remaining_capacity(), "read_into");
+        tracing::debug!(offset, capacity = data.remaining_capacity(), "read_into");
         let len = stream.read_into(&mut data).await?;
-        tracing::info!(len, "read_into_result");
+        tracing::debug!(len, "read_into_result");
         if len == 0 {
             break;
         }
@@ -203,13 +203,13 @@ async fn check_write(stream: &mut Stream, delays: &Delays, amount: usize) -> io:
     info!(writing = amount);
     let mut data = Data::new(amount as _);
     while !data.is_finished() {
-        tracing::info!(
+        tracing::debug!(
             offset = data.offset(),
             remaining = data.buffered_len(),
             "write_from"
         );
         let len = stream.write_from(&mut data).await?;
-        tracing::info!(len, "write_from_result");
+        tracing::debug!(len, "write_from_result");
     }
     info!(amount, "write_from_finished");
     Ok(())
@@ -356,15 +356,19 @@ impl Harness {
         }
 
         if is_sim {
-            // TODO limit concurrency
+            // TODO limit concurrency - for now we just stagger clients
 
             for idx in 0..self.client.count {
                 let config = self.client;
                 let requests = requests.clone();
                 let client = client.clone();
                 let server = server.clone();
                 let task = async move {
+                    // delay connecting by 1us per client
+                    (1.us() * idx as u32).sleep().await;
+
                     info!("connecting");
+
                     let stream = client.connect_to(&server).await.unwrap();
                     info!(peer_addr = %stream.peer_addr().unwrap(), local_addr = %stream.local_addr().unwrap());
 
@@ -429,8 +433,8 @@ struct Runtime {
 }
 
 impl Runtime {
-    fn new(harness: Harness) -> Self {
-        let rt = tokio::runtime::Builder::new_multi_thread()
+    fn new(harness: &Harness) -> Self {
+        let rt = tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()
             .unwrap();
@@ -452,6 +456,10 @@ impl Runtime {
         }
     }
 
+    fn run(&self, harness: Harness) {
+        self.rt.block_on(harness.run());
+    }
+
     fn run_with(&self, client: Client, server: Server, requests: Vec<Request>) {
         let harness = Harness {
             client,
@@ -528,11 +536,6 @@ macro_rules! tests {
         $test!(multi_request_test, {
             let harness = harness();
 
-            // TODO make this not flaky with UDP
-            if harness.protocol.is_udp() {
-                return;
-            }
-
             Harness {
                 requests: vec![Request {
                     count: 10,
@@ -628,46 +631,54 @@ macro_rules! tests {
 
 macro_rules! tokio_test {
     ($name:ident, $harness:expr) => {
-        #[tokio::test]
-        async fn $name() {
-            $harness.run().await;
+        #[test]
+        fn $name() {
+            let harness = $harness;
+            without_tracing(|| {
+                Runtime::new(&harness).run(harness);
+            });
         }
     };
 }
 
 macro_rules! tokio_fuzz_test {
     () => {
         #[test]
-        #[ignore = "TODO the CI currently doesn't like running these tests - need to figure out why"]
         fn fuzz_test() {
-            use std::sync::OnceLock;
-
-            bolero::check!()
-                .with_generator((produce(), produce(), produce::<Vec<_>>().with().len(1..5)))
-                .cloned()
-                .with_test_time(Duration::from_secs(45))
-                .for_each(|(client, server, requests)| {
-                    static RUNTIME: OnceLock<Runtime> = OnceLock::new();
-                    RUNTIME
-                        .get_or_init(|| Runtime::new(harness()))
-                        .run_with(client, server, requests);
-                });
+            without_tracing(|| {
+                use std::sync::OnceLock;
+
+                bolero::check!()
+                    .with_generator((produce(), produce(), produce::<Vec<_>>().with().len(1..5)))
+                    .cloned()
+                    // limit the amount of time in tests since they can produce a log of tracing data
+                    .with_test_time(Duration::from_secs(10))
+                    .for_each(|(client, server, requests)| {
+                        static RUNTIME: OnceLock<Runtime> = OnceLock::new();
+                        RUNTIME
+                            .get_or_init(|| Runtime::new(&harness()))
+                            .run_with(client, server, requests);
+                    });
+            });
         }
-    }
+    };
 }
 
 macro_rules! sim_test {
     ($name:ident, $harness:expr) => {
         #[test]
         fn $name() {
-            sim(|| {
-                spawn_named($harness.run().primary(), "harness");
-
-                async {
-                    sleep(Duration::from_secs(120)).await;
-                    panic!("test timed out after {}", bach::time::Instant::now());
-                }
-                .spawn();
+            // The tracing logs end up consuming a bunch of memory and failing the tests
+            without_tracing(|| {
+                sim(|| {
+                    spawn_named($harness.run().primary(), "harness");
+
+                    async {
+                        sleep(Duration::from_secs(120)).await;
+                        panic!("test timed out after {}", bach::time::Instant::now());
+                    }
+                    .spawn();
+                });
             });
         }
     };
@@ -715,24 +726,26 @@ mod udp_sim {
     tests!(sim_test);
 
     #[test]
-    #[ignore = "TODO the CI currently doesn't like running these tests - need to figure out why"]
     fn fuzz_test() {
-        bolero::check!()
-            .with_generator((produce(), produce(), produce::<Vec<_>>().with().len(1..5)))
-            .cloned()
-            .with_test_time(Duration::from_secs(60))
-            .for_each(|(client, server, requests)| {
-                crate::testing::sim(|| {
-                    Harness {
-                        client,
-                        server,
-                        requests,
-                        ..harness()
-                    }
-                    .run()
-                    .primary()
-                    .spawn();
-                });
-            });
+        // The tracing logs end up consuming a bunch of memory and failing the tests
+        without_tracing(|| {
+            bolero::check!()
+                .with_generator((produce(), produce(), produce::<Vec<_>>().with().len(1..5)))
+                .cloned()
+                .with_test_time(Duration::from_secs(60))
+                .for_each(|(client, server, requests)| {
+                    crate::testing::sim(|| {
+                        Harness {
+                            client,
+                            server,
+                            requests,
+                            ..harness()
+                        }
+                        .run()
+                        .primary()
+                        .spawn();
+                    });
+                })
+        });
     }
 }

dc/s2n-quic-dc/src/testing.rs

@@ -1,7 +1,7 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::time::Duration;
+use std::{sync::OnceLock, time::Duration};
 
 pub use bach::{ext, rand};
 
@@ -79,7 +79,10 @@ pub fn init_tracing() {
             .with_timer(Uptime::default())
             .compact(); // Use a less verbose output format.
 
-        let default_level = if cfg!(debug_assertions) {
+        let default_level = if std::env::var("CI").is_ok() {
+            // The CI runs out of memory if we log too much tracing data
+            tracing::Level::INFO
+        } else if cfg!(debug_assertions) {
             tracing::Level::DEBUG
         } else {
             tracing::Level::WARN
@@ -99,6 +102,20 @@ pub fn init_tracing() {
     });
 }
 
+pub fn without_tracing<F: FnOnce() -> T, T>(f: F) -> T {
+    // make sure the global subscriber is initialized before setting a local one
+    init_tracing();
+
+    static FORCED: OnceLock<bool> = OnceLock::new();
+
+    // add the option to get logs with `S2N_LOG_FORCED=1`
+    if *FORCED.get_or_init(|| std::env::var("S2N_LOG_FORCED").is_ok()) {
+        return f();
+    }
+
+    tracing::subscriber::with_default(tracing::subscriber::NoSubscriber::new(), f)
+}
+
 #[derive(Default)]
 struct Uptime(tracing_subscriber::fmt::time::SystemTime);