Skip to content
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

fix(quic): use Provider::send_to for UDP datagram #28

Merged
merged 2 commits into from
Jul 4, 2023
Merged

fix(quic): use Provider::send_to for UDP datagram #28

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
459dedd
fix(quic): use Provider::send_to for UDP datagram
mxinden Jun 28, 2023
f31da2e
Simplify and hide dependency types
mxinden Jul 2, 2023
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
12 changes: 7 additions & 5 deletions transports/quic/src/hole_punching.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ use futures::future::Either;
use rand::{distributions, Rng};

use std::{
io,
net::{SocketAddr, UdpSocket},
time::Duration,
};
Expand All @@ -24,6 +23,11 @@ pub(crate) async fn hole_puncher<P: Provider>(
}

async fn punch_holes<P: Provider>(socket: UdpSocket, remote_addr: SocketAddr) -> Error {
let socket = match P::from_std_udp_socket(socket) {
Ok(s) => s,
Err(e) => return Error::Io(e),
};

loop {
let sleep_duration = Duration::from_millis(rand::thread_rng().gen_range(10..=200));
P::sleep(sleep_duration).await;
Expand All @@ -33,10 +37,8 @@ async fn punch_holes<P: Provider>(socket: UdpSocket, remote_addr: SocketAddr) ->
.take(64)
.collect();

if let Err(e) = socket.send_to(&contents, remote_addr) {
if !matches!(e.kind(), io::ErrorKind::WouldBlock) {
return Error::Io(e);
}
if let Err(e) = P::send_to(&socket, &contents, remote_addr).await {
return Error::Io(e);
}
Comment on lines +35 to 37
Copy link

Choose a reason for hiding this comment

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

Isn't this just what ? would do?

Copy link
Author

Choose a reason for hiding this comment

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

I wondered the same. Note the function signature:

async fn punch_holes<P: Provider>(socket: UdpSocket, remote_addr: SocketAddr) -> Error {

The function never returns, unless there is an Error. In words, the function returns Error and not Result<_, Error>. Thus one can not use ?.

Copy link

Choose a reason for hiding this comment

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

Perhaps we should change it to Result<Infallible, Error>?

Copy link
Author

Choose a reason for hiding this comment

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

We discussed this in libp2p#3964 as well, can't find the concrete conversation right now. I suggest we keep as is unless you feel strongly about it @thomaseizinger.

}
}
12 changes: 12 additions & 0 deletions transports/quic/src/provider.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ use futures::{future::BoxFuture, Future};
use if_watch::IfEvent;
use std::{
io,
net::SocketAddr,
task::{Context, Poll},
time::Duration,
};
Expand All @@ -42,6 +43,7 @@ pub enum Runtime {
/// Provider for a corresponding quinn runtime and spawning tasks.
pub trait Provider: Unpin + Send + Sized + 'static {
type IfWatcher: Unpin + Send;
type UdpSocket: Unpin + Send + Sync;

/// Run the corresponding runtime
fn runtime() -> Runtime;
Expand All @@ -62,4 +64,14 @@ pub trait Provider: Unpin + Send + Sized + 'static {

/// Sleep for specified amount of time.
fn sleep(duration: Duration) -> BoxFuture<'static, ()>;

/// Creates new UdpSocket from a previously bound std::net::UdpSocket.
fn from_std_udp_socket(socket: std::net::UdpSocket) -> io::Result<Self::UdpSocket>;

/// Sends data on the socket to the given address. On success, returns the number of bytes written.
fn send_to<'a>(
udp_socket: &'a Self::UdpSocket,
buf: &'a [u8],
target: SocketAddr,
) -> BoxFuture<'a, io::Result<usize>>;
}
13 changes: 13 additions & 0 deletions transports/quic/src/provider/async_std.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ pub struct Provider;

impl super::Provider for Provider {
type IfWatcher = if_watch::smol::IfWatcher;
type UdpSocket = async_std::net::UdpSocket;
kpp marked this conversation as resolved.
Show resolved Hide resolved

fn runtime() -> super::Runtime {
super::Runtime::AsyncStd
Expand All @@ -59,4 +60,16 @@ impl super::Provider for Provider {
fn sleep(duration: Duration) -> BoxFuture<'static, ()> {
async_std::task::sleep(duration).boxed()
}

fn from_std_udp_socket(socket: std::net::UdpSocket) -> io::Result<Self::UdpSocket> {
Ok(socket.into())
}

fn send_to<'a>(
udp_socket: &'a Self::UdpSocket,
buf: &'a [u8],
target: std::net::SocketAddr,
) -> BoxFuture<'a, io::Result<usize>> {
udp_socket.send_to(buf, target).boxed()
}
Copy link

@thomaseizinger thomaseizinger Jun 28, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

Instead of two functions, we could also make one function that accepts an std::net::UdpSocket, converts it internally to the runtime-specific one, sends the packet and returns the socket.

That would avoid the associated type and reduce this to one function that needs to be implemented.

}
14 changes: 14 additions & 0 deletions transports/quic/src/provider/tokio.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
use futures::{future::BoxFuture, Future, FutureExt};
use std::{
io,
net::SocketAddr,
task::{Context, Poll},
time::Duration,
};
Expand All @@ -35,6 +36,7 @@ pub struct Provider;

impl super::Provider for Provider {
type IfWatcher = if_watch::tokio::IfWatcher;
type UdpSocket = tokio::net::UdpSocket;

fn runtime() -> super::Runtime {
super::Runtime::Tokio
Expand All @@ -58,4 +60,16 @@ impl super::Provider for Provider {
fn sleep(duration: Duration) -> BoxFuture<'static, ()> {
tokio::time::sleep(duration).boxed()
}

fn from_std_udp_socket(socket: std::net::UdpSocket) -> io::Result<Self::UdpSocket> {
tokio::net::UdpSocket::from_std(socket)
}

fn send_to<'a>(
udp_socket: &'a Self::UdpSocket,
buf: &'a [u8],
target: SocketAddr,
) -> BoxFuture<'a, io::Result<usize>> {
udp_socket.send_to(buf, target).boxed()
}
}