OXYGENHUB
# f3b4fdeb2bc2 oxygenflow / crates / flow / src / home.rs
4.1 KiB · Rust · last changed The runner runs up to max_jobs jobs at once within a memory and vCPU budget, asking the hub only for jobs that fit (a lone job may use the whole budget); preparing workspaces stays one at a time 1 month ago source blame history raw
1 
//! `$FLOW_HOME` (default `~/.local/share/flow`): the pinned parts of each
2 
//! isolation tier, rootfs images by hash, scratch space and the runner's
3 
//! configuration, all described by `flow.toml`.
4 
//!
5 
//! ```toml
6 
//! [microvm]
7 
//! firecracker = { path = "bin/firecracker", blake3 = "…" }
8 
//! kernel = { path = "kernel/vmlinux", blake3 = "…" }
9 
//! init = { path = "bin/flow-init", blake3 = "…" }
10 
//!
11 
//! [runner]                       # only for `flow runner`
12 
//! hub = "https://hub.example:7343"
13 
//! token = "…"                    # a runner token from the hub's [flow] runner_tokens
14 
//! ca = "/path/to/hub-ca.pem"     # when the hub's certificate is self-signed
15 
//! name = "earth-1"
16 
//! ```
17
18 
use anyhow::{bail, Context, Result};
19 
use serde::Deserialize;
20 
use std::collections::HashMap;
21 
use std::path::{Path, PathBuf};
22 
use std::sync::Mutex;
23
24 
#[derive(Clone, Debug, Deserialize)]
25 
pub struct Pinned {
26 
    pub path: PathBuf,
27 
    pub blake3: String,
28 
}
29
30 
#[derive(Clone, Debug, Deserialize)]
31 
pub struct MicrovmConfig {
32 
    pub firecracker: Pinned,
33 
    pub kernel: Pinned,
34 
    pub init: Pinned,
35 
}
36
37 
#[derive(Clone, Debug, Deserialize)]
38 
pub struct RunnerConfig {
39 
    pub hub: String,
40 
    pub token: String,
41 
    pub ca: Option<PathBuf>,
42 
    pub name: Option<String>,
43 
    /// Jobs run at once (default 1).
44 
    #[serde(default)]
45 
    pub max_jobs: Option<usize>,
46 
    /// Memory (MiB) and vCPUs the jobs running at once may use together.
47 
    #[serde(default)]
48 
    pub memory_mib: Option<u32>,
49 
    #[serde(default)]
50 
    pub cpus: Option<u32>,
51 
}
52
53 
#[derive(Clone, Debug, Default, Deserialize)]
54 
pub struct Config {
55 
    pub microvm: Option<MicrovmConfig>,
56 
    pub runner: Option<RunnerConfig>,
57 
}
58
59 
pub struct Home {
60 
    pub root: PathBuf,
61 
    pub config: Config,
62 
    /// Files already verified, by path, with the (len, mtime) they had then.
63 
    verified: Mutex<HashMap<PathBuf, (u64, std::time::SystemTime)>>,
64 
}
65
66 
impl Home {
67 
    pub fn locate(explicit: Option<&Path>) -> Result<Home> {
68 
        let root = match explicit {
69 
            Some(p) => p.to_path_buf(),
70 
            None => match std::env::var_os("FLOW_HOME") {
71 
                Some(p) => PathBuf::from(p),
72 
                None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),
73 
            },
74 
        };
75 
        let file = root.join("flow.toml");
76 
        let config = if file.is_file() {
77 
            let text = std::fs::read_to_string(&file).with_context(|| format!("read {}", file.display()))?;
78 
            toml::from_str(&text).with_context(|| format!("parse {}", file.display()))?
79 
        } else {
80 
            Config::default()
81 
        };
82 
        for d in ["cas", "work", "repos"] {
83 
            std::fs::create_dir_all(root.join(d)).with_context(|| format!("create {}", root.join(d).display()))?;
84 
        }
85 
        Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })
86 
    }
87
88 
    pub fn resolve(&self, p: &Path) -> PathBuf {
89 
        if p.is_absolute() {
90 
            p.to_path_buf()
91 
        } else {
92 
            self.root.join(p)
93 
        }
94 
    }
95
96 
    /// Where a rootfs image with this hash lives.
97 
    pub fn rootfs(&self, hash: &str) -> PathBuf {
98 
        self.root.join("cas").join(format!("{hash}.ext4"))
99 
    }
100
101 
    /// A pinned file, checked against its BLAKE3 before use (again only
102 
    /// when its size or modification time changed since).
103 
    pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {
104 
        let meta = std::fs::metadata(path).with_context(|| format!("{} is missing", path.display()))?;
105 
        let stamp = (meta.len(), meta.modified()?);
106 
        if self.verified.lock().unwrap().get(path) == Some(&stamp) {
107 
            return Ok(path.to_path_buf());
108 
        }
109 
        let got = hash_file(path)?;
110 
        if got != blake3_hex {
111 
            bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());
112 
        }
113 
        self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);
114 
        Ok(path.to_path_buf())
115 
    }
116 
}
117
118 
pub fn hash_file(path: &Path) -> Result<String> {
119 
    let mut h = blake3::Hasher::new();
120 
    h.update_reader(std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?)?;
121 
    Ok(h.finalize().to_hex().to_string())
122 
}