/// 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;
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));
#[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)
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]),