Instrument background processor loop timing

Scaling to 16 tokio workers and a larger instance had zero
impact on the heartbeat polling delay metric, ruling out
worker starvation as the cause of the remaining 1-2s jitter
(and 4s spikes every 60s).

The jitter must come from synchronous operations inside the
process_events_async loop that block a worker thread without
yielding. The main suspects are peer_manager.process_events()
(blocks on ChannelManager locks every iteration) and
channel_manager.timer_tick_occurred() (fires every 60s,
correlates with the 4s spikes).

Wraps every synchronous call in the loop with Instant timing
and logs a full breakdown whenever a loop iteration exceeds
500ms.

Author: amackillop

lightning-background-processor/src/lib.rs

@@ -987,11 +987,21 @@ where
 	let mut last_forwards_processing_call = sleeper(batch_delay.get());
 
 	loop {
+		let loop_start = Instant::now();
+
+		let t = Instant::now();
 		channel_manager.get_cm().process_pending_events_async(async_event_handler).await;
+		let cm_events_ms = t.elapsed().as_millis();
+
+		let t = Instant::now();
 		chain_monitor.process_pending_events_async(async_event_handler).await;
+		let chain_mon_events_ms = t.elapsed().as_millis();
+
+		let t = Instant::now();
 		if let Some(om) = &onion_messenger {
 			om.get_om().process_pending_events_async(async_event_handler).await
 		}
+		let om_events_ms = t.elapsed().as_millis();
 
 		// Note that the PeerManager::process_events may block on ChannelManager's locks,
 		// hence it comes last here. When the ChannelManager finishes whatever it's doing,
@@ -1004,7 +1014,11 @@ where
 		// ChannelManager, we want to minimize methods blocking on a ChannelManager
 		// generally, and as a fallback place such blocking only immediately before
 		// persistence.
+		let t = Instant::now();
 		peer_manager.as_ref().process_events();
+		let peer_mgr_events_ms = t.elapsed().as_millis();
+
+		let t = Instant::now();
 		match check_and_reset_sleeper(&mut last_forwards_processing_call, || {
 			sleeper(batch_delay.next())
 		}) {
@@ -1014,6 +1028,7 @@ where
 			Some(true) => break,
 			None => {},
 		}
+		let htlc_fwd_ms = t.elapsed().as_millis();
 
 		// We wait up to 100ms, but track how long it takes to detect being put to sleep,
 		// see `await_start`'s use below.
@@ -1067,14 +1082,18 @@ where
 		} else {
 			false
 		};
+		let t = Instant::now();
+		let mut cm_tick_fired = false;
 		match check_and_reset_sleeper(&mut last_freshness_call, || sleeper(FRESHNESS_TIMER)) {
 			Some(false) => {
 				log_trace!(logger, "Calling ChannelManager's timer_tick_occurred");
 				channel_manager.get_cm().timer_tick_occurred();
+				cm_tick_fired = true;
 			},
 			Some(true) => break,
 			None => {},
 		}
+		let cm_tick_ms = t.elapsed().as_millis();
 
 		let mut futures = Joiner::new();
 
@@ -1132,9 +1151,12 @@ where
 			GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed,
 			_ => prune_timer_elapsed,
 		};
+		let t = Instant::now();
+		let mut prune_fired = false;
 		if should_prune {
 			// The network graph must not be pruned while rapid sync completion is pending
 			if let Some(network_graph) = gossip_sync.prunable_network_graph() {
+				prune_fired = true;
 				if let Some(duration_since_epoch) = fetch_time() {
 					log_trace!(logger, "Pruning and persisting network graph.");
 					network_graph.remove_stale_channels_and_tracking_with_time(
@@ -1166,6 +1188,8 @@ where
 				have_pruned = true;
 			}
 		}
+		let prune_ms = t.elapsed().as_millis();
+
 		if !have_decayed_scorer {
 			if let Some(ref scorer) = scorer {
 				if let Some(duration_since_epoch) = fetch_time() {
@@ -1247,6 +1271,7 @@ where
 			res?;
 		}
 
+		let t = Instant::now();
 		match check_and_reset_sleeper(&mut last_onion_message_handler_call, || {
 			sleeper(ONION_MESSAGE_HANDLER_TIMER)
 		}) {
@@ -1259,8 +1284,10 @@ where
 			Some(true) => break,
 			None => {},
 		}
+		let om_tick_ms = t.elapsed().as_millis();
 
 		// Peer manager timer tick. If we were interrupted on a mobile platform, we disconnect all peers.
+		let t = Instant::now();
 		if await_slow {
 			// On various platforms, we may be starved of CPU cycles for several reasons.
 			// E.g. on iOS, if we've been in the background, we will be entirely paused.
@@ -1288,7 +1315,10 @@ where
 			}
 		}
 
+		let peer_mgr_tick_ms = t.elapsed().as_millis();
+
 		// Rebroadcast pending claims.
+		let t = Instant::now();
 		match check_and_reset_sleeper(&mut last_rebroadcast_call, || sleeper(REBROADCAST_TIMER)) {
 			Some(false) => {
 				log_trace!(logger, "Rebroadcasting monitor's pending claims");
@@ -1297,6 +1327,24 @@ where
 			Some(true) => break,
 			None => {},
 		}
+		let rebroadcast_ms = t.elapsed().as_millis();
+
+		let loop_ms = loop_start.elapsed().as_millis();
+		if loop_ms > 500 {
+			log_warn!(logger,
+				"[bg-processor] loop took {}ms: \
+				cm_events={}ms chain_mon={}ms om_events={}ms \
+				peer_mgr={}ms htlc_fwd={}ms \
+				cm_tick={}ms{} prune={}ms{} \
+				om_tick={}ms peer_tick={}ms rebroadcast={}ms",
+				loop_ms,
+				cm_events_ms, chain_mon_events_ms, om_events_ms,
+				peer_mgr_events_ms, htlc_fwd_ms,
+				cm_tick_ms, if cm_tick_fired { "(fired)" } else { "" },
+				prune_ms, if prune_fired { "(fired)" } else { "" },
+				om_tick_ms, peer_mgr_tick_ms, rebroadcast_ms,
+			);
+		}
 	}
 	log_trace!(logger, "Terminating background processor.");