vv_agent/runtime/backends/distributed/
backend.rs

1use std::time::Duration;
2
3use std::sync::Arc;
4
5use super::super::RuntimeRecipe;
6use super::capabilities::DistributedCapabilityRegistry;
7use super::dispatch::CycleDispatcher;
8use super::driver::CycleEnqueuer;
9use super::{DEFAULT_CYCLE_NAME, DEFAULT_LEASE_DURATION_MS};
10
11#[derive(Clone)]
12pub struct DistributedBackend {
13    pub(super) runtime_recipe: Option<RuntimeRecipe>,
14    pub(super) cycle_dispatcher: Option<Arc<dyn CycleDispatcher>>,
15    pub(super) capability_registry: Option<DistributedCapabilityRegistry>,
16    pub(super) cycle_enqueuer: Option<Arc<dyn CycleEnqueuer>>,
17    pub(super) cycle_name: String,
18    pub(super) dispatch_timeout: Duration,
19    pub(super) lease_duration_ms: u64,
20}
21
22impl std::fmt::Debug for DistributedBackend {
23    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        formatter
25            .debug_struct("DistributedBackend")
26            .field("runtime_recipe", &self.runtime_recipe)
27            .field("has_cycle_dispatcher", &self.cycle_dispatcher.is_some())
28            .field(
29                "has_capability_registry",
30                &self.capability_registry.is_some(),
31            )
32            .field("has_cycle_enqueuer", &self.cycle_enqueuer.is_some())
33            .field("cycle_name", &self.cycle_name)
34            .field("dispatch_timeout", &self.dispatch_timeout)
35            .field("lease_duration_ms", &self.lease_duration_ms)
36            .finish()
37    }
38}
39
40impl DistributedBackend {
41    pub fn inline_fallback() -> Self {
42        Self {
43            runtime_recipe: None,
44            cycle_dispatcher: None,
45            capability_registry: None,
46            cycle_enqueuer: None,
47            cycle_name: DEFAULT_CYCLE_NAME.to_string(),
48            dispatch_timeout: Duration::from_secs(10 * 60),
49            lease_duration_ms: DEFAULT_LEASE_DURATION_MS,
50        }
51    }
52
53    pub fn new(runtime_recipe: RuntimeRecipe, cycle_dispatcher: Arc<dyn CycleDispatcher>) -> Self {
54        Self {
55            runtime_recipe: Some(runtime_recipe),
56            cycle_dispatcher: Some(cycle_dispatcher),
57            capability_registry: None,
58            cycle_enqueuer: None,
59            cycle_name: DEFAULT_CYCLE_NAME.to_string(),
60            dispatch_timeout: Duration::from_secs(10 * 60),
61            lease_duration_ms: DEFAULT_LEASE_DURATION_MS,
62        }
63    }
64
65    pub fn with_cycle_name(mut self, cycle_name: impl Into<String>) -> Self {
66        self.cycle_name = cycle_name.into();
67        self
68    }
69
70    pub fn with_dispatch_timeout(mut self, timeout: Duration) -> Self {
71        assert!(!timeout.is_zero(), "dispatch timeout must be positive");
72        self.dispatch_timeout = timeout;
73        self
74    }
75
76    pub fn with_lease_duration(mut self, duration: Duration) -> Self {
77        let duration_ms = u64::try_from(duration.as_millis())
78            .expect("lease duration milliseconds must fit in u64");
79        assert!(duration_ms > 0, "lease duration must be positive");
80        self.lease_duration_ms = duration_ms;
81        self
82    }
83
84    pub fn runtime_recipe(&self) -> Option<&RuntimeRecipe> {
85        self.runtime_recipe.as_ref()
86    }
87
88    pub fn cycle_name(&self) -> &str {
89        &self.cycle_name
90    }
91
92    pub fn lease_duration_ms(&self) -> u64 {
93        self.lease_duration_ms
94    }
95
96    pub(crate) fn has_nonblocking_driver(&self) -> bool {
97        self.runtime_recipe.is_some()
98            && self.capability_registry.is_some()
99            && self.cycle_enqueuer.is_some()
100    }
101
102    pub fn parallel_map<T, R, F>(&self, function: F, items: Vec<T>) -> Vec<R>
103    where
104        F: Fn(T) -> R,
105    {
106        items.into_iter().map(function).collect()
107    }
108}
vv_agent/runtime/backends/distributed/backend.rs

vv_agent/runtime/backends/distributed/
backend.rs
