daemon: sched/ — the concurrency governor + schedule-window evaluation (build step 4)
The scheduling brain, built against CORE's headers (vdm/engine.hpp,
vdm/segment/budget.hpp) — signatures only; the Engine bodies land in
CORE stage 8 and the Scheduler that wires governor <-> store <-> engine
<-> timer comes after that (daemon/docs/deferrals.md D4).
- sched/governor — a pure decision function. In: a snapshot of every
task's coarse RunState and every queue's state (schedule windows
pre-resolved). Out: {to_start, to_resume, to_pause, pause_reasons,
priority_order}. Enforces, all in TASK units per ADR 0011 §1:
connection.maxConcurrentDownloads; the min(that, maxActiveSegments)
clamp (§2); Queue.maxConcurrent; the per-host task cap (§4); and a
stopped queue / closed window runs nothing. Never touches a task
paused for `user` or CORE's `auto` (ADR 0013 §3) — only Schedule /
QueueStopped / AdmissionReconcile are auto-resumable. Deterministic:
main-list before queued-in-queue, then queue order, then FIFO, then
task_id.
- sched/schedule_window — window_open(Schedule, local tm): disabled =>
always open; `once` => date + time match; `periodic` => weekday in
daysOfWeek (empty = every day) + time in [start, stop); null start =>
midnight, null stop => end of day, stop < start => overnight window.
Pure; re-evaluated every tick, no cached instants.
- veloxd_sched static lib; veloxd links it (nothing calls it yet).
Tests (ASan+UBSan and TSan clean): veloxd.sched_window (10 window
cases incl. overnight, once, null bounds), veloxd.sched_governor
(global/clamp/per-queue/per-host caps, stop vs window pause reasons,
resume-only-governor-reasons, auth-pause untouched, admission
reconcile, determinism under shuffled input). 32 daemon/cli tests
green; full tree green.
Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Upd9WhG9oppieig5nRDLig
This commit is contained in:
Claude Sonnet 5
+13
-1
@@ -44,6 +44,18 @@ target_compile_features(veloxd_store PUBLIC cxx_std_23)
|
||||
target_compile_options(veloxd_store PRIVATE -Wall -Wextra -Wpedantic -Werror)
|
||||
target_link_libraries(veloxd_store PUBLIC SQLite::SQLite3 PRIVATE OpenSSL::Crypto)
|
||||
|
||||
# --- veloxd_sched — the concurrency governor (pure; no engine link yet, see
|
||||
# daemon/docs/deferrals.md D4) ---------------------------------------------------
|
||||
add_library(veloxd_sched STATIC
|
||||
src/sched/schedule_window.cpp
|
||||
src/sched/governor.cpp
|
||||
)
|
||||
add_library(velox::daemon_sched ALIAS veloxd_sched)
|
||||
target_include_directories(veloxd_sched PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/src)
|
||||
target_compile_features(veloxd_sched PUBLIC cxx_std_23)
|
||||
target_compile_options(veloxd_sched PRIVATE -Wall -Wextra -Wpedantic -Werror)
|
||||
target_link_libraries(veloxd_sched PUBLIC velox::proto nlohmann_json::nlohmann_json)
|
||||
|
||||
# --- veloxd_rpc — the RPC transports + dispatcher ------------------------------------
|
||||
add_library(veloxd_rpc STATIC
|
||||
src/rpc/runtime_dir.cpp
|
||||
@@ -68,7 +80,7 @@ target_link_libraries(veloxd_rpc
|
||||
add_executable(veloxd src/main.cpp)
|
||||
target_compile_features(veloxd PRIVATE cxx_std_23)
|
||||
target_compile_options(veloxd PRIVATE -Wall -Wextra -Wpedantic -Werror)
|
||||
target_link_libraries(veloxd PRIVATE veloxd_rpc)
|
||||
target_link_libraries(veloxd PRIVATE veloxd_rpc veloxd_sched)
|
||||
|
||||
if(VELOX_BUILD_TESTS AND EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/tests/CMakeLists.txt)
|
||||
add_subdirectory(tests)
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
# DAEMON — deferred work, tracked
|
||||
|
||||
Things that are deliberately incomplete in `daemon/` right now, with why and when they
|
||||
close. Kept here (not buried in commit messages) so the next pass can see them at a glance.
|
||||
|
||||
| # | What | Where | Why deferred | Closes when |
|
||||
|---|---|---|---|---|
|
||||
| D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) |
|
||||
| D2 | `download.add` → `-32603`, `download.probe` → `-32603` | `rpc/dispatcher.cpp` | Need path canonicalisation + allowed-root check (`-32011`) and the probe path (`-32013`); those need the engine link | `download.add` glue (after `sched/`) |
|
||||
| D3 | Stub handlers for everything except `session.*`, `download.list`, `download.get` | `rpc/dispatcher.cpp` | No store behind them yet | Per method, as the store/scheduler wire in |
|
||||
| D4 | `sched/` is the pure `Governor` + schedule window only; no `Scheduler` wiring to store/engine/timer | `sched/` | `Engine` bodies land in CORE stage 8; `Scheduler` needs the UUID↔`vdm::TaskId` map, a store query layer, and a timer | After CORE stage 8 lands `Engine::start()` |
|
||||
| D5 | `event.*` fan-out not implemented; `session.subscribe` accepts and echoes but nothing is emitted | `rpc/uds_server.cpp`, `rpc/ws_server.cpp` | No task state to broadcast until the engine is wired | With the callback → `event.*` projection |
|
||||
@@ -0,0 +1,158 @@
|
||||
#include "sched/governor.hpp"
|
||||
|
||||
#include <algorithm>
|
||||
#include <limits>
|
||||
#include <tuple>
|
||||
#include <unordered_map>
|
||||
|
||||
namespace velox::daemon::sched {
|
||||
|
||||
namespace {
|
||||
|
||||
constexpr std::int64_t kUnlimited = std::numeric_limits<std::int64_t>::max();
|
||||
|
||||
// A total order over tasks: main-list (no queue) outranks queued-in-queue; within a queue
|
||||
// by run order; then FIFO by admission rank; task_id breaks any remaining tie so the
|
||||
// result is deterministic. "Lower" == higher priority == admitted first, paused last.
|
||||
auto priority_key(const TaskView& t) {
|
||||
const int tier = t.queue_id ? 1 : 0;
|
||||
const std::string& q = t.queue_id ? *t.queue_id : "";
|
||||
return std::make_tuple(tier, q, t.queue_position, t.admit_rank, t.task_id);
|
||||
}
|
||||
|
||||
bool less_priority(const TaskView* a, const TaskView* b) {
|
||||
return priority_key(*a) < priority_key(*b);
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
Decision Governor::evaluate(const std::vector<TaskView>& tasks,
|
||||
const std::vector<QueueView>& queues) const {
|
||||
Decision d;
|
||||
|
||||
std::unordered_map<std::string, const QueueView*> qby;
|
||||
for (const auto& q : queues) qby.emplace(q.queue_id, &q);
|
||||
|
||||
const std::int64_t global_cap =
|
||||
std::min(cfg_.max_concurrent_downloads, cfg_.max_active_segments);
|
||||
|
||||
auto queue_runnable = [&](const std::optional<std::string>& qid) -> bool {
|
||||
if (!qid) return true; // the main list is always "runnable"
|
||||
const auto it = qby.find(*qid);
|
||||
return it != qby.end() && it->second->running && it->second->window_open;
|
||||
};
|
||||
auto queue_cap = [&](const std::string& qid) -> std::int64_t {
|
||||
const auto it = qby.find(qid);
|
||||
return it == qby.end() ? 0 : std::max<std::int64_t>(1, it->second->max_concurrent);
|
||||
};
|
||||
auto host_cap = [&](const std::string& host) -> std::int64_t {
|
||||
if (host.empty()) return kUnlimited;
|
||||
const auto it = cfg_.host_caps.find(host);
|
||||
return it == cfg_.host_caps.end() ? kUnlimited : it->second;
|
||||
};
|
||||
|
||||
// ---- phase 1: pause running tasks whose queue can no longer host them -------------
|
||||
std::vector<const TaskView*> running;
|
||||
for (const auto& t : tasks) {
|
||||
if (t.run_state == RunState::Running) running.push_back(&t);
|
||||
}
|
||||
std::sort(running.begin(), running.end(), less_priority);
|
||||
|
||||
std::vector<const TaskView*> survivors;
|
||||
for (const auto* t : running) {
|
||||
if (t->queue_id && !queue_runnable(t->queue_id)) {
|
||||
const auto it = qby.find(*t->queue_id);
|
||||
const bool stopped_not_windowed =
|
||||
it != qby.end() && it->second->running && !it->second->window_open;
|
||||
const PauseReason why =
|
||||
stopped_not_windowed ? PauseReason::Schedule : PauseReason::QueueStopped;
|
||||
d.to_pause.push_back(t->task_id);
|
||||
d.pause_reasons.emplace(t->task_id, why);
|
||||
} else {
|
||||
survivors.push_back(t);
|
||||
}
|
||||
}
|
||||
|
||||
// ---- phase 2a: per-queue reconcile (a lowered Queue.maxConcurrent) ---------------
|
||||
{
|
||||
std::unordered_map<std::string, std::int64_t> per_queue;
|
||||
std::vector<const TaskView*> kept;
|
||||
for (const auto* t : survivors) { // already priority-sorted
|
||||
if (!t->queue_id) {
|
||||
kept.push_back(t);
|
||||
continue;
|
||||
}
|
||||
auto& n = per_queue[*t->queue_id];
|
||||
if (n < queue_cap(*t->queue_id)) {
|
||||
++n;
|
||||
kept.push_back(t);
|
||||
} else {
|
||||
d.to_pause.push_back(t->task_id);
|
||||
d.pause_reasons.emplace(t->task_id, PauseReason::AdmissionReconcile);
|
||||
}
|
||||
}
|
||||
survivors.swap(kept);
|
||||
}
|
||||
|
||||
// ---- phase 2b: global reconcile (a lowered maxConcurrentDownloads / clamp) -------
|
||||
if (static_cast<std::int64_t>(survivors.size()) > global_cap) {
|
||||
for (std::size_t i = static_cast<std::size_t>(std::max<std::int64_t>(global_cap, 0));
|
||||
i < survivors.size(); ++i) {
|
||||
d.to_pause.push_back(survivors[i]->task_id);
|
||||
d.pause_reasons.emplace(survivors[i]->task_id, PauseReason::AdmissionReconcile);
|
||||
}
|
||||
survivors.resize(static_cast<std::size_t>(std::max<std::int64_t>(global_cap, 0)));
|
||||
}
|
||||
|
||||
// ---- phase 3: fill free slots from Queued + governor-owned Paused ----------------
|
||||
std::int64_t global_running = static_cast<std::int64_t>(survivors.size());
|
||||
std::unordered_map<std::string, std::int64_t> per_queue_running;
|
||||
std::unordered_map<std::string, std::int64_t> per_host_running;
|
||||
for (const auto* t : survivors) {
|
||||
if (t->queue_id) ++per_queue_running[*t->queue_id];
|
||||
if (!t->host.empty()) ++per_host_running[t->host];
|
||||
}
|
||||
|
||||
auto resumable = [](const TaskView& t) {
|
||||
return t.run_state == RunState::Paused && t.pause_reason &&
|
||||
(*t.pause_reason == PauseReason::Schedule ||
|
||||
*t.pause_reason == PauseReason::QueueStopped ||
|
||||
*t.pause_reason == PauseReason::AdmissionReconcile);
|
||||
};
|
||||
|
||||
std::vector<const TaskView*> candidates;
|
||||
for (const auto& t : tasks) {
|
||||
if (t.run_state == RunState::Queued || resumable(t)) {
|
||||
if (queue_runnable(t.queue_id)) candidates.push_back(&t);
|
||||
}
|
||||
}
|
||||
std::sort(candidates.begin(), candidates.end(), less_priority);
|
||||
|
||||
std::vector<const TaskView*> newly_running;
|
||||
for (const auto* c : candidates) {
|
||||
if (global_running >= global_cap) break;
|
||||
if (c->queue_id && per_queue_running[*c->queue_id] >= queue_cap(*c->queue_id)) continue;
|
||||
if (!c->host.empty() && per_host_running[c->host] >= host_cap(c->host)) continue;
|
||||
|
||||
if (c->run_state == RunState::Queued) {
|
||||
d.to_start.push_back(c->task_id);
|
||||
} else {
|
||||
d.to_resume.push_back(c->task_id);
|
||||
}
|
||||
newly_running.push_back(c);
|
||||
++global_running;
|
||||
if (c->queue_id) ++per_queue_running[*c->queue_id];
|
||||
if (!c->host.empty()) ++per_host_running[c->host];
|
||||
}
|
||||
|
||||
// ---- priority order: everyone who will be running after this decision ------------
|
||||
std::vector<const TaskView*> will_run = survivors;
|
||||
will_run.insert(will_run.end(), newly_running.begin(), newly_running.end());
|
||||
std::sort(will_run.begin(), will_run.end(), less_priority);
|
||||
d.priority_order.reserve(will_run.size());
|
||||
for (const auto* t : will_run) d.priority_order.push_back(t->task_id);
|
||||
|
||||
return d;
|
||||
}
|
||||
|
||||
} // namespace velox::daemon::sched
|
||||
@@ -0,0 +1,89 @@
|
||||
#pragma once
|
||||
|
||||
// The concurrency governor — a pure decision function. Given a snapshot of every task's
|
||||
// coarse run-state and every queue's state, it decides which tasks should start, which
|
||||
// should resume, which should pause, and the priority order to push to CORE's segment
|
||||
// budget. No I/O, no store, no engine, no clock — the caller supplies an already-evaluated
|
||||
// world (schedule windows resolved via sched/schedule_window.hpp) and applies the result.
|
||||
//
|
||||
// Axes it enforces, all in TASK units (ADR 0011 §1: DAEMON counts tasks, CORE counts
|
||||
// segments):
|
||||
// * connection.maxConcurrentDownloads — global running-task ceiling
|
||||
// * min(that, connection.maxActiveSegments) — the one ADR 0011 §2 clamp
|
||||
// * Queue.maxConcurrent — per-queue running-task ceiling
|
||||
// * per-host task cap (== that host's segment cap) — ADR 0011 §4
|
||||
// * queue running state + schedule window — a stopped/closed queue runs nothing
|
||||
//
|
||||
// What it never does: touch a task paused for a reason it does not own (user, or CORE's
|
||||
// auto-pause for auth_required / server_file_changed / disk_full) — ADR 0013 §3.
|
||||
|
||||
#include <cstdint>
|
||||
#include <map>
|
||||
#include <optional>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
namespace velox::daemon::sched {
|
||||
|
||||
// The governor's coarse view of a task. Derived from the SQLite row + the engine's last
|
||||
// reported state; the governor does not need the fine-grained EngineState.
|
||||
enum class RunState {
|
||||
Queued, // wants a slot; eligible for admission
|
||||
Running, // probing / connecting / downloading / retry_wait / assembling / verifying
|
||||
Paused, // see pause_reason for whether the governor may resume it
|
||||
Terminal, // complete / failed / cancelled — ignored
|
||||
};
|
||||
|
||||
// Why a paused task is paused. Only the first three are governor-owned and auto-resumable;
|
||||
// `User` and `Auto` (CORE's auth/decision/disk auto-pause) are never touched here.
|
||||
enum class PauseReason { User, Schedule, QueueStopped, AdmissionReconcile, Auto };
|
||||
|
||||
struct TaskView {
|
||||
std::string task_id; // wire UUID
|
||||
RunState run_state = RunState::Queued;
|
||||
std::optional<PauseReason> pause_reason; // set iff run_state == Paused
|
||||
std::optional<std::string> queue_id; // nullopt => the main list (no queue)
|
||||
std::int64_t queue_position = 0; // order within the queue
|
||||
std::string host; // for the per-host cap; "" opts out
|
||||
std::int64_t admit_rank = 0; // FIFO tiebreak (created_at, then rowid)
|
||||
};
|
||||
|
||||
struct QueueView {
|
||||
std::string queue_id;
|
||||
bool running = false; // Queue.state == "running"
|
||||
std::int64_t max_concurrent = 1;
|
||||
bool window_open = true; // sched/schedule_window.hpp result for right now
|
||||
};
|
||||
|
||||
struct GovernorConfig {
|
||||
std::int64_t max_concurrent_downloads = 5; // connection.maxConcurrentDownloads
|
||||
std::int64_t max_active_segments = 32; // connection.maxActiveSegments (§2 clamp)
|
||||
std::map<std::string, std::int64_t> host_caps{}; // host -> max running tasks; absent => unlimited
|
||||
};
|
||||
|
||||
struct Decision {
|
||||
std::vector<std::string> to_start; // Queued -> admit: caller invokes Engine::start()
|
||||
std::vector<std::string> to_resume; // Paused (governor-owned reason) -> Engine::resume()
|
||||
std::vector<std::string> to_pause; // Running -> Engine::pause(); reason in pause_reasons
|
||||
std::map<std::string, PauseReason> pause_reasons; // task_id -> why, for the DB column
|
||||
std::vector<std::string> priority_order; // running + starting + resuming, for set_task_order()
|
||||
};
|
||||
|
||||
class Governor {
|
||||
public:
|
||||
Governor() = default;
|
||||
explicit Governor(GovernorConfig cfg) : cfg_(std::move(cfg)) {}
|
||||
|
||||
void set_config(GovernorConfig cfg) { cfg_ = std::move(cfg); }
|
||||
const GovernorConfig& config() const noexcept { return cfg_; }
|
||||
|
||||
// Pure: same inputs -> same Decision. `tasks` and `queues` are snapshots; order within
|
||||
// them does not matter (the governor sorts by its own keys).
|
||||
Decision evaluate(const std::vector<TaskView>& tasks,
|
||||
const std::vector<QueueView>& queues) const;
|
||||
|
||||
private:
|
||||
GovernorConfig cfg_;
|
||||
};
|
||||
|
||||
} // namespace velox::daemon::sched
|
||||
@@ -0,0 +1,89 @@
|
||||
#include "sched/schedule_window.hpp"
|
||||
|
||||
#include <array>
|
||||
#include <charconv>
|
||||
#include <optional>
|
||||
#include <string_view>
|
||||
|
||||
namespace velox::daemon::sched {
|
||||
|
||||
namespace proto = velox::proto;
|
||||
|
||||
namespace {
|
||||
|
||||
// "HH:MM" -> minutes since midnight. Returns nullopt on a malformed value (the schema
|
||||
// pattern should prevent that, but the schedule can also come from an older DB row).
|
||||
std::optional<int> minutes_of(std::string_view hhmm) {
|
||||
if (hhmm.size() != 5 || hhmm[2] != ':') return std::nullopt;
|
||||
int h = 0;
|
||||
int m = 0;
|
||||
if (std::from_chars(hhmm.data(), hhmm.data() + 2, h).ec != std::errc{}) return std::nullopt;
|
||||
if (std::from_chars(hhmm.data() + 3, hhmm.data() + 5, m).ec != std::errc{}) return std::nullopt;
|
||||
if (h < 0 || h > 23 || m < 0 || m > 59) return std::nullopt;
|
||||
return h * 60 + m;
|
||||
}
|
||||
|
||||
bool in_window(int now_min, std::optional<int> start_min, std::optional<int> stop_min) {
|
||||
const int start = start_min.value_or(0);
|
||||
if (!stop_min) return now_min >= start; // null stop => until end of day
|
||||
const int stop = *stop_min;
|
||||
if (start <= stop) return now_min >= start && now_min < stop;
|
||||
return now_min >= start || now_min < stop; // overnight window
|
||||
}
|
||||
|
||||
// tm_year/mon/mday -> "YYYY-MM-DD" to compare against onceDate.
|
||||
std::array<char, 11> iso_date(const std::tm& t) {
|
||||
std::array<char, 11> out{};
|
||||
const int y = t.tm_year + 1900;
|
||||
const int mo = t.tm_mon + 1;
|
||||
const int d = t.tm_mday;
|
||||
out[0] = static_cast<char>('0' + (y / 1000) % 10);
|
||||
out[1] = static_cast<char>('0' + (y / 100) % 10);
|
||||
out[2] = static_cast<char>('0' + (y / 10) % 10);
|
||||
out[3] = static_cast<char>('0' + y % 10);
|
||||
out[4] = '-';
|
||||
out[5] = static_cast<char>('0' + mo / 10);
|
||||
out[6] = static_cast<char>('0' + mo % 10);
|
||||
out[7] = '-';
|
||||
out[8] = static_cast<char>('0' + d / 10);
|
||||
out[9] = static_cast<char>('0' + d % 10);
|
||||
out[10] = '\0';
|
||||
return out;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
bool window_open(const proto::Schedule& s, const std::tm& now) {
|
||||
if (!s.enabled) return true;
|
||||
|
||||
const int now_min = now.tm_hour * 60 + now.tm_min;
|
||||
const std::optional<int> start = s.startTime ? minutes_of(*s.startTime) : std::nullopt;
|
||||
const std::optional<int> stop = s.stopTime ? minutes_of(*s.stopTime) : std::nullopt;
|
||||
|
||||
if (s.mode == proto::ScheduleMode::Once) {
|
||||
if (!s.onceDate) return false; // "once" with no date never runs
|
||||
const auto today = iso_date(now);
|
||||
if (std::string_view(today.data()) != *s.onceDate) return false;
|
||||
return in_window(now_min, start, stop);
|
||||
}
|
||||
|
||||
// periodic
|
||||
if (s.daysOfWeek && !s.daysOfWeek->empty()) {
|
||||
bool day_ok = false;
|
||||
for (const auto d : *s.daysOfWeek) {
|
||||
if (d == now.tm_wday) {
|
||||
day_ok = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!day_ok) return false;
|
||||
}
|
||||
return in_window(now_min, start, stop);
|
||||
}
|
||||
|
||||
bool window_open(const std::optional<proto::Schedule>& s, const std::tm& now) {
|
||||
if (!s) return true;
|
||||
return window_open(*s, now);
|
||||
}
|
||||
|
||||
} // namespace velox::daemon::sched
|
||||
@@ -0,0 +1,33 @@
|
||||
#pragma once
|
||||
|
||||
// Evaluates a queue's Schedule against the current local wall-clock time: is the queue
|
||||
// allowed to be running right now? Times are local HH:MM and re-evaluated on every tick
|
||||
// (the daemon never caches an absolute instant — docs, Schedule.schema.json), so this is
|
||||
// a pure function of (schedule, broken-down local time).
|
||||
//
|
||||
// A disabled schedule, or no schedule at all, means "always open" — the queue runs
|
||||
// whenever queue.state is 'running'.
|
||||
|
||||
#include <ctime>
|
||||
|
||||
#include "velox_proto.hpp"
|
||||
|
||||
namespace velox::daemon::sched {
|
||||
|
||||
// `local_now` must be a fully populated std::tm in local time (tm_wday and tm_year/mon/mday
|
||||
// used). Returns true when the schedule permits the queue to run at that moment.
|
||||
//
|
||||
// - mode "once": open iff the date matches onceDate and the time is within the window.
|
||||
// daysOfWeek is ignored.
|
||||
// - mode "periodic": open iff today's weekday is in daysOfWeek (an empty/absent list means
|
||||
// every day) and the time is within the window.
|
||||
//
|
||||
// The window is [startTime, stopTime). A null startTime means 00:00. A null stopTime
|
||||
// means "until end of day" (and, for a queue, "until it drains"). A stopTime earlier than
|
||||
// startTime is an overnight window (e.g. 22:00-06:00): open if now >= start OR now < stop.
|
||||
bool window_open(const velox::proto::Schedule& schedule, const std::tm& local_now);
|
||||
|
||||
// Convenience: the queue's schedule is optional. nullopt or disabled => always open.
|
||||
bool window_open(const std::optional<velox::proto::Schedule>& schedule, const std::tm& local_now);
|
||||
|
||||
} // namespace velox::daemon::sched
|
||||
@@ -16,3 +16,5 @@ veloxd_test(store_migrations LIBS veloxd_store)
|
||||
veloxd_test(pairings LIBS veloxd_store veloxd_rpc)
|
||||
veloxd_test(ws_frame LIBS veloxd_rpc)
|
||||
veloxd_test(ws_server LIBS veloxd_rpc)
|
||||
veloxd_test(sched_window LIBS veloxd_sched)
|
||||
veloxd_test(sched_governor LIBS veloxd_sched)
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
#include "sched/governor.hpp"
|
||||
|
||||
#include <algorithm>
|
||||
#include <string>
|
||||
|
||||
#include "check.hpp"
|
||||
|
||||
using namespace velox::daemon::sched;
|
||||
|
||||
namespace {
|
||||
|
||||
bool has(const std::vector<std::string>& v, const std::string& x) {
|
||||
return std::find(v.begin(), v.end(), x) != v.end();
|
||||
}
|
||||
|
||||
TaskView task(std::string id, RunState st, std::optional<std::string> queue = std::nullopt,
|
||||
std::int64_t pos = 0, std::string host = "", std::int64_t rank = 0) {
|
||||
TaskView t;
|
||||
t.task_id = std::move(id);
|
||||
t.run_state = st;
|
||||
t.queue_id = std::move(queue);
|
||||
t.queue_position = pos;
|
||||
t.host = std::move(host);
|
||||
t.admit_rank = rank;
|
||||
return t;
|
||||
}
|
||||
|
||||
QueueView queue(std::string id, bool running, std::int64_t cap, bool window = true) {
|
||||
QueueView q;
|
||||
q.queue_id = std::move(id);
|
||||
q.running = running;
|
||||
q.max_concurrent = cap;
|
||||
q.window_open = window;
|
||||
return q;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
void run() {
|
||||
// --- global cap: 5 queued, max 3 -> start 3, order is all 3 --------------------
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 3, .max_active_segments = 32});
|
||||
std::vector<TaskView> t;
|
||||
for (int i = 0; i < 5; ++i)
|
||||
t.push_back(task("t" + std::to_string(i), RunState::Queued, std::nullopt, 0, "", i));
|
||||
const auto d = g.evaluate(t, {});
|
||||
CHECK_EQ(d.to_start.size(), 3u);
|
||||
CHECK(has(d.to_start, "t0") && has(d.to_start, "t1") && has(d.to_start, "t2"));
|
||||
CHECK_EQ(d.priority_order.size(), 3u);
|
||||
CHECK_EQ(d.priority_order.front(), std::string("t0")); // FIFO by admit_rank
|
||||
CHECK(d.to_pause.empty());
|
||||
}
|
||||
|
||||
// --- the ADR 0011 §2 clamp: maxActiveSegments below maxConcurrentDownloads ------
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 2});
|
||||
std::vector<TaskView> t;
|
||||
for (int i = 0; i < 6; ++i) t.push_back(task("t" + std::to_string(i), RunState::Queued));
|
||||
const auto d = g.evaluate(t, {});
|
||||
CHECK_EQ(d.to_start.size(), 2u); // clamped to the segment budget
|
||||
}
|
||||
|
||||
// --- per-queue cap: queue running, cap 2, 4 queued in it -----------------------
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32});
|
||||
std::vector<TaskView> t;
|
||||
for (int i = 0; i < 4; ++i)
|
||||
t.push_back(task("q" + std::to_string(i), RunState::Queued, "Q", i));
|
||||
const auto d = g.evaluate(t, {queue("Q", true, 2)});
|
||||
CHECK_EQ(d.to_start.size(), 2u);
|
||||
CHECK(has(d.to_start, "q0") && has(d.to_start, "q1")); // lowest queue_position first
|
||||
}
|
||||
|
||||
// --- a stopped queue: its running tasks are paused (QueueStopped) -------------
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32});
|
||||
std::vector<TaskView> t{
|
||||
task("r0", RunState::Running, "Q", 0),
|
||||
task("r1", RunState::Running, "Q", 1),
|
||||
task("m0", RunState::Running, std::nullopt, 0, "", 5), // main list, unaffected
|
||||
};
|
||||
const auto d = g.evaluate(t, {queue("Q", /*running=*/false, 4)});
|
||||
CHECK_EQ(d.to_pause.size(), 2u);
|
||||
CHECK(has(d.to_pause, "r0") && has(d.to_pause, "r1"));
|
||||
CHECK(d.pause_reasons.at("r0") == PauseReason::QueueStopped);
|
||||
CHECK(!has(d.to_pause, "m0"));
|
||||
}
|
||||
|
||||
// --- schedule window closed: pause reason is Schedule, not QueueStopped -------
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32});
|
||||
std::vector<TaskView> t{task("r0", RunState::Running, "Q", 0)};
|
||||
const auto d = g.evaluate(t, {queue("Q", /*running=*/true, 4, /*window=*/false)});
|
||||
CHECK_EQ(d.to_pause.size(), 1u);
|
||||
CHECK(d.pause_reasons.at("r0") == PauseReason::Schedule);
|
||||
}
|
||||
|
||||
// --- resume: a task paused for Schedule comes back when the window reopens ----
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32});
|
||||
TaskView p = task("p0", RunState::Paused, "Q", 0);
|
||||
p.pause_reason = PauseReason::Schedule;
|
||||
TaskView u = task("u0", RunState::Paused, "Q", 1);
|
||||
u.pause_reason = PauseReason::User; // must NOT be auto-resumed
|
||||
const auto d = g.evaluate({p, u}, {queue("Q", true, 4, true)});
|
||||
CHECK_EQ(d.to_resume.size(), 1u);
|
||||
CHECK(has(d.to_resume, "p0"));
|
||||
CHECK(!has(d.to_resume, "u0"));
|
||||
}
|
||||
|
||||
// --- an Auto (auth) pause is never touched, even with slots free -------------
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32});
|
||||
TaskView a = task("a0", RunState::Paused, std::nullopt);
|
||||
a.pause_reason = PauseReason::Auto;
|
||||
const auto d = g.evaluate({a}, {});
|
||||
CHECK(d.to_resume.empty());
|
||||
CHECK(d.to_pause.empty());
|
||||
}
|
||||
|
||||
// --- per-host cap: 3 queued on the same host, cap 1 ------------------------
|
||||
{
|
||||
GovernorConfig cfg{.max_concurrent_downloads = 10, .max_active_segments = 32};
|
||||
cfg.host_caps["cdn.example"] = 1;
|
||||
Governor g(cfg);
|
||||
std::vector<TaskView> t{
|
||||
task("h0", RunState::Queued, std::nullopt, 0, "cdn.example", 0),
|
||||
task("h1", RunState::Queued, std::nullopt, 0, "cdn.example", 1),
|
||||
task("h2", RunState::Queued, std::nullopt, 0, "other.example", 2),
|
||||
};
|
||||
const auto d = g.evaluate(t, {});
|
||||
CHECK_EQ(d.to_start.size(), 2u); // one per host
|
||||
CHECK(has(d.to_start, "h0") && has(d.to_start, "h2"));
|
||||
CHECK(!has(d.to_start, "h1"));
|
||||
}
|
||||
|
||||
// --- admission reconcile: 4 running, cap drops to 2 -> pause the 2 lowest -----
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 2, .max_active_segments = 32});
|
||||
std::vector<TaskView> t;
|
||||
for (int i = 0; i < 4; ++i)
|
||||
t.push_back(task("t" + std::to_string(i), RunState::Running, std::nullopt, 0, "", i));
|
||||
const auto d = g.evaluate(t, {});
|
||||
CHECK_EQ(d.to_pause.size(), 2u);
|
||||
CHECK(has(d.to_pause, "t2") && has(d.to_pause, "t3")); // lowest priority (highest rank)
|
||||
CHECK(d.pause_reasons.at("t2") == PauseReason::AdmissionReconcile);
|
||||
CHECK_EQ(d.priority_order.size(), 2u);
|
||||
}
|
||||
|
||||
// --- determinism: shuffled input, identical Decision -----------------------
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 2, .max_active_segments = 32});
|
||||
std::vector<TaskView> a{
|
||||
task("b", RunState::Queued, std::nullopt, 0, "", 1),
|
||||
task("a", RunState::Queued, std::nullopt, 0, "", 0),
|
||||
task("c", RunState::Queued, std::nullopt, 0, "", 2),
|
||||
};
|
||||
std::vector<TaskView> b{a[2], a[0], a[1]};
|
||||
const auto da = g.evaluate(a, {});
|
||||
const auto db = g.evaluate(b, {});
|
||||
CHECK(da.to_start == db.to_start);
|
||||
CHECK(da.priority_order == db.priority_order);
|
||||
CHECK_EQ(da.priority_order.front(), std::string("a"));
|
||||
}
|
||||
|
||||
// --- main list outranks queues in priority_order -------------------------
|
||||
{
|
||||
Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32});
|
||||
std::vector<TaskView> t{
|
||||
task("qtask", RunState::Running, "Q", 0),
|
||||
task("mtask", RunState::Running, std::nullopt, 0, "", 99),
|
||||
};
|
||||
const auto d = g.evaluate(t, {queue("Q", true, 4)});
|
||||
CHECK_EQ(d.priority_order.front(), std::string("mtask"));
|
||||
}
|
||||
}
|
||||
|
||||
TEST_MAIN()
|
||||
@@ -0,0 +1,120 @@
|
||||
#include "sched/schedule_window.hpp"
|
||||
|
||||
#include <cstring>
|
||||
#include <ctime>
|
||||
|
||||
#include "check.hpp"
|
||||
|
||||
namespace proto = velox::proto;
|
||||
using velox::daemon::sched::window_open;
|
||||
|
||||
namespace {
|
||||
|
||||
// A std::tm for a given weekday (0=Sun) and HH:MM on 2026-09-14 (a Monday) + offset days.
|
||||
std::tm at(int wday, int hh, int mm, const char* date = "2026-09-14") {
|
||||
std::tm t{};
|
||||
t.tm_year = 126; // 2026
|
||||
// parse "YYYY-MM-DD"
|
||||
int y = 0, mo = 0, d = 0;
|
||||
std::sscanf(date, "%d-%d-%d", &y, &mo, &d);
|
||||
t.tm_year = y - 1900;
|
||||
t.tm_mon = mo - 1;
|
||||
t.tm_mday = d;
|
||||
t.tm_wday = wday;
|
||||
t.tm_hour = hh;
|
||||
t.tm_min = mm;
|
||||
return t;
|
||||
}
|
||||
|
||||
proto::Schedule periodic(const char* start, const char* stop, std::vector<std::int64_t> days) {
|
||||
proto::Schedule s;
|
||||
s.enabled = true;
|
||||
s.mode = proto::ScheduleMode::Periodic;
|
||||
if (start) s.startTime = start;
|
||||
if (stop) s.stopTime = stop;
|
||||
if (!days.empty()) s.daysOfWeek = std::move(days);
|
||||
return s;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
void run() {
|
||||
// --- disabled schedule: always open --------------------------------------------
|
||||
{
|
||||
proto::Schedule s;
|
||||
s.enabled = false;
|
||||
s.mode = proto::ScheduleMode::Periodic;
|
||||
CHECK(window_open(s, at(3, 3, 0)));
|
||||
std::optional<proto::Schedule> none;
|
||||
CHECK(window_open(none, at(3, 3, 0)));
|
||||
}
|
||||
|
||||
// --- periodic, weekdays 01:00-06:00 -------------------------------------------
|
||||
{
|
||||
const auto s = periodic("01:00", "06:00", {1, 2, 3, 4, 5}); // Mon-Fri
|
||||
CHECK(window_open(s, at(1, 2, 30))); // Monday 02:30 -> open
|
||||
CHECK(!window_open(s, at(1, 6, 0))); // 06:00 is the exclusive end
|
||||
CHECK(!window_open(s, at(1, 0, 59))); // before start
|
||||
CHECK(!window_open(s, at(0, 2, 30))); // Sunday -> wrong day
|
||||
CHECK(!window_open(s, at(6, 2, 30))); // Saturday -> wrong day
|
||||
}
|
||||
|
||||
// --- periodic, no daysOfWeek -> every day ------------------------------------
|
||||
{
|
||||
const auto s = periodic("09:00", "17:00", {});
|
||||
CHECK(window_open(s, at(0, 12, 0))); // Sunday still fine
|
||||
CHECK(window_open(s, at(3, 9, 0))); // inclusive start
|
||||
CHECK(!window_open(s, at(3, 17, 0)));
|
||||
}
|
||||
|
||||
// --- overnight window 22:00-06:00 ------------------------------------------
|
||||
{
|
||||
const auto s = periodic("22:00", "06:00", {});
|
||||
CHECK(window_open(s, at(3, 23, 0))); // late evening
|
||||
CHECK(window_open(s, at(3, 2, 0))); // early morning
|
||||
CHECK(!window_open(s, at(3, 12, 0))); // midday -> closed
|
||||
CHECK(window_open(s, at(3, 22, 0))); // inclusive start
|
||||
CHECK(!window_open(s, at(3, 6, 0))); // exclusive end
|
||||
}
|
||||
|
||||
// --- null stopTime -> until end of day ----------------------------------
|
||||
{
|
||||
auto s = periodic("20:00", nullptr, {});
|
||||
CHECK(!window_open(s, at(3, 19, 59)));
|
||||
CHECK(window_open(s, at(3, 20, 0)));
|
||||
CHECK(window_open(s, at(3, 23, 59)));
|
||||
}
|
||||
|
||||
// --- null startTime -> from midnight -----------------------------------
|
||||
{
|
||||
auto s = periodic(nullptr, "08:00", {});
|
||||
CHECK(window_open(s, at(3, 0, 0)));
|
||||
CHECK(window_open(s, at(3, 7, 59)));
|
||||
CHECK(!window_open(s, at(3, 8, 0)));
|
||||
}
|
||||
|
||||
// --- mode "once": only on the named date ------------------------------
|
||||
{
|
||||
proto::Schedule s;
|
||||
s.enabled = true;
|
||||
s.mode = proto::ScheduleMode::Once;
|
||||
s.startTime = "10:00";
|
||||
s.stopTime = "12:00";
|
||||
s.onceDate = "2026-09-14";
|
||||
CHECK(window_open(s, at(1, 11, 0, "2026-09-14"))); // right date + time
|
||||
CHECK(!window_open(s, at(1, 9, 0, "2026-09-14"))); // right date, before window
|
||||
CHECK(!window_open(s, at(2, 11, 0, "2026-09-15"))); // wrong date
|
||||
s.daysOfWeek = {2}; // ignored for "once"
|
||||
CHECK(window_open(s, at(1, 11, 0, "2026-09-14")));
|
||||
}
|
||||
|
||||
// --- "once" with no onceDate never runs ------------------------------
|
||||
{
|
||||
proto::Schedule s;
|
||||
s.enabled = true;
|
||||
s.mode = proto::ScheduleMode::Once;
|
||||
CHECK(!window_open(s, at(1, 11, 0)));
|
||||
}
|
||||
}
|
||||
|
||||
TEST_MAIN()
|
||||
Reference in New Issue
Block a user