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

nautd/webui: scripting, labels, settings, set-location, pause fix

Browse source 
Session checkpoint on webui-plugin:
- engine dump (nautctl dump) + engine endgame integration
- per-file move locations persistence; torrent-level "Set location"
  with reset/keep-relative/leave-separate handling + residual prune
- Lua: naut.get_labels, define_settings/get_setting (script_host struct)
- daemon-owned labels (category+tags) + taxonomy persistence; webui write-through
- fix: pausing a completed/seeding torrent now sticks (stop wins over result)
- automation tab responsive layout; anime_sort label gating + settings

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
ookami125 2026-06-21 23:19:41 -04:00
parent 6dc711cf57
commit b633b7d216
40 changed files with 3305 additions and 3267 deletions
Download patch file Download diff file Expand all files Collapse all files

214
apps/echo/main.c
View file

@ -1,214 +0,0 @@
/* naut_echo — Phase 1 gate.
*
* A single-reactor io_uring echo server that proves the foundation works end to
* end: multishot accept, recv/send driven entirely off the page-aligned buffer
* pool with ZERO per-operation allocation in steady state. Throughput on
* loopback should be limited by memory bandwidth / the single core, not by the
* allocator or syscalls.
*
* It is intentionally one-in-flight-op-per-connection (recv -> send -> recv).
* The real peer reactor (later phase) uses multishot recv + provided buffers
* and pipelines; this is the minimal honest exercise of the primitives.
*
* usage: naut_echo [port] (default 9000)
*/
#include "naut/uring.h"
#include "naut/net.h"
#include "naut/buf.h"
#include "naut/log.h"
#include "naut/system.h"
#include <liburing.h>
#include <errno.h>
#include <signal.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#define ECHO_BLOCK (128u * 1024u)
#define ECHO_BUFS 4096u
#define RING_ENTRIES 4096u
/* user_data tagging: low 3 bits = op, high bits = conn* (16-byte aligned). */
enum { TAG_ACCEPT = 1, TAG_RECV = 2, TAG_SEND = 3 };
#define UD(p, tag) ((__u64)(uintptr_t)(p) | (unsigned)(tag))
#define UD_TAG(ud) ((unsigned)((ud) & 0x7u))
#define UD_PTR(ud) ((conn *)(uintptr_t)((ud) & ~(__u64)0x7u))
typedef struct conn {
int fd;
uint32_t sent; /* bytes of buf->len already written (partial sends) */
naut_buf *buf;
bool awaiting_notif;
bool recv_fixed; /* the in-flight recv used the fixed buffer */
} conn;
static volatile sig_atomic_t g_stop = 0;
static void on_signal(int s) { (void)s; g_stop = 1; }
static naut_bufpool *g_pool;
static _Atomic uint64_t g_bytes = 0, g_conns = 0, g_zc_copied = 0;
static void arm_recv(naut_ring *owner, conn *c) {
struct io_uring_sqe *sqe = io_uring_get_sqe(&owner->ring);
c->recv_fixed =
naut_ring_prep_recv(owner, sqe, c->fd, c->buf->data, c->buf->cap, 0);
io_uring_sqe_set_data64(sqe, UD(c, TAG_RECV));
}
static void arm_send(naut_ring *owner, conn *c) {
struct io_uring_sqe *sqe = io_uring_get_sqe(&owner->ring);
c->awaiting_notif = naut_ring_prep_send(
owner, sqe, c->fd, c->buf->data + c->sent,
c->buf->len - c->sent, MSG_NOSIGNAL, true);
io_uring_sqe_set_data64(sqe, UD(c, TAG_SEND));
}
static void conn_close(conn *c) {
close(c->fd);
naut_buf_put(c->buf);
free(c);
}
int main(int argc, char **argv) {
uint16_t port = (argc > 1) ? (uint16_t)atoi(argv[1]) : 9000;
int cpu = getenv("NAUT_CPU") ? atoi(getenv("NAUT_CPU")) : -1;
int numa_node =
getenv("NAUT_NUMA_NODE") ? atoi(getenv("NAUT_NUMA_NODE")) : -1;
bool sqpoll = getenv("NAUT_SQPOLL") != NULL;
bool hugepages = getenv("NAUT_HUGEPAGES") != NULL;
signal(SIGINT, on_signal);
signal(SIGTERM, on_signal);
signal(SIGPIPE, SIG_IGN);
if (cpu >= 0 && naut_pin_current_thread(cpu) != NAUT_OK)
NAUT_WARN("failed to pin reactor to CPU %d", cpu);
naut_ring r;
if (naut_ring_init_cpu(&r, RING_ENTRIES, sqpoll, cpu) != NAUT_OK)
return 1;
if (naut_ring_probe(&r) != NAUT_OK) { naut_ring_close(&r); return 1; }
struct io_uring *ring = &r.ring;
int lfd = naut_net_listen(port, 1024, true);
if (lfd < 0) { naut_ring_close(&r); return 1; }
g_pool = naut_bufpool_create_on_node(
ECHO_BLOCK, ECHO_BUFS, hugepages, numa_node);
if (!g_pool) { close(lfd); naut_ring_close(&r); return 1; }
(void)naut_ring_register_bufpool(&r, g_pool);
/* prime the multishot accept */
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
io_uring_prep_multishot_accept(sqe, lfd, NULL, NULL, 0);
io_uring_sqe_set_data64(sqe, UD(NULL, TAG_ACCEPT));
NAUT_INFO("echo listening on :%u", port);
while (!g_stop) {
int rc = io_uring_submit_and_wait(ring, 1);
if (rc < 0 && rc != -EINTR) { NAUT_ERROR("submit_and_wait: %s", strerror(-rc)); break; }
unsigned head, count = 0;
struct io_uring_cqe *cqe;
io_uring_for_each_cqe(ring, head, cqe) {
count++;
__u64 ud = cqe->user_data;
int res = cqe->res;
switch (UD_TAG(ud)) {
case TAG_ACCEPT: {
if (res < 0) {
if (res != -ECANCELED) NAUT_WARN("accept: %s", strerror(-res));
} else {
int cfd = res;
naut_net_tune_peer(cfd);
naut_buf *b = naut_buf_get(g_pool);
if (!b) { NAUT_WARN("pool exhausted, dropping conn"); close(cfd); }
else {
conn *c = calloc(1, sizeof(*c));
c->fd = cfd; c->buf = b;
atomic_fetch_add(&g_conns, 1);
arm_recv(&r, c);
}
}
/* re-arm if the kernel dropped the multishot registration */
if (!(cqe->flags & IORING_CQE_F_MORE)) {
struct io_uring_sqe *s = io_uring_get_sqe(ring);
io_uring_prep_multishot_accept(s, lfd, NULL, NULL, 0);
io_uring_sqe_set_data64(s, UD(NULL, TAG_ACCEPT));
}
break;
}
case TAG_RECV: {
conn *c = UD_PTR(ud);
if (res <= 0) {
/* A fixed-buffer recv rejected with -EINVAL means this
* kernel doesn't support IORING_RECVSEND_FIXED_BUF on plain
* recv. Disable it ring-wide and retry THIS connection
* unfixed. We key off the per-conn flag, not the ring flag,
* so every connection that armed a fixed recv before the
* flag flipped recovers too (otherwise all but the first
* would be torn down). */
if (res == -EINVAL && c->recv_fixed) {
if (r.recv_fixed) {
NAUT_WARN("fixed-buffer recv unsupported at runtime; "
"falling back to normal recv");
r.recv_fixed = false;
}
arm_recv(&r, c);
break;
}
if (res < 0)
NAUT_WARN("recv completion: %s", strerror(-res));
conn_close(c);
break;
}
c->buf->len = (uint32_t)res;
c->sent = 0;
arm_send(&r, c);
break;
}
case TAG_SEND: {
conn *c = UD_PTR(ud);
if (cqe->flags & IORING_CQE_F_NOTIF) {
if (res & IORING_NOTIF_USAGE_ZC_COPIED) {
uint64_t copied =
atomic_fetch_add(&g_zc_copied, 1) + 1;
if (copied == 8) {
NAUT_WARN("SEND_ZC is copying on this transport; "
"disabling it for this ring");
r.send_zc = false;
}
}
c->awaiting_notif = false;
if (c->sent < c->buf->len) arm_send(&r, c);
else { c->buf->len = 0; arm_recv(&r, c); }
break;
}
if (res <= 0) { conn_close(c); break; }
c->sent += (uint32_t)res;
atomic_fetch_add(&g_bytes, (uint64_t)res);
if (!c->awaiting_notif) {
if (c->sent < c->buf->len) arm_send(&r, c);
else { c->buf->len = 0; arm_recv(&r, c); }
}
break;
}
default:
NAUT_PANIC("bad user_data tag %u", UD_TAG(ud));
}
}
io_uring_cq_advance(ring, count);
}
NAUT_INFO("shutting down: %llu conns, %llu bytes echoed, %llu SEND_ZC copied notifications",
(unsigned long long)atomic_load(&g_conns),
(unsigned long long)atomic_load(&g_bytes),
(unsigned long long)atomic_load(&g_zc_copied));
close(lfd);
naut_ring_unregister_buffers(&r);
naut_bufpool_destroy(g_pool);
naut_ring_close(&r);
return 0;
}

