Repository: paritytech/substrate
File: client/rpc-spec-v2/src/chain_head/subscription/inner.rs (Rust)

Code before:
	}

	fn advance_unpin(&mut self) {
		match self {
			BlockStateMachine::Registered => *self = BlockStateMachine::Unpinned,
			BlockStateMachine::FullyRegistered => *self = BlockStateMachine::FullyUnpinned,
			_ => (),
		}
	}

	fn was_unpinned(&self) -> bool {
		match self {
			BlockStateMachine::Unpinned => true,
			BlockStateMachine::FullyUnpinned => true,
			_ => false,
		}
	}
}

/// Limit the number of ongoing operations across methods.
#[derive(Clone)]
struct LimitOperations {
	/// Maximum number of operations that can be executed at the same time.
	max_operations: usize,
	/// The number of ongoing operations for this subscription.
	ongoing: Arc<Mutex<usize>>,
}

impl LimitOperations {
	/// Constructs a new [`LimitOperations`].
	fn new(max_operations: usize) -> Self {
		LimitOperations { max_operations, ongoing: Default::default() }
	}

	/// Reserves capacity to execute at least one operation and at most the requested items.
	///
	/// Dropping [`PermitOperations`] without executing an operation will release
	/// the reserved capacity.
	///
	/// Returns nothing if there's no space available, else returns a permit
	/// that guarantees that at least one operation can be executed.
	fn reserve_at_most(&self, to_reserve: usize) -> Option<PermitOperations> {
		let mut ongoing = self.ongoing.lock();

		// The maximum capacity.
		let capacity = self.max_operations.saturating_sub(*ongoing);
		// The number of operations we can reserve.
		let reserved = capacity.min(to_reserve);

		if reserved == 0 {
			None

Review comment (suggestion):
maybe you could replace this by [tokio::sync::Semaphore](https://docs.rs/tokio/latest/tokio/sync/struct.Semaphore.html) to avoid implement similar functionality here or implement a wrapper on-top of the semaphore would be cleaner I reckon.

Code after:

	fn advance_unpin(&mut self) {
		match self {
			BlockStateMachine::Registered => *self = BlockStateMachine::Unpinned,
			BlockStateMachine::FullyRegistered => *self = BlockStateMachine::FullyUnpinned,
			_ => (),
		}
	}

	fn was_unpinned(&self) -> bool {
		match self {
			BlockStateMachine::Unpinned => true,
			BlockStateMachine::FullyUnpinned => true,
			_ => false,
		}
	}
}

/// Limit the number of ongoing operations across methods.
struct LimitOperations {
	/// Limit the number of ongoing operations for this subscription.
	semaphore: Arc<tokio::sync::Semaphore>,
}

impl LimitOperations {
	/// Constructs a new [`LimitOperations`].
	fn new(max_operations: usize) -> Self {
		LimitOperations { semaphore: Arc::new(tokio::sync::Semaphore::new(max_operations)) }
	}

	/// Reserves capacity to execute at least one operation and at most the requested items.
	///
	/// Dropping [`PermitOperations`] without executing an operation will release
	/// the reserved capacity.
	///
	/// Returns nothing if there's no space available, else returns a permit
	/// that guarantees that at least one operation can be executed.
	fn reserve_at_most(&self, to_reserve: usize) -> Option<PermitOperations> {
		let num_ops = std::cmp::min(self.semaphore.available_permits(), to_reserve);

		if num_ops == 0 {
			return None
		}

		let permits = Arc::clone(&self.semaphore)
			.try_acquire_many_owned(num_ops.try_into().ok()?)
			.ok()?;

		Some(PermitOperations { num_ops, _permit: permits })
	}
}