moirai-executor 0.9.1

Hybrid executor implementation for Moirai concurrency library
Documentation 
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
47
48
49
50

use std::sync::Arc;

use moirai_core::{
    Priority,
    error::ExecutorResult,
    task::{TaskHandle, TaskId},
};

use super::super::AsyncFutureState;
use crate::metrics::ExecutorMetrics;
use crate::registry::TaskRegistry;
use crate::schedule::{WorkClass, WorkSubmit};

impl WorkSubmit for Arc<TaskRegistry> {
    fn schedule<C, F>(
        &self,
        _priority: Priority,
        _locality_hint: Option<usize>,
        task: F,
    ) -> ExecutorResult<()>
    where
        C: WorkClass,
        F: FnOnce(usize) + Send + 'static,
    {
        task(0);
        Ok(())
    }
}

#[test]
fn scheduled_lifecycle_retires_before_its_registry_owner() {
    let registry = Arc::new(TaskRegistry::new());
    // SAFETY: the async state owns the only remaining registry Arc
    // through its scheduler field, which is declared after lifecycle.
    let (task_id, lifecycle) = unsafe { registry.register_next_scheduled_task() };
    let registry_owner = Arc::downgrade(&registry);
    let (_handle, result_sender) = TaskHandle::<()>::new_pending(TaskId(task_id));
    let state = AsyncFutureState::new(
        Arc::clone(&registry),
        std::future::pending::<()>(),
        lifecycle,
        result_sender,
        Arc::new(ExecutorMetrics::new()),
    );

    drop(registry);
    assert!(registry_owner.upgrade().is_some());
    drop(state);
    assert!(registry_owner.upgrade().is_none());
}