Cleaning up

This commit is contained in:
Igor Katson 2024-05-03 13:35:31 +01:00
parent 43008f2857
commit 0f1d07409d
3 changed files with 23 additions and 11 deletions

View file

@ -1,10 +1,14 @@
/* /*
A storage middleware that slows down the underlying storage. A storage middleware that slows down the underlying storage.
To be used for debugging. The input to it are 2 files with line-delimited numbers of microseconds
it takes to read and write respectively.
*/ */
use std::{ use std::{
fs::File, fs::File,
io::{BufRead, BufReader, Lines}, io::{BufRead, BufReader},
time::Duration, time::Duration,
}; };
@ -33,7 +37,7 @@ impl<U: StorageFactory + Clone> StorageFactory for SlowStorageFactory<U> {
underlying: self.underlying_factory.init_storage(info)?, underlying: self.underlying_factory.init_storage(info)?,
pwrite_all_bufread: Mutex::new(Box::new( pwrite_all_bufread: Mutex::new(Box::new(
BufReader::new( BufReader::new(
File::open("/Users/igor/Downloads/rqbit-log-slow-disk.log-pwrite_all").unwrap(), File::open(std::env::var("DBG_PWRITE_ALL_FILENAME").unwrap()).unwrap(),
) )
.lines() .lines()
.map(|l| l.unwrap().parse().unwrap()) .map(|l| l.unwrap().parse().unwrap())
@ -42,8 +46,7 @@ impl<U: StorageFactory + Clone> StorageFactory for SlowStorageFactory<U> {
)), )),
pread_exact_bufread: Mutex::new(Box::new( pread_exact_bufread: Mutex::new(Box::new(
BufReader::new( BufReader::new(
File::open("/Users/igor/Downloads/rqbit-log-slow-disk.log-pread_exact") File::open(std::env::var("DBG_PWRITE_ALL_FILENAME").unwrap()).unwrap(),
.unwrap(),
) )
.lines() .lines()
.map(|l| l.unwrap().parse().unwrap()) .map(|l| l.unwrap().parse().unwrap())

View file

@ -19,6 +19,7 @@ webui = ["librqbit/webui"]
timed_existence = ["librqbit/timed_existence"] timed_existence = ["librqbit/timed_existence"]
default-tls = ["librqbit/default-tls"] default-tls = ["librqbit/default-tls"]
rust-tls = ["librqbit/rust-tls"] rust-tls = ["librqbit/rust-tls"]
debug_slow_disk = ["librqbit/storage_middleware"]
[dependencies] [dependencies]
librqbit = { path = "../librqbit", default-features = false, version = "5.6.4" } librqbit = { path = "../librqbit", default-features = false, version = "5.6.4" }

View file

@ -9,7 +9,6 @@ use librqbit::{
http_api_client, librqbit_spawn, http_api_client, librqbit_spawn,
storage::{ storage::{
filesystem::{FilesystemStorageFactory, MmapFilesystemStorageFactory}, filesystem::{FilesystemStorageFactory, MmapFilesystemStorageFactory},
middleware::{slow::SlowStorageFactory, timing::TimingStorageFactory},
StorageFactory, StorageFactoryExt, StorageFactory, StorageFactoryExt,
}, },
tracing_subscriber_config_utils::{init_logging, InitLoggingOptions}, tracing_subscriber_config_utils::{init_logging, InitLoggingOptions},
@ -99,17 +98,19 @@ struct Opts {
#[command(subcommand)] #[command(subcommand)]
subcommand: SubCommand, subcommand: SubCommand,
/// How many blocking tokio threads to spawn to process disk reads/writes. /// How many maximum blocking tokio threads to spawn to process disk reads/writes.
/// Might want to increase if the disk is slow. /// This will indicate how many parallel reads/writes can happen at a moment in time.
/// The higher the number, the more the memory usage.
#[arg(long = "max-blocking-threads", default_value = "8")] #[arg(long = "max-blocking-threads", default_value = "8")]
max_blocking_threads: u16, max_blocking_threads: u16,
/// If set, will write to disk in background and not inline with peer. /// If set, will write to disk in background and not inline with peer.
/// Might be useful if the disk is slow. /// Useful if the disk is slow or its latency is very unstable (e.g. HDD or old SSD).
#[arg(long = "defer-writes", default_value = "false")] #[arg(long = "defer-writes", default_value = "false")]
defer_writes: bool, defer_writes: bool,
/// Use mmap (file-backed) for storage. /// Use mmap (file-backed) for storage. Any advantages are questionable and unproven.
/// If you use it, you know what you are doing.
#[arg(long)] #[arg(long)]
experimental_mmap_storage: bool, experimental_mmap_storage: bool,
} }
@ -300,8 +301,15 @@ async fn async_main(opts: Opts) -> anyhow::Result<()> {
default_defer_writes: opts.defer_writes, default_defer_writes: opts.defer_writes,
default_storage_factory: Some({ default_storage_factory: Some({
fn wrap<S: StorageFactory + Clone>(s: S) -> impl StorageFactory { fn wrap<S: StorageFactory + Clone>(s: S) -> impl StorageFactory {
// TimingStorageFactory::new("hdd".to_owned(), SlowStorageFactory::new(s)) #[cfg(feature = "debug_slow_disk")]
TimingStorageFactory::new("hdd".to_owned(), s) {
use librqbit::middleware::{
slow::SlowStorageFactory, timing::TimingStorageFactory,
};
TimingStorageFactory::new("hdd".to_owned(), SlowStorageFactory::new(s))
}
#[cfg(not(feature = "debug_slow_disk"))]
s
} }
if opts.experimental_mmap_storage { if opts.experimental_mmap_storage {