some_global_executor 0.1.6

Reference thread-per-core executor for the some_executor crate.
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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

// SPDX-License-Identifier: MIT OR Apache-2.0

//! Exposes the thread-pool registry through exfiltrate's `snapshot` command.

use exfiltrate::provider::{Provider, ProviderResult, Row, SnapshotRequest};
use wasm_lite_std::time::Instant;

use crate::registry;

/// Registers the `thread_pool` subsystem. Idempotent.
///
/// ```no_run
/// # // no_run because: `begin` opens exfiltrate's listening socket and installs
/// # // process-global state, which a doctest process must not do.
/// exfiltrate::begin();
/// some_global_executor::exfiltrate_provider::install();
/// ```
pub fn install() {
    exfiltrate::provider::add_provider(ThreadPools);
}

struct ThreadPools;

impl Provider for ThreadPools {
    fn subsystem(&self) -> &'static str {
        "thread_pool"
    }

    fn description(&self) -> &'static str {
        "executor thread pools: worker count, tasks in flight, and whether the pool is alive"
    }

    fn snapshot(&self, request: &SnapshotRequest<'_>) -> ProviderResult {
        let Some(entries) = registry::entries() else {
            return ProviderResult::Busy;
        };
        let stats = registry::stats();

        let wanted: Option<u64> = match request.selector() {
            Some(selector) => match selector.parse::<u64>() {
                Ok(id) => Some(id),
                Err(_) => {
                    return ProviderResult::Unavailable(format!(
                        "--id must be a pool id; got {selector:?}"
                    ));
                }
            },
            None => None,
        };

        let now = Instant::now();
        let mut rows = Vec::new();
        for entry in entries {
            if request.should_stop() {
                return ProviderResult::Partial(rows, "deadline".to_string());
            }
            if let Some(id) = wanted
                && entry.id != id
            {
                continue;
            }
            rows.push(
                Row::new()
                    .support("id", entry.id)
                    .support("threads", entry.threads as u64)
                    // Read live rather than mirrored: a copy taken at
                    // registration would be stale exactly when it matters.
                    .support("running_tasks", entry.running_tasks())
                    .support("alive", entry.alive)
                    .support(
                        "age_ms",
                        now.duration_since(entry.created).as_millis() as u64,
                    )
                    // The name is chosen by whoever built the pool, so it is
                    // the application's own vocabulary.
                    .local("name", entry.name.clone()),
            );
        }

        if stats.dropped > 0 {
            return ProviderResult::Partial(
                rows,
                format!(
                    "the registry has dropped {} pool(s) to stay within its capacity of {}; \
                     raise SOME_GLOBAL_EXECUTOR_REGISTRY_CAPACITY to keep more",
                    stats.dropped, stats.capacity
                ),
            );
        }
        ProviderResult::Rows(rows)
    }
}