Cover worker teardown with completions in the active post batch

This commit is contained in:
evgeny committed 2026-10-26 19:15:32 +01:00
1 parent 93ec12226f
commit 8642b04e6d
3 files changed
+37 -4

No files matched your search

+1 -1
View File
@@ -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);
}
+2 -2
View File
@@ -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 допустим. */
+34 -1
View File
@@ -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 <assert.h>
#include <errno.h>
#include <limits.h>
@@ -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);