212
apps/leech/main.c
View file

@ -1,212 +0,0 @@
/* naut_leech — Phase 3 gate: download a torrent from a single peer and write a
* byte-correct, hash-verified file to disk.
*
* Blocking-socket driver around the sans-IO peer codec + download engine. The
* point of this phase is protocol correctness and interop (it downloads from a
* libtorrent seed in the integration test), not peak throughput — the io_uring
* reactor that drives thousands of these comes in Phase 6.
*
* usage: naut_leech [--mse] <file.torrent> <output-dir> <ip> <port>
*/
#include "naut/metainfo.h"
#include "naut/storage.h"
#include "naut/piece.h"
#include "naut/peer.h"
#include "naut/mse.h"
#include "naut/log.h"
#include <arpa/inet.h>
#include <errno.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 PIPELINE_DEPTH 512 /* outstanding requests (~8 MiB in flight) */
static double now(void) {
struct timespec t; clock_gettime(CLOCK_MONOTONIC, &t);
return t.tv_sec + t.tv_nsec * 1e-9;
}
static uint8_t *slurp(const char *path, size_t *len) {
FILE *f = fopen(path, "rb");
if (!f) { NAUT_ERROR("open %s: %s", path, strerror(errno)); 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 int connect_peer(const char *ip, uint16_t port) {
int fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd < 0) return -1;
struct sockaddr_in a; memset(&a, 0, sizeof a);
a.sin_family = AF_INET; a.sin_port = htons(port);
if (inet_pton(AF_INET, ip, &a.sin_addr) != 1) { close(fd); return -1; }
if (connect(fd, (struct sockaddr *)&a, sizeof a) != 0) {
NAUT_ERROR("connect %s:%u: %s", ip, port, strerror(errno));
close(fd); return -1;
}
int one = 1; setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
return fd;
}
/* send up to PIPELINE_DEPTH outstanding requests */
static bool refill(int fd, naut_mse_stream *mse,
naut_download *d, int *outstanding) {
uint32_t idx, begin, len;
while (*outstanding < PIPELINE_DEPTH) {
if (!naut_download_next_request(d, &idx, &begin, &len)) break;
uint8_t req[17];
naut_peer_msg_request(req, idx, begin, len);
if (!naut_mse_send_all(fd, mse, req, sizeof req)) return false;
(*outstanding)++;
}
return true;
}
int main(int argc, char **argv) {
bool use_mse = argc > 1 && strcmp(argv[1], "--mse") == 0;
int arg = use_mse ? 2 : 1;
if (argc - arg != 4) {
fprintf(stderr, "usage: %s [--mse] <file.torrent> <output-dir> <ip> <port>\n",
argv[0]);
return 2;
}
naut_log_set_level(NAUT_LOG_INFO);
size_t tlen;
uint8_t *tor = slurp(argv[arg], &tlen);
if (!tor) return 1;
naut_metainfo mi;
if (naut_metainfo_parse(tor, tlen, &mi) != NAUT_OK) { NAUT_ERROR("bad torrent"); return 1; }
free(tor);
char hex[41]; naut_infohash_v1_hex(&mi, hex);
NAUT_INFO("torrent '%s': %u pieces, %lld bytes, infohash %s",
mi.name, mi.num_pieces, (long long)mi.total_length, hex);
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, argv[arg + 1], &storage_opts, &err);
if (!st) { NAUT_ERROR("storage: %s", naut_strerror(err)); return 1; }
naut_download *d = naut_download_create(&mi, st);
if (!d) return 1;
int fd = connect_peer(argv[arg + 2], (uint16_t)atoi(argv[arg + 3]));
if (fd < 0) return 1;
/* handshake */
uint8_t peerid[20]; memcpy(peerid, "-NT0001-", 8);
for (int i = 8; i < 20; i++) peerid[i] = (uint8_t)(rand() & 0xff);
uint8_t hs[NAUT_HANDSHAKE_LEN];
naut_peer_handshake_build(hs, mi.infohash_v1, peerid, 0);
naut_mse_stream mse = {0};
uint8_t remote_hs[NAUT_HANDSHAKE_LEN];
bool hs_done = false;
if (use_mse) {
naut_err mse_err = naut_mse_client_handshake(
fd, mi.infohash_v1, peerid, 0, &mse, remote_hs);
if (mse_err != NAUT_OK) {
NAUT_ERROR("MSE handshake failed: %s", naut_strerror(mse_err));
return 1;
}
hs_done = true;
NAUT_INFO("MSE/RC4 peer transport established");
uint8_t intr[5];
naut_peer_msg_simple(intr, NAUT_MSG_INTERESTED);
if (!naut_mse_send_all(fd, &mse, intr, sizeof intr)) {
NAUT_ERROR("interested send failed");
return 1;
}
} else if (!naut_mse_send_all(fd, &mse, hs, sizeof hs)) {
NAUT_ERROR("handshake send failed");
return 1;
}
/* recv buffer */
size_t cap = 4u << 20, len = 0;
uint8_t *buf = malloc(cap);
bool unchoked = false;
int outstanding = 0;
double t0 = now();
while (!naut_download_complete(d)) {
if (len == cap) { cap *= 2; buf = realloc(buf, cap); }
ssize_t r = naut_mse_recv(fd, &mse, buf + len, cap - len);
if (r < 0) { NAUT_ERROR("recv: %s", strerror(errno)); break; }
if (r == 0) { NAUT_ERROR("peer closed (%.1f%% done)",
100.0 * naut_download_pieces_done(d) / mi.num_pieces); break; }
len += (size_t)r;
size_t pos = 0;
if (!hs_done) {
if (len < NAUT_HANDSHAKE_LEN) continue;
uint8_t ih[20], pid[20];
if (!naut_peer_handshake_parse(buf, ih, pid, NULL) ||
memcmp(ih, mi.infohash_v1, 20) != 0) {
NAUT_ERROR("handshake mismatch"); break;
}
pos = NAUT_HANDSHAKE_LEN;
hs_done = true;
uint8_t intr[5]; naut_peer_msg_simple(intr, NAUT_MSG_INTERESTED);
if (!naut_mse_send_all(fd, &mse, intr, 5)) break;
}
/* parse all complete messages */
for (;;) {
naut_msg m;
int c = naut_peer_msg_parse(buf + pos, len - pos, &m);
if (c == 0) break;
if (c < 0) { NAUT_ERROR("protocol error"); goto done; }
pos += (size_t)c;
switch (m.type) {
case NAUT_MSG_UNCHOKE: unchoked = true; break;
case NAUT_MSG_CHOKE: unchoked = false; break;
case NAUT_MSG_PIECE: {
outstanding--;
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) { NAUT_ERROR("block rejected: %s", naut_strerror(e)); goto done; }
break;
}
default: break; /* bitfield/have/keepalive/port: ignore for a seed */
}
}
/* compact consumed bytes */
memmove(buf, buf + pos, len - pos);
len -= pos;
if (unchoked && !refill(fd, &mse, d, &outstanding)) {
NAUT_ERROR("request send failed"); break;
}
}
done:;
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, %.1f MB in %.2fs (%.1f MB/s), all SHA-1 verified",
naut_download_pieces_done(d), mi.num_pieces, mb, dt, mb / dt);
} else {
NAUT_ERROR("INCOMPLETE: %u/%u pieces", naut_download_pieces_done(d), mi.num_pieces);
}
naut_storage_sync(st);
close(fd);
naut_download_destroy(d);
naut_storage_close(st);
naut_metainfo_free(&mi);
free(buf);
return ok ? 0 : 1;
}

