media_delivery: streaming, fallback, admission control, 8/9 tests
- md_handle_block_req: real chunk streaming via etcp_router_on_send_ready - stream_ctx state machine: backpressure waiter → send chunk → repeat - content_hash + block_sig on stream completion - Fallback to author when supernode has no block info - media_download: author_node_id for peer fallback - test_media_delivery_full: 8/9 pass (SUPER_HELLO, HAVE_BLOCK, replication, admission, file create) - Known issue: streaming test hung (BGP route for n1→n2 not established in time)
This commit is contained in:
evgeny committed
1 parent
57e6ffed16
commit
a6321e15b3
4 files changed
+268
-62
No files matched your search
@@ -497,9 +497,92 @@ static void md_super_conn_cb(int result, uint64_t node_id, void* arg) {
|
||||
/* wait for HELLO response — md_handle_super_hello will start replication */
|
||||
}
|
||||
|
||||
/* ── handle incoming BLOCK_REQ (admission control) ── */
|
||||
/* ── handle incoming BLOCK_REQ (admission control + streaming) ── */
|
||||
|
||||
#define MD_MAX_STREAMS 3
|
||||
#define MD_CHUNK_SIZE 1024
|
||||
|
||||
struct stream_ctx {
|
||||
struct media_delivery_ctx* md;
|
||||
uint64_t dst_node_id;
|
||||
uint8_t media_id[16];
|
||||
uint8_t block_id[16];
|
||||
uint32_t chunk;
|
||||
FILE* file;
|
||||
uint64_t offset; // absolute offset in file
|
||||
uint64_t remaining; // bytes left in this block
|
||||
uint64_t block_start; // absolute start of this block in file
|
||||
uint8_t* block_data; // accumulated data for signature (all chunk data)
|
||||
size_t block_data_len;
|
||||
struct queue_waiter_handle waiter;
|
||||
};
|
||||
|
||||
static void stream_send_chunk_cb(struct ll_queue* q, void* arg) {
|
||||
(void)q;
|
||||
struct stream_ctx* sc = (struct stream_ctx*)arg;
|
||||
struct media_delivery_ctx* md = sc->md;
|
||||
if (!md || sc->remaining == 0) { u_free(sc); return; }
|
||||
|
||||
/* read up to MD_CHUNK_SIZE bytes */
|
||||
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) { /* EOF or error */
|
||||
u_free(sc->block_data); u_free(sc); return;
|
||||
}
|
||||
|
||||
/* build CHUNK packet */
|
||||
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);
|
||||
|
||||
md_send(md->inst, sc->dst_node_id, pkt, MEDIA_BLOCK_CHUNK_HDR_SIZE + rd);
|
||||
|
||||
sc->offset += rd;
|
||||
sc->remaining -= rd;
|
||||
|
||||
if (sc->remaining == 0) {
|
||||
/* all data sent — send BLOCK_DONE */
|
||||
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 = sc->chunk;
|
||||
bd->total_size = (uint32_t)sc->block_data_len;
|
||||
|
||||
/* sign block data */
|
||||
uint8_t sig[64];
|
||||
uint64_t node_id = md->inst->node_id;
|
||||
uint8_t smsg[8192]; size_t soff = 0;
|
||||
memcpy(smsg + soff, sc->block_data, sc->block_data_len); soff += sc->block_data_len;
|
||||
memcpy(smsg + soff, &node_id, 8); soff += 8;
|
||||
if (sc_ed25519_sign(md->inst->my_ed25519_privkey, smsg, soff, sig) == SC_OK)
|
||||
memcpy(bd->block_sig, sig, 64);
|
||||
|
||||
md_send(md->inst, sc->dst_node_id, done_pkt, MEDIA_BLOCK_DONE_SIZE);
|
||||
media_delivery_stream_done(md->inst);
|
||||
fclose(sc->file);
|
||||
u_free(sc->block_data);
|
||||
u_free(sc);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_DONE sent chunk=%d total=%u",
|
||||
MD_ID, sc->chunk, bd->total_size);
|
||||
} else {
|
||||
/* continue streaming — register waiter for backpressure */
|
||||
etcp_router_on_send_ready(md->inst, TOPO_GROUP_UTUN,
|
||||
sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY,
|
||||
&sc->waiter, stream_send_chunk_cb, sc);
|
||||
}
|
||||
}
|
||||
|
||||
static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_node,
|
||||
const uint8_t* data, size_t len) {
|
||||
@@ -509,7 +592,11 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_nod
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_REQ from 0x%016llx block=%02x%02x... active_streams=%d",
|
||||
MD_ID, (unsigned long long)from_node, req->block_id[0], req->block_id[1], md->active_streams);
|
||||
|
||||
if (md->active_streams >= MD_MAX_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);
|
||||
@@ -522,9 +609,71 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_nod
|
||||
return;
|
||||
}
|
||||
|
||||
md->active_streams++;
|
||||
/* hand off to download module for actual file streaming */
|
||||
/* the stream completion (BLOCK_DONE sent) MUST decrement active_streams */
|
||||
/* find file in media_files DB */
|
||||
sqlite3* db = md->db;
|
||||
if (!db) 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) 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); return; }
|
||||
|
||||
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 */
|
||||
int64_t block_start = (int64_t)chunk * chunk_size;
|
||||
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[512] = "/tmp/utun_media";
|
||||
{
|
||||
sqlite3_stmt* ms = NULL;
|
||||
sqlite3_prepare_v2(db, "SELECT value FROM ui_state WHERE key='media_base'", -1, &ms, NULL);
|
||||
if (ms && sqlite3_step(ms) == SQLITE_ROW) {
|
||||
const char* mb = (const char*)sqlite3_column_text(ms, 0);
|
||||
if (mb) snprintf(media_base, sizeof(media_base), "%s", mb);
|
||||
}
|
||||
if (ms) sqlite3_finalize(ms);
|
||||
}
|
||||
(void)media_base;
|
||||
|
||||
char path[2048];
|
||||
snprintf(path, sizeof(path), "%s/%s", media_base, loc);
|
||||
FILE* f = fopen(path, "rb");
|
||||
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;
|
||||
}
|
||||
struct stream_ctx* sc = u_calloc(1, sizeof(*sc));
|
||||
if (!sc) { fclose(f); md->active_streams--; return; }
|
||||
sc->md = md;
|
||||
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_DEBUG, "%s: streaming file=%s chunk=%d start=%lld len=%lld to 0x%016llx",
|
||||
MD_ID, path, chunk, (long long)block_start, (long long)block_length, (unsigned long long)from_node);
|
||||
|
||||
/* start streaming — first chunk immediately, then backpressure */
|
||||
memset(&sc->waiter, 0, sizeof(sc->waiter));
|
||||
stream_send_chunk_cb(NULL, sc);
|
||||
}
|
||||
|
||||
/* ── decrement stream counter (called when stream finished/cancelled internally) ── */
|
||||
|
||||
@@ -231,6 +231,17 @@ void media_download_handle_query_resp(struct UTUN_INSTANCE* inst,
|
||||
}
|
||||
if (!grp || !grp->conn_mgr) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: QUERY_RESP — no conn_mgr to connect peers", MDL_ID); return; }
|
||||
|
||||
/* if no peers found, try querying the author directly */
|
||||
if (dl->num_peers == 0 && dl->author_node_id && dl->author_node_id != inst->node_id) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: no holders from supernode, querying author 0x%016llx",
|
||||
MDL_ID, (unsigned long long)dl->author_node_id);
|
||||
dl->super_nodes[0] = dl->author_node_id;
|
||||
dl->super_count = 1;
|
||||
dl->super_current = 0;
|
||||
md_dl_send_query(inst, dl); /* retry QUERY to author */
|
||||
return;
|
||||
}
|
||||
|
||||
for (int pi = 0; pi < dl->num_peers; pi++) {
|
||||
conn_mgr_connect_node(grp->conn_mgr, dl->peers[pi].node_id, 0, md_dl_conn_cb, dl);
|
||||
}
|
||||
|
||||
@@ -55,6 +55,7 @@ struct media_download {
|
||||
uint64_t super_nodes[10];
|
||||
int super_count;
|
||||
int super_current;
|
||||
uint64_t author_node_id; /* src_node_id from chat message, for fallback when supernode has no info */
|
||||
void* query_timer;
|
||||
void* timeout_timer;
|
||||
void (*done_cb)(void* arg, int err);
|
||||
|
||||
Reference in new issue
Block a user