use std::sync::Arc;
use anyhow::Result;
use dynamo_memory::nixl::NixlAgent;
use kvbm_config::KvbmConfig;
use tokio::runtime::{Handle, Runtime};
use velo::Messenger;
pub enum RuntimeHandle {
Owned(Arc<Runtime>),
Handle(Handle),
}
impl RuntimeHandle {
pub fn handle(&self) -> Handle {
match self {
RuntimeHandle::Owned(rt) => rt.handle().clone(),
RuntimeHandle::Handle(h) => h.clone(),
}
}
}
pub struct KvbmRuntimeBuilder {
config: KvbmConfig,
runtime: Option<RuntimeHandle>,
messenger: Option<Arc<Messenger>>,
nixl_agent: Option<NixlAgent>,
}
impl KvbmRuntimeBuilder {
pub fn new(config: KvbmConfig) -> Self {
Self {
config,
runtime: None,
messenger: None,
nixl_agent: None,
}
}
pub fn from_env() -> Result<Self, kvbm_config::ConfigError> {
Ok(Self::new(KvbmConfig::from_env()?))
}
pub fn from_json(json: &str) -> Result<Self, kvbm_config::ConfigError> {
Ok(Self::new(KvbmConfig::from_figment_with_json(json)?))
}
pub fn with_runtime(mut self, runtime: Arc<Runtime>) -> Self {
self.runtime = Some(RuntimeHandle::Owned(runtime));
self
}
pub fn with_runtime_handle(mut self, handle: Handle) -> Self {
self.runtime = Some(RuntimeHandle::Handle(handle));
self
}
pub fn with_messenger(mut self, messenger: Arc<Messenger>) -> Self {
self.messenger = Some(messenger);
self
}
pub fn with_nixl_agent(mut self, agent: NixlAgent) -> Self {
self.nixl_agent = Some(agent);
self
}
pub async fn build_leader(self) -> Result<super::KvbmRuntime> {
self.build_internal().await
}
pub async fn build_worker(self) -> Result<super::KvbmRuntime> {
self.build_internal().await
}
async fn build_internal(self) -> Result<super::KvbmRuntime> {
let runtime = match self.runtime {
Some(rt) => rt,
None => RuntimeHandle::Owned(Arc::new(self.config.tokio.build_runtime()?)),
};
let messenger = match self.messenger {
Some(m) => m,
None => self.config.messenger.build_messenger().await?,
};
let nixl_agent = match self.nixl_agent {
Some(agent) => Some(agent),
None => match &self.config.nixl {
Some(nixl_config) => {
let agent_name = format!("nixl-{}", messenger.instance_id());
let backend_config = nixl_config.clone().into();
Some(NixlAgent::from_nixl_backend_config(
&agent_name,
backend_config,
)?)
}
None => None, },
};
Ok(super::KvbmRuntime {
config: self.config,
runtime,
messenger,
nixl_agent,
})
}
}