Merge branch 'main-v1' into sync-v2-to-v1

src/block/manager.rs

@@ -48,6 +48,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 {
@@ -85,6 +87,7 @@ pub struct BlockManager {
 	disable_scrub: bool,
 
 	mutation_lock: Vec<Mutex<BlockManagerLocked>>,
+	read_semaphore: Semaphore,
 
 	pub rc: BlockRc,
 	pub resync: BlockResyncManager,
@@ -174,6 +177,8 @@ impl BlockManager {
 				.iter()
 				.map(|_| Mutex::new(BlockManagerLocked()))
 				.collect::<Vec<_>>(),
+
+			read_semaphore: Semaphore::new(config.block_max_concurrent_reads),
 			rc,
 			resync,
 			system,
@@ -416,8 +421,8 @@ impl BlockManager {
 	}
 
 	/// Get number of items in the refcount table
-	pub fn rc_len(&self) -> Result<usize, Error> {
-		Ok(self.rc.rc_table.len()?)
+	pub fn rc_approximate_len(&self) -> Result<usize, Error> {
+		Ok(self.rc.rc_table.approximate_len()?)
 	}
 
 	/// Send command to start/stop/manager scrub worker
@@ -435,7 +440,7 @@ impl BlockManager {
 
 	/// List all resync errors
 	pub fn list_resync_errors(&self) -> Result<Vec<BlockResyncErrorInfo>, Error> {
-		let mut blocks = Vec::with_capacity(self.resync.errors.len()?);
+		let mut blocks = Vec::with_capacity(self.resync.errors.approximate_len()?);
 		for ent in self.resync.errors.iter()? {
 			let (hash, cnt) = ent?;
 			let cnt = ErrorCounter::decode(&cnt);
@@ -565,9 +570,6 @@ impl BlockManager {
 			match self.find_block(hash).await {
 				Some(p) => self.read_block_from(hash, &p).await,
 				None => {
-					// Not found but maybe we should have had it ??
-					self.resync
-						.put_to_resync(hash, 2 * self.system.rpc_helper().rpc_timeout())?;
 					return Err(Error::Message(format!(
 						"block {:?} not found on node",
 						hash
@@ -589,6 +591,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?;
@@ -613,6 +624,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>,
@@ -50,7 +51,7 @@ impl BlockManagerMetrics {
 				.init(),
 			_rc_size: meter
 				.u64_value_observer("block.rc_size", move |observer| {
-					if let Ok(value) = rc_tree.len() {
+					if let Ok(value) = rc_tree.approximate_len() {
 						observer.observe(value as u64, &[])
 					}
 				})
@@ -58,7 +59,7 @@ impl BlockManagerMetrics {
 				.init(),
 			_resync_queue_len: meter
 				.u64_value_observer("block.resync_queue_length", move |observer| {
-					if let Ok(value) = resync_queue.len() {
+					if let Ok(value) = resync_queue.approximate_len() {
 						observer.observe(value as u64, &[]);
 					}
 				})
@@ -68,7 +69,7 @@ impl BlockManagerMetrics {
 				.init(),
 			_resync_errored_blocks: meter
 				.u64_value_observer("block.resync_errored_blocks", move |observer| {
-					if let Ok(value) = resync_errors.len() {
+					if let Ok(value) = resync_errors.approximate_len() {
 						observer.observe(value as 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/block/resync.rs

@@ -104,13 +104,13 @@ impl BlockResyncManager {
 	}
 
 	/// Get length of resync queue
-	pub fn queue_len(&self) -> Result<usize, Error> {
-		Ok(self.queue.len()?)
+	pub fn queue_approximate_len(&self) -> Result<usize, Error> {
+		Ok(self.queue.approximate_len()?)
 	}
 
 	/// Get number of blocks that have an error
-	pub fn errors_len(&self) -> Result<usize, Error> {
-		Ok(self.errors.len()?)
+	pub fn errors_approximate_len(&self) -> Result<usize, Error> {
+		Ok(self.errors.approximate_len()?)
 	}
 
 	/// Clear the error counter for a block and put it in queue immediately
@@ -540,9 +540,11 @@ impl Worker for ResyncWorker {
 		}
 
 		WorkerStatus {
-			queue_length: Some(self.manager.resync.queue_len().unwrap_or(0) as u64),
+			queue_length: Some(self.manager.resync.queue_approximate_len().unwrap_or(0) as u64),
 			tranquility: Some(tranquility),
-			persistent_errors: Some(self.manager.resync.errors_len().unwrap_or(0) as u64),
+			persistent_errors: Some(
+				self.manager.resync.errors_approximate_len().unwrap_or(0) as u64
+			),
 			..Default::default()
 		}
 	}