diff --git a/src/chat/chat_join_port.c b/src/chat/chat_join_port.c index ae60e7fa..7f439e67 100644 --- a/src/chat/chat_join_port.c +++ b/src/chat/chat_join_port.c @@ -104,7 +104,7 @@ void chat_join_port_close(struct chat_join_port* p) { struct join_command* c = commands; commands = c->next; c->completed = 1; uint64_t request = c->request, channel = c->channel, peer = c->peer; - /* Уже извлечённый batch исполнит пустой callback и снимет свою ссылку. */ + /* Успешная отмена снимает ссылку команды; уже начатый execute снимает её самостоятельно. */ if (uasync_cancel_post(ua, &c->post) == ERR_OK) chat_join_port_release(p); chat_sync_join_result(inst, request, channel, peer, CHAT_CONNECT_CANCELLED); } diff --git a/src/member_files/worker.h b/src/member_files/worker.h index 2591b7c5..fc836d6f 100644 --- a/src/member_files/worker.h +++ b/src/member_files/worker.h @@ -1,7 +1,7 @@ /* Единственный дисковый поток модуля. submit/done выполняются в uasync, work — * в pthread. Очередь ограничена; completion резервируется до передачи работы. * Для обязательной очистки reserve выполняется до открытия файлов. destroy вне - * callbacks ждёт работ и доставляет done, включая поставленные ими cleanup jobs. + * собственных callbacks ждёт работ и доставляет done, включая поставленные ими cleanup jobs. * data/arg заимствованы до done; worker освобождает только собственную оболочку работы. * ua/work/done и worker обязательны; ua живёт до конца destroy. Останавливать worker из done нельзя. */ #ifndef MEMBER_FILES_WORKER_H @@ -24,7 +24,7 @@ void mf_worker_submit(struct mf_worker* worker, mf_work_fn work, void* data, mf_ * После вызова резерв нельзя освобождать/использовать повторно. Work выполняется FIFO, done ровно один раз. */ void mf_worker_submit_reserved(struct mf_worker* worker, struct mf_work* reserved, mf_work_fn work, void* data, mf_done_fn done, void* arg); -// После quiesce владельца, вне callbacks: дождаться работ и их cleanup continuations, не останавливая pthread. +// После quiesce владельца, вне собственных callbacks: дождаться работ и cleanup continuations без остановки pthread. // Не запускает uasync рекурсивно; порядок done при drain не обещается. NULL допустим. void mf_worker_drain(struct mf_worker* worker); /* Drain, затем join и освобождение. Работы не прерываются и получают обычный done(0); NULL допустим. */ diff --git a/tests/test_media_primitives.c b/tests/test_media_primitives.c index a02be870..e10b0e19 100644 --- a/tests/test_media_primitives.c +++ b/tests/test_media_primitives.c @@ -6,6 +6,7 @@ #include "../lib/u_async.h" #include "../src/media_async/media_async.h" #include "../src/media_delivery/media_index.h" +#include "../src/member_files/worker.h" #include #include #include @@ -413,6 +414,38 @@ static void engine_tests(struct UASYNC *ua) { uasync_poll(ua, 0); assert(state.done == 2 && u_get_allocated_count() == before); DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS 16 finished jobs retain slots; failures, reentrant submit, destroy and once-only cancellation"); } + +struct batch_destroy { + struct mf_worker *disk; + struct media_async *media; + unsigned calls; +}; +static void destroy_in_batch(void *arg) { + struct batch_destroy *state = arg; + mf_worker_destroy(state->disk); media_async_destroy(state->media); state->calls++; +} + +/* Барьер гарантирует очередь [destroy, completion] до poll, без зависимости от планировщика. */ +static void worker_batch_tests(struct UASYNC *ua) { + size_t before = u_get_allocated_count(); + for (int media = 0; media < 2; media++) { + struct batch_destroy state = {0}; + struct job job = {.ua = ua, .event_thread = pthread_self()}; + if (media) { state.media = media_async_create(); assert(state.media); } + else { state.disk = mf_worker_create(ua); assert(state.disk); } + uasync_post(ua, destroy_in_batch, &state); + pthread_barrier_t barrier; assert(!pthread_barrier_init(&barrier, NULL, 2)); + __atomic_store_n(&post_barrier, &barrier, __ATOMIC_RELEASE); + if (media) media_async_submit(state.media, ua, job_work, &job, job_done, &job); + else mf_worker_submit(state.disk, job_work, &job, job_done, &job); + pthread_barrier_wait(&barrier); __atomic_store_n(&post_barrier, NULL, __ATOMIC_RELEASE); + uasync_poll(ua, 0); uasync_poll(ua, 0); + assert(state.calls == 1 && job.work == 1 && job.done == 1); + assert(job.error == (media ? MEDIA_ASYNC_CANCELLED : 0)); + assert(!pthread_barrier_destroy(&barrier) && u_get_allocated_count() == before); + } + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS disk drain and media destroy cancel batched completions exactly once"); +} #endif int main(void) { @@ -429,7 +462,7 @@ int main(void) { struct UASYNC *ua = uasync_create(); assert(ua); registration_tests(ua); #ifdef __linux__ - temporary_thread_tests(); engine_tests(ua); + temporary_thread_tests(); engine_tests(ua); worker_batch_tests(ua); #endif uasync_destroy(ua, 0); assert(!ma_dir_walk(root, remove_file, NULL) && !rmdir(root)); assert(u_get_allocated_count() == before);