Correct `peer_handler::test_process_events_multithreaded`

This test was added some time ago in
0c034e9a82e4339fb32af9da63832ac2a64abb0b, but never made any sense.
`PeerManager::process_events` will go around its loop as many times
is required to ensure we've always processed all events which were
pending prior to a `process_events` call, so having a test that
checks that we never go around more than twice is obviously broken.

And, indeed, in CI this tests fails with some regularity.

Instead, the test here is changed to ensure that we detectably go
around the loop again at least once.

Fixes #2385

diff --git a/lightning/src/ln/peer_handler.rs b/lightning/src/ln/peer_handler.rs
index b4c242627371b6d636b91ab18386376431a3ed6b..6b03b36545779f33a146a786b5e2a35de39bf196 100644 (file)
--- a/lightning/src/ln/peer_handler.rs
+++ b/lightning/src/ln/peer_handler.rs
@@ -3452,48 +3452,44 @@ mod tests {
        #[cfg(feature = "std")]
        fn test_process_events_multithreaded() {
                use std::time::{Duration, Instant};
-               // Test that `process_events` getting called on multiple threads doesn't generate too many
-               // loop iterations.
+               // `process_events` shouldn't block on another thread processing events and instead should
+               // simply signal the currently processing thread to go around the loop again.
+               // Here we test that this happens by spawning a few threads and checking that we see one go
+               // around again at least once.
+               //
                // Each time `process_events` goes around the loop we call
-               // `get_and_clear_pending_msg_events`, which we count using the `TestMessageHandler`.
-               // Because the loop should go around once more after a call which fails to take the
-               // single-threaded lock, if we write zero to the counter before calling `process_events` we
-               // should never observe there having been more than 2 loop iterations.
-               // Further, because the last thread to exit will call `process_events` before returning, we
-               // should always have at least one count at the end.
+               // `get_and_clear_pending_msg_events`, which we count using the `TestMessageHandler`. Thus,
+               // to test we simply write zero to the counter before calling `process_events` and make
+               // sure we observe a value greater than one at least once.
                let cfg = Arc::new(create_peermgr_cfgs(1));
                // Until we have std::thread::scoped we have to unsafe { turn off the borrow checker }.
                let peer = Arc::new(create_network(1, unsafe { &*(&*cfg as *const _) as &'static _ }).pop().unwrap());
 
-               let exit_flag = Arc::new(AtomicBool::new(false));
-               macro_rules! spawn_thread { () => { {
-                       let thread_cfg = Arc::clone(&cfg);
+               let end_time = Instant::now() + Duration::from_millis(100);
+               let observed_loop = Arc::new(AtomicBool::new(false));
+               let thread_fn = || {
                        let thread_peer = Arc::clone(&peer);
-                       let thread_exit = Arc::clone(&exit_flag);
-                       std::thread::spawn(move || {
-                               while !thread_exit.load(Ordering::Acquire) {
-                                       thread_cfg[0].chan_handler.message_fetch_counter.store(0, Ordering::Release);
+                       let thread_observed_loop = Arc::clone(&observed_loop);
+                       move || {
+                               while Instant::now() < end_time || !thread_observed_loop.load(Ordering::Acquire) {
+                                       test_utils::TestChannelMessageHandler::MESSAGE_FETCH_COUNTER.with(|val| val.store(0, Ordering::Relaxed));
                                        thread_peer.process_events();
+                                       if test_utils::TestChannelMessageHandler::MESSAGE_FETCH_COUNTER.with(|val| val.load(Ordering::Relaxed)) > 1 {
+                                               thread_observed_loop.store(true, Ordering::Release);
+                                               return;
+                                       }
                                        std::thread::sleep(Duration::from_micros(1));
                                }
-                       })
-               } } }
-
-               let thread_a = spawn_thread!();
-               let thread_b = spawn_thread!();
-               let thread_c = spawn_thread!();
-
-               let start_time = Instant::now();
-               while start_time.elapsed() < Duration::from_millis(100) {
-                       let val = cfg[0].chan_handler.message_fetch_counter.load(Ordering::Acquire);
-                       assert!(val <= 2);
-                       std::thread::yield_now(); // Winblowz seemingly doesn't ever interrupt threads?!
-               }
+                       }
+               };
 
-               exit_flag.store(true, Ordering::Release);
+               let thread_a = std::thread::spawn(thread_fn());
+               let thread_b = std::thread::spawn(thread_fn());
+               let thread_c = std::thread::spawn(thread_fn());
+               thread_fn()();
                thread_a.join().unwrap();
                thread_b.join().unwrap();
                thread_c.join().unwrap();
-               assert!(cfg[0].chan_handler.message_fetch_counter.load(Ordering::Acquire) >= 1);
+               assert!(observed_loop.load(Ordering::Acquire));
        }
 }

diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs
index 3f2226d741d3a0180143ffd43de257bfadbbec70..050c2a3e007940592598b49bfbe221bb4b13e896 100644 (file)
--- a/lightning/src/util/test_utils.rs
+++ b/lightning/src/util/test_utils.rs
@@ -759,17 +759,21 @@ pub struct TestChannelMessageHandler {
        pub pending_events: Mutex<Vec<events::MessageSendEvent>>,
        expected_recv_msgs: Mutex<Option<Vec<wire::Message<()>>>>,
        connected_peers: Mutex<HashSet<PublicKey>>,
-       pub message_fetch_counter: AtomicUsize,
        chain_hash: ChainHash,
 }
 
+impl TestChannelMessageHandler {
+       thread_local! {
+               pub static MESSAGE_FETCH_COUNTER: AtomicUsize = AtomicUsize::new(0);
+       }
+}
+
 impl TestChannelMessageHandler {
        pub fn new(chain_hash: ChainHash) -> Self {
                TestChannelMessageHandler {
                        pending_events: Mutex::new(Vec::new()),
                        expected_recv_msgs: Mutex::new(None),
                        connected_peers: Mutex::new(new_hash_set()),
-                       message_fetch_counter: AtomicUsize::new(0),
                        chain_hash,
                }
        }
@@ -940,7 +944,7 @@ impl msgs::ChannelMessageHandler for TestChannelMessageHandler {
 
 impl events::MessageSendEventsProvider for TestChannelMessageHandler {
        fn get_and_clear_pending_msg_events(&self) -> Vec<events::MessageSendEvent> {
-               self.message_fetch_counter.fetch_add(1, Ordering::AcqRel);
+               Self::MESSAGE_FETCH_COUNTER.with(|val| val.fetch_add(1, Ordering::AcqRel));
                let mut pending_events = self.pending_events.lock().unwrap();
                let mut ret = Vec::new();
                mem::swap(&mut ret, &mut *pending_events);