mirror
/
rqbit
mirror of https://github.com/ikatson/rqbit
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[feat] librqbit: support ephemeral ports when listening. 

Fixes https://github.com/ikatson/rqbit/issues/454.

The ephemeral port is found using OS (if TCP listen is enabled).
It's possible, but very unlikely (given the sheer number of listen ports available), that uTP won't be able to bind to the same port.
In this case uTP is disabled and error will be printed.
This commit is contained in:
Igor Katson 2025-07-10 09:06:19 +01:00
parent 314564955d
commit 67cf71850b
No known key found for this signature in database
GPG Key ID: B4EC22B66D61A3F5
2 changed files with 48 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
crates/librqbit/src/listen.rs
View File

@ -1,5 +1,5 @@
use std::{
net::{Ipv4Addr, SocketAddr},
net::{Ipv6Addr, SocketAddr},
sync::Arc,
};
@ -58,7 +58,7 @@ impl Default for ListenerOptions {
Self {
// TODO: once uTP is stable upgrade default to both
mode: ListenerMode::TcpOnly,
listen_addr: (Ipv4Addr::LOCALHOST, 0).into(),
listen_addr: (Ipv6Addr::UNSPECIFIED, 0).into(),
enable_upnp_port_forwarding: false,
utp_opts: None,
}
@ -72,63 +72,72 @@ impl ListenerOptions {
cancellation_token: CancellationToken,
bind_device: Option<&BindDevice>,
) -> anyhow::Result<ListenResult> {
if self.listen_addr.port() == 0 {
anyhow::bail!("you must set the listen port explicitly")
}
let mut utp_opts = self.utp_opts.take().unwrap_or_default();
utp_opts.cancellation_token = cancellation_token.clone();
utp_opts.parent_span = parent_span;
utp_opts.dont_wait_for_lastack = true;
let tcp = async {
if !self.mode.tcp_enabled() {
return Ok::<_, anyhow::Error>(None);
}
let mut listen_addr = self.listen_addr;
let tcp_socket = if self.mode.tcp_enabled() {
let listener = TcpListener::bind_tcp(
self.listen_addr,
BindOpts {
request_dualstack: true,
reuseport: true,
reuseport: false,
device: bind_device,
},
)
.context("error starting TCP listener")?;
listen_addr = listener.bind_addr();
info!(
"Listening on TCP {:?} for incoming peer connections",
self.listen_addr
listen_addr
);
Ok(Some(listener))
Some(listener)
} else {
None
};
let utp = async {
if !self.mode.utp_enabled() {
return Ok::<_, anyhow::Error>(None);
}
let socket = UtpSocketUdp::new_udp_with_opts(
self.listen_addr,
let utp_socket = if self.mode.utp_enabled() {
let bind_result = UtpSocketUdp::new_udp_with_opts(
listen_addr,
utp_opts,
UtpSocketUdpOpts { bind_device },
)
.await
.context("error starting uTP listener")?;
info!(
"Listening on UDP {:?} for incoming uTP peer connections",
self.listen_addr
);
Ok(Some(socket))
.await;
match bind_result {
Ok(sock) => {
listen_addr = sock.bind_addr();
info!(
"Listening on UDP {:?} for incoming uTP peer connections",
listen_addr
);
Some(sock)
}
Err(e) if tcp_socket.is_some() => {
// If we listen over TCP, it's not a fatal error if we can't listen over uTP.
tracing::error!("Error listening on UDP {listen_addr:?}: {e:#}");
None
}
Err(e) => {
return Err(e.into());
}
}
} else {
None
};
let announce_port = if self.listen_addr.ip().is_loopback() {
let announce_port = if listen_addr.ip().is_loopback() {
None
} else {
Some(self.listen_addr.port())
Some(listen_addr.port())
};
let (tcp_socket, utp_socket) = tokio::try_join!(tcp, utp)?;
Ok(ListenResult {
tcp_socket,
utp_socket,
announce_port,
addr: self.listen_addr,
addr: listen_addr,
enable_upnp_port_forwarding: self.enable_upnp_port_forwarding,
})
}

35
crates/rqbit/src/main.rs
View File

@ -528,7 +528,7 @@ async fn async_main(mut opts: Opts, cancel: CancellationToken) -> anyhow::Result
};
let listen = listen_mode.map(|mode| ListenerOptions {
mode,
listen_addr: (opts.listen_ip, opts.listen_port.unwrap_or(4240)).into(),
listen_addr: (opts.listen_ip, opts.listen_port.unwrap_or(0)).into(),
enable_upnp_port_forwarding: !opts.disable_upnp_port_forward,
..Default::default()
});
@ -618,6 +618,13 @@ async fn async_main(mut opts: Opts, cancel: CancellationToken) -> anyhow::Result
match &opts.subcommand {
SubCommand::Server(server_opts) => match &server_opts.subcommand {
ServerSubcommand::Start(start_opts) => {
// If the listen port wasn't set, default to 4240
if let Some(l) = sopts.listen.as_mut() {
if l.listen_addr.port() == 0 {
l.listen_addr.set_port(4240);
}
}
if !start_opts.disable_persistence {
if let Some(p) = start_opts.persistence_location.as_ref() {
if p.starts_with("postgres://") {
@ -685,9 +692,8 @@ async fn async_main(mut opts: Opts, cancel: CancellationToken) -> anyhow::Result
}
if let Some(listen) = sopts.listen.as_mut() {
// We are creating ephemeral ports, no point in port forwarding.
// We are creating an ephemeral download, no point in port forwarding.
listen.enable_upnp_port_forwarding = false;
maybe_set_ephemeral_port(&opts.listen_port, listen)?;
}
let torrent_opts = || AddTorrentOptions {
@ -822,9 +828,7 @@ async fn async_main(mut opts: Opts, cancel: CancellationToken) -> anyhow::Result
sopts.disable_dht_persistence = true;
sopts.persistence = None;
if let Some(listen) = sopts.listen.as_mut() {
maybe_set_ephemeral_port(&opts.listen_port, listen)?;
} else {
if sopts.listen.is_none() {
anyhow::bail!("you disabled all listeners, can't share");
}
@ -875,25 +879,6 @@ async fn async_main(mut opts: Opts, cancel: CancellationToken) -> anyhow::Result
}
}
fn maybe_set_ephemeral_port(
forced_listen_port: &Option<u16>,
listen: &mut ListenerOptions,
) -> anyhow::Result<()> {
// If the user hasn't specified a specific port, find a free random port.
// It needs to be the same for TCP and UDP, as we announce that port
// to trackers and DHT.
if forced_listen_port.is_none() {
let mut addr = listen.listen_addr;
addr.set_port(0);
let ephemeral_port = std::net::TcpListener::bind(addr)
.and_then(|l| l.local_addr())
.context("failed finding an ephemeral TCP/UDP listen port to use")?
.port();
listen.listen_addr.set_port(ephemeral_port);
}
Ok(())
}
async fn start_http_api(
cancel: CancellationToken,
session: Arc<Session>,