diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index fcfaffea..af170d4f 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -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, diff --git a/src/media_delivery/media_delivery.h b/src/media_delivery/media_delivery.h index 0af1edcb..c61cd30e 100644 --- a/src/media_delivery/media_delivery.h +++ b/src/media_delivery/media_delivery.h @@ -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; // монотонный счётчик запущенных стримов (для тестов)