diff --git a/docs/architecture/scheduler_resource_integration.md b/docs/architecture/scheduler_resource_integration.md new file mode 100644 index 0000000..6bb4fff --- /dev/null +++ b/docs/architecture/scheduler_resource_integration.md @@ -0,0 +1,53 @@ +# Intégration du scheduler et du gouverneur + +## Responsabilités + +Le scheduler conserve l'ordre FIFO et exécute les callbacks. Il ne calcule +jamais les budgets : la RAM, le GPU, les CPU, les slots IO et la taille de lot +sont exclusivement arbitrés par le Resource Governor. + +Le cycle d'exécution est strict : + +```text +Task + → Resource Estimate + → Governor + → Reservation + → Task Queue + → Worker + → Release Reservation +``` + +Chaque tâche reçoit une estimation immuable à sa création. Son passage de +`PENDING` à `RUNNING` est interdit tant que le gouverneur n'a pas créé une +réservation active. Le callback ne reçoit pas l'objet opaque : il consulte une +copie du contrat contenant le lot, la RAM, la mémoire GPU, les CPU et les slots +accordés. + +## Admission + +Le worker examine la première tâche FIFO. `WAIT` la laisse en attente et le +worker dort sur la condition de la file. Une libération de ressources suivie de +`lardon3d_task_queue_resources_changed()` le réveille sans polling. +`REDUCE_BATCH` crée un contrat avec le lot réduit, qui est transmis à la tâche. +`REJECT` place la tâche en échec sans appeler son callback et conserve la raison +explicite du gouverneur. + +Après succès, échec ou annulation, le worker libère exactement une fois la +réservation puis réveille la file. La destruction annule les tâches, rejoint le +worker et libère toute réservation détenue avant de détruire les tâches. + +## Pause + +Dans cette première version, une tâche déjà démarrée conserve sa réservation +pendant `PAUSED`. Ses ressources restent donc indisponibles pour les autres +tâches. Ce choix évite de reprendre un callback avec un contrat qui aurait été +attribué entre-temps à un autre travail. + +## Limites et extensions + +La file possède un seul worker, reste strictement FIFO et ne gère ni priorité +ni dépendance. Une notification explicite est nécessaire lorsqu'un composant +extérieur libère une réservation. Les prochaines étapes pourront ajouter un +DAG, des priorités, des séquences de lots adaptatives et des pools distincts +CPU, IO et GPU sans déplacer les décisions de ressources hors du gouverneur. diff --git a/include/lardon3d/resource_governor.h b/include/lardon3d/resource_governor.h index 0b51d72..76d733c 100644 --- a/include/lardon3d/resource_governor.h +++ b/include/lardon3d/resource_governor.h @@ -139,6 +139,12 @@ bool lardon3d_resource_governor_reserve( Lardon3DResourceDecision *decision, Lardon3DResourceReservation **reservation ); +bool lardon3d_resource_governor_reserve_available( + Lardon3DResourceGovernor *governor, + const Lardon3DResourceEstimate *estimate, + Lardon3DResourceDecision *decision, + Lardon3DResourceReservation **reservation +); bool lardon3d_resource_governor_release( Lardon3DResourceGovernor *governor, Lardon3DResourceReservation *reservation @@ -152,6 +158,11 @@ bool lardon3d_resource_reservation_get( const Lardon3DResourceReservation *reservation, Lardon3DResourceReservationInfo *information ); +bool lardon3d_resource_reservation_get_active( + Lardon3DResourceGovernor *governor, + const Lardon3DResourceReservation *reservation, + Lardon3DResourceReservationInfo *information +); size_t lardon3d_resource_governor_list_reservations( Lardon3DResourceGovernor *governor, Lardon3DResourceReservationInfo *reservations, diff --git a/include/lardon3d/task.h b/include/lardon3d/task.h index 2531c74..d771a57 100644 --- a/include/lardon3d/task.h +++ b/include/lardon3d/task.h @@ -5,6 +5,8 @@ #include #include +#include + enum { LARDON3D_TASK_NAME_CAPACITY = 128, LARDON3D_TASK_MESSAGE_CAPACITY = 256, @@ -32,14 +34,28 @@ typedef struct { struct timespec finished_at; } Lardon3DTaskSnapshot; +typedef struct { + size_t batch_size; + uint64_t memory_bytes; + uint64_t gpu_memory_bytes; + unsigned int cpu_threads; + unsigned int gpu_slots; + unsigned int io_slots; +} Lardon3DTaskExecutionContract; + Lardon3DTask *lardon3d_task_create( const char *name, + const Lardon3DResourceEstimate *estimate, Lardon3DTaskCallback callback, void *userdata ); void lardon3d_task_destroy(Lardon3DTask *task); /* Exécute le callback dans le thread appelant. */ -bool lardon3d_task_start(Lardon3DTask *task); +bool lardon3d_task_start( + Lardon3DTask *task, + Lardon3DResourceGovernor *governor, + const Lardon3DResourceReservation *reservation +); void lardon3d_task_request_cancel(Lardon3DTask *task); bool lardon3d_task_pause(Lardon3DTask *task); bool lardon3d_task_resume(Lardon3DTask *task); @@ -57,6 +73,15 @@ bool lardon3d_task_snapshot( ); uint64_t lardon3d_task_id(const Lardon3DTask *task); bool lardon3d_task_assign_id(Lardon3DTask *task, uint64_t id); +bool lardon3d_task_resource_estimate( + const Lardon3DTask *task, + Lardon3DResourceEstimate *estimate +); +bool lardon3d_task_execution_contract( + const Lardon3DTask *task, + Lardon3DTaskExecutionContract *contract +); +bool lardon3d_task_reject(Lardon3DTask *task, const char *message); const char *lardon3d_task_state_name(Lardon3DTaskState state); #endif diff --git a/include/lardon3d/task_queue.h b/include/lardon3d/task_queue.h index 094592b..a6154ad 100644 --- a/include/lardon3d/task_queue.h +++ b/include/lardon3d/task_queue.h @@ -16,7 +16,9 @@ typedef struct { size_t total; } Lardon3DTaskQueueSummary; -Lardon3DTaskQueue *lardon3d_task_queue_create(void); +Lardon3DTaskQueue *lardon3d_task_queue_create( + Lardon3DResourceGovernor *governor +); void lardon3d_task_queue_destroy(Lardon3DTaskQueue *queue); /* La file devient propriétaire de task uniquement en cas de succès. */ bool lardon3d_task_queue_add( @@ -25,6 +27,8 @@ bool lardon3d_task_queue_add( 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); size_t lardon3d_task_queue_count(Lardon3DTaskQueue *queue); bool lardon3d_task_queue_get( Lardon3DTaskQueue *queue, diff --git a/meson.build b/meson.build index 50b4ca6..6680500 100644 --- a/meson.build +++ b/meson.build @@ -36,6 +36,8 @@ executable( 'src/project.c', 'src/task.c', 'src/task_queue.c', + 'src/resource_governor.c', + 'src/resource_snapshot.c', 'src/hardware_profile.c', 'src/resource_snapshot.c', 'src/resource_governor.c', @@ -100,6 +102,8 @@ task_test = executable( sources: [ 'tests/test_task.c', 'src/task.c', + 'src/resource_governor.c', + 'src/resource_snapshot.c', ], include_directories: include_directories('include'), dependencies: [threads], @@ -113,6 +117,8 @@ task_queue_test = executable( 'tests/test_task_queue.c', 'src/task.c', 'src/task_queue.c', + 'src/resource_governor.c', + 'src/resource_snapshot.c', ], include_directories: include_directories('include'), dependencies: [threads], @@ -148,6 +154,7 @@ resource_governor_test = executable( sources: [ 'tests/test_resource_governor.c', 'src/resource_governor.c', + 'src/resource_snapshot.c', ], include_directories: include_directories('include'), dependencies: [threads], @@ -160,6 +167,7 @@ resource_reservation_test = executable( sources: [ 'tests/test_resource_reservation.c', 'src/resource_governor.c', + 'src/resource_snapshot.c', ], include_directories: include_directories('include'), dependencies: [threads], diff --git a/src/app.c b/src/app.c index 3db2a1e..2a66ebd 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.task_queue = lardon3d_task_queue_create(state.resource_governor); if (!state.task_queue) { lardon3d_resource_governor_destroy(state.resource_governor); return EXIT_FAILURE; diff --git a/src/resource_governor.c b/src/resource_governor.c index e556448..f34579c 100644 --- a/src/resource_governor.c +++ b/src/resource_governor.c @@ -558,6 +558,33 @@ lardon3d_resource_governor_reserve( return true; } +bool +lardon3d_resource_governor_reserve_available( + Lardon3DResourceGovernor *governor, + const Lardon3DResourceEstimate *estimate, + Lardon3DResourceDecision *decision, + Lardon3DResourceReservation **reservation +) +{ + if (!governor || !estimate || !decision || !reservation) { + return false; + } + (void)pthread_mutex_lock(&governor->mutex); + Lardon3DHardwareProfile profile = governor->profile; + (void)pthread_mutex_unlock(&governor->mutex); + Lardon3DResourceSnapshot snapshot; + if (!lardon3d_resource_snapshot_capture(&profile, &snapshot, NULL, 0)) { + return false; + } + return lardon3d_resource_governor_reserve( + governor, + &snapshot, + estimate, + decision, + reservation + ); +} + static Lardon3DResourceReservation * find_reservation( Lardon3DResourceReservation *head, @@ -652,6 +679,28 @@ lardon3d_resource_reservation_get( return found != NULL; } +bool +lardon3d_resource_reservation_get_active( + Lardon3DResourceGovernor *governor, + const Lardon3DResourceReservation *reservation, + Lardon3DResourceReservationInfo *information +) +{ + if (!governor || !reservation || !information) { + return false; + } + (void)pthread_mutex_lock(&governor->mutex); + Lardon3DResourceReservation *found = find_reservation( + governor->active, + reservation + ); + if (found) { + *information = found->information; + } + (void)pthread_mutex_unlock(&governor->mutex); + return found != NULL; +} + size_t lardon3d_resource_governor_list_reservations( Lardon3DResourceGovernor *governor, diff --git a/src/task.c b/src/task.c index 740e8f7..c77113e 100644 --- a/src/task.c +++ b/src/task.c @@ -17,6 +17,9 @@ struct Lardon3DTask { struct timespec finished_at; Lardon3DTaskCallback callback; void *userdata; + Lardon3DResourceEstimate estimate; + Lardon3DTaskExecutionContract contract; + bool has_contract; bool pause_requested; bool cancel_requested; bool executing; @@ -57,11 +60,12 @@ finish_locked( Lardon3DTask * lardon3d_task_create( const char *name, + const Lardon3DResourceEstimate *estimate, Lardon3DTaskCallback callback, void *userdata ) { - if (!name || !name[0] || !callback) { + if (!name || !name[0] || !estimate || !callback) { return NULL; } Lardon3DTask *task = calloc(1, sizeof(*task)); @@ -82,6 +86,7 @@ lardon3d_task_create( task->state = TASK_PENDING; task->callback = callback; task->userdata = userdata; + task->estimate = *estimate; copy_text(task->message, sizeof(task->message), "En attente."); return task; } @@ -100,9 +105,18 @@ lardon3d_task_destroy(Lardon3DTask *task) } bool -lardon3d_task_start(Lardon3DTask *task) +lardon3d_task_start( + Lardon3DTask *task, + Lardon3DResourceGovernor *governor, + const Lardon3DResourceReservation *reservation +) { - if (!task) { + Lardon3DResourceReservationInfo information; + if (!task || !lardon3d_resource_reservation_get_active( + governor, + reservation, + &information + )) { return false; } (void)pthread_mutex_lock(&task->mutex); @@ -111,6 +125,15 @@ lardon3d_task_start(Lardon3DTask *task) return false; } task->executing = true; + task->contract = (Lardon3DTaskExecutionContract) { + .batch_size = information.batch_size, + .memory_bytes = information.memory_bytes, + .gpu_memory_bytes = information.gpu_memory_bytes, + .cpu_threads = information.cpu_threads, + .gpu_slots = information.gpu_slots, + .io_slots = information.io_slots, + }; + task->has_contract = true; (void)clock_gettime(CLOCK_REALTIME, &task->started_at); if (task->cancel_requested) { finish_locked(task, TASK_CANCELLED, "Tâche annulée."); @@ -333,6 +356,60 @@ lardon3d_task_assign_id(Lardon3DTask *task, uint64_t id) return accepted; } +bool +lardon3d_task_resource_estimate( + const Lardon3DTask *task, + Lardon3DResourceEstimate *estimate +) +{ + if (!task || !estimate) { + return false; + } + Lardon3DTask *mutable_task = (Lardon3DTask *)task; + (void)pthread_mutex_lock(&mutable_task->mutex); + *estimate = task->estimate; + (void)pthread_mutex_unlock(&mutable_task->mutex); + return true; +} + +bool +lardon3d_task_execution_contract( + const Lardon3DTask *task, + Lardon3DTaskExecutionContract *contract +) +{ + if (!task || !contract) { + return false; + } + Lardon3DTask *mutable_task = (Lardon3DTask *)task; + (void)pthread_mutex_lock(&mutable_task->mutex); + bool available = task->has_contract; + if (available) { + *contract = task->contract; + } + (void)pthread_mutex_unlock(&mutable_task->mutex); + return available; +} + +bool +lardon3d_task_reject(Lardon3DTask *task, const char *message) +{ + if (!task) { + return false; + } + (void)pthread_mutex_lock(&task->mutex); + bool accepted = !task->executing && !is_terminal(task->state); + if (accepted) { + finish_locked( + task, + TASK_FAILED, + message ? message : "Ressources impossibles à réserver." + ); + } + (void)pthread_mutex_unlock(&task->mutex); + return accepted; +} + const char * lardon3d_task_state_name(Lardon3DTaskState state) { diff --git a/src/task_queue.c b/src/task_queue.c index 8a89201..677bbb3 100644 --- a/src/task_queue.c +++ b/src/task_queue.c @@ -17,6 +17,7 @@ struct Lardon3DTaskQueue { pthread_t worker; bool worker_started; bool stopping; + Lardon3DResourceGovernor *governor; uint64_t next_id; TaskNode *all_head; TaskNode *all_tail; @@ -47,6 +48,48 @@ queue_worker(void *context) return NULL; } TaskNode *node = queue->pending_head; + Lardon3DTaskSnapshot task_snapshot; + if (!lardon3d_task_snapshot(node->task, &task_snapshot)) { + (void)pthread_mutex_unlock(&queue->mutex); + continue; + } + if (terminal_state(task_snapshot.state)) { + queue->pending_head = node->next_pending; + if (!queue->pending_head) { + queue->pending_tail = NULL; + } + node->next_pending = NULL; + (void)pthread_mutex_unlock(&queue->mutex); + continue; + } + Lardon3DResourceEstimate estimate; + Lardon3DResourceDecision decision; + Lardon3DResourceReservation *reservation = NULL; + bool evaluated = lardon3d_task_resource_estimate(node->task, &estimate) + && lardon3d_resource_governor_reserve_available( + queue->governor, + &estimate, + &decision, + &reservation + ); + if (!evaluated) { + (void)lardon3d_task_reject( + node->task, + "Impossible d'évaluer les ressources disponibles." + ); + (void)pthread_mutex_unlock(&queue->mutex); + continue; + } + if (decision.kind == LARDON3D_RESOURCE_WAIT) { + (void)pthread_cond_wait(&queue->condition, &queue->mutex); + (void)pthread_mutex_unlock(&queue->mutex); + continue; + } + if (decision.kind == LARDON3D_RESOURCE_REJECT || !reservation) { + (void)lardon3d_task_reject(node->task, decision.reason); + (void)pthread_mutex_unlock(&queue->mutex); + continue; + } queue->pending_head = node->next_pending; if (!queue->pending_head) { queue->pending_tail = NULL; @@ -55,7 +98,20 @@ queue_worker(void *context) queue->active = node->task; (void)pthread_mutex_unlock(&queue->mutex); - (void)lardon3d_task_start(node->task); + if (!lardon3d_task_start( + node->task, + queue->governor, + reservation + )) { + (void)lardon3d_task_reject( + node->task, + "Réservation de ressources invalide." + ); + } + (void)lardon3d_resource_governor_release( + queue->governor, + reservation + ); (void)pthread_mutex_lock(&queue->mutex); queue->active = NULL; @@ -65,8 +121,11 @@ queue_worker(void *context) } Lardon3DTaskQueue * -lardon3d_task_queue_create(void) +lardon3d_task_queue_create(Lardon3DResourceGovernor *governor) { + if (!governor) { + return NULL; + } Lardon3DTaskQueue *queue = calloc(1, sizeof(*queue)); if (!queue) { return NULL; @@ -81,6 +140,7 @@ lardon3d_task_queue_create(void) return NULL; } queue->next_id = 1; + queue->governor = governor; if (pthread_create(&queue->worker, NULL, queue_worker, queue) != 0) { (void)pthread_cond_destroy(&queue->condition); (void)pthread_mutex_destroy(&queue->mutex); @@ -91,6 +151,36 @@ lardon3d_task_queue_create(void) return queue; } +bool +lardon3d_task_queue_cancel(Lardon3DTaskQueue *queue, uint64_t task_id) +{ + if (!queue || task_id == 0) { + return false; + } + (void)pthread_mutex_lock(&queue->mutex); + TaskNode *node = queue->all_head; + while (node && lardon3d_task_id(node->task) != task_id) { + node = node->next_all; + } + if (node) { + lardon3d_task_request_cancel(node->task); + (void)pthread_cond_broadcast(&queue->condition); + } + (void)pthread_mutex_unlock(&queue->mutex); + return node != NULL; +} + +void +lardon3d_task_queue_resources_changed(Lardon3DTaskQueue *queue) +{ + if (!queue) { + return; + } + (void)pthread_mutex_lock(&queue->mutex); + (void)pthread_cond_broadcast(&queue->condition); + (void)pthread_mutex_unlock(&queue->mutex); +} + void lardon3d_task_queue_destroy(Lardon3DTaskQueue *queue) { diff --git a/tests/test_task.c b/tests/test_task.c index bc43cd3..3629ed3 100644 --- a/tests/test_task.c +++ b/tests/test_task.c @@ -20,6 +20,12 @@ typedef struct { long pause_ns; } Work; +typedef struct { + Lardon3DTask *task; + Lardon3DResourceGovernor *governor; + Lardon3DResourceReservation *reservation; +} StartContext; + static void short_pause(long nanoseconds) { @@ -54,8 +60,12 @@ failure_callback(Lardon3DTask *task, void *userdata) static void * start_task(void *context) { - Lardon3DTask *task = context; - return (void *)(uintptr_t)(lardon3d_task_start(task) ? 1 : 0); + StartContext *start = context; + return (void *)(uintptr_t)(lardon3d_task_start( + start->task, + start->governor, + start->reservation + ) ? 1 : 0); } static bool @@ -77,14 +87,46 @@ wait_for_state(Lardon3DTask *task, Lardon3DTaskState expected) static bool run_test(void) { - CHECK(!lardon3d_task_create(NULL, work_callback, NULL)); - CHECK(!lardon3d_task_create("", work_callback, NULL)); - CHECK(!lardon3d_task_create("invalide", NULL, NULL)); + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + Lardon3DHardwareProfile profile = { + .logical_cpu_count = 4, + .page_size_bytes = 4096, + .memory_total_bytes = UINT64_MAX, + .cpu_architecture = "test", + }; + Lardon3DResourcePolicy policy = { + .maximum_cpu_load_ratio = 1.0, + .maximum_io_pressure_avg10 = 100.0, + .io_slot_capacity = 1, + }; + Lardon3DResourceGovernor *governor = lardon3d_resource_governor_create( + &profile, + &policy + ); + Lardon3DResourceSnapshot resource_snapshot = { + .memory_available_bytes = UINT64_MAX, + .cpu_load_1m = 0.0, + }; + Lardon3DResourceDecision decision; + Lardon3DResourceReservation *reservation; + CHECK(governor); + CHECK(!lardon3d_task_create(NULL, &estimate, work_callback, NULL)); + CHECK(!lardon3d_task_create("", &estimate, work_callback, NULL)); + CHECK(!lardon3d_task_create("invalide", &estimate, NULL, NULL)); lardon3d_task_destroy(NULL); CHECK(!lardon3d_task_join(NULL)); Work work = {.steps = 100, .pause_ns = 1000000}; - Lardon3DTask *task = lardon3d_task_create("Tâche de test", work_callback, &work); + Lardon3DTask *task = lardon3d_task_create( + "Tâche de test", + &estimate, + work_callback, + &work + ); CHECK(task); CHECK(lardon3d_task_assign_id(task, 42)); CHECK(!lardon3d_task_assign_id(task, 43)); @@ -96,7 +138,11 @@ run_test(void) CHECK(strcmp(snapshot.name, "Tâche de test") == 0); pthread_t thread; - CHECK(pthread_create(&thread, NULL, start_task, task) == 0); + CHECK(lardon3d_resource_governor_reserve( + governor, &resource_snapshot, &estimate, &decision, &reservation + )); + StartContext start = {task, governor, reservation}; + CHECK(pthread_create(&thread, NULL, start_task, &start) == 0); CHECK(wait_for_state(task, TASK_RUNNING)); CHECK(lardon3d_task_pause(task)); CHECK(wait_for_state(task, TASK_PAUSED)); @@ -111,34 +157,51 @@ run_test(void) void *thread_result; CHECK(pthread_join(thread, &thread_result) == 0); CHECK((uintptr_t)thread_result == 1); + CHECK(lardon3d_resource_governor_release(governor, reservation)); CHECK(lardon3d_task_snapshot(task, &snapshot)); CHECK(snapshot.state == TASK_COMPLETED); CHECK(snapshot.progress == 100); CHECK(snapshot.started_at.tv_sec > 0); CHECK(snapshot.finished_at.tv_sec > 0); - CHECK(!lardon3d_task_start(task)); + CHECK(!lardon3d_task_start(task, NULL, NULL)); lardon3d_task_destroy(task); work = (Work) {.steps = 1000, .pause_ns = 1000000}; - task = lardon3d_task_create("Annulation", work_callback, &work); - CHECK(task && pthread_create(&thread, NULL, start_task, task) == 0); + task = lardon3d_task_create("Annulation", &estimate, work_callback, &work); + CHECK(task); + CHECK(lardon3d_resource_governor_reserve( + governor, &resource_snapshot, &estimate, &decision, &reservation + )); + start = (StartContext) {task, governor, reservation}; + CHECK(pthread_create(&thread, NULL, start_task, &start) == 0); CHECK(wait_for_state(task, TASK_RUNNING)); lardon3d_task_request_cancel(task); CHECK(lardon3d_task_join(task)); CHECK(pthread_join(thread, NULL) == 0); + CHECK(lardon3d_resource_governor_release(governor, reservation)); CHECK(lardon3d_task_snapshot(task, &snapshot)); CHECK(snapshot.state == TASK_CANCELLED); CHECK(snapshot.progress < 100); lardon3d_task_destroy(task); - task = lardon3d_task_create("Échec", failure_callback, NULL); - CHECK(task && lardon3d_task_start(task)); + task = lardon3d_task_create("Échec", &estimate, failure_callback, NULL); + CHECK(task); + CHECK(lardon3d_resource_governor_reserve( + governor, &resource_snapshot, &estimate, &decision, &reservation + )); + CHECK(lardon3d_task_start(task, governor, reservation)); + CHECK(lardon3d_resource_governor_release(governor, reservation)); CHECK(lardon3d_task_snapshot(task, &snapshot)); CHECK(snapshot.state == TASK_FAILED); CHECK(strcmp(snapshot.message, "Erreur contrôlée.") == 0); lardon3d_task_destroy(task); - task = lardon3d_task_create("Pause avant départ", work_callback, &work); + task = lardon3d_task_create( + "Pause avant départ", + &estimate, + work_callback, + &work + ); CHECK(task && lardon3d_task_pause(task)); CHECK(lardon3d_task_snapshot(task, &snapshot)); CHECK(snapshot.state == TASK_PAUSED); @@ -146,6 +209,7 @@ run_test(void) lardon3d_task_request_cancel(task); CHECK(lardon3d_task_join(task)); lardon3d_task_destroy(task); + lardon3d_resource_governor_destroy(governor); return true; } diff --git a/tests/test_task_queue.c b/tests/test_task_queue.c index 31dc9fa..0e84cba 100644 --- a/tests/test_task_queue.c +++ b/tests/test_task_queue.c @@ -28,6 +28,9 @@ typedef struct { OrderLog *log; size_t value; size_t steps; + Lardon3DTaskExecutionContract contract; + bool contract_seen; + bool fail; } QueueWork; static void @@ -41,6 +44,10 @@ static bool queue_callback(Lardon3DTask *task, void *userdata) { QueueWork *work = userdata; + if (!lardon3d_task_execution_contract(task, &work->contract)) { + return false; + } + work->contract_seen = true; (void)pthread_mutex_lock(&work->log->mutex); work->log->order[work->log->count++] = work->value; (void)pthread_mutex_unlock(&work->log->mutex); @@ -54,7 +61,7 @@ queue_callback(Lardon3DTask *task, void *userdata) short_pause(); } } - return true; + return !work->fail; } static bool @@ -76,9 +83,33 @@ wait_terminal(Lardon3DTaskQueue *queue, uint64_t id, Lardon3DTaskSnapshot *resul static bool run_test(void) { + 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 + ); + const Lardon3DResourceEstimate estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + CHECK(governor); lardon3d_task_queue_destroy(NULL); CHECK(lardon3d_task_queue_count(NULL) == 0); - Lardon3DTaskQueue *queue = lardon3d_task_queue_create(); + CHECK(!lardon3d_task_queue_create(NULL)); + Lardon3DTaskQueue *queue = lardon3d_task_queue_create(governor); CHECK(queue); short_pause(); @@ -93,7 +124,12 @@ run_test(void) .value = index, .steps = 1, }; - tasks[index] = lardon3d_task_create("FIFO", queue_callback, &work[index]); + tasks[index] = lardon3d_task_create( + "FIFO", + &estimate, + queue_callback, + &work[index] + ); CHECK(tasks[index]); CHECK(lardon3d_task_queue_add(queue, tasks[index], &ids[index])); CHECK(ids[index] == index + 1); @@ -105,6 +141,8 @@ run_test(void) CHECK(log.count == TASK_COUNT); for (size_t index = 0; index < TASK_COUNT; ++index) { CHECK(log.order[index] == index); + CHECK(work[index].contract_seen); + CHECK(work[index].contract.batch_size == 1); } Lardon3DTaskQueueSummary summary; Lardon3DTaskSnapshot listed[8]; @@ -122,15 +160,21 @@ run_test(void) lardon3d_task_queue_destroy(queue); CHECK(pthread_mutex_destroy(&log.mutex) == 0); - queue = lardon3d_task_queue_create(); + queue = lardon3d_task_queue_create(governor); CHECK(queue); OrderLog control_log = {0}; CHECK(pthread_mutex_init(&control_log.mutex, NULL) == 0); QueueWork slow = {.log = &control_log, .value = 1, .steps = 500}; QueueWork cancelled = {.log = &control_log, .value = 2, .steps = 1}; - Lardon3DTask *slow_task = lardon3d_task_create("Longue", queue_callback, &slow); + Lardon3DTask *slow_task = lardon3d_task_create( + "Longue", + &estimate, + queue_callback, + &slow + ); Lardon3DTask *cancelled_task = lardon3d_task_create( "Annulée en attente", + &estimate, queue_callback, &cancelled ); @@ -155,7 +199,8 @@ run_test(void) short_pause(); } CHECK(snapshot.state == TASK_PAUSED); - lardon3d_task_request_cancel(cancelled_task); + CHECK(lardon3d_resource_governor_reservation_count(governor) == 1); + CHECK(lardon3d_task_queue_cancel(queue, cancelled_id)); CHECK(lardon3d_task_resume(slow_task)); CHECK(wait_terminal(queue, slow_id, &snapshot)); CHECK(snapshot.state == TASK_COMPLETED); @@ -167,7 +212,7 @@ run_test(void) lardon3d_task_queue_destroy(queue); CHECK(pthread_mutex_destroy(&control_log.mutex) == 0); - queue = lardon3d_task_queue_create(); + queue = lardon3d_task_queue_create(governor); CHECK(queue); OrderLog destruction_log = {0}; CHECK(pthread_mutex_init(&destruction_log.mutex, NULL) == 0); @@ -178,6 +223,7 @@ run_test(void) }; Lardon3DTask *destruction_task = lardon3d_task_create( "Destruction sûre", + &estimate, queue_callback, &destruction_work ); @@ -185,7 +231,132 @@ run_test(void) CHECK(lardon3d_task_queue_add(queue, destruction_task, NULL)); short_pause(); lardon3d_task_queue_destroy(queue); + CHECK(lardon3d_resource_governor_reservation_count(governor) == 0); CHECK(pthread_mutex_destroy(&destruction_log.mutex) == 0); + + Lardon3DResourceSnapshot blocking_snapshot = { + .memory_available_bytes = UINT64_MAX, + .cpu_load_1m = 0.0, + }; + Lardon3DResourceEstimate blocking_estimate = { + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1024, + }; + Lardon3DResourceDecision decision; + Lardon3DResourceReservation *blocking_reservation; + CHECK(lardon3d_resource_governor_reserve( + governor, + &blocking_snapshot, + &blocking_estimate, + &decision, + &blocking_reservation + )); + CHECK(blocking_reservation); + queue = lardon3d_task_queue_create(governor); + CHECK(queue); + OrderLog wait_log = {0}; + CHECK(pthread_mutex_init(&wait_log.mutex, NULL) == 0); + QueueWork waiting = {.log = &wait_log, .value = 1, .steps = 1}; + Lardon3DTask *waiting_task = lardon3d_task_create( + "Attente ressources", + &estimate, + queue_callback, + &waiting + ); + uint64_t waiting_id; + CHECK(waiting_task); + CHECK(lardon3d_task_queue_add(queue, waiting_task, &waiting_id)); + short_pause(); + CHECK(lardon3d_task_queue_get(queue, waiting_id, &snapshot)); + CHECK(snapshot.state == TASK_PENDING); + CHECK(!waiting.contract_seen); + CHECK(lardon3d_task_queue_cancel(queue, waiting_id)); + CHECK(wait_terminal(queue, waiting_id, &snapshot)); + CHECK(snapshot.state == TASK_CANCELLED); + CHECK(!waiting.contract_seen); + QueueWork awakened = {.log = &wait_log, .value = 2, .steps = 1}; + Lardon3DTask *awakened_task = lardon3d_task_create( + "Réveil ressources", + &estimate, + queue_callback, + &awakened + ); + uint64_t awakened_id; + CHECK(awakened_task); + CHECK(lardon3d_task_queue_add(queue, awakened_task, &awakened_id)); + short_pause(); + CHECK(!awakened.contract_seen); + CHECK(lardon3d_resource_governor_release(governor, blocking_reservation)); + lardon3d_task_queue_resources_changed(queue); + CHECK(wait_terminal(queue, awakened_id, &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + CHECK(awakened.contract_seen); + CHECK(lardon3d_resource_governor_reservation_count(governor) == 0); + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&wait_log.mutex) == 0); + + queue = lardon3d_task_queue_create(governor); + CHECK(queue); + OrderLog contract_log = {0}; + CHECK(pthread_mutex_init(&contract_log.mutex, NULL) == 0); + Lardon3DResourceEstimate reduced_estimate = estimate; + reduced_estimate.desired_cpu_threads = 2048; + QueueWork reduced = {.log = &contract_log, .value = 1, .steps = 1}; + Lardon3DTask *reduced_task = lardon3d_task_create( + "Contrat réduit", + &reduced_estimate, + queue_callback, + &reduced + ); + uint64_t reduced_id; + CHECK(reduced_task); + CHECK(lardon3d_task_queue_add(queue, reduced_task, &reduced_id)); + CHECK(wait_terminal(queue, reduced_id, &snapshot)); + CHECK(snapshot.state == TASK_COMPLETED); + CHECK(reduced.contract.cpu_threads == 1024); + Lardon3DResourceEstimate rejected_estimate = { + .memory_fixed_bytes = UINT64_MAX, + .memory_bytes_per_item = 1, + .minimum_batch_size = 1, + .maximum_batch_size = 1, + .desired_cpu_threads = 1, + }; + QueueWork rejected = {.log = &contract_log, .value = 2, .steps = 1}; + Lardon3DTask *rejected_task = lardon3d_task_create( + "Impossible", + &rejected_estimate, + queue_callback, + &rejected + ); + uint64_t rejected_id; + CHECK(rejected_task); + CHECK(lardon3d_task_queue_add(queue, rejected_task, &rejected_id)); + CHECK(wait_terminal(queue, rejected_id, &snapshot)); + CHECK(snapshot.state == TASK_FAILED); + CHECK(!rejected.contract_seen); + QueueWork failure = { + .log = &contract_log, + .value = 3, + .steps = 1, + .fail = true, + }; + Lardon3DTask *failure_task = lardon3d_task_create( + "Échec callback", + &estimate, + queue_callback, + &failure + ); + uint64_t failure_id; + CHECK(failure_task); + CHECK(lardon3d_task_queue_add(queue, failure_task, &failure_id)); + CHECK(wait_terminal(queue, failure_id, &snapshot)); + CHECK(snapshot.state == TASK_FAILED); + CHECK(failure.contract_seen); + CHECK(lardon3d_resource_governor_reservation_count(governor) == 0); + lardon3d_task_queue_destroy(queue); + CHECK(pthread_mutex_destroy(&contract_log.mutex) == 0); + lardon3d_resource_governor_destroy(governor); return true; }