X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fln%2Fpeer_handler.rs;h=4c2e68d9a43993a021c5a0bc796b548aed1a649e;hb=0456b0e311a4996b00d715f9211adcabc06b37c7;hp=92826244e09ee2b03f674e49c9e229e0a65a20ff;hpb=7aefa3131c6b0ee826ba4d86510745c9b2f95d1d;p=rust-lightning diff --git a/lightning/src/ln/peer_handler.rs b/lightning/src/ln/peer_handler.rs index 92826244..4c2e68d9 100644 --- a/lightning/src/ln/peer_handler.rs +++ b/lightning/src/ln/peer_handler.rs @@ -19,17 +19,20 @@ use bitcoin::blockdata::constants::ChainHash; use bitcoin::secp256k1::{self, Secp256k1, SecretKey, PublicKey}; use crate::sign::{KeysManager, NodeSigner, Recipient}; -use crate::events::{MessageSendEvent, MessageSendEventsProvider, OnionMessageProvider}; +use crate::events::{MessageSendEvent, MessageSendEventsProvider}; use crate::ln::ChannelId; use crate::ln::features::{InitFeatures, NodeFeatures}; use crate::ln::msgs; use crate::ln::msgs::{ChannelMessageHandler, LightningError, SocketAddress, OnionMessageHandler, RoutingMessageHandler}; +#[cfg(not(c_bindings))] use crate::ln::channelmanager::{SimpleArcChannelManager, SimpleRefChannelManager}; use crate::util::ser::{VecWriter, Writeable, Writer}; -use crate::ln::peer_channel_encryptor::{PeerChannelEncryptor,NextNoiseStep}; +use crate::ln::peer_channel_encryptor::{PeerChannelEncryptor, NextNoiseStep, MessageBuf, MSG_BUF_ALLOC_SIZE}; use crate::ln::wire; use crate::ln::wire::{Encode, Type}; -use crate::onion_message::{CustomOnionMessageContents, CustomOnionMessageHandler, OffersMessage, OffersMessageHandler, SimpleArcOnionMessenger, SimpleRefOnionMessenger}; +#[cfg(not(c_bindings))] +use crate::onion_message::{SimpleArcOnionMessenger, SimpleRefOnionMessenger}; +use crate::onion_message::{CustomOnionMessageHandler, OffersMessage, OffersMessageHandler, OnionMessageContents, PendingOnionMessage}; use crate::routing::gossip::{NetworkGraph, P2PGossipSync, NodeId, NodeAlias}; use crate::util::atomic_counter::AtomicCounter; use crate::util::logger::Logger; @@ -37,7 +40,7 @@ use crate::util::string::PrintableString; use crate::prelude::*; use crate::io; -use alloc::collections::LinkedList; +use alloc::collections::VecDeque; use crate::sync::{Arc, Mutex, MutexGuard, FairRwLock}; use core::sync::atomic::{AtomicBool, AtomicU32, AtomicI32, Ordering}; use core::{cmp, hash, fmt, mem}; @@ -107,11 +110,9 @@ impl RoutingMessageHandler for IgnoringMessageHandler { } fn processing_queue_high(&self) -> bool { false } } -impl OnionMessageProvider for IgnoringMessageHandler { - fn next_onion_message_for_peer(&self, _peer_node_id: PublicKey) -> Option { None } -} impl OnionMessageHandler for IgnoringMessageHandler { fn handle_onion_message(&self, _their_node_id: &PublicKey, _msg: &msgs::OnionMessage) {} + fn next_onion_message_for_peer(&self, _peer_node_id: PublicKey) -> Option { None } fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init, _inbound: bool) -> Result<(), ()> { Ok(()) } fn peer_disconnected(&self, _their_node_id: &PublicKey) {} fn provided_node_features(&self) -> NodeFeatures { NodeFeatures::empty() } @@ -131,9 +132,12 @@ impl CustomOnionMessageHandler for IgnoringMessageHandler { fn read_custom_message(&self, _msg_type: u64, _buffer: &mut R) -> Result, msgs::DecodeError> where Self: Sized { Ok(None) } + fn release_pending_custom_messages(&self) -> Vec> { + vec![] + } } -impl CustomOnionMessageContents for Infallible { +impl OnionMessageContents for Infallible { fn tlv_type(&self) -> u64 { unreachable!(); } } @@ -227,6 +231,18 @@ impl ChannelMessageHandler for ErroringMessageHandler { fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) { ErroringMessageHandler::push_error(self, their_node_id, msg.channel_id); } + fn handle_stfu(&self, their_node_id: &PublicKey, msg: &msgs::Stfu) { + ErroringMessageHandler::push_error(&self, their_node_id, msg.channel_id); + } + fn handle_splice(&self, their_node_id: &PublicKey, msg: &msgs::Splice) { + ErroringMessageHandler::push_error(&self, their_node_id, msg.channel_id); + } + fn handle_splice_ack(&self, their_node_id: &PublicKey, msg: &msgs::SpliceAck) { + ErroringMessageHandler::push_error(&self, their_node_id, msg.channel_id); + } + fn handle_splice_locked(&self, their_node_id: &PublicKey, msg: &msgs::SpliceLocked) { + ErroringMessageHandler::push_error(&self, their_node_id, msg.channel_id); + } fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) { ErroringMessageHandler::push_error(self, their_node_id, msg.channel_id); } @@ -279,7 +295,7 @@ impl ChannelMessageHandler for ErroringMessageHandler { features } - fn get_genesis_hashes(&self) -> Option> { + fn get_chain_hashes(&self) -> Option> { // We don't enforce any chains upon peer connection for `ErroringMessageHandler` and leave it up // to users of `ErroringMessageHandler` to make decisions on network compatiblility. // There's not really any way to pull in specific networks here, and hardcoding can cause breakages. @@ -485,13 +501,13 @@ struct Peer { their_features: Option, their_socket_address: Option, - pending_outbound_buffer: LinkedList>, + pending_outbound_buffer: VecDeque>, pending_outbound_buffer_first_msg_offset: usize, /// Queue gossip broadcasts separately from `pending_outbound_buffer` so we can easily /// prioritize channel messages over them. /// /// Note that these messages are *not* encrypted/MAC'd, and are only serialized. - gossip_broadcast_buffer: LinkedList>, + gossip_broadcast_buffer: VecDeque, awaiting_write_event: bool, pending_read_buffer: Vec, @@ -607,12 +623,13 @@ impl Peer { /// SimpleRefPeerManager is the more appropriate type. Defining these type aliases prevents /// issues such as overly long function definitions. /// -/// This is not exported to bindings users as `Arc`s don't make sense in bindings. +/// This is not exported to bindings users as type aliases aren't supported in most languages. +#[cfg(not(c_bindings))] pub type SimpleArcPeerManager = PeerManager< SD, Arc>, Arc>>, C, Arc>>, - Arc>, + Arc>, Arc, IgnoringMessageHandler, Arc @@ -625,14 +642,15 @@ pub type SimpleArcPeerManager = PeerManager< /// But if this is not necessary, using a reference is more efficient. Defining these type aliases /// helps with issues such as long function definitions. /// -/// This is not exported to bindings users as general type aliases don't make sense in bindings. +/// This is not exported to bindings users as type aliases aren't supported in most languages. +#[cfg(not(c_bindings))] pub type SimpleRefPeerManager< - 'a, 'b, 'c, 'd, 'e, 'f, 'logger, 'h, 'i, 'j, 'graph, SD, M, T, F, C, L + 'a, 'b, 'c, 'd, 'e, 'f, 'logger, 'h, 'i, 'j, 'graph, 'k, SD, M, T, F, C, L > = PeerManager< SD, &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'graph, 'logger, 'i, M, T, F, L>, &'f P2PGossipSync<&'graph NetworkGraph<&'logger L>, C, &'logger L>, - &'h SimpleRefOnionMessenger<'logger, 'i, 'j, L>, + &'h SimpleRefOnionMessenger<'a, 'b, 'c, 'd, 'e, 'graph, 'logger, 'i, 'j, 'k, M, T, F, L>, &'logger L, IgnoringMessageHandler, &'c KeysManager @@ -779,7 +797,7 @@ impl From for MessageHandlingError { macro_rules! encode_msg { ($msg: expr) => {{ - let mut buffer = VecWriter(Vec::new()); + let mut buffer = VecWriter(Vec::with_capacity(MSG_BUF_ALLOC_SIZE)); wire::write($msg, &mut buffer).unwrap(); buffer.0 }} @@ -991,9 +1009,9 @@ impl>(); + let large_capacity = peer.pending_outbound_buffer.capacity() > 4096 / VEC_SIZE; + let lots_of_slack = peer.pending_outbound_buffer.len() + < peer.pending_outbound_buffer.capacity() / 2; + if large_capacity && lots_of_slack { + peer.pending_outbound_buffer.shrink_to_fit(); + } } else { peer.awaiting_write_event = true; } @@ -1238,8 +1263,9 @@ impl) { + fn enqueue_encoded_gossip_broadcast(&self, peer: &mut Peer, encoded_message: MessageBuf) { peer.msgs_sent_since_pong += 1; + debug_assert!(peer.gossip_broadcast_buffer.len() <= OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP); peer.gossip_broadcast_buffer.push_back(encoded_message); } @@ -1367,7 +1393,7 @@ impl= 2); + debug_assert!(peer.pending_read_buffer.len() >= 2 + 16); + try_potential_handleerror!(peer, + peer.channel_encryptor.decrypt_message(&mut peer.pending_read_buffer[..])); + + let mut reader = io::Cursor::new(&peer.pending_read_buffer[..peer.pending_read_buffer.len() - 16]); + let message_result = wire::read(&mut reader, &*self.message_handler.custom_message_handler); // Reset read buffer if peer.pending_read_buffer.capacity() > 8192 { peer.pending_read_buffer = Vec::new(); } peer.pending_read_buffer.resize(18, 0); peer.pending_read_is_header = true; - let mut reader = io::Cursor::new(&msg_data[..]); - let message_result = wire::read(&mut reader, &*self.message_handler.custom_message_handler); let message = match message_result { Ok(x) => x, Err(e) => { @@ -1498,7 +1525,7 @@ impl { + self.message_handler.chan_handler.handle_stfu(&their_node_id, &msg); + } + + // Splicing messages: + wire::Message::Splice(msg) => { + self.message_handler.chan_handler.handle_splice(&their_node_id, &msg); + } + wire::Message::SpliceAck(msg) => { + self.message_handler.chan_handler.handle_splice_ack(&their_node_id, &msg); + } + wire::Message::SpliceLocked(msg) => { + self.message_handler.chan_handler.handle_splice_locked(&their_node_id, &msg); + } + // Interactive transaction construction messages: wire::Message::TxAddInput(msg) => { self.message_handler.chan_handler.handle_tx_add_input(&their_node_id, &msg); @@ -1785,7 +1828,7 @@ impl { @@ -1812,7 +1855,7 @@ impl { @@ -1834,7 +1877,7 @@ impl debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages"), @@ -1870,15 +1913,13 @@ impl { @@ -1956,6 +1997,30 @@ impl { + log_debug!(self.logger, "Handling SendStfu event in peer_handler for node {} for channel {}", + log_pubkey!(node_id), + &msg.channel_id); + self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg); + } + MessageSendEvent::SendSplice { ref node_id, ref msg} => { + log_debug!(self.logger, "Handling SendSplice event in peer_handler for node {} for channel {}", + log_pubkey!(node_id), + &msg.channel_id); + self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg); + } + MessageSendEvent::SendSpliceAck { ref node_id, ref msg} => { + log_debug!(self.logger, "Handling SendSpliceAck event in peer_handler for node {} for channel {}", + log_pubkey!(node_id), + &msg.channel_id); + self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg); + } + MessageSendEvent::SendSpliceLocked { ref node_id, ref msg} => { + log_debug!(self.logger, "Handling SendSpliceLocked event in peer_handler for node {} for channel {}", + log_pubkey!(node_id), + &msg.channel_id); + self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg); + } MessageSendEvent::SendTxAddInput { ref node_id, ref msg } => { log_debug!(self.logger, "Handling SendTxAddInput event in peer_handler for node {} for channel {}", log_pubkey!(node_id),