Cieric/Naut
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions
Naut/apps/swarm/main.c
ookami125 187e8f2db2 swarm: populate per-peer stats for the web UI peers tab 
report_progress now fills peer_stats via engine_peer_list, so the peer
list (ip, progress, dl rate, state) shows again. Mark #12 done.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-22 00:35:23 -04:00

907 lines
36 KiB
C
Raw Blame History

/* naut_swarm — multi-peer download driver built on the torrent-peer engine.
*
* The engine (../torrent-peer, engine.h) owns all peer sockets, the wire
* protocol, the request pipeline, transports (TCP/µTP/MSE), and piece selection.
* This driver:
* - parses the .torrent / magnet and (for magnet) fetches the info dict,
* - discovers peers via HTTP/UDP trackers + DHT (src/discovery),
* - feeds discovered endpoints + a piece-priority vector to the engine,
* - drains delivered blocks, verifies+persists them through naut_download,
* - re-arms hash-failed pieces and reports progress for nautd / the web UI.
*
* usage: naut_swarm <file.torrent|magnet-uri> <output-dir> [<ip:port> ...]
*/
#include "naut/dht.h"
#include "naut/extension.h"
#include "naut/metainfo.h"
#include "naut/storage.h"
#include "naut/piece.h"
#include "naut/tracker.h"
#include "naut/log.h"
#include "naut/system.h"
#include "naut/swarm.h"
#include "engine.h" /* torrent-peer multi-peer engine */
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#define DEFAULT_TARGET_PEERS 80
#define MAX_TARGET_PEERS 512
#define TRACKER_DEFAULT_INTERVAL 1800.0
#define TRACKER_MIN_INTERVAL 60.0
#define TRACKER_FAILURE_RETRY_INTERVAL 300.0
#define DHT_REFRESH_INTERVAL 300.0
#define READY_BATCH 64
typedef struct {
naut_peer_addr addr;
char name[32];
} endpoint_t;
static double now(void) { struct timespec t; clock_gettime(CLOCK_MONOTONIC, &t); return t.tv_sec + t.tv_nsec*1e-9; }
static uint32_t target_peer_count(void) {
const char *env = getenv("NAUT_TARGET_PEERS");
if (!env || !*env) return DEFAULT_TARGET_PEERS;
char *end = NULL;
unsigned long value = strtoul(env, &end, 10);
if (!end || *end || value == 0) return DEFAULT_TARGET_PEERS;
return (uint32_t)NAUT_MIN(value, MAX_TARGET_PEERS);
}
static double tracker_delay_seconds(int32_t interval) {
if (interval <= 0) return TRACKER_DEFAULT_INTERVAL;
if (interval < (int32_t)TRACKER_MIN_INTERVAL)
return TRACKER_MIN_INTERVAL;
return (double)interval;
}
static void random_bytes(uint8_t *output, size_t length) {
int fd = open("/dev/urandom", O_RDONLY);
size_t offset = 0;
while (fd >= 0 && offset < length) {
ssize_t count = read(fd, output + offset, length - offset);
if (count > 0) {
offset += (size_t)count;
} else if (count < 0 && errno == EINTR) {
continue;
} else {
break;
}
}
if (fd >= 0) close(fd);
uint64_t fallback = (uint64_t)(now() * 1e9) ^
(uint64_t)(uintptr_t)output ^
(uint64_t)getpid();
while (offset < length) {
fallback ^= fallback << 13;
fallback ^= fallback >> 7;
fallback ^= fallback << 17;
output[offset++] = (uint8_t)fallback;
}
}
static void emit_event(const naut_swarm_config *config, naut_event_type type,
uint32_t index, const char *message, const char *path) {
if (!config->events) return;
naut_event event = {
.type = type,
.torrent_id = config->torrent_id,
.index = index,
.message = message,
.path = path,
};
naut_event_emit(config->events, &event);
}
static void on_file_complete(void *opaque, uint32_t index, const char *path) {
const naut_swarm_config *config = opaque;
char full_path[PATH_MAX];
const char *event_path = path;
if (path && path[0] != '/') {
int length = snprintf(full_path, sizeof full_path, "%s/%s",
config->output_dir, path);
if (length >= 0 && (size_t)length < sizeof full_path)
event_path = full_path;
}
emit_event(config, NAUT_EVENT_FILE_COMPLETE, index, NULL, event_path);
}
static void on_piece_complete(void *opaque, uint32_t index) {
const naut_swarm_config *config = opaque;
emit_event(config, NAUT_EVENT_PIECE_COMPLETE, index, NULL, NULL);
}
/* --- tracker stats ------------------------------------------------------- */
static uint32_t init_tracker_stats(char *const *trackers,
const uint32_t *tracker_tiers,
size_t num_trackers,
naut_swarm_tracker_stats *stats,
uint32_t capacity) {
uint32_t count = 0;
if (!stats) return 0;
for (size_t i = 0; i < num_trackers && count < capacity; i++) {
naut_swarm_tracker_stats *out = &stats[count++];
memset(out, 0, sizeof(*out));
snprintf(out->url, sizeof out->url, "%s", trackers[i]);
out->tier = tracker_tiers ? (int32_t)tracker_tiers[i] : (int32_t)i;
snprintf(out->status, sizeof out->status, "not contacted");
out->seeds = -1;
out->peers = -1;
out->leeches = -1;
out->downloaded = -1;
}
return count;
}
static naut_swarm_tracker_stats *tracker_stat_for(
naut_swarm_tracker_stats *stats, uint32_t count, const char *url) {
if (!stats || !url) return NULL;
for (uint32_t i = 0; i < count; i++)
if (strcmp(stats[i].url, url) == 0) return &stats[i];
return NULL;
}
static void tracker_set_status(naut_swarm_tracker_stats *tracker,
const char *status, const char *message) {
if (!tracker) return;
snprintf(tracker->status, sizeof tracker->status, "%s",
status ? status : "");
snprintf(tracker->message, sizeof tracker->message, "%s",
message ? message : "");
}
static void snapshot_tracker_stats(naut_swarm_stats *stats,
const naut_swarm_tracker_stats *trackers,
uint32_t tracker_count) {
if (!stats || !trackers) return;
if (tracker_count > NAUT_SWARM_MAX_TRACKER_STATS)
tracker_count = NAUT_SWARM_MAX_TRACKER_STATS;
stats->tracker_count = tracker_count;
for (uint32_t i = 0; i < tracker_count; i++)
stats->tracker_stats[i] = trackers[i];
}
static void snapshot_file_stats(naut_swarm_stats *stats,
const naut_download *download,
const naut_metainfo *metainfo) {
if (!stats || !metainfo || !metainfo->files) return;
uint64_t offset = 0;
uint32_t count = 0;
for (size_t i = 0;
i < metainfo->num_files && count < NAUT_SWARM_MAX_FILE_STATS;
i++) {
const naut_file *file = &metainfo->files[i];
uint64_t size = file->length > 0 ? (uint64_t)file->length : 0;
uint64_t done = 0;
if (download && size > 0) {
uint64_t start = offset;
uint64_t end = offset + size;
uint32_t first = (uint32_t)(start / (uint64_t)metainfo->piece_length);
uint32_t last = (uint32_t)((end - 1) /
(uint64_t)metainfo->piece_length);
for (uint32_t p = first; p <= last; p++) {
if (!naut_download_have(download, p)) continue;
uint64_t piece_start = (uint64_t)p *
(uint64_t)metainfo->piece_length;
uint64_t piece_end = piece_start +
(uint64_t)metainfo->piece_length;
if (piece_end > (uint64_t)metainfo->total_length)
piece_end = (uint64_t)metainfo->total_length;
uint64_t lo = piece_start > start ? piece_start : start;
uint64_t hi = piece_end < end ? piece_end : end;
if (hi > lo) done += hi - lo;
}
}
naut_swarm_file_stats *out = &stats->file_stats[count++];
memset(out, 0, sizeof(*out));
snprintf(out->path, sizeof out->path, "%s",
file->path ? file->path : "");
out->size = size;
out->progress = size ? (double)done / (double)size : 1.0;
if (out->progress > 1.0) out->progress = 1.0;
out->priority = 1;
out->availability = 1.0;
offset += size;
}
stats->file_count = count;
}
/* Fill and deliver a progress snapshot from engine status + naut_download.
* Per-peer detail collapses to the engine's aggregate counts for now; rich
* per-peer rows are a later web-UI feature. */
static void report_progress(const naut_swarm_config *config,
engine *eng, uint32_t torrent_id,
const naut_download *download,
const naut_metainfo *metainfo,
const naut_swarm_tracker_stats *trackers,
uint32_t tracker_count,
double started_at) {
if (!config->on_progress) return;
torrent_status ts;
memset(&ts, 0, sizeof ts);
if (eng) engine_torrent_status(eng, torrent_id, &ts);
uint32_t connecting = ts.peers > ts.peers_connected + ts.peers_failed
? ts.peers - ts.peers_connected - ts.peers_failed : 0;
naut_swarm_stats stats = {
.total_bytes = (uint64_t)metainfo->total_length,
.bytes_done = download ? naut_download_bytes_done(download) : 0,
.total_pieces = metainfo->num_pieces,
.pieces_done = download ? naut_download_pieces_done(download) : 0,
.peers_total = ts.peers,
.peers_connecting = connecting,
.peers_active = ts.peers_connected,
.peers_failed = ts.peers_failed,
.stalled = ts.peers_connected == 0,
.elapsed_seconds = now() - started_at,
};
snapshot_tracker_stats(&stats, trackers, tracker_count);
snapshot_file_stats(&stats, download, metainfo);
if (download)
stats.piece_state_count = (uint32_t)naut_download_piece_states(
download, stats.piece_states, NAUT_SWARM_MAX_PIECE_STATS);
if (eng) {
engine_peer_info peers[NAUT_SWARM_MAX_PEER_STATS];
uint32_t pc = engine_peer_list(eng, torrent_id, peers,
NAUT_SWARM_MAX_PEER_STATS);
stats.peer_count = pc;
for (uint32_t i = 0; i < pc; i++) {
naut_swarm_peer_stats *o = &stats.peer_stats[i];
memset(o, 0, sizeof *o);
snprintf(o->ip, sizeof o->ip, "%.45s", peers[i].ip);
o->port = peers[i].port;
snprintf(o->connection, sizeof o->connection, "BT");
snprintf(o->flags, sizeof o->flags, "%s",
peers[i].state == PEER_STATE_RUNNING
? (peers[i].unchoked ? "D" : "d") : "K");
o->progress = peers[i].num_pieces
? (double)peers[i].have_pieces / (double)peers[i].num_pieces
: 0.0;
o->relevance = o->progress;
o->dlspeed = peers[i].rate_bps;
o->downloaded = peers[i].bytes_received;
}
}
config->on_progress(config->context, &stats);
}
static bool stop_requested(const naut_swarm_config *config) {
return config->should_stop && config->should_stop(config->context);
}
static void service_control(const naut_swarm_config *config,
naut_storage *storage) {
if (config->on_control) config->on_control(config->context, storage);
}
/* Render a full diagnostic snapshot (block assembly + engine piece selection)
* when the caller requests one, and hand the text back through on_dump. Runs on
* the owner thread, the only place engine + download state can be read safely. */
static void service_dump(const naut_swarm_config *config, engine *eng,
uint32_t torrent_id, const naut_download *download) {
if (!config->should_dump || !config->on_dump) return;
if (!config->should_dump(config->context)) return;
char *buf = NULL;
size_t len = 0;
FILE *f = open_memstream(&buf, &len);
if (!f) {
config->on_dump(config->context, "dump: out of memory\n");
return;
}
naut_download_dump(download, f);
engine_dump_torrent(eng, torrent_id, f);
fclose(f);
config->on_dump(config->context, buf ? buf : "dump: render failed\n");
free(buf);
}
static uint8_t *slurp(const char *path, size_t *len) {
FILE *f = fopen(path, "rb"); if (!f) return NULL;
fseek(f, 0, SEEK_END); long n = ftell(f); fseek(f, 0, SEEK_SET);
uint8_t *b = malloc(n);
if (fread(b, 1, n, f) != (size_t)n) { fclose(f); free(b); return NULL; }
fclose(f); *len = (size_t)n; return b;
}
/* --- endpoint collection ------------------------------------------------- */
static bool endpoint_add(endpoint_t **v, size_t *n, size_t *cap,
const naut_peer_addr *addr) {
if (addr->port == 0) return true;
for (size_t i = 0; i < *n; i++)
if ((*v)[i].addr.port == addr->port &&
memcmp((*v)[i].addr.ip, addr->ip, sizeof addr->ip) == 0)
return true;
if (*n == *cap) {
size_t newcap = *cap ? *cap * 2 : 32;
endpoint_t *next = realloc(*v, newcap * sizeof(*next));
if (!next) return false;
*v = next;
*cap = newcap;
}
endpoint_t *ep = &(*v)[(*n)++];
ep->addr = *addr;
snprintf(ep->name, sizeof ep->name, "%u.%u.%u.%u:%u",
addr->ip[0], addr->ip[1], addr->ip[2], addr->ip[3], addr->port);
return true;
}
static bool endpoint_parse(const char *s, naut_peer_addr *out) {
char host[INET_ADDRSTRLEN];
const char *colon = strrchr(s, ':');
if (!colon || colon == s || (size_t)(colon - s) >= sizeof host) return false;
char *end = NULL;
unsigned long port = strtoul(colon + 1, &end, 10);
if (!end || *end || port == 0 || port > UINT16_MAX) return false;
memcpy(host, s, (size_t)(colon - s));
host[colon - s] = 0;
struct in_addr ip;
if (inet_pton(AF_INET, host, &ip) != 1) return false;
memcpy(out->ip, &ip, sizeof out->ip);
out->port = (uint16_t)port;
return true;
}
static bool parse_udp_tracker(const char *url, char *host, size_t hostsz,
uint16_t *port) {
if (strncmp(url, "udp://", 6) != 0) return false;
const char *start = url + 6;
const char *slash = strchr(start, '/');
const char *end = slash ? slash : start + strlen(start);
const char *colon = memchr(start, ':', (size_t)(end - start));
if (!colon || colon == start) return false;
size_t hlen = (size_t)(colon - start);
if (hlen >= hostsz) return false;
char pbuf[16];
size_t plen = (size_t)(end - colon - 1);
if (plen == 0 || plen >= sizeof pbuf) return false;
memcpy(host, start, hlen);
host[hlen] = 0;
memcpy(pbuf, colon + 1, plen);
pbuf[plen] = 0;
char *tail = NULL;
unsigned long parsed = strtoul(pbuf, &tail, 10);
if (!tail || *tail || parsed == 0 || parsed > UINT16_MAX) return false;
*port = (uint16_t)parsed;
return true;
}
static bool discover_trackers(const uint8_t info_hash[20], uint64_t total_length,
char *const *trackers, size_t num_trackers,
const uint32_t *tracker_tiers,
const uint8_t peerid[20],
uint64_t downloaded, uint64_t left,
naut_tracker_event event,
endpoint_t **eps, size_t *neps, size_t *cap,
int32_t *announce_interval,
naut_swarm_tracker_stats *tracker_stats,
uint32_t tracker_count) {
naut_announce_req req;
memset(&req, 0, sizeof req);
memcpy(req.info_hash, info_hash, sizeof req.info_hash);
memcpy(req.peer_id, peerid, sizeof req.peer_id);
req.port = 6881;
req.downloaded = downloaded;
req.left = total_length;
if (left <= total_length) req.left = left;
req.event = event;
req.numwant = 100;
memcpy(&req.key, peerid + 8, sizeof req.key);
if (announce_interval) *announce_interval = 0;
size_t tier_start = 0;
while (tier_start < num_trackers) {
uint32_t tier = tracker_tiers
? tracker_tiers[tier_start] : (uint32_t)tier_start;
size_t tier_end = tier_start + 1;
if (tracker_tiers)
while (tier_end < num_trackers &&
tracker_tiers[tier_end] == tier)
tier_end++;
size_t tier_count = tier_end - tier_start;
uint32_t random = 0;
random_bytes((uint8_t *)&random, sizeof random);
size_t first = tier_count ? random % tier_count : 0;
bool tier_succeeded = false;
for (size_t n = 0; n < tier_count; n++) {
size_t i = tier_start + (first + n) % tier_count;
const char *tracker = trackers[i];
naut_tracker_response response;
memset(&response, 0, sizeof response);
response.seeders = response.leechers = -1;
naut_swarm_tracker_stats *tracker_stat =
tracker_stat_for(tracker_stats, tracker_count, tracker);
naut_err e = NAUT_ERR_INVAL;
if (strncmp(tracker, "http://", 7) == 0) {
char url[4096];
if (naut_tracker_http_url(tracker, &req, url,
sizeof url) != 0)
e = naut_tracker_announce_http(url, &response);
} else if (strncmp(tracker, "udp://", 6) == 0) {
char host[256];
uint16_t port;
if (parse_udp_tracker(tracker, host, sizeof host, &port))
e = naut_tracker_announce_udp(host, port, &req, &response);
} else {
tracker_set_status(tracker_stat, "unsupported",
"unsupported tracker scheme");
NAUT_WARN("tracker scheme unsupported: %s", tracker);
continue;
}
if (e != NAUT_OK) {
tracker_set_status(tracker_stat, "error",
response.failure
? response.failure : "announce failed");
NAUT_WARN("tracker announce failed: %s", tracker);
naut_tracker_response_free(&response);
continue;
}
tier_succeeded = true;
if (announce_interval && response.interval > 0)
*announce_interval = response.interval;
if (tracker_stat) {
tracker_set_status(tracker_stat, "working", "");
tracker_stat->seeds = response.seeders;
tracker_stat->leeches = response.leechers;
tracker_stat->peers = response.num_peers > INT32_MAX
? INT32_MAX : (int32_t)response.num_peers;
tracker_stat->downloaded = -1;
}
NAUT_INFO("tracker %s returned %zu peers",
tracker, response.num_peers);
for (size_t p = 0; p < response.num_peers; p++) {
if (!endpoint_add(eps, neps, cap, &response.peers[p])) {
naut_tracker_response_free(&response);
return false;
}
}
naut_tracker_response_free(&response);
/* Trackers within a tier are alternatives, not a fan-out set.
* Once one accepts the announce, do not load the rest. */
break;
}
if (tier_succeeded) break;
tier_start = tier_end;
}
return true;
}
static bool discover_dht(const uint8_t info_hash[20],
endpoint_t **eps, size_t *neps, size_t *cap) {
static const char *defaults[] = {
"router.bittorrent.com:6881",
"router.utorrent.com:6881",
"dht.transmissionbt.com:6881",
};
const char *const *bootstrap = defaults;
size_t num_bootstrap = NAUT_ARRAY_LEN(defaults);
char *copy = NULL;
const char *custom[32];
const char *env = getenv("NAUT_DHT_BOOTSTRAP");
if (env && *env) {
copy = strdup(env);
if (!copy) return false;
num_bootstrap = 0;
char *save = NULL;
for (char *part = strtok_r(copy, ",", &save);
part && num_bootstrap < NAUT_ARRAY_LEN(custom);
part = strtok_r(NULL, ",", &save))
custom[num_bootstrap++] = part;
bootstrap = custom;
}
naut_peer_addr *peers = NULL;
size_t num_peers = 0;
naut_err e = num_bootstrap
? naut_dht_get_peers(bootstrap, num_bootstrap, info_hash,
&peers, &num_peers)
: NAUT_ERR_INVAL;
free(copy);
if (e != NAUT_OK) {
NAUT_WARN("DHT lookup found no peers");
return true;
}
NAUT_INFO("DHT returned %zu peers", num_peers);
for (size_t i = 0; i < num_peers; i++) {
if (!endpoint_add(eps, neps, cap, &peers[i])) {
free(peers);
return false;
}
}
free(peers);
return true;
}
/* Hand every not-yet-fed endpoint to the engine, which owns the connection. */
static void feed_engine(engine *eng, uint32_t torrent_id,
const endpoint_t *eps, size_t neps, size_t *fed) {
for (size_t i = *fed; i < neps; i++) {
char ip[16];
snprintf(ip, sizeof ip, "%u.%u.%u.%u",
eps[i].addr.ip[0], eps[i].addr.ip[1],
eps[i].addr.ip[2], eps[i].addr.ip[3]);
engine_add_peer(eng, torrent_id, ip, eps[i].addr.port);
}
*fed = neps;
}
naut_err naut_swarm_run(const naut_swarm_config *config) {
if (!config || !config->source || !*config->source ||
!config->output_dir || !*config->output_dir)
return NAUT_ERR_INVAL;
uint8_t peerid[20];
memcpy(peerid, "-NT0001-", 8);
random_bytes(peerid + 8, sizeof peerid - 8);
bool from_magnet = strncmp(config->source, "magnet:?", 8) == 0;
naut_metainfo mi;
memset(&mi, 0, sizeof mi);
naut_magnet magnet;
memset(&magnet, 0, sizeof magnet);
if (from_magnet) {
if (naut_magnet_parse(config->source, &magnet) != NAUT_OK ||
!magnet.has_v1) {
NAUT_ERROR("magnet must contain a v1 btih hash");
naut_magnet_free(&magnet);
return NAUT_ERR_INVAL;
}
} else {
size_t tlen;
uint8_t *tor = slurp(config->source, &tlen);
if (!tor) { NAUT_ERROR("read torrent"); return NAUT_ERR_IO; }
if (naut_metainfo_parse(tor, tlen, &mi) != NAUT_OK) {
NAUT_ERROR("parse torrent");
free(tor);
return NAUT_ERR_PROTO;
}
free(tor);
}
endpoint_t *endpoints = NULL;
size_t neps = 0, epcap = 0;
uint32_t target_peers = target_peer_count();
int32_t tracker_interval = 0;
naut_swarm_tracker_stats tracker_stats[NAUT_SWARM_MAX_TRACKER_STATS];
uint32_t tracker_count = 0;
if (config->num_peers > 0) {
for (size_t i = 0; i < config->num_peers; i++) {
naut_peer_addr addr;
if (!endpoint_parse(config->peers[i], &addr)) {
NAUT_WARN("invalid peer address: %s", config->peers[i]);
continue;
}
if (!endpoint_add(&endpoints, &neps, &epcap, &addr)) {
NAUT_ERROR("out of memory collecting peers");
naut_metainfo_free(&mi);
naut_magnet_free(&magnet);
free(endpoints);
return NAUT_ERR_NOMEM;
}
}
} else {
const uint8_t *hash =
from_magnet ? magnet.infohash_v1 : mi.infohash_v1;
char *const *trackers =
from_magnet ? magnet.trackers : mi.trackers;
const uint32_t *tracker_tiers =
from_magnet ? NULL : mi.tracker_tiers;
size_t num_trackers =
from_magnet ? magnet.num_trackers : mi.num_trackers;
tracker_count = init_tracker_stats(trackers, tracker_tiers,
num_trackers, tracker_stats,
NAUT_SWARM_MAX_TRACKER_STATS);
if (from_magnet) {
if (!discover_trackers(hash, 0, trackers, num_trackers,
tracker_tiers, peerid, 0, 0,
NAUT_TEV_STARTED, &endpoints, &neps, &epcap,
&tracker_interval, tracker_stats,
tracker_count) ||
(neps < target_peers &&
!discover_dht(hash, &endpoints, &neps, &epcap))) {
NAUT_ERROR("out of memory collecting discovered peers");
naut_metainfo_free(&mi);
naut_magnet_free(&magnet);
free(endpoints);
return NAUT_ERR_NOMEM;
}
}
}
/* Magnet: resolve the info dict from a peer before we can size the torrent.
* Neither the engine nor torrent-tracker does BEP-9; use Naut's own fetch. */
if (from_magnet && neps) {
uint8_t *info = NULL;
size_t info_len = 0;
naut_err metadata_error = NAUT_ERR_EMPTY;
for (size_t i = 0; i < neps; i++) {
metadata_error = naut_metadata_fetch(
&endpoints[i].addr, magnet.infohash_v1, peerid,
&info, &info_len);
if (metadata_error == NAUT_OK) break;
NAUT_WARN("metadata fetch from %s failed", endpoints[i].name);
}
if (metadata_error != NAUT_OK ||
naut_metainfo_parse_info(
info, info_len, (const char *const *)magnet.trackers,
magnet.num_trackers, &mi) != NAUT_OK) {
NAUT_ERROR("unable to fetch valid magnet metadata");
free(info);
naut_magnet_free(&magnet);
free(endpoints);
return metadata_error != NAUT_OK ? metadata_error :
NAUT_ERR_PROTO;
}
free(info);
NAUT_INFO("magnet metadata verified: %u pieces, %lld bytes",
mi.num_pieces, (long long)mi.total_length);
}
naut_magnet_free(&magnet);
if (neps == 0) {
if (from_magnet) {
NAUT_ERROR("magnet metadata unavailable: no peers discovered");
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOTFOUND;
}
if (config->num_peers > 0)
NAUT_WARN("no valid peer addresses; torrent stalled");
}
naut_err err;
/* Reopen any files relocated on a prior run in place (no re-download). */
const char **overrides = NULL;
if (config->num_locations && mi.num_files) {
overrides = calloc(mi.num_files, sizeof *overrides);
if (overrides)
for (size_t i = 0; i < config->num_locations; i++) {
const naut_swarm_file_location *loc = &config->locations[i];
if (loc->path && loc->file_index < mi.num_files)
overrides[loc->file_index] = loc->path;
}
}
naut_storage_opts storage_opts = {
.direct_io = getenv("NAUT_DIRECT_IO") != NULL,
.preallocate = true,
.overrides = overrides,
};
naut_storage *st = naut_storage_open_opts(
mi.files, mi.num_files, config->output_dir, &storage_opts, &err);
free(overrides);
if (!st) {
NAUT_ERROR("storage: %s", naut_strerror(err));
naut_metainfo_free(&mi);
free(endpoints);
return err != NAUT_OK ? err : NAUT_ERR_IO;
}
naut_download *d = naut_download_create(&mi, st);
if (!d) {
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOMEM;
}
naut_download_set_file_cb(d, on_file_complete, (void *)config);
naut_download_set_piece_cb(d, on_piece_complete, (void *)config);
err = naut_download_resume(d);
if (err != NAUT_OK) {
NAUT_ERROR("resume scan: %s", naut_strerror(err));
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return err;
}
uint64_t resumed_bytes = naut_download_bytes_done(d);
uint64_t resumed_left = (uint64_t)mi.total_length > resumed_bytes
? (uint64_t)mi.total_length - resumed_bytes : 0;
NAUT_INFO("resume scan complete: %u/%u pieces correct, %llu bytes left",
naut_download_pieces_done(d), mi.num_pieces,
(unsigned long long)resumed_left);
if (!from_magnet && config->num_peers == 0 && !naut_download_complete(d)) {
if (!discover_trackers(mi.infohash_v1, (uint64_t)mi.total_length,
mi.trackers, mi.num_trackers,
mi.tracker_tiers, peerid,
resumed_bytes, resumed_left,
NAUT_TEV_STARTED, &endpoints, &neps, &epcap,
&tracker_interval, tracker_stats,
tracker_count) ||
(neps < target_peers &&
!discover_dht(mi.infohash_v1, &endpoints, &neps, &epcap))) {
NAUT_ERROR("out of memory collecting discovered peers");
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOMEM;
}
if (neps == 0)
NAUT_WARN("tracker and DHT discovery returned no peers; torrent stalled");
}
/* Spin up the engine and register the torrent. The engine owns sockets,
* the wire protocol, the pipeline, transports, and piece selection. */
engine_config ecfg;
memset(&ecfg, 0, sizeof ecfg);
ecfg.encryption = 1; /* offer MSE (RC4) + plaintext: most compatible */
ecfg.fallback = 1; /* retry transport/encryption combos per endpoint */
engine *eng = engine_create(&ecfg);
int32_t tid = eng ? engine_add_torrent(eng, mi.infohash_v1, peerid,
(uint64_t)mi.piece_length,
(uint64_t)mi.total_length,
mi.num_pieces)
: -1;
if (!eng || tid < 0) {
NAUT_ERROR("unable to create download engine");
if (eng) engine_destroy(eng);
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOMEM;
}
uint32_t torrent_id = (uint32_t)tid;
/* Priority vector: skip what resume already verified, request the rest. */
uint8_t *prio = malloc(mi.num_pieces ? mi.num_pieces : 1);
if (!prio) {
engine_destroy(eng);
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOMEM;
}
for (uint32_t p = 0; p < mi.num_pieces; p++)
prio[p] = naut_download_have(d, p) ? 0 : 1;
engine_set_priorities(eng, torrent_id, prio, mi.num_pieces);
size_t fed = 0;
feed_engine(eng, torrent_id, endpoints, neps, &fed);
double t0 = now();
double next_tracker_announce =
t0 + (tracker_interval > 0
? tracker_delay_seconds(tracker_interval)
: TRACKER_FAILURE_RETRY_INTERVAL);
double next_dht_lookup = t0 + DHT_REFRESH_INTERVAL;
naut_err run_error = NAUT_OK;
bool cancelled = false;
NAUT_INFO("swarm: engine started, %zu peers queued, %u pieces, %lld bytes",
neps, mi.num_pieces, (long long)mi.total_length);
emit_event(config, NAUT_EVENT_TORRENT_ADDED, 0, NULL, NULL);
report_progress(config, eng, torrent_id, d, &mi,
tracker_stats, tracker_count, t0);
engine_block blocks[READY_BATCH];
uint64_t applied_rate = UINT64_MAX; /* force first apply */
while (!naut_download_complete(d) && run_error == NAUT_OK) {
service_control(config, st);
service_dump(config, eng, torrent_id, d);
if (stop_requested(config)) { cancelled = true; break; }
/* Apply the live download throttle when it changes. */
if (config->download_rate) {
uint64_t rate = config->download_rate(config->context);
if (rate != applied_rate) {
engine_set_download_rate(eng, rate);
applied_rate = rate;
}
}
/* Top up the swarm from trackers / DHT when it runs thin. */
if (config->num_peers == 0) {
torrent_status ts;
engine_torrent_status(eng, torrent_id, &ts);
if (ts.peers_connected < target_peers) {
double t = now();
if (mi.num_trackers > 0 && t >= next_tracker_announce) {
uint64_t downloaded = naut_download_bytes_done(d);
uint64_t left = (uint64_t)mi.total_length > downloaded
? (uint64_t)mi.total_length - downloaded : 0;
int32_t interval = 0;
if (!discover_trackers(mi.infohash_v1,
(uint64_t)mi.total_length,
mi.trackers, mi.num_trackers,
mi.tracker_tiers, peerid,
downloaded, left, NAUT_TEV_NONE,
&endpoints, &neps, &epcap,
&interval, tracker_stats,
tracker_count)) {
run_error = NAUT_ERR_NOMEM;
break;
}
next_tracker_announce = t + (interval > 0
? tracker_delay_seconds(interval)
: TRACKER_FAILURE_RETRY_INTERVAL);
feed_engine(eng, torrent_id, endpoints, neps, &fed);
}
if (t >= next_dht_lookup) {
if (!discover_dht(mi.infohash_v1, &endpoints, &neps,
&epcap)) {
run_error = NAUT_ERR_NOMEM;
break;
}
next_dht_lookup = t + DHT_REFRESH_INTERVAL;
feed_engine(eng, torrent_id, endpoints, neps, &fed);
}
}
}
engine_wait(eng, 200);
uint32_t n;
while ((n = engine_poll_ready(eng, blocks, READY_BATCH)) > 0) {
for (uint32_t i = 0; i < n; i++) {
engine_block *b = &blocks[i];
uint8_t *data = (uint8_t *)engine_arena_base(eng, b->loop) +
(uint64_t)b->slot * PEER_BLOCK_SIZE;
bool done = false;
naut_err be = naut_download_on_block(d, b->piece, b->begin,
data, b->len, &done);
engine_release_slot(eng, b->loop, b->slot);
if (be == NAUT_ERR_PROTO) {
/* Bad/failed piece: re-arm it for another fetch. */
engine_request_piece(eng, torrent_id, b->piece);
} else if (be != NAUT_OK && be != NAUT_ERR_RANGE) {
run_error = be;
break;
} else if (done) {
engine_set_priority(eng, torrent_id, b->piece, 0);
}
}
if (run_error != NAUT_OK) break;
}
report_progress(config, eng, torrent_id, d, &mi,
tracker_stats, tracker_count, t0);
}
double dt = now() - t0;
bool ok = naut_download_complete(d);
if (ok) {
double mb = (double)mi.total_length / 1e6;
NAUT_INFO("COMPLETE: %u/%u pieces in %.2fs (%.1f MB/s), all SHA-1 verified",
naut_download_pieces_done(d), mi.num_pieces, dt,
dt > 0 ? mb / dt : 0.0);
report_progress(config, eng, torrent_id, d, &mi,
tracker_stats, tracker_count, t0);
emit_event(config, NAUT_EVENT_TORRENT_FINISHED, 0, NULL, NULL);
while (config->keep_alive && !stop_requested(config)) {
service_control(config, st);
service_dump(config, eng, torrent_id, d);
usleep(100000);
}
} else {
if (run_error != NAUT_OK)
NAUT_ERROR("swarm stopped: %s", naut_strerror(run_error));
NAUT_ERROR("INCOMPLETE: %u/%u pieces",
naut_download_pieces_done(d), mi.num_pieces);
}
service_control(config, st);
naut_storage_sync(st);
engine_destroy(eng);
free(prio);
free(endpoints);
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
if (ok) return NAUT_OK;
if (cancelled) return NAUT_ERR_AGAIN;
return run_error != NAUT_OK ? run_error : NAUT_ERR_IO;
}
Reference in a new issue View git blame Copy permalink