sami/vdm
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

core: segment/segmenter + segment/budget (stage 6)

Browse Source 
vdm/ids.hpp — TaskId, an opaque engine handle (DAEMON keeps the wire
UUID <-> TaskId map; the engine never sees the UUID).

segment/segmenter — per-download range management (docs/04 §3). Initial
lazy split; assign_slot() splits the largest remaining range when the
budget grants a slot; on_complete(may_steal) either *steals* the second
half of the largest remaining range for the same worker (slot-neutral) or
returns nullopt so the caller *yields* the slot (ADR 0011 A1); on_failed()
returns requeue only on the 3rd consecutive connection error with a mirror
present — the remaining range is orphaned and re-split. Non-resumable or
unknown-size => exactly 1 segment; never split below min_segment_bytes
(1 MiB). Resume ctor rebuilds from a persisted table (falls back to a
fresh layout if it doesn't tile [0,total)). One mutex == "the task lock";
segment fields are std::atomic and the store is a std::deque so a steal's
append never moves a worker's record.

segment/budget — the global allocator (ADR 0011). Owns exactly one
ceiling (maxActiveSegments) and min-1-before-seconds fairness: a two-pass
allocation (guarantee pass gives every wanting task 1 slot in DAEMON's
priority order, then a growth pass round-robins the rest up to each
task's effective cap = min(per_task_cap, host cap, 1 if non-resumable)),
recomputed from scratch on every edge so a live set_max_active_segments
cut naturally yields the excess lowest-priority-first, never a
mid-segment kill. DAEMON-facing surface exactly as promised in
daemon/docs/core-requests-m1.md / ADR 0011: budget(), segments_active(),
starved_tasks(), starved_since(), set_max_active_segments (drain),
set_host_segment_cap, set_task_order, on_budget_changed (a jthread
coalesces at <=4 Hz; the tasks_starved 0<->nonzero edge fires
immediately). Callbacks are copied out and run after the lock is
dropped.

Tests: segmenter split/steal/requeue/resume math + a concurrent
steal-and-advance run; budget min-1 under a tight budget, round-robin
growth, host-cap and non-resumable clamps, live-lower shedding
lowest-priority-first, starvation below the task count, starved-edge
notification, and a concurrent set_want hammer. Green under ASan/UBSan;
the steal path and the budget green under TSan.

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS
This commit is contained in:
sami
2026-09-10 15:14:14 +04:00
co-authored by Claude Sonnet 5
parent 1fabaf805d
commit 5d81b4cdae
9 changed files with 1485 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/tests/segment/budget_test.cpp
+264
View File
@@ -0,0 +1,264 @@
#include "vdm/segment/budget.hpp"
#include <atomic>
#include <chrono>
#include <mutex>
#include <thread>
#include <vector>
#include "vtest.hpp"
using namespace vdm;
using namespace vdm::segment;
using EB = SegmentBudget::EngineBudget;
namespace {
TaskId tid(std::uint64_t v) {
return TaskId{v};
}
// A test task that reacts to slot targets the way stage 8's download_task will: start
// workers up to the target, release them when the target drops. Purely bookkeeping.
struct FakeTask {
SegmentBudget *budget = nullptr;
TaskId id{};
std::mutex mu;
std::uint32_t workers = 0;
std::uint32_t target = 0;
FakeTask() = default;
FakeTask(SegmentBudget *b, TaskId i) : budget(b), id(i) {}
void on_target(std::uint32_t t) {
std::lock_guard lk(mu);
target = t;
while (workers < target) {
if (!budget->confirm_slot(id))
break;
++workers;
}
// over target -> yield the excess immediately (a real task waits for a boundary)
while (workers > target) {
budget->release_slot(id);
--workers;
}
}
std::uint32_t held() {
std::lock_guard lk(mu);
return workers;
}
};
} // namespace
VT_TEST(budget_single_task_grows_to_cap) {
SegmentBudget b({.max_active_segments = 32});
FakeTask t{&b, tid(1)};
b.register_task(tid(1), {.host = "h", .per_task_cap = 8, .resumable = true},
[&](std::uint32_t n) { t.on_target(n); });
b.set_want(tid(1), 8);
VT_CHECK_EQ(t.held(), 8u);
VT_CHECK_EQ(b.segments_active(tid(1)), 8u);
VT_CHECK_EQ(b.budget().active, 8u);
VT_CHECK_EQ(b.budget().tasks_starved, 0u);
}
VT_TEST(budget_min_one_before_seconds) {
// Budget of 3, two tasks each wanting 8. min-1 first: each gets 1, then the higher-
// priority one grows to 2.
SegmentBudget b({.max_active_segments = 3});
FakeTask a{&b, tid(1)}, c{&b, tid(2)};
b.register_task(tid(1), {.host = "h1", .per_task_cap = 8, .resumable = true},
[&](std::uint32_t n) { a.on_target(n); });
b.register_task(tid(2), {.host = "h2", .per_task_cap = 8, .resumable = true},
[&](std::uint32_t n) { c.on_target(n); });
std::vector<TaskId> order = {tid(1), tid(2)};
b.set_task_order(order);
b.set_want(tid(1), 8);
b.set_want(tid(2), 8);
VT_CHECK(a.held() >= 1); // guarantee
VT_CHECK(c.held() >= 1); // guarantee — the load-bearing property
VT_CHECK_EQ(a.held() + c.held(), 3u);
VT_CHECK_EQ(a.held(), 2u); // higher priority took the growth slot
}
VT_TEST(budget_new_high_priority_task_gets_min_one_via_yield) {
SegmentBudget b({.max_active_segments = 4});
FakeTask a{&b, tid(1)};
b.register_task(tid(1), {.host = "h", .per_task_cap = 8, .resumable = true},
[&](std::uint32_t n) { a.on_target(n); });
b.set_task_order(std::vector<TaskId>{tid(1)});
b.set_want(tid(1), 8);
VT_CHECK_EQ(a.held(), 4u); // hogging the whole budget
// a second, higher-priority task arrives
FakeTask c{&b, tid(2)};
b.register_task(tid(2), {.host = "h2", .per_task_cap = 8, .resumable = true},
[&](std::uint32_t n) { c.on_target(n); });
b.set_task_order(std::vector<TaskId>{tid(2), tid(1)});
b.set_want(tid(2), 8);
// a yields so c gets at least its guaranteed slot; the surplus is shared round-robin.
VT_CHECK(c.held() >= 1); // min-1 — the load-bearing guarantee
VT_CHECK(a.held() >= 1); // a keeps its own min-1
VT_CHECK(a.held() < 4); // a really did yield at least one
VT_CHECK_EQ(a.held() + c.held(), 4u);
VT_CHECK_EQ(b.budget().active, 4u);
VT_CHECK_EQ(b.budget().tasks_starved, 0u);
}
VT_TEST(budget_host_cap_clamps_effective_target) {
SegmentBudget b({.max_active_segments = 32});
FakeTask t{&b, tid(1)};
b.set_host_segment_cap("slowcdn", 4);
b.register_task(tid(1), {.host = "slowcdn", .per_task_cap = 16, .resumable = true},
[&](std::uint32_t n) { t.on_target(n); });
b.set_want(tid(1), 16);
VT_CHECK_EQ(t.held(), 4u); // clamped by the host cap, not per_task_cap
b.set_host_segment_cap("slowcdn", 0); // clear
VT_CHECK_EQ(t.held(), 16u);
}
VT_TEST(budget_non_resumable_task_capped_at_one) {
SegmentBudget b({.max_active_segments = 32});
FakeTask t{&b, tid(1)};
b.register_task(tid(1), {.host = "h", .per_task_cap = 8, .resumable = false},
[&](std::uint32_t n) { t.on_target(n); });
b.set_want(tid(1), 8);
VT_CHECK_EQ(t.held(), 1u);
}
VT_TEST(budget_live_lower_sheds_via_yield_lowest_priority_first) {
SegmentBudget b({.max_active_segments = 24});
FakeTask a{&b, tid(1)}, c{&b, tid(2)}, d{&b, tid(3)};
for (auto *ft : {&a, &c, &d})
b.register_task(ft->id, {.host = "h", .per_task_cap = 8, .resumable = true},
[ft](std::uint32_t n) { ft->on_target(n); });
b.set_task_order(std::vector<TaskId>{tid(1), tid(2), tid(3)});
for (auto id : {tid(1), tid(2), tid(3)})
b.set_want(id, 8);
VT_CHECK_EQ(a.held() + c.held() + d.held(), 24u); // 8 + 8 + 8
b.set_max_active_segments(10); // live cut
VT_CHECK_EQ(a.held() + c.held() + d.held(), 10u);
VT_CHECK(a.held() >= c.held() && c.held() >= d.held()); // priority order preserved
VT_CHECK(a.held() >= 1 && c.held() >= 1 && d.held() >= 1); // min-1 still honoured
}
VT_TEST(budget_live_lower_below_task_count_starves_the_tail) {
SegmentBudget b({.max_active_segments = 6});
std::vector<FakeTask> ts(4);
for (std::uint32_t i = 0; i < 4; ++i) {
ts[i].budget = &b;
ts[i].id = tid(i + 1);
}
for (auto &ft : ts)
b.register_task(ft.id, {.host = "h", .per_task_cap = 4, .resumable = true},
[&ft](std::uint32_t n) { ft.on_target(n); });
b.set_task_order(std::vector<TaskId>{tid(1), tid(2), tid(3), tid(4)});
for (auto &ft : ts)
b.set_want(ft.id, 4);
VT_CHECK_EQ(b.budget().tasks_starved, 0u);
b.set_max_active_segments(3); // below the running-task count
VT_CHECK_EQ(ts[0].held(), 1u);
VT_CHECK_EQ(ts[3].held(), 0u); // lowest priority shed to zero
VT_CHECK_EQ(b.budget().tasks_starved, 1u);
VT_REQUIRE(b.starved_tasks().size() == 1);
VT_CHECK_EQ(b.starved_tasks()[0], tid(4));
VT_CHECK(b.starved_since(tid(4)).has_value());
VT_CHECK(!b.starved_since(tid(1)).has_value());
}
VT_TEST(budget_deregister_frees_slots_to_starved) {
SegmentBudget b({.max_active_segments = 4});
FakeTask a{&b, tid(1)}, c{&b, tid(2)};
b.register_task(tid(1), {.host = "h", .per_task_cap = 8, .resumable = true},
[&](std::uint32_t n) { a.on_target(n); });
b.set_task_order(std::vector<TaskId>{tid(1)});
b.set_want(tid(1), 8);
VT_CHECK_EQ(a.held(), 4u);
b.register_task(tid(2), {.host = "h", .per_task_cap = 8, .resumable = true},
[&](std::uint32_t n) { c.on_target(n); });
b.set_task_order(std::vector<TaskId>{tid(1), tid(2)});
b.set_want(tid(2), 8);
VT_CHECK(c.held() >= 1); // min-1 from a's yield
b.deregister_task(tid(1));
VT_CHECK_EQ(c.held(), 4u); // c grows into the whole freed budget
VT_CHECK_EQ(b.budget().active, 4u);
}
VT_TEST(budget_on_changed_fires_on_starved_edge) {
SegmentBudget b({.max_active_segments = 1, .notify_period = std::chrono::milliseconds{40}});
std::mutex m;
std::vector<EB> seen;
b.on_budget_changed([&](EB e) {
std::lock_guard lk(m);
seen.push_back(e);
});
FakeTask a{&b, tid(1)}, c{&b, tid(2)};
b.register_task(tid(1), {.host = "h", .per_task_cap = 4, .resumable = true},
[&](std::uint32_t n) { a.on_target(n); });
b.register_task(tid(2), {.host = "h", .per_task_cap = 4, .resumable = true},
[&](std::uint32_t n) { c.on_target(n); });
b.set_task_order(std::vector<TaskId>{tid(1), tid(2)});
b.set_want(tid(1), 4);
b.set_want(tid(2), 4); // budget is 1 -> tid(2) is starved: 0 -> nonzero edge
// the edge fire is synchronous on the triggering call
bool saw_starved = false;
{
std::lock_guard lk(m);
for (auto &e : seen)
if (e.tasks_starved > 0)
saw_starved = true;
}
VT_CHECK(saw_starved);
b.deregister_task(tid(1)); // frees the slot -> tid(2) no longer starved: edge back
std::this_thread::sleep_for(std::chrono::milliseconds(120));
bool saw_unstarved_after = false;
{
std::lock_guard lk(m);
VT_CHECK(!seen.empty());
saw_unstarved_after = seen.back().tasks_starved == 0;
}
VT_CHECK(saw_unstarved_after);
}
VT_TEST(budget_concurrent_confirm_release_stays_consistent) {
SegmentBudget b({.max_active_segments = 16});
constexpr int kTasks = 6;
std::vector<std::unique_ptr<FakeTask>> ts;
for (int i = 0; i < kTasks; ++i) {
ts.push_back(std::make_unique<FakeTask>());
ts.back()->budget = &b;
ts.back()->id = tid(i + 1);
FakeTask *ft = ts.back().get();
b.register_task(ft->id, {.host = "h", .per_task_cap = 6, .resumable = true},
[ft](std::uint32_t n) { ft->on_target(n); });
}
std::vector<std::jthread> drivers;
for (int i = 0; i < kTasks; ++i) {
drivers.emplace_back([&, id = tid(i + 1)] {
for (int r = 0; r < 4000; ++r)
b.set_want(id, (r % 7));
});
}
drivers.clear(); // join
for (auto &ft : ts)
b.set_want(ft->id, 0);
// With everyone wanting nothing, the budget must be fully released.
VT_CHECK_EQ(b.budget().active, 0u);
std::uint32_t sum = 0;
for (auto &ft : ts)
sum += b.segments_active(ft->id);
VT_CHECK_EQ(sum, 0u);
}
core/tests/segment/segmenter_test.cpp
+257
View File
@@ -0,0 +1,257 @@
#include "vdm/segment/segmenter.hpp"
#include <atomic>
#include <cstdint>
#include <thread>
#include <vector>
#include "vtest.hpp"
using namespace vdm::segment;
namespace {
constexpr std::uint64_t MiB = 1u << 20;
// Assign `n` slots (bounded by what the segmenter hands out) and return the indices.
std::vector<std::uint32_t> fill(Segmenter &s, int n) {
std::vector<std::uint32_t> idx;
for (int i = 0; i < n; ++i) {
auto a = s.assign_slot();
if (!a)
break;
idx.push_back(*a);
}
return idx;
}
// Do the segments (by their [start,end] at this instant) tile [0,total) with no overlap?
bool tiles_exactly(const Segmenter &s) {
auto snap = s.snapshot();
std::vector<std::pair<std::uint64_t, std::uint64_t>> r;
for (auto &v : snap)
if (v.state != SegState::failed)
r.emplace_back(v.start, v.end);
std::sort(r.begin(), r.end());
std::uint64_t cursor = 0;
for (auto [a, b] : r) {
if (a != cursor)
return false;
cursor = b + 1;
}
return cursor == s.total_size();
}
} // namespace
VT_TEST(seg_target_count_clamps) {
VT_CHECK_EQ(Segmenter(100 * MiB, 8, true).target_segment_count(), 8u);
VT_CHECK_EQ(Segmenter(100 * MiB, 64, true).target_segment_count(), 32u); // max 32
VT_CHECK_EQ(Segmenter(100 * MiB, 0, true).target_segment_count(), 8u); // default
VT_CHECK_EQ(Segmenter(3 * MiB + 1, 8, true).target_segment_count(), 3u); // total/min
VT_CHECK_EQ(Segmenter(100 * MiB, 8, false).target_segment_count(), 1u); // non-resumable
VT_CHECK_EQ(Segmenter(0, 8, true).target_segment_count(), 1u); // chunked
}
VT_TEST(seg_non_resumable_is_one_segment) {
Segmenter s(50 * MiB, 8, false);
auto idx = fill(s, 8);
VT_REQUIRE(idx.size() == 1);
auto snap = s.snapshot();
VT_REQUIRE(snap.size() == 1);
VT_CHECK_EQ(snap[0].start, 0u);
VT_CHECK_EQ(snap[0].end, 50u * MiB - 1);
}
VT_TEST(seg_initial_split_covers_range) {
Segmenter s(80 * MiB, 8, true);
auto idx = fill(s, 8);
VT_CHECK_EQ(idx.size(), 8u);
VT_CHECK(tiles_exactly(s));
// no segment below the 1 MiB floor
for (auto &v : s.snapshot())
VT_CHECK(v.length() >= MiB);
}
VT_TEST(seg_split_stops_at_min_floor) {
// 5 MiB, floor 1 MiB, ask for 8: only ~5 splits possible (each half >= 1 MiB needs
// the parent >= 2 MiB), so we get fewer than 8.
Segmenter s(5 * MiB, 8, true);
auto idx = fill(s, 8);
VT_CHECK(idx.size() >= 1 && idx.size() <= 5);
VT_CHECK(tiles_exactly(s));
}
VT_TEST(seg_steal_takes_second_half_of_largest_remaining) {
Segmenter s(80 * MiB, 4, true);
auto idx = fill(s, 4);
VT_REQUIRE(idx.size() == 4);
// Spread progress unevenly. Index != file position after splits, so identify the
// largest-remaining segment by scanning the snapshot, not by index.
s.advance(idx[1], 3 * MiB);
s.advance(idx[2], 7 * MiB);
s.advance(idx[3], 12 * MiB);
SegmentView pre_victim{};
std::uint64_t worst = 0;
for (auto &v : s.snapshot())
if (v.index != idx[0] && v.remaining() > worst) {
worst = v.remaining();
pre_victim = v;
}
auto cont = s.on_complete(idx[0], /*may_steal=*/true);
VT_REQUIRE(cont.has_value());
SegmentView victim{}, fresh{};
for (auto &v : s.snapshot()) {
if (v.index == pre_victim.index)
victim = v;
if (v.index == *cont)
fresh = v;
}
VT_CHECK_EQ(fresh.end, pre_victim.end); // fresh takes the tail of the victim's range
VT_CHECK_EQ(victim.end + 1, fresh.start); // contiguous, no gap / no overlap
VT_CHECK(victim.end < pre_victim.end); // the victim really did shrink
VT_CHECK(fresh.length() >= MiB);
VT_CHECK(victim.remaining() >= MiB);
// fresh got roughly the back half of what was remaining
VT_CHECK(fresh.length() >= worst / 2 - MiB && fresh.length() <= worst / 2 + MiB);
VT_CHECK(tiles_exactly(s));
}
VT_TEST(seg_complete_without_steal_releases) {
Segmenter s(4 * MiB, 2, true);
auto idx = fill(s, 2); // 2 x 2 MiB
VT_REQUIRE(idx.size() == 2);
s.advance(idx[1], 2 * MiB);
s.set_segment_state(idx[1], SegState::complete);
// idx[0] done, nothing left worth >= 1 MiB to steal -> release
s.advance(idx[0], 2 * MiB);
auto cont = s.on_complete(idx[0], true);
VT_CHECK(!cont.has_value());
}
VT_TEST(seg_yield_returns_nullopt) {
Segmenter s(80 * MiB, 4, true);
auto idx = fill(s, 4);
s.advance(idx[0], 20 * MiB);
auto cont = s.on_complete(idx[0], /*may_steal=*/false); // yielding
VT_CHECK(!cont.has_value());
}
VT_TEST(seg_third_connection_failure_with_mirror_requeues) {
Segmenter s(40 * MiB, 2, true);
auto idx = fill(s, 2);
s.advance(idx[0], 4 * MiB);
VT_CHECK(s.on_failed(idx[0], /*conn=*/true, /*mirror=*/true) == FailAction::retry);
VT_CHECK(s.on_failed(idx[0], true, true) == FailAction::retry);
VT_CHECK(s.on_failed(idx[0], true, true) == FailAction::requeue); // 3rd
VT_CHECK_EQ(s.segment_state(idx[0]), SegState::failed);
// the orphaned tail is now assignable again
auto again = s.assign_slot();
VT_REQUIRE(again.has_value());
auto snap = s.snapshot();
SegmentView reborn{};
for (auto &v : snap)
if (v.index == *again)
reborn = v;
VT_CHECK_EQ(reborn.start, 4u * MiB); // resumes where the failed one stopped
VT_CHECK_EQ(reborn.end, 20u * MiB - 1); // its half of the file
}
VT_TEST(seg_failure_without_mirror_always_retries) {
Segmenter s(40 * MiB, 2, true);
auto idx = fill(s, 2);
for (int i = 0; i < 6; ++i)
VT_CHECK(s.on_failed(idx[0], true, /*mirror=*/false) == FailAction::retry);
// a non-connection error also retries regardless of count
VT_CHECK(s.on_failed(idx[1], /*conn=*/false, /*mirror=*/true) == FailAction::retry);
}
VT_TEST(seg_note_connected_resets_failure_count) {
Segmenter s(40 * MiB, 2, true);
auto idx = fill(s, 2);
s.on_failed(idx[0], true, true);
s.on_failed(idx[0], true, true);
s.note_connected(idx[0]);
VT_CHECK(s.on_failed(idx[0], true, true) == FailAction::retry); // count restarted
}
VT_TEST(seg_resume_from_meta_table) {
std::vector<ResumedRange> table = {
{0, 9 * MiB - 1, 9 * MiB}, // fully done
{9 * MiB, 19 * MiB - 1, 3 * MiB}, // partial
{19 * MiB, 40 * MiB - 1, 0}, // untouched
};
Segmenter s(40 * MiB, 8, table, true);
auto snap = s.snapshot();
VT_REQUIRE(snap.size() == 3);
VT_CHECK_EQ(snap[0].state, SegState::complete);
VT_CHECK_EQ(snap[1].completed, 3u * MiB);
VT_CHECK_EQ(s.downloaded(), 12u * MiB);
VT_CHECK(tiles_exactly(s));
// assign hands out the two incomplete ranges before splitting
auto a = s.assign_slot();
auto b = s.assign_slot();
VT_REQUIRE(a && b);
}
VT_TEST(seg_resume_from_bad_table_falls_back) {
std::vector<ResumedRange> gappy = {{0, 4 * MiB - 1, 0}, {8 * MiB, 40 * MiB - 1, 0}};
Segmenter s(40 * MiB, 8, gappy, true);
VT_CHECK(s.snapshot().empty()); // lazy fresh layout
auto idx = fill(s, 8);
VT_CHECK(idx.size() >= 1);
VT_CHECK(tiles_exactly(s));
}
VT_TEST(seg_all_complete_and_downloaded) {
Segmenter s(8 * MiB, 4, true);
auto idx = fill(s, 4);
VT_CHECK(!s.all_complete());
for (auto i : idx) {
std::uint64_t len = s.segment_end(i) - s.segment_start(i) + 1;
s.advance(i, len);
s.set_segment_state(i, SegState::complete);
}
VT_CHECK(s.all_complete());
VT_CHECK_EQ(s.downloaded(), 8u * MiB);
}
// --- the steal path under the sanitizers -----------------------------------------------
VT_TEST(seg_concurrent_steal_and_advance) {
constexpr std::uint64_t total = 64 * MiB;
Segmenter s(total, 8, true);
auto idx = fill(s, 8);
VT_REQUIRE(idx.size() == 8);
std::vector<std::jthread> workers;
for (std::uint32_t w = 0; w < 8; ++w) {
workers.emplace_back([&s, seg = idx[w]]() mutable {
std::uint32_t cur = seg;
for (int guard = 0; guard < 200000; ++guard) {
const std::uint64_t start = s.segment_start(cur);
const std::uint64_t end = s.segment_end(cur); // may shrink under a steal
const std::uint64_t len = end - start + 1;
const std::uint64_t done = s.segment_completed(cur);
if (done >= len) {
auto nxt = s.on_complete(cur, /*may_steal=*/true);
if (!nxt)
return; // nothing left to steal — this worker is finished
cur = *nxt;
continue;
}
s.advance(cur, std::min(done + 128 * 1024, len));
}
});
}
workers.clear(); // join
VT_CHECK(s.all_complete());
VT_CHECK_EQ(s.downloaded(), total);
VT_CHECK(tiles_exactly(s));
}