Skip to content

hack: Prioritise UDP disco messages #3237

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

hack: Prioritise UDP disco messages #3237

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 34 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions iroh/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ workspace = true
[dependencies]
aead = { version = "0.5.2", features = ["bytes"] }
anyhow = { version = "1" }
async-channel = "2.3.1"
atomic-waker = "1.1.2"
concurrent-queue = "2.5"
backon = { version = "1.4" }
Expand Down
83 changes: 80 additions & 3 deletions iroh/src/magicsock.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,7 +228,9 @@ pub(crate) struct MagicSock {
disco_secrets: DiscoSecrets,

/// UDP disco (ping) queue
udp_disco_sender: mpsc::Sender<(SocketAddr, PublicKey, disco::Message)>,
udp_disco_sender: async_channel::Sender<(SocketAddr, PublicKey, disco::Message)>,
udp_disco_receiver: async_channel::Receiver<(SocketAddr, PublicKey, disco::Message)>,
udp_disco_blocked: std::sync::Mutex<ConcurrentQueue<(SocketAddr, PublicKey, disco::Message)>>,

/// Optional discovery service
discovery: Option<Box<dyn Discovery>>,
Expand Down Expand Up @@ -474,6 +476,79 @@ impl MagicSock {
"connection closed",
));
}

// First empty out any disco messages, these have priority. Best-effort though, do
// not propagate any errors.
let mut pop_udp_disco_receiver = true;
while let Ok((dst, dst_key, msg)) = self.udp_disco_blocked.lock().unwrap().pop() {
let span = info_span!("udp-disco", %dst, node = %dst_key.fmt_short(), ?msg);
let _guard = span.enter();
let packet = self.encode_disco_message(dst_key, &msg);
inc!(MagicsockMetrics, send_disco_udp);
let transmit = quinn_udp::Transmit {
destination: dst,
ecn: None,
contents: &packet,
segment_size: None,
src_ip: None,
};
match self.try_send_udp(dst, &transmit) {
Ok(_) => {
inc!(MagicsockMetrics, send_disco_udp);
trace!("sent message");
}
Err(err) if err.kind() != io::ErrorKind::WouldBlock => {
warn!(?err, "failed to send");
pop_udp_disco_receiver = false;
break;
}
_ => {
trace!("would block");
self.udp_disco_blocked
.lock()
.unwrap()
.push((dst, dst_key, msg))
.ok();
pop_udp_disco_receiver = false;
break;
}
}
}
if pop_udp_disco_receiver {
while let Ok((dst, dst_key, msg)) = self.udp_disco_receiver.try_recv() {
let span = info_span!("udp-disco", %dst, node = %dst_key.fmt_short(), ?msg);
let _guard = span.enter();
let packet = self.encode_disco_message(dst_key, &msg);
inc!(MagicsockMetrics, send_disco_udp);
let transmit = quinn_udp::Transmit {
destination: dst,
ecn: None,
contents: &packet,
segment_size: None,
src_ip: None,
};
match self.try_send_udp(dst, &transmit) {
Ok(_) => {
inc!(MagicsockMetrics, send_disco_udp);
trace!("sent message");
}
Err(err) if err.kind() != io::ErrorKind::WouldBlock => {
warn!(?err, "failed to send");
break;
}
_ => {
trace!("would block");
self.udp_disco_blocked
.lock()
.unwrap()
.push((dst, dst_key, msg))
.ok();
break;
}
}
}
}

match MappedAddr::from(transmit.destination) {
MappedAddr::None(dest) => {
error!(%dest, "Cannot convert to a mapped address, voiding transmit.");
Expand Down Expand Up @@ -1704,7 +1779,7 @@ impl Handle {
let (relay_actor_sender, relay_actor_receiver) = mpsc::channel(256);
let (relay_datagram_send_tx, relay_datagram_send_rx) = relay_datagram_send_channel();
let relay_datagram_recv_queue = Arc::new(RelayDatagramRecvQueue::new());
let (udp_disco_sender, mut udp_disco_receiver) = mpsc::channel(256);
let (udp_disco_sender, udp_disco_receiver) = async_channel::bounded(256);

// load the node data
let node_map = node_map.unwrap_or_default();
Expand Down Expand Up @@ -1736,6 +1811,8 @@ impl Handle {
node_map,
ip_mapped_addrs,
udp_disco_sender,
udp_disco_receiver: udp_disco_receiver.clone(),
udp_disco_blocked: std::sync::Mutex::new(ConcurrentQueue::unbounded()),
discovery,
discovery_user_data: RwLock::new(discovery_user_data),
direct_addrs: Default::default(),
Expand Down Expand Up @@ -1783,7 +1860,7 @@ impl Handle {
let _ = actor_tasks.spawn({
let msock = msock.clone();
async move {
while let Some((dst, dst_key, msg)) = udp_disco_receiver.recv().await {
while let Ok((dst, dst_key, msg)) = udp_disco_receiver.recv().await {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So this task was previously the only place which would send these queued disco messages. Internally in the actual socket it is taking a lock which is contended with the same socket with the same internal lock that is called from try_send whcih is modified above to also read from this same udp_disco_receiver now.

if let Err(err) = msock.send_disco_message_udp(dst, dst_key, &msg).await {
warn!(%dst, node = %dst_key.fmt_short(), ?err, "failed to send disco message (UDP)");
}
Expand Down
Loading