udp: handle udp requests concurrently

Author: da2ce7

.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,
+}