gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

uses sendmmsg in streamer (backport #23062) (#23080)

Browse Source 
* uses sendmmsg in streamer (#23062)

packet::send_to sends packets one by one:
https://github.com/solana-labs/solana/blob/9213fcb11/streamer/src/packet.rs#L63-L75

sendmmsg uses a single system call for multiple messages:
https://github.com/solana-labs/solana/blob/9213fcb11/streamer/src/sendmmsg.rs#L94

(cherry picked from commit c078ca3fb3)

# Conflicts:
#	streamer/src/streamer.rs

* removes mergify merge conflicts

Co-authored-by: behzad nouri <behzadnouri@gmail.com>
This commit is contained in:
mergify[bot]
2022-02-11 15:47:45 +00:00
committed by GitHub
parent 5a5244ecf8
commit 781609b27a
1 changed files with 12 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
streamer/src/streamer.rs
View File

@ -3,8 +3,9 @@
use { use {
crate::{ crate::{
packet::{self, send_to, PacketBatch, PacketBatchRecycler, PACKETS_PER_BATCH}, packet::{self, PacketBatch, PacketBatchRecycler, PACKETS_PER_BATCH},
recvmmsg::NUM_RCVMMSGS, recvmmsg::NUM_RCVMMSGS,
sendmmsg::{batch_send, SendPktsError},
socket::SocketAddrSpace, socket::SocketAddrSpace,
}, },
solana_sdk::timing::timestamp, solana_sdk::timing::timestamp,
@ -34,6 +35,9 @@ pub enum StreamerError {
#[error("send packets error")] #[error("send packets error")]
Send(#[from] SendError<PacketBatch>), Send(#[from] SendError<PacketBatch>),
#[error(transparent)]
SendPktsError(#[from] SendPktsError),
} }
pub type Result<T> = std::result::Result<T, StreamerError>; pub type Result<T> = std::result::Result<T, StreamerError>;
@ -126,7 +130,13 @@ fn recv_send(
) -> Result<()> { ) -> Result<()> {
let timer = Duration::new(1, 0); let timer = Duration::new(1, 0);
let packet_batch = r.recv_timeout(timer)?; let packet_batch = r.recv_timeout(timer)?;
send_to(&packet_batch, sock, socket_addr_space)?; let packets = packet_batch.packets.iter().filter_map(|pkt| {
let addr = pkt.meta.addr();
socket_addr_space
.check(&addr)
.then(|| (&pkt.data[..pkt.meta.size], addr))
});
batch_send(sock, &packets.collect::<Vec<_>>())?;
Ok(()) Ok(())
} }