app

Local-first trade for farms and co-ops
git clone https://radroots.dev/git/app.git
Log | Files | Refs | README | LICENSE

blocking.rs (3738B)

      1 use std::sync::Arc;
      2 use std::time::Instant;
      3 
      4 use tokio::runtime::Handle;
      5 use tokio::sync::Semaphore;
      6 
      7 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
      8 pub(crate) enum BlockingExecutionError {
      9     DeadlineElapsed,
     10     Saturated,
     11     TaskFailed,
     12 }
     13 
     14 #[derive(Clone)]
     15 pub(crate) struct BoundedBlockingExecutor {
     16     permits: Arc<Semaphore>,
     17     runtime: Handle,
     18 }
     19 
     20 impl BoundedBlockingExecutor {
     21     pub(crate) fn new(capacity: usize, runtime: &Handle) -> Self {
     22         Self {
     23             permits: Arc::new(Semaphore::new(capacity)),
     24             runtime: runtime.clone(),
     25         }
     26     }
     27 
     28     pub(crate) async fn execute<T, F>(
     29         &self,
     30         deadline: Instant,
     31         operation: F,
     32     ) -> Result<T, BlockingExecutionError>
     33     where
     34         T: Send + 'static,
     35         F: FnOnce() -> T + Send + 'static,
     36     {
     37         if Instant::now() >= deadline {
     38             return Err(BlockingExecutionError::DeadlineElapsed);
     39         }
     40         let permit = self
     41             .permits
     42             .clone()
     43             .try_acquire_owned()
     44             .map_err(|_| BlockingExecutionError::Saturated)?;
     45         self.runtime
     46             .spawn_blocking(move || {
     47                 let _permit = permit;
     48                 operation()
     49             })
     50             .await
     51             .map_err(|_| BlockingExecutionError::TaskFailed)
     52     }
     53 }
     54 
     55 #[cfg(test)]
     56 mod tests {
     57     use std::sync::{Arc, Condvar, Mutex};
     58     use std::time::{Duration, Instant};
     59 
     60     use tokio::sync::oneshot;
     61 
     62     use super::{BlockingExecutionError, BoundedBlockingExecutor};
     63 
     64     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     65     async fn executor_rejects_saturation_without_starting_excess_work() {
     66         let executor = BoundedBlockingExecutor::new(1, &tokio::runtime::Handle::current());
     67         let release = Arc::new((Mutex::new(false), Condvar::new()));
     68         let first_release = Arc::clone(&release);
     69         let (started, started_rx) = oneshot::channel();
     70         let first_executor = executor.clone();
     71         let first = tokio::spawn(async move {
     72             first_executor
     73                 .execute(Instant::now() + Duration::from_secs(5), move || {
     74                     let _ = started.send(());
     75                     let (lock, ready) = &*first_release;
     76                     let mut released = lock.lock().expect("release lock");
     77                     while !*released {
     78                         released = ready.wait(released).expect("release wait");
     79                     }
     80                     7
     81                 })
     82                 .await
     83         });
     84         started_rx.await.expect("first work starts");
     85 
     86         let second = executor
     87             .execute(Instant::now() + Duration::from_secs(5), || 9)
     88             .await;
     89         assert_eq!(second, Err(BlockingExecutionError::Saturated));
     90 
     91         let (lock, ready) = &*release;
     92         *lock.lock().expect("release lock") = true;
     93         ready.notify_all();
     94         assert_eq!(first.await.expect("first join"), Ok(7));
     95     }
     96 
     97     #[tokio::test]
     98     async fn executor_rejects_expired_work_before_spawn() {
     99         let executor = BoundedBlockingExecutor::new(1, &tokio::runtime::Handle::current());
    100         let result = executor.execute(Instant::now(), || 1).await;
    101         assert_eq!(result, Err(BlockingExecutionError::DeadlineElapsed));
    102     }
    103 
    104     #[tokio::test]
    105     async fn executor_classifies_panicked_work_without_panicking_the_actor() {
    106         let executor = BoundedBlockingExecutor::new(1, &tokio::runtime::Handle::current());
    107         let result = executor
    108             .execute::<(), _>(Instant::now() + Duration::from_secs(1), || {
    109                 panic!("test-only blocking task failure");
    110             })
    111             .await;
    112         assert_eq!(result, Err(BlockingExecutionError::TaskFailed));
    113     }
    114 }