Add persistence_lock to ChannelManager

This will allow the ChannelManager to signal when it has new
updates to persist.

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,41 @@ 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.
+	/// Example usage:
+	/// ```
+	/// use std::sync::Arc;
+	/// type TxBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface;
+	/// type FeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator;
+	/// type Logger = dyn lightning::util::logger::Logger;
+	/// type ChainAccess = dyn lightning::chain::Access;
+	/// type ChainFilter = dyn lightning::chain::Filter;
+	/// type DataPersister = dyn lightning::chain::channelmonitor::Persist<lightning::chain::keysinterface::InMemoryChannelKeys>;
+	/// type ChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::chain::keysinterface::InMemoryChannelKeys, Arc<ChainFilter>, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>, Arc<DataPersister>>;
+	/// type ChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor, TxBroadcaster, FeeEstimator, Logger>;
+	/// fn start(manager: ChannelManager) {
+	///   let mutcond = Arc::clone(&manager.persistence_lock);
+	///   // ChannelManager persistence should generally be run in a background thread since it can be
+	///   // time-consuming to persist.
+	///   std::thread::spawn(move || {
+	///     let &(ref mtx, ref cnd) = &*mutcond;
+	///     loop {
+	///       let mut guard = mtx.lock().unwrap();
+	///       // If there's a new update, we break out of the while loop, reset the condition variable,
+	///       // and persist the ChannelManager to disk/backups.
+	///       while !*guard {
+	///         guard = cnd.wait(guard).unwrap();
+	///       }
+	///       *guard = false;
+	///       std::mem::drop(guard); // Don't hold the lock while persisting the ChannelManager.
+	///       // Persist the ChannelManager here.
+	///     }
+	///   });
+	/// }
+	/// ```
+	pub persistence_lock: Arc<(Mutex<bool>, Condvar)>,
+
 	keys_manager: K,
 
 	logger: L,
@@ -759,6 +794,8 @@ 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,
@@ -912,6 +949,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();
 		}
 	}
 
@@ -1312,6 +1350,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 								commitment_signed,
 							},
 						});
+						self.persist_updates();
 					},
 					None => {},
 				}
@@ -1706,6 +1745,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 									commitment_signed: commitment_msg,
 								},
 							});
+							self.persist_updates();
 						}
 					} else {
 						unreachable!();
@@ -2125,6 +2165,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 							}
 						});
 					}
+					self.persist_updates();
 					return Ok(())
 				},
 				Err(e) => {
@@ -2339,6 +2380,15 @@ 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();
@@ -2373,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) {
@@ -2411,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,
@@ -2706,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))
@@ -2797,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)
+			}
 		}
 	}
 
@@ -2940,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))
@@ -2989,6 +3051,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 								commitment_signed,
 							},
 						});
+						self.persist_updates();
 					}
 				},
 			}
@@ -3988,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,