f test

Author: TheBlueMatt

lightning/src/ln/peer_handler.rs

@@ -2876,4 +2876,52 @@ mod tests {
 		// For (None)
 		assert_eq!(filter_addresses(None), None);
 	}
+
+	#[test]
+	#[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.
+		// 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.
+		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 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);
+					thread_peer.process_events();
+					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);
+		}
+
+		exit_flag.store(true, Ordering::Release);
+		thread_a.join().unwrap();
+		thread_b.join().unwrap();
+		thread_c.join().unwrap();
+		assert!(cfg[0].chan_handler.message_fetch_counter.load(Ordering::Acquire) >= 1);
+	}
 }

lightning/src/util/test_utils.rs

@@ -359,6 +359,7 @@ 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,
 }
 
 impl TestChannelMessageHandler {
@@ -367,6 +368,7 @@ impl TestChannelMessageHandler {
 			pending_events: Mutex::new(Vec::new()),
 			expected_recv_msgs: Mutex::new(None),
 			connected_peers: Mutex::new(HashSet::new()),
+			message_fetch_counter: AtomicUsize::new(0),
 		}
 	}
 
@@ -517,6 +519,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);
 		let mut pending_events = self.pending_events.lock().unwrap();
 		let mut ret = Vec::new();
 		mem::swap(&mut ret, &mut *pending_events);