Skip to content

Commit f212c5e

Browse files
SWvheerdenSWvheerden
committed
moved checking for buffer
1 parent 0fcb2b1 commit f212c5e

File tree

3 files changed

+42
-40
lines changed

3 files changed

+42
-40
lines changed

‎src/ln/msgs.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -586,8 +586,8 @@ pub trait RoutingMessageHandler : Send + Sync {
586586
fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result<bool, HandleError>;
587587
/// Handle some updates to the route graph that we learned due to an outbound failed payment.
588588
fn handle_htlc_fail_channel_update(&self, update: &HTLCFailChannelUpdate);
589-
///This returns a vec tuple of (ChannelAnnouncements, ChannelUpdates, NodeAnnouncements NodeAnnouncements). It also returns the id of the last channel passed through
590-
/// This function requires -1 to start at the beginning.
589+
///Gets a subset of the messages required to dump our routing table to a remote node, starting at the short_channel_id indicated by starting_point and including batch_amount entries
590+
/// This function will start iterating at 0 if the starting_point is < 0.
591591
fn get_next_announcements(&self, starting_point: i64, batch_amount: u8)->(Vec<(ChannelAnnouncement, ChannelUpdate, NodeAnnouncement, NodeAnnouncement)>, i64);
592592
}
593593

‎src/ln/peer_handler.rs

Lines changed: 39 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -264,16 +264,34 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
264264
Ok(())
265265
}
266266

