Skip to content

Commit ea33f1c

Browse files
josecelanojosecelano
committed
Merge torrust#747: Profiling for the UDP tracker
9015668 refactor: [torrust#746] rename functions and extract named closures (Jose Celano) cc1cbc1 test: [torrust#746] add a new binary for profiling (Jose Celano) 8395c42 test: [torrust#746] disable tracker stats for profiling (Jose Celano) Pull request description: Profiling for the UDP tracker. - [x] Add a new binary for profiling. - [x] Minor refactorings. Rename functions and extract named closures. ACKs for top commit: josecelano: ACK 9015668 Tree-SHA512: 3940ed11f04950da34941e45da89f43d74ce8d5bc97e2b6a11e8473b617e73cd742376c197456352ae9dfff201b72705d26866b8de4535d93c5ac8e5be73977b
2 parents b777d3a + 9015668 commit ea33f1c

File tree

7 files changed

+256
-34
lines changed

7 files changed

+256
-34
lines changed

‎.gitignore

+2-1
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,11 @@
33
/.coverage/
44
/.idea/
55
/.vscode/launch.json
6-
/tracker.toml
76
/data.db
87
/database.db
98
/database.json.bz2
109
/storage/
1110
/target
1211
/tracker.*
12+
/tracker.toml
13+
callgrind.out

‎cSpell.json

+5
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
"bufs",
2222
"Buildx",
2323
"byteorder",
24+
"callgrind",
2425
"canonicalize",
2526
"canonicalized",
2627
"certbot",
@@ -35,6 +36,7 @@
3536
"Cyberneering",
3637
"datagram",
3738
"datetime",
39+
"debuginfo",
3840
"Deque",
3941
"Dijke",
4042
"distroless",
@@ -60,6 +62,7 @@
6062
"infoschema",
6163
"Intermodal",
6264
"intervali",
65+
"kcachegrind",
6366
"keyout",
6467
"lcov",
6568
"leecher",
@@ -134,7 +137,9 @@
134137
"untuple",
135138
"uroot",
136139
"Vagaa",
140+
"valgrind",
137141
"Vuze",
142+
"Weidendorfer",
138143
"Werror",
139144
"whitespaces",
140145
"XBTT",

‎share/default/config/tracker.udp.benchmarking.toml

+2-2
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,8 @@ min_announce_interval = 120
99
mode = "public"
1010
on_reverse_proxy = false
1111
persistent_torrent_completed_stat = false
12-
remove_peerless_torrents = true
13-
tracker_usage_statistics = true
12+
remove_peerless_torrents = false
13+
tracker_usage_statistics = false
1414

1515
[[udp_trackers]]
1616
bind_address = "0.0.0.0:6969"

‎src/bin/profiling.rs

+8
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
//! This binary is used for profiling with [valgrind](https://valgrind.org/)
2+
//! and [kcachegrind](https://kcachegrind.github.io/).
3+
use torrust_tracker::console::profiling::run;
4+
5+
#[tokio::main]
6+
async fn main() {
7+
run().await;
8+
}

‎src/console/mod.rs

+1
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
11
//! Console apps.
22
pub mod ci;
33
pub mod clients;
4+
pub mod profiling;

‎src/console/profiling.rs

+202
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,202 @@
1+
//! This binary is used for profiling with [valgrind](https://valgrind.org/)
2+
//! and [kcachegrind](https://kcachegrind.github.io/).
3+
//!
4+
//! # Requirements
5+
//!
6+
//! [valgrind](https://valgrind.org/) and [kcachegrind](https://kcachegrind.github.io/).
7+
//!
8+
//! On Ubuntu you can install them with:
9+
//!
10+
//! ```text
11+
//! sudo apt install valgrind kcachegrind
12+
//! ```
13+
//!
14+
//! > NOTICE: valgrind executes the program you wan to profile and waits until
15+
//! it ends. Since the tracker is a service and does not end the profiling
16+
//! binary accepts an arguments with the duration you want to run the tracker,
17+
//! so that it terminates automatically after that period of time.
18+
//!
19+
//! # Run profiling
20+
//!
21+
//! To run the profiling you have to:
22+
//!
23+
//! 1. Build and run the tracker for profiling.
24+
//! 2. Run the aquatic UDP load test tool to start collecting data in the tracker.
25+
//!
26+
//! Build and run the tracker for profiling:
27+
//!
28+
//! ```text
29+
//! RUSTFLAGS='-g' cargo build --release --bin profiling \
30+
//! && export TORRUST_TRACKER_PATH_CONFIG="./share/default/config/tracker.udp.benchmarking.toml" \
31+
//! && valgrind \
32+
//! --tool=callgrind \
33+
//! --callgrind-out-file=callgrind.out \
34+
//! --collect-jumps=yes \
35+
//! --simulate-cache=yes \
36+
//! ./target/release/profiling 60
37+
//! ```
38+
//!
39+
//! The output should be something like:
40+
//!
41+
//! ```text
42+
//! RUSTFLAGS='-g' cargo build --release --bin profiling \
43+
//! && export TORRUST_TRACKER_PATH_CONFIG="./share/default/config/tracker.udp.benchmarking.toml" \
44+
//! && valgrind \
45+
//! --tool=callgrind \
46+
//! --callgrind-out-file=callgrind.out \
47+
//! --collect-jumps=yes \
48+
//! --simulate-cache=yes \
49+
//! ./target/release/profiling 60
50+
//!
51+
//! Compiling torrust-tracker v3.0.0-alpha.12-develop (/home/developer/Documents/git/committer/me/github/torrust/torrust-tracker)
52+
//! Finished `release` profile [optimized + debuginfo] target(s) in 1m 15s
53+
//! ==122801== Callgrind, a call-graph generating cache profiler
54+
//! ==122801== Copyright (C) 2002-2017, and GNU GPL'd, by Josef Weidendorfer et al.
55+
//! ==122801== Using Valgrind-3.19.0 and LibVEX; rerun with -h for copyright info
56+
//! ==122801== Command: ./target/release/profiling 60
57+
//! ==122801==
58+
//! --122801-- warning: L3 cache found, using its data for the LL simulation.
59+
//! ==122801== For interactive control, run 'callgrind_control -h'.
60+
//! Loading configuration file: `./share/default/config/tracker.udp.benchmarking.toml` ...
61+
//! Torrust successfully shutdown.
62+
//! ==122801==
63+
//! ==122801== Events : Ir Dr Dw I1mr D1mr D1mw ILmr DLmr DLmw
64+
//! ==122801== Collected : 1160654816 278135882 247755311 24453652 12650490 16315690 10932 2481624 4832145
65+
//! ==122801==
66+
//! ==122801== I refs: 1,160,654,816
67+
//! ==122801== I1 misses: 24,453,652
68+
//! ==122801== LLi misses: 10,932
69+
//! ==122801== I1 miss rate: 2.11%
70+
//! ==122801== LLi miss rate: 0.00%
71+
//! ==122801==
72+
//! ==122801== D refs: 525,891,193 (278,135,882 rd + 247,755,311 wr)
73+
//! ==122801== D1 misses: 28,966,180 ( 12,650,490 rd + 16,315,690 wr)
74+
//! ==122801== LLd misses: 7,313,769 ( 2,481,624 rd + 4,832,145 wr)
75+
//! ==122801== D1 miss rate: 5.5% ( 4.5% + 6.6% )
76+
//! ==122801== LLd miss rate: 1.4% ( 0.9% + 2.0% )
77+
//! ==122801==
78+
//! ==122801== LL refs: 53,419,832 ( 37,104,142 rd + 16,315,690 wr)
79+
//! ==122801== LL misses: 7,324,701 ( 2,492,556 rd + 4,832,145 wr)
80+
//! ==122801== LL miss rate: 0.4% ( 0.2% + 2.0% )
81+
//! ```
82+
//!
83+
//! > NOTICE: We are using an specific tracker configuration for profiling that
84+
//! removes all features except the UDP tracker and sets the logging level to `error`.
85+
//!
86+
//! Build the aquatic UDP load test command:
87+
//!
88+
//! ```text
89+
//! cd /tmp
90+
//! git clone git@github.com:greatest-ape/aquatic.git
91+
//! cd aquatic
92+
//! cargo build --profile=release-debug -p aquatic_udp_load_test
93+
//! ./target/release-debug/aquatic_udp_load_test -p > "load-test-config.toml"
94+
//! ```
95+
//!
96+
//! Modify the "load-test-config.toml" file to change the UDP tracker port from
97+
//! `3000` to `6969`.
98+
//!
99+
//! Running the aquatic UDP load test command:
100+
//!
101+
//! ```text
102+
//! ./target/release-debug/aquatic_udp_load_test -c "load-test-config.toml"
103+
//! ```
104+
//!
105+
//! The output should be something like this:
106+
//!
107+
//! ```text
108+
//! Starting client with config: Config {
109+
//! server_address: 127.0.0.1:6969,
110+
//! log_level: Error,
111+
//! workers: 1,
112+
//! duration: 0,
113+
//! summarize_last: 0,
114+
//! extra_statistics: true,
115+
//! network: NetworkConfig {
116+
//! multiple_client_ipv4s: true,
117+
//! sockets_per_worker: 4,
118+
//! recv_buffer: 8000000,
119+
//! },
120+
//! requests: RequestConfig {
121+
//! number_of_torrents: 1000000,
122+
//! number_of_peers: 2000000,
123+
//! scrape_max_torrents: 10,
124+
//! announce_peers_wanted: 30,
125+
//! weight_connect: 50,
126+
//! weight_announce: 50,
127+
//! weight_scrape: 1,
128+
//! peer_seeder_probability: 0.75,
129+
//! },
130+
//! }
131+
//!
132+
//! Requests out: 45097.51/second
133+
//! Responses in: 4212.70/second
134+
//! - Connect responses: 2098.15
135+
//! - Announce responses: 2074.95
136+
//! - Scrape responses: 39.59
137+
//! - Error responses: 0.00
138+
//! Peers per announce response: 0.00
139+
//! Announce responses per info hash:
140+
//! - p10: 1
141+
//! - p25: 1
142+
//! - p50: 1
143+
//! - p75: 2
144+
//! - p90: 3
145+
//! - p95: 4
146+
//! - p99: 6
147+
//! - p99.9: 8
148+
//! - p100: 10
149+
//! ```
150+
//!
151+
//! After running the tracker for some seconds the tracker will automatically stop
152+
//! and `valgrind`will write the file `callgrind.out` with the data.
153+
//!
154+
//! You can now analyze the collected data with:
155+
//!
156+
//! ```text
157+
//! kcachegrind callgrind.out
158+
//! ```
159+
use std::env;
160+
use std::time::Duration;
161+
162+
use log::info;
163+
use tokio::time::sleep;
164+
165+
use crate::{app, bootstrap};
166+
167+
pub async fn run() {
168+
// Parse command line arguments
169+
let args: Vec<String> = env::args().collect();
170+
171+
// Ensure an argument for duration is provided
172+
if args.len() != 2 {
173+
eprintln!("Usage: {} <duration_in_seconds>", args[0]);
174+
return;
175+
}
176+
177+
// Parse duration argument
178+
let Ok(duration_secs) = args[1].parse::<u64>() else {
179+
eprintln!("Invalid duration provided");
180+
return;
181+
};
182+
183+
let (config, tracker) = bootstrap::app::setup();
184+
185+
let jobs = app::start(&config, tracker).await;
186+
187+
// Run the tracker for a fixed duration
188+
let run_duration = sleep(Duration::from_secs(duration_secs));
189+
190+
tokio::select! {
191+
() = run_duration => {
192+
info!("Torrust timed shutdown..");
193+
},
194+
_ = tokio::signal::ctrl_c() => {
195+
info!("Torrust shutting down via Ctrl+C..");
196+
// Await for all jobs to shutdown
197+
futures::future::join_all(jobs).await;
198+
}
199+
}
200+
201+
println!("Torrust successfully shutdown.");
202+
}

‎src/servers/udp/server.rs

+36-31
Original file line numberDiff line numberDiff line change
@@ -255,24 +255,7 @@ impl Udp {
255255

256256
let running = tokio::task::spawn(async move {
257257
debug!(target: "UDP TRACKER", "Started: Waiting for packets on socket address: udp://{address} ...");
258-
259-
let tracker = tracker.clone();
260-
let socket = socket.clone();
261-
262-
let reqs = &mut ActiveRequests::default();
263-
264-
// Main Waiting Loop, awaits on async [`receive_request`].
265-
loop {
266-
if let Some(h) = reqs.rb.push_overwrite(
267-
Self::do_request(Self::receive_request(socket.clone()).await, tracker.clone(), socket.clone()).abort_handle(),
268-
) {
269-
if !h.is_finished() {
270-
// the task is still running, lets yield and give it a chance to flush.
271-
tokio::task::yield_now().await;
272-
h.abort();
273-
}
274-
}
275-
}
258+
Self::run_udp_server(tracker, socket).await;
276259
});
277260

278261
tx_start
@@ -292,6 +275,27 @@ impl Udp {
292275
task::yield_now().await; // lets allow the other threads to complete.
293276
}
294277

278+
async fn run_udp_server(tracker: Arc<Tracker>, socket: Arc<UdpSocket>) {
279+
let tracker = tracker.clone();
280+
let socket = socket.clone();
281+
282+
let reqs = &mut ActiveRequests::default();
283+
284+
// Main Waiting Loop, awaits on async [`receive_request`].
285+
loop {
286+
if let Some(h) = reqs.rb.push_overwrite(
287+
Self::spawn_request_processor(Self::receive_request(socket.clone()).await, tracker.clone(), socket.clone())
288+
.abort_handle(),
289+
) {
290+
if !h.is_finished() {
291+
// the task is still running, lets yield and give it a chance to flush.
292+
tokio::task::yield_now().await;
293+
h.abort();
294+
}
295+
}
296+
}
297+
}
298+
295299
async fn receive_request(socket: Arc<UdpSocket>) -> Result<UdpRequest, Box<std::io::Error>> {
296300
// Wait for the socket to be readable
297301
socket.readable().await?;
@@ -309,26 +313,27 @@ impl Udp {
309313
}
310314
}
311315

312-
fn do_request(
316+
fn spawn_request_processor(
313317
result: Result<UdpRequest, Box<std::io::Error>>,
314318
tracker: Arc<Tracker>,
315319
socket: Arc<UdpSocket>,
316320
) -> JoinHandle<()> {
317-
// timeout not needed, as udp is non-blocking.
318-
tokio::task::spawn(async move {
319-
match result {
320-
Ok(udp_request) => {
321-
trace!("Received Request from: {}", udp_request.from);
322-
Self::make_response(tracker.clone(), socket.clone(), udp_request).await;
323-
}
324-
Err(error) => {
325-
debug!("error: {error}");
326-
}
321+
tokio::task::spawn(Self::process_request(result, tracker, socket))
322+
}
323+
324+
async fn process_request(result: Result<UdpRequest, Box<std::io::Error>>, tracker: Arc<Tracker>, socket: Arc<UdpSocket>) {
325+
match result {
326+
Ok(udp_request) => {
327+
trace!("Received Request from: {}", udp_request.from);
328+
Self::process_valid_request(tracker.clone(), socket.clone(), udp_request).await;
327329
}
328-
})
330+
Err(error) => {
331+
debug!("error: {error}");
332+
}
333+
}
329334
}
330335

331-
async fn make_response(tracker: Arc<Tracker>, socket: Arc<UdpSocket>, udp_request: UdpRequest) {
336+
async fn process_valid_request(tracker: Arc<Tracker>, socket: Arc<UdpSocket>, udp_request: UdpRequest) {
332337
trace!("Making Response to {udp_request:?}");
333338
let from = udp_request.from;
334339
let response = handlers::handle_packet(udp_request, &tracker.clone(), socket.clone()).await;

0 commit comments

Comments
 (0)