Cieric/Naut
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions
Naut/apps/nautd/main.c
ookami125 6ae72907b0 Expose peer and piece state to web UI
2026-06-17 01:29:44 -04:00

1014 lines
36 KiB
C
Raw Blame History

#include "naut/event.h"
#include "naut/log.h"
#include "naut/plugin.h"
#include "naut/rpc.h"
#include "naut/script.h"
#include "naut/storage.h"
#include "naut/swarm.h"
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <poll.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#define DEFAULT_SOCKET "/tmp/nautd.sock"
#define MOVE_QUEUE_CAPACITY 64
#define MAX_SUBSCRIBERS 64
#define MAX_TORRENTS 128
typedef struct {
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 {
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;
};
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_t *piece_states = json_array();
if (piece_states) {
uint32_t state_count = task->stats.piece_state_count;
if (state_count > NAUT_SWARM_MAX_PIECE_STATS)
state_count = NAUT_SWARM_MAX_PIECE_STATS;
for (uint32_t i = 0; i < state_count; i++)
json_array_append_new(piece_states,
json_integer(task->stats.piece_states[i]));
json_object_set_new(result, "piece_states", piece_states);
}
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_t *peer_list = json_array();
if (peer_list) {
uint32_t peer_count = task->stats.peer_count;
if (peer_count > NAUT_SWARM_MAX_PEER_STATS)
peer_count = NAUT_SWARM_MAX_PEER_STATS;
for (uint32_t i = 0; i < peer_count; i++) {
const naut_swarm_peer_stats *peer =
&task->stats.peer_stats[i];
json_t *item = json_pack(
"{s:s,s:i,s:s,s:s,s:s,s:f,s:f,s:I,s:I,s:I,s:I}",
"ip", peer->ip,
"port", (int)peer->port,
"client", peer->client,
"connection", peer->connection,
"flags", peer->flags,
"progress", peer->progress,
"relevance", peer->relevance,
"downloaded", (json_int_t)peer->downloaded,
"uploaded", (json_int_t)peer->uploaded,
"dlspeed", (json_int_t)peer->dlspeed,
"upspeed", (json_int_t)peer->upspeed);
if (item) json_array_append_new(peer_list, item);
}
json_object_set_new(result, "peer_list", peer_list);
}
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;
json_t *result = json_object();
if (!result) {
*error = NAUT_ERR_NOMEM;
return NULL;
}
json_object_set_new(result, "protocol", json_integer(NAUT_RPC_VERSION));
json_object_set_new(result, "service", json_string("nautd"));
*error = NAUT_OK;
return result;
}
static json_t *rpc_status(void *opaque, const json_t *params,
naut_err *error) {
(void)params;
daemon_state *state = opaque;
naut_script_stats stats = {0};
if (state->script) naut_script_get_stats(state->script, &stats);
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) {
json_decref(result);
json_decref(script);
*error = NAUT_ERR_NOMEM;
return NULL;
}
json_object_set_new(result, "protocol", json_integer(NAUT_RPC_VERSION));
json_object_set_new(result, "plugins",
json_integer((json_int_t)naut_plugin_count(
state->plugins)));
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));
json_object_set_new(script, "errors", json_integer(stats.errors));
json_object_set_new(script, "move_requests",
json_integer(stats.move_requests));
json_object_set_new(result, "script", script);
json_object_set_new(result, "move_commands", json_integer(moves));
json_object_set_new(result, "pending_move_commands",
json_integer((json_int_t)pending));
*error = NAUT_OK;
return result;
}
static json_t *rpc_plugins(void *opaque, const json_t *params,
naut_err *error) {
(void)params;
daemon_state *state = opaque;
json_t *plugins = json_array();
json_t *storage = json_array();
if (!plugins || !storage) {
json_decref(plugins);
json_decref(storage);
*error = NAUT_ERR_NOMEM;
return NULL;
}
for (size_t i = 0; i < naut_plugin_count(state->plugins); i++)
json_array_append_new(plugins,
json_string(naut_plugin_name(state->plugins, i)));
for (size_t i = 0; i < naut_plugin_storage_count(state->plugins); i++)
json_array_append_new(storage,
json_string(naut_plugin_storage_name(state->plugins, i)));
json_t *result = json_object();
if (!result) {
json_decref(plugins);
json_decref(storage);
*error = NAUT_ERR_NOMEM;
return NULL;
}
json_object_set_new(result, "plugins", plugins);
json_object_set_new(result, "storage_backends", storage);
*error = NAUT_OK;
return result;
}
static json_t *rpc_emit(void *opaque, const json_t *params, naut_err *error) {
daemon_state *state = opaque;
if (!json_is_object(params)) {
*error = NAUT_ERR_INVAL;
return NULL;
}
const char *type_name = json_string_value(json_object_get(params, "type"));
naut_event_type type;
if (!type_name || !naut_event_type_parse(type_name, &type)) {
*error = NAUT_ERR_INVAL;
return NULL;
}
json_int_t torrent_id =
json_integer_value(json_object_get(params, "torrent_id"));
json_int_t index = json_integer_value(json_object_get(params, "index"));
if (torrent_id < 0 || index < 0 || (uint64_t)index > UINT32_MAX) {
*error = NAUT_ERR_RANGE;
return NULL;
}
naut_event event = {
.type = type,
.torrent_id = (uint64_t)torrent_id,
.index = (uint32_t)index,
.message = json_string_value(json_object_get(params, "message")),
.path = json_string_value(json_object_get(params, "path")),
};
naut_event_emit(state->events, &event);
*error = NAUT_OK;
return json_true();
}
static json_t *rpc_shutdown(void *opaque, const json_t *params,
naut_err *error) {
(void)params;
daemon_state *state = opaque;
state->stopping = true;
*error = NAUT_OK;
return json_true();
}
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;
}
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; }
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;
}
/* 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;
}
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;
}
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;
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 = (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 broadcast_event(void *opaque, const naut_event *event) {
daemon_state *state = opaque;
json_t *payload = naut_rpc_event_json(event);
if (!payload) return;
pthread_mutex_lock(&state->subscriber_lock);
for (size_t i = 0; i < state->subscriber_count;) {
if (naut_rpc_send_json(state->subscribers[i], NAUT_RPC_EVENT,
payload) == NAUT_OK) {
i++;
continue;
}
close(state->subscribers[i]);
state->subscribers[i] =
state->subscribers[--state->subscriber_count];
}
pthread_mutex_unlock(&state->subscriber_lock);
json_decref(payload);
}
static int listen_unix(const char *path) {
int fd = socket(AF_UNIX, SOCK_STREAM, 0);
if (fd < 0) return -1;
struct sockaddr_un address;
memset(&address, 0, sizeof address);
address.sun_family = AF_UNIX;
if (strlen(path) >= sizeof address.sun_path) {
close(fd);
return -1;
}
strcpy(address.sun_path, path);
unlink(path);
if (bind(fd, (struct sockaddr *)&address, sizeof address) != 0 ||
listen(fd, 32) != 0) {
close(fd);
return -1;
}
return fd;
}
static json_t *response(bool ok, json_t *result, naut_err error) {
json_t *reply = json_object();
if (!reply) return NULL;
json_object_set_new(reply, "ok", json_boolean(ok));
if (ok) {
json_object_set(reply, "result", result ? result : json_null());
} else {
json_object_set_new(reply, "code", json_integer(error));
json_object_set_new(reply, "error",
json_string(naut_strerror(error)));
}
return reply;
}
static void add_subscriber(daemon_state *state, int fd) {
int flags = fcntl(fd, F_GETFL, 0);
if (flags >= 0) fcntl(fd, F_SETFL, flags | O_NONBLOCK);
pthread_mutex_lock(&state->subscriber_lock);
if (state->subscriber_count < MAX_SUBSCRIBERS) {
state->subscribers[state->subscriber_count++] = fd;
fd = -1;
}
pthread_mutex_unlock(&state->subscriber_lock);
if (fd >= 0) close(fd);
}
static void handle_client(daemon_state *state, int fd) {
struct timeval timeout = {.tv_sec = 2};
setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout);
naut_rpc_frame_type type;
json_t *request = NULL;
naut_err error = naut_rpc_recv_json(fd, &type, &request);
if (error != NAUT_OK || type != NAUT_RPC_REQUEST ||
!json_is_object(request)) {
json_decref(request);
close(fd);
return;
}
const char *method =
json_string_value(json_object_get(request, "method"));
json_t *params = json_object_get(request, "params");
if (method && strcmp(method, "subscribe") == 0) {
json_t *subscribed = json_string("subscribed");
json_t *reply = response(true, subscribed, NAUT_OK);
json_decref(subscribed);
if (reply &&
naut_rpc_send_json(fd, NAUT_RPC_RESPONSE, reply) == NAUT_OK)
add_subscriber(state, fd);
else
close(fd);
json_decref(reply);
json_decref(request);
return;
}
if (!method) error = NAUT_ERR_INVAL;
json_t *result = method
? naut_rpc_dispatch(state->rpc, method, params, &error) : NULL;
json_t *reply = response(error == NAUT_OK && result, result, error);
json_decref(result);
if (reply) {
naut_rpc_send_json(fd, NAUT_RPC_RESPONSE, reply);
json_decref(reply);
}
json_decref(request);
close(fd);
}
static bool register_commands(daemon_state *state) {
return naut_rpc_register(state->rpc, "ping", rpc_ping, state) == NAUT_OK &&
naut_rpc_register(state->rpc, "status", rpc_status, state) == NAUT_OK &&
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",
program);
}
int main(int argc, char **argv) {
const char *socket_path = DEFAULT_SOCKET;
const char *script_path = NULL;
const char *plugin_paths[64];
size_t plugin_count = 0;
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "--socket") == 0 && i + 1 < argc)
socket_path = argv[++i];
else if (strcmp(argv[i], "--plugin") == 0 && i + 1 < argc &&
plugin_count < NAUT_ARRAY_LEN(plugin_paths))
plugin_paths[plugin_count++] = argv[++i];
else if (strcmp(argv[i], "--script") == 0 && i + 1 < argc)
script_path = argv[++i];
else {
usage(argv[0]);
return 2;
}
}
signal(SIGINT, on_signal);
signal(SIGTERM, on_signal);
signal(SIGPIPE, SIG_IGN);
daemon_state state = {0};
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);
if (!state.events || !state.rpc || !state.plugins ||
!register_commands(&state)) {
fprintf(stderr, "nautd: failed to initialize control plane\n");
return 1;
}
for (size_t i = 0; i < plugin_count; i++) {
if (naut_plugin_load(state.plugins, plugin_paths[i]) != NAUT_OK) {
fprintf(stderr, "nautd: failed to load plugin %s\n",
plugin_paths[i]);
return 1;
}
}
if (script_path) {
naut_err error;
state.script = naut_script_create(state.events, script_path, 256,
queue_move, &state, &error);
if (!state.script) {
fprintf(stderr, "nautd: failed to load script %s: %s\n",
script_path, naut_strerror(error));
return 1;
}
}
uint64_t event_subscription;
if (naut_event_subscribe(state.events, broadcast_event, &state,
&event_subscription) != NAUT_OK)
return 1;
int listener = listen_unix(socket_path);
if (listener < 0) {
perror("nautd: listen");
return 1;
}
NAUT_INFO("nautd listening on %s", socket_path);
while (!state.stopping && !interrupted) {
struct pollfd pollfd = {.fd = listener, .events = POLLIN};
int ready = poll(&pollfd, 1, 100);
if (ready > 0 && (pollfd.revents & POLLIN)) {
int client = accept(listener, NULL, NULL);
if (client >= 0) handle_client(&state, client);
} else if (ready < 0 && errno != EINTR) {
break;
}
reap_torrents(&state);
}
close(listener);
unlink(socket_path);
/* Tear down plugins first: a plugin like webui runs its own threads that
* call back into the daemon via RPC, so it must be stopped (and its
* threads joined) before we free the torrent tasks those calls touch. */
naut_plugin_manager_destroy(state.plugins);
state.plugins = NULL;
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_rpc_registry_destroy(state.rpc);
naut_event_bus_destroy(state.events);
pthread_mutex_destroy(&state.subscriber_lock);
pthread_mutex_destroy(&state.torrent_lock);
return 0;
}
Reference in a new issue View git blame Copy permalink