YGGverse/aquatic
1
0
Fork 0
mirror of https://github.com/YGGverse/aquatic.git synced 2026-03-31 17:55:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

udp: distribute announce swarm responses among socket workers

Browse source 
They don't have to be sent from the same worker that received the
request, so we can decrease performance loss from underutilized
threads this way.
This commit is contained in:
Joakim Frostegård 2024-01-01 13:32:28 +01:00
parent 7d17213a20
commit b527af7195
2 changed files with 91 additions and 57 deletions
Download patch file Download diff file Expand all files Collapse all files

23
crates/udp/src/common.rs
View file

@ -171,13 +171,17 @@ impl ConnectedRequestSender {
pub struct ConnectedResponseSender {
senders: Vec<thingbuf::mpsc::blocking::Sender<ConnectedResponseWithAddr, Recycler>>,
to_any_last_index_picked: usize,
}
impl ConnectedResponseSender {
pub fn new(
senders: Vec<thingbuf::mpsc::blocking::Sender<ConnectedResponseWithAddr, Recycler>>,
) -> Self {
Self { senders }
Self {
senders,
to_any_last_index_picked: 0,
}
}
pub fn try_send_ref_to(
@ -193,6 +197,23 @@ impl ConnectedResponseSender {
) -> Result<SendRef<ConnectedResponseWithAddr>, thingbuf::mpsc::errors::Closed> {
self.senders[index.0].send_ref()
}
pub fn send_ref_to_any(
&mut self,
) -> Result<SendRef<ConnectedResponseWithAddr>, thingbuf::mpsc::errors::Closed> {
let start = self.to_any_last_index_picked + 1;
for i in (start..start + self.senders.len()).map(|i| i % self.senders.len()) {
if let Ok(sender) = self.senders[i].try_send_ref() {
self.to_any_last_index_picked = i;
return Ok(sender);
}
}
self.to_any_last_index_picked = start % self.senders.len();
self.send_ref_to(SocketWorkerIndex(self.to_any_last_index_picked))
}
}
pub type ConnectedResponseReceiver =

125
crates/udp/src/workers/swarm/mod.rs
View file

@ -23,7 +23,7 @@ pub fn run_swarm_worker(
state: State,
server_start_instant: ServerStartInstant,
request_receiver: Receiver<(SocketWorkerIndex, ConnectedRequest, CanonicalSocketAddr)>,
response_sender: ConnectedResponseSender,
mut response_sender: ConnectedResponseSender,
statistics_sender: Sender<StatisticsMessage>,
worker_index: SwarmWorkerIndex,
) {
@ -43,65 +43,78 @@ pub fn run_swarm_worker(
loop {
if let Ok((sender_index, request, src)) = request_receiver.recv_timeout(timeout) {
// It is OK to block here as long as we don't do blocking sends
// in socket workers, which could cause a deadlock
match response_sender.send_ref_to(sender_index) {
Ok(mut send_ref) => {
// It is OK to block here as long as we don't also do blocking
// sends in socket workers (doing both could cause a deadlock)
match (request, src.get().ip()) {
(ConnectedRequest::Announce(request), IpAddr::V4(ip)) => {
// It doesn't matter which socket worker receives announce responses
let mut send_ref = response_sender
.send_ref_to_any()
.expect("swarm response channel is closed");
send_ref.addr = src;
send_ref.kind = ConnectedResponseKind::AnnounceIpv4;
match (request, src.get().ip()) {
(ConnectedRequest::Announce(request), IpAddr::V4(ip)) => {
send_ref.kind = ConnectedResponseKind::AnnounceIpv4;
torrents
.ipv4
.0
.entry(request.info_hash)
.or_default()
.announce(
&config,
&statistics_sender,
&mut rng,
&request,
ip.into(),
peer_valid_until,
&mut send_ref.announce_ipv4,
);
}
(ConnectedRequest::Announce(request), IpAddr::V6(ip)) => {
send_ref.kind = ConnectedResponseKind::AnnounceIpv6;
torrents
.ipv6
.0
.entry(request.info_hash)
.or_default()
.announce(
&config,
&statistics_sender,
&mut rng,
&request,
ip.into(),
peer_valid_until,
&mut send_ref.announce_ipv6,
);
}
(ConnectedRequest::Scrape(request), IpAddr::V4(_)) => {
send_ref.kind = ConnectedResponseKind::Scrape;
torrents.ipv4.scrape(request, &mut send_ref.scrape);
}
(ConnectedRequest::Scrape(request), IpAddr::V6(_)) => {
send_ref.kind = ConnectedResponseKind::Scrape;
torrents.ipv6.scrape(request, &mut send_ref.scrape);
}
};
torrents
.ipv4
.0
.entry(request.info_hash)
.or_default()
.announce(
&config,
&statistics_sender,
&mut rng,
&request,
ip.into(),
peer_valid_until,
&mut send_ref.announce_ipv4,
);
}
Err(_) => {
panic!("swarm response channel closed");
(ConnectedRequest::Announce(request), IpAddr::V6(ip)) => {
// It doesn't matter which socket worker receives announce responses
let mut send_ref = response_sender
.send_ref_to_any()
.expect("swarm response channel is closed");
send_ref.addr = src;
send_ref.kind = ConnectedResponseKind::AnnounceIpv6;
torrents
.ipv6
.0
.entry(request.info_hash)
.or_default()
.announce(
&config,
&statistics_sender,
&mut rng,
&request,
ip.into(),
peer_valid_until,
&mut send_ref.announce_ipv6,
);
}
}
(ConnectedRequest::Scrape(request), IpAddr::V4(_)) => {
let mut send_ref = response_sender
.send_ref_to(sender_index)
.expect("swarm response channel is closed");
send_ref.addr = src;
send_ref.kind = ConnectedResponseKind::Scrape;
torrents.ipv4.scrape(request, &mut send_ref.scrape);
}
(ConnectedRequest::Scrape(request), IpAddr::V6(_)) => {
let mut send_ref = response_sender
.send_ref_to(sender_index)
.expect("swarm response channel is closed");
send_ref.addr = src;
send_ref.kind = ConnectedResponseKind::Scrape;
torrents.ipv6.scrape(request, &mut send_ref.scrape);
}
};
}
// Run periodic tasks