Merge #644: udp: handle udp requests concurrently

72c8348 udp: handle udp requests concurrently (Cameron Garnham)

Pull request description:

  Extracted from #557

  Upgrades the udp implementation to handle requests concurrently, no timeout is needed as udp is non-blocking.

  Hard-coded limit of 50 concurrent requests. With trivial changes can make this limit configurable, however I don't think that it would really be much benefit for it, the main concern is memory usage, and since udp is no blocking each request should return very quickly, could set to 5000 and it should still be fine.

ACKs for top commit:
  josecelano:
    ACK 72c8348

Tree-SHA512: 83c4d893d5a88a59655942dcafceed25ca3b620a73fbd70d43ade4684394348c170dbfa427dfb236bcb7a8175e307547c894ec79d3c0b992441e8aa1490885b8

Author: josecelano

.github/workflows/coverage.yaml

@@ -55,6 +55,9 @@ jobs:
         name: Run Build Checks
         run: cargo check --tests --benches --examples --workspace --all-targets --all-features
 
+      # Run Test Locally:
+      # RUSTFLAGS="-Z profile -C codegen-units=1 -C inline-threshold=0 -C link-dead-code -C overflow-checks=off -C panic=abort -Z panic_abort_tests" RUSTDOCFLAGS="-Z profile -C codegen-units=1 -C inline-threshold=0 -C link-dead-code -C overflow-checks=off -C panic=abort -Z panic_abort_tests" CARGO_INCREMENTAL="0" RUST_BACKTRACE=1 cargo test --tests --benches --examples --workspace --all-targets --all-features
+
       - id: test
         name: Run Unit Tests
         run: cargo test --tests --benches --examples --workspace --all-targets --all-features

.vscode/settings.json

@@ -2,6 +2,12 @@
   "[rust]": {
     "editor.formatOnSave": true
   },
+  "[ignore]": { "rust-analyzer.cargo.extraEnv" : {
+    "RUSTFLAGS": "-Z profile -C codegen-units=1 -C inline-threshold=0 -C link-dead-code -C overflow-checks=off -C panic=abort -Z panic_abort_tests",
+    "RUSTDOCFLAGS": "-Z profile -C codegen-units=1 -C inline-threshold=0 -C link-dead-code -C overflow-checks=off -C panic=abort -Z panic_abort_tests",
+    "CARGO_INCREMENTAL": "0",
+    "RUST_BACKTRACE": "1"
+  }},
   "rust-analyzer.checkOnSave": true,
   "rust-analyzer.check.command": "clippy",
   "rust-analyzer.check.allTargets": true,

Cargo.lock

@@ -56,6 +56,7 @@ serde = { version = "1", features = ["derive"] }
 serde_bencode = "0"
 serde_bytes = "0"
 serde_json = "1"
+ringbuf = "0.4.0-rc.2"
 serde_with = "3"
 serde_repr = "0"
 tdyne-peer-id = "1"

Cargo.toml

@@ -94,8 +94,10 @@
         "reannounce",
         "Registar",
         "repr",
+        "reqs",
         "reqwest",
         "rerequests",
+        "ringbuf",
         "rngs",
         "routable",
         "rusqlite",

cSpell.json

@@ -11,6 +11,7 @@ use log::{debug, info};
 use torrust_tracker_located_error::DynError;
 
 use super::connection_cookie::{check, from_connection_id, into_connection_id, make};
+use super::UdpRequest;
 use crate::core::{statistics, ScrapeData, Tracker};
 use crate::servers::udp::error::Error;
 use crate::servers::udp::peer_builder;
@@ -27,10 +28,13 @@ use crate::shared::bit_torrent::info_hash::InfoHash;
 /// type.
 ///
 /// It will return an `Error` response if the request is invalid.
-pub async fn handle_packet(remote_addr: SocketAddr, payload: Vec<u8>, tracker: &Tracker) -> Response {
-    match Request::from_bytes(&payload[..payload.len()], MAX_SCRAPE_TORRENTS).map_err(|e| Error::InternalServer {
-        message: format!("{e:?}"),
-        location: Location::caller(),
+pub(crate) async fn handle_packet(udp_request: UdpRequest, tracker: &Arc<Tracker>) -> Response {
+    debug!("Handling Packets: {udp_request:?}");
+    match Request::from_bytes(&udp_request.payload[..udp_request.payload.len()], MAX_SCRAPE_TORRENTS).map_err(|e| {
+        Error::InternalServer {
+            message: format!("{e:?}"),
+            location: Location::caller(),
+        }
     }) {
         Ok(request) => {
             let transaction_id = match &request {
@@ -39,7 +43,7 @@ pub async fn handle_packet(remote_addr: SocketAddr, payload: Vec<u8>, tracker: &
                 Request::Scrape(scrape_request) => scrape_request.transaction_id,
             };
 
-            match handle_request(request, remote_addr, tracker).await {
+            match handle_request(request, udp_request.from, tracker).await {
                 Ok(response) => response,
                 Err(e) => handle_error(&e, transaction_id),
             }
@@ -60,6 +64,8 @@ pub async fn handle_packet(remote_addr: SocketAddr, payload: Vec<u8>, tracker: &
 ///
 /// If a error happens in the `handle_request` function, it will just return the  `ServerError`.
 pub async fn handle_request(request: Request, remote_addr: SocketAddr, tracker: &Tracker) -> Result<Response, Error> {
+    debug!("Handling Request: {request:?} to: {remote_addr:?}");
+
     match request {
         Request::Connect(connect_request) => handle_connect(remote_addr, &connect_request, tracker).await,
         Request::Announce(announce_request) => handle_announce(remote_addr, &announce_request, tracker).await,

src/servers/udp/handlers.rs

@@ -638,6 +638,9 @@
 //! documentation by [Arvid Norberg](https://github.com/arvidn) was very
 //! supportive in the development of this documentation. Some descriptions were
 //! taken from the [libtorrent](https://www.rasterbar.com/products/libtorrent/udp_tracker_protocol.html).
+
+use std::net::SocketAddr;
+
 pub mod connection_cookie;
 pub mod error;
 pub mod handlers;
@@ -652,3 +655,9 @@ pub type Port = u16;
 /// The transaction id. A random number generated byt the peer that is used to
 /// match requests and responses.
 pub type TransactionId = i64;
+
+#[derive(Clone, Debug)]
+pub(crate) struct UdpRequest {
+    payload: Vec<u8>,
+    from: SocketAddr,
+}