sort_governor/service/
handle.rs

1//! The cheap, cloneable client side of the Sorter — the only way callers
2//! reach the governor.
3
4use std::path::PathBuf;
5use std::sync::Arc;
6
7use tokio::sync::{
8    mpsc,
9    oneshot,
10};
11
12use crate::config::SorterConfig;
13use crate::error::SorterError;
14use crate::service::actor::SorterActor;
15use crate::service::command::SorterCommand;
16use crate::service::lease::SortLease;
17use crate::service::pressure::MemoryPressure;
18use crate::service::stats::SorterStats;
19use crate::spec::SortSpec;
20
21/// Bounded depth of the admission queue.
22const COMMAND_CHANNEL_CAPACITY: usize = 256;
23
24/// A handle to the process Sorter. Clone freely; all clones address the one
25/// governor actor.
26#[derive(Clone)]
27pub struct SorterHandle {
28    tx: mpsc::Sender<SorterCommand>,
29}
30
31impl SorterHandle {
32    /// Spawn the governor and return a handle to it. `fd_budget` is the
33    /// number of usable file descriptors the Sorter may ration (the process
34    /// soft limit minus a safety margin); sorts spill under unique
35    /// directories beneath `scratch_root`.
36    #[must_use]
37    pub fn spawn(
38        config: SorterConfig,
39        fd_budget: u32,
40        pressure: Arc<dyn MemoryPressure>,
41        scratch_root: PathBuf,
42    ) -> Self {
43        let (tx, rx) = mpsc::channel(COMMAND_CHANNEL_CAPACITY);
44        let actor = SorterActor::new(config, fd_budget, pressure, scratch_root);
45        tokio::spawn(actor.run(rx));
46        Self { tx }
47    }
48
49    /// Admit a sort: the governor plans it, reserves its resources, and
50    /// returns a lease. The returned lease must be held until the sort's
51    /// value stream is fully consumed.
52    ///
53    /// # Errors
54    ///
55    /// Returns [`SorterError::Gone`] if the governor is no longer running.
56    pub async fn submit(&self, spec: SortSpec) -> Result<SortLease, SorterError> {
57        let (reply, rx) = oneshot::channel();
58        self.tx
59            .send(SorterCommand::Submit { spec, reply })
60            .await
61            .map_err(|_| SorterError::Gone)?;
62        rx.await.map_err(|_| SorterError::Gone)?
63    }
64
65    /// Read the governor's current counters.
66    ///
67    /// # Errors
68    ///
69    /// Returns [`SorterError::Gone`] if the governor is no longer running.
70    pub async fn stats(&self) -> Result<SorterStats, SorterError> {
71        let (reply, rx) = oneshot::channel();
72        self.tx
73            .send(SorterCommand::Stats { reply })
74            .await
75            .map_err(|_| SorterError::Gone)?;
76        rx.await.map_err(|_| SorterError::Gone)
77    }
78}
sort_governor/service/handle.rs

sort_governor/service/
handle.rs