21
apps/nautctl/main.c
View file

@ -17,6 +17,7 @@ static void usage(const char *program) {
" add SOURCE OUTPUT [IP:PORT ...]\n"
" list\n"
" show TORRENT_ID\n"
" dump TORRENT_ID\n"
" remove TORRENT_ID\n"
" script PATH | unscript\n"
" status | events | shutdown\n"
@ -93,6 +94,7 @@ int main(int argc, char **argv) {
return stream_events(socket_path);
json_t *params = NULL;
bool raw_dump = false;
if (strcmp(method, "add") == 0) {
if (arg + 1 >= argc) { usage(argv[0]); return 2; }
method = "add_torrent";
@ -113,14 +115,16 @@ int main(int argc, char **argv) {
method = "torrents";
params = json_object();
} else if (strcmp(method, "show") == 0 ||
strcmp(method, "dump") == 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";
if (strcmp(method, "show") == 0) method = "torrent";
else if (strcmp(method, "dump") == 0) { method = "dump_torrent"; raw_dump = true; }
else method = "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; }
@ -145,8 +149,19 @@ int main(int argc, char **argv) {
fprintf(stderr, "nautctl: RPC failed: %s\n", naut_strerror(error));
return 1;
}
int result = print_json(reply);
bool ok = json_is_true(json_object_get(reply, "ok"));
int result;
/* `dump` returns a multi-line text blob; print it raw instead of escaped JSON. */
const char *dump = raw_dump
? json_string_value(json_object_get(
json_object_get(reply, "result"), "dump"))
: NULL;
if (dump) {
fputs(dump, stdout);
result = 0;
} else {
result = print_json(reply);
}
json_decref(reply);
return result || !ok;
}

1882
apps/nautd/main.c
View file

File diff suppressed because it is too large Load diff

1157
apps/swarm/main.c
View file

File diff suppressed because it is too large Load diff