From 68129c01511c03daf02600d177b6fbff81e20c19 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 19:07:04 +0200 Subject: [PATCH] Retain media downloads across reentrant progress callbacks --- src/media_delivery/media_download.c | 40 +++++++++++++++++----------- src/media_delivery/media_download.h | 6 ++++- tests/test_media_delivery_download.c | 2 ++ 3 files changed, 32 insertions(+), 16 deletions(-) diff --git a/src/media_delivery/media_download.c b/src/media_delivery/media_download.c index 3011053c..bf783fed 100644 --- a/src/media_delivery/media_download.c +++ b/src/media_delivery/media_download.c @@ -282,15 +282,24 @@ int md_dl_check_stall(struct media_download* dl, uint64_t now_tb) { return md_dl_pump(dl); } -/* ── завершение загрузки (успех/провал/отмена) — освобождает всё ── */ - -static void md_dl_finish(struct media_download* dl, int err) { - struct UTUN_INSTANCE* inst = dl->inst; - struct media_delivery_ctx* md = &inst->md; - uint8_t* blks = (uint8_t*)dl->blocks; - uint8_t* sigs = dl->block_sigs; +/* Последний обработчик освобождает данные до done_cb. Отмена из progress_cb не уничтожает его аргумент. */ +static void md_dl_release(struct media_download* dl) { + if (--dl->refs) return; void (*cb)(void*, int) = dl->done_cb; void* cb_arg = dl->done_arg; + int err = dl->err; + u_free(dl->blocks); + u_free(dl->block_sigs); + queue_entry_free(dl->entry); + if (cb) cb(cb_arg, err); +} + +/* Завершение снимает активное владение; выполняющийся обработчик может ещё удерживать dl. */ + +static void md_dl_finish(struct media_download* dl, int err) { + if (!dl->active) return; + struct UTUN_INSTANCE* inst = dl->inst; + struct media_delivery_ctx* md = &inst->md; DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: finish media=%02x%02x err=%d validated=%d/%d inflight=%d peers=%d query=%d", MDL_ID, dl->media_id[0], dl->media_id[1], err, dl->blocks_validated, dl->num_blocks, @@ -302,13 +311,8 @@ static void md_dl_finish(struct media_download* dl, int err) { for (int pi = 0; pi < dl->num_peers; pi++) { if (dl->peers[pi].cm_handle) { conn_mgr_close(dl->peers[pi].cm_handle); dl->peers[pi].cm_handle = NULL; } } - if (md->downloads) { - struct ll_entry* e = queue_find_data_by_index(md->downloads, dl->media_id); - if (e) { queue_remove_data(md->downloads, e); queue_entry_free(e); } - } - if (blks) u_free(blks); - if (sigs) u_free(sigs); - if (cb) cb(cb_arg, err); + queue_remove_data(md->downloads, dl->entry); + md_dl_release(dl); } /* Один таймер обслуживает ближайший дедлайн поиска, блока или задержанного повтора. */ @@ -701,7 +705,9 @@ void media_download_handle_done(struct UTUN_INSTANCE* inst, DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_DONE block=%d total=%d/%d", MDL_ID, bi, dl->blocks_validated, dl->num_blocks); + dl->refs++; if (dl->progress_cb) dl->progress_cb(dl->progress_arg, dl->blocks_validated, dl->num_blocks); + if (!dl->active) { md_dl_release(dl); return; } md_dl_send_have_block(inst, dl, bi); @@ -745,17 +751,19 @@ void media_download_handle_done(struct UTUN_INSTANCE* inst, { int hash_zero = 1; for (int hi = 0; hi < 32; hi++) if (dl->content_hash[hi]) { hash_zero = 0; break; } if (!hash_zero) { uint8_t fhash[32]; ma_sha256_file(dl->dest_path, fhash); - if (memcmp(fhash, dl->content_hash, 32) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: content_hash mismatch for %s", MDL_ID, dl->dest_path); remove(dl->dest_path); dl->blocks_validated--; return; } + if (memcmp(fhash, dl->content_hash, 32) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: content_hash mismatch for %s", MDL_ID, dl->dest_path); remove(dl->dest_path); md_dl_finish(dl, -1); md_dl_release(dl); return; } } } dl->assembled = 1; DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: file assembled: %s", MDL_ID, dl->dest_path); int registered = md_dl_register_servable(dl); md_dl_finish(dl, registered); + md_dl_release(dl); return; } /* освободился слот — добрать ожидающие блоки */ if (md_dl_pump(dl) < 0) md_dl_finish(dl, -1); + md_dl_release(dl); } /* ── handle OVERLOADED ── */ @@ -916,6 +924,8 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, queue_data_put_with_index(md->downloads, qe); dl = (struct media_download*)qe->data; + dl->entry = qe; + dl->refs = 1; md_dl_collect_supernodes(inst, dl); diff --git a/src/media_delivery/media_download.h b/src/media_delivery/media_download.h index 25b021e7..42791a94 100644 --- a/src/media_delivery/media_download.h +++ b/src/media_delivery/media_download.h @@ -69,6 +69,8 @@ struct media_download_peer { /* ── внутреннее состояние загрузки (доступно для тестов) ── */ struct media_download { struct ll_entry ll; + struct ll_entry* entry; /* Владеющая запись; после удаления из индекса живёт до последнего refs. */ + unsigned refs; /* Активная загрузка и обработчики, вызывающие пользовательские callbacks. */ uint8_t media_id[16]; uint64_t group_id; char dest_path[1024]; @@ -109,7 +111,9 @@ struct media_download { * Проверки идут до создания каталогов. 0 — принят: done_cb ровно один раз, в том числе синхронно при отсутствии суперузлов; * -1 — отказ без done_cb, включая уже активный media_id. Аргументы callbacks заимствованы до done_cb. * Успех включает атомарную регистрацию блоков собранного файла. Ошибка регистрации не удаляет собранный файл; - * путь остаётся в media_preparations для удаления канала. Остальные поля result копируются до возврата. */ + * путь остаётся в media_preparations для удаления канала. Остальные поля result копируются до возврата. + * progress_cb может отменить загрузку и запустить новую с тем же media_id; done_cb старой загрузки + * вызывается после возврата из progress_cb, когда её последний обработчик освободил контекст. */ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, const struct media_index_result* result, const char* dest_path, const char* media_base, diff --git a/tests/test_media_delivery_download.c b/tests/test_media_delivery_download.c index 4de38a4b..c9d487c2 100644 --- a/tests/test_media_delivery_download.c +++ b/tests/test_media_delivery_download.c @@ -136,6 +136,8 @@ static struct media_download* make_test_dl(sqlite3* db, struct UTUN_INSTANCE* in memcpy(qe->data, dl, sizeof(*dl)); u_free(dl); queue_data_put_with_index(inst->md.downloads, qe); + ((struct media_download*)qe->data)->entry = qe; + ((struct media_download*)qe->data)->refs = 1; return (struct media_download*)qe->data; }