blockworks-foundation
/
solana
mirror of https://github.com/blockworks-foundation/solana.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

renames Packet Meta::{,set_}addr methods to {,set_}socket_addr (#25478) 

In order to distinguish between Meta.addr field which is an IpAddr and
the methods which refer to a SocketAddr.
This commit is contained in:
behzad nouri 2022-05-23 15:48:59 +00:00 committed by GitHub
parent bac05dc55a
commit c248fb3f51
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 52 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
bench-streamer/src/main.rs
View File

@ -25,7 +25,7 @@ fn producer(addr: &SocketAddr, exit: Arc<AtomicBool>) -> JoinHandle<()> {
packet_batch.packets.resize(10, Packet::default()); packet_batch.packets.resize(10, Packet::default());
for w in packet_batch.packets.iter_mut() { for w in packet_batch.packets.iter_mut() {
w.meta.size = PACKET_DATA_SIZE; w.meta.size = PACKET_DATA_SIZE;
w.meta.set_addr(addr); w.meta.set_socket_addr(addr);
} }
let packet_batch = Arc::new(packet_batch); let packet_batch = Arc::new(packet_batch);
spawn(move || loop { spawn(move || loop {
@ -34,7 +34,7 @@ fn producer(addr: &SocketAddr, exit: Arc<AtomicBool>) -> JoinHandle<()> {
} }
let mut num = 0; let mut num = 0;
for p in &packet_batch.packets { for p in &packet_batch.packets {
let a = p.meta.addr(); let a = p.meta.socket_addr();
assert!(p.meta.size <= PACKET_DATA_SIZE); assert!(p.meta.size <= PACKET_DATA_SIZE);
send.send_to(&p.data[..p.meta.size], &a).unwrap(); send.send_to(&p.data[..p.meta.size], &a).unwrap();
num += 1; num += 1;

6
core/src/ancestor_hashes_service.rs
View File

@ -321,7 +321,7 @@ impl AncestorHashesService {
outstanding_requests: &RwLock<OutstandingAncestorHashesRepairs>, outstanding_requests: &RwLock<OutstandingAncestorHashesRepairs>,
blockstore: &Blockstore, blockstore: &Blockstore,
) -> Option<(Slot, DuplicateAncestorDecision)> { ) -> Option<(Slot, DuplicateAncestorDecision)> {
let from_addr = packet.meta.addr(); let from_addr = packet.meta.socket_addr();
limited_deserialize(&packet.data[..packet.meta.size.saturating_sub(SIZE_OF_NONCE)]) limited_deserialize(&packet.data[..packet.meta.size.saturating_sub(SIZE_OF_NONCE)])
.ok() .ok()
.and_then(|ancestor_hashes_response| { .and_then(|ancestor_hashes_response| {
@ -1117,7 +1117,7 @@ mod test {
.recv_timeout(Duration::from_millis(10_000)) .recv_timeout(Duration::from_millis(10_000))
.unwrap(); .unwrap();
let packet = &mut response_packet.packets[0]; let packet = &mut response_packet.packets[0];
packet.meta.set_addr(&responder_info.serve_repair); packet.meta.set_socket_addr(&responder_info.serve_repair);
let decision = AncestorHashesService::verify_and_process_ancestor_response( let decision = AncestorHashesService::verify_and_process_ancestor_response(
packet, packet,
&ancestor_hashes_request_statuses, &ancestor_hashes_request_statuses,
@ -1478,7 +1478,7 @@ mod test {
.recv_timeout(Duration::from_millis(10_000)) .recv_timeout(Duration::from_millis(10_000))
.unwrap(); .unwrap();
let packet = &mut response_packet.packets[0]; let packet = &mut response_packet.packets[0];
packet.meta.set_addr(&responder_info.serve_repair); packet.meta.set_socket_addr(&responder_info.serve_repair);
let decision = AncestorHashesService::verify_and_process_ancestor_response( let decision = AncestorHashesService::verify_and_process_ancestor_response(
packet, packet,
&ancestor_hashes_request_statuses, &ancestor_hashes_request_statuses,

6
core/src/find_packet_sender_stake_stage.rs
View File

@ -135,8 +135,10 @@ impl FindPacketSenderStakeStage {
.into_par_iter() .into_par_iter()
.flat_map(|batch| batch.packets.par_iter_mut()) .flat_map(|batch| batch.packets.par_iter_mut())
.for_each(|packet| { .for_each(|packet| {
packet.meta.sender_stake = packet.meta.sender_stake = ip_to_stake
*ip_to_stake.get(&packet.meta.addr().ip()).unwrap_or(&0); .get(&packet.meta.addr)
.copied()
.unwrap_or_default();
}); });
}); });
} }

2
core/src/repair_response.rs
View File

@ -33,7 +33,7 @@ pub fn repair_response_packet_from_bytes(
if packet.meta.size > packet.data.len() { if packet.meta.size > packet.data.len() {
return None; return None;
} }
packet.meta.set_addr(dest); packet.meta.set_socket_addr(dest);
packet.data[..bytes.len()].copy_from_slice(&bytes); packet.data[..bytes.len()].copy_from_slice(&bytes);
let mut wr = io::Cursor::new(&mut packet.data[bytes.len()..]); let mut wr = io::Cursor::new(&mut packet.data[bytes.len()..]);
bincode::serialize_into(&mut wr, &nonce).expect("Buffer not large enough to fit nonce"); bincode::serialize_into(&mut wr, &nonce).expect("Buffer not large enough to fit nonce");

2
core/src/serve_repair.rs
View File

@ -432,7 +432,7 @@ impl ServeRepair {
) { ) {
// iter over the packets // iter over the packets
packet_batch.packets.iter().for_each(|packet| { packet_batch.packets.iter().for_each(|packet| {
let from_addr = packet.meta.addr(); let from_addr = packet.meta.socket_addr();
limited_deserialize(&packet.data[..packet.meta.size]) limited_deserialize(&packet.data[..packet.meta.size])
.into_iter() .into_iter()
.for_each(|request| { .for_each(|request| {

5
core/src/window_service.rs
View File

@ -374,7 +374,7 @@ where
} }
if packet.meta.repair() { if packet.meta.repair() {
let repair_info = RepairMeta { let repair_info = RepairMeta {
_from_addr: packet.meta.addr(), _from_addr: packet.meta.socket_addr(),
// If can't parse the nonce, dump the packet. // If can't parse the nonce, dump the packet.
nonce: repair_response::nonce(&packet.data)?, nonce: repair_response::nonce(&packet.data)?,
}; };
@ -408,7 +408,8 @@ where
stats.num_packets += packets.iter().map(|pkt| pkt.packets.len()).sum::<usize>(); stats.num_packets += packets.iter().map(|pkt| pkt.packets.len()).sum::<usize>();
for packet in packets.iter().flat_map(|pkt| pkt.packets.iter()) { for packet in packets.iter().flat_map(|pkt| pkt.packets.iter()) {
*stats.addrs.entry(packet.meta.addr()).or_default() += 1; let addr = packet.meta.socket_addr();
*stats.addrs.entry(addr).or_default() += 1;
} }
stats.elapsed += now.elapsed(); stats.elapsed += now.elapsed();
Ok(()) Ok(())

6
gossip/src/cluster_info.rs
View File

@ -1901,7 +1901,7 @@ impl ClusterInfo {
} }
check check
}; };
// Because pull-responses are sent back to packet.meta.addr() of // Because pull-responses are sent back to packet.meta.socket_addr() of
// incoming pull-requests, pings are also sent to request.from_addr (as // incoming pull-requests, pings are also sent to request.from_addr (as
// opposed to caller.gossip address). // opposed to caller.gossip address).
move |request| { move |request| {
@ -2470,7 +2470,7 @@ impl ClusterInfo {
let protocol: Protocol = limited_deserialize(data).ok()?; let protocol: Protocol = limited_deserialize(data).ok()?;
protocol.sanitize().ok()?; protocol.sanitize().ok()?;
let protocol = protocol.par_verify(&self.stats)?; let protocol = protocol.par_verify(&self.stats)?;
Some((packet.meta.addr(), protocol)) Some((packet.meta.socket_addr(), protocol))
}; };
let packets: Vec<_> = { let packets: Vec<_> = {
let _st = ScopedTimer::from(&self.stats.verify_gossip_packets_time); let _st = ScopedTimer::from(&self.stats.verify_gossip_packets_time);
@ -3234,7 +3234,7 @@ mod tests {
remote_nodes.into_iter(), remote_nodes.into_iter(),
pongs.into_iter() pongs.into_iter()
) { ) {
assert_eq!(packet.meta.addr(), socket); assert_eq!(packet.meta.socket_addr(), socket);
let bytes = serialize(&pong).unwrap(); let bytes = serialize(&pong).unwrap();
match limited_deserialize(&packet.data[..packet.meta.size]).unwrap() { match limited_deserialize(&packet.data[..packet.meta.size]).unwrap() {
Protocol::PongMessage(pong) => assert_eq!(serialize(&pong).unwrap(), bytes), Protocol::PongMessage(pong) => assert_eq!(serialize(&pong).unwrap(), bytes),

2
perf/src/packet.rs
View File

@ -98,7 +98,7 @@ impl PacketBatch {
pub fn set_addr(&mut self, addr: &SocketAddr) { pub fn set_addr(&mut self, addr: &SocketAddr) {
for p in self.packets.iter_mut() { for p in self.packets.iter_mut() {
p.meta.set_addr(addr); p.meta.set_socket_addr(addr);
} }
} }

18
sdk/src/packet.rs
View File

@ -17,11 +17,11 @@ pub const PACKET_DATA_SIZE: usize = 1280 - 40 - 8;
bitflags! { bitflags! {
#[repr(C)] #[repr(C)]
pub struct PacketFlags: u8 { pub struct PacketFlags: u8 {
const DISCARD = 0b00000001; const DISCARD = 0b0000_0001;
const FORWARDED = 0b00000010; const FORWARDED = 0b0000_0010;
const REPAIR = 0b00000100; const REPAIR = 0b0000_0100;
const SIMPLE_VOTE_TX = 0b00001000; const SIMPLE_VOTE_TX = 0b0000_1000;
const TRACER_TX = 0b00010000; const TRACER_TX = 0b0001_0000;
} }
} }
@ -63,7 +63,7 @@ impl Packet {
let len = wr.position() as usize; let len = wr.position() as usize;
packet.meta.size = len; packet.meta.size = len;
if let Some(dest) = dest { if let Some(dest) = dest {
packet.meta.set_addr(dest); packet.meta.set_socket_addr(dest);
} }
Ok(()) Ok(())
} }
@ -75,7 +75,7 @@ impl fmt::Debug for Packet {
f, f,
"Packet {{ size: {:?}, addr: {:?} }}", "Packet {{ size: {:?}, addr: {:?} }}",
self.meta.size, self.meta.size,
self.meta.addr() self.meta.socket_addr()
) )
} }
} }
@ -99,11 +99,11 @@ impl PartialEq for Packet {
} }
impl Meta { impl Meta {
pub fn addr(&self) -> SocketAddr { pub fn socket_addr(&self) -> SocketAddr {
SocketAddr::new(self.addr, self.port) SocketAddr::new(self.addr, self.port)
} }
pub fn set_addr(&mut self, socket_addr: &SocketAddr) { pub fn set_socket_addr(&mut self, socket_addr: &SocketAddr) {
self.addr = socket_addr.ip(); self.addr = socket_addr.ip();
self.port = socket_addr.port(); self.port = socket_addr.port();
} }

16
streamer/src/nonblocking/recvmmsg.rs
View File

@ -28,7 +28,7 @@ pub async fn recv_mmsg(
} }
Ok((nrecv, from)) => { Ok((nrecv, from)) => {
p.meta.size = nrecv; p.meta.size = nrecv;
p.meta.set_addr(&from); p.meta.set_socket_addr(&from);
} }
} }
i += 1; i += 1;
@ -68,7 +68,7 @@ mod tests {
assert_eq!(sent, recv); assert_eq!(sent, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr); assert_eq!(packet.meta.socket_addr(), saddr);
} }
} }
@ -94,7 +94,7 @@ mod tests {
assert_eq!(TEST_NUM_MSGS, recv); assert_eq!(TEST_NUM_MSGS, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr); assert_eq!(packet.meta.socket_addr(), saddr);
} }
packets packets
@ -104,7 +104,7 @@ mod tests {
assert_eq!(sent - TEST_NUM_MSGS, recv); assert_eq!(sent - TEST_NUM_MSGS, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr); assert_eq!(packet.meta.socket_addr(), saddr);
} }
} }
@ -136,7 +136,7 @@ mod tests {
assert_eq!(TEST_NUM_MSGS, recv); assert_eq!(TEST_NUM_MSGS, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr); assert_eq!(packet.meta.socket_addr(), saddr);
} }
packets packets
@ -175,11 +175,11 @@ mod tests {
assert_eq!(TEST_NUM_MSGS, recv); assert_eq!(TEST_NUM_MSGS, recv);
for packet in packets.iter().take(sent1) { for packet in packets.iter().take(sent1) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr1); assert_eq!(packet.meta.socket_addr(), saddr1);
} }
for packet in packets.iter().skip(sent1).take(recv - sent1) { for packet in packets.iter().skip(sent1).take(recv - sent1) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr2); assert_eq!(packet.meta.socket_addr(), saddr2);
} }
packets packets
@ -189,7 +189,7 @@ mod tests {
assert_eq!(sent1 + sent2 - TEST_NUM_MSGS, recv); assert_eq!(sent1 + sent2 - TEST_NUM_MSGS, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr2); assert_eq!(packet.meta.socket_addr(), saddr2);
} }
} }
} }

10
streamer/src/packet.rs
View File

@ -66,7 +66,7 @@ pub fn send_to(
socket_addr_space: &SocketAddrSpace, socket_addr_space: &SocketAddrSpace,
) -> Result<()> { ) -> Result<()> {
for p in &batch.packets { for p in &batch.packets {
let addr = p.meta.addr(); let addr = p.meta.socket_addr();
if socket_addr_space.check(&addr) { if socket_addr_space.check(&addr) {
socket.send_to(&p.data[..p.meta.size], &addr)?; socket.send_to(&p.data[..p.meta.size], &addr)?;
} }
@ -92,7 +92,7 @@ mod tests {
let packets = vec![Packet::default()]; let packets = vec![Packet::default()];
let mut packet_batch = PacketBatch::new(packets); let mut packet_batch = PacketBatch::new(packets);
packet_batch.set_addr(&send_addr); packet_batch.set_addr(&send_addr);
assert_eq!(packet_batch.packets[0].meta.addr(), send_addr); assert_eq!(packet_batch.packets[0].meta.socket_addr(), send_addr);
} }
#[test] #[test]
@ -107,7 +107,7 @@ mod tests {
batch.packets.resize(10, Packet::default()); batch.packets.resize(10, Packet::default());
for m in batch.packets.iter_mut() { for m in batch.packets.iter_mut() {
m.meta.set_addr(&addr); m.meta.set_socket_addr(&addr);
m.meta.size = PACKET_DATA_SIZE; m.meta.size = PACKET_DATA_SIZE;
} }
send_to(&batch, &send_socket, &SocketAddrSpace::Unspecified).unwrap(); send_to(&batch, &send_socket, &SocketAddrSpace::Unspecified).unwrap();
@ -122,7 +122,7 @@ mod tests {
for m in &batch.packets { for m in &batch.packets {
assert_eq!(m.meta.size, PACKET_DATA_SIZE); assert_eq!(m.meta.size, PACKET_DATA_SIZE);
assert_eq!(m.meta.addr(), saddr); assert_eq!(m.meta.socket_addr(), saddr);
} }
} }
@ -164,7 +164,7 @@ mod tests {
let mut batch = PacketBatch::default(); let mut batch = PacketBatch::default();
batch.packets.resize(1, Packet::default()); batch.packets.resize(1, Packet::default());
for m in batch.packets.iter_mut() { for m in batch.packets.iter_mut() {
m.meta.set_addr(&addr); m.meta.set_socket_addr(&addr);
m.meta.size = 1; m.meta.size = 1;
} }
send_to(&batch, &send_socket, &SocketAddrSpace::Unspecified).unwrap(); send_to(&batch, &send_socket, &SocketAddrSpace::Unspecified).unwrap();

2
streamer/src/quic.rs
View File

@ -178,7 +178,7 @@ fn handle_chunk(
if maybe_batch.is_none() { if maybe_batch.is_none() {
let mut batch = PacketBatch::with_capacity(1); let mut batch = PacketBatch::with_capacity(1);
let mut packet = Packet::default(); let mut packet = Packet::default();
packet.meta.set_addr(remote_addr); packet.meta.set_socket_addr(remote_addr);
packet.meta.sender_stake = stake; packet.meta.sender_stake = stake;
batch.packets.push(packet); batch.packets.push(packet);
*maybe_batch = Some(batch); *maybe_batch = Some(batch);

18
streamer/src/recvmmsg.rs
View File

@ -31,7 +31,7 @@ pub fn recv_mmsg(socket: &UdpSocket, packets: &mut [Packet]) -> io::Result</*num
} }
Ok((nrecv, from)) => { Ok((nrecv, from)) => {
p.meta.size = nrecv; p.meta.size = nrecv;
p.meta.set_addr(&from); p.meta.set_socket_addr(&from);
if i == 0 { if i == 0 {
socket.set_nonblocking(true)?; socket.set_nonblocking(true)?;
} }
@ -107,7 +107,7 @@ pub fn recv_mmsg(sock: &UdpSocket, packets: &mut [Packet]) -> io::Result</*num p
for (addr, hdr, pkt) in izip!(addrs, hdrs, packets.iter_mut()).take(nrecv) { for (addr, hdr, pkt) in izip!(addrs, hdrs, packets.iter_mut()).take(nrecv) {
pkt.meta.size = hdr.msg_len as usize; pkt.meta.size = hdr.msg_len as usize;
if let Some(addr) = cast_socket_addr(&addr, &hdr) { if let Some(addr) = cast_socket_addr(&addr, &hdr) {
pkt.meta.set_addr(&addr.to_std()); pkt.meta.set_socket_addr(&addr.to_std());
} }
} }
Ok(nrecv) Ok(nrecv)
@ -148,7 +148,7 @@ mod tests {
assert_eq!(sent, recv); assert_eq!(sent, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr); assert_eq!(packet.meta.socket_addr(), saddr);
} }
}; };
@ -174,7 +174,7 @@ mod tests {
assert_eq!(TEST_NUM_MSGS, recv); assert_eq!(TEST_NUM_MSGS, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr); assert_eq!(packet.meta.socket_addr(), saddr);
} }
packets packets
@ -184,7 +184,7 @@ mod tests {
assert_eq!(sent - TEST_NUM_MSGS, recv); assert_eq!(sent - TEST_NUM_MSGS, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr); assert_eq!(packet.meta.socket_addr(), saddr);
} }
}; };
@ -216,7 +216,7 @@ mod tests {
assert_eq!(TEST_NUM_MSGS, recv); assert_eq!(TEST_NUM_MSGS, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr); assert_eq!(packet.meta.socket_addr(), saddr);
} }
reader.set_nonblocking(true).unwrap(); reader.set_nonblocking(true).unwrap();
@ -256,11 +256,11 @@ mod tests {
assert_eq!(TEST_NUM_MSGS, recv); assert_eq!(TEST_NUM_MSGS, recv);
for packet in packets.iter().take(sent1) { for packet in packets.iter().take(sent1) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr1); assert_eq!(packet.meta.socket_addr(), saddr1);
} }
for packet in packets.iter().skip(sent1).take(recv - sent1) { for packet in packets.iter().skip(sent1).take(recv - sent1) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr2); assert_eq!(packet.meta.socket_addr(), saddr2);
} }
packets packets
@ -270,7 +270,7 @@ mod tests {
assert_eq!(sent1 + sent2 - TEST_NUM_MSGS, recv); assert_eq!(sent1 + sent2 - TEST_NUM_MSGS, recv);
for packet in packets.iter().take(recv) { for packet in packets.iter().take(recv) {
assert_eq!(packet.meta.size, PACKET_DATA_SIZE); assert_eq!(packet.meta.size, PACKET_DATA_SIZE);
assert_eq!(packet.meta.addr(), saddr2); assert_eq!(packet.meta.socket_addr(), saddr2);
} }
} }
} }

4
streamer/src/streamer.rs
View File

@ -295,7 +295,7 @@ fn recv_send(
packet_batch.packets.iter().for_each(|p| stats.record(p)); packet_batch.packets.iter().for_each(|p| stats.record(p));
} }
let packets = packet_batch.packets.iter().filter_map(|pkt| { let packets = packet_batch.packets.iter().filter_map(|pkt| {
let addr = pkt.meta.addr(); let addr = pkt.meta.socket_addr();
socket_addr_space socket_addr_space
.check(&addr) .check(&addr)
.then(|| (&pkt.data[..pkt.meta.size], addr)) .then(|| (&pkt.data[..pkt.meta.size], addr))
@ -480,7 +480,7 @@ mod test {
{ {
p.data[0] = i as u8; p.data[0] = i as u8;
p.meta.size = PACKET_DATA_SIZE; p.meta.size = PACKET_DATA_SIZE;
p.meta.set_addr(&addr); p.meta.set_socket_addr(&addr);
} }
packet_batch.packets.push(p); packet_batch.packets.push(p);
} }