atagen

macltok

public
Star0
Watch0
5d2974de5826bd24a1e122358050983ec5eb0644
3
ssh://brewery@pls.lobotomise.me:33/atagen/macltok.git

Blame: src/engine/executor.rs

1fe480c16a3catagen2026-09-26 08:44 UTC1
use std::future::Future;
1fe480c16a3catagen2026-09-26 08:44 UTC2
1fe480c16a3catagen2026-09-26 08:44 UTC3
use futures_util::{StreamExt, stream::FuturesUnordered};
1fe480c16a3catagen2026-09-26 08:44 UTC4
use matrix_sdk::SendOutsideWasm;
1fe480c16a3catagen2026-09-26 08:44 UTC5
pub(super) use matrix_sdk::executor::{AbortHandle, JoinError, JoinHandle, spawn};
1fe480c16a3catagen2026-09-26 08:44 UTC6
1fe480c16a3catagen2026-09-26 08:44 UTC7
#[cfg(not(target_family = "wasm"))]
a319bae745d8atagen2026-09-26 09:59 UTC8
pub(crate) type EngineBoxFuture<'a, T> = futures_util::future::BoxFuture<'a, T>;
1fe480c16a3catagen2026-09-26 08:44 UTC9
#[cfg(target_family = "wasm")]
a319bae745d8atagen2026-09-26 09:59 UTC10
pub(crate) type EngineBoxFuture<'a, T> = futures_util::future::LocalBoxFuture<'a, T>;
1fe480c16a3catagen2026-09-26 08:44 UTC11
1fe480c16a3catagen2026-09-26 08:44 UTC12
#[cfg(not(target_family = "wasm"))]
1fe480c16a3catagen2026-09-26 08:44 UTC13
pub(super) fn is_panic(error: &JoinError) -> bool {
1fe480c16a3catagen2026-09-26 08:44 UTC14
    error.is_panic()
1fe480c16a3catagen2026-09-26 08:44 UTC15
}
1fe480c16a3catagen2026-09-26 08:44 UTC16
7b376d620a3datagen2026-09-26 12:53 UTC17
/// the sdk's wasm join error has no payload, and a real panic traps before it gets here
1fe480c16a3catagen2026-09-26 08:44 UTC18
#[cfg(target_family = "wasm")]
1fe480c16a3catagen2026-09-26 08:44 UTC19
pub(super) fn is_panic(error: &JoinError) -> bool {
1fe480c16a3catagen2026-09-26 08:44 UTC20
    !error.is_cancelled()
1fe480c16a3catagen2026-09-26 08:44 UTC21
}
1fe480c16a3catagen2026-09-26 08:44 UTC22
1fe480c16a3catagen2026-09-26 08:44 UTC23
#[cfg(not(target_family = "wasm"))]
1fe480c16a3catagen2026-09-26 08:44 UTC24
pub(super) fn resume_unwind(error: JoinError) -> ! {
1fe480c16a3catagen2026-09-26 08:44 UTC25
    std::panic::resume_unwind(error.into_panic())
1fe480c16a3catagen2026-09-26 08:44 UTC26
}
1fe480c16a3catagen2026-09-26 08:44 UTC27
1fe480c16a3catagen2026-09-26 08:44 UTC28
#[cfg(target_family = "wasm")]
1fe480c16a3catagen2026-09-26 08:44 UTC29
pub(super) fn resume_unwind(error: JoinError) -> ! {
1fe480c16a3catagen2026-09-26 08:44 UTC30
    panic!("{error}")
1fe480c16a3catagen2026-09-26 08:44 UTC31
}
1fe480c16a3catagen2026-09-26 08:44 UTC32
1fe480c16a3catagen2026-09-26 08:44 UTC33
/// a `JoinSet` that runs on wasm too, and like it aborts whatever's left when dropped
1fe480c16a3catagen2026-09-26 08:44 UTC34
pub(super) struct TaskSet<T: 'static> {
1fe480c16a3catagen2026-09-26 08:44 UTC35
    handles: FuturesUnordered<JoinHandle<T>>,
