jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Unify media source and relay transmission with owned buffers

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 19:11:26 +02:00
1 parent f15d53947e
commit 9b4974f02d
2 files changed
+324 -387
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.c
+311 -380
View File
@@ -20,12 +20,14 @@
#include "../lib/sqlite3.h"
#include <string.h>
#include <time.h>
#include <errno.h>
#include "../lib/platform_compat.h"
#define MD_ID "media_delivery"
/* ── forward decls ── */
static void md_relay_cancel_block(struct media_delivery_ctx* md, struct relay_block_ctx* block);
static void md_block_wake(struct media_block_source* source);
static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry);
static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg);
static void md_on_bgp_node(struct TOPO_GROUP* group, uint64_t node_id, int event, void* arg);
@@ -246,44 +248,95 @@ static void md_super_peer_remove(struct media_delivery_ctx* md, uint64_t node_id
queue_entry_free(e);
}
/* ── relay block context helpers ── */
/* Назначения общие для загрузок и исходящих фоновых подготовок. */
struct media_destination {
struct media_destination* next;
char path[MEDIA_PATH_MAX];
};
struct relay_block_ctx* md_relay_find(struct media_delivery_ctx* md, const uint8_t* block_id) {
if (!md->relay_blocks) return NULL;
struct ll_entry* e = queue_find_data_by_index(md->relay_blocks, block_id);
return e ? (struct relay_block_ctx*)e->data : NULL;
}
struct relay_block_ctx* md_relay_add(struct media_delivery_ctx* md,
const uint8_t* block_id,
const uint8_t* media_id,
uint32_t chunk) {
struct relay_block_ctx* rc = md_relay_find(md, block_id);
if (rc) return rc;
if (!md->relay_blocks) {
md->relay_blocks = queue_new(md->inst->ua, 64, offsetof(struct relay_block_ctx, block_id), 16, "md_relay");
if (!md->relay_blocks) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_new(relay_blocks) failed", MD_ID); return NULL; }
struct media_destination* media_delivery_reserve(struct UTUN_INSTANCE* inst, const char* path) {
char base[MEDIA_BASE_MAX], location[MEDIA_PATH_MAX], normalized[MEDIA_PATH_MAX];
if (!inst || !inst->md.initialized || media_index_get_base(inst->topo_sqlite_db, base)
|| media_index_location(path, base, location) || media_index_path(base, location, normalized)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: destination reservation failed path=%s", MD_ID, path ? path : "(null)");
return NULL;
}
struct ll_entry* qe = queue_entry_new(sizeof(struct relay_block_ctx));
if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new(relay_block) failed", MD_ID); return NULL; }
rc = (struct relay_block_ctx*)qe->data;
memset(rc, 0, sizeof(*rc));
memcpy(rc->block_id, block_id, 16);
memcpy(rc->media_id, media_id, 16);
rc->chunk = chunk;
rc->inst = md->inst;
memcpy(qe->data, block_id, 16);
queue_data_put_with_index(md->relay_blocks, qe);
return rc;
for (struct media_destination* d = inst->md.destinations; d; d = d->next) {
if (!strcmp(d->path, normalized)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: destination busy path=%s", MD_ID, normalized); return NULL;
}
}
struct media_destination* d = u_calloc(1, sizeof(*d));
if (!d) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: destination allocation failed", MD_ID); return NULL; }
memcpy(d->path, normalized, strlen(normalized) + 1);
d->next = inst->md.destinations; inst->md.destinations = d;
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: reserved destination=%s", MD_ID, normalized);
return d;
}
void md_relay_remove(struct media_delivery_ctx* md, const uint8_t* block_id) {
if (!md->relay_blocks) return;
struct ll_entry* e = queue_find_data_by_index(md->relay_blocks, block_id);
if (!e) return;
md_relay_cancel_block(md, (struct relay_block_ctx*)e->data);
queue_remove_data(md->relay_blocks, e);
queue_entry_free(e);
void media_delivery_release(struct UTUN_INSTANCE* inst, struct media_destination* destination) {
if (!destination) return;
struct media_destination** slot = &inst->md.destinations;
while (*slot && *slot != destination) slot = &(*slot)->next;
if (!*slot) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: unknown destination reservation", MD_ID); return; }
*slot = destination->next;
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: released destination=%s", MD_ID, destination->path);
u_free(destination);
}
static void md_block_detach(struct media_block_source* source) {
if (!source->indexed) return;
queue_remove_data(source->md->block_sources, source->entry);
source->indexed = 0;
}
static struct media_block_source* md_block_find(struct media_delivery_ctx* md, const uint8_t block_id[16]) {
struct ll_entry* e = md->block_sources ? queue_find_data_by_index(md->block_sources, block_id) : NULL;
return e ? (struct media_block_source*)e->data : NULL;
}
void md_block_release(struct media_block_source* source) {
if (!source || --source->refs) return;
md_block_detach(source);
if (source->owned) file_io_temp_discard(&source->storage);
queue_entry_free(source->entry);
}
struct media_block_source* md_block_create(struct media_delivery_ctx* md, uint64_t group_id,
const uint8_t media_id[16], const uint8_t block_id[16], uint32_t chunk, uint32_t size, const char* destination) {
if (!md->block_sources)
md->block_sources = queue_new(md->inst->ua, 64, offsetof(struct media_block_source, block_id), 16, "md_sources");
if (!md->block_sources) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: source index allocation failed", MD_ID); return NULL; }
struct ll_entry* entry = queue_entry_new(sizeof(struct media_block_source));
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: block source allocation failed", MD_ID); return NULL; }
struct media_block_source* source = (struct media_block_source*)entry->data;
memset(source, 0, sizeof(*source));
source->entry = entry; source->md = md; source->refs = 1; source->owned = 1;
source->group_id = group_id; source->chunk = chunk; source->size = size;
memcpy(source->media_id, media_id, 16); memcpy(source->block_id, block_id, 16);
if (file_io_temp_open(&source->storage, destination) || file_io_temp_finish(&source->storage)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: block temp creation failed chunk=%u", MD_ID, chunk);
md_block_release(source); return NULL;
}
struct media_block_source* previous = md_block_find(md, block_id);
if (previous) md_block_detach(previous);
source->indexed = 1; queue_data_put_with_index(md->block_sources, entry);
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: block source created chunk=%u size=%u path=%s",
MD_ID, chunk, size, source->storage.temporary);
return source;
}
void md_block_publish(struct media_block_source* source, uint32_t available, int verified) {
source->available = available; source->verified = verified != 0;
md_block_wake(source);
}
void md_block_abort(struct media_block_source* source) {
if (!source || source->failed) return;
source->failed = 1; md_block_detach(source);
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: block source aborted chunk=%u available=%u/%u",
MD_ID, source->chunk, source->available, source->size);
md_block_wake(source);
}
/* ── file_load helpers (per-file source download tracking) ── */
@@ -688,377 +741,255 @@ static void md_super_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64
/* ── handle incoming BLOCK_REQ (admission control + streaming) ── */
#define MD_MAX_STREAMS 3
#define MD_CHUNK_SIZE 1024
struct stream_ctx {
struct stream_ctx* next;
struct media_delivery_ctx* md;
uint64_t group_id;
uint64_t dst_node_id;
uint8_t media_id[16];
uint8_t block_id[16];
uint32_t chunk;
FILE* file;
uint64_t offset;
uint64_t remaining;
uint64_t block_start;
size_t block_data_len;
struct media_block_source* source;
uint64_t group_id, dst_node_id;
uint8_t transfer_id[16];
uint32_t offset;
uint64_t last_progress_tb;
void* pump;
void* timer;
struct queue_waiter_handle waiter;
};
static void stream_pump(void* arg);
static void stream_free(struct stream_ctx* sc) {
struct media_delivery_ctx* md = sc->md;
etcp_router_cancel_send_ready(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, &sc->waiter);
struct stream_ctx** p = &md->streams;
while (*p && *p != sc) p = &(*p)->next;
if (*p) *p = sc->next;
md_file_load_dec(md, sc->media_id, sc->dst_node_id);
if (sc->pump) uasync_call_soon_cancel(md->inst->ua, sc->pump);
if (sc->timer) uasync_cancel_timeout(md->inst->ua, sc->timer);
struct stream_ctx** slot = &md->streams;
while (*slot && *slot != sc) slot = &(*slot)->next;
if (*slot) *slot = sc->next;
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: TX released peer=%016llx chunk=%u sent=%u/%u attempt=%02x%02x",
MD_ID, (unsigned long long)sc->dst_node_id, sc->source->chunk, sc->offset, sc->source->size,
sc->transfer_id[0], sc->transfer_id[1]);
md_file_load_dec(md, sc->source->media_id, sc->dst_node_id);
media_delivery_stream_done(md->inst);
if (sc->file) fclose(sc->file);
md_block_release(sc->source);
u_free(sc);
}
static void stream_send_chunk_cb(struct ll_queue* q, void* arg) {
(void)q;
struct stream_ctx* sc = (struct stream_ctx*)arg;
static int stream_schedule(struct stream_ctx* sc) {
if (!sc->pump) sc->pump = uasync_call_soon(sc->md->inst->ua, sc, stream_pump);
if (sc->pump) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: cannot schedule TX peer=%016llx chunk=%u",
MD_ID, (unsigned long long)sc->dst_node_id, sc->source->chunk);
return -1;
}
static void stream_ready(struct ll_queue* queue, void* arg) {
(void)queue;
struct stream_ctx* sc = arg;
if (stream_schedule(sc)) stream_free(sc);
}
static void md_block_wake(struct media_block_source* source) {
struct stream_ctx* sc = source->md->streams;
while (sc) {
struct stream_ctx* next = sc->next;
if (sc->source == source && stream_schedule(sc)) stream_free(sc);
sc = next;
}
}
static void stream_watchdog(void* arg) {
struct stream_ctx* sc = arg;
sc->timer = NULL;
uint64_t timeout = sc->md->dl_stall_timeout_tb > 0 ? (uint64_t)sc->md->dl_stall_timeout_tb : MEDIA_DL_STALL_TIMEOUT_TB;
if (get_time_tb() - sc->last_progress_tb >= timeout) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: TX stalled peer=%016llx chunk=%u offset=%u available=%u verified=%u",
MD_ID, (unsigned long long)sc->dst_node_id, sc->source->chunk, sc->offset,
sc->source->available, sc->source->verified);
stream_free(sc); return;
}
sc->timer = uasync_set_timeout(sc->md->inst->ua, 2000, sc, stream_watchdog, "md_tx");
if (!sc->timer || stream_schedule(sc)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: TX watchdog scheduling failed", MD_ID); stream_free(sc);
}
}
/* Единственный путь выдачи готового диапазона и растущего temp. Waiter может сработать синхронно:
* callback только ставит pump, не входит в него рекурсивно. DONE проходит тот же backpressure, что DATA. */
static void stream_pump(void* arg) {
struct stream_ctx* sc = arg;
struct media_delivery_ctx* md = sc->md;
if (!md->initialized || sc->remaining == 0) { stream_free(sc); return; }
while (sc->remaining > 0) {
size_t to_read = sc->remaining < MD_CHUNK_SIZE ? (size_t)sc->remaining : MD_CHUNK_SIZE;
uint8_t buf[MD_CHUNK_SIZE];
fseeko(sc->file, (off_t)sc->offset, SEEK_SET);
size_t rd = fread(buf, 1, to_read, sc->file);
if (rd == 0) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "stream read failed"); stream_free(sc); return; }
uint8_t pkt[MEDIA_BLOCK_CHUNK_HDR_SIZE + MD_CHUNK_SIZE];
struct media_pkt_block_chunk* ch = (struct media_pkt_block_chunk*)pkt;
memset(ch, 0, MEDIA_BLOCK_CHUNK_HDR_SIZE);
ch->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK;
memcpy(ch->media_id, sc->media_id, 16);
memcpy(ch->block_id, sc->block_id, 16);
ch->chunk = sc->chunk;
ch->offset = (uint32_t)(sc->offset - sc->block_start);
ch->data_len = (uint16_t)rd;
memcpy(pkt + MEDIA_BLOCK_CHUNK_HDR_SIZE, buf, rd);
int snd = md_send(md->inst, sc->group_id, sc->dst_node_id, pkt, MEDIA_BLOCK_CHUNK_HDR_SIZE + rd);
if (snd != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: stream chunk send FAILED rc=%d — abort stream chunk=%d to 0x%016llx",
MD_ID, snd, sc->chunk, (unsigned long long)sc->dst_node_id);
stream_free(sc);
return;
}
sc->offset += rd;
sc->remaining -= rd;
{
uint64_t sent = sc->offset - sc->block_start;
uint64_t mb_off = (sent - rd) >> 20;
uint64_t mb_end = sent >> 20;
if (sent == rd || mb_end != mb_off || sent >= sc->block_data_len)
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: stream chunk=%d sent=%llu/%zu (%.0f%%) to 0x%016llx",
MD_ID, sc->chunk, (unsigned long long)sent, sc->block_data_len,
sc->block_data_len ? 100.0 * (double)sent / (double)sc->block_data_len : 0.0,
(unsigned long long)sc->dst_node_id);
}
if (sc->remaining == 0) break;
struct media_block_source* source = sc->source;
sc->pump = NULL;
if (!md->initialized) { stream_free(sc); return; }
etcp_router_cancel_send_ready(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, &sc->waiter);
for (int budget = 0; budget < 4; budget++) {
if (!source->failed && sc->offset == source->available && !source->verified) return;
if (!etcp_router_send_q_has_room(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY)) {
etcp_router_on_send_ready(md->inst, sc->group_id,
sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY,
&sc->waiter, stream_send_chunk_cb, sc);
etcp_router_on_send_ready(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY,
&sc->waiter, stream_ready, sc);
return;
}
if (source->failed || (source->verified && sc->offset == source->size)) {
struct media_pkt_block_done done = {.subcmd = MEDIA_SUBCMD_BLOCK_DONE, .chunk = source->chunk,
.total_size = source->size, .status = source->failed ? MEDIA_BLOCK_UNAVAILABLE : MEDIA_BLOCK_OK};
memcpy(done.media_id, source->media_id, 16); memcpy(done.block_id, source->block_id, 16);
memcpy(done.transfer_id, sc->transfer_id, 16);
if (!md_send(md->inst, sc->group_id, sc->dst_node_id, (const uint8_t*)&done, sizeof(done)))
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: TX DONE peer=%016llx chunk=%u bytes=%u status=%u",
MD_ID, (unsigned long long)sc->dst_node_id, source->chunk, sc->offset, done.status);
stream_free(sc); return;
}
uint32_t count = source->available - sc->offset;
if (count > MEDIA_BLOCK_DATA_MAX) count = MEDIA_BLOCK_DATA_MAX;
uint8_t packet[MEDIA_BLOCK_CHUNK_HDR_SIZE + MEDIA_BLOCK_DATA_MAX];
struct media_pkt_block_chunk* ch = (struct media_pkt_block_chunk*)packet;
memset(ch, 0, sizeof(*ch)); ch->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK;
memcpy(ch->media_id, source->media_id, 16); memcpy(ch->block_id, source->block_id, 16);
memcpy(ch->transfer_id, sc->transfer_id, 16);
ch->chunk = source->chunk; ch->offset = sc->offset; ch->data_len = (uint16_t)count;
FILE* file = file_io_open_read(source->storage.temporary);
int io_error = !file;
if (file) {
if (fseeko(file, (off_t)(source->base + sc->offset), SEEK_SET)
|| fread(packet + MEDIA_BLOCK_CHUNK_HDR_SIZE, 1, count, file) != count) io_error = 1;
if (fclose(file)) io_error = 1;
}
if (io_error) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: TX read failed path=%s chunk=%u offset=%u count=%u errno=%d",
MD_ID, source->storage.temporary, source->chunk, sc->offset, count, errno);
source->failed = 1; md_block_detach(source); md_block_wake(source);
continue;
}
if (md_send(md->inst, sc->group_id, sc->dst_node_id, packet, sizeof(*ch) + count)) {
stream_free(sc); return;
}
sc->offset += count; sc->last_progress_tb = get_time_tb();
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: TX peer=%016llx chunk=%u offset=%u available=%u/%u verified=%u",
MD_ID, (unsigned long long)sc->dst_node_id, source->chunk, sc->offset,
source->available, source->size, source->verified);
}
/* all data sent — send BLOCK_DONE */
{
uint32_t chunk = sc->chunk;
uint8_t done_pkt[MEDIA_BLOCK_DONE_SIZE];
struct media_pkt_block_done* bd = (struct media_pkt_block_done*)done_pkt;
memset(bd, 0, sizeof(*bd));
bd->subcmd = MEDIA_SUBCMD_BLOCK_DONE;
memcpy(bd->media_id, sc->media_id, 16);
memcpy(bd->block_id, sc->block_id, 16);
bd->chunk = chunk;
bd->total_size = (uint32_t)sc->block_data_len;
md_send(md->inst, sc->group_id, sc->dst_node_id, done_pkt, MEDIA_BLOCK_DONE_SIZE);
stream_free(sc);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_DONE sent chunk=%d total=%u",
MD_ID, chunk, bd->total_size);
}
if (stream_schedule(sc)) stream_free(sc);
}
/* ── relay chunk forwarding ── */
struct relay_fwd_ctx {
struct relay_fwd_ctx* next;
struct media_delivery_ctx* md;
uint64_t dst_node_id;
uint64_t group_id;
struct relay_block_ctx* rc;
int ds_idx; // индекс в rc->downstream[]
struct queue_waiter_handle waiter;
};
static void relay_fwd_free(struct relay_fwd_ctx* fc) {
etcp_router_cancel_send_ready(fc->md->inst, fc->group_id, fc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, &fc->waiter);
struct relay_fwd_ctx** p = &fc->md->forwards;
while (*p && *p != fc) p = &(*p)->next;
if (*p) *p = fc->next;
u_free(fc);
}
static void md_relay_cancel_block(struct media_delivery_ctx* md, struct relay_block_ctx* block) {
struct relay_fwd_ctx* fc = md->forwards;
while (fc) {
struct relay_fwd_ctx* next = fc->next;
if (fc->rc == block) relay_fwd_free(fc);
fc = next;
}
}
static void md_relay_fwd_cb(struct ll_queue* q, void* arg) {
(void)q;
struct relay_fwd_ctx* fc = (struct relay_fwd_ctx*)arg;
struct relay_block_ctx* rc = fc->rc;
struct relay_downstream* ds = &rc->downstream[fc->ds_idx];
if (ds->sent_offset >= rc->file_offset) { relay_fwd_free(fc); return; }
FILE* f = fopen(rc->chunk_file, "rb");
if (!f) { relay_fwd_free(fc); return; }
fseeko(f, (off_t)ds->sent_offset, SEEK_SET);
size_t to_read = rc->file_offset - ds->sent_offset;
if (to_read > 1024) to_read = 1024;
uint8_t buf[1024];
size_t rd = fread(buf, 1, to_read, f);
fclose(f);
if (rd == 0) { relay_fwd_free(fc); return; }
uint8_t pkt[MEDIA_BLOCK_CHUNK_HDR_SIZE + 1024];
struct media_pkt_block_chunk* ch = (struct media_pkt_block_chunk*)pkt;
memset(ch, 0, MEDIA_BLOCK_CHUNK_HDR_SIZE);
ch->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK;
memcpy(ch->media_id, rc->media_id, 16);
memcpy(ch->block_id, rc->block_id, 16);
ch->chunk = rc->chunk;
ch->offset = (uint32_t)ds->sent_offset;
ch->data_len = (uint16_t)rd;
memcpy(pkt + MEDIA_BLOCK_CHUNK_HDR_SIZE, buf, rd);
size_t pkt_len = MEDIA_BLOCK_CHUNK_HDR_SIZE + rd;
if (md_send(fc->md->inst, fc->group_id, fc->dst_node_id, pkt, pkt_len) == 0) {
ds->sent_offset += rd;
}
if (ds->sent_offset < rc->file_offset) {
etcp_router_on_send_ready(fc->md->inst, fc->group_id,
fc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY,
&fc->waiter, md_relay_fwd_cb, fc);
static void md_block_reply(struct media_delivery_ctx* md, uint64_t group, uint64_t peer,
const struct media_pkt_block_req* request, int overloaded) {
if (overloaded) {
struct media_pkt_block_overloaded reply = {.subcmd = MEDIA_SUBCMD_BLOCK_OVERLOADED,
.chunk = request->chunk, .retry_after_ms = 2000};
memcpy(reply.media_id, request->media_id, 16); memcpy(reply.block_id, request->block_id, 16);
memcpy(reply.transfer_id, request->transfer_id, 16);
md_send(md->inst, group, peer, (const uint8_t*)&reply, sizeof(reply));
} else {
relay_fwd_free(fc);
struct media_pkt_block_done reply = {.subcmd = MEDIA_SUBCMD_BLOCK_DONE, .chunk = request->chunk,
.status = MEDIA_BLOCK_UNAVAILABLE};
memcpy(reply.media_id, request->media_id, 16); memcpy(reply.block_id, request->block_id, 16);
memcpy(reply.transfer_id, request->transfer_id, 16);
md_send(md->inst, group, peer, (const uint8_t*)&reply, sizeof(reply));
}
}
static void md_relay_catchup(struct media_delivery_ctx* md, struct relay_block_ctx* rc,
uint64_t dst_node_id, uint64_t group_id, int ds_idx) {
struct relay_downstream* ds = &rc->downstream[ds_idx];
if (rc->file_offset == 0 || ds->sent_offset >= rc->file_offset) return;
struct relay_fwd_ctx* fc = u_calloc(1, sizeof(*fc));
if (!fc) return;
fc->next = md->forwards; md->forwards = fc;
fc->md = md; fc->dst_node_id = dst_node_id; fc->group_id = group_id;
fc->rc = rc; fc->ds_idx = ds_idx;
memset(&fc->waiter, 0, sizeof(fc->waiter));
md_relay_fwd_cb(NULL, fc);
}
static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_node,
const uint8_t* data, size_t len) {
if (len < MEDIA_BLOCK_REQ_SIZE) return;
struct media_pkt_block_req* req = (struct media_pkt_block_req*)data;
uint64_t group_id = req->group_id ? req->group_id : TOPO_GROUP_UTUN;
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_REQ from 0x%016llx group=%016llx block=%02x%02x... active_streams=%d",
MD_ID, (unsigned long long)from_node, (unsigned long long)group_id, req->block_id[0], req->block_id[1], md->active_streams);
int limit_reached = md->active_streams >= MD_MAX_STREAMS;
md->active_streams++;
if (limit_reached) {
md->active_streams--; /* undo */
struct media_pkt_block_overloaded ov;
ov.subcmd = MEDIA_SUBCMD_BLOCK_OVERLOADED;
memcpy(ov.media_id, req->media_id, 16);
memcpy(ov.block_id, req->block_id, 16);
ov.chunk = req->chunk;
ov.retry_after_ms = 2000;
md_send(md->inst, group_id, from_node, (const uint8_t*)&ov, sizeof(ov));
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ OVERLOADED (streams=%d) to 0x%016llx",
MD_ID, md->active_streams, (unsigned long long)from_node);
static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t group_id, uint64_t from_node,
const uint8_t* data, size_t len) {
if (len != MEDIA_BLOCK_REQ_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed BLOCK_REQ bytes=%zu", MD_ID, len); return;
}
const struct media_pkt_block_req* req = (const struct media_pkt_block_req*)data;
int relay_count = 0;
struct media_block_source* source = md_block_find(md, req->block_id);
if (source && (source->group_id != group_id || source->chunk != req->chunk
|| memcmp(source->media_id, req->media_id, 16))) source = NULL;
for (struct stream_ctx* sc = md->streams; sc; sc = sc->next) {
if (sc->source == source) relay_count++;
if (sc->group_id != group_id || sc->dst_node_id != from_node || memcmp(sc->transfer_id, req->transfer_id, 16)) continue;
if (sc->source->chunk != req->chunk || memcmp(sc->source->media_id, req->media_id, 16)
|| memcmp(sc->source->block_id, req->block_id, 16))
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: conflicting BLOCK_REQ attempt peer=%016llx", MD_ID, (unsigned long long)from_node);
return;
}
/* establish direct connection if not already connected */
if (!instance_find_conn(md->inst, from_node) && group_id != TOPO_GROUP_UTUN) {
struct TOPO_GROUP* grp = topo_groups_find(md->inst->topo_groups, group_id);
if (!grp) grp = topo_groups_get_default(md->inst->topo_groups);
if (grp && grp->conn_mgr) {
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: no direct conn to 0x%016llx — connecting via conn_mgr",
MD_ID, (unsigned long long)from_node);
conn_mgr_open(grp->instance, grp->group_id, from_node, NULL, NULL, NULL);
struct md_file_load* load = md_file_load_find(md, req->media_id);
if (md->active_streams >= MD_MAX_STREAMS || relay_count >= MD_MAX_RELAY_DOWNSTREAM
|| (load && md->max_downloads_per_file && load->active_downloads >= md->max_downloads_per_file)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_REQ admission busy peer=%016llx active=%u file=%d relay=%d",
MD_ID, (unsigned long long)from_node, md->active_streams, load ? load->active_downloads : 0, relay_count);
md_block_reply(md, group_id, from_node, req, 1); return;
}
if (source) source->refs++;
else {
sqlite3_stmt* st = NULL;
const char* sql = "SELECT location,chunk_size,chunk,offset,file_size,chat_id FROM media_files"
" WHERE block_id=? AND media_id=? AND node_id=? LIMIT 1";
int rc = sqlite3_prepare_v2(md->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, req->block_id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, req->media_id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)md->self_node_id);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
char base[MEDIA_BASE_MAX], path[MEDIA_PATH_MAX], channel[32];
snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group_id);
const char* location = rc == SQLITE_ROW ? (const char*)sqlite3_column_text(st, 0) : NULL;
int64_t size = rc == SQLITE_ROW ? sqlite3_column_int64(st, 1) : 0;
int64_t offset = rc == SQLITE_ROW ? sqlite3_column_int64(st, 3) : -1;
int64_t file_size = rc == SQLITE_ROW ? sqlite3_column_int64(st, 4) : 0;
const char* chat = rc == SQLITE_ROW ? (const char*)sqlite3_column_text(st, 5) : NULL;
int valid = location && sqlite3_column_bytes(st, 0) == (int)strlen(location)
&& chat && sqlite3_column_bytes(st, 5) == (int)strlen(chat) && !strcmp(chat, channel)
&& sqlite3_column_int64(st, 2) == req->chunk && size > 0 && size <= MEDIA_BLOCK_SIZE_MAX
&& offset >= 0 && offset < file_size && size <= file_size - offset
&& !media_index_get_base(md->db, base) && !media_index_path(base, location, path);
int final_rc = sqlite3_finalize(st);
struct file_io_info info;
if (!valid || final_rc != SQLITE_OK || file_io_inspect(path, 1, &info)
|| info.type != FILE_IO_REGULAR || info.size != file_size) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_REQ unavailable peer=%016llx chunk=%u sql=%d finalize=%d",
MD_ID, (unsigned long long)from_node, req->chunk, rc, final_rc);
md_block_reply(md, group_id, from_node, req, 0); return;
}
}
/* find file in media_files DB */
sqlite3* db = md->db;
if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ no DB", MD_ID); md->active_streams--; return; }
const char* sql = "SELECT location,chunk_size,chunk,offset,file_size FROM media_files WHERE block_id=? AND node_id=? LIMIT 1";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ prepare failed: %s", MD_ID, sqlite3_errmsg(db)); md->active_streams--; return; }
sqlite3_bind_blob(st, 1, req->block_id, 16, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)md->inst->node_id);
if (sqlite3_step(st) != SQLITE_ROW) {
sqlite3_finalize(st);
/* fallback: check relay_blocks — maybe we are downloading this block and can relay */
struct relay_block_ctx* rc = md_relay_find(md, req->block_id);
if (rc && rc->downstream_count < MD_MAX_RELAY_DOWNSTREAM) {
int dsi = rc->downstream_count++;
rc->downstream[dsi].node_id = from_node;
rc->downstream[dsi].group_id = group_id;
rc->downstream[dsi].sent_offset = 0;
memset(&rc->downstream[dsi].waiter, 0, sizeof(rc->downstream[dsi].waiter));
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: relay downstream[%d] added node=0x%016llx file_offset=%llu",
MD_ID, dsi, (unsigned long long)from_node, (unsigned long long)rc->file_offset);
md_relay_catchup(md, rc, from_node, group_id, dsi);
md->active_streams--; return;
struct ll_entry* entry = queue_entry_new(sizeof(*source));
if (!entry) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: indexed source allocation failed", MD_ID);
md_block_reply(md, group_id, from_node, req, 0); return;
}
if (rc) {
/* relay full — send RELAY_FULL with downstream list */
uint8_t rfp[256]; int roff = 0;
rfp[roff++] = MEDIA_SUBCMD_BLOCK_RELAY_FULL;
memcpy(rfp + roff, req->media_id, 16); roff += 16;
memcpy(rfp + roff, req->block_id, 16); roff += 16;
uint32_t ck = req->chunk; memcpy(rfp + roff, &ck, 4); roff += 4;
uint16_t nrn = (uint16_t)rc->downstream_count;
memcpy(rfp + roff, &nrn, 2); roff += 2;
for (int i = 0; i < rc->downstream_count; i++) {
memcpy(rfp + roff, &rc->downstream[i].node_id, 8); roff += 8;
}
md_send(md->inst, group_id, from_node, rfp, (size_t)roff);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: relay full for block=%02x%02x..., sent %d downstream nodes to 0x%016llx",
MD_ID, req->block_id[0], req->block_id[1], rc->downstream_count, (unsigned long long)from_node);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ block_id=%02x%02x... not found in media_files",
MD_ID, req->block_id[0], req->block_id[1]);
}
md->active_streams--; return;
source = (struct media_block_source*)entry->data; memset(source, 0, sizeof(*source));
source->entry = entry; source->md = md; source->refs = 1; source->verified = 1;
source->group_id = group_id; source->chunk = req->chunk; source->size = source->available = (uint32_t)size;
source->base = (uint64_t)offset;
memcpy(source->block_id, req->block_id, 16); memcpy(source->media_id, req->media_id, 16);
memcpy(source->storage.temporary, path, strlen(path) + 1);
}
const char* loc = (const char*)sqlite3_column_text(st, 0);
int64_t chunk_size = sqlite3_column_int64(st, 1);
int chunk = sqlite3_column_int(st, 2);
int64_t offset = sqlite3_column_int64(st, 3);
int64_t file_size = sqlite3_column_int64(st, 4);
/* compute chunk start and size: offset column хранит n * block_size
(для последнего усечённого блока chunk * chunk_size даёт неверный offset) */
int64_t block_start = offset;
int64_t block_end = block_start + chunk_size;
if (block_end > file_size) block_end = file_size;
int64_t block_length = block_end - block_start;
/* find media_base */
char media_base[MEDIA_BASE_MAX], path[MEDIA_PATH_MAX];
if (!loc || sqlite3_column_bytes(st, 0) != (int)strlen(loc) || media_index_get_base(db, media_base)
|| media_index_path(media_base, loc, path)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: invalid stored media path", MD_ID);
sqlite3_finalize(st); md->active_streams--; return;
}
FILE* f = file_io_open_read(path);
sqlite3_finalize(st);
if (!f) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: cannot open file %s for streaming", MD_ID, path);
md->active_streams--; /* undo pre-increment */
return;
}
/* per-file download limit: redirect to current downloaders if at capacity */
if (md->max_downloads_per_file > 0) {
struct md_file_load* fl = md_file_load_find(md, req->media_id);
if (fl && fl->active_downloads >= (int)md->max_downloads_per_file) {
/* source is at capacity — send RELAY_FULL with current downloader list */
uint8_t rfp[256]; int roff = 0;
rfp[roff++] = MEDIA_SUBCMD_BLOCK_RELAY_FULL;
memcpy(rfp + roff, req->media_id, 16); roff += 16;
memcpy(rfp + roff, req->block_id, 16); roff += 16;
uint32_t ck = req->chunk; memcpy(rfp + roff, &ck, 4); roff += 4;
uint16_t nrn = (uint16_t)fl->active_downloads;
memcpy(rfp + roff, &nrn, 2); roff += 2;
for (int i = 0; i < fl->active_downloads && i < 10; i++) {
memcpy(rfp + roff, &fl->downloader_ids[i], 8); roff += 8;
}
md_send(md->inst, group_id, from_node, rfp, (size_t)roff);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: source at capacity (%d/%d) for media=%02x%02x..., redirecting to %d nodes",
MD_ID, fl->active_downloads, md->max_downloads_per_file,
req->media_id[0], req->media_id[1], fl->active_downloads);
fclose(f);
md->active_streams--; return;
}
}
md_file_load_inc(md, req->media_id, from_node);
struct stream_ctx* sc = u_calloc(1, sizeof(*sc));
if (!sc) { fclose(f); md->active_streams--; md_file_load_dec(md, req->media_id, from_node); return; }
sc->md = md; sc->next = md->streams; md->streams = sc;
sc->group_id = group_id;
sc->dst_node_id = from_node;
memcpy(sc->media_id, req->media_id, 16);
memcpy(sc->block_id, req->block_id, 16);
sc->chunk = req->chunk;
sc->file = f;
sc->offset = (uint64_t)block_start;
sc->remaining = (uint64_t)block_length;
sc->block_start = (uint64_t)block_start;
sc->block_data_len = (size_t)block_length;
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: streaming file=%s chunk=%d start=%lld len=%lld group=%016llx to 0x%016llx",
MD_ID, path, chunk, (long long)block_start, (long long)block_length, (unsigned long long)group_id, (unsigned long long)from_node);
md->streams_started++; /* monotonic — for test admission checks */
/* start streaming — first chunk immediately, then backpressure */
memset(&sc->waiter, 0, sizeof(sc->waiter));
stream_send_chunk_cb(NULL, sc);
if (!sc || !etcp_router_conn_get(md->inst, group_id, from_node, ETCP_RT_ID_MEDIA_DELIVERY)
|| md_file_load_inc(md, req->media_id, from_node) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: cannot start TX peer=%016llx chunk=%u", MD_ID,
(unsigned long long)from_node, req->chunk);
u_free(sc); md_block_release(source); md_block_reply(md, group_id, from_node, req, 0); return;
}
sc->md = md; sc->source = source; sc->group_id = group_id; sc->dst_node_id = from_node;
memcpy(sc->transfer_id, req->transfer_id, 16); sc->last_progress_tb = get_time_tb();
sc->next = md->streams; md->streams = sc; md->active_streams++; md->streams_started++;
sc->timer = uasync_set_timeout(md->inst->ua, 2000, sc, stream_watchdog, "md_tx");
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: TX started peer=%016llx group=%016llx chunk=%u size=%u available=%u relay=%u",
MD_ID, (unsigned long long)from_node, (unsigned long long)group_id, source->chunk,
source->size, source->available, source->owned);
if (!sc->timer || stream_schedule(sc)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: TX startup scheduling failed", MD_ID); stream_free(sc);
}
}
/* ── handle RELAY_FULL: relay node is overloaded, retry one of its downstream nodes ── */
static void md_handle_block_relay_full(struct media_delivery_ctx* md, uint64_t from_node,
const uint8_t* data, size_t len) {
if (len < MEDIA_RELAY_FULL_HDR_SIZE) return;
struct media_pkt_block_relay_full* rf = (struct media_pkt_block_relay_full*)data;
uint16_t nrn = rf->num_relay_nodes;
if (nrn > MD_MAX_RELAY_DOWNSTREAM) nrn = MD_MAX_RELAY_DOWNSTREAM;
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: RELAY_FULL from 0x%016llx block=%02x%02x... nodes=%u",
MD_ID, (unsigned long long)from_node, rf->block_id[0], rf->block_id[1], nrn);
/* forward to media_download layer to retry from one of the listed nodes */
const uint8_t* rp = data + MEDIA_RELAY_FULL_HDR_SIZE;
uint64_t node_ids[MD_MAX_RELAY_DOWNSTREAM];
for (int i = 0; i < (int)nrn && i < MD_MAX_RELAY_DOWNSTREAM; i++) {
memcpy(&node_ids[i], rp, 8); rp += 8;
static void md_handle_cancel(struct media_delivery_ctx* md, uint64_t group, uint64_t peer,
const uint8_t* data, size_t len) {
if (len != sizeof(struct media_pkt_cancel)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed CANCEL bytes=%zu", MD_ID, len); return;
}
media_download_handle_relay_full(md->inst, rf->media_id, rf->block_id,
rf->chunk, node_ids, (int)nrn);
const struct media_pkt_cancel* can = (const struct media_pkt_cancel*)data;
for (struct stream_ctx* sc = md->streams; sc; sc = sc->next) {
if (sc->group_id == group && sc->dst_node_id == peer && sc->source->chunk == can->chunk
&& !memcmp(sc->transfer_id, can->transfer_id, 16) && !memcmp(sc->source->media_id, can->media_id, 16)
&& !memcmp(sc->source->block_id, can->block_id, 16)) {
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: TX cancelled peer=%016llx chunk=%u", MD_ID,
(unsigned long long)peer, can->chunk);
stream_free(sc); return;
}
}
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: stale CANCEL peer=%016llx chunk=%u", MD_ID, (unsigned long long)peer, can->chunk);
}
/* ── decrement stream counter (called when stream finished/cancelled internally) ── */
@@ -1288,22 +1219,22 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
case MEDIA_SUBCMD_SERVE_ACK:
break;
case MEDIA_SUBCMD_BLOCK_REQ:
md_handle_block_req(md, from_node, data, len);
md_handle_block_req(md, group_id, from_node, data, len);
break;
case MEDIA_SUBCMD_BLOCK_OVERLOADED:
media_download_handle_overloaded(inst, data, len);
media_download_handle_overloaded(inst, group_id, from_node, data, len);
break;
case MEDIA_SUBCMD_BLOCK_RELAY_FULL:
md_handle_block_relay_full(md, from_node, data, len);
media_download_handle_relay_full(inst, group_id, from_node, data, len);
break;
case MEDIA_SUBCMD_BLOCK_CHUNK:
media_download_handle_chunk(inst, data, len);
media_download_handle_chunk(inst, group_id, from_node, data, len);
break;
case MEDIA_SUBCMD_BLOCK_DONE:
media_download_handle_done(inst, data, len);
media_download_handle_done(inst, group_id, from_node, data, len);
break;
case MEDIA_SUBCMD_CANCEL:
/* TODO: handle cancel from remote */
md_handle_cancel(md, group_id, from_node, data, len);
break;
default:
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: unknown subcmd 0x%02x from 0x%016llx",
@@ -1408,7 +1339,6 @@ void media_delivery_destroy(struct UTUN_INSTANCE* inst) {
md->initialized = 0;
while (md->streams) stream_free(md->streams);
while (md->forwards) relay_fwd_free(md->forwards);
etcp_router_unbind(inst, ETCP_RT_ID_MEDIA_DELIVERY);
etcp_remove_conn_status_cbk(inst, md_on_conn_status, md);
@@ -1436,9 +1366,10 @@ void media_delivery_destroy(struct UTUN_INSTANCE* inst) {
}
queue_free(md->downloads); md->downloads = NULL;
}
if (md->relay_blocks) { queue_free(md->relay_blocks); md->relay_blocks = NULL; }
if (md->block_sources) { queue_free(md->block_sources); md->block_sources = NULL; }
if (md->file_loads) { queue_free(md->file_loads); md->file_loads = NULL; }
if (md->destinations) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: destroy with active destination writers", MD_ID);
memset(md, 0, sizeof(*md));
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: destroyed", MD_ID);
}
src/routing_layer/topo_node_sqlite.c
+13 -7
View File
@@ -907,18 +907,24 @@ int topo_node_sqlite_node_get_online(sqlite3* db, uint64_t node_id) {
int topo_node_sqlite_get_ed25519_pubkey(sqlite3* db, uint64_t node_id, uint8_t pubkey_out[32])
{
if (!db || !pubkey_out) return -1;
if (!db || !pubkey_out) {
DEBUG_ERROR(DEBUG_CATEGORY_ROUTING, "get_ed25519_pubkey: invalid arguments node=%016llx", (unsigned long long)node_id);
return -1;
}
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, "SELECT ed25519_pubkey FROM nodes WHERE node_id=?",
-1, &stmt, NULL) != SQLITE_OK) return -1;
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id);
int sql_rc = sqlite3_prepare_v2(db, "SELECT ed25519_pubkey FROM nodes WHERE node_id=?", -1, &stmt, NULL);
if (sql_rc == SQLITE_OK) sql_rc = sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id);
if (sql_rc == SQLITE_OK) sql_rc = sqlite3_step(stmt);
int rc = -1;
if (sqlite3_step(stmt) == SQLITE_ROW) {
if (sql_rc == SQLITE_ROW) {
const void* b = sqlite3_column_blob(stmt, 0);
int bytes = sqlite3_column_bytes(stmt, 0);
if (b && bytes >= 32) { uint64_t chk; memcpy(&chk, b, 8); if (chk != 0) { memcpy(pubkey_out, b, 32); rc = 0; } }
if (b && bytes == 32) { memcpy(pubkey_out, b, 32); rc = 0; }
}
sqlite3_finalize(stmt);
int final_rc = sqlite3_finalize(stmt);
if (final_rc != SQLITE_OK) rc = -1;
if (rc) DEBUG_WARN(DEBUG_CATEGORY_ROUTING, "get_ed25519_pubkey: unavailable node=%016llx sql=%d finalize=%d: %s",
(unsigned long long)node_id, sql_rc, final_rc, sqlite3_errmsg(db));
return rc;
}