block: maximum number of simultaneous reads

src/block/manager.rs

@@ -50,6 +50,8 @@ pub const INLINE_THRESHOLD: usize = 3072;
 // to delete the block locally.
 pub(crate) const BLOCK_GC_DELAY: Duration = Duration::from_secs(600);
 
+const BLOCK_READ_SEMAPHORE_TIMEOUT: Duration = Duration::from_secs(15);
+
 /// RPC messages used to share blocks of data between nodes
 #[derive(Debug, Serialize, Deserialize)]
 pub enum BlockRpc {
@@ -87,6 +89,7 @@ pub struct BlockManager {
 	disable_scrub: bool,
 
 	mutation_lock: Vec<Mutex<BlockManagerLocked>>,
+	read_semaphore: Semaphore,
 
 	pub rc: BlockRc,
 	pub resync: BlockResyncManager,
@@ -176,6 +179,8 @@ impl BlockManager {
 				.iter()
 				.map(|_| Mutex::new(BlockManagerLocked()))
 				.collect::<Vec<_>>(),
+
+			read_semaphore: Semaphore::new(config.block_max_concurrent_reads),
 			rc,
 			resync,
 			system,
@@ -581,6 +586,15 @@ impl BlockManager {
 	) -> Result<DataBlock, Error> {
 		let (header, path) = block_path.as_parts_ref();
 
+		let permit = tokio::select! {
+			sem = self.read_semaphore.acquire() => sem.ok_or_message("acquire read semaphore")?,
+			_ = tokio::time::sleep(BLOCK_READ_SEMAPHORE_TIMEOUT) => {
+				self.metrics.block_read_semaphore_timeouts.add(1);
+				debug!("read block {:?}: read_semaphore acquire timeout", hash);
+				return Err(Error::Message("read block: read_semaphore acquire timeout".into()));
+			}
+		};
+
 		let mut f = fs::File::open(&path).await?;
 		let mut data = vec![];
 		f.read_to_end(&mut data).await?;
@@ -605,6 +619,8 @@ impl BlockManager {
 			return Err(Error::CorruptData(*hash));
 		}
 
+		drop(permit);
+
 		Ok(data)
 	}
 

src/block/metrics.rs

@@ -22,6 +22,7 @@ pub struct BlockManagerMetrics {
 
 	pub(crate) bytes_read: BoundCounter<u64>,
 	pub(crate) block_read_duration: BoundValueRecorder<f64>,
+	pub(crate) block_read_semaphore_timeouts: BoundCounter<u64>,
 	pub(crate) bytes_written: BoundCounter<u64>,
 	pub(crate) block_write_duration: BoundValueRecorder<f64>,
 	pub(crate) delete_counter: BoundCounter<u64>,
@@ -119,6 +120,11 @@ impl BlockManagerMetrics {
 				.with_description("Duration of block read operations")
 				.init()
 				.bind(&[]),
+			block_read_semaphore_timeouts: meter
+				.u64_counter("block.read_semaphore_timeouts")
+				.with_description("Number of block reads that failed due to semaphore acquire timeout")
+				.init()
+				.bind(&[]),
 			bytes_written: meter
 				.u64_counter("block.bytes_written")
 				.with_description("Number of bytes written to disk")

src/util/config.rs

@@ -75,6 +75,10 @@ pub struct Config {
 	)]
 	pub block_ram_buffer_max: usize,
 
+	/// Maximum number of concurrent reads of block files on disk
+	#[serde(default = "default_block_max_concurrent_reads")]
+	pub block_max_concurrent_reads: usize,
+
 	/// Skip the permission check of secret files. Useful when
 	/// POSIX ACLs (or more complex chmods) are used.
 	#[serde(default)]
@@ -280,6 +284,9 @@ fn default_block_size() -> usize {
 fn default_block_ram_buffer_max() -> usize {
 	256 * 1024 * 1024
 }
+fn default_block_max_concurrent_reads() -> usize {
+	16
+}
 
 fn default_consistency_mode() -> String {
 	"consistent".into()