xet_runtime/core/
context.rs

1use std::sync::Arc;
2
3#[cfg(not(target_family = "wasm"))]
4use tokio::runtime::Handle as TokioRuntimeHandle;
5#[cfg(not(target_family = "wasm"))]
6use tracing::info;
7
8use super::XetCommon;
9use super::runtime::XetRuntime;
10use crate::config::XetConfig;
11use crate::error::RuntimeError;
12
13/// Bundles the thread pool, configuration, and shared state into a single clonable handle.
14///
15/// Every major struct in the codebase should accept `&XetContext` in its constructor
16/// and store a clone. This replaces the thread-local globals, allowing multiple
17/// independent runtimes within the same process.
18#[derive(Clone)]
19pub struct XetContext {
20    pub runtime: Arc<XetRuntime>,
21    pub config: Arc<XetConfig>,
22    pub common: Arc<XetCommon>,
23}
24
25impl XetContext {
26    /// Creates a context from a pre-built thread pool and configuration.
27    ///
28    /// The reconstruction buffer values are normalized here, before the config is
29    /// frozen, so the buffer semaphore bounds and the per-download scaling target
30    /// always read identical, coherent values. `XetConfig::new` already normalizes
31    /// env-provided values; this covers configs mutated after construction (Rust or
32    /// Python `with_config`), since this is the choke point every entry path
33    /// funnels through.
34    pub fn new(mut config: XetConfig, runtime: Arc<XetRuntime>) -> Self {
35        config.reconstruction.normalize();
36        let config = Arc::new(config);
37        let common = Arc::new(XetCommon::new(&config));
38        Self {
39            runtime,
40            config,
41            common,
42        }
43    }
44
45    /// Creates a context with default configuration and an auto-detected thread pool.
46    ///
47    /// If called from an owned runtime worker thread, reuses that owned [`XetRuntime`].
48    /// Otherwise, if called from within an existing tokio runtime, wraps that runtime.
49    /// If neither is available, spins up a new owned tokio thread pool.
50    #[allow(clippy::should_implement_trait)]
51    pub fn default() -> Result<Self, RuntimeError> {
52        Self::with_config(XetConfig::new())
53    }
54
55    /// Creates a context with the given configuration and an auto-detected thread pool.
56    ///
57    /// Follows the same runtime selection as [`default`](Self::default):
58    /// reuse an owned runtime if available, wrap an existing tokio handle, or create a new one.
59    pub fn with_config(config: XetConfig) -> Result<Self, RuntimeError> {
60        #[cfg(not(target_family = "wasm"))]
61        let runtime = if let Some(runtime) = XetRuntime::current_if_exists() {
62            runtime
63        } else if let Ok(handle) = TokioRuntimeHandle::try_current()
64            && Self::handle_meets_requirements(&handle)
65        {
66            info!(
67                "Detected compatible existing Tokio runtime; using external handle instead of creating a new thread pool"
68            );
69            XetRuntime::from_external(handle)
70        } else {
71            XetRuntime::new(&config)?
72        };
73
74        #[cfg(target_family = "wasm")]
75        let runtime = XetRuntime::new(&config)?;
76
77        Ok(Self::new(config, runtime))
78    }
79
80    /// Wraps a caller-provided tokio handle with the given configuration.
81    #[cfg(not(target_family = "wasm"))]
82    pub fn from_external(rt_handle: TokioRuntimeHandle, config: XetConfig) -> Self {
83        Self::new(config, XetRuntime::from_external(rt_handle))
84    }
85
86    /// Checks whether a tokio handle meets the requirements for use with xet.
87    #[cfg(not(target_family = "wasm"))]
88    pub fn handle_meets_requirements(handle: &TokioRuntimeHandle) -> bool {
89        XetRuntime::handle_meets_requirements(handle)
90    }
91
92    /// Returns an error if the runtime is in the middle of a SIGINT shutdown.
93    #[inline]
94    pub fn check_sigint_shutdown(&self) -> Result<(), RuntimeError> {
95        if self.runtime.in_sigint_shutdown() {
96            Err(RuntimeError::KeyboardInterrupt)
97        } else {
98            Ok(())
99        }
100    }
101}
102
103impl std::fmt::Debug for XetContext {
104    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105        f.debug_struct("XetContext")
106            .field("runtime", &self.runtime)
107            .field("config", &"...")
108            .field("common", &"...")
109            .finish()
110    }
111}
xet_runtime/core/context.rs

xet_runtime/core/
context.rs
