vv_agent/runtime/backends/distributed/
backend.rs

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

vv_agent/runtime/backends/distributed/
backend.rs