1fe480c16a3catagen2026-09-26 08:44 UTC36
}
1fe480c16a3catagen2026-09-26 08:44 UTC37
1fe480c16a3catagen2026-09-26 08:44 UTC38
impl<T: 'static> TaskSet<T> {
1fe480c16a3catagen2026-09-26 08:44 UTC39
    pub(super) fn new() -> Self {
1fe480c16a3catagen2026-09-26 08:44 UTC40
        Self {
1fe480c16a3catagen2026-09-26 08:44 UTC41
            handles: FuturesUnordered::new(),
1fe480c16a3catagen2026-09-26 08:44 UTC42
        }
1fe480c16a3catagen2026-09-26 08:44 UTC43
    }
1fe480c16a3catagen2026-09-26 08:44 UTC44
1fe480c16a3catagen2026-09-26 08:44 UTC45
    pub(super) fn spawn(
1fe480c16a3catagen2026-09-26 08:44 UTC46
        &mut self,
1fe480c16a3catagen2026-09-26 08:44 UTC47
        task: impl Future<Output = T> + SendOutsideWasm + 'static,
1fe480c16a3catagen2026-09-26 08:44 UTC48
    ) -> AbortHandle
1fe480c16a3catagen2026-09-26 08:44 UTC49
    where
1fe480c16a3catagen2026-09-26 08:44 UTC50
        T: SendOutsideWasm,
1fe480c16a3catagen2026-09-26 08:44 UTC51
    {
1fe480c16a3catagen2026-09-26 08:44 UTC52
        let handle = spawn(task);
1fe480c16a3catagen2026-09-26 08:44 UTC53
        let abort = handle.abort_handle();
1fe480c16a3catagen2026-09-26 08:44 UTC54
        self.handles.push(handle);
1fe480c16a3catagen2026-09-26 08:44 UTC55
        abort
1fe480c16a3catagen2026-09-26 08:44 UTC56
    }
1fe480c16a3catagen2026-09-26 08:44 UTC57
1fe480c16a3catagen2026-09-26 08:44 UTC58
    pub(super) fn is_empty(&self) -> bool {
1fe480c16a3catagen2026-09-26 08:44 UTC59
        self.handles.is_empty()
1fe480c16a3catagen2026-09-26 08:44 UTC60
    }
1fe480c16a3catagen2026-09-26 08:44 UTC61
1fe480c16a3catagen2026-09-26 08:44 UTC62
    pub(super) async fn join_next(&mut self) -> Option<Result<T, JoinError>> {
1fe480c16a3catagen2026-09-26 08:44 UTC63
        self.handles.next().await
1fe480c16a3catagen2026-09-26 08:44 UTC64
    }
1fe480c16a3catagen2026-09-26 08:44 UTC65
1fe480c16a3catagen2026-09-26 08:44 UTC66
    fn abort_all(&self) {
1fe480c16a3catagen2026-09-26 08:44 UTC67
        for handle in &self.handles {
1fe480c16a3catagen2026-09-26 08:44 UTC68
            handle.abort();
1fe480c16a3catagen2026-09-26 08:44 UTC69
        }
1fe480c16a3catagen2026-09-26 08:44 UTC70
    }
1fe480c16a3catagen2026-09-26 08:44 UTC71
1fe480c16a3catagen2026-09-26 08:44 UTC72
    pub(super) async fn shutdown(&mut self) {
1fe480c16a3catagen2026-09-26 08:44 UTC73
        self.abort_all();
1fe480c16a3catagen2026-09-26 08:44 UTC74
        while self.handles.next().await.is_some() {}
1fe480c16a3catagen2026-09-26 08:44 UTC75
    }
1fe480c16a3catagen2026-09-26 08:44 UTC76
}
1fe480c16a3catagen2026-09-26 08:44 UTC77
1fe480c16a3catagen2026-09-26 08:44 UTC78
impl<T: 'static> Drop for TaskSet<T> {
1fe480c16a3catagen2026-09-26 08:44 UTC79
    fn drop(&mut self) {
1fe480c16a3catagen2026-09-26 08:44 UTC80
        self.abort_all();
1fe480c16a3catagen2026-09-26 08:44 UTC81
    }
1fe480c16a3catagen2026-09-26 08:44 UTC82
}