// 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.
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.");
/// could setup `process_events_async` like this:
/// ```
/// # use lightning::io;
-/// # use std::sync::{Arc, Mutex};
+/// # use std::sync::{Arc, RwLock};
/// # use std::sync::atomic::{AtomicBool, Ordering};
/// # use lightning_background_processor::{process_events_async, GossipSync};
/// # struct MyStore {}
/// # impl lightning::util::persist::KVStore for MyStore {
-/// # fn read(&self, namespace: &str, sub_namespace: &str, key: &str) -> io::Result<Vec<u8>> { Ok(Vec::new()) }
-/// # fn write(&self, namespace: &str, sub_namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
-/// # fn remove(&self, namespace: &str, sub_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) }
-/// # fn list(&self, namespace: &str, sub_namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
+/// # fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> io::Result<Vec<u8>> { Ok(Vec::new()) }
+/// # fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) }
+/// # fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) }
+/// # fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> { Ok(Vec::new()) }
/// # }
/// # struct MyEventHandler {}
/// # impl MyEventHandler {
/// # type MyFilter = dyn lightning::chain::Filter + Send + Sync;
/// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync;
/// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::sign::InMemorySigner, Arc<MyFilter>, Arc<MyBroadcaster>, Arc<MyFeeEstimator>, Arc<MyLogger>, Arc<MyStore>>;
-/// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyUtxoLookup, MyLogger>;
+/// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<MySocketDescriptor, MyChainMonitor, MyBroadcaster, MyFeeEstimator, Arc<MyUtxoLookup>, MyLogger>;
/// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph<Arc<MyLogger>>;
/// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync<Arc<MyNetworkGraph>, Arc<MyUtxoLookup>, Arc<MyLogger>>;
/// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<MyChainMonitor, MyBroadcaster, MyFeeEstimator, MyLogger>;
-/// # type MyScorer =
Mutex<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
+/// # type MyScorer =
RwLock<lightning::routing::scoring::ProbabilisticScorer<Arc<MyNetworkGraph>, Arc<MyLogger>>>;
///
/// # async fn setup_background_processing(my_persister: Arc<MyStore>, my_event_handler: Arc<MyEventHandler>, my_chain_monitor: Arc<MyChainMonitor>, my_channel_manager: Arc<MyChannelManager>, my_gossip_sync: Arc<MyGossipSync>, my_logger: Arc<MyLogger>, my_scorer: Arc<MyScorer>, my_peer_manager: Arc<MyPeerManager>) {
/// let background_persister = Arc::clone(&my_persister);
EventHandlerFuture: core::future::Future<Output = ()>,
EventHandler: Fn(Event) -> EventHandlerFuture,
PS: 'static + Deref + Send,
- M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
+ M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::EcdsaSigner, CF, T, F, L, P>> + Send + Sync,
CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
where
UL::Target: 'static + UtxoLookup,
CF::Target: 'static + chain::Filter,
- CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
+ CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
T::Target: 'static + BroadcasterInterface,
ES::Target: 'static + EntropySource,
NS::Target: 'static + NodeSigner,
F::Target: 'static + FeeEstimator,
R::Target: 'static + Router,
L::Target: 'static + Logger,
- P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
+ P::Target: 'static + Persist<<SP::Target as SignerProvider>::EcdsaSigner>,
PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
{
let mut should_break = false;
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)),
P: 'static + Deref + Send + Sync,
EH: 'static + EventHandler + Send,
PS: 'static + Deref + Send,
- M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::Signer, CF, T, F, L, P>> + Send + Sync,
+ M: 'static + Deref<Target = ChainMonitor<<SP::Target as SignerProvider>::EcdsaSigner, CF, T, F, L, P>> + Send + Sync,
CM: 'static + Deref<Target = ChannelManager<CW, T, ES, NS, SP, F, R, L>> + Send + Sync,
PGS: 'static + Deref<Target = P2PGossipSync<G, UL, L>> + Send + Sync,
RGS: 'static + Deref<Target = RapidGossipSync<G, L>> + Send,
where
UL::Target: 'static + UtxoLookup,
CF::Target: 'static + chain::Filter,
- CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::Signer>,
+ CW::Target: 'static + chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
T::Target: 'static + BroadcasterInterface,
ES::Target: 'static + EntropySource,
NS::Target: 'static + NodeSigner,
F::Target: 'static + FeeEstimator,
R::Target: 'static + Router,
L::Target: 'static + Logger,
- P::Target: 'static + Persist<<SP::Target as SignerProvider>::Signer>,
+ P::Target: 'static + Persist<<SP::Target as SignerProvider>::EcdsaSigner>,
PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>,
{
let stop_thread = Arc::new(AtomicBool::new(false));
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) }
#[cfg(all(feature = "std", test))]
mod tests {
use bitcoin::blockdata::constants::{genesis_block, ChainHash};
- use bitcoin::blockdata::locktime::PackedLockTime;
+ use bitcoin::blockdata::locktime::absolute::LockTime;
use bitcoin::blockdata::transaction::{Transaction, TxOut};
use bitcoin::network::constants::Network;
use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1};
use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync};
use lightning::routing::router::{DefaultRouter, Path, RouteHop};
- use lightning::routing::scoring::{ChannelUsage, ScoreUpdate, ScoreLookUp};
+ use lightning::routing::scoring::{ChannelUsage, ScoreUpdate, ScoreLookUp, LockableScore};
use lightning::util::config::UserConfig;
use lightning::util::ser::Writeable;
use lightning::util::test_utils;
- use lightning::util::persist::{KVStore, CHANNEL_MANAGER_PERSISTENCE_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SUB_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SUB_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY, SCORER_PERSISTENCE_NAMESPACE, SCORER_PERSISTENCE_SUB_NAMESPACE, SCORER_PERSISTENCE_KEY};
+ use lightning::util::persist::{KVStore,
+ CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY,
+ NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY,
+ SCORER_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE, SCORER_PERSISTENCE_KEY};
use lightning_persister::fs_store::FilesystemStore;
use std::collections::VecDeque;
use std::{fs, env};
fn disconnect_socket(&mut self) {}
}
+ #[cfg(c_bindings)]
+ type LockingWrapper<T> = lightning::routing::scoring::MultiThreadedLockableScore<T>;
+ #[cfg(not(c_bindings))]
+ type LockingWrapper<T> = Mutex<T>;
+
type ChannelManager =
channelmanager::ChannelManager<
Arc<ChainMonitor>,
Arc<DefaultRouter<
Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
Arc<test_utils::TestLogger>,
- Arc<Mutex<TestScorer>>,
+ Arc<LockingWrapper<TestScorer>>,
(),
TestScorer>
>,
network_graph: Arc<NetworkGraph<Arc<test_utils::TestLogger>>>,
logger: Arc<test_utils::TestLogger>,
best_block: BestBlock,
- scorer: Arc<Mutex<TestScorer>>,
+ scorer: Arc<LockingWrapper<TestScorer>>,
}
impl Node {
}
impl KVStore for Persister {
- fn read(&self, namespace: &str, sub_namespace: &str, key: &str) -> lightning::io::Result<Vec<u8>> {
- self.kv_store.read(namespace, sub_namespace, key)
+ fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> lightning::io::Result<Vec<u8>> {
+ self.kv_store.read(primary_namespace, secondary_namespace, key)
}
- fn write(&self, namespace: &str, sub_namespace: &str, key: &str, buf: &[u8]) -> lightning::io::Result<()> {
- if namespace == CHANNEL_MANAGER_PERSISTENCE_NAMESPACE &&
- sub_namespace == CHANNEL_MANAGER_PERSISTENCE_SUB_NAMESPACE &&
+ fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> lightning::io::Result<()> {
+ if primary_namespace == CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE &&
+ secondary_namespace == CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE &&
key == CHANNEL_MANAGER_PERSISTENCE_KEY
{
if let Some((error, message)) = self.manager_error {
}
}
- if namespace == NETWORK_GRAPH_PERSISTENCE_NAMESPACE &&
- sub_namespace == NETWORK_GRAPH_PERSISTENCE_SUB_NAMESPACE &&
+ if primary_namespace == NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE &&
+ secondary_namespace == NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE &&
key == NETWORK_GRAPH_PERSISTENCE_KEY
{
if let Some(sender) = &self.graph_persistence_notifier {
}
}
- if namespace == SCORER_PERSISTENCE_NAMESPACE &&
- sub_namespace == SCORER_PERSISTENCE_SUB_NAMESPACE &&
+ if primary_namespace == SCORER_PERSISTENCE_PRIMARY_NAMESPACE &&
+ secondary_namespace == SCORER_PERSISTENCE_SECONDARY_NAMESPACE &&
key == SCORER_PERSISTENCE_KEY
{
if let Some((error, message)) = self.scorer_error {
}
}
- self.kv_store.write(namespace, sub_namespace, key, buf)
+ self.kv_store.write(primary_namespace, secondary_namespace, key, buf)
}
- fn remove(&self, namespace: &str, sub_namespace: &str, key: &str, lazy: bool) -> lightning::io::Result<()> {
- self.kv_store.remove(namespace, sub_namespace, key, lazy)
+ fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> lightning::io::Result<()> {
+ self.kv_store.remove(primary_namespace, secondary_namespace, key, lazy)
}
- fn list(&self, namespace: &str, sub_namespace: &str) -> lightning::io::Result<Vec<String>> {
- self.kv_store.list(namespace, sub_namespace)
+ fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> lightning::io::Result<Vec<String>> {
+ self.kv_store.list(primary_namespace, secondary_namespace)
}
}
}
}
+ #[cfg(c_bindings)]
+ impl lightning::routing::scoring::Score for TestScorer {}
+
impl Drop for TestScorer {
fn drop(&mut self) {
if std::thread::panicking() {
let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
let genesis_block = genesis_block(network);
let network_graph = Arc::new(NetworkGraph::new(network, logger.clone()));
- let scorer = Arc::new(Mutex::new(TestScorer::new()));
+ let scorer = Arc::new(LockingWrapper::new(TestScorer::new()));
let seed = [i as u8; 32];
- let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), ()));
+ let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), Default::default()));
let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
let kv_store = Arc::new(FilesystemStore::new(format!("{}_persister_{}", &persist_dir, i).into()));
let now = Duration::from_secs(genesis_block.header.time as u64);
macro_rules! begin_open_channel {
($node_a: expr, $node_b: expr, $channel_value: expr) => {{
- $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap();
+ $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None, None).unwrap();
$node_b.node.handle_open_channel(&$node_a.node.get_our_node_id(), &get_event_msg!($node_a, MessageSendEvent::SendOpenChannel, $node_b.node.get_our_node_id()));
$node_a.node.handle_accept_channel(&$node_b.node.get_our_node_id(), &get_event_msg!($node_b, MessageSendEvent::SendAcceptChannel, $node_a.node.get_our_node_id()));
}}
assert_eq!(channel_value_satoshis, $channel_value);
assert_eq!(user_channel_id, 42);
- let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut {
+ let tx = Transaction { version: 1 as i32, lock_time: LockTime::ZERO, input: Vec::new(), output: vec![TxOut {
value: channel_value_satoshis, script_pubkey: output_script.clone(),
}]};
(temporary_channel_id, tx)
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.
// 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
channel_features: ChannelFeatures::empty(),
fee_msat: 0,
cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32,
+ maybe_announced_channel: true,
}], blinded_tail: None };
- $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
+ $nodes[0].scorer.write_lock().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid });
$nodes[0].node.push_pending_event(Event::PaymentPathFailed {
payment_id: None,
payment_hash: PaymentHash([42; 32]),
// Ensure we'll score payments that were explicitly failed back by the destination as
// ProbeSuccess.
- $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
+ $nodes[0].scorer.write_lock().expect(TestResult::ProbeSuccess { path: path.clone() });
$nodes[0].node.push_pending_event(Event::PaymentPathFailed {
payment_id: None,
payment_hash: PaymentHash([42; 32]),
_ => panic!("Unexpected event"),
}
- $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() });
+ $nodes[0].scorer.write_lock().expect(TestResult::PaymentSuccess { path: path.clone() });
$nodes[0].node.push_pending_event(Event::PaymentPathSuccessful {
payment_id: PaymentId([42; 32]),
payment_hash: None,
_ => panic!("Unexpected event"),
}
- $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() });
+ $nodes[0].scorer.write_lock().expect(TestResult::ProbeSuccess { path: path.clone() });
$nodes[0].node.push_pending_event(Event::ProbeSuccessful {
payment_id: PaymentId([42; 32]),
payment_hash: PaymentHash([42; 32]),
_ => panic!("Unexpected event"),
}
- $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() });
+ $nodes[0].scorer.write_lock().expect(TestResult::ProbeFailure { path: path.clone() });
$nodes[0].node.push_pending_event(Event::ProbeFailed {
payment_id: PaymentId([42; 32]),
payment_hash: PaymentHash([42; 32]),
projects / rust-lightning / blobdiff