#ifndef HYPEROVERLAY_SCHEDULER_H
#define HYPEROVERLAY_SCHEDULER_H
// Who fetches which chunk, from whom, and when to give up on them.
//
// This is the multi-source half of the transfer manager, and it is deliberately a
// PURE STATE MACHINE: no disk, no sockets, no clock of its own. It is told which
// chunks are wanted, which peers are available and what time it is; it answers
// with batches of chunk indices to request. `transfer.h` owns the bytes and
// `protocol.h` owns the wire.
//
// That split is not tidiness. Every behaviour here is a timing behaviour —
// windows, stalls, hotswap, watchdogs — and the only way to assert a timing
// behaviour without a race is to hand the machine its clock and drive it to an
// exact state. A scheduler that called now() would be untestable in precisely the
// places it is most likely to be wrong.
//
// THE ONE INVARIANT: a chunk is requested from EXACTLY ONE peer at a time. Chunks
// are content-addressed, so any holder can serve any chunk and the naive version
// asks everyone — which works, wastes N times the bandwidth, and is the reason
// this file exists rather than a loop that asks every peer for everything.
//
// Modelled on hypercore's block replicator, which the reference follows closely
// (lib/chunk-scheduler.js). The constants below are its constants; they are tuned
// DOWN from hypercore's [16,512] because overlay chunks are far larger than
// hypercore blocks — 16 KB to 4 MB, against 4 KB.
#include <algorithm>
#include <cstdint>
#include <functional>
#include <map>
#include <set>
#include <vector>
#include "transport.h"
namespace hyperoverlay {
namespace sched {
// Per-peer request window, in chunks.
static constexpr uint32_t INFLIGHT_MIN = 4;
static constexpr uint32_t INFLIGHT_MAX = 64;
// At or under this RTT the window stays at the minimum: the link is close enough
// that a small number of outstanding requests already keeps it busy.
static constexpr uint32_t SCALE_LATENCY_MS = 50;
struct Options {
// >0 pins the per-peer window (tests, and callers who know their link);
// 0 scales it from each peer's RTT.
uint32_t cap = 0;
// NO-PROGRESS timeout, not a total one. The reference records that a fixed
// overall timeout was tried and aborted every large transfer mid-stream — at
// any bandwidth, a multi-GB file eventually exceeds any constant. What is
// actually pathological is silence, so that is what is measured.
uint64_t idle_timeout_ms = 30000;
// A chunk outstanding to one peer for this long, while another peer sits idle,
// may be reassigned. Negative disables hotswap entirely.
int64_t hotswap_stall_ms = 2000;
};
class ChunkScheduler {
public:
// Emit a batch of chunk requests to one peer. Batched rather than per chunk
// because that is what the wire carries — one chunkNeed with many indices.
std::function<void(Transport *, const std::vector<uint64_t> &)> send_need;
// Best-effort "stop serving these" to a peer whose work was reassigned. Purely
// a bandwidth saving: a duplicate arrival is handled idempotently anyway, so
// this never has to succeed.
std::function<void(Transport *, const std::vector<uint64_t> &)> send_cancel;
void configure(const Options &o) { opt_ = o; }
const Options &options() const { return opt_; }
// Arm the watchdog before any holder has answered. A request that draws NO
// reply at all — every holder silent, or none of them actually has the file —
// must still end. The reference arms its idle timer when the fetch is created,
// not when the first chunk list arrives, and a fetch armed only on the first
// response is a fetch that hangs forever in exactly the case it should not.
void arm(uint64_t now_ms) {
armed_ = true;
last_progress_ms_ = now_ms;
}
// Begin, with the set of chunks still wanted (a resumed partial contributes
// far fewer than all of them).
void start(const std::vector<uint64_t> &wanted, uint64_t now_ms) {
needed_.clear();
for (uint64_t i : wanted) needed_.insert(i);
inflight_.clear();
inflight_at_.clear();
for (PeerSlot &s : peers_) s.load = 0;
accepted_by_.clear();
last_progress_ms_ = now_ms;
started_ = true;
stalled_ = false;
}
// A peer that has answered with a chunk list and can therefore serve.
void add_peer(Transport *t, uint64_t now_ms) {
if (!t) return;
if (!slot_for(t)) peers_.push_back({ t, 0 });
// A new holder is forward progress in its own right: it changes what is
// possible even before it delivers a byte.
last_progress_ms_ = now_ms;
}
// A peer went away. Its outstanding chunks return to the pool so someone else
// can take them — the difference between a transfer that survives a
// disconnection and one that hangs holding indices nobody will ever send.
void remove_peer(Transport *t, uint64_t now_ms) {
for (auto it = inflight_.begin(); it != inflight_.end();) {
if (it->second == t) {
inflight_at_.erase(it->first);
it = inflight_.erase(it); // still in needed_; simply unassigned
} else {
++it;
}
}
peers_.erase(std::remove_if(peers_.begin(), peers_.end(),
[&](const PeerSlot &s) { return s.peer == t; }),
peers_.end());
if (peers_.empty() && !needed_.empty()) {
// Distinct from a stall: nobody is coming. Reported separately so the
// caller can fail fast instead of waiting out the idle window.
abandoned_ = true;
return;
}
assign(now_ms);
}
// A page of a large chunk list arrived and was buffered. Receiving it IS
// forward progress — a chunk map big enough to need paging can take longer to
// stream than the idle window, and a transfer that has not started yet would
// otherwise be declared stalled while its holder is busy sending it.
void note_progress(uint64_t now_ms) { last_progress_ms_ = now_ms; }
// A chunk arrived and was accepted (verified, on disk).
//
// The assignment is cleared REGARDLESS of who was holding it, which is a
// deliberate departure from the reference. There, a chunk delivered by a peer
// that is no longer its assigned holder — the ordinary aftermath of a hotswap,
// since the slow peer usually still sends its copy — leaves the assignment
// standing, and the transfer only completes because the new holder eventually
// delivers too. The chunk is on disk and verified; nobody needs to be holding
// it any more.
void on_accepted(Transport *from, uint64_t index, uint64_t now_ms) {
auto it = inflight_.find(index);
if (it != inflight_.end()) {
if (PeerSlot *s = slot_for(it->second)) { if (s->load) s->load--; }
inflight_.erase(it);
inflight_at_.erase(index);
}
// A duplicate of a chunk already satisfied is real and harmless, but it is
// not a contribution — counting it would overstate how many peers actually
// served, which is the one thing multi-source has to be able to prove.
if (needed_.erase(index) == 0) return;
accepted_by_[from]++;
last_progress_ms_ = now_ms; // bytes landed: the watchdog resets
assign(now_ms);
}
// A chunk arrived and was rejected, or a request will not be answered. The
// chunk stays wanted; only the assignment is released, so it can go to someone
// else. NOT progress — a peer that reliably serves garbage must not hold the
// watchdog open forever.
void on_rejected(Transport *from, uint64_t index, uint64_t now_ms) {
release(index, from);
assign(now_ms);
}
// Hand out work. Called after every event; idempotent when there is nothing to
// do.
void assign(uint64_t now_ms) {
if (!started_ || stalled_ || abandoned_) return;
// Batches keyed in connection order too, so the emitted sequence of
// send_need calls is reproducible.
std::vector<std::pair<Transport *, std::vector<uint64_t>>> batches;
auto batch_for = [&](Transport *p) -> std::vector<uint64_t> & {
for (auto &b : batches) if (b.first == p) return b.second;
batches.push_back({ p, {} });
return batches.back().second;
};
for (size_t pi = 0; pi < peers_.size(); pi++) {
Transport *peer = peers_[pi].peer;
const uint32_t window = max_inflight(peer);
if (peers_[pi].load >= window) continue;
uint32_t slots = window - peers_[pi].load;
for (uint64_t index : needed_) {
if (!slots) break;
if (inflight_.find(index) != inflight_.end()) continue; // someone has it
inflight_[index] = peer;
inflight_at_[index] = now_ms;
peers_[pi].load++;
batch_for(peer).push_back(index);
slots--;
}
// Spare capacity and nothing unassigned left for it: take over the work
// of a peer that has gone quiet, rather than idling while one slow
// holder owns the tail.
if (slots) hotswap(peer, slots, batch_for, now_ms);
}
for (auto &b : batches)
if (!b.second.empty() && send_need) send_need(b.first, b.second);
}
// Drive the watchdog. Returns false once the transfer has been silent for
// longer than the idle window — the caller fails it.
bool tick(uint64_t now_ms) {
if (stalled_ || abandoned_) return false;
if (!started_ && !armed_) return true;
if (started_ && needed_.empty()) return true;
if (now_ms - last_progress_ms_ > opt_.idle_timeout_ms) { stalled_ = true; return false; }
// Hotswap is time-driven, so it needs the clock too — otherwise a chunk
// stalls only if some OTHER event happens to call assign().
if (started_) assign(now_ms);
return true;
}
bool done() const { return started_ && needed_.empty() && inflight_.empty(); }
bool stalled() const { return stalled_; }
bool abandoned() const { return abandoned_; }
size_t remaining() const { return needed_.size(); }
size_t inflight_count() const { return inflight_.size(); }
size_t peer_count() const { return peers_.size(); }
uint32_t load_of(Transport *t) const {
const PeerSlot *s = slot_for(t);
return s ? s->load : 0;
}
Transport *holder_of(uint64_t index) const {
auto it = inflight_.find(index);
return it == inflight_.end() ? nullptr : it->second;
}
size_t accepted_from(Transport *t) const {
auto it = accepted_by_.find(t);
return it == accepted_by_.end() ? 0 : it->second;
}
// How many peers actually contributed. The proof that "multi-source" is a
// behaviour and not just a configuration.
size_t contributing_peers() const { return accepted_by_.size(); }
size_t hotswaps() const { return hotswaps_; }
// The window this peer would get right now. Exposed so the RTT policy can be
// asserted directly rather than inferred from how many requests appeared.
uint32_t max_inflight(Transport *peer) const {
if (opt_.cap) return opt_.cap;
const uint32_t rtt = peer ? peer->rtt_ms() : 0;
// No RTT signal: a modest fixed window. Not the minimum — that would
// serialise a link we simply know nothing about.
if (rtt == 0) return std::min(INFLIGHT_MAX, INFLIGHT_MIN * 3);
if (rtt <= SCALE_LATENCY_MS) return INFLIGHT_MIN;
// Grow with latency, shrink toward a fair share as holders join: two peers
// each pulling a full window from a shared link help nobody.
const double share = std::min(1.0, 2.0 / (double)std::max<size_t>(1, peers_.size()));
const double scale = ((double)rtt / (double)SCALE_LATENCY_MS) * share;
const double scaled = (double)INFLIGHT_MIN * scale;
const uint32_t capped = (uint32_t)(scaled < (double)INFLIGHT_MAX ? (scaled + 0.5) : (double)INFLIGHT_MAX);
return std::max(INFLIGHT_MIN, capped);
}
private:
void release(uint64_t index, Transport *from) {
auto it = inflight_.find(index);
// Only the CURRENT holder may release it. After a hotswap the old holder
// may still deliver its copy; letting that late arrival clear the new
// holder's slot would corrupt the accounting for a chunk still in flight.
if (it == inflight_.end() || it->second != from) return;
PeerSlot *s = slot_for(from);
if (s && s->load > 0) s->load--;
inflight_.erase(it);
inflight_at_.erase(index);
}
// Reassign chunks that have been outstanding to OTHER peers beyond the stall
// window, so one slow holder cannot own the tail of a transfer while faster
// peers sit idle. Mirrors hypercore's hotswap queue.
template <typename BatchFor>
void hotswap(Transport *peer, uint32_t slots, BatchFor &&batch_for, uint64_t now_ms) {
if (opt_.hotswap_stall_ms < 0) return;
const uint64_t stall = (uint64_t)opt_.hotswap_stall_ms;
std::vector<uint64_t> cands;
for (const auto &kv : inflight_) {
if (kv.second == peer) continue;
auto at = inflight_at_.find(kv.first);
if (at == inflight_at_.end()) continue;
if (now_ms - at->second < stall) continue;
cands.push_back(kv.first);
}
if (cands.empty()) return;
// Most-stalled first.
std::sort(cands.begin(), cands.end(), [&](uint64_t a, uint64_t b) {
return inflight_at_[a] < inflight_at_[b];
});
std::vector<std::pair<Transport *, std::vector<uint64_t>>> cancels;
auto cancel_for = [&](Transport *p) -> std::vector<uint64_t> & {
for (auto &c : cancels) if (c.first == p) return c.second;
cancels.push_back({ p, {} });
return cancels.back().second;
};
for (uint64_t index : cands) {
if (!slots) break;
Transport *loser = inflight_[index];
inflight_[index] = peer;
inflight_at_[index] = now_ms;
if (PeerSlot *w = slot_for(peer)) w->load++;
if (PeerSlot *l = slot_for(loser)) { if (l->load > 0) l->load--; }
batch_for(peer).push_back(index);
cancel_for(loser).push_back(index);
hotswaps_++;
slots--;
}
for (auto &c : cancels)
if (send_cancel) send_cancel(c.first, c.second);
}
// Peers are held in CONNECTION ORDER, not in a std::map keyed by pointer.
// Which peer is offered work first decides who gets which chunks, and ordering
// by pointer value makes that depend on where the allocator happened to put
// things — different on every run, and untestable. The reference iterates an
// insertion-ordered Map; this is that.
struct PeerSlot { Transport *peer; uint32_t load; };
std::vector<PeerSlot> peers_;
PeerSlot *slot_for(Transport *t) {
for (PeerSlot &s : peers_) if (s.peer == t) return &s;
return nullptr;
}
const PeerSlot *slot_for(Transport *t) const {
for (const PeerSlot &s : peers_) if (s.peer == t) return &s;
return nullptr;
}
Options opt_;
bool started_ = false, armed_ = false, stalled_ = false, abandoned_ = false;
// Sorted, so requests go out in file order and the receiver's hash frontier
// advances as early as it can rather than waiting on a late low chunk.
std::set<uint64_t> needed_;
std::map<uint64_t, Transport *> inflight_;
std::map<uint64_t, uint64_t> inflight_at_;
std::map<Transport *, size_t> accepted_by_;
uint64_t last_progress_ms_ = 0;
size_t hotswaps_ = 0;
};
} // namespace sched
} // namespace hyperoverlay
#endif // HYPEROVERLAY_SCHEDULER_H