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

Bound relay readers independently and redirect only to matching block recipients

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 19:32:15 +02:00
1 parent b09a020855
commit 99389f4bd3
2 files changed
+35 -13
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.c
+34 -12
View File
@@ -27,7 +27,7 @@
/* ── forward decls ── */
static void md_block_wake(struct media_block_source* source);
static void md_block_wake(struct media_block_source* source, struct stream_ctx* executing);
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);
@@ -328,7 +328,7 @@ struct media_block_source* md_block_create(struct media_delivery_ctx* md, uint64
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);
md_block_wake(source, NULL);
}
void md_block_abort(struct media_block_source* source) {
@@ -336,7 +336,7 @@ void md_block_abort(struct media_block_source* source) {
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);
md_block_wake(source, NULL);
}
/* ── file_load helpers (per-file source download tracking) ── */
@@ -772,8 +772,10 @@ static void stream_free(struct stream_ctx* sc) {
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->source->owned) {
md_file_load_dec(md, sc->source->media_id, sc->dst_node_id);
media_delivery_stream_done(md->inst);
}
md_block_release(sc->source);
u_free(sc);
}
@@ -792,11 +794,11 @@ static void stream_ready(struct ll_queue* queue, void* arg) {
if (stream_schedule(sc)) stream_free(sc);
}
static void md_block_wake(struct media_block_source* source) {
static void md_block_wake(struct media_block_source* source, struct stream_ctx* executing) {
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);
if (sc != executing && sc->source == source && stream_schedule(sc)) stream_free(sc);
sc = next;
}
}
@@ -861,7 +863,7 @@ static void stream_pump(void* arg) {
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);
source->failed = 1; md_block_detach(source); md_block_wake(source, sc);
continue;
}
if (md_send(md->inst, sc->group_id, sc->dst_node_id, packet, sizeof(*ch) + count)) {
@@ -911,10 +913,28 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t group_id
return;
}
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)) {
if ((source && relay_count >= MD_MAX_RELAY_DOWNSTREAM)
|| (!source && (md->active_streams >= MD_MAX_STREAMS
|| (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);
uint64_t nodes[MD_MAX_RELAY_DOWNSTREAM]; unsigned count = 0;
for (struct stream_ctx* sc = md->streams; sc && count < MD_MAX_RELAY_DOWNSTREAM; sc = sc->next) {
if (sc->group_id != group_id || memcmp(sc->source->block_id, req->block_id, 16)
|| memcmp(sc->source->media_id, req->media_id, 16) || sc->source->chunk != req->chunk
|| sc->dst_node_id == from_node || sc->dst_node_id == md->self_node_id) continue;
unsigned i = 0;
while (i < count && nodes[i] != sc->dst_node_id) i++;
if (i == count) nodes[count++] = sc->dst_node_id;
}
if (count) {
uint8_t packet[MEDIA_RELAY_FULL_HDR_SIZE + sizeof(nodes)] = {0};
struct media_pkt_block_relay_full* full = (struct media_pkt_block_relay_full*)packet;
full->subcmd = MEDIA_SUBCMD_BLOCK_RELAY_FULL; full->chunk = req->chunk; full->num_relay_nodes = (uint16_t)count;
memcpy(full->media_id, req->media_id, 16); memcpy(full->block_id, req->block_id, 16);
memcpy(full->transfer_id, req->transfer_id, 16); memcpy(packet + sizeof(*full), nodes, count * sizeof(uint64_t));
md_send(md->inst, group_id, from_node, packet, sizeof(*full) + count * sizeof(uint64_t)); return;
}
md_block_reply(md, group_id, from_node, req, 1); return;
}
if (source) source->refs++;
@@ -961,14 +981,16 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t group_id
}
struct stream_ctx* sc = u_calloc(1, sizeof(*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) {
|| (!source->owned && 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->next = md->streams; md->streams = sc;
if (!source->owned) 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,
src/media_delivery/media_delivery.h
+1 -1
View File
@@ -117,7 +117,7 @@ struct media_delivery_ctx {
uint64_t self_node_id;
int initialized;
uint8_t is_supernode; // из adm_tags в peers_*
uint8_t active_streams; // текущее число исходящих стримов (admission)
uint8_t active_streams; // Выдачи индексированных файлов; растущий relay ограничен отдельно на блок.
uint8_t max_downloads_per_file; // лимит параллельных загрузок с одного файла (source node), default 3
uint8_t stream_completed; // 1 = хотя бы один стрим завершился (для тестов)
uint8_t streams_started; // монотонный счётчик запущенных стримов (для тестов)