Merge pull request #2521 from TheBlueMatt/2023-08-one-less-write

Avoid persisting ChannelManager in some cases and separate event from persist notifies

Author: TheBlueMatt

fuzz/src/chanmon_consistency.rs

@@ -125,7 +125,6 @@ struct TestChainMonitor {
 	// "fails" if we ever force-close a channel, we avoid doing so, always saving the latest
 	// fully-serialized monitor state here, as well as the corresponding update_id.
 	pub latest_monitors: Mutex<HashMap<OutPoint, (u64, Vec<u8>)>>,
-	pub should_update_manager: atomic::AtomicBool,
 }
 impl TestChainMonitor {
 	pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
@@ -135,7 +134,6 @@ impl TestChainMonitor {
 			keys,
 			persister,
 			latest_monitors: Mutex::new(HashMap::new()),
-			should_update_manager: atomic::AtomicBool::new(false),
 		}
 	}
 }
@@ -146,7 +144,6 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 		if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
 			panic!("Already had monitor pre-watch_channel");
 		}
-		self.should_update_manager.store(true, atomic::Ordering::Relaxed);
 		self.chain_monitor.watch_channel(funding_txo, monitor)
 	}
 
@@ -162,7 +159,6 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 		let mut ser = VecWriter(Vec::new());
 		deserialized_monitor.write(&mut ser).unwrap();
 		map_entry.insert((update.update_id, ser.0));
-		self.should_update_manager.store(true, atomic::Ordering::Relaxed);
 		self.chain_monitor.update_channel(funding_txo, update)
 	}
 
@@ -1101,11 +1097,9 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out) {
 				if !chan_a_disconnected {
 					nodes[1].peer_disconnected(&nodes[0].get_our_node_id());
 					chan_a_disconnected = true;
-					drain_msg_events_on_disconnect!(0);
-				}
-				if monitor_a.should_update_manager.load(atomic::Ordering::Relaxed) {
-					node_a_ser.0.clear();
-					nodes[0].write(&mut node_a_ser).unwrap();
+					push_excess_b_events!(nodes[1].get_and_clear_pending_msg_events().drain(..), Some(0));
+					ab_events.clear();
+					ba_events.clear();
 				}
 				let (new_node_a, new_monitor_a) = reload_node!(node_a_ser, 0, monitor_a, keys_manager_a, fee_est_a);
 				nodes[0] = new_node_a;
@@ -1134,11 +1128,9 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out) {
 				if !chan_b_disconnected {
 					nodes[1].peer_disconnected(&nodes[2].get_our_node_id());
 					chan_b_disconnected = true;
-					drain_msg_events_on_disconnect!(2);
-				}
-				if monitor_c.should_update_manager.load(atomic::Ordering::Relaxed) {
-					node_c_ser.0.clear();
-					nodes[2].write(&mut node_c_ser).unwrap();
+					push_excess_b_events!(nodes[1].get_and_clear_pending_msg_events().drain(..), Some(2));
+					bc_events.clear();
+					cb_events.clear();
 				}
 				let (new_node_c, new_monitor_c) = reload_node!(node_c_ser, 2, monitor_c, keys_manager_c, fee_est_c);
 				nodes[2] = new_node_c;
@@ -1304,15 +1296,18 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out) {
 			_ => test_return!(),
 		}
 
-		node_a_ser.0.clear();
-		nodes[0].write(&mut node_a_ser).unwrap();
-		monitor_a.should_update_manager.store(false, atomic::Ordering::Relaxed);
-		node_b_ser.0.clear();
-		nodes[1].write(&mut node_b_ser).unwrap();
-		monitor_b.should_update_manager.store(false, atomic::Ordering::Relaxed);
-		node_c_ser.0.clear();
-		nodes[2].write(&mut node_c_ser).unwrap();
-		monitor_c.should_update_manager.store(false, atomic::Ordering::Relaxed);
+		if nodes[0].get_and_clear_needs_persistence() == true {
+			node_a_ser.0.clear();
+			nodes[0].write(&mut node_a_ser).unwrap();
+		}
+		if nodes[1].get_and_clear_needs_persistence() == true {
+			node_b_ser.0.clear();
+			nodes[1].write(&mut node_b_ser).unwrap();
+		}
+		if nodes[2].get_and_clear_needs_persistence() == true {
+			node_c_ser.0.clear();
+			nodes[2].write(&mut node_c_ser).unwrap();
+		}
 	}
 }
 

lightning-background-processor/src/lib.rs

@@ -315,7 +315,7 @@ macro_rules! define_run_body {
 			// see `await_start`'s use below.
 			let mut await_start = None;
 			if $check_slow_await { await_start = Some($get_timer(1)); }
-			let updates_available = $await;
+			$await;
 			let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
 
 			// Exit the loop if the background processor was requested to stop.
@@ -324,7 +324,7 @@ macro_rules! define_run_body {
 				break;
 			}
 
-			if updates_available {
+			if $channel_manager.get_and_clear_needs_persistence() {
 				log_trace!($logger, "Persisting ChannelManager...");
 				$persister.persist_manager(&*$channel_manager)?;
 				log_trace!($logger, "Done persisting ChannelManager.");
@@ -655,16 +655,14 @@ where
 		channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
 		gossip_sync, peer_manager, logger, scorer, should_break, {
 			let fut = Selector {
-				a: channel_manager.get_persistable_update_future(),
+				a: channel_manager.get_event_or_persistence_needed_future(),
 				b: chain_monitor.get_update_future(),
 				c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
 			};
 			match fut.await {
-				SelectorOutput::A => true,
-				SelectorOutput::B => false,
+				SelectorOutput::A|SelectorOutput::B => {},
 				SelectorOutput::C(exit) => {
 					should_break = exit;
-					false
 				}
 			}
 		}, |t| sleeper(Duration::from_secs(t)),
@@ -787,10 +785,10 @@ impl BackgroundProcessor {
 			define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
 				channel_manager, channel_manager.process_pending_events(&event_handler),
 				gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
-				Sleeper::from_two_futures(
-					channel_manager.get_persistable_update_future(),
+				{ Sleeper::from_two_futures(
+					channel_manager.get_event_or_persistence_needed_future(),
 					chain_monitor.get_update_future()
-				).wait_timeout(Duration::from_millis(100)),
+				).wait_timeout(Duration::from_millis(100)); },
 				|_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
 		});
 		Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
@@ -1326,7 +1324,7 @@ mod tests {
 		check_persisted_data!(nodes[0].node, filepath.clone());
 
 		loop {
-			if !nodes[0].node.get_persistence_condvar_value() { break }
+			if !nodes[0].node.get_event_or_persist_condvar_value() { break }
 		}
 
 		// Force-close the channel.
@@ -1335,7 +1333,7 @@ mod tests {
 		// Check that the force-close updates are persisted.
 		check_persisted_data!(nodes[0].node, filepath.clone());
 		loop {
-			if !nodes[0].node.get_persistence_condvar_value() { break }
+			if !nodes[0].node.get_event_or_persist_condvar_value() { break }
 		}
 
 		// Check network graph is persisted