mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(scheduler-gate): throttle background AI on battery / busy CPU (#1062)
This commit is contained in:
Steven Enamakel
Generated
+67
-1
@@ -1166,7 +1166,7 @@ dependencies = [
|
||||
"jni 0.21.1",
|
||||
"js-sys",
|
||||
"libc",
|
||||
"mach2",
|
||||
"mach2 0.4.3",
|
||||
"ndk",
|
||||
"ndk-context",
|
||||
"oboe",
|
||||
@@ -3280,6 +3280,12 @@ version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
|
||||
|
||||
[[package]]
|
||||
name = "lazycell"
|
||||
version = "1.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
|
||||
|
||||
[[package]]
|
||||
name = "leb128fmt"
|
||||
version = "0.1.0"
|
||||
@@ -3453,6 +3459,15 @@ dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mach2"
|
||||
version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6a1b95cd5421ec55b445b5ae102f5ea0e768de1f82bd3001e11f426c269c3aea"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "macroific"
|
||||
version = "2.0.0"
|
||||
@@ -4609,6 +4624,7 @@ dependencies = [
|
||||
"sha2 0.10.9",
|
||||
"shellexpand",
|
||||
"socketioxide",
|
||||
"starship-battery",
|
||||
"sysinfo",
|
||||
"tar",
|
||||
"tempfile",
|
||||
@@ -5014,6 +5030,19 @@ version = "0.2.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6"
|
||||
|
||||
[[package]]
|
||||
name = "plist"
|
||||
version = "1.8.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "740ebea15c5d1428f910cd1a5f52cebf8d25006245ed8ade92702f4943d91e07"
|
||||
dependencies = [
|
||||
"base64",
|
||||
"indexmap",
|
||||
"quick-xml",
|
||||
"serde",
|
||||
"time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "png"
|
||||
version = "0.18.1"
|
||||
@@ -5327,6 +5356,15 @@ version = "2.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3"
|
||||
|
||||
[[package]]
|
||||
name = "quick-xml"
|
||||
version = "0.38.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quinn"
|
||||
version = "0.11.9"
|
||||
@@ -6630,6 +6668,24 @@ version = "1.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
|
||||
|
||||
[[package]]
|
||||
name = "starship-battery"
|
||||
version = "0.10.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fd0efc2c44c92705be724265a0c758e3b7c120ea63817d2d684bab86fbeced9a"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"core-foundation 0.10.1",
|
||||
"lazycell",
|
||||
"libc",
|
||||
"mach2 0.5.0",
|
||||
"nix",
|
||||
"num-traits",
|
||||
"plist",
|
||||
"uom",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "stop-token"
|
||||
version = "0.7.0"
|
||||
@@ -7479,6 +7535,16 @@ version = "0.0.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6d49784317cd0d1ee7ec5c716dd598ec5b4483ea832a2dced265471cc0f690ae"
|
||||
|
||||
[[package]]
|
||||
name = "uom"
|
||||
version = "0.37.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cd5cfe7d84f6774726717f358a37f5bca8fca273bed4de40604ad129d1107b49"
|
||||
dependencies = [
|
||||
"num-traits",
|
||||
"typenum",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ureq"
|
||||
version = "3.3.0"
|
||||
|
||||
@@ -117,6 +117,11 @@ enigo = "0.3"
|
||||
arboard = "3"
|
||||
rdev = "0.5"
|
||||
fs2 = "0.4"
|
||||
# Cross-platform battery probe for the scheduler gate. Maintained fork of
|
||||
# the abandoned `battery` crate; same `use battery::*;` API surface. Used
|
||||
# only by `openhuman::scheduler_gate::signals` to decide when to throttle
|
||||
# background LLM work on laptops.
|
||||
starship-battery = "0.10"
|
||||
|
||||
matrix-sdk = { version = "0.16", optional = true, default-features = false, features = ["e2e-encryption", "rustls-tls", "markdown"] }
|
||||
fantoccini = { version = "0.22.0", optional = true, default-features = false, features = ["rustls-tls"] }
|
||||
|
||||
@@ -858,6 +858,11 @@ fn register_domain_subscribers(
|
||||
}
|
||||
crate::openhuman::composio::register_composio_trigger_subscriber();
|
||||
crate::openhuman::composio::start_periodic_sync();
|
||||
// Initialise the scheduler gate before any background AI workers
|
||||
// start so they observe a real policy on their first iteration
|
||||
// (otherwise they fall back to `Policy::Normal` and miss the
|
||||
// initial throttle decision on battery-powered hosts).
|
||||
crate::openhuman::scheduler_gate::init_global(&config);
|
||||
crate::openhuman::memory::tree::jobs::start(config.clone());
|
||||
|
||||
// Restart requests go through a subscriber so every trigger path shares
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
//!
|
||||
//! Design notes:
|
||||
//!
|
||||
//! * One global tick (60s) drives every provider — we don't spawn a
|
||||
//! * One global tick (5min) drives every provider — we don't spawn a
|
||||
//! task per connection, because the number of connections per user
|
||||
//! is small and a single tick keeps the bookkeeping trivial.
|
||||
//! * Per-connection state (last sync timestamp) lives in a
|
||||
@@ -36,7 +36,14 @@ use super::providers::{get_provider, ProviderContext, SyncReason};
|
||||
/// How often the scheduler wakes up to look for due syncs. Independent
|
||||
/// from per-provider `sync_interval_secs` — this just bounds how long
|
||||
/// past a provider's interval we might fire.
|
||||
const TICK_SECONDS: u64 = 60;
|
||||
///
|
||||
/// 5 min trades a little staleness for noticeably less foreground load:
|
||||
/// each tick triggers an HTTP fetch + DB write per due connection, and
|
||||
/// for users with several connected providers the old 60s cadence kept
|
||||
/// the laptop visibly busy. Per-provider `sync_interval_secs` still
|
||||
/// caps the *minimum* delay between actual syncs — this only loosens
|
||||
/// the upper bound.
|
||||
const TICK_SECONDS: u64 = 300;
|
||||
|
||||
/// Process-wide guard so the scheduler is only started once even
|
||||
/// when both `start_channels` and `bootstrap_skill_runtime` call into
|
||||
@@ -228,7 +235,7 @@ mod tests {
|
||||
fn tick_seconds_is_sane_default() {
|
||||
// Sanity check: don't accidentally ship a 1-second tick.
|
||||
assert!(TICK_SECONDS >= 30);
|
||||
assert!(TICK_SECONDS <= 300);
|
||||
assert!(TICK_SECONDS <= 600);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -32,10 +32,11 @@ pub use schema::{
|
||||
LocalAiConfig, MatrixConfig, MemoryConfig, ModelRouteConfig, MultimodalConfig,
|
||||
ObservabilityConfig, ProxyConfig, ProxyScope, ReflectionSource, ReliabilityConfig,
|
||||
ResourceLimitsConfig, RuntimeConfig, SandboxBackend, SandboxConfig, SchedulerConfig,
|
||||
ScreenIntelligenceConfig, SecretsConfig, SecurityConfig, SlackConfig, StorageConfig,
|
||||
StorageProviderConfig, StorageProviderSection, StreamMode, TelegramConfig, UpdateConfig,
|
||||
VoiceActivationMode, VoiceServerConfig, WebSearchConfig, WebhookConfig, DEFAULT_MODEL,
|
||||
MODEL_AGENTIC_V1, MODEL_CODING_V1, MODEL_REASONING_V1,
|
||||
SchedulerGateConfig, SchedulerGateMode, ScreenIntelligenceConfig, SecretsConfig,
|
||||
SecurityConfig, SlackConfig, StorageConfig, StorageProviderConfig, StorageProviderSection,
|
||||
StreamMode, TelegramConfig, UpdateConfig, VoiceActivationMode, VoiceServerConfig,
|
||||
WebSearchConfig, WebhookConfig, DEFAULT_MODEL, MODEL_AGENTIC_V1, MODEL_CODING_V1,
|
||||
MODEL_REASONING_V1,
|
||||
};
|
||||
pub use schema::{
|
||||
clear_active_user, default_root_openhuman_dir, pre_login_user_dir, read_active_user_id,
|
||||
|
||||
@@ -24,6 +24,7 @@ mod observability;
|
||||
mod proxy;
|
||||
mod routes;
|
||||
mod runtime;
|
||||
mod scheduler_gate;
|
||||
mod storage_memory;
|
||||
mod tools;
|
||||
mod update;
|
||||
@@ -53,6 +54,7 @@ pub use proxy::{
|
||||
};
|
||||
pub use routes::{EmbeddingRouteConfig, ModelRouteConfig};
|
||||
pub use runtime::{DockerRuntimeConfig, ReliabilityConfig, RuntimeConfig, SchedulerConfig};
|
||||
pub use scheduler_gate::{SchedulerGateConfig, SchedulerGateMode};
|
||||
pub use storage_memory::{
|
||||
MemoryConfig, MemoryTreeConfig, StorageConfig, StorageProviderConfig, StorageProviderSection,
|
||||
};
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
//! Scheduler-gate configuration — controls when background AI work runs.
|
||||
//!
|
||||
//! Consumed by [`crate::openhuman::scheduler_gate`].
|
||||
|
||||
use schemars::JsonSchema;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum SchedulerGateMode {
|
||||
/// Decide based on power + CPU + deployment-mode signals.
|
||||
Auto,
|
||||
/// Always run background AI flat-out (server / power-user setting).
|
||||
AlwaysOn,
|
||||
/// Never run background AI. User can still trigger work explicitly.
|
||||
Off,
|
||||
}
|
||||
|
||||
impl SchedulerGateMode {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Auto => "auto",
|
||||
Self::AlwaysOn => "always_on",
|
||||
Self::Off => "off",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for SchedulerGateMode {
|
||||
fn default() -> Self {
|
||||
Self::Auto
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
|
||||
pub struct SchedulerGateConfig {
|
||||
/// Top-level mode — `auto` (default), `always_on`, or `off`.
|
||||
#[serde(default)]
|
||||
pub mode: SchedulerGateMode,
|
||||
|
||||
/// Battery charge floor in `auto` mode, 0.0..=1.0. Below this and not on
|
||||
/// AC, the gate throttles. Default: 0.80.
|
||||
#[serde(default = "default_battery_floor")]
|
||||
pub battery_floor: f32,
|
||||
|
||||
/// CPU busy threshold (recent global usage, 0..100). Above this, the gate
|
||||
/// throttles even when plugged in. Default: 70.0 (i.e. <30% headroom).
|
||||
#[serde(default = "default_cpu_busy_threshold")]
|
||||
pub cpu_busy_threshold_pct: f32,
|
||||
|
||||
/// In `Throttled` mode, sleep this many ms before each LLM-bound job to
|
||||
/// serialise workers and let the host catch up. Default: 30_000 (30s).
|
||||
#[serde(default = "default_throttled_backoff_ms")]
|
||||
pub throttled_backoff_ms: u64,
|
||||
|
||||
/// In `Paused` mode, re-check the policy every this many ms so workers
|
||||
/// resume promptly when the user toggles the gate back on. Default:
|
||||
/// 60_000 (60s).
|
||||
#[serde(default = "default_paused_poll_ms")]
|
||||
pub paused_poll_ms: u64,
|
||||
}
|
||||
|
||||
fn default_battery_floor() -> f32 {
|
||||
0.80
|
||||
}
|
||||
fn default_cpu_busy_threshold() -> f32 {
|
||||
70.0
|
||||
}
|
||||
fn default_throttled_backoff_ms() -> u64 {
|
||||
30_000
|
||||
}
|
||||
fn default_paused_poll_ms() -> u64 {
|
||||
60_000
|
||||
}
|
||||
|
||||
impl Default for SchedulerGateConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
mode: SchedulerGateMode::default(),
|
||||
battery_floor: default_battery_floor(),
|
||||
cpu_busy_threshold_pct: default_cpu_busy_threshold(),
|
||||
throttled_backoff_ms: default_throttled_backoff_ms(),
|
||||
paused_poll_ms: default_paused_poll_ms(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -54,6 +54,13 @@ pub struct Config {
|
||||
#[serde(default)]
|
||||
pub scheduler: SchedulerConfig,
|
||||
|
||||
/// Background-AI scheduler gate — throttles memory-tree digests,
|
||||
/// embeddings, and other LLM-bound background work based on power
|
||||
/// state, CPU pressure, and deployment mode. See
|
||||
/// [`crate::openhuman::scheduler_gate`].
|
||||
#[serde(default)]
|
||||
pub scheduler_gate: SchedulerGateConfig,
|
||||
|
||||
#[serde(default)]
|
||||
pub agent: AgentConfig,
|
||||
|
||||
@@ -251,6 +258,7 @@ impl Default for Config {
|
||||
autocomplete: AutocompleteConfig::default(),
|
||||
reliability: ReliabilityConfig::default(),
|
||||
scheduler: SchedulerConfig::default(),
|
||||
scheduler_gate: SchedulerGateConfig::default(),
|
||||
agent: AgentConfig::default(),
|
||||
context: ContextConfig::default(),
|
||||
model_routes: Vec::new(),
|
||||
|
||||
@@ -73,6 +73,19 @@ pub async fn run_once(config: &Config) -> Result<bool> {
|
||||
}
|
||||
|
||||
async fn run_once_with_semaphore(config: &Config, llm_slots: Arc<Semaphore>) -> Result<bool> {
|
||||
// Cooperative throttle BEFORE `claim_next()`. Holding the DB claim
|
||||
// across an awaited `wait_for_capacity()` would let `Paused` mode
|
||||
// sit on the row past `DEFAULT_LOCK_DURATION_MS`, after which
|
||||
// `recover_stale_locks()` would requeue it for another worker to
|
||||
// pick up — duplicating side effects. Throttling here means
|
||||
// non-LLM jobs (AppendBuffer/FlushStale) also experience the same
|
||||
// gate delay, but that's fine: in Throttled mode the host is
|
||||
// already overloaded and a 30s breather between any DB-write batch
|
||||
// is welcome; in Paused mode the user has explicitly asked us to
|
||||
// stand down. Returns immediately in Aggressive/Normal so plugged-in
|
||||
// desktops with headroom pay zero cost.
|
||||
crate::openhuman::scheduler_gate::wait_for_capacity().await;
|
||||
|
||||
let Some(job) = claim_next(config, DEFAULT_LOCK_DURATION_MS)? else {
|
||||
return Ok(false);
|
||||
};
|
||||
|
||||
@@ -47,6 +47,7 @@ pub mod providers;
|
||||
pub mod redirect_links;
|
||||
pub mod referral;
|
||||
pub mod routing;
|
||||
pub mod scheduler_gate;
|
||||
pub mod screen_intelligence;
|
||||
pub mod security;
|
||||
pub mod service;
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
//! Process-wide singleton: cached policy + cooperative throttling.
|
||||
//!
|
||||
//! One sampler task refreshes [`Signals`] every 30s and recomputes the
|
||||
//! [`Policy`]. Workers call [`current_policy`] for cheap reads or
|
||||
//! [`wait_for_capacity`] to cooperatively block until the host is ready.
|
||||
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use std::time::Duration;
|
||||
|
||||
use parking_lot::RwLock;
|
||||
|
||||
use crate::openhuman::config::{Config, SchedulerGateConfig};
|
||||
use crate::openhuman::scheduler_gate::policy::{decide, Policy};
|
||||
use crate::openhuman::scheduler_gate::signals::Signals;
|
||||
|
||||
struct State {
|
||||
cfg: SchedulerGateConfig,
|
||||
signals: Signals,
|
||||
policy: Policy,
|
||||
}
|
||||
|
||||
static STATE: OnceLock<Arc<RwLock<State>>> = OnceLock::new();
|
||||
static STARTED: std::sync::Once = std::sync::Once::new();
|
||||
|
||||
const SAMPLE_INTERVAL: Duration = Duration::from_secs(30);
|
||||
|
||||
/// Initialise the gate and spawn the background sampler.
|
||||
///
|
||||
/// Idempotent — repeat calls during bootstrap are no-ops. Subsequent config
|
||||
/// reloads should call [`update_config`] instead.
|
||||
pub fn init_global(config: &Config) {
|
||||
let cfg = config.scheduler_gate.clone();
|
||||
STARTED.call_once(|| {
|
||||
let signals = Signals::sample();
|
||||
let policy = decide(&signals, &cfg);
|
||||
log::info!(
|
||||
"[scheduler_gate] startup policy={} mode={} on_ac={} charge={:?} cpu={:.1}% server={}",
|
||||
policy.as_str(),
|
||||
cfg.mode.as_str(),
|
||||
signals.on_ac_power,
|
||||
signals.battery_charge,
|
||||
signals.cpu_usage_pct,
|
||||
signals.server_mode,
|
||||
);
|
||||
let state = Arc::new(RwLock::new(State {
|
||||
cfg,
|
||||
signals,
|
||||
policy,
|
||||
}));
|
||||
let _ = STATE.set(state.clone());
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
tokio::time::sleep(SAMPLE_INTERVAL).await;
|
||||
// Sampling does a brief blocking sleep + sysinfo refresh —
|
||||
// push it off the async runtime.
|
||||
let signals = match tokio::task::spawn_blocking(Signals::sample).await {
|
||||
Ok(s) => s,
|
||||
Err(err) => {
|
||||
log::warn!("[scheduler_gate] sampler join error: {err:#}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let mut guard = state.write();
|
||||
let next = decide(&signals, &guard.cfg);
|
||||
if next != guard.policy {
|
||||
log::info!(
|
||||
"[scheduler_gate] policy {} -> {} (on_ac={} charge={:?} cpu={:.1}% server={})",
|
||||
guard.policy.as_str(),
|
||||
next.as_str(),
|
||||
signals.on_ac_power,
|
||||
signals.battery_charge,
|
||||
signals.cpu_usage_pct,
|
||||
signals.server_mode,
|
||||
);
|
||||
}
|
||||
guard.signals = signals;
|
||||
guard.policy = next;
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/// Update the gate's view of user config (e.g. after a settings change).
|
||||
pub fn update_config(cfg: SchedulerGateConfig) {
|
||||
if let Some(state) = STATE.get() {
|
||||
let mut guard = state.write();
|
||||
guard.cfg = cfg;
|
||||
guard.policy = decide(&guard.signals, &guard.cfg);
|
||||
}
|
||||
}
|
||||
|
||||
/// Current policy. Defaults to [`Policy::Normal`] before [`init_global`] runs
|
||||
/// (e.g. in unit tests) so callers don't deadlock waiting on a sampler that
|
||||
/// will never start.
|
||||
pub fn current_policy() -> Policy {
|
||||
STATE
|
||||
.get()
|
||||
.map(|s| s.read().policy)
|
||||
.unwrap_or(Policy::Normal)
|
||||
}
|
||||
|
||||
/// Most recent sampled signals, or a neutral default if the sampler hasn't run.
|
||||
pub fn current_signals() -> Signals {
|
||||
STATE.get().map(|s| s.read().signals).unwrap_or(Signals {
|
||||
on_ac_power: true,
|
||||
battery_charge: None,
|
||||
cpu_usage_pct: 0.0,
|
||||
server_mode: false,
|
||||
})
|
||||
}
|
||||
|
||||
/// Cooperatively block a worker until the host is ready for LLM-bound work.
|
||||
///
|
||||
/// * **Aggressive / Normal** — returns immediately.
|
||||
/// * **Throttled** — sleeps `throttled_backoff_ms` so concurrent workers
|
||||
/// serialise themselves and the host catches its breath between jobs.
|
||||
/// * **Paused** — polls every `paused_poll_ms` until the policy changes.
|
||||
///
|
||||
/// Designed so existing semaphore-bounded worker pools can keep their pool
|
||||
/// size and just gain a per-job throttle in front of the existing
|
||||
/// `semaphore.acquire()` call.
|
||||
pub async fn wait_for_capacity() {
|
||||
loop {
|
||||
let (policy, throttled_ms, paused_ms) = match STATE.get() {
|
||||
Some(state) => {
|
||||
let g = state.read();
|
||||
(g.policy, g.cfg.throttled_backoff_ms, g.cfg.paused_poll_ms)
|
||||
}
|
||||
None => return,
|
||||
};
|
||||
match policy {
|
||||
Policy::Aggressive | Policy::Normal => return,
|
||||
Policy::Throttled => {
|
||||
tokio::time::sleep(Duration::from_millis(throttled_ms)).await;
|
||||
return;
|
||||
}
|
||||
Policy::Paused => {
|
||||
tokio::time::sleep(Duration::from_millis(paused_ms)).await;
|
||||
// re-evaluate; user may have toggled the gate back on.
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
//! Scheduler gate — gates background AI work on host conditions.
|
||||
//!
|
||||
//! Background AI tasks (memory-tree digests, embeddings, summarisation) used
|
||||
//! to run flat-out and made the host visibly lag, especially on battery.
|
||||
//! This module exposes a single decision point — [`current_policy`] — that
|
||||
//! background workers consult before spending CPU/GPU on LLM-bound work.
|
||||
//!
|
||||
//! Signals (refreshed every 30s in a background sampler):
|
||||
//! * power state — on AC, or battery >= 80%
|
||||
//! * CPU usage — recent global usage; <70% means "idle enough"
|
||||
//! * deployment mode — server/container hosts run flat-out
|
||||
//!
|
||||
//! Resulting [`Policy`]:
|
||||
//! * [`Policy::Aggressive`] — server-mode; bypass throttles entirely
|
||||
//! * [`Policy::Normal`] — desktop with headroom; run as scheduled
|
||||
//! * [`Policy::Throttled`] — busy or on battery; serialise + slow down
|
||||
//! * [`Policy::Paused`] — user opted out; defer indefinitely
|
||||
//!
|
||||
//! Cooperative throttling: callers `await gate::wait_for_capacity()` before
|
||||
//! each unit of LLM-bound work. The future resolves immediately in
|
||||
//! Aggressive/Normal, sleeps in Throttled, and re-polls in Paused so the
|
||||
//! caller resumes the moment the user toggles the gate back on.
|
||||
|
||||
pub mod gate;
|
||||
pub mod policy;
|
||||
pub mod signals;
|
||||
|
||||
pub use gate::{current_policy, current_signals, init_global, wait_for_capacity};
|
||||
pub use policy::Policy;
|
||||
pub use signals::Signals;
|
||||
@@ -0,0 +1,188 @@
|
||||
//! Decision logic — turn raw [`Signals`] + user config into a [`Policy`].
|
||||
|
||||
use crate::openhuman::config::SchedulerGateConfig;
|
||||
use crate::openhuman::scheduler_gate::signals::Signals;
|
||||
|
||||
/// Background-AI scheduling tier. See module docs in `mod.rs` for semantics.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum Policy {
|
||||
Aggressive,
|
||||
Normal,
|
||||
Throttled,
|
||||
Paused,
|
||||
}
|
||||
|
||||
impl Policy {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Aggressive => "aggressive",
|
||||
Self::Normal => "normal",
|
||||
Self::Throttled => "throttled",
|
||||
Self::Paused => "paused",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Compute the current [`Policy`] from sampled signals + user config.
|
||||
///
|
||||
/// Order of evaluation matters — explicit user overrides win first, then
|
||||
/// deployment mode, then dynamic host signals.
|
||||
pub fn decide(signals: &Signals, cfg: &SchedulerGateConfig) -> Policy {
|
||||
use crate::openhuman::config::SchedulerGateMode;
|
||||
|
||||
match cfg.mode {
|
||||
SchedulerGateMode::Off => return Policy::Paused,
|
||||
SchedulerGateMode::AlwaysOn => return Policy::Aggressive,
|
||||
SchedulerGateMode::Auto => {}
|
||||
}
|
||||
|
||||
if signals.server_mode {
|
||||
return Policy::Aggressive;
|
||||
}
|
||||
|
||||
// Clamp config-supplied thresholds so a malformed config.toml (e.g.
|
||||
// `battery_floor = 1.5` or a negative cpu threshold) can't silently
|
||||
// disable / force throttling — the field is `f32` and serde won't
|
||||
// reject out-of-domain values for us.
|
||||
let battery_floor = cfg.battery_floor.clamp(0.0, 1.0);
|
||||
let cpu_threshold = cfg.cpu_busy_threshold_pct.clamp(0.0, 100.0);
|
||||
|
||||
let battery_ok = signals.on_ac_power
|
||||
|| signals
|
||||
.battery_charge
|
||||
.map(|c| c >= battery_floor)
|
||||
.unwrap_or(true); // no battery present == treat as plugged in
|
||||
|
||||
let cpu_ok = signals.cpu_usage_pct <= cpu_threshold;
|
||||
|
||||
if battery_ok && cpu_ok {
|
||||
Policy::Normal
|
||||
} else {
|
||||
Policy::Throttled
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::openhuman::config::{SchedulerGateConfig, SchedulerGateMode};
|
||||
|
||||
fn cfg(mode: SchedulerGateMode) -> SchedulerGateConfig {
|
||||
SchedulerGateConfig {
|
||||
mode,
|
||||
battery_floor: 0.8,
|
||||
cpu_busy_threshold_pct: 70.0,
|
||||
throttled_backoff_ms: 30_000,
|
||||
paused_poll_ms: 60_000,
|
||||
}
|
||||
}
|
||||
|
||||
fn signals(on_ac: bool, charge: Option<f32>, cpu: f32, server: bool) -> Signals {
|
||||
Signals {
|
||||
on_ac_power: on_ac,
|
||||
battery_charge: charge,
|
||||
cpu_usage_pct: cpu,
|
||||
server_mode: server,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn off_mode_pauses() {
|
||||
let p = decide(
|
||||
&signals(true, None, 5.0, true),
|
||||
&cfg(SchedulerGateMode::Off),
|
||||
);
|
||||
assert_eq!(p, Policy::Paused);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn always_on_overrides_signals() {
|
||||
// discharging laptop at 10% with 99% CPU — still Aggressive.
|
||||
let p = decide(
|
||||
&signals(false, Some(0.10), 99.0, false),
|
||||
&cfg(SchedulerGateMode::AlwaysOn),
|
||||
);
|
||||
assert_eq!(p, Policy::Aggressive);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn server_mode_is_aggressive() {
|
||||
let p = decide(
|
||||
&signals(false, None, 50.0, true),
|
||||
&cfg(SchedulerGateMode::Auto),
|
||||
);
|
||||
assert_eq!(p, Policy::Aggressive);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plugged_in_idle_is_normal() {
|
||||
let p = decide(
|
||||
&signals(true, Some(0.45), 20.0, false),
|
||||
&cfg(SchedulerGateMode::Auto),
|
||||
);
|
||||
assert_eq!(p, Policy::Normal);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn battery_above_floor_is_normal() {
|
||||
let p = decide(
|
||||
&signals(false, Some(0.85), 20.0, false),
|
||||
&cfg(SchedulerGateMode::Auto),
|
||||
);
|
||||
assert_eq!(p, Policy::Normal);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn battery_below_floor_throttles() {
|
||||
let p = decide(
|
||||
&signals(false, Some(0.30), 20.0, false),
|
||||
&cfg(SchedulerGateMode::Auto),
|
||||
);
|
||||
assert_eq!(p, Policy::Throttled);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn busy_cpu_throttles_even_when_plugged_in() {
|
||||
let p = decide(
|
||||
&signals(true, Some(0.95), 90.0, false),
|
||||
&cfg(SchedulerGateMode::Auto),
|
||||
);
|
||||
assert_eq!(p, Policy::Throttled);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn out_of_range_battery_floor_is_clamped() {
|
||||
// 1.5 clamped to 1.0 — with charge < 1.0 on battery, must throttle.
|
||||
let mut c = cfg(SchedulerGateMode::Auto);
|
||||
c.battery_floor = 1.5;
|
||||
let p = decide(&signals(false, Some(0.99), 10.0, false), &c);
|
||||
assert_eq!(p, Policy::Throttled);
|
||||
// -1.0 clamped to 0.0 — any non-zero charge passes the floor.
|
||||
c.battery_floor = -1.0;
|
||||
let p = decide(&signals(false, Some(0.05), 10.0, false), &c);
|
||||
assert_eq!(p, Policy::Normal);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn out_of_range_cpu_threshold_is_clamped() {
|
||||
// 200.0 clamped to 100.0 — nothing above it, never throttles on CPU.
|
||||
let mut c = cfg(SchedulerGateMode::Auto);
|
||||
c.cpu_busy_threshold_pct = 200.0;
|
||||
let p = decide(&signals(true, None, 99.0, false), &c);
|
||||
assert_eq!(p, Policy::Normal);
|
||||
// -10.0 clamped to 0.0 — any positive CPU usage throttles.
|
||||
c.cpu_busy_threshold_pct = -10.0;
|
||||
let p = decide(&signals(true, None, 5.0, false), &c);
|
||||
assert_eq!(p, Policy::Throttled);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_battery_treated_as_plugged_in() {
|
||||
// Desktop / server with no battery sensor — treat as AC.
|
||||
let p = decide(
|
||||
&signals(false, None, 20.0, false),
|
||||
&cfg(SchedulerGateMode::Auto),
|
||||
);
|
||||
assert_eq!(p, Policy::Normal);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
//! Host signals: power state, CPU pressure, deployment mode.
|
||||
//!
|
||||
//! Sampled on a 30s cadence by [`crate::openhuman::scheduler_gate::gate`]; this
|
||||
//! file just captures one snapshot at a time.
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
|
||||
use once_cell::sync::Lazy;
|
||||
use sysinfo::System;
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct Signals {
|
||||
pub on_ac_power: bool,
|
||||
/// 0.0..=1.0, or `None` when no battery sensor is present (most servers).
|
||||
pub battery_charge: Option<f32>,
|
||||
/// Recent global CPU usage, 0..100.
|
||||
pub cpu_usage_pct: f32,
|
||||
pub server_mode: bool,
|
||||
}
|
||||
|
||||
impl Signals {
|
||||
/// Sample once. Cheap (~ms-scale) — safe to call from a 30s background task.
|
||||
pub fn sample() -> Self {
|
||||
let (on_ac, charge) = sample_power();
|
||||
let cpu_usage_pct = sample_cpu();
|
||||
let server_mode = detect_server_mode(charge.is_none());
|
||||
Self {
|
||||
on_ac_power: on_ac,
|
||||
battery_charge: charge,
|
||||
cpu_usage_pct,
|
||||
server_mode,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- power ---------------------------------------------------------------
|
||||
|
||||
fn sample_power() -> (bool, Option<f32>) {
|
||||
// Env overrides win — useful for CI, container hosts that misreport,
|
||||
// and manual debugging of the throttle path on a desktop. Only
|
||||
// explicit truthy/falsy tokens count: garbage values yield None so
|
||||
// the real probe still gets to answer (vs. silently coercing to
|
||||
// "on battery" and triggering throttling on every misconfigured host).
|
||||
let env_on_ac = std::env::var("OPENHUMAN_ON_AC_POWER").ok().and_then(|v| {
|
||||
match v.to_ascii_lowercase().as_str() {
|
||||
"1" | "true" | "yes" => Some(true),
|
||||
"0" | "false" | "no" => Some(false),
|
||||
_ => None,
|
||||
}
|
||||
});
|
||||
let env_charge = std::env::var("OPENHUMAN_BATTERY_CHARGE")
|
||||
.ok()
|
||||
.and_then(|v| v.parse::<f32>().ok())
|
||||
.map(|v| v.clamp(0.0, 1.0));
|
||||
if let (Some(ac), Some(c)) = (env_on_ac, env_charge) {
|
||||
return (ac, Some(c));
|
||||
}
|
||||
|
||||
match probe_battery() {
|
||||
Ok(probe) => (
|
||||
env_on_ac.unwrap_or(probe.on_ac),
|
||||
env_charge.or(probe.charge),
|
||||
),
|
||||
Err(err) => {
|
||||
// Probe failure on Linux often just means no /sys/class/power_supply
|
||||
// entries (server, container) — treat as "plugged in, no battery"
|
||||
// which yields Normal/Aggressive, not Throttled. Log once at debug
|
||||
// because this fires every 30s on the sampler tick.
|
||||
log::debug!("[scheduler_gate] battery probe failed: {err:#}");
|
||||
(env_on_ac.unwrap_or(true), env_charge)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct BatteryProbe {
|
||||
on_ac: bool,
|
||||
charge: Option<f32>,
|
||||
}
|
||||
|
||||
fn probe_battery() -> Result<BatteryProbe, starship_battery::Error> {
|
||||
let manager = starship_battery::Manager::new()?;
|
||||
let mut any = false;
|
||||
let mut on_ac = true; // if all batteries report Charging/Full, we're on AC.
|
||||
let mut total: f32 = 0.0;
|
||||
let mut count: f32 = 0.0;
|
||||
for maybe in manager.batteries()? {
|
||||
let battery = maybe?;
|
||||
any = true;
|
||||
// Discharging is the only state that conclusively means "on battery".
|
||||
// Unknown / Empty / Full / Charging all imply the AC adapter is
|
||||
// present (or at minimum that the OS isn't draining the pack).
|
||||
if matches!(battery.state(), starship_battery::State::Discharging) {
|
||||
on_ac = false;
|
||||
}
|
||||
total += battery.state_of_charge().value;
|
||||
count += 1.0;
|
||||
}
|
||||
let charge = if any && count > 0.0 {
|
||||
Some((total / count).clamp(0.0, 1.0))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
Ok(BatteryProbe { on_ac, charge })
|
||||
}
|
||||
|
||||
// ---- cpu -----------------------------------------------------------------
|
||||
|
||||
static CPU_SYS: Lazy<Mutex<System>> = Lazy::new(|| Mutex::new(System::new()));
|
||||
|
||||
fn sample_cpu() -> f32 {
|
||||
// Two refreshes spaced ~MINIMUM_CPU_UPDATE_INTERVAL apart give sysinfo
|
||||
// a real delta to compute usage from. The interval is small enough to
|
||||
// run on the 30s sampler tick without noticeable cost.
|
||||
let mut sys = match CPU_SYS.lock() {
|
||||
Ok(g) => g,
|
||||
Err(p) => p.into_inner(),
|
||||
};
|
||||
sys.refresh_cpu_usage();
|
||||
std::thread::sleep(Duration::from_millis(
|
||||
sysinfo::MINIMUM_CPU_UPDATE_INTERVAL.as_millis() as u64 + 50,
|
||||
));
|
||||
sys.refresh_cpu_usage();
|
||||
sys.global_cpu_usage()
|
||||
}
|
||||
|
||||
// ---- deployment mode -----------------------------------------------------
|
||||
|
||||
fn detect_server_mode(no_battery: bool) -> bool {
|
||||
if let Ok(v) = std::env::var("OPENHUMAN_DEPLOYMENT") {
|
||||
if v.eq_ignore_ascii_case("server") {
|
||||
return true;
|
||||
}
|
||||
if matches!(v.to_ascii_lowercase().as_str(), "desktop" | "laptop") {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
if std::env::var("KUBERNETES_SERVICE_HOST").is_ok() {
|
||||
return true;
|
||||
}
|
||||
if Path::new("/.dockerenv").exists() {
|
||||
return true;
|
||||
}
|
||||
// Heuristic of last resort: a Linux box with no battery and no display
|
||||
// server set is almost certainly a server. We *don't* infer server-mode
|
||||
// from "no battery" alone — desktops have no battery either.
|
||||
if cfg!(target_os = "linux")
|
||||
&& no_battery
|
||||
&& std::env::var("DISPLAY").is_err()
|
||||
&& std::env::var("WAYLAND_DISPLAY").is_err()
|
||||
{
|
||||
return true;
|
||||
}
|
||||
false
|
||||
}
|
||||
Reference in New Issue
Block a user