Cieric/Naut
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions
Naut/apps/swarm/main.c
ookami125 6dc711cf57 Limit active pieces and expose file list
2026-06-17 02:23:15 -04:00

1558 lines
58 KiB
C
Raw Blame History

/* naut_swarm — Phase 4 gate: download from a SWARM of peers concurrently.
*
* A poll()-based multi-socket driver around the same sans-IO peer codec and the
* multi-peer engine (rarest-first + bounded endgame duplication). Peers can be
* supplied explicitly for deterministic testing, or discovered from the
* torrent's HTTP/UDP trackers.
*
* 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/peer.h"
#include "naut/tracker.h"
#include "naut/bitfield.h"
#include "naut/log.h"
#include "naut/pipeline.h"
#include "naut/system.h"
#include "naut/swarm.h"
#include "naut/worker.h"
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <poll.h>
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#define REQUEST_TIMEOUT 15.0
#define CONNECT_TIMEOUT_MS 5000
#define EXT_RESERVED 0x0000000000100000ULL
#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
typedef struct {
uint32_t piece, begin, length;
double sent_at;
} req_t;
typedef struct {
naut_peer_addr addr;
char name[32];
} endpoint_t;
typedef struct {
int fd;
naut_peer_addr addr;
char name[40];
uint8_t peer_id[20];
bool have_peer_id;
bool connecting;
uint8_t *rbuf; size_t rcap, rlen;
bool hs_done, peer_choking;
naut_bitfield have;
req_t *inflight; size_t nflight, cflight;
uint64_t blocks_received;
uint64_t bytes_received;
uint64_t rate_last_bytes;
double rate_last_at;
double download_rate;
naut_ext_handshake extensions;
uint64_t pex_received;
naut_pipeline pipeline;
bool dead, availability_removed;
} peer_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);
}
static void peer_client_label(const peer_t *peer, char out[64]) {
if (!peer || !peer->have_peer_id) {
snprintf(out, 64, "Unknown");
return;
}
if (peer->peer_id[0] == '-' && peer->peer_id[7] == '-') {
char code[3] = {
(char)peer->peer_id[1],
(char)peer->peer_id[2],
0,
};
char version[5];
memcpy(version, peer->peer_id + 3, 4);
version[4] = 0;
for (size_t i = 0; i < sizeof version - 1; i++)
if (!isprint((unsigned char)version[i])) version[i] = '?';
snprintf(out, 64, "%s %s", code, version);
return;
}
char id[21];
memcpy(id, peer->peer_id, sizeof peer->peer_id);
id[20] = 0;
for (size_t i = 0; i < sizeof id - 1; i++)
if (!isprint((unsigned char)id[i])) id[i] = '.';
snprintf(out, 64, "%s", id);
}
static void peer_flags(const peer_t *peer, char out[16]) {
size_t n = 0;
if (peer && !peer->peer_choking && n + 1 < 16) out[n++] = 'D';
if (peer && (peer->extensions.ut_pex || peer->pex_received) && n + 1 < 16)
out[n++] = 'X';
out[n] = 0;
}
static void peer_update_rate(peer_t *peer, double sampled_at) {
if (!peer || peer->dead) return;
if (peer->rate_last_at <= 0) {
peer->rate_last_at = sampled_at;
peer->rate_last_bytes = peer->bytes_received;
return;
}
double dt = sampled_at - peer->rate_last_at;
if (dt < 0.25) return;
uint64_t delta = peer->bytes_received - peer->rate_last_bytes;
double rate = (double)delta / dt;
peer->download_rate = peer->download_rate <= 0.0
? rate : peer->download_rate * 0.7 + rate * 0.3;
peer->rate_last_at = sampled_at;
peer->rate_last_bytes = peer->bytes_received;
}
static void snapshot_peer_stats(naut_swarm_stats *stats,
peer_t *peers, int npeers,
uint32_t total_pieces) {
if (!stats || !peers || npeers <= 0) return;
double sampled_at = now();
for (int i = 0; i < npeers &&
stats->peer_count < NAUT_SWARM_MAX_PEER_STATS; i++) {
peer_t *peer = &peers[i];
if (peer->dead || !peer->hs_done) continue;
peer_update_rate(peer, sampled_at);
naut_swarm_peer_stats *out =
&stats->peer_stats[stats->peer_count++];
snprintf(out->ip, sizeof out->ip, "%u.%u.%u.%u",
peer->addr.ip[0], peer->addr.ip[1],
peer->addr.ip[2], peer->addr.ip[3]);
out->port = peer->addr.port;
peer_client_label(peer, out->client);
snprintf(out->connection, sizeof out->connection, "TCP");
peer_flags(peer, out->flags);
size_t have = peer->have.words ? naut_bitfield_count(&peer->have) : 0;
double ratio = total_pieces
? (double)have / (double)total_pieces : 0.0;
if (ratio > 1.0) ratio = 1.0;
out->progress = ratio;
out->relevance = ratio;
out->dlspeed = peer->download_rate;
out->upspeed = 0.0;
out->downloaded = peer->bytes_received;
out->uploaded = 0;
}
}
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;
}
static void report_progress(const naut_swarm_config *config,
const naut_download *download,
const naut_metainfo *metainfo,
uint32_t peers_total, uint32_t peers_connecting,
uint32_t peers_active, uint32_t peers_failed,
peer_t *peers, int npeers,
const naut_swarm_tracker_stats *trackers,
uint32_t tracker_count,
double started_at) {
if (!config->on_progress) return;
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 = peers_total,
.peers_connecting = peers_connecting,
.peers_active = peers_active,
.peers_failed = peers_failed,
.elapsed_seconds = now() - started_at,
};
snapshot_peer_stats(&stats, peers, npeers, metainfo->num_pieces);
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);
}
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);
}
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;
}
static bool send_all(int fd, const void *p, size_t n) {
const uint8_t *b = p;
while (n) { ssize_t w = send(fd, b, n, MSG_NOSIGNAL);
if (w < 0) { if (errno == EINTR) continue; return false; } b += w; n -= (size_t)w; }
return true;
}
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;
}
static bool peer_reserve_inflight(peer_t *p) {
if (p->nflight == p->cflight) {
size_t cap = p->cflight ? p->cflight * 2 : 64;
req_t *v = realloc(p->inflight, cap * sizeof(*v));
if (!v) return false;
p->inflight = v;
p->cflight = cap;
}
return true;
}
static void peer_add_inflight(peer_t *p, uint32_t piece, uint32_t begin,
uint32_t length) {
p->inflight[p->nflight].piece = piece;
p->inflight[p->nflight].begin = begin;
p->inflight[p->nflight].length = length;
p->inflight[p->nflight].sent_at = now();
p->nflight++;
}
static bool peer_del_inflight(peer_t *p, uint32_t piece, uint32_t begin,
req_t *removed) {
for (size_t i = 0; i < p->nflight; i++)
if (p->inflight[i].piece == piece && p->inflight[i].begin == begin) {
if (removed) *removed = p->inflight[i];
p->inflight[i] = p->inflight[--p->nflight];
return true;
}
return false;
}
static bool peer_has_request(void *ctx, uint32_t piece, uint32_t begin) {
peer_t *p = ctx;
for (size_t i = 0; i < p->nflight; i++)
if (p->inflight[i].piece == piece && p->inflight[i].begin == begin)
return true;
return false;
}
/* return remaining inflight blocks to the picker (choke / disconnect) */
static void peer_release(naut_download *d, peer_t *p) {
for (size_t i = 0; i < p->nflight; i++)
naut_download_unrequest(d, p->inflight[i].piece, p->inflight[i].begin);
p->nflight = 0;
}
static void peer_drop(naut_download *d, peer_t *p) {
if (!p->availability_removed) {
naut_download_remove_bitfield(d, &p->have);
p->availability_removed = true;
}
peer_release(d, p);
if (p->fd >= 0) close(p->fd);
p->fd = -1;
p->connecting = false;
p->dead = true;
}
static bool refill_one(naut_download *d, peer_t *p) {
if (p->dead || !p->hs_done || p->peer_choking ||
p->nflight >= naut_pipeline_depth(&p->pipeline))
return false;
if (!peer_reserve_inflight(p)) {
peer_drop(d, p);
return false;
}
uint32_t i, b, l;
if (!naut_download_pick_for_peer(d, &p->have, peer_has_request, p,
&i, &b, &l))
return false;
uint8_t req[17];
naut_peer_msg_request(req, i, b, l);
if (!send_all(p->fd, req, sizeof req)) {
naut_download_unrequest(d, i, b);
peer_drop(d, p);
return false;
}
peer_add_inflight(p, i, b, l);
return true;
}
static void expire_requests(naut_download *d, peer_t *p, double t) {
size_t i = 0;
while (i < p->nflight) {
if (t - p->inflight[i].sent_at < REQUEST_TIMEOUT) {
i++;
continue;
}
naut_download_unrequest(d, p->inflight[i].piece, p->inflight[i].begin);
p->inflight[i] = p->inflight[--p->nflight];
}
}
static int connect_start(const endpoint_t *ep, bool *connected) {
*connected = false;
int fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd < 0) return -1;
int flags = fcntl(fd, F_GETFL, 0);
if (flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0) {
close(fd);
return -1;
}
struct sockaddr_in a; memset(&a, 0, sizeof a);
a.sin_family = AF_INET;
a.sin_port = htons(ep->addr.port);
memcpy(&a.sin_addr, ep->addr.ip, sizeof ep->addr.ip);
if (connect(fd, (struct sockaddr *)&a, sizeof a) == 0) {
*connected = true;
} else if (errno != EINPROGRESS) {
close(fd);
return -1;
}
return fd;
}
static bool connect_finish(int fd) {
int error = 0;
socklen_t length = sizeof error;
if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &length) != 0 ||
error != 0)
return false;
int flags = fcntl(fd, F_GETFL, 0);
if (flags < 0 || fcntl(fd, F_SETFL, flags & ~O_NONBLOCK) != 0)
return false;
int one = 1; setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
return true;
}
static bool peer_start(naut_download *download, const naut_metainfo *metainfo,
const uint8_t peerid[20], const endpoint_t *endpoint,
peer_t *peer, int fd) {
peer->fd = fd;
peer->addr = endpoint->addr;
snprintf(peer->name, sizeof peer->name, "%s", endpoint->name);
peer->connecting = false;
peer->peer_choking = true;
naut_pipeline_init(&peer->pipeline, NAUT_BLOCK, 4, 1024, 32);
peer->rcap = 1 << 18;
peer->rbuf = malloc(peer->rcap);
if (!peer->rbuf ||
naut_bitfield_init(&peer->have, metainfo->num_pieces) != NAUT_OK) {
free(peer->rbuf);
peer->rbuf = NULL;
close(fd);
peer->fd = -1;
peer->dead = true;
return false;
}
uint8_t handshake[NAUT_HANDSHAKE_LEN];
naut_peer_handshake_build(handshake, metainfo->infohash_v1, peerid,
EXT_RESERVED);
uint8_t interested[5];
naut_peer_msg_simple(interested, NAUT_MSG_INTERESTED);
uint8_t *extension = NULL;
size_t extension_length = 0;
naut_err extension_error = naut_ext_build_handshake(
NAUT_EXT_UT_METADATA, NAUT_EXT_UT_PEX, 0, 0, &extension,
&extension_length);
bool sent = extension_error == NAUT_OK &&
send_all(fd, handshake, sizeof handshake) &&
send_all(fd, extension, extension_length) &&
send_all(fd, interested, sizeof interested);
free(extension);
if (!sent) {
peer_drop(download, peer);
return false;
}
return true;
}
static bool ensure_peer_capacity(peer_t **peers, struct pollfd **pfd,
int **idx_map, int *capacity, int want) {
if (want <= *capacity) return true;
int next = *capacity > 0 ? *capacity : 1;
while (next < want) next *= 2;
peer_t *next_peers = realloc(*peers, (size_t)next * sizeof(*next_peers));
if (!next_peers) return false;
for (int i = *capacity; i < next; i++) {
memset(&next_peers[i], 0, sizeof(next_peers[i]));
next_peers[i].fd = -1;
next_peers[i].dead = true;
}
struct pollfd *next_pfd =
realloc(*pfd, ((size_t)next + 1) * sizeof(*next_pfd));
if (!next_pfd) {
*peers = next_peers;
return false;
}
int *next_idx = realloc(*idx_map, ((size_t)next + 1) * sizeof(*next_idx));
if (!next_idx) {
*peers = next_peers;
*pfd = next_pfd;
return false;
}
*peers = next_peers;
*pfd = next_pfd;
*idx_map = next_idx;
*capacity = next;
return true;
}
static bool peer_addr_seen(const peer_t *peers, int npeers,
const naut_peer_addr *addr);
static uint32_t peer_count_active(const peer_t *peers, int npeers,
bool include_connecting) {
uint32_t count = 0;
for (int i = 0; i < npeers; i++) {
if (peers[i].dead) continue;
if (peers[i].connecting && !include_connecting) continue;
count++;
}
return count;
}
static bool connect_one_candidate(const naut_swarm_config *config,
naut_download *download,
const naut_metainfo *metainfo,
const uint8_t peerid[20],
peer_t **peers, int *npeers,
int *peer_capacity,
struct pollfd **pfd, int **idx_map,
endpoint_t endpoint,
uint32_t *failed) {
if (!ensure_peer_capacity(peers, pfd, idx_map, peer_capacity,
*npeers + 1))
return false;
peer_t *peer = &(*peers)[*npeers];
memset(peer, 0, sizeof(*peer));
peer->fd = -1;
peer->addr = endpoint.addr;
snprintf(peer->name, sizeof peer->name, "%s", endpoint.name);
bool connected = false;
int fd = connect_start(&endpoint, &connected);
if (fd < 0) {
if (failed) (*failed)++;
return true;
}
peer->fd = fd;
(*npeers)++;
if (connected) {
if (!connect_finish(fd) ||
!peer_start(download, metainfo, peerid, &endpoint, peer, fd)) {
if (peer->fd >= 0) close(peer->fd);
peer->fd = -1;
peer->dead = true;
if (failed) (*failed)++;
return true;
}
emit_event(config, NAUT_EVENT_PEER_CONNECTED, 0, peer->name, NULL);
return true;
}
peer->connecting = true;
return true;
}
static bool connect_pending_peers(const naut_swarm_config *config,
naut_download *download,
const naut_metainfo *metainfo,
const uint8_t peerid[20],
peer_t **peers, int *npeers,
int *peer_capacity,
struct pollfd **pfd, int **idx_map,
endpoint_t *pending, size_t *npending,
uint32_t target, uint32_t *failed) {
while (*npending > 0 &&
peer_count_active(*peers, *npeers, true) < target) {
endpoint_t endpoint = pending[--(*npending)];
if (peer_addr_seen(*peers, *npeers, &endpoint.addr)) continue;
if (!connect_one_candidate(config, download, metainfo, peerid,
peers, npeers, peer_capacity, pfd, idx_map,
endpoint, failed))
return false;
}
return true;
}
/* process all complete messages currently buffered for peer p */
static void cancel_block(naut_download *d, peer_t *peers, int npeers,
peer_t *source, uint32_t piece, uint32_t begin) {
for (int i = 0; i < npeers; i++) {
peer_t *p = &peers[i];
if (p == source || p->dead) continue;
req_t old;
if (!peer_del_inflight(p, piece, begin, &old)) continue;
uint8_t msg[17];
naut_peer_msg_cancel(msg, old.piece, old.begin, old.length);
if (!send_all(p->fd, msg, sizeof msg)) peer_drop(d, p);
}
}
static void cancel_piece(naut_download *d, peer_t *peers, int npeers,
uint32_t piece) {
for (int i = 0; i < npeers; i++) {
peer_t *p = &peers[i];
size_t j = 0;
while (j < p->nflight) {
req_t old = p->inflight[j];
if (old.piece != piece) {
j++;
continue;
}
p->inflight[j] = p->inflight[--p->nflight];
naut_download_unrequest(d, old.piece, old.begin);
if (!p->dead) {
uint8_t msg[17];
naut_peer_msg_cancel(msg, old.piece, old.begin, old.length);
if (!send_all(p->fd, msg, sizeof msg)) {
peer_drop(d, p);
break;
}
}
}
}
}
static void merge_bitfield(naut_download *d, const naut_metainfo *mi,
peer_t *p, const uint8_t *wire, size_t wire_len) {
naut_bitfield incoming;
if (naut_bitfield_init(&incoming, mi->num_pieces) != NAUT_OK) {
peer_drop(d, p);
return;
}
naut_bitfield_from_wire(&incoming, wire, wire_len);
for (uint32_t piece = 0; piece < mi->num_pieces; piece++) {
if (naut_bitfield_test(&incoming, piece) &&
!naut_bitfield_test(&p->have, piece)) {
naut_bitfield_set(&p->have, piece);
naut_download_inc_avail(d, piece);
}
}
naut_bitfield_free(&incoming);
}
static bool peer_addr_seen(const peer_t *peers, int npeers,
const naut_peer_addr *addr) {
for (int i = 0; i < npeers; i++)
if (peers[i].addr.port == addr->port &&
memcmp(peers[i].addr.ip, addr->ip, sizeof addr->ip) == 0)
return true;
return false;
}
static naut_err peer_process(naut_download *d, const naut_metainfo *mi,
peer_t *peers, int npeers, peer_t *p,
endpoint_t **pending, size_t *npending,
size_t *pending_cap) {
size_t pos = 0;
if (!p->hs_done) {
if (p->rlen < NAUT_HANDSHAKE_LEN) return NAUT_OK;
uint8_t ih[20], pid[20];
if (!naut_peer_handshake_parse(p->rbuf, ih, pid, NULL) ||
memcmp(ih, mi->infohash_v1, 20) != 0) {
p->dead = true;
return NAUT_OK;
}
memcpy(p->peer_id, pid, sizeof p->peer_id);
p->have_peer_id = true;
pos = NAUT_HANDSHAKE_LEN;
p->hs_done = true;
}
for (;;) {
naut_msg m;
int c = naut_peer_msg_parse(p->rbuf + pos, p->rlen - pos, &m);
if (c == 0) break;
if (c < 0) { p->dead = true; break; }
pos += (size_t)c;
switch (m.type) {
case NAUT_MSG_BITFIELD:
merge_bitfield(d, mi, p, m.payload, m.payload_len);
if (p->dead) goto parsed;
break;
case NAUT_MSG_HAVE:
if (m.index < mi->num_pieces && !naut_bitfield_test(&p->have, m.index)) {
naut_bitfield_set(&p->have, m.index);
naut_download_inc_avail(d, m.index);
}
break;
case NAUT_MSG_UNCHOKE: p->peer_choking = false; break;
case NAUT_MSG_CHOKE: p->peer_choking = true; peer_release(d, p); break;
case NAUT_MSG_EXTENDED:
if (m.payload_len < 1) {
peer_drop(d, p);
goto parsed;
}
if (m.payload[0] == 0) {
if (naut_ext_parse_handshake(
m.payload + 1, m.payload_len - 1,
&p->extensions) != NAUT_OK) {
peer_drop(d, p);
goto parsed;
}
} else if (p->extensions.ut_pex &&
m.payload[0] == p->extensions.ut_pex) {
naut_pex_msg pex;
if (naut_pex_parse(m.payload + 1, m.payload_len - 1,
&pex) != NAUT_OK) {
peer_drop(d, p);
goto parsed;
}
for (size_t i = 0; i < pex.num_added; i++)
if (!peer_addr_seen(peers, npeers, &pex.added[i]) &&
endpoint_add(pending, npending, pending_cap,
&pex.added[i])) {
p->pex_received++;
}
naut_pex_free(&pex);
}
break;
case NAUT_MSG_PIECE: {
req_t request;
bool expected = peer_del_inflight(p, m.index, m.begin, &request);
if (!expected) break;
naut_pipeline_on_block(&p->pipeline, request.length,
request.sent_at, now());
naut_download_unrequest(d, m.index, m.begin);
if (request.length != m.payload_len) {
peer_drop(d, p);
goto parsed;
}
bool pdone = false;
naut_err e = naut_download_on_block(d, m.index, m.begin, m.payload,
(uint32_t)m.payload_len, &pdone);
if (e == NAUT_OK) {
p->blocks_received++;
p->bytes_received += m.payload_len;
cancel_block(d, peers, npeers, p, m.index, m.begin);
} else if (e == NAUT_ERR_PROTO) {
cancel_piece(d, peers, npeers, m.index);
} else {
return e;
}
break;
}
default: break;
}
}
parsed:
memmove(p->rbuf, p->rbuf + pos, p->rlen - pos);
p->rlen -= pos;
return NAUT_OK;
}
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);
uint64_t total = from_magnet ? 0 : (uint64_t)mi.total_length;
if (!discover_trackers(hash, total, trackers, num_trackers,
tracker_tiers, peerid, 0, total,
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;
}
}
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) {
NAUT_ERROR(config->num_peers > 0 ? "no valid peer addresses" :
"tracker and DHT discovery returned no peers");
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOTFOUND;
}
naut_err err;
naut_storage_opts storage_opts = {
.direct_io = getenv("NAUT_DIRECT_IO") != NULL,
.preallocate = true,
};
naut_storage *st = naut_storage_open_opts(
mi.files, mi.num_files, config->output_dir, &storage_opts, &err);
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);
int online_cpus = naut_online_cpus();
uint32_t worker_count =
(uint32_t)NAUT_MAX(1, NAUT_MIN(8, online_cpus / 2));
if (getenv("NAUT_WORKERS")) {
unsigned long configured = strtoul(getenv("NAUT_WORKERS"), NULL, 10);
if (configured > 0 && configured <= 256)
worker_count = (uint32_t)configured;
}
int worker_cpu_base = getenv("NAUT_WORKER_CPU_BASE")
? atoi(getenv("NAUT_WORKER_CPU_BASE")) : -1;
naut_worker_pool *workers =
naut_worker_pool_create(worker_count, 1024, worker_cpu_base);
if (!workers) {
NAUT_ERROR("unable to create hash worker pool");
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOMEM;
}
naut_download_set_worker_pool(d, workers);
endpoint_t *pending = NULL;
size_t npending = 0, pending_cap = 0;
size_t initial_endpoints = neps;
if (config->num_peers == 0 && initial_endpoints > target_peers)
initial_endpoints = target_peers;
for (size_t i = neps; i > initial_endpoints; i--) {
if (!endpoint_add(&pending, &npending, &pending_cap,
&endpoints[i - 1].addr)) {
NAUT_ERROR("out of memory queueing discovered peers");
naut_worker_pool_destroy(workers);
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOMEM;
}
}
int npeers = (int)initial_endpoints;
int peer_capacity = (int)initial_endpoints;
peer_t *peers = calloc(initial_endpoints, sizeof(*peers));
struct pollfd *pfd = calloc(initial_endpoints + 1, sizeof(*pfd));
int *idx_map = calloc(initial_endpoints + 1, sizeof(*idx_map));
if (!peers || !pfd || !idx_map) {
NAUT_ERROR("out of memory creating swarm");
free(peers); free(pfd); free(idx_map); free(pending);
naut_worker_pool_destroy(workers);
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return NAUT_ERR_NOMEM;
}
for (int i = 0; i < npeers; i++) peers[i].fd = -1;
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;
uint32_t connecting = 0;
uint32_t active = 0;
uint32_t failed = 0;
struct pollfd *connect_fds = pfd;
for (int i = 0; i < npeers; i++) connect_fds[i].fd = -1;
report_progress(config, d, &mi, (uint32_t)npeers, 0, 0, 0,
peers, npeers, tracker_stats, tracker_count, t0);
for (int i = 0; i < npeers; i++) {
bool connected = false;
int fd = connect_start(&endpoints[i], &connected);
if (fd < 0) {
NAUT_WARN("connect %s failed", endpoints[i].name);
peers[i].dead = true;
failed++;
continue;
}
peers[i].fd = fd;
if (connected) {
if (!connect_finish(fd) ||
!peer_start(d, &mi, peerid, &endpoints[i], &peers[i], fd)) {
NAUT_WARN("connect %s failed", endpoints[i].name);
if (peers[i].fd >= 0) close(peers[i].fd);
peers[i].fd = -1;
peers[i].dead = true;
failed++;
continue;
}
active++;
emit_event(config, NAUT_EVENT_PEER_CONNECTED, 0,
peers[i].name, NULL);
} else {
connect_fds[i].fd = fd;
connect_fds[i].events = POLLOUT;
connecting++;
}
}
double connect_deadline = now() + CONNECT_TIMEOUT_MS / 1000.0;
while (connecting > 0 && now() < connect_deadline &&
!stop_requested(config)) {
report_progress(config, d, &mi, (uint32_t)npeers, connecting,
active, failed, peers, npeers,
tracker_stats, tracker_count, t0);
int ready = poll(connect_fds, (nfds_t)neps, 100);
if (ready < 0) {
if (errno == EINTR) continue;
run_error = NAUT_ERR_IO;
break;
}
if (ready == 0) continue;
for (int i = 0; i < npeers; i++) {
if (connect_fds[i].fd < 0 ||
!(connect_fds[i].revents &
(POLLOUT | POLLERR | POLLHUP | POLLNVAL)))
continue;
int fd = connect_fds[i].fd;
connect_fds[i].fd = -1;
connecting--;
if (!connect_finish(fd) ||
!peer_start(d, &mi, peerid, &endpoints[i], &peers[i], fd)) {
NAUT_WARN("connect %s failed", endpoints[i].name);
if (peers[i].fd >= 0) close(peers[i].fd);
peers[i].fd = -1;
peers[i].dead = true;
failed++;
continue;
}
active++;
emit_event(config, NAUT_EVENT_PEER_CONNECTED, 0,
peers[i].name, NULL);
}
}
for (int i = 0; i < npeers; i++) {
if (connect_fds[i].fd < 0) continue;
close(connect_fds[i].fd);
connect_fds[i].fd = -1;
peers[i].fd = -1;
peers[i].dead = true;
connecting--;
failed++;
NAUT_WARN("connect %s timed out", endpoints[i].name);
}
free(endpoints);
if (!active) {
NAUT_ERROR("no peers reachable");
if (run_error == NAUT_OK) run_error = NAUT_ERR_IO;
goto done;
}
NAUT_INFO("swarm: %u/%d peers connected, %u pieces, %lld bytes",
active, npeers, mi.num_pieces, (long long)mi.total_length);
emit_event(config, NAUT_EVENT_TORRENT_ADDED, 0, NULL, NULL);
report_progress(config, d, &mi, (uint32_t)npeers, 0, active, failed,
peers, npeers, tracker_stats, tracker_count, t0);
while (!naut_download_complete(d) && run_error == NAUT_OK) {
service_control(config, st);
if (stop_requested(config)) {
cancelled = true;
break;
}
if (config->num_peers == 0 && npending == 0 &&
peer_count_active(peers, npeers, true) < 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;
size_t before = npending;
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,
&pending, &npending, &pending_cap,
&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);
if (npending > before)
NAUT_INFO("tracker refresh queued %zu peers",
npending - before);
}
if (npending == 0 && t >= next_dht_lookup) {
size_t before = npending;
if (!discover_dht(mi.infohash_v1, &pending, &npending,
&pending_cap)) {
run_error = NAUT_ERR_NOMEM;
break;
}
next_dht_lookup = t + DHT_REFRESH_INTERVAL;
if (npending > before)
NAUT_INFO("DHT refresh queued %zu peers",
npending - before);
}
}
if (!connect_pending_peers(config, d, &mi, peerid, &peers, &npeers,
&peer_capacity, &pfd, &idx_map,
pending, &npending, target_peers, &failed)) {
run_error = NAUT_ERR_NOMEM;
break;
}
int nf = 0;
int connecting_peers = 0;
int live_peers = 0;
for (int i = 0; i < npeers; i++) {
if (peers[i].dead) continue;
pfd[nf].fd = peers[i].fd;
pfd[nf].events = peers[i].connecting
? (POLLOUT | POLLERR | POLLHUP | POLLNVAL)
: POLLIN;
pfd[nf].revents = 0;
idx_map[nf] = i; nf++;
if (peers[i].connecting) connecting_peers++;
else live_peers++;
}
pfd[nf].fd = naut_worker_eventfd(workers);
pfd[nf].events = POLLIN;
pfd[nf].revents = 0;
idx_map[nf] = -1;
nf++;
if (live_peers == 0 && connecting_peers == 0) {
uint32_t completed = 0;
run_error = naut_download_poll(d, &completed);
if (naut_download_complete(d)) break;
NAUT_ERROR("all peers gone (%.0f%% done)",
100.0 * naut_download_pieces_done(d) / mi.num_pieces);
run_error = NAUT_ERR_IO;
break;
}
report_progress(config, d, &mi, (uint32_t)npeers,
(uint32_t)connecting_peers,
(uint32_t)live_peers,
failed,
peers, npeers, tracker_stats, tracker_count, t0);
int r = poll(pfd, nf, 200);
if (r < 0) {
if (errno == EINTR) continue;
run_error = NAUT_ERR_IO;
break;
}
for (int k = 0; k < nf; k++) {
if (idx_map[k] < 0) {
if (pfd[k].revents & POLLIN) {
uint64_t count;
(void)read(pfd[k].fd, &count, sizeof count);
uint32_t completed = 0;
run_error = naut_download_poll(d, &completed);
}
continue;
}
peer_t *p = &peers[idx_map[k]];
if (p->connecting) {
if (!(pfd[k].revents &
(POLLOUT | POLLERR | POLLHUP | POLLNVAL)))
continue;
endpoint_t endpoint = { .addr = p->addr };
for (size_t c = 0; c + 1 < sizeof endpoint.name; c++) {
endpoint.name[c] = p->name[c];
if (p->name[c] == 0) break;
}
endpoint.name[sizeof endpoint.name - 1] = 0;
if (!connect_finish(p->fd) ||
!peer_start(d, &mi, peerid, &endpoint, p, p->fd)) {
NAUT_WARN("connect %s failed", p->name);
peer_drop(d, p);
failed++;
continue;
}
emit_event(config, NAUT_EVENT_PEER_CONNECTED, 0,
p->name, NULL);
continue;
}
if (!(pfd[k].revents & (POLLIN | POLLHUP | POLLERR))) continue;
if (p->rlen == p->rcap) {
size_t cap = p->rcap * 2;
uint8_t *buf = realloc(p->rbuf, cap);
if (!buf) { peer_drop(d, p); continue; }
p->rbuf = buf;
p->rcap = cap;
}
ssize_t got = recv(p->fd, p->rbuf + p->rlen, p->rcap - p->rlen, 0);
if (got <= 0) { peer_drop(d, p); continue; }
p->rlen += (size_t)got;
run_error = peer_process(d, &mi, peers, npeers, p,
&pending, &npending, &pending_cap);
if (run_error != NAUT_OK) break;
if (p->dead) peer_drop(d, p);
}
if (run_error != NAUT_OK) break;
double t = now();
for (int i = 0; i < npeers; i++) {
peer_t *p = &peers[i];
if (!p->dead) expire_requests(d, p, t);
}
for (;;) {
bool sent = false;
for (int i = 0; i < npeers; i++)
if (refill_one(d, &peers[i])) sent = true;
if (!sent) break;
}
}
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 from swarm in %.2fs (%.1f MB/s), all SHA-1 verified%s",
naut_download_pieces_done(d), mi.num_pieces, dt, mb/dt,
naut_download_in_endgame(d) ? " (passed through endgame)" : "");
report_progress(config, d, &mi, (uint32_t)npeers, 0, 0,
(uint32_t)npeers, peers, npeers,
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);
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);
}
done:
ok = naut_download_complete(d);
service_control(config, st);
naut_storage_sync(st);
double elapsed = now() - t0;
for (int i = 0; i < npeers; i++) {
if (peers[i].blocks_received)
NAUT_INFO("peer %s delivered %llu blocks (pipeline %u, RTT %.1f ms, avg %.1f MB/s)",
peers[i].name,
(unsigned long long)peers[i].blocks_received,
naut_pipeline_depth(&peers[i].pipeline),
peers[i].pipeline.rtt_seconds * 1000.0,
elapsed > 0 ? peers[i].bytes_received / elapsed / 1e6 : 0.0);
if (peers[i].pex_received)
NAUT_INFO("peer %s advertised %llu peers through PEX",
peers[i].name,
(unsigned long long)peers[i].pex_received);
if (!peers[i].dead) peer_drop(d, &peers[i]);
free(peers[i].rbuf); free(peers[i].inflight);
if (peers[i].have.words) naut_bitfield_free(&peers[i].have);
}
free(peers); free(pfd); free(idx_map);
free(pending);
naut_worker_pool_destroy(workers);
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;
}
#ifndef NAUT_SWARM_LIBRARY
int main(int argc, char **argv) {
if (argc < 3) {
fprintf(stderr,
"usage: %s <file.torrent|magnet-uri> <out-dir> [ip:port ...]\n",
argv[0]);
return 2;
}
naut_log_set_level(NAUT_LOG_INFO);
naut_swarm_config config = {
.source = argv[1],
.output_dir = argv[2],
.peers = argc > 3 ? (const char *const *)&argv[3] : NULL,
.num_peers = argc > 3 ? (size_t)(argc - 3) : 0,
};
return naut_swarm_run(&config) == NAUT_OK ? 0 : 1;
}
#endif
Reference in a new issue View git blame Copy permalink