channel_encryptor: PeerChannelEncryptor,
outbound: bool,
their_node_id: Option<PublicKey>,
- their_global_features: Option<msgs::GlobalFeatures>,
- their_local_features: Option<msgs::LocalFeatures>,
+ their_features: Option<msgs::InitFeatures>,
pending_outbound_buffer: LinkedList<Vec<u8>>,
pending_outbound_buffer_first_msg_offset: usize,
pending_read_is_header: bool,
sync_status: InitSyncTracker,
+
+ awaiting_pong: bool,
}
impl Peer {
pub fn get_peer_node_ids(&self) -> Vec<PublicKey> {
let peers = self.peers.lock().unwrap();
peers.peers.values().filter_map(|p| {
- if !p.channel_encryptor.is_ready_for_encryption() || p.their_global_features.is_none() {
+ if !p.channel_encryptor.is_ready_for_encryption() || p.their_features.is_none() {
return None;
}
p.their_node_id
channel_encryptor: peer_encryptor,
outbound: true,
their_node_id: None,
- their_global_features: None,
- their_local_features: None,
+ their_features: None,
pending_outbound_buffer: LinkedList::new(),
pending_outbound_buffer_first_msg_offset: 0,
pending_read_is_header: false,
sync_status: InitSyncTracker::NoSyncRequested,
+
+ awaiting_pong: false,
}).is_some() {
panic!("PeerManager driver duplicated descriptors!");
};
channel_encryptor: peer_encryptor,
outbound: false,
their_node_id: None,
- their_global_features: None,
- their_local_features: None,
+ their_features: None,
pending_outbound_buffer: LinkedList::new(),
pending_outbound_buffer_first_msg_offset: 0,
pending_read_is_header: false,
sync_status: InitSyncTracker::NoSyncRequested,
+
+ awaiting_pong: false,
}).is_some() {
panic!("PeerManager driver duplicated descriptors!");
};
peer.their_node_id = Some(their_node_id);
insert_node_id!();
- let mut local_features = msgs::LocalFeatures::new();
+ let mut features = msgs::InitFeatures::new();
if self.initial_syncs_sent.load(Ordering::Acquire) < INITIAL_SYNCS_TO_SEND {
self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel);
- local_features.set_initial_routing_sync();
+ features.set_initial_routing_sync();
}
encode_and_send_msg!(msgs::Init {
- global_features: msgs::GlobalFeatures::new(),
- local_features,
+ features,
}, 16);
},
NextNoiseStep::ActThree => {
let msg_type = byte_utils::slice_to_be16(&msg_data[0..2]);
log_trace!(self, "Received message of type {} from {}", msg_type, log_pubkey!(peer.their_node_id.unwrap()));
- if msg_type != 16 && peer.their_global_features.is_none() {
+ if msg_type != 16 && peer.their_features.is_none() {
// Need an init message as first message
log_trace!(self, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
return Err(PeerHandleError{ no_connection_possible: false });
// Connection control:
16 => {
let msg = try_potential_decodeerror!(msgs::Init::read(&mut reader));
- if msg.global_features.requires_unknown_bits() {
+ if msg.features.requires_unknown_bits() {
log_info!(self, "Peer global features required unknown version bits");
return Err(PeerHandleError{ no_connection_possible: true });
}
- if msg.local_features.requires_unknown_bits() {
+ if msg.features.requires_unknown_bits() {
log_info!(self, "Peer local features required unknown version bits");
return Err(PeerHandleError{ no_connection_possible: true });
}
- if peer.their_global_features.is_some() {
+ if peer.their_features.is_some() {
return Err(PeerHandleError{ no_connection_possible: false });
}
log_info!(self, "Received peer Init message: data_loss_protect: {}, initial_routing_sync: {}, upfront_shutdown_script: {}, unkown local flags: {}, unknown global flags: {}",
- if msg.local_features.supports_data_loss_protect() { "supported" } else { "not supported"},
- if msg.local_features.initial_routing_sync() { "requested" } else { "not requested" },
- if msg.local_features.supports_upfront_shutdown_script() { "supported" } else { "not supported"},
- if msg.local_features.supports_unknown_bits() { "present" } else { "none" },
- if msg.global_features.supports_unknown_bits() { "present" } else { "none" });
+ if msg.features.supports_data_loss_protect() { "supported" } else { "not supported"},
+ if msg.features.initial_routing_sync() { "requested" } else { "not requested" },
+ if msg.features.supports_upfront_shutdown_script() { "supported" } else { "not supported"},
+ if msg.features.supports_unknown_bits() { "present" } else { "none" },
+ if msg.features.supports_unknown_bits() { "present" } else { "none" });
- if msg.local_features.initial_routing_sync() {
+ if msg.features.initial_routing_sync() {
peer.sync_status = InitSyncTracker::ChannelsSyncing(0);
peers.peers_needing_send.insert(peer_descriptor.clone());
}
- peer.their_global_features = Some(msg.global_features);
- peer.their_local_features = Some(msg.local_features);
+ peer.their_features = Some(msg.features);
if !peer.outbound {
- let mut local_features = msgs::LocalFeatures::new();
+ let mut features = msgs::InitFeatures::new();
if self.initial_syncs_sent.load(Ordering::Acquire) < INITIAL_SYNCS_TO_SEND {
self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel);
- local_features.set_initial_routing_sync();
+ features.set_initial_routing_sync();
}
encode_and_send_msg!(msgs::Init {
- global_features: msgs::GlobalFeatures::new(),
- local_features,
+ features,
}, 16);
}
}
},
19 => {
+ peer.awaiting_pong = false;
try_potential_decodeerror!(msgs::Pong::read(&mut reader));
},
-
// Channel control:
32 => {
let msg = try_potential_decodeerror!(msgs::OpenChannel::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_open_channel(&peer.their_node_id.unwrap(), peer.their_local_features.clone().unwrap(), &msg));
+ self.message_handler.chan_handler.handle_open_channel(&peer.their_node_id.unwrap(), peer.their_features.clone().unwrap(), &msg);
},
33 => {
let msg = try_potential_decodeerror!(msgs::AcceptChannel::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_accept_channel(&peer.their_node_id.unwrap(), peer.their_local_features.clone().unwrap(), &msg));
+ self.message_handler.chan_handler.handle_accept_channel(&peer.their_node_id.unwrap(), peer.their_features.clone().unwrap(), &msg);
},
34 => {
let msg = try_potential_decodeerror!(msgs::FundingCreated::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_funding_created(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_funding_created(&peer.their_node_id.unwrap(), &msg);
},
35 => {
let msg = try_potential_decodeerror!(msgs::FundingSigned::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_funding_signed(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_funding_signed(&peer.their_node_id.unwrap(), &msg);
},
36 => {
let msg = try_potential_decodeerror!(msgs::FundingLocked::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_funding_locked(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_funding_locked(&peer.their_node_id.unwrap(), &msg);
},
38 => {
let msg = try_potential_decodeerror!(msgs::Shutdown::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg);
},
39 => {
let msg = try_potential_decodeerror!(msgs::ClosingSigned::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg);
},
128 => {
let msg = try_potential_decodeerror!(msgs::UpdateAddHTLC::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_update_add_htlc(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_update_add_htlc(&peer.their_node_id.unwrap(), &msg);
},
130 => {
let msg = try_potential_decodeerror!(msgs::UpdateFulfillHTLC::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fulfill_htlc(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_update_fulfill_htlc(&peer.their_node_id.unwrap(), &msg);
},
131 => {
let msg = try_potential_decodeerror!(msgs::UpdateFailHTLC::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg);
},
135 => {
let msg = try_potential_decodeerror!(msgs::UpdateFailMalformedHTLC::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg);
},
132 => {
let msg = try_potential_decodeerror!(msgs::CommitmentSigned::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_commitment_signed(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_commitment_signed(&peer.their_node_id.unwrap(), &msg);
},
133 => {
let msg = try_potential_decodeerror!(msgs::RevokeAndACK::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg);
},
134 => {
let msg = try_potential_decodeerror!(msgs::UpdateFee::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fee(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_update_fee(&peer.their_node_id.unwrap(), &msg);
},
136 => {
let msg = try_potential_decodeerror!(msgs::ChannelReestablish::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg);
},
// Routing control:
259 => {
let msg = try_potential_decodeerror!(msgs::AnnouncementSignatures::read(&mut reader));
- try_potential_handleerror!(self.message_handler.chan_handler.handle_announcement_signatures(&peer.their_node_id.unwrap(), &msg));
+ self.message_handler.chan_handler.handle_announcement_signatures(&peer.their_node_id.unwrap(), &msg);
},
256 => {
let msg = try_potential_decodeerror!(msgs::ChannelAnnouncement::read(&mut reader));
};
match peers.peers.get_mut(&descriptor) {
Some(peer) => {
- if peer.their_global_features.is_none() {
+ if peer.their_features.is_none() {
$handle_no_such_peer;
continue;
}
let encoded_update_msg = encode_msg!(update_msg, 258);
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
- if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() ||
+ if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
!peer.should_forward_channel(msg.contents.short_channel_id) {
continue
}
let encoded_msg = encode_msg!(msg, 258);
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
- if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() ||
+ if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
!peer.should_forward_channel(msg.contents.short_channel_id) {
continue
}
}
};
}
+
+ /// This function should be called roughly once every 30 seconds.
+ /// It will send pings to each peer and disconnect those which did not respond to the last round of pings.
+
+ /// Will most likely call send_data on all of the registered descriptors, thus, be very careful with reentrancy issues!
+ pub fn timer_tick_occured(&self) {
+ let mut peers_lock = self.peers.lock().unwrap();
+ {
+ let peers = peers_lock.borrow_parts();
+ let peers_needing_send = peers.peers_needing_send;
+ let node_id_to_descriptor = peers.node_id_to_descriptor;
+ let peers = peers.peers;
+
+ peers.retain(|descriptor, peer| {
+ if peer.awaiting_pong == true {
+ peers_needing_send.remove(descriptor);
+ match peer.their_node_id {
+ Some(node_id) => {
+ node_id_to_descriptor.remove(&node_id);
+ self.message_handler.chan_handler.peer_disconnected(&node_id, true);
+ },
+ None => {}
+ }
+ }
+
+ let ping = msgs::Ping {
+ ponglen: 0,
+ byteslen: 64,
+ };
+ peer.pending_outbound_buffer.push_back(encode_msg!(ping, 18));
+ let mut descriptor_clone = descriptor.clone();
+ self.do_attempt_write_data(&mut descriptor_clone, peer);
+
+ if peer.awaiting_pong {
+ false // Drop the peer
+ } else {
+ peer.awaiting_pong = true;
+ true
+ }
+ });
+ }
+ }
}
#[cfg(test)]
peers[0].process_events();
assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
}
+ #[test]
+ fn test_timer_tick_occured(){
+ // Create peers, a vector of two peer managers, perform initial set up and check that peers[0] has one Peer.
+ let peers = create_network(2);
+ establish_connection(&peers[0], &peers[1]);
+ assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
+
+ // peers[0] awaiting_pong is set to true, but the Peer is still connected
+ peers[0].timer_tick_occured();
+ assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
+
+ // Since timer_tick_occured() is called again when awaiting_pong is true, all Peers are disconnected
+ peers[0].timer_tick_occured();
+ assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
+ }
}
projects / rust-lightning / blobdiff