rust-data-processing 0.1.8

Schema-first ingestion (CSV, JSON, Parquet, Excel) into an in-memory DataSet, plus Polars-backed pipelines, SQL, profiling, validation, and map/reduce-style processing.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

use std::sync::{Condvar, Mutex};
use std::time::{Duration, Instant};

/// A small, blocking counting semaphore.
///
/// Used to implement throttling/backpressure for chunked execution.
pub struct Semaphore {
    permits: Mutex<usize>,
    cv: Condvar,
}

impl Semaphore {
    pub fn new(permits: usize) -> Self {
        assert!(permits > 0, "permits must be > 0");
        Self {
            permits: Mutex::new(permits),
            cv: Condvar::new(),
        }
    }

    /// Acquire one permit, blocking until available.
    ///
    /// Returns the time spent waiting (zero if no wait was required).
    pub fn acquire(&self) -> Duration {
        let start = Instant::now();
        let mut waited = false;
        let mut g = self.permits.lock().expect("semaphore mutex poisoned");
        while *g == 0 {
            waited = true;
            g = self.cv.wait(g).expect("semaphore mutex poisoned");
        }
        *g -= 1;
        if waited {
            start.elapsed()
        } else {
            Duration::ZERO
        }
    }

    /// Release one permit.
    pub fn release(&self) {
        let mut g = self.permits.lock().expect("semaphore mutex poisoned");
        *g += 1;
        self.cv.notify_one();
    }
}