Add block_until_needs_persist to ChannelManager

This will allow the ChannelManager to signal when it has new
updates to persist. Also add signalling for persistence.

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -58,7 +58,7 @@ use util::errors::APIError;
 use std::{cmp, mem};
 use std::collections::{HashMap, hash_map, HashSet};
 use std::io::{Cursor, Read};
-use std::sync::{Arc, Mutex, MutexGuard, RwLock};
+use std::sync::{Arc, Mutex, MutexGuard, RwLock, Condvar};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
 use std::marker::{Sync, Send};
@@ -439,6 +439,14 @@ pub struct ChannelManager<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref,
 	/// Taken first everywhere where we are making changes before any other locks.
 	total_consistency_lock: RwLock<()>,
 
+	/// Used to signal to the ChannelManager persister that the manager needs to be re-persisted to
+	/// disk/backups. Users won't access the persistence_lock directly, but rather wait on its bool
+	/// using block_until_needs_persist.
+	#[cfg(any(test, feature = "_test_utils"))]
+	pub persistence_lock: Arc<(Mutex<bool>, Condvar)>,
+	#[cfg(not(any(test, feature = "_test_utils")))]
+	persistence_lock: Arc<(Mutex<bool>, Condvar)>,
+
 	keys_manager: K,
 
 	logger: L,
@@ -760,12 +768,36 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 			pending_events: Mutex::new(Vec::new()),
 			total_consistency_lock: RwLock::new(()),
 
+			persistence_lock: Arc::new((Mutex::new(false), Condvar::new())),
+
 			keys_manager,
 
 			logger,
 		}
 	}
 
+	/// Used to signal to users when the ChannelManager needs to be persisted to disk. This function
+	/// blocks either for the given Duration or indefinitely until the ChannelManager has updates and
+	/// needs persistence. It returns a bool indicating whether persistence is necessary (which will
+	/// always be true if max_wait is None).
+	pub fn block_until_needs_persist(&self, max_wait: Option<Duration>) -> bool {
+		loop {
+			let mutcond = Arc::clone(&self.persistence_lock);
+			let &(ref mtx, ref cvar) = &*mutcond;
+			let mut guard = mtx.lock().unwrap();
+			let result = match max_wait {
+				Some(wait) => cvar.wait_timeout(guard, wait).unwrap().0,
+				None => cvar.wait(guard).unwrap(),
+			};
+			guard = result;
+			let saved_res = *guard;
+			*guard = false;
+			if saved_res || max_wait.is_some() {
+				return saved_res;
+			}
+		}
+	}
+
 	/// Creates a new outbound channel to the given remote node and with the given value.
 	///
 	/// user_id will be provided back as user_channel_id in FundingGenerationReady and
@@ -913,6 +945,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 			// the latest local state, which is the best we can do anyway. Thus, it is safe to
 			// ignore the result here.
 			let _ = self.chain_monitor.update_channel(funding_txo, monitor_update);
+			self.persist_updates();
 		}
 	}
 
@@ -1313,6 +1346,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 								commitment_signed,
 							},
 						});
+						self.persist_updates();
 					},
 					None => {},
 				}
@@ -1707,6 +1741,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 									commitment_signed: commitment_msg,
 								},
 							});
+							self.persist_updates();
 						}
 					} else {
 						unreachable!();
@@ -2126,6 +2161,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 							}
 						});
 					}
+					self.persist_updates();
 					return Ok(())
 				},
 				Err(e) => {
@@ -2340,6 +2376,14 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 		Ok(())
 	}
 
+	// Signal to the ChannelManager persister that there are updates necessitating persisting to disk.
+	fn persist_updates(&self) {
+		let &(ref persist_mtx, ref cnd) = &*self.persistence_lock;
+		let mut persistence_lock = persist_mtx.lock().unwrap();
+		*persistence_lock = true;
+		cnd.notify_all();
+	}
+
 	fn internal_funding_created(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> {
 		let ((funding_msg, monitor), mut chan) = {
 			let mut channel_lock = self.channel_state.lock().unwrap();
@@ -2379,6 +2423,9 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 				},
 			}
 		}
+
+		self.persist_updates();
+
 		let mut channel_state_lock = self.channel_state.lock().unwrap();
 		let channel_state = &mut *channel_state_lock;
 		match channel_state.by_id.entry(funding_msg.channel_id) {
@@ -2417,6 +2464,9 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 				hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
 			}
 		};
+
+		self.persist_updates();
+
 		let mut pending_events = self.pending_events.lock().unwrap();
 		pending_events.push(events::Event::FundingBroadcastSafe {
 			funding_txo,
@@ -2712,6 +2762,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 						msg,
 					});
 				}
+				self.persist_updates();
 				Ok(())
 			},
 			hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
@@ -2803,9 +2854,13 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 					self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2);
 				}
 				self.forward_htlcs(&mut [(short_channel_id, channel_outpoint, pending_forwards)]);
+				self.persist_updates();
 				Ok(())
 			},
-			Err(e) => Err(e)
+			Err(e) => {
+				self.persist_updates();
+				Err(e)
+			}
 		}
 	}
 
@@ -2946,6 +3001,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 						msg,
 					});
 				}
+				self.persist_updates();
 				Ok(())
 			},
 			hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
@@ -2995,6 +3051,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 								commitment_signed,
 							},
 						});
+						self.persist_updates();
 					}
 				},
 			}
@@ -3994,6 +4051,7 @@ impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: De
 
 			pending_events: Mutex::new(pending_events_read),
 			total_consistency_lock: RwLock::new(()),
+			persistence_lock: Arc::new((Mutex::new(false), Condvar::new())),
 			keys_manager: args.keys_manager,
 			logger: args.logger,
 			default_configuration: args.default_config,