// Multi-source scheduling: who fetches which chunk, from whom, and when to stop
// waiting.
//
// The scheduler is a pure state machine with an injected clock, and this suite
// exists because of that: every behaviour it has is a TIMING behaviour, and the
// only way to assert one without a race is to drive the machine to an exact state
// and look. No disk, no sockets, no sleeping.
//
// Build:
// g++ -O2 -std=c++17 -I liboverlay/include \
// -o /tmp/ho_sched liboverlay/test/test_scheduler.cpp
#include <cstdio>
#include <set>
#include <string>
#include <vector>
#include "overlay/scheduler.h"
using namespace hyperoverlay;
static int passed = 0, failed = 0;
static void check(bool c, const std::string &what) {
if (c) { passed++; printf(" ok %s\n", what.c_str()); }
else { failed++; printf(" FAIL %s\n", what.c_str()); }
}
// A peer that carries nothing: the scheduler never sends through it, it only
// keys state and answers for its RTT.
struct FakePeer : Transport {
std::string name;
uint32_t rtt = 0;
FakePeer(const char *n, uint32_t r = 0) : name(n), rtt(r) {}
bool send(MsgId, const std::vector<uint8_t> &) override { return true; }
const char *peer_label() const override { return name.c_str(); }
uint32_t rtt_ms() const override { return rtt; }
};
struct Emitted {
std::vector<std::pair<Transport *, std::vector<uint64_t>>> needs, cancels;
void wire(sched::ChunkScheduler &s) {
s.send_need = [this](Transport *p, const std::vector<uint64_t> &v) { needs.push_back({ p, v }); };
s.send_cancel = [this](Transport *p, const std::vector<uint64_t> &v) { cancels.push_back({ p, v }); };
}
std::vector<uint64_t> to(Transport *p) const {
std::vector<uint64_t> out;
for (const auto &n : needs) if (n.first == p) out.insert(out.end(), n.second.begin(), n.second.end());
return out;
}
size_t total_requested() const {
size_t n = 0;
for (const auto &e : needs) n += e.second.size();
return n;
}
void clear() { needs.clear(); cancels.clear(); }
};
static std::vector<uint64_t> range(uint64_t n) {
std::vector<uint64_t> v;
for (uint64_t i = 0; i < n; i++) v.push_back(i);
return v;
}
int main() {
printf("hyper-overlay C++ chunk scheduler\n");
// ---- one peer per chunk, and the window is respected --------------------
//
// The invariant the whole file exists for. Chunks are content-addressed, so
// asking every holder for everything WORKS — it just costs N times the
// bandwidth the layer was built to save.
{
printf("\n-- assignment\n");
FakePeer a("a"), b("b"), c("c");
sched::ChunkScheduler s;
sched::Options o; o.cap = 4;
s.configure(o);
Emitted em; em.wire(s);
s.start(range(30), 0);
s.add_peer(&a, 0); s.add_peer(&b, 0); s.add_peer(&c, 0);
s.assign(0);
check(em.total_requested() == 12, "three peers at a window of 4 request 12 chunks, not 30 (" +
std::to_string(em.total_requested()) + ")");
std::set<uint64_t> seen;
bool dup = false;
for (const auto &e : em.needs)
for (uint64_t i : e.second) if (!seen.insert(i).second) dup = true;
check(!dup, "no chunk is requested from more than one peer");
check(s.load_of(&a) == 4 && s.load_of(&b) == 4 && s.load_of(&c) == 4,
"each peer is loaded to exactly its window");
check(s.inflight_count() == 12 && s.remaining() == 30,
"twelve are outstanding; all thirty are still wanted until they land");
// Peers are offered work in CONNECTION order, so the split is reproducible
// rather than dependent on where the allocator put things.
check(em.to(&a) == std::vector<uint64_t>({ 0, 1, 2, 3 }), "the first peer takes the first chunks");
check(em.to(&c) == std::vector<uint64_t>({ 8, 9, 10, 11 }), "the third takes the next ones along");
}
// ---- the per-peer window scales with RTT --------------------------------
//
// A far peer needs more requests outstanding to keep the link busy; a near one
// does not. Asserted directly rather than inferred from how many requests
// appeared, so the policy is pinned and not just its consequences.
{
printf("\n-- the RTT window\n");
sched::ChunkScheduler s; // cap 0 => scaled
FakePeer none("none", 0), near("near", 10), at("at", 50), far("far", 200), absurd("absurd", 100000);
s.add_peer(&none, 0);
check(s.max_inflight(&none) == 12, "no RTT signal gets a modest fixed window, not the minimum");
sched::ChunkScheduler s2;
s2.add_peer(&near, 0);
check(s2.max_inflight(&near) == sched::INFLIGHT_MIN, "a near peer stays at the minimum");
sched::ChunkScheduler s3;
s3.add_peer(&at, 0);
check(s3.max_inflight(&at) == sched::INFLIGHT_MIN, "and so does one exactly at the threshold");
sched::ChunkScheduler s4;
s4.add_peer(&far, 0);
check(s4.max_inflight(&far) == 16, "a far peer alone gets a wider window (16)");
// Fair share: the same peer asks for less once other holders are pulling on
// the same link.
sched::ChunkScheduler s5;
FakePeer f2("f2", 200), f3("f3", 200), f4("f4", 200);
s5.add_peer(&far, 0); s5.add_peer(&f2, 0); s5.add_peer(&f3, 0); s5.add_peer(&f4, 0);
check(s5.max_inflight(&far) == 8, "with four holders the same peer's window halves (8)");
sched::ChunkScheduler s6;
s6.add_peer(&absurd, 0);
check(s6.max_inflight(&absurd) == sched::INFLIGHT_MAX, "an absurd RTT is capped, not unbounded");
sched::ChunkScheduler s7;
sched::Options o; o.cap = 7; s7.configure(o);
s7.add_peer(&far, 0);
check(s7.max_inflight(&far) == 7, "an explicit cap overrides the policy entirely");
}
// ---- several peers actually contribute ----------------------------------
{
printf("\n-- multi-source\n");
FakePeer a("a"), b("b");
sched::ChunkScheduler s;
sched::Options o; o.cap = 2; s.configure(o);
s.configure(o);
Emitted em; em.wire(s);
s.start(range(8), 0);
s.add_peer(&a, 0); s.add_peer(&b, 0);
s.assign(0);
// Deliver everything each peer was asked for, alternating as it arrives.
uint64_t t = 1;
for (int round = 0; round < 8 && !s.done(); round++) {
std::vector<std::pair<Transport *, std::vector<uint64_t>>> pending = em.needs;
em.clear();
for (const auto &e : pending)
for (uint64_t i : e.second) s.on_accepted(e.first, i, t++);
}
check(s.done(), "the transfer completes");
check(s.contributing_peers() == 2, "and both peers actually served chunks");
check(s.accepted_from(&a) + s.accepted_from(&b) == 8, "every chunk is accounted to a peer");
}
// ---- a peer that leaves hands its work back -----------------------------
{
printf("\n-- a peer disappears\n");
FakePeer a("a"), b("b");
sched::ChunkScheduler s;
sched::Options o; o.cap = 4; s.configure(o);
Emitted em; em.wire(s);
s.start(range(8), 0);
s.add_peer(&a, 0); s.add_peer(&b, 0);
s.assign(0);
check(s.inflight_count() == 8, "all eight are outstanding across two peers");
const std::vector<uint64_t> a_had = em.to(&a);
em.clear();
s.remove_peer(&a, 10);
check(s.remaining() == 8, "nothing was delivered, so everything is still wanted");
check(s.peer_count() == 1, "the departed peer is gone");
// The chunks return to the POOL — unassigned but still wanted. They are not
// re-requested on the spot, because the survivor is already at its window;
// handing them over anyway would make the window meaningless exactly when
// the link is under most pressure. (A first version of this test assumed
// immediate reassignment and was simply wrong about how a window works.)
bool all_unassigned = true;
for (uint64_t i : a_had) if (s.holder_of(i) != nullptr) all_unassigned = false;
check(!a_had.empty() && all_unassigned, "its chunks are released, held by nobody");
check(s.inflight_count() == 4, "only the survivor's own four remain outstanding");
check(em.to(&b).empty(), "and nothing is re-requested while the survivor is at its window");
// As soon as capacity frees, the orphaned work is picked up.
const std::vector<uint64_t> b_had = em.to(&b);
em.clear();
s.on_accepted(&b, 4, 20); // one of b's own chunks lands
const std::vector<uint64_t> next = em.to(&b);
check(next.size() == 1, "a freed slot draws exactly one more request");
// Guarded indexing throughout: under a mutated build these vectors can be
// empty, and an unguarded [0] turns "the control fired" into "the control
// segfaulted", which reports nothing at all.
check(!next.empty() && !a_had.empty() && next[0] == a_had[0],
"and it is the lowest orphaned chunk, in file order");
check(!a_had.empty() && s.holder_of(a_had[0]) == &b, "now assigned to the survivor");
}
// ---- the last peer leaving is not a stall, it is abandonment ------------
{
printf("\n-- nobody left\n");
FakePeer a("a");
sched::ChunkScheduler s;
sched::Options o; o.cap = 2; s.configure(o);
Emitted em; em.wire(s);
s.start(range(4), 0);
s.add_peer(&a, 0);
s.assign(0);
s.remove_peer(&a, 5);
check(s.abandoned(), "losing the last holder with work outstanding is reported at once");
check(!s.stalled(), "and is distinct from a stall — there is nothing to wait for");
}
// ---- the watchdog measures SILENCE, not elapsed time --------------------
//
// The reference records that a fixed OVERALL timeout was tried and aborted
// every large transfer mid-stream: at any bandwidth a big enough file exceeds
// any constant. What is actually pathological is no progress.
{
printf("\n-- the idle watchdog\n");
FakePeer a("a");
sched::ChunkScheduler s;
sched::Options o; o.cap = 4; o.idle_timeout_ms = 1000; o.hotswap_stall_ms = -1;
s.configure(o);
Emitted em; em.wire(s);
s.start(range(100), 0);
s.add_peer(&a, 0);
s.assign(0);
check(s.tick(900), "a quiet moment inside the window is fine");
check(s.tick(1000), "and so is the boundary itself");
// Progress re-arms it, so a long transfer that keeps delivering never trips.
s.on_accepted(&a, 0, 1000);
check(s.tick(1900), "a chunk landing resets the clock");
for (uint64_t t = 2000; t <= 20000; t += 900) {
s.on_accepted(&a, (t / 900) % 90 + 1, t);
if (!s.tick(t + 800)) break;
}
check(!s.stalled(), "so twenty seconds of steady delivery never trips a one-second idle window");
// Silence does trip it.
const uint64_t quiet_from = 21000;
s.on_accepted(&a, 95, quiet_from);
check(s.tick(quiet_from + 1000), "still fine at exactly the window");
check(!s.tick(quiet_from + 1001), "one millisecond past it, the transfer is declared stalled");
check(s.stalled(), "and stays that way");
}
// ---- hotswap: one slow holder must not own the tail ---------------------
{
printf("\n-- hotswap\n");
FakePeer slow("slow"), fast("fast");
sched::ChunkScheduler s;
sched::Options o; o.cap = 1; o.hotswap_stall_ms = 2000; o.idle_timeout_ms = 100000;
s.configure(o);
Emitted em; em.wire(s);
s.start(range(1), 0);
s.add_peer(&slow, 0);
s.assign(0);
check(s.holder_of(0) == &slow, "the only holder takes the chunk");
s.add_peer(&fast, 0);
s.assign(0);
check(s.holder_of(0) == &slow, "a fresh idle peer does NOT steal it immediately");
check(s.hotswaps() == 0, "nothing has stalled yet");
em.clear();
check(s.tick(1999), "just inside the stall window, still nothing");
check(s.hotswaps() == 0, "no steal yet");
s.tick(2001);
check(s.holder_of(0) == &fast, "past the window, the idle peer takes it over");
check(s.hotswaps() == 1, "and that is recorded as a hotswap");
check(em.to(&fast) == std::vector<uint64_t>({ 0 }), "the new holder is asked for it");
check(em.cancels.size() == 1 && !em.cancels.empty() && em.cancels[0].first == &slow &&
em.cancels[0].second == std::vector<uint64_t>({ 0 }),
"and the slow peer is told to stop — best effort, purely to save bandwidth");
// The slow peer usually still delivers. Its bytes are perfectly good.
s.on_accepted(&slow, 0, 2100);
check(s.done(), "a late delivery from the peer that was swapped out still completes the transfer");
check(s.inflight_count() == 0,
"and clears the assignment, rather than leaving the new holder marked as still fetching");
}
// ---- hotswap can be switched off ----------------------------------------
{
printf("\n-- hotswap disabled\n");
FakePeer slow("slow"), fast("fast");
sched::ChunkScheduler s;
sched::Options o; o.cap = 1; o.hotswap_stall_ms = -1; o.idle_timeout_ms = 100000;
s.configure(o);
Emitted em; em.wire(s);
s.start(range(1), 0);
s.add_peer(&slow, 0);
s.assign(0);
s.add_peer(&fast, 0);
s.tick(1000000);
check(s.holder_of(0) == &slow, "with hotswap off the chunk stays with the slow peer forever");
check(s.hotswaps() == 0 && em.cancels.empty(), "no steal, no cancel");
}
// ---- a rejection frees the assignment but is not progress ---------------
{
printf("\n-- rejections\n");
FakePeer a("a"), b("b");
sched::ChunkScheduler s;
sched::Options o; o.cap = 1; o.idle_timeout_ms = 1000; o.hotswap_stall_ms = -1;
s.configure(o);
Emitted em; em.wire(s);
s.start(range(2), 0);
s.add_peer(&a, 0);
s.assign(0);
const std::vector<uint64_t> first = em.to(&a);
check(!first.empty(), "the peer was asked for something to reject");
const uint64_t held = first.empty() ? 0 : first[0];
em.clear();
s.on_rejected(&a, held, 100);
check(s.remaining() == 2, "a rejected chunk is still wanted");
check(s.holder_of(held) == &a, "and is re-offered (here, to the only peer there is)");
// A peer that reliably serves garbage must not hold the watchdog open.
check(!s.tick(1101), "rejections alone do not count as progress, so the watchdog still fires");
// And a rejection from a peer that is NOT the current holder must not free
// the real holder's slot.
sched::ChunkScheduler s2;
sched::Options o2; o2.cap = 1; o2.hotswap_stall_ms = -1; s2.configure(o2);
Emitted em2; em2.wire(s2);
s2.start(range(1), 0);
s2.add_peer(&a, 0); s2.add_peer(&b, 0);
s2.assign(0);
Transport *holder = s2.holder_of(0);
Transport *other = (holder == &a) ? (Transport *)&b : (Transport *)&a;
const uint32_t before = s2.load_of(holder);
s2.on_rejected(other, 0, 10);
check(s2.load_of(holder) == before && s2.holder_of(0) == holder,
"a stale rejection from a non-holder leaves the real assignment alone");
}
printf("\n %d assertions pass, %d fail\n", passed, failed);
printf("%s\n", failed == 0 ? "PASS" : "FAIL");
return failed == 0 ? 0 : 1;
}