#include "overlay_node.h"
#include <algorithm>
#include <chrono>
#include <cstdio>
#include <godot_cpp/core/class_db.hpp>
#include <godot_cpp/variant/utility_functions.hpp>
#include "debug_log.h"
#include "hyperswarm_peer.h"
using namespace godot;
// A monotonic millisecond clock, read ONLY here. liboverlay never calls now() —
// it is handed the time — so this is the single place the library's notion of
// time comes from, and the single place to change if it ever needs to be faked
// outside the C++ tests.
static uint64_t now_ms() {
using namespace std::chrono;
return (uint64_t)duration_cast<milliseconds>(steady_clock::now().time_since_epoch()).count();
}
HyperOverlay::HyperOverlay() {}
HyperOverlay::~HyperOverlay() {
close();
}
void HyperOverlay::_bind_methods() {
ClassDB::bind_method(D_METHOD("open", "storage_dir"), &HyperOverlay::open);
ClassDB::bind_method(D_METHOD("close"), &HyperOverlay::close);
ClassDB::bind_method(D_METHOD("attach", "hyperswarm_peer"), &HyperOverlay::attach);
ClassDB::bind_method(D_METHOD("detach"), &HyperOverlay::detach);
ClassDB::bind_method(D_METHOD("peer_count"), &HyperOverlay::peer_count);
ClassDB::bind_method(D_METHOD("content_hash_of", "file_path"), &HyperOverlay::content_hash_of);
ClassDB::bind_method(D_METHOD("is_open"), &HyperOverlay::is_open);
ClassDB::bind_method(D_METHOD("serve", "content_hash", "file_path"), &HyperOverlay::serve);
ClassDB::bind_method(D_METHOD("fetch", "content_hash", "dest_path"), &HyperOverlay::fetch);
ClassDB::bind_method(D_METHOD("cancel", "content_hash", "discard"), &HyperOverlay::cancel);
ClassDB::bind_method(D_METHOD("poll"), &HyperOverlay::poll);
ClassDB::bind_method(D_METHOD("active_fetches"), &HyperOverlay::active_fetches);
ClassDB::bind_method(D_METHOD("describe"), &HyperOverlay::describe);
// Emitted from poll(), on the main thread. Never from the worker: a signal
// crossing threads is how a GDScript handler ends up touching the scene tree
// from somewhere Godot does not expect it.
ADD_SIGNAL(MethodInfo("content_fetched",
PropertyInfo(Variant::STRING, "content_hash"),
PropertyInfo(Variant::STRING, "path")));
ADD_SIGNAL(MethodInfo("content_failed",
PropertyInfo(Variant::STRING, "content_hash"),
PropertyInfo(Variant::STRING, "reason")));
ADD_SIGNAL(MethodInfo("content_progress",
PropertyInfo(Variant::STRING, "content_hash"),
PropertyInfo(Variant::INT, "received_bytes"),
PropertyInfo(Variant::INT, "total_bytes")));
}
bool HyperOverlay::open(const String &storage_dir) {
if (open_) return true;
storage_dir_ = std::string(storage_dir.utf8().get_data());
index_.reset(new hyperoverlay::index::FileIndex());
if (!index_->open(storage_dir_ + "/overlay-index.log")) {
UtilityFunctions::push_error("HyperOverlay: cannot open index under " + storage_dir);
index_.reset();
return false;
}
if (index_->truncated_bytes()) {
// The process is a game; it gets killed. A torn tail is expected and is
// reported rather than hidden, because "entries vanished" is otherwise a
// mystery later.
UtilityFunctions::push_warning("HyperOverlay: discarded " +
String::num_int64((int64_t)index_->truncated_bytes()) +
" bytes of torn index tail from an earlier kill");
}
// What this instance can serve is whatever it registered, resolved through the
// index rather than a map that dies with the process.
proto_.reset(new hyperoverlay::Protocol(index_.get(),
[this](const std::string &hash, hyperoverlay::ContentLocation &out) {
const hyperoverlay::index::FileMeta *m =
index_->get_file(hyperoverlay::index::FileIndex::content_register_key(hash));
if (!m || m->linkname.empty()) return false;
out.path = m->linkname; // where the bytes actually are
out.size = m->size;
return true;
}));
proto_->on_complete = [this](const std::string &hash, const std::string &path) {
Event e; e.kind = Event::FETCHED; e.hash = hash; e.detail = path;
push_event(e);
};
proto_->on_failed = [this](const std::string &hash, const std::string &why) {
Event e; e.kind = Event::FAILED; e.hash = hash; e.detail = why;
push_event(e);
};
started_ms_ = now_ms();
proto_->tick(started_ms_);
open_ = true;
return true;
}
void HyperOverlay::close() {
if (!open_) return;
detach();
proto_.reset();
if (index_) index_->close();
index_.reset();
open_ = false;
}
bool HyperOverlay::attach(Object *hyperswarm_peer) {
if (!open_) {
UtilityFunctions::push_error("HyperOverlay: attach() before open()");
return false;
}
HyperswarmMultiplayerPeer *peer = Object::cast_to<HyperswarmMultiplayerPeer>(hyperswarm_peer);
if (!peer) {
UtilityFunctions::push_error("HyperOverlay: attach() needs a HyperswarmMultiplayerPeer");
return false;
}
// detach() FIRST, and while the previous peer is still held: it tears channels
// down on that peer's loop thread, which has to exist to run them.
detach();
peer_ = Ref<HyperswarmMultiplayerPeer>(peer);
transport_ = peer->transport_internal();
// Both hooks fire ON THE UDX THREAD, which is where a channel must be created
// and where the Protocol lives. Nothing here touches Godot.
//
// Through the setter, so peers that are ALREADY meshed are replayed to us.
// attach() is normally called after the swarm is up — you wait for peers before
// you want to fetch from them — and registering for future connections only
// meant those peers never existed as far as the overlay was concerned.
transport_->set_mux_callbacks(
[this](int32_t id, Protomux *mux) { udx_add_peer(id, mux); },
[this](int32_t id) { udx_remove_peer(id); });
return true;
}
void HyperOverlay::detach() {
if (!transport_) {
peer_.unref();
return;
}
UdxTransport *t = transport_;
transport_ = nullptr;
// Clear on the loop thread rather than from here: the UDX thread may be inside
// these callbacks right now. Queued first, and the teardown below is queued
// second, so FIFO ordering guarantees no new peer arrives after the sweep.
t->set_mux_callbacks(nullptr, nullptr);
// Tear the channels down on the thread that owns them, and wait, so no
// callback can land on a half-destroyed object afterwards.
std::mutex m;
std::condition_variable cv;
bool done = false;
t->run_on_loop([this, &m, &cv, &done]() {
for (auto &kv : peers_) {
if (proto_) proto_->remove_peer(kv.second);
delete kv.second;
}
peers_.clear();
{ std::lock_guard<std::mutex> lk(m); done = true; }
cv.notify_one();
});
std::unique_lock<std::mutex> lk(m);
cv.wait_for(lk, std::chrono::seconds(2), [&done]() { return done; });
// Only now: nothing of ours points into that peer any more, so it may go.
peer_.unref();
}
String HyperOverlay::content_hash_of(const String &file_path) const {
const std::string path(file_path.utf8().get_data());
uint64_t size = 0;
if (!hyperoverlay::io::file_size(path, size)) return String();
// Streamed, so hashing a file larger than RAM is not a different code path
// from hashing a small one.
hyperoverlay::io::File f;
if (!f.open_read(path)) return String();
hyperoverlay::cdc::StreamingHasher hasher;
hasher.init(size);
std::vector<uint8_t> buf(1u << 20);
uint64_t off = 0;
while (off < size) {
const size_t want = (size_t)std::min<uint64_t>(buf.size(), size - off);
if (!f.read_at(off, buf.data(), want)) return String();
hasher.update(buf.data(), want);
off += want;
}
return String(hasher.digest().c_str());
}
int HyperOverlay::peer_count() const {
return proto_ ? (int)proto_->peer_count() : 0;
}
void HyperOverlay::udx_add_peer(int32_t id, Protomux *mux) {
DHT_LOG("[overlay] on_mux_ready peer=%d mux=%p proto=%p\n",
(int)id, (void *)mux, (void *)proto_.get());
if (!proto_ || !mux) return;
if (peers_.count(id)) return;
// The pool hook goes in through the CONSTRUCTOR, and covers every way a channel
// can open: already paired when we attach, paired later by the remote, or built
// again after the remote rejected the first one because its own overlay had not
// attached yet. Assigning it afterwards missed the first of those — the channel
// is already open before the assignment runs — and there is no version of this
// that a ready() check on the next line covers, because the later re-openings
// have no line to check on. add_peer ignores a peer it already holds.
hyperswarm::OverlayTransport *ot =
new hyperswarm::OverlayTransport(mux, proto_.get(), std::to_string(id),
[this](hyperswarm::OverlayTransport *t) {
DHT_LOG("[overlay] channel open, joining pool\n");
if (proto_) proto_->add_peer(t);
});
if (ot->slot_mismatch()) {
// Protomux handed back a slot that is not the registration index, so every
// frame on this connection would land in the wrong handler. Refuse loudly
// rather than move misrouted bytes.
UtilityFunctions::push_error("HyperOverlay: slot-order mismatch on peer " +
String::num_int64(id) + " — refusing the channel");
// It may already have joined the pool from the constructor, and the pool
// must not be left holding a pointer we are about to free.
proto_->remove_peer(ot);
delete ot;
return;
}
peers_[id] = ot;
DHT_LOG("[overlay] peer=%d channel %s\n", (int)id,
ot->ready() ? "ready at attach" : "not ready yet, waiting on pair()");
}
void HyperOverlay::udx_remove_peer(int32_t id) {
auto it = peers_.find(id);
if (it == peers_.end()) return;
if (proto_) proto_->remove_peer(it->second); // hands its chunks back to the pool
delete it->second;
peers_.erase(it);
}
// Everything the game asks for becomes a command. The worker owns the Protocol
// outright, so nothing else ever touches it — no lock around liboverlay, and no
// question about which thread a given call came from.
// Commands arrive on Godot's main thread and are marshalled onto the UDX thread,
// because that is where the Protocol lives and where sockets may be touched.
// Before attach() there is no loop to marshal to, so they run inline — safe,
// because nothing else is touching the Protocol yet.
void HyperOverlay::on_loop(std::function<void()> fn) {
if (transport_) transport_->run_on_loop(std::move(fn));
else fn();
}
void HyperOverlay::serve(const String &content_hash, const String &file_path) {
if (!open_) return;
const std::string hash(content_hash.utf8().get_data());
const std::string path(file_path.utf8().get_data());
on_loop([this, hash, path]() {
// Registered by CONTENT hash with the disk path in linkname, so the answer
// survives a restart without re-scanning anything.
hyperoverlay::index::FileMeta m;
m.contentHash = hash;
m.linkname = path;
uint64_t sz = 0;
if (!hyperoverlay::io::file_size(path, sz)) {
push_event({ Event::FAILED, hash, "serve: no such file: " + path, 0, 0 });
return;
}
m.size = sz;
index_->put_file(hyperoverlay::index::FileIndex::content_register_key(hash), m);
// Precompute the chunk map HERE — when the game registers an asset — rather
// than when a peer first asks for it. Same work either way, but this way it
// is not sitting in front of the first byte of a transfer, and it is the
// difference between a one-off cost at load and a stall mid-session.
if (!index_->has_chunk_map_by_hash(hash)) {
const std::vector<hyperoverlay::index::ChunkRef> map =
hyperoverlay::Protocol::build_chunk_map(path, sz);
if (!map.empty()) index_->put_chunk_map_by_hash(hash, map);
}
});
}
void HyperOverlay::fetch(const String &content_hash, const String &dest_path) {
if (!open_) return;
const std::string hash(content_hash.utf8().get_data());
const std::string path(dest_path.utf8().get_data());
on_loop([this, hash, path]() { proto_->request(hash, path); });
}
void HyperOverlay::cancel(const String &content_hash, bool discard) {
if (!open_) return;
const std::string hash(content_hash.utf8().get_data());
on_loop([this, hash, discard]() {
push_event({ Event::FAILED, hash, "cancel is not implemented yet", 0, 0 });
});
}
void HyperOverlay::push_event(const Event &e) {
std::lock_guard<std::mutex> lk(evt_mutex_);
if (events_.size() >= MAX_EVENTS) {
// Drop PROGRESS under pressure; never a terminal event. A caller that
// misses a progress tick sees a coarser bar; one that misses a FETCHED
// waits forever.
if (e.kind == Event::PROGRESS) { dropped_progress_++; return; }
for (auto it = events_.begin(); it != events_.end(); ++it) {
if (it->kind == Event::PROGRESS) { events_.erase(it); dropped_progress_++; break; }
}
}
events_.push_back(e);
}
void HyperOverlay::poll() {
if (!open_) return;
// Advance liboverlay's clock. It never calls now() itself, so without this no
// watchdog fires and no hotswap happens — the library would be correct and
// completely inert. Marshalled, because the Protocol belongs to the UDX thread.
const uint64_t t = now_ms();
on_loop([this, t]() { if (proto_) proto_->tick(t); });
std::deque<Event> batch;
{
std::lock_guard<std::mutex> lk(evt_mutex_);
batch.swap(events_);
}
for (const Event &e : batch) {
switch (e.kind) {
case Event::FETCHED:
emit_signal("content_fetched", String(e.hash.c_str()), String(e.detail.c_str()));
break;
case Event::FAILED:
emit_signal("content_failed", String(e.hash.c_str()), String(e.detail.c_str()));
break;
case Event::PROGRESS:
emit_signal("content_progress", String(e.hash.c_str()),
(int64_t)e.received, (int64_t)e.total);
break;
}
}
}
void HyperOverlay::_process(double) {
poll();
}
int HyperOverlay::active_fetches() const {
// Deliberately reports the queue, not the Protocol's download map: that map
// belongs to the UDX thread, and reading it from here is exactly the race this
// structure exists to avoid.
std::lock_guard<std::mutex> lk(evt_mutex_);
return (int)events_.size();
}
String HyperOverlay::describe() const {
std::lock_guard<std::mutex> lk(evt_mutex_);
return String("HyperOverlay(open=") + (open_ ? "true" : "false") +
", attached=" + (transport_ ? "true" : "false") +
", storage=" + String(storage_dir_.c_str()) +
", queued_events=" + String::num_int64((int64_t)events_.size()) +
", dropped_progress=" + String::num_int64((int64_t)dropped_progress_) + ")";
}