#include "content.h"
#include <algorithm>
#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <mutex>
namespace swarmd {
// The single place the library's notion of time comes from. liboverlay never
// calls now() — it is handed the time — so this is also the single place to fake
// if it ever needs faking.
static uint64_t now_ms() {
using namespace std::chrono;
return (uint64_t)duration_cast<milliseconds>(steady_clock::now().time_since_epoch()).count();
}
// How often a running fetch reports. Fast enough for a progress bar, slow enough
// that a 100 MB transfer does not spend the socket on telling us about itself.
static constexpr uint64_t PROGRESS_INTERVAL_MS = 250;
bool Content::open(const std::string &storage_dir, UdxTransport *transport) {
if (open_) return true;
storage_dir_ = storage_dir;
t_ = transport;
index_.reset(new hyperoverlay::index::FileIndex());
if (!index_->open(storage_dir_ + "/overlay-index.log")) {
fprintf(stderr, "[swarmd] content: cannot open the index under %s\n",
storage_dir_.c_str());
index_.reset();
return false;
}
if (index_->truncated_bytes()) {
// A daemon gets killed, not shut down. A torn tail is expected; it is
// reported rather than hidden, because "entries vanished" is otherwise a
// mystery months later.
fprintf(stderr, "[swarmd] content: discarded %llu bytes of torn index tail "
"from an earlier kill\n",
(unsigned long long)index_->truncated_bytes());
}
// What this daemon can serve is whatever a client 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;
out.size = m->size;
return true;
}));
proto_->on_complete = [this](const std::string &hash, const std::string &path) {
Event e; e.kind = Event::DONE; e.hash = hash; e.detail = path;
_push(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(e);
};
proto_->tick(now_ms());
open_ = true;
return true;
}
void Content::close() {
if (!open_) return;
open_ = false;
// ⛔ WAIT FOR THE LOOP, DO NOT JUST POST TO IT.
//
// Every command — serve, fetch, cancel, the progress sample — is a lambda
// queued onto the UDX loop that captures `this` and reads index_ and proto_.
// Posting the teardown and then resetting them here races those: a serve
// posted a moment earlier runs afterwards, against an index that has been
// destroyed. Measured, as an intermittent SEGV on the UDX thread inside
// Content::serve's lambda, in two runs out of three.
//
// run_on_loop is FIFO, so a teardown posted last runs last — but only if we
// wait for it before freeing anything. That is what HyperOverlay::detach
// does (src/overlay_node.cpp) and this is the same hazard.
if (t_) {
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);
// Bounded: a transport already stopped will never run this, and hanging
// a shutdown on it would be worse than the leak of not running it.
cv.wait_for(lk, std::chrono::seconds(2), [&done]() { return done; });
}
// Only now is nothing on the loop pointing at either of these.
proto_.reset();
if (index_) index_->close();
index_.reset();
t_ = nullptr;
}
std::string Content::hash_of_file(const std::string &path) {
uint64_t size = 0;
if (!hyperoverlay::io::file_size(path, size)) return std::string();
hyperoverlay::io::File f;
if (!f.open_read(path)) return std::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 std::string();
hasher.update(buf.data(), want);
off += want;
}
return hasher.digest();
}
void Content::serve(const std::string &hash, const std::string &path) {
if (!open_ || !t_) return;
t_->run_on_loop([this, hash, path]() {
uint64_t sz = 0;
if (!hyperoverlay::io::file_size(path, sz)) {
Event e; e.kind = Event::FAILED; e.hash = hash;
e.detail = "serve: no such file: " + path;
_push(e);
return;
}
hyperoverlay::index::FileMeta m;
m.contentHash = hash;
m.linkname = path; // a LOCATOR: the bytes are never read whole
m.size = sz;
index_->put_file(hyperoverlay::index::FileIndex::content_register_key(hash), m);
// Precompute the chunk map HERE — when the file is registered — rather
// than when a peer first asks. Same work either way, but this way it is
// not sitting in front of the first byte of a transfer.
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 Content::fetch(const std::string &hash, const std::string &dest_path) {
if (!open_ || !t_) return;
t_->run_on_loop([this, hash, dest_path]() { proto_->request(hash, dest_path); });
}
void Content::cancel(const std::string &hash, bool discard) {
if (!open_ || !t_) return;
t_->run_on_loop([this, hash, discard]() {
if (proto_->cancel(hash, discard)) return;
Event e; e.kind = Event::FAILED; e.hash = hash;
e.detail = "cancel: no such fetch";
_push(e);
});
}
void Content::add_peer(int32_t peer_id, Protomux *mux) {
// UDX THREAD ONLY.
if (!open_ || !proto_ || !mux) return;
if (peers_.count(peer_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 rebuilt after the remote's overlay came up second. Assigning it
// on the next line misses the first of those, because the channel can be
// fully open before the constructor returns.
hyperswarm::OverlayTransport *ot = new hyperswarm::OverlayTransport(
mux, proto_.get(), std::to_string(peer_id),
[this](hyperswarm::OverlayTransport *t) {
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.
fprintf(stderr, "[swarmd] content: slot-order mismatch on peer %d — "
"refusing the overlay channel\n", (int)peer_id);
proto_->remove_peer(ot); // it may already have joined from the ctor
delete ot;
return;
}
peers_[peer_id] = ot;
}
void Content::remove_peer(int32_t peer_id) {
// UDX THREAD ONLY.
auto it = peers_.find(peer_id);
if (it == peers_.end()) return;
if (proto_) proto_->remove_peer(it->second); // hands its chunks back
delete it->second;
peers_.erase(it);
}
void Content::_push(const Event &e) {
std::lock_guard<std::mutex> lk(m_);
if (events_.size() >= MAX_EVENTS) {
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 Content::poll(std::vector<Event> &out) {
if (!open_ || !t_) return;
const uint64_t t = now_ms();
const bool want_progress = (t - last_progress_ms_) >= PROGRESS_INTERVAL_MS;
if (want_progress) last_progress_ms_ = t;
// Both of these read Protocol state, so both go to the thread that owns it.
t_->run_on_loop([this, t, want_progress]() {
if (!proto_) return;
proto_->tick(t);
if (!want_progress) return;
for (const std::string &hash : proto_->active_downloads()) {
uint64_t bytes = 0;
size_t got = 0, total = 0;
if (!proto_->progress(hash, bytes, got, total)) continue;
Event e;
e.kind = Event::PROGRESS;
e.hash = hash;
e.received = (uint64_t)got;
e.total = (uint64_t)total;
_push(e);
}
});
std::deque<Event> batch;
{
std::lock_guard<std::mutex> lk(m_);
batch.swap(events_);
}
out.assign(batch.begin(), batch.end());
}
} // namespace swarmd