Cieric/Naut
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

webui: replace nautctl web server with a loadable plugin

Browse source 
Drop the web UI that was compiled into nautctl and serve the
torrent-ui front end (../torrent-ui/public) from a native plugin
(plugins/webui) loaded via `nautd --plugin`. The plugin talks to the
engine only through the host call_rpc ABI and adapts the daemon's RPC
surface to the qBittorrent-style contract the UI expects (snapshot/SSE,
torrent detail tabs, add/delete, cookie auth).

Also folds in the daemon refactor that owns per-torrent worker threads
and the swarm engine (naut_swarm) used by the plugin's data source.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
ookami125 2026-06-17 00:21:48 -04:00
parent 50a357968a
commit 8dde48c05a
27 changed files with 2706 additions and 403 deletions
Download patch file Download diff file Expand all files Collapse all files

67
apps/nautctl/main.c
View file

@ -11,8 +11,16 @@
static void usage(const char *program) {
fprintf(stderr,
"usage: %s [--socket PATH] METHOD [PARAMS_JSON]\n"
" %s [--socket PATH] events\n", program, program);
"usage: %s [--socket PATH] COMMAND [ARGS]\n"
"\n"
"commands:\n"
" add SOURCE OUTPUT [IP:PORT ...]\n"
" list\n"
" show TORRENT_ID\n"
" remove TORRENT_ID\n"
" script PATH | unscript\n"
" status | events | shutdown\n"
" METHOD [PARAMS_JSON] (raw RPC)\n", program);
}
static json_t *parse_params(const char *text) {
@ -54,6 +62,15 @@ static int stream_events(const char *socket_path) {
}
}
static bool parse_id(const char *text, json_int_t *id) {
if (!text || !*text || *text == '-') return false;
char *end = NULL;
unsigned long long value = strtoull(text, &end, 10);
if (!end || *end || value > (unsigned long long)INT64_MAX) return false;
*id = (json_int_t)value;
return true;
}
int main(int argc, char **argv) {
const char *socket_path = getenv("NAUT_SOCKET");
if (!socket_path || !*socket_path) socket_path = DEFAULT_SOCKET;
@ -66,7 +83,7 @@ int main(int argc, char **argv) {
socket_path = argv[arg + 1];
arg += 2;
}
if (arg >= argc || arg + 2 < argc) {
if (arg >= argc) {
usage(argv[0]);
return 2;
}
@ -74,7 +91,49 @@ int main(int argc, char **argv) {
const char *method = argv[arg++];
if (strcmp(method, "events") == 0)
return stream_events(socket_path);
json_t *params = parse_params(arg < argc ? argv[arg] : NULL);
json_t *params = NULL;
if (strcmp(method, "add") == 0) {
if (arg + 1 >= argc) { usage(argv[0]); return 2; }
method = "add_torrent";
params = json_pack("{s:s,s:s}", "source", argv[arg],
"output", argv[arg + 1]);
arg += 2;
json_t *peers = json_array();
if (!params || !peers) {
json_decref(params);
json_decref(peers);
return 1;
}
while (arg < argc)
json_array_append_new(peers, json_string(argv[arg++]));
json_object_set_new(params, "peers", peers);
} else if (strcmp(method, "list") == 0) {
if (arg != argc) { usage(argv[0]); return 2; }
method = "torrents";
params = json_object();
} else if (strcmp(method, "show") == 0 ||
strcmp(method, "remove") == 0) {
json_int_t id;
if (arg + 1 != argc || !parse_id(argv[arg], &id)) {
usage(argv[0]);
return 2;
}
method = strcmp(method, "show") == 0 ? "torrent" :
"remove_torrent";
params = json_pack("{s:I}", "torrent_id", id);
} else if (strcmp(method, "script") == 0) {
if (arg + 1 != argc) { usage(argv[0]); return 2; }
method = "load_script";
params = json_pack("{s:s}", "path", argv[arg]);
} else if (strcmp(method, "unscript") == 0) {
if (arg != argc) { usage(argv[0]); return 2; }
method = "unload_script";
params = json_object();
} else {
if (arg + 1 < argc) { usage(argv[0]); return 2; }
params = parse_params(arg < argc ? argv[arg] : NULL);
}
if (!params) {
fprintf(stderr, "nautctl: invalid JSON parameters\n");
return 2;

688
apps/nautd/main.c
View file

@ -1,11 +1,10 @@
#include "naut/event.h"
#include "naut/log.h"
#include "naut/metainfo.h"
#include "naut/plugin.h"
#include "naut/rpc.h"
#include "naut/script.h"
#include "naut/session.h"
#include "naut/storage.h"
#include "naut/swarm.h"
#include <errno.h>
#include <fcntl.h>
@ -23,37 +22,221 @@
#define DEFAULT_SOCKET "/tmp/nautd.sock"
#define MOVE_QUEUE_CAPACITY 64
#define MAX_SUBSCRIBERS 64
#define MAX_TORRENTS 128
typedef struct {
uint64_t torrent_id;
uint32_t file_index;
char destination[PATH_MAX];
} move_command;
typedef enum {
TORRENT_QUEUED,
TORRENT_RUNNING,
TORRENT_COMPLETE,
TORRENT_STOPPING,
TORRENT_STOPPED,
TORRENT_ERROR,
} torrent_state;
typedef struct daemon_state daemon_state;
typedef struct {
naut_event_bus *events;
naut_rpc_registry *rpc;
naut_plugin_manager *plugins;
naut_script *script;
naut_session *session;
pthread_mutex_t move_lock;
daemon_state *daemon;
uint64_t id;
char *source;
bool source_is_temp; /* source is a daemon-owned upload; unlink on destroy */
char *output_dir;
char **peers;
size_t num_peers;
pthread_t thread;
bool thread_started;
bool thread_done;
pthread_mutex_t lock;
torrent_state state;
naut_err result;
bool stop_requested;
bool remove_requested;
naut_swarm_stats stats;
move_command moves[MOVE_QUEUE_CAPACITY];
size_t move_head;
size_t move_count;
uint64_t moves_processed;
uint64_t moves_failed;
} torrent_task;
struct daemon_state {
naut_event_bus *events;
naut_rpc_registry *rpc;
naut_plugin_manager *plugins;
naut_script *script;
pthread_mutex_t torrent_lock;
torrent_task *torrents[MAX_TORRENTS];
size_t torrent_count;
uint64_t next_torrent_id;
pthread_mutex_t subscriber_lock;
int subscribers[MAX_SUBSCRIBERS];
size_t subscriber_count;
bool stopping;
} daemon_state;
};
static volatile sig_atomic_t interrupted;
static naut_err queue_move(void *opaque, uint64_t torrent_id,
uint32_t file_index, const char *destination);
static void on_signal(int signal_number) {
(void)signal_number;
interrupted = 1;
}
static const char *torrent_state_name(torrent_state state) {
static const char *names[] = {
[TORRENT_QUEUED] = "queued",
[TORRENT_RUNNING] = "downloading",
[TORRENT_COMPLETE] = "complete",
[TORRENT_STOPPING] = "stopping",
[TORRENT_STOPPED] = "stopped",
[TORRENT_ERROR] = "error",
};
return (size_t)state < NAUT_ARRAY_LEN(names) ? names[state] : "unknown";
}
static torrent_task *find_torrent_locked(daemon_state *state, uint64_t id) {
for (size_t i = 0; i < state->torrent_count; i++)
if (state->torrents[i]->id == id) return state->torrents[i];
return NULL;
}
static void torrent_progress(void *opaque, const naut_swarm_stats *stats) {
torrent_task *task = opaque;
pthread_mutex_lock(&task->lock);
task->stats = *stats;
if (stats->total_pieces > 0 &&
stats->pieces_done == stats->total_pieces)
task->state = TORRENT_COMPLETE;
else if (task->state == TORRENT_QUEUED)
task->state = TORRENT_RUNNING;
pthread_mutex_unlock(&task->lock);
}
static bool torrent_should_stop(void *opaque) {
torrent_task *task = opaque;
pthread_mutex_lock(&task->lock);
bool stop = task->stop_requested;
pthread_mutex_unlock(&task->lock);
return stop;
}
static void torrent_control(void *opaque, naut_storage *storage) {
torrent_task *task = opaque;
for (;;) {
move_command command;
pthread_mutex_lock(&task->lock);
if (task->move_count == 0) {
pthread_mutex_unlock(&task->lock);
return;
}
command = task->moves[task->move_head];
task->move_head = (task->move_head + 1) % MOVE_QUEUE_CAPACITY;
task->move_count--;
pthread_mutex_unlock(&task->lock);
naut_err error = naut_storage_relocate(
storage, command.file_index, command.destination);
pthread_mutex_lock(&task->lock);
if (error == NAUT_OK)
task->moves_processed++;
else
task->moves_failed++;
pthread_mutex_unlock(&task->lock);
if (error == NAUT_OK)
NAUT_INFO("moved torrent=%llu file=%u -> %s",
(unsigned long long)task->id, command.file_index,
command.destination);
else
NAUT_WARN("move torrent=%llu file=%u -> %s failed: %s",
(unsigned long long)task->id, command.file_index,
command.destination, naut_strerror(error));
}
}
static void *torrent_worker(void *opaque) {
torrent_task *task = opaque;
pthread_mutex_lock(&task->lock);
task->state = TORRENT_RUNNING;
pthread_mutex_unlock(&task->lock);
naut_swarm_config config = {
.source = task->source,
.output_dir = task->output_dir,
.peers = (const char *const *)task->peers,
.num_peers = task->num_peers,
.torrent_id = task->id,
.events = task->daemon->events,
.keep_alive = true,
.on_progress = torrent_progress,
.on_control = torrent_control,
.should_stop = torrent_should_stop,
.context = task,
};
naut_err result = naut_swarm_run(&config);
pthread_mutex_lock(&task->lock);
task->result = result;
if (result == NAUT_OK)
task->state = TORRENT_COMPLETE;
else if (task->stop_requested)
task->state = TORRENT_STOPPED;
else
task->state = TORRENT_ERROR;
task->thread_done = true;
pthread_mutex_unlock(&task->lock);
return NULL;
}
static json_t *torrent_json(torrent_task *task) {
pthread_mutex_lock(&task->lock);
json_t *result = json_object();
if (result) {
json_object_set_new(result, "torrent_id",
json_integer((json_int_t)task->id));
json_object_set_new(result, "source", json_string(task->source));
json_object_set_new(result, "output",
json_string(task->output_dir));
json_object_set_new(result, "state",
json_string(torrent_state_name(task->state)));
json_object_set_new(result, "bytes_done",
json_integer((json_int_t)task->stats.bytes_done));
json_object_set_new(result, "total_bytes",
json_integer((json_int_t)task->stats.total_bytes));
json_object_set_new(result, "pieces_done",
json_integer(task->stats.pieces_done));
json_object_set_new(result, "total_pieces",
json_integer(task->stats.total_pieces));
json_object_set_new(result, "peers",
json_integer(task->stats.peers_active));
json_object_set_new(result, "peers_discovered",
json_integer(task->stats.peers_total));
json_object_set_new(result, "peers_connecting",
json_integer(task->stats.peers_connecting));
json_object_set_new(result, "peers_failed",
json_integer(task->stats.peers_failed));
json_object_set_new(result, "elapsed_seconds",
json_real(task->stats.elapsed_seconds));
json_object_set_new(result, "pending_moves",
json_integer((json_int_t)task->move_count));
json_object_set_new(result, "moves_processed",
json_integer((json_int_t)task->moves_processed));
json_object_set_new(result, "moves_failed",
json_integer((json_int_t)task->moves_failed));
if (task->state == TORRENT_ERROR)
json_object_set_new(result, "error",
json_string(naut_strerror(task->result)));
}
pthread_mutex_unlock(&task->lock);
return result;
}
static json_t *rpc_ping(void *opaque, const json_t *params, naut_err *error) {
(void)opaque;
(void)params;
@ -74,10 +257,23 @@ static json_t *rpc_status(void *opaque, const json_t *params,
daemon_state *state = opaque;
naut_script_stats stats = {0};
if (state->script) naut_script_get_stats(state->script, &stats);
pthread_mutex_lock(&state->move_lock);
uint64_t moves = state->moves_processed;
size_t pending = state->move_count;
pthread_mutex_unlock(&state->move_lock);
size_t torrent_count;
size_t active = 0;
uint64_t moves = 0;
size_t pending = 0;
pthread_mutex_lock(&state->torrent_lock);
torrent_count = state->torrent_count;
for (size_t i = 0; i < torrent_count; i++) {
torrent_task *task = state->torrents[i];
pthread_mutex_lock(&task->lock);
if (task->state == TORRENT_RUNNING ||
task->state == TORRENT_STOPPING)
active++;
moves += task->moves_processed;
pending += task->move_count;
pthread_mutex_unlock(&task->lock);
}
pthread_mutex_unlock(&state->torrent_lock);
json_t *result = json_object();
json_t *script = json_object();
if (!result || !script) {
@ -93,6 +289,12 @@ static json_t *rpc_status(void *opaque, const json_t *params,
json_object_set_new(result, "storage_backends",
json_integer((json_int_t)naut_plugin_storage_count(
state->plugins)));
json_object_set_new(result, "torrents",
json_integer((json_int_t)torrent_count));
json_object_set_new(result, "active_torrents",
json_integer((json_int_t)active));
json_object_set_new(result, "script_loaded",
json_boolean(state->script != NULL));
json_object_set_new(script, "queued", json_integer(stats.queued));
json_object_set_new(script, "handled", json_integer(stats.handled));
json_object_set_new(script, "dropped", json_integer(stats.dropped));
@ -178,122 +380,328 @@ static json_t *rpc_shutdown(void *opaque, const json_t *params,
return json_true();
}
static uint8_t *read_file(const char *path, size_t *len) {
FILE *f = fopen(path, "rb");
if (!f) return NULL;
if (fseek(f, 0, SEEK_END) != 0) { fclose(f); return NULL; }
long n = ftell(f);
if (n < 0 || fseek(f, 0, SEEK_SET) != 0) { fclose(f); return NULL; }
uint8_t *buf = malloc((size_t)n);
if (!buf) { fclose(f); return NULL; }
if (fread(buf, 1, (size_t)n, f) != (size_t)n) {
free(buf); fclose(f); return NULL;
}
fclose(f);
*len = (size_t)n;
return buf;
static int b64_val(int c) {
if (c >= 'A' && c <= 'Z') return c - 'A';
if (c >= 'a' && c <= 'z') return c - 'a' + 26;
if (c >= '0' && c <= '9') return c - '0' + 52;
if (c == '+') return 62;
if (c == '/') return 63;
return -1; /* padding / whitespace / invalid -> skipped */
}
/* Decode standard base64 (padding optional, whitespace ignored). */
static unsigned char *b64_decode(const char *in, size_t *out_len) {
size_t cap = strlen(in) / 4 * 3 + 4;
unsigned char *out = malloc(cap);
if (!out) return NULL;
size_t o = 0;
int acc = 0, bits = 0;
for (const char *p = in; *p; p++) {
if (*p == '=') break;
int v = b64_val((unsigned char)*p);
if (v < 0) continue;
acc = (acc << 6) | v;
bits += 6;
if (bits >= 8) { bits -= 8; out[o++] = (unsigned char)((acc >> bits) & 0xff); }
}
*out_len = o;
return out;
}
static bool write_all_fd(int fd, const void *buf, size_t len) {
const char *p = buf;
while (len) {
ssize_t n = write(fd, p, len);
if (n < 0) { if (errno == EINTR) continue; return false; }
p += n;
len -= (size_t)n;
}
return true;
}
/* A browser uploads a .torrent's bytes as base64 in "data"; the daemon writes
* its own temp file and owns its lifecycle (no shared path with the client).
* Returns the temp path in `out` (size cap) or sets *error. */
static bool add_torrent_write_upload(const char *data_b64, char *out,
size_t cap, naut_err *error) {
size_t raw_len = 0;
unsigned char *raw = b64_decode(data_b64, &raw_len);
if (!raw || raw_len == 0) { free(raw); *error = NAUT_ERR_INVAL; return false; }
char tmpl[] = "/tmp/naut-upload-XXXXXX";
int fd = mkstemp(tmpl);
if (fd < 0) { free(raw); *error = NAUT_ERR_IO; return false; }
bool ok = write_all_fd(fd, raw, raw_len);
close(fd);
free(raw);
if (!ok) { unlink(tmpl); *error = NAUT_ERR_IO; return false; }
snprintf(out, cap, "%s", tmpl);
return true;
}
/* add_torrent {torrent_id, torrent: <.torrent path>, root: <output dir>} opens
* the torrent's storage and registers it so move_file can later relocate one of
* its files. This is the control-plane seam that binds a script's move command
* to a concrete naut_storage; it runs on the daemon owner thread. */
static json_t *rpc_add_torrent(void *opaque, const json_t *params,
naut_err *error) {
daemon_state *state = opaque;
if (!json_is_object(params)) { *error = NAUT_ERR_INVAL; return NULL; }
json_int_t torrent_id =
json_integer_value(json_object_get(params, "torrent_id"));
const char *torrent_path =
json_string_value(json_object_get(params, "torrent"));
const char *root = json_string_value(json_object_get(params, "root"));
if (torrent_id < 0 || !torrent_path || !root) {
const char *source =
json_string_value(json_object_get(params, "source"));
if (!source)
source = json_string_value(json_object_get(params, "torrent"));
const char *data_b64 =
json_string_value(json_object_get(params, "data"));
const char *output =
json_string_value(json_object_get(params, "output"));
if (!output)
output = json_string_value(json_object_get(params, "root"));
json_t *peers_json = json_object_get(params, "peers");
/* need an output and either a source (path/magnet) or uploaded bytes */
if (!output || !*output ||
((!source || !*source) && (!data_b64 || !*data_b64)) ||
(peers_json && !json_is_array(peers_json))) {
*error = NAUT_ERR_INVAL;
return NULL;
}
if (naut_session_has(state->session, (uint64_t)torrent_id)) {
/* materialize an upload into a daemon-owned temp .torrent */
char temp_source[PATH_MAX];
bool is_temp = false;
if ((!source || !*source) && data_b64 && *data_b64) {
if (!add_torrent_write_upload(data_b64, temp_source, sizeof temp_source,
error))
return NULL;
source = temp_source;
is_temp = true;
}
torrent_task *task = calloc(1, sizeof(*task));
if (!task) {
if (is_temp) unlink(source);
*error = NAUT_ERR_NOMEM;
return NULL;
}
task->daemon = state;
task->state = TORRENT_QUEUED;
task->result = NAUT_ERR_AGAIN;
task->source = strdup(source);
task->source_is_temp = is_temp;
task->output_dir = strdup(output);
if (!task->source || !task->output_dir) {
if (is_temp) unlink(temp_source);
free(task->source);
free(task->output_dir);
free(task);
*error = NAUT_ERR_NOMEM;
return NULL;
}
if (pthread_mutex_init(&task->lock, NULL) != 0) {
if (is_temp) unlink(temp_source);
free(task->source);
free(task->output_dir);
free(task);
*error = NAUT_ERR_NOMEM;
return NULL;
}
task->num_peers = peers_json ? json_array_size(peers_json) : 0;
if (task->num_peers) {
task->peers = calloc(task->num_peers, sizeof(*task->peers));
if (!task->peers) { *error = NAUT_ERR_NOMEM; goto fail_task; }
for (size_t i = 0; i < task->num_peers; i++) {
const char *peer =
json_string_value(json_array_get(peers_json, i));
if (!peer || !*peer) { *error = NAUT_ERR_INVAL; goto fail_task; }
task->peers[i] = strdup(peer);
if (!task->peers[i]) { *error = NAUT_ERR_NOMEM; goto fail_task; }
}
}
json_t *id_json = json_object_get(params, "torrent_id");
pthread_mutex_lock(&state->torrent_lock);
if (state->torrent_count == MAX_TORRENTS) {
pthread_mutex_unlock(&state->torrent_lock);
*error = NAUT_ERR_FULL;
goto fail_task;
}
if (id_json) {
if (!json_is_integer(id_json) || json_integer_value(id_json) < 0) {
pthread_mutex_unlock(&state->torrent_lock);
*error = NAUT_ERR_INVAL;
goto fail_task;
}
task->id = (uint64_t)json_integer_value(id_json);
if (task->id < (uint64_t)INT64_MAX &&
task->id >= state->next_torrent_id)
state->next_torrent_id = task->id + 1;
} else {
while (find_torrent_locked(state, state->next_torrent_id))
state->next_torrent_id++;
if (state->next_torrent_id > (uint64_t)INT64_MAX) {
pthread_mutex_unlock(&state->torrent_lock);
*error = NAUT_ERR_FULL;
goto fail_task;
}
task->id = state->next_torrent_id++;
}
if (find_torrent_locked(state, task->id)) {
pthread_mutex_unlock(&state->torrent_lock);
*error = NAUT_ERR_INVAL;
goto fail_task;
}
state->torrents[state->torrent_count++] = task;
pthread_mutex_unlock(&state->torrent_lock);
if (pthread_create(&task->thread, NULL, torrent_worker, task) != 0) {
pthread_mutex_lock(&state->torrent_lock);
state->torrent_count--;
pthread_mutex_unlock(&state->torrent_lock);
*error = NAUT_ERR_NOMEM;
goto fail_task;
}
task->thread_started = true;
*error = NAUT_OK;
return torrent_json(task);
fail_task:
if (is_temp) unlink(temp_source);
for (size_t i = 0; i < task->num_peers; i++) free(task->peers[i]);
free(task->peers);
free(task->source);
free(task->output_dir);
pthread_mutex_destroy(&task->lock);
free(task);
return NULL;
}
static json_t *rpc_torrents(void *opaque, const json_t *params,
naut_err *error) {
(void)params;
daemon_state *state = opaque;
json_t *result = json_array();
if (!result) { *error = NAUT_ERR_NOMEM; return NULL; }
pthread_mutex_lock(&state->torrent_lock);
for (size_t i = 0; i < state->torrent_count; i++) {
json_t *item = torrent_json(state->torrents[i]);
if (!item || json_array_append_new(result, item) != 0) {
json_decref(item);
json_decref(result);
pthread_mutex_unlock(&state->torrent_lock);
*error = NAUT_ERR_NOMEM;
return NULL;
}
}
pthread_mutex_unlock(&state->torrent_lock);
*error = NAUT_OK;
return result;
}
static bool parse_torrent_id(const json_t *params, uint64_t *id) {
if (!json_is_object(params)) return false;
json_t *value = json_object_get(params, "torrent_id");
if (!json_is_integer(value) || json_integer_value(value) < 0) return false;
*id = (uint64_t)json_integer_value(value);
return true;
}
static json_t *rpc_torrent(void *opaque, const json_t *params,
naut_err *error) {
daemon_state *state = opaque;
uint64_t id;
if (!parse_torrent_id(params, &id)) {
*error = NAUT_ERR_INVAL;
return NULL;
}
size_t len = 0;
uint8_t *raw = read_file(torrent_path, &len);
if (!raw) { *error = NAUT_ERR_IO; return NULL; }
naut_metainfo mi;
naut_err err = naut_metainfo_parse(raw, len, &mi);
free(raw);
if (err != NAUT_OK) { *error = err; return NULL; }
naut_storage *storage =
naut_storage_open(mi.files, mi.num_files, root, &err);
if (!storage) {
naut_metainfo_free(&mi);
*error = err != NAUT_OK ? err : NAUT_ERR_IO;
pthread_mutex_lock(&state->torrent_lock);
torrent_task *task = find_torrent_locked(state, id);
json_t *result = task ? torrent_json(task) : NULL;
pthread_mutex_unlock(&state->torrent_lock);
*error = task ? (result ? NAUT_OK : NAUT_ERR_NOMEM) : NAUT_ERR_NOTFOUND;
return result;
}
static json_t *rpc_remove_torrent(void *opaque, const json_t *params,
naut_err *error) {
daemon_state *state = opaque;
uint64_t id;
if (!parse_torrent_id(params, &id)) {
*error = NAUT_ERR_INVAL;
return NULL;
}
naut_metainfo_free(&mi);
err = naut_session_add(state->session, (uint64_t)torrent_id, storage);
if (err != NAUT_OK) {
naut_storage_close(storage);
*error = err;
pthread_mutex_lock(&state->torrent_lock);
torrent_task *task = find_torrent_locked(state, id);
if (task) {
pthread_mutex_lock(&task->lock);
task->stop_requested = true;
task->remove_requested = true;
if (!task->thread_done)
task->state = TORRENT_STOPPING;
pthread_mutex_unlock(&task->lock);
}
pthread_mutex_unlock(&state->torrent_lock);
if (!task) {
*error = NAUT_ERR_NOTFOUND;
return NULL;
}
*error = NAUT_OK;
return torrent_json(task);
}
static json_t *rpc_load_script(void *opaque, const json_t *params,
naut_err *error) {
daemon_state *state = opaque;
const char *path = json_is_object(params)
? json_string_value(json_object_get(params, "path")) : NULL;
if (!path || !*path) { *error = NAUT_ERR_INVAL; return NULL; }
naut_script *script = naut_script_create(
state->events, path, 256, queue_move, state, error);
if (!script) return NULL;
naut_script *old = state->script;
state->script = script;
naut_script_destroy(old);
*error = NAUT_OK;
return json_string(path);
}
static json_t *rpc_unload_script(void *opaque, const json_t *params,
naut_err *error) {
(void)params;
daemon_state *state = opaque;
naut_script *old = state->script;
state->script = NULL;
naut_script_destroy(old);
*error = NAUT_OK;
return json_true();
}
static naut_err queue_move(void *opaque, uint64_t torrent_id,
uint32_t file_index, const char *destination) {
daemon_state *state = opaque;
pthread_mutex_lock(&state->move_lock);
if (state->move_count == MOVE_QUEUE_CAPACITY) {
pthread_mutex_unlock(&state->move_lock);
if (!destination || strlen(destination) >= PATH_MAX)
return NAUT_ERR_RANGE;
pthread_mutex_lock(&state->torrent_lock);
torrent_task *task = find_torrent_locked(state, torrent_id);
if (!task) {
pthread_mutex_unlock(&state->torrent_lock);
return NAUT_ERR_NOTFOUND;
}
pthread_mutex_lock(&task->lock);
pthread_mutex_unlock(&state->torrent_lock);
if (task->remove_requested) {
pthread_mutex_unlock(&task->lock);
return NAUT_ERR_NOTFOUND;
}
if (task->move_count == MOVE_QUEUE_CAPACITY) {
pthread_mutex_unlock(&task->lock);
return NAUT_ERR_FULL;
}
size_t tail = (state->move_head + state->move_count) %
MOVE_QUEUE_CAPACITY;
state->moves[tail] = (move_command) {
.torrent_id = torrent_id,
.file_index = file_index,
};
snprintf(state->moves[tail].destination,
sizeof state->moves[tail].destination, "%s", destination);
state->move_count++;
pthread_mutex_unlock(&state->move_lock);
size_t tail = (task->move_head + task->move_count) % MOVE_QUEUE_CAPACITY;
task->moves[tail].file_index = file_index;
snprintf(task->moves[tail].destination,
sizeof task->moves[tail].destination, "%s", destination);
task->move_count++;
pthread_mutex_unlock(&task->lock);
return NAUT_OK;
}
static void drain_moves(daemon_state *state) {
/* Copy each pending command out under the lock, then perform the relocate
* with the lock released (so the script thread can keep enqueuing). All
* relocates run on this, the owner thread, as the session requires. */
for (;;) {
move_command command;
pthread_mutex_lock(&state->move_lock);
if (state->move_count == 0) {
pthread_mutex_unlock(&state->move_lock);
return;
}
command = state->moves[state->move_head];
state->move_head = (state->move_head + 1) % MOVE_QUEUE_CAPACITY;
state->move_count--;
state->moves_processed++;
pthread_mutex_unlock(&state->move_lock);
naut_err err = naut_session_move_file(state->session,
command.torrent_id,
command.file_index,
command.destination);
if (err == NAUT_OK)
NAUT_INFO("moved torrent=%llu file=%u -> %s",
(unsigned long long)command.torrent_id,
command.file_index, command.destination);
else
NAUT_WARN("move torrent=%llu file=%u -> %s failed: %s",
(unsigned long long)command.torrent_id,
command.file_index, command.destination,
naut_strerror(err));
}
}
static void broadcast_event(void *opaque, const naut_event *event) {
daemon_state *state = opaque;
json_t *payload = naut_rpc_event_json(event);
@ -406,9 +814,68 @@ static bool register_commands(daemon_state *state) {
naut_rpc_register(state->rpc, "plugins", rpc_plugins, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "emit", rpc_emit, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "add_torrent", rpc_add_torrent, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "torrents", rpc_torrents, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "torrent", rpc_torrent, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "remove_torrent", rpc_remove_torrent, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "load_script", rpc_load_script, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "unload_script", rpc_unload_script, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "shutdown", rpc_shutdown, state) == NAUT_OK;
}
static void stop_torrents(daemon_state *state) {
pthread_mutex_lock(&state->torrent_lock);
for (size_t i = 0; i < state->torrent_count; i++) {
torrent_task *task = state->torrents[i];
pthread_mutex_lock(&task->lock);
task->stop_requested = true;
if (task->state == TORRENT_RUNNING)
task->state = TORRENT_STOPPING;
pthread_mutex_unlock(&task->lock);
}
pthread_mutex_unlock(&state->torrent_lock);
}
static void destroy_torrent(torrent_task *task) {
if (task->thread_started) pthread_join(task->thread, NULL);
/* uploaded torrents live in a daemon-owned temp file; remove it now that
* the worker has finished reading it */
if (task->source_is_temp && task->source) unlink(task->source);
for (size_t p = 0; p < task->num_peers; p++) free(task->peers[p]);
free(task->peers);
free(task->source);
free(task->output_dir);
pthread_mutex_destroy(&task->lock);
free(task);
}
static void reap_torrents(daemon_state *state) {
for (;;) {
torrent_task *task = NULL;
pthread_mutex_lock(&state->torrent_lock);
for (size_t i = 0; i < state->torrent_count; i++) {
torrent_task *candidate = state->torrents[i];
pthread_mutex_lock(&candidate->lock);
bool reap = candidate->remove_requested &&
candidate->thread_done;
pthread_mutex_unlock(&candidate->lock);
if (!reap) continue;
task = candidate;
state->torrents[i] =
state->torrents[--state->torrent_count];
break;
}
pthread_mutex_unlock(&state->torrent_lock);
if (!task) return;
destroy_torrent(task);
}
}
static void destroy_torrents(daemon_state *state) {
for (size_t i = 0; i < state->torrent_count; i++)
destroy_torrent(state->torrents[i]);
state->torrent_count = 0;
}
static void usage(const char *program) {
fprintf(stderr,
"usage: %s [--socket PATH] [--plugin PATH]... [--script PATH]\n",
@ -438,13 +905,13 @@ int main(int argc, char **argv) {
signal(SIGTERM, on_signal);
signal(SIGPIPE, SIG_IGN);
daemon_state state = {0};
pthread_mutex_init(&state.move_lock, NULL);
state.next_torrent_id = 1;
pthread_mutex_init(&state.torrent_lock, NULL);
pthread_mutex_init(&state.subscriber_lock, NULL);
state.events = naut_event_bus_create();
state.rpc = naut_rpc_registry_create();
state.plugins = naut_plugin_manager_create(state.rpc, state.events);
state.session = naut_session_create();
if (!state.events || !state.rpc || !state.plugins || !state.session ||
if (!state.events || !state.rpc || !state.plugins ||
!register_commands(&state)) {
fprintf(stderr, "nautd: failed to initialize control plane\n");
return 1;
@ -485,23 +952,24 @@ int main(int argc, char **argv) {
} else if (ready < 0 && errno != EINTR) {
break;
}
drain_moves(&state);
reap_torrents(&state);
}
close(listener);
unlink(socket_path);
naut_script_destroy(state.script);
state.script = NULL;
stop_torrents(&state);
destroy_torrents(&state);
naut_event_unsubscribe(state.events, event_subscription);
pthread_mutex_lock(&state.subscriber_lock);
for (size_t i = 0; i < state.subscriber_count; i++)
close(state.subscribers[i]);
pthread_mutex_unlock(&state.subscriber_lock);
naut_script_destroy(state.script); /* joins the script thread */
drain_moves(&state); /* flush any moves it left queued */
naut_plugin_manager_destroy(state.plugins);
naut_rpc_registry_destroy(state.rpc);
naut_session_destroy(state.session);
naut_event_bus_destroy(state.events);
pthread_mutex_destroy(&state.subscriber_lock);
pthread_mutex_destroy(&state.move_lock);
pthread_mutex_destroy(&state.torrent_lock);
return 0;
}

472
apps/swarm/main.c
View file

@ -18,10 +18,13 @@
#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 <stdio.h>
#include <stdlib.h>
@ -33,6 +36,7 @@
#include <netinet/tcp.h>
#define REQUEST_TIMEOUT 15.0
#define CONNECT_TIMEOUT_MS 5000
#define EXT_RESERVED 0x0000000000100000ULL
typedef struct {
@ -61,6 +65,92 @@ typedef struct {
static double now(void) { struct timespec t; clock_gettime(CLOCK_MONOTONIC, &t); return t.tv_sec + t.tv_nsec*1e-9; }
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 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,
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,
};
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);
@ -138,6 +228,7 @@ static bool parse_udp_tracker(const char *url, char *host, size_t hostsz,
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],
endpoint_t **eps, size_t *neps, size_t *cap) {
naut_announce_req req;
@ -148,37 +239,62 @@ static bool discover_trackers(const uint8_t info_hash[20], uint64_t total_length
req.left = total_length;
req.event = NAUT_TEV_STARTED;
req.numwant = 100;
req.key = (uint32_t)rand();
memcpy(&req.key, peerid + 8, sizeof req.key);
for (size_t i = 0; i < num_trackers; i++) {
const char *tracker = trackers[i];
naut_tracker_response response;
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 {
NAUT_WARN("tracker scheme unsupported: %s", tracker);
continue;
}
if (e != NAUT_OK) {
NAUT_WARN("tracker announce failed: %s", tracker);
continue;
}
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;
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;
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 {
NAUT_WARN("tracker scheme unsupported: %s", tracker);
continue;
}
if (e != NAUT_OK) {
NAUT_WARN("tracker announce failed: %s", tracker);
continue;
}
tier_succeeded = true;
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;
}
naut_tracker_response_free(&response);
if (tier_succeeded) break;
tier_start = tier_end;
}
return true;
}
@ -321,18 +437,81 @@ static void expire_requests(naut_download *d, peer_t *p, double t) {
}
}
static int connect_to(const endpoint_t *ep) {
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) { close(fd); return -1; }
int one = 1; setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
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;
snprintf(peer->name, sizeof peer->name, "%s", endpoint->name);
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;
}
/* 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) {
@ -481,50 +660,46 @@ parsed:
return NAUT_OK;
}
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_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);
srand((unsigned)time(NULL) ^ (unsigned)getpid());
for (int i = 8; i < 20; i++) peerid[i] = (uint8_t)(rand() & 0xff);
uint8_t peerid[20];
memcpy(peerid, "-NT0001-", 8);
random_bytes(peerid + 8, sizeof peerid - 8);
bool from_magnet = strncmp(argv[1], "magnet:?", 8) == 0;
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(argv[1], &magnet) != NAUT_OK ||
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 1;
return NAUT_ERR_INVAL;
}
} else {
size_t tlen;
uint8_t *tor = slurp(argv[1], &tlen);
if (!tor) { NAUT_ERROR("read torrent"); return 1; }
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 1;
return NAUT_ERR_PROTO;
}
free(tor);
}
endpoint_t *endpoints = NULL;
size_t neps = 0, epcap = 0;
if (argc > 3) {
for (int i = 3; i < argc; i++) {
if (config->num_peers > 0) {
for (size_t i = 0; i < config->num_peers; i++) {
naut_peer_addr addr;
if (!endpoint_parse(argv[i], &addr)) {
NAUT_WARN("invalid peer address: %s", argv[i]);
if (!endpoint_parse(config->peers[i], &addr)) {
NAUT_WARN("invalid peer address: %s", config->peers[i]);
continue;
}
if (!endpoint_add(&endpoints, &neps, &epcap, &addr)) {
@ -532,7 +707,7 @@ int main(int argc, char **argv) {
naut_metainfo_free(&mi);
naut_magnet_free(&magnet);
free(endpoints);
return 1;
return NAUT_ERR_NOMEM;
}
}
} else {
@ -540,18 +715,21 @@ int main(int argc, char **argv) {
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;
uint64_t total = from_magnet ? 0 : (uint64_t)mi.total_length;
if (!discover_trackers(hash, total, trackers, num_trackers, peerid,
&endpoints, &neps, &epcap) ||
if (!discover_trackers(hash, total, trackers, num_trackers,
tracker_tiers, peerid, &endpoints, &neps,
&epcap) ||
(neps == 0 &&
!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 1;
return NAUT_ERR_NOMEM;
}
}
@ -574,7 +752,8 @@ int main(int argc, char **argv) {
free(info);
naut_magnet_free(&magnet);
free(endpoints);
return 1;
return metadata_error != NAUT_OK ? metadata_error :
NAUT_ERR_PROTO;
}
free(info);
NAUT_INFO("magnet metadata verified: %u pieces, %lld bytes",
@ -583,11 +762,11 @@ int main(int argc, char **argv) {
naut_magnet_free(&magnet);
if (neps == 0) {
NAUT_ERROR(argc > 3 ? "no valid peer addresses" :
"tracker and DHT discovery returned no peers");
NAUT_ERROR(config->num_peers > 0 ? "no valid peer addresses" :
"tracker and DHT discovery returned no peers");
naut_metainfo_free(&mi);
free(endpoints);
return 1;
return NAUT_ERR_NOTFOUND;
}
naut_err err;
@ -596,20 +775,22 @@ int main(int argc, char **argv) {
.preallocate = true,
};
naut_storage *st = naut_storage_open_opts(
mi.files, mi.num_files, argv[2], &storage_opts, &err);
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 1;
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 1;
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));
@ -628,7 +809,7 @@ int main(int argc, char **argv) {
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return 1;
return NAUT_ERR_NOMEM;
}
naut_download_set_worker_pool(d, workers);
@ -644,61 +825,109 @@ int main(int argc, char **argv) {
naut_storage_close(st);
naut_metainfo_free(&mi);
free(endpoints);
return 1;
return NAUT_ERR_NOMEM;
}
for (int i = 0; i < npeers; i++) peers[i].fd = -1;
int active = 0;
double t0 = now();
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, t0);
for (int i = 0; i < npeers; i++) {
int fd = connect_to(&endpoints[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;
}
peer_t *p = &peers[i];
p->fd = fd;
snprintf(p->name, sizeof p->name, "%s", endpoints[i].name);
p->peer_choking = true;
naut_pipeline_init(&p->pipeline, NAUT_BLOCK, 4, 1024, 32);
p->rcap = 1 << 18;
p->rbuf = malloc(p->rcap);
if (!p->rbuf || naut_bitfield_init(&p->have, mi.num_pieces) != NAUT_OK) {
free(p->rbuf);
p->rbuf = NULL;
close(fd);
p->fd = -1;
p->dead = true;
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++;
}
uint8_t hs[NAUT_HANDSHAKE_LEN];
naut_peer_handshake_build(hs, mi.infohash_v1, peerid, EXT_RESERVED);
uint8_t intr[5]; naut_peer_msg_simple(intr, NAUT_MSG_INTERESTED);
uint8_t *ext = NULL;
size_t ext_len = 0;
naut_err ext_error = naut_ext_build_handshake(
NAUT_EXT_UT_METADATA, NAUT_EXT_UT_PEX, 0, 0, &ext, &ext_len);
bool sent = ext_error == NAUT_OK &&
send_all(fd, hs, sizeof hs) &&
send_all(fd, ext, ext_len) &&
send_all(fd, intr, 5);
free(ext);
if (!sent) {
peer_drop(d, p);
continue;
}
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, t0);
int ready = poll(connect_fds, (nfds_t)neps, 100);
if (ready < 0) {
if (errno == EINTR) continue;
run_error = NAUT_ERR_IO;
break;
}
active++;
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: %d peers, %u pieces, %lld bytes", active, mi.num_pieces, (long long)mi.total_length);
NAUT_INFO("swarm: %u/%d peers connected, %u pieces, %lld bytes",
active, npeers, mi.num_pieces, (long long)mi.total_length);
double t0 = now();
naut_err run_error = NAUT_OK;
emit_event(config, NAUT_EVENT_TORRENT_ADDED, 0, NULL, NULL);
report_progress(config, d, &mi, (uint32_t)npeers, 0, active, failed, t0);
while (!naut_download_complete(d) && run_error == NAUT_OK) {
service_control(config, st);
if (stop_requested(config)) {
cancelled = true;
break;
}
int nf = 0;
for (int i = 0; i < npeers; i++) {
if (peers[i].dead) continue;
@ -717,10 +946,18 @@ int main(int argc, char **argv) {
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, 0,
(uint32_t)live_peers,
(uint32_t)npeers - (uint32_t)live_peers, t0);
int r = poll(pfd, nf, 200);
if (r < 0) {
if (errno == EINTR) continue;
run_error = NAUT_ERR_IO;
break;
}
int r = poll(pfd, nf, 2000);
if (r < 0) { if (errno == EINTR) continue; break; }
for (int k = 0; k < nf; k++) {
if (idx_map[k] < 0) {
@ -769,6 +1006,13 @@ int main(int argc, char **argv) {
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, 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));
@ -777,6 +1021,7 @@ int main(int argc, char **argv) {
done:
ok = naut_download_complete(d);
service_control(config, st);
naut_storage_sync(st);
for (int i = 0; i < npeers; i++) {
if (peers[i].blocks_received)
@ -797,5 +1042,26 @@ done:
free(peers); free(pfd); free(idx_map);
naut_worker_pool_destroy(workers);
naut_download_destroy(d); naut_storage_close(st); naut_metainfo_free(&mi);
return ok ? 0 : 1;
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