reactor
/
ptth
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

update: add throttled endless random garbage

main
_ 2021-02-20 17:28:39 +00:00
parent 09464c548a
commit 498d69eeb9
2 changed files with 48 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
crates/ptth_relay/src/lib.rs
View File

@ -303,22 +303,54 @@ async fn handle_server_list (
Ok (ok_reply (s)?) Ok (ok_reply (s)?)
} }
async fn handle_endless_source (gib: usize) -> Result <Response <Body>, http::Error> { async fn handle_endless_source (gib: usize, throttle: Option <usize>)
-> Result <Response <Body>, http::Error>
{
use futures::stream::StreamExt; use futures::stream::StreamExt;
use tokio::sync::mpsc;
let random_block = { let block_bytes = 64 * 1024;
use rand::RngCore; let num_blocks = (1024 * 1024 * 1024 / block_bytes) * gib;
let (tx, rx) = mpsc::channel (1);
tokio::spawn (async move {
let mut tx = tx;
let mut rng = rand::thread_rng (); let random_block = {
let mut block = vec! [0u8; 64 * 1024]; use rand::RngCore;
rng.fill_bytes (&mut block);
block let mut rng = rand::thread_rng ();
}; let mut block = vec! [0u8; 64 * 1024];
rng.fill_bytes (&mut block);
block
};
let mut interval = tokio::time::interval (Duration::from_millis (1000));
let mut blocks_sent = 0;
while blocks_sent < num_blocks {
if throttle.is_some () {
interval.tick ().await;
}
for _ in 0..throttle.unwrap_or (1) {
let item = Ok::<_, Infallible> (random_block.clone ());
if let Err (_) = tx.send (item).await {
debug! ("Endless source dropped");
return;
}
blocks_sent += 1;
}
}
debug! ("Endless source ended");
});
Response::builder () Response::builder ()
.status (StatusCode::OK) .status (StatusCode::OK)
.header ("content-type", "application/octet-stream") .header ("content-type", "application/octet-stream")
.body (Body::wrap_stream (futures::stream::repeat (Ok::<_, Infallible> (random_block)).take (gib * 1024 * 1024 / 64))) .body (Body::wrap_stream (rx))
} }
#[instrument (level = "trace", skip (req, state, handlebars))] #[instrument (level = "trace", skip (req, state, handlebars))]
@ -379,7 +411,10 @@ async fn handle_all (
Ok (ok_reply (s)?) Ok (ok_reply (s)?)
} }
else if rest == "endless_source" { else if rest == "endless_source" {
Ok (handle_endless_source (1).await?) Ok (handle_endless_source (1, None).await?)
}
else if rest == "endless_source_throttled" {
Ok (handle_endless_source (1, Some (1024 / 64)).await?)
} }
else { else {
Ok (error_reply (StatusCode::OK, "Can't route URL")?) Ok (error_reply (StatusCode::OK, "Can't route URL")?)

5
handlebars/relay/debug.hbs
View File

@ -41,12 +41,13 @@ AIABAACAAQAAgAEAAIABAACAAQAAgAEAAIABAACAAQAA" rel="icon" type="image/x-icon" />
<h1>Debugging tools (relay)</h1> <h1>Debugging tools (relay)</h1>
<p> <p>
Lorem ipsum dolor set amet Lorem ipsum dolor set amet
<p> <p>
<a href="endless_source">1 GiB random garbage data source</a> <a href="endless_source">1 GiB random garbage data source</a>
<p>
<a href="endless_source_throttled">1 GiB random garbage data source (Throttled to 1 MiB/s)</a>
</body> </body>
</html> </html>