From 7259746b3de6995e83362d9416c56cf52101ea63 Mon Sep 17 00:00:00 2001 From: fy59 Date: Sat, 8 Aug 2026 09:53:42 +0200 Subject: [PATCH] feat(task_queue): add bounded backpressure --- include/lardon3d/task_queue.h | 13 +- src/app.c | 2 +- src/task_queue.c | 147 +++++++--- tests/test_task_queue.c | 506 +++++++++++++++++++++++++++++++++- 4 files changed, 617 insertions(+), 51 deletions(-) diff --git a/include/lardon3d/task_queue.h b/include/lardon3d/task_queue.h index a6154ad..418dac1 100644 --- a/include/lardon3d/task_queue.h +++ b/include/lardon3d/task_queue.h @@ -17,15 +17,24 @@ typedef struct { } Lardon3DTaskQueueSummary; Lardon3DTaskQueue *lardon3d_task_queue_create( - Lardon3DResourceGovernor *governor + Lardon3DResourceGovernor *governor, + size_t capacity ); void lardon3d_task_queue_destroy(Lardon3DTaskQueue *queue); -/* La file devient propriétaire de task uniquement en cas de succès. */ +/* La file devient propriétaire de task uniquement en cas de succès. + Bloquante : attend une place libre si la file est pleine. */ bool lardon3d_task_queue_add( Lardon3DTaskQueue *queue, Lardon3DTask *task, uint64_t *task_id ); +/* Non-bloquante : retourne false si la file est pleine ou en arrêt. + La file devient propriétaire de task uniquement en cas de succès. */ +bool lardon3d_task_queue_try_add( + Lardon3DTaskQueue *queue, + Lardon3DTask *task, + uint64_t *task_id +); bool lardon3d_task_queue_remove(Lardon3DTaskQueue *queue, uint64_t task_id); bool lardon3d_task_queue_cancel(Lardon3DTaskQueue *queue, uint64_t task_id); void lardon3d_task_queue_resources_changed(Lardon3DTaskQueue *queue); diff --git a/src/app.c b/src/app.c index 2a66ebd..f5a7f1d 100644 --- a/src/app.c +++ b/src/app.c @@ -38,7 +38,7 @@ lardon3d_app_run(void) if (!state.resource_governor) { return EXIT_FAILURE; } - state.task_queue = lardon3d_task_queue_create(state.resource_governor); + state.task_queue = lardon3d_task_queue_create(state.resource_governor, 64); if (!state.task_queue) { lardon3d_resource_governor_destroy(state.resource_governor); return EXIT_FAILURE; diff --git a/src/task_queue.c b/src/task_queue.c index fdf7f65..735510b 100644 --- a/src/task_queue.c +++ b/src/task_queue.c @@ -13,7 +13,8 @@ typedef struct TaskNode { struct Lardon3DTaskQueue { pthread_mutex_t mutex; - pthread_cond_t condition; + pthread_cond_t not_empty; + pthread_cond_t not_full; pthread_t worker; bool worker_started; bool stopping; @@ -24,7 +25,9 @@ struct Lardon3DTaskQueue { TaskNode *pending_head; TaskNode *pending_tail; Lardon3DTask *active; - size_t count; + size_t capacity; + size_t pending_count; + size_t active_producers; }; static bool @@ -37,6 +40,7 @@ terminal_state(Lardon3DTaskState state) static void unlink_pending(Lardon3DTaskQueue *queue, TaskNode *previous, TaskNode *node) { + bool was_full = queue->pending_count >= queue->capacity; if (previous) { previous->next_pending = node->next_pending; } else { @@ -46,6 +50,10 @@ unlink_pending(Lardon3DTaskQueue *queue, TaskNode *previous, TaskNode *node) queue->pending_tail = previous; } node->next_pending = NULL; + --queue->pending_count; + if (was_full) { + (void)pthread_cond_signal(&queue->not_full); + } } /* Parcourt la file d'attente et sélectionne la première tâche admissible. @@ -117,7 +125,7 @@ queue_worker(void *context) for (;;) { (void)pthread_mutex_lock(&queue->mutex); while (!queue->stopping && !queue->pending_head) { - (void)pthread_cond_wait(&queue->condition, &queue->mutex); + (void)pthread_cond_wait(&queue->not_empty, &queue->mutex); } if (queue->stopping) { (void)pthread_mutex_unlock(&queue->mutex); @@ -126,7 +134,7 @@ queue_worker(void *context) Lardon3DResourceReservation *reservation = NULL; Lardon3DTask *selected = select_admissible(queue, &reservation); if (!selected) { - (void)pthread_cond_wait(&queue->condition, &queue->mutex); + (void)pthread_cond_wait(&queue->not_empty, &queue->mutex); (void)pthread_mutex_unlock(&queue->mutex); continue; } @@ -150,15 +158,15 @@ queue_worker(void *context) (void)pthread_mutex_lock(&queue->mutex); queue->active = NULL; - (void)pthread_cond_broadcast(&queue->condition); + (void)pthread_cond_broadcast(&queue->not_empty); (void)pthread_mutex_unlock(&queue->mutex); } } Lardon3DTaskQueue * -lardon3d_task_queue_create(Lardon3DResourceGovernor *governor) +lardon3d_task_queue_create(Lardon3DResourceGovernor *governor, size_t capacity) { - if (!governor) { + if (!governor || capacity < 1) { return NULL; } Lardon3DTaskQueue *queue = calloc(1, sizeof(*queue)); @@ -169,15 +177,23 @@ lardon3d_task_queue_create(Lardon3DResourceGovernor *governor) free(queue); return NULL; } - if (pthread_cond_init(&queue->condition, NULL) != 0) { + if (pthread_cond_init(&queue->not_empty, NULL) != 0) { + (void)pthread_mutex_destroy(&queue->mutex); + free(queue); + return NULL; + } + if (pthread_cond_init(&queue->not_full, NULL) != 0) { + (void)pthread_cond_destroy(&queue->not_empty); (void)pthread_mutex_destroy(&queue->mutex); free(queue); return NULL; } queue->next_id = 1; queue->governor = governor; + queue->capacity = capacity; if (pthread_create(&queue->worker, NULL, queue_worker, queue) != 0) { - (void)pthread_cond_destroy(&queue->condition); + (void)pthread_cond_destroy(&queue->not_full); + (void)pthread_cond_destroy(&queue->not_empty); (void)pthread_mutex_destroy(&queue->mutex); free(queue); return NULL; @@ -198,8 +214,15 @@ lardon3d_task_queue_cancel(Lardon3DTaskQueue *queue, uint64_t task_id) node = node->next_all; } if (node) { + Lardon3DTaskSnapshot task_snapshot; + bool was_pending = lardon3d_task_snapshot(node->task, &task_snapshot) + && task_snapshot.state != TASK_RUNNING + && task_snapshot.state != TASK_PAUSED; lardon3d_task_request_cancel(node->task); - (void)pthread_cond_broadcast(&queue->condition); + if (was_pending && queue->pending_count == queue->capacity) { + (void)pthread_cond_signal(&queue->not_full); + } + (void)pthread_cond_broadcast(&queue->not_empty); } (void)pthread_mutex_unlock(&queue->mutex); return node != NULL; @@ -212,7 +235,7 @@ lardon3d_task_queue_resources_changed(Lardon3DTaskQueue *queue) return; } (void)pthread_mutex_lock(&queue->mutex); - (void)pthread_cond_broadcast(&queue->condition); + (void)pthread_cond_broadcast(&queue->not_empty); (void)pthread_mutex_unlock(&queue->mutex); } @@ -227,7 +250,11 @@ lardon3d_task_queue_destroy(Lardon3DTaskQueue *queue) for (TaskNode *node = queue->all_head; node; node = node->next_all) { lardon3d_task_request_cancel(node->task); } - (void)pthread_cond_broadcast(&queue->condition); + (void)pthread_cond_broadcast(&queue->not_empty); + (void)pthread_cond_broadcast(&queue->not_full); + while (queue->active_producers > 0) { + (void)pthread_cond_wait(&queue->not_empty, &queue->mutex); + } (void)pthread_mutex_unlock(&queue->mutex); if (queue->worker_started) { (void)pthread_join(queue->worker, NULL); @@ -239,11 +266,47 @@ lardon3d_task_queue_destroy(Lardon3DTaskQueue *queue) free(node); node = next; } - (void)pthread_cond_destroy(&queue->condition); + (void)pthread_cond_destroy(&queue->not_full); + (void)pthread_cond_destroy(&queue->not_empty); (void)pthread_mutex_destroy(&queue->mutex); free(queue); } +/* Appelée sous le mutex queue. Ne signale pas not_empty sur échec. */ +static bool +enqueue_locked( + Lardon3DTaskQueue *queue, + TaskNode *node, + Lardon3DTask *task, + uint64_t *task_id +) +{ + if (queue->stopping || queue->next_id == 0 + || !lardon3d_task_assign_id(task, queue->next_id)) { + return false; + } + uint64_t id = queue->next_id++; + node->task = task; + if (queue->all_tail) { + queue->all_tail->next_all = node; + } else { + queue->all_head = node; + } + queue->all_tail = node; + if (queue->pending_tail) { + queue->pending_tail->next_pending = node; + } else { + queue->pending_head = node; + } + queue->pending_tail = node; + ++queue->pending_count; + if (task_id) { + *task_id = id; + } + (void)pthread_cond_signal(&queue->not_empty); + return true; +} + bool lardon3d_task_queue_add( Lardon3DTaskQueue *queue, @@ -259,33 +322,42 @@ lardon3d_task_queue_add( return false; } (void)pthread_mutex_lock(&queue->mutex); - if (queue->stopping || queue->next_id == 0 - || !lardon3d_task_assign_id(task, queue->next_id)) { - (void)pthread_mutex_unlock(&queue->mutex); + ++queue->active_producers; + while (!queue->stopping && queue->pending_count >= queue->capacity) { + (void)pthread_cond_wait(&queue->not_full, &queue->mutex); + } + --queue->active_producers; + (void)pthread_cond_broadcast(&queue->not_empty); + bool accepted = enqueue_locked(queue, node, task, task_id); + (void)pthread_mutex_unlock(&queue->mutex); + if (!accepted) { free(node); + } + return accepted; +} + +bool +lardon3d_task_queue_try_add( + Lardon3DTaskQueue *queue, + Lardon3DTask *task, + uint64_t *task_id +) +{ + if (!queue || !task) { return false; } - uint64_t id = queue->next_id++; - node->task = task; - if (queue->all_tail) { - queue->all_tail->next_all = node; - } else { - queue->all_head = node; + TaskNode *node = calloc(1, sizeof(*node)); + if (!node) { + return false; } - queue->all_tail = node; - if (queue->pending_tail) { - queue->pending_tail->next_pending = node; - } else { - queue->pending_head = node; - } - queue->pending_tail = node; - ++queue->count; - if (task_id) { - *task_id = id; - } - (void)pthread_cond_signal(&queue->condition); + (void)pthread_mutex_lock(&queue->mutex); + bool accepted = queue->pending_count < queue->capacity + && enqueue_locked(queue, node, task, task_id); (void)pthread_mutex_unlock(&queue->mutex); - return true; + if (!accepted) { + free(node); + } + return accepted; } bool @@ -308,7 +380,7 @@ lardon3d_task_queue_remove(Lardon3DTaskQueue *queue, uint64_t task_id) return false; } while (node->task == queue->active) { - (void)pthread_cond_wait(&queue->condition, &queue->mutex); + (void)pthread_cond_wait(&queue->not_empty, &queue->mutex); } if (previous) { previous->next_all = node->next_all; @@ -327,7 +399,6 @@ lardon3d_task_queue_remove(Lardon3DTaskQueue *queue, uint64_t task_id) if (pending) { unlink_pending(queue, pending_previous, pending); } - --queue->count; (void)pthread_mutex_unlock(&queue->mutex); lardon3d_task_destroy(node->task); free(node); @@ -341,7 +412,7 @@ lardon3d_task_queue_count(Lardon3DTaskQueue *queue) return 0; } (void)pthread_mutex_lock(&queue->mutex); - size_t count = queue->count; + size_t count = queue->pending_count; (void)pthread_mutex_unlock(&queue->mutex); return count; } diff --git a/tests/test_task_queue.c b/tests/test_task_queue.c index 056a7bb..21b4b9f 100644 --- a/tests/test_task_queue.c +++ b/tests/test_task_queue.c @@ -80,6 +80,67 @@ wait_terminal(Lardon3DTaskQueue *queue, uint64_t id, Lardon3DTaskSnapshot *resul return false; } +typedef struct { + Lardon3DTaskQueue *queue; + Lardon3DResourceEstimate estimate; + QueueWork *work; + uint64_t id; + bool added; +} ProducerThread; + +static void * +producer_thread(void *context) +{ + ProducerThread *producer = context; + Lardon3DTask *task = lardon3d_task_create( + "Producteur", + &producer->estimate, + queue_callback, + producer->work + ); + if (!task) { + producer->added = false; + return NULL; + } + producer->added = lardon3d_task_queue_add( + producer->queue, + task, + &producer->id + ); + if (!producer->added) { + lardon3d_task_destroy(task); + } + return NULL; +} + +/* Réserve toutes les ressources CPU pour que le worker ne puisse plus + consommer de tâche : la file d'attente reste pleine et les producteurs + bloquent de façon déterministe. */ +static bool +hold_resources( + Lardon3DResourceGovernor *governor, + Lardon3DResourceReservation **reservation +) +{ + const Lardon3DResourceSnapshot blocking_snapshot = { + .memory_available_bytes = UINT64_MAX, + .cpu_load_1m = 0.0, + }; + const Lardon3DResourceEstimate blocking_estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1024, + }; + Lardon3DResourceDecision decision; + return lardon3d_resource_governor_reserve( + governor, + &blocking_snapshot, + &blocking_estimate, + &decision, + reservation + ); +} + static bool run_test(void) { @@ -108,8 +169,8 @@ run_test(void) CHECK(governor); lardon3d_task_queue_destroy(NULL); CHECK(lardon3d_task_queue_count(NULL) == 0); - CHECK(!lardon3d_task_queue_create(NULL)); - Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor); + CHECK(!lardon3d_task_queue_create(NULL, 0)); + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 1024); CHECK(queue); short_pause(); @@ -134,7 +195,7 @@ run_test(void) CHECK(lardon3d_task_queue_add(queue, tasks[index], &ids[index])); CHECK(ids[index] == index + 1); } - CHECK(lardon3d_task_queue_count(queue) == TASK_COUNT); + CHECK(lardon3d_task_queue_count(queue) <= TASK_COUNT); Lardon3DTaskSnapshot snapshot; CHECK(wait_terminal(queue, ids[TASK_COUNT - 1], &snapshot)); CHECK(snapshot.state == TASK_COMPLETED); @@ -156,11 +217,11 @@ run_test(void) CHECK(!lardon3d_task_queue_get_at(queue, TASK_COUNT, &snapshot)); CHECK(lardon3d_task_queue_remove(queue, ids[0])); CHECK(!lardon3d_task_queue_get(queue, ids[0], &snapshot)); - CHECK(lardon3d_task_queue_count(queue) == TASK_COUNT - 1); + CHECK(lardon3d_task_queue_count(queue) == 0); lardon3d_task_queue_destroy(queue); CHECK(pthread_mutex_destroy(&log.mutex) == 0); - queue = lardon3d_task_queue_create(governor); + queue = lardon3d_task_queue_create(governor, 1024); CHECK(queue); OrderLog control_log = {0}; CHECK(pthread_mutex_init(&control_log.mutex, NULL) == 0); @@ -212,7 +273,7 @@ run_test(void) lardon3d_task_queue_destroy(queue); CHECK(pthread_mutex_destroy(&control_log.mutex) == 0); - queue = lardon3d_task_queue_create(governor); + queue = lardon3d_task_queue_create(governor, 1024); CHECK(queue); OrderLog destruction_log = {0}; CHECK(pthread_mutex_init(&destruction_log.mutex, NULL) == 0); @@ -253,7 +314,7 @@ run_test(void) &blocking_reservation )); CHECK(blocking_reservation); - queue = lardon3d_task_queue_create(governor); + queue = lardon3d_task_queue_create(governor, 1024); CHECK(queue); OrderLog wait_log = {0}; CHECK(pthread_mutex_init(&wait_log.mutex, NULL) == 0); @@ -318,7 +379,7 @@ run_test(void) &io_blocking_reservation )); CHECK(io_blocking_reservation); - queue = lardon3d_task_queue_create(governor); + queue = lardon3d_task_queue_create(governor, 1024); CHECK(queue); OrderLog bypass_log = {0}; CHECK(pthread_mutex_init(&bypass_log.mutex, NULL) == 0); @@ -368,7 +429,7 @@ run_test(void) lardon3d_task_queue_destroy(queue); CHECK(pthread_mutex_destroy(&bypass_log.mutex) == 0); - queue = lardon3d_task_queue_create(governor); + queue = lardon3d_task_queue_create(governor, 1024); CHECK(queue); OrderLog contract_log = {0}; CHECK(pthread_mutex_init(&contract_log.mutex, NULL) == 0); @@ -432,8 +493,433 @@ run_test(void) return true; } +static bool +test_enqueue_under_capacity(Lardon3DResourceGovernor *governor) +{ + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 4); + CHECK(queue); + OrderLog log = {0}; + CHECK(pthread_mutex_init(&log.mutex, NULL) == 0); + QueueWork work[3]; + Lardon3DTask *tasks[3]; + uint64_t ids[3]; + for (size_t index = 0; index < 3; ++index) { + work[index] = (QueueWork) { + .log = &log, + .value = index, + .steps = 1, + }; + tasks[index] = lardon3d_task_create( + "Sous capacité", + &estimate, + queue_callback, + &work[index] + ); + CHECK(tasks[index]); + CHECK(lardon3d_task_queue_add(queue, tasks[index], &ids[index])); + } + CHECK(lardon3d_task_queue_count(queue) <= 3); + Lardon3DTaskSnapshot snapshot; + for (size_t index = 0; index < 3; ++index) { + CHECK(wait_terminal(queue, ids[index], &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + } + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&log.mutex) == 0); + return true; +} + +static bool +test_capacity_reached_blocks(Lardon3DResourceGovernor *governor) +{ + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DResourceReservation *reservation; + CHECK(hold_resources(governor, &reservation)); + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 2); + CHECK(queue); + OrderLog log = {0}; + CHECK(pthread_mutex_init(&log.mutex, NULL) == 0); + QueueWork work[3]; + Lardon3DTask *tasks[2]; + uint64_t ids[2]; + for (size_t index = 0; index < 2; ++index) { + work[index] = (QueueWork) { + .log = &log, + .value = index, + .steps = 1, + }; + tasks[index] = lardon3d_task_create( + "File pleine", + &estimate, + queue_callback, + &work[index] + ); + CHECK(tasks[index]); + CHECK(lardon3d_task_queue_add(queue, tasks[index], &ids[index])); + } + CHECK(lardon3d_task_queue_count(queue) == 2); + work[2] = (QueueWork) {.log = &log, .value = 2, .steps = 1}; + ProducerThread producer = { + .queue = queue, + .estimate = estimate, + .work = &work[2], + }; + pthread_t thread; + CHECK(pthread_create(&thread, NULL, producer_thread, &producer) == 0); + short_pause(); + CHECK(!producer.added); + CHECK(lardon3d_resource_governor_release(governor, reservation)); + lardon3d_task_queue_resources_changed(queue); + CHECK(pthread_join(thread, NULL) == 0); + CHECK(producer.added); + Lardon3DTaskSnapshot snapshot; + for (size_t index = 0; index < 2; ++index) { + CHECK(wait_terminal(queue, ids[index], &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + } + CHECK(wait_terminal(queue, producer.id, &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&log.mutex) == 0); + return true; +} + +static bool +test_release_unblocks_producer(Lardon3DResourceGovernor *governor) +{ + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DResourceReservation *reservation; + CHECK(hold_resources(governor, &reservation)); + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 1); + CHECK(queue); + OrderLog log = {0}; + CHECK(pthread_mutex_init(&log.mutex, NULL) == 0); + QueueWork work[2]; + work[0] = (QueueWork) {.log = &log, .value = 0, .steps = 1}; + Lardon3DTask *first = lardon3d_task_create( + "Première", + &estimate, + queue_callback, + &work[0] + ); + CHECK(first); + uint64_t first_id; + CHECK(lardon3d_task_queue_add(queue, first, &first_id)); + CHECK(lardon3d_task_queue_count(queue) == 1); + work[1] = (QueueWork) {.log = &log, .value = 1, .steps = 1}; + ProducerThread producer = { + .queue = queue, + .estimate = estimate, + .work = &work[1], + }; + pthread_t thread; + CHECK(pthread_create(&thread, NULL, producer_thread, &producer) == 0); + short_pause(); + CHECK(!producer.added); + CHECK(lardon3d_resource_governor_release(governor, reservation)); + lardon3d_task_queue_resources_changed(queue); + Lardon3DTaskSnapshot snapshot; + CHECK(wait_terminal(queue, first_id, &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + CHECK(lardon3d_task_queue_remove(queue, first_id)); + CHECK(pthread_join(thread, NULL) == 0); + CHECK(producer.added); + CHECK(wait_terminal(queue, producer.id, &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&log.mutex) == 0); + return true; +} + +static bool +test_multiple_producers(Lardon3DResourceGovernor *governor) +{ + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 4); + CHECK(queue); + OrderLog log = {0}; + CHECK(pthread_mutex_init(&log.mutex, NULL) == 0); + QueueWork work[4]; + ProducerThread producers[4]; + pthread_t threads[4]; + for (size_t index = 0; index < 4; ++index) { + work[index] = (QueueWork) { + .log = &log, + .value = index, + .steps = 1, + }; + producers[index] = (ProducerThread) { + .queue = queue, + .estimate = estimate, + .work = &work[index], + }; + CHECK(pthread_create( + &threads[index], + NULL, + producer_thread, + &producers[index] + ) == 0); + } + for (size_t index = 0; index < 4; ++index) { + CHECK(pthread_join(threads[index], NULL) == 0); + CHECK(producers[index].added); + } + CHECK(lardon3d_task_queue_count(queue) <= 4); + Lardon3DTaskSnapshot snapshot; + for (size_t index = 0; index < 4; ++index) { + CHECK(wait_terminal(queue, producers[index].id, &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + } + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&log.mutex) == 0); + return true; +} + +static bool +test_shutdown_unblocks_producer(Lardon3DResourceGovernor *governor) +{ + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DResourceReservation *reservation; + CHECK(hold_resources(governor, &reservation)); + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 1); + CHECK(queue); + OrderLog log = {0}; + CHECK(pthread_mutex_init(&log.mutex, NULL) == 0); + QueueWork work[2]; + work[0] = (QueueWork) {.log = &log, .value = 0, .steps = 1}; + Lardon3DTask *first = lardon3d_task_create( + "Première", + &estimate, + queue_callback, + &work[0] + ); + CHECK(first); + uint64_t first_id; + CHECK(lardon3d_task_queue_add(queue, first, &first_id)); + CHECK(lardon3d_task_queue_count(queue) == 1); + work[1] = (QueueWork) {.log = &log, .value = 1, .steps = 1}; + ProducerThread producer = { + .queue = queue, + .estimate = estimate, + .work = &work[1], + }; + pthread_t thread; + CHECK(pthread_create(&thread, NULL, producer_thread, &producer) == 0); + short_pause(); + CHECK(!producer.added); + // destroy réveille le producteur via broadcast(¬_full) + lardon3d_task_queue_destroy(queue); + // Le producteur peut maintenant sortir de add et terminer + CHECK(pthread_join(thread, NULL) == 0); + CHECK(!producer.added); + CHECK(lardon3d_resource_governor_release(governor, reservation)); + CHECK(pthread_mutex_destroy(&log.mutex) == 0); + return true; +} + +static bool +test_shutdown_empty_queue(Lardon3DResourceGovernor *governor) +{ + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 1); + CHECK(queue); + lardon3d_task_queue_destroy(queue); + return true; +} + +static bool +test_pending_count_correct(Lardon3DResourceGovernor *governor) +{ + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DResourceReservation *reservation; + CHECK(hold_resources(governor, &reservation)); + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 4); + CHECK(queue); + OrderLog log = {0}; + CHECK(pthread_mutex_init(&log.mutex, NULL) == 0); + QueueWork work[2]; + Lardon3DTask *tasks[2]; + uint64_t ids[2]; + for (size_t index = 0; index < 2; ++index) { + work[index] = (QueueWork) { + .log = &log, + .value = index, + .steps = 1, + }; + tasks[index] = lardon3d_task_create( + "Comptage", + &estimate, + queue_callback, + &work[index] + ); + CHECK(tasks[index]); + CHECK(lardon3d_task_queue_add(queue, tasks[index], &ids[index])); + } + CHECK(lardon3d_task_queue_count(queue) == 2); + Lardon3DTaskSnapshot snapshot; + CHECK(lardon3d_task_queue_cancel(queue, ids[0])); + CHECK(wait_terminal(queue, ids[0], &snapshot)); + CHECK(snapshot.state == TASK_CANCELLED); + CHECK(lardon3d_task_queue_remove(queue, ids[0])); + CHECK(lardon3d_task_queue_count(queue) == 1); + CHECK(lardon3d_task_queue_cancel(queue, ids[1])); + CHECK(wait_terminal(queue, ids[1], &snapshot)); + CHECK(snapshot.state == TASK_CANCELLED); + CHECK(lardon3d_task_queue_remove(queue, ids[1])); + CHECK(lardon3d_task_queue_count(queue) == 0); + CHECK(lardon3d_resource_governor_release(governor, reservation)); + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&log.mutex) == 0); + return true; +} + +static bool +test_capacity_one(Lardon3DResourceGovernor *governor) +{ + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DResourceReservation *reservation; + CHECK(hold_resources(governor, &reservation)); + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 1); + CHECK(queue); + OrderLog log = {0}; + CHECK(pthread_mutex_init(&log.mutex, NULL) == 0); + QueueWork work = {.log = &log, .value = 0, .steps = 1}; + Lardon3DTask *task = lardon3d_task_create( + "Capacité un", + &estimate, + queue_callback, + &work + ); + CHECK(task); + uint64_t id; + CHECK(lardon3d_task_queue_add(queue, task, &id)); + CHECK(lardon3d_task_queue_count(queue) == 1); + CHECK(lardon3d_resource_governor_release(governor, reservation)); + lardon3d_task_queue_resources_changed(queue); + Lardon3DTaskSnapshot snapshot; + CHECK(wait_terminal(queue, id, &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&log.mutex) == 0); + return true; +} + +static bool +test_stress_concurrent(Lardon3DResourceGovernor *governor) +{ + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor, 8); + CHECK(queue); + OrderLog log = {0}; + CHECK(pthread_mutex_init(&log.mutex, NULL) == 0); + enum { PRODUCERS = 4, PER_PRODUCER = 4 }; + QueueWork work[PRODUCERS * PER_PRODUCER]; + ProducerThread producers[PRODUCERS * PER_PRODUCER]; + pthread_t threads[PRODUCERS * PER_PRODUCER]; + size_t index = 0; + for (size_t producer = 0; producer < PRODUCERS; ++producer) { + for (size_t item = 0; item < PER_PRODUCER; ++item) { + work[index] = (QueueWork) { + .log = &log, + .value = index, + .steps = 1, + }; + producers[index] = (ProducerThread) { + .queue = queue, + .estimate = estimate, + .work = &work[index], + }; + CHECK(pthread_create( + &threads[index], + NULL, + producer_thread, + &producers[index] + ) == 0); + ++index; + } + } + for (size_t i = 0; i < index; ++i) { + CHECK(pthread_join(threads[i], NULL) == 0); + CHECK(producers[i].added); + } + Lardon3DTaskSnapshot snapshot; + for (size_t i = 0; i < index; ++i) { + CHECK(wait_terminal(queue, producers[i].id, &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + } + CHECK(log.count == index); + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&log.mutex) == 0); + return true; +} + int main(void) { - return run_test() ? EXIT_SUCCESS : EXIT_FAILURE; + if (!run_test()) { + return EXIT_FAILURE; + } + Lardon3DHardwareProfile profile = { + .logical_cpu_count = 1024, + .page_size_bytes = 4096, + .memory_total_bytes = UINT64_MAX, + .cpu_architecture = "test", + }; + Lardon3DResourcePolicy policy = { + .system_memory_reserve_bytes = 0, + .system_cpu_reserve = 0, + .maximum_cpu_load_ratio = 1.0, + .maximum_io_pressure_avg10 = 100.0, + .io_slot_capacity = 1, + }; + Lardon3DResourceGovernor *governor = lardon3d_resource_governor_create( + &profile, + &policy + ); + if (!governor) { + return EXIT_FAILURE; + } + bool ok = test_enqueue_under_capacity(governor) + && test_capacity_reached_blocks(governor) + && test_release_unblocks_producer(governor) + && test_multiple_producers(governor) + && test_shutdown_unblocks_producer(governor) + && test_shutdown_empty_queue(governor) + && test_pending_count_correct(governor) + && test_capacity_one(governor) + && test_stress_concurrent(governor); + lardon3d_resource_governor_destroy(governor); + return ok ? EXIT_SUCCESS : EXIT_FAILURE; }