vv_agent/runtime/backends/distributed/
backend.rs

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

vv_agent/runtime/backends/distributed/
backend.rs
