use std::{
fmt,
sync::{
Arc,
atomic::{AtomicU64, AtomicUsize, Ordering},
},
time::Duration,
};
use reovim_arch::sync::{Condvar, Mutex};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct ScopeId(u64);
impl ScopeId {
pub(crate) fn new() -> Self {
static COUNTER: AtomicU64 = AtomicU64::new(1);
Self(COUNTER.fetch_add(1, Ordering::Relaxed))
}
#[inline]
#[must_use]
pub const fn as_u64(&self) -> u64 {
self.0
}
}
impl fmt::Display for ScopeId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "scope-{}", self.0)
}
}
struct ScopeInner {
id: ScopeId,
in_flight: AtomicUsize,
condvar: (Mutex<()>, Condvar),
}
#[derive(Clone)]
pub struct EventScope {
inner: Arc<ScopeInner>,
}
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(3);
impl EventScope {
#[must_use]
pub fn new() -> Self {
Self {
inner: Arc::new(ScopeInner {
id: ScopeId::new(),
in_flight: AtomicUsize::new(0),
condvar: (Mutex::new(()), Condvar::new()),
}),
}
}
#[inline]
#[must_use]
pub fn id(&self) -> ScopeId {
self.inner.id
}
#[inline]
pub fn increment(&self) {
self.inner.in_flight.fetch_add(1, Ordering::SeqCst);
}
#[inline]
pub fn decrement(&self) {
let prev = self.inner.in_flight.fetch_sub(1, Ordering::SeqCst);
debug_assert!(prev > 0, "EventScope::decrement called when counter is 0");
if prev == 1 {
let (_lock, condvar) = &self.inner.condvar;
condvar.notify_all();
}
}
#[inline]
#[must_use]
pub fn in_flight(&self) -> usize {
self.inner.in_flight.load(Ordering::SeqCst)
}
#[inline]
#[must_use]
pub fn is_complete(&self) -> bool {
self.in_flight() == 0
}
pub fn wait(&self) {
let (mutex, condvar) = &self.inner.condvar;
let mut guard = mutex.lock();
while self.in_flight() > 0 {
condvar.wait(&mut guard);
}
}
#[must_use]
#[allow(clippy::significant_drop_tightening)] #[cfg_attr(coverage_nightly, coverage(off))]
pub fn wait_timeout(&self, timeout: Duration) -> bool {
let (mutex, condvar) = &self.inner.condvar;
let mut guard = mutex.lock();
let deadline = std::time::Instant::now() + timeout;
while self.in_flight() > 0 {
let remaining = deadline.saturating_duration_since(std::time::Instant::now());
if remaining.is_zero() {
return false;
}
let result = condvar.wait_for(&mut guard, remaining);
if result.timed_out() && self.in_flight() > 0 {
return false;
}
}
true
}
#[must_use]
pub fn wait_with_default_timeout(&self) -> bool {
self.wait_timeout(DEFAULT_TIMEOUT)
}
}
impl Default for EventScope {
fn default() -> Self {
Self::new()
}
}
impl fmt::Debug for EventScope {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("EventScope")
.field("id", &self.inner.id)
.field("in_flight", &self.in_flight())
.finish()
}
}