isothrum-runner 1.0.1

The Isothrum runner: executes the run manifest Isoloom writes next to a target's files (lifecycle, state and locks, drivers, checks, events).
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

//! Picking the host ports a fresh run pins (ported from the launcher's `pick_pins`): a free
//! loopback port for each ephemeral publish, every socket held until all are picked so the OS
//! doesn't hand one out twice, and each pick reserved in the runner until the run's start ends, so
//! another run of this runner starting meanwhile can't take it either (R9). With the
//! `ISOLOOM_PUBLISH_FIXED` param set, a port takes its manifest's `preferred` number instead, once
//! per number and protocol.

use std::collections::BTreeSet;

use isoloom_manifest::{Native, Protocol};

use super::config::{self, ConfigPort};
use crate::executor::Cx;
use crate::net::HeldPort;

/// The param asking for each port's own `publish:` number (run-manifest-v0.md ยง6).
const PUBLISH_FIXED: &str = "ISOLOOM_PUBLISH_FIXED";

/// The pinned-ports override for `config` (`docker compose config --format json`), or `None`
/// when nothing is ephemeral or no port could be picked.
pub(super) fn pick(cx: &Cx<'_>, config: &str) -> Option<String> {
    let ports = config::ports(config)?;
    let fixed = cx.vars.params().contains_key(PUBLISH_FIXED);
    let mut held: Vec<HeldPort> = Vec::new();
    let mut taken: BTreeSet<(u16, Protocol)> = BTreeSet::new();
    let pins = config::pinned_ports(&ports, |p| {
        let mine = |n: &u16| cx.run.runner.reserve(cx.run.name(), *n, p.protocol);
        if fixed && let Some(preferred) = preferred(cx, p).filter(|n| !taken.contains(&(*n, p.protocol)) && mine(n)) {
            taken.insert((preferred, p.protocol));
            return Some(preferred);
        }
        let port = free_port(cx, p.protocol)?;
        taken.insert((port.port, p.protocol));
        let number = port.port;
        held.push(port);
        Some(number)
    });
    drop(held);
    pins
}

/// A free port over `protocol` that no other run of this runner holds, reserved for this run.
fn free_port(cx: &Cx<'_>, protocol: Protocol) -> Option<HeldPort> {
    let runner = cx.run.runner;
    // A port another run reserved is refused here, and held meanwhile so it isn't offered again.
    let mut refused = Vec::new();
    for _ in 0..64 {
        let port = runner.config.ports.pick_free(protocol)?;
        if runner.reserve(cx.run.name(), port.port, protocol) {
            return Some(port);
        }
        refused.push(port);
    }
    None
}

/// The manifest's `preferred` number for the machine port `p` publishes.
fn preferred(cx: &Cx<'_>, p: &ConfigPort) -> Option<u16> {
    let machine = cx.run.manifest().machines.iter().find(|m| match &m.native {
        Some(Native::Compose { service }) => service.as_str() == p.service,
        _ => m.name.as_str() == p.service,
    })?;
    let port = machine.ports.iter().find(|q| q.port.get() == p.target && q.kind.transport() == p.protocol)?;
    port.publish.as_ref()?.preferred.map(std::num::NonZeroU16::get)
}