267-
fn do_attempt_write_data(descriptor: &mut Descriptor, peer: &mut Peer) {
267+
fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
268+
macro_rules! encode_and_send_msg {
269+
($msg: expr, $msg_code: expr) => {
270+
{
271+
log_trace!(self, "Encoding and sending message of type {} to {}", $msg_code, log_pubkey!(peer.their_node_id.unwrap()));
272+
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
273+
}
274+
}
275+
}
268276
while !peer.awaiting_write_event {
269277
if {
278+
let should_be_reading = peer.pending_outbound_buffer.len() < 10;
279+
if (peer.last_synced_channel > -1) &&(should_be_reading){
280+
let all_messages_tuple = self.message_handler.route_handler.get_next_announcements(peer.last_synced_channel,(peer.pending_outbound_buffer.len()-10) as u8);
281+
for i in 0..all_messages_tuple.0.len(){
282+
encode_and_send_msg!(all_messages_tuple.0[i].0.clone(), 256);
283+
encode_and_send_msg!(all_messages_tuple.0[i].2.clone(), 257);
284+
encode_and_send_msg!(all_messages_tuple.0[i].3.clone(), 257);
285+
encode_and_send_msg!(all_messages_tuple.0[i].1.clone(), 258);
286+
}
287+
peer.last_synced_channel = all_messages_tuple.1;
288+
}
270289
let next_buff = match peer.pending_outbound_buffer.front() {
271290
None => return,
272291
Some(buff) => buff,
273292
};
274-
let should_be_reading = peer.pending_outbound_buffer.len() < 10;
275293

276-
let data_sent = descriptor.send_data(next_buff, peer.pending_outbound_buffer_first_msg_offset, should_be_reading);
294+
let data_sent = descriptor.send_data(&next_buff, peer.pending_outbound_buffer_first_msg_offset, should_be_reading);
277295
peer.pending_outbound_buffer_first_msg_offset += data_sent;
278296
if peer.pending_outbound_buffer_first_msg_offset == next_buff.len() { true } else { false }
279297
} {
@@ -300,7 +318,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
300318
None => panic!("Descriptor for write_event is not already known to PeerManager"),
301319
Some(peer) => {
302320
peer.awaiting_write_event = false;
303-
Self::do_attempt_write_data(descriptor, peer);
321+
self.do_attempt_write_data(descriptor, peer);
304322
}
305323
};
306324
Ok(())
@@ -351,22 +369,6 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
351369
peer.pending_read_buffer_pos = 0;
352370

353371
macro_rules! encode_and_send_msg {
354-
($msg: expr, $msg_code: expr) => {
355-
{
356-
encode_and_send_actual_msg!($msg, $msg_code);
357-
if peer.last_synced_channel > -1{
358-
let all_messages_tuple = self.message_handler.route_handler.get_next_announcements(peer.last_synced_channel,5);
359-
for i in 0..all_messages_tuple.0.len(){
360-
encode_and_send_actual_msg!(all_messages_tuple.0[i].0.clone(), 256);
361-
encode_and_send_actual_msg!(all_messages_tuple.0[i].2.clone(), 257);
362-
encode_and_send_actual_msg!(all_messages_tuple.0[i].3.clone(), 257);
363-
encode_and_send_actual_msg!(all_messages_tuple.0[i].1.clone(), 258);
364-
}
365-
peer.last_synced_channel = all_messages_tuple.1;
366-
}
367-
}
368-
}
369-
}macro_rules! encode_and_send_actual_msg {
370372
($msg: expr, $msg_code: expr) => {
371373
{
372374
log_trace!(self, "Encoding and sending message of type {} to {}", $msg_code, log_pubkey!(peer.their_node_id.unwrap()));
@@ -691,7 +693,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
691693
}
692694
}
693695

694-
Self::do_attempt_write_data(peer_descriptor, peer);
696+
self.do_attempt_write_data(peer_descriptor, peer);
695697

696698
peer.pending_outbound_buffer.len() > 10 // pause_read
697699
}
@@ -748,7 +750,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
748750
//TODO: Drop the pending channel? (or just let it timeout, but that sucks)
749751
});
750752
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 33)));
751-
Self::do_attempt_write_data(&mut descriptor, peer);
753+
self.do_attempt_write_data(&mut descriptor, peer);
752754
},
753755
MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
754756
log_trace!(self, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
@@ -758,7 +760,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
758760
//TODO: Drop the pending channel? (or just let it timeout, but that sucks)
759761
});
760762
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 32)));
761-
Self::do_attempt_write_data(&mut descriptor, peer);
763+
self.do_attempt_write_data(&mut descriptor, peer);
762764
},
763765
MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
764766
log_trace!(self, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
@@ -770,7 +772,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
770772
//they should just throw away this funding transaction
771773
});
772774
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 34)));
773-
Self::do_attempt_write_data(&mut descriptor, peer);
775+
self.do_attempt_write_data(&mut descriptor, peer);
774776
},
775777
MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
776778
log_trace!(self, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
@@ -781,7 +783,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
781783
//they should just throw away this funding transaction
782784
});
783785
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 35)));
784-
Self::do_attempt_write_data(&mut descriptor, peer);
786+
self.do_attempt_write_data(&mut descriptor, peer);
785787
},
786788
MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
787789
log_trace!(self, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
@@ -791,7 +793,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
791793
//TODO: Do whatever we're gonna do for handling dropped messages
792794
});
793795
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 36)));
794-
Self::do_attempt_write_data(&mut descriptor, peer);
796+
self.do_attempt_write_data(&mut descriptor, peer);
795797
},
796798
MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
797799
log_trace!(self, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
@@ -802,7 +804,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
802804
//they should just throw away this funding transaction
803805
});
804806
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 259)));
805-
Self::do_attempt_write_data(&mut descriptor, peer);
807+
self.do_attempt_write_data(&mut descriptor, peer);
806808
},
807809
MessageSendEvent::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref update_fee, ref commitment_signed } } => {
808810
log_trace!(self, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
@@ -830,7 +832,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
830832
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 134)));
831833
}
832834
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed, 132)));
833-
Self::do_attempt_write_data(&mut descriptor, peer);
835+
self.do_attempt_write_data(&mut descriptor, peer);
834836
},
835837
MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
836838
log_trace!(self, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
@@ -840,7 +842,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
840842
//TODO: Do whatever we're gonna do for handling dropped messages
841843
});
842844
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 133)));
843-
Self::do_attempt_write_data(&mut descriptor, peer);
845+
self.do_attempt_write_data(&mut descriptor, peer);
844846
},
845847
MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
846848
log_trace!(self, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
@@ -850,7 +852,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
850852
//TODO: Do whatever we're gonna do for handling dropped messages
851853
});
852854
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 39)));
853-
Self::do_attempt_write_data(&mut descriptor, peer);
855+
self.do_attempt_write_data(&mut descriptor, peer);
854856
},
855857
MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
856858
log_trace!(self, "Handling Shutdown event in peer_handler for node {} for channel {}",
@@ -860,7 +862,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
860862
//TODO: Do whatever we're gonna do for handling dropped messages
861863
});
862864
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 38)));
863-
Self::do_attempt_write_data(&mut descriptor, peer);
865+
self.do_attempt_write_data(&mut descriptor, peer);
864866
},
865867
MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
866868
log_trace!(self, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
@@ -870,7 +872,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
870872
//TODO: Do whatever we're gonna do for handling dropped messages
871873
});
872874
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 136)));
873-
Self::do_attempt_write_data(&mut descriptor, peer);
875+
self.do_attempt_write_data(&mut descriptor, peer);
874876
},
875877
MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
876878
log_trace!(self, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
@@ -892,7 +894,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
892894
}
893895
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
894896
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_update_msg[..]));
895-
Self::do_attempt_write_data(&mut (*descriptor).clone(), peer);
897+
self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
896898
}
897899
}
898900
},
@@ -906,7 +908,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
906908
continue
907909
}
908910
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
909-
Self::do_attempt_write_data(&mut (*descriptor).clone(), peer);
911+
self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
910912
}
911913
}
912914
},
@@ -927,7 +929,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
927929
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
928930
// This isn't guaranteed to work, but if there is enough free
929931
// room in the send buffer, put the error message there...
930-
Self::do_attempt_write_data(&mut descriptor, &mut peer);
932+
self.do_attempt_write_data(&mut descriptor, &mut peer);
931933
} else {
932934
log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with no message", log_pubkey!(node_id));
933935
}
@@ -945,7 +947,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
945947
//TODO: Do whatever we're gonna do for handling dropped messages
946948
});
947949
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
948-
Self::do_attempt_write_data(&mut descriptor, peer);
950+
self.do_attempt_write_data(&mut descriptor, peer);
949951
},
950952
}
951953
} else {
@@ -957,7 +959,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
957959

958960
for mut descriptor in peers.peers_needing_send.drain() {
959961
match peers.peers.get_mut(&descriptor) {
960-
Some(peer) => Self::do_attempt_write_data(&mut descriptor, peer),
962+
Some(peer) => self.do_attempt_write_data(&mut descriptor, peer),
961963
None => panic!("Inconsistent peers set state!"),
962964
}
963965
}

‎src/ln/router.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -506,7 +506,7 @@ impl RoutingMessageHandler for Router {
506506
let mut result = Vec::new();
507507
let network = self.network_map.read().unwrap();
508508
let mut starting_for_next = starting_point;
509-
if starting_point == -1{
509+
if starting_point < 0{
510510
starting_for_next = 0;
511511
}
512512
let mut iter = network.channels.range((starting_for_next as u64)..);

0 commit comments

Comments
 (0)