fy59/lardon3d
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(task): add durable task checkpoint foundations

Browse source
This commit is contained in:
fy59 2026-08-08 10:17:28 +02:00
parent 5647f780b0
commit 2b6e9bf444
12 changed files with 1110 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
README.md
View file

@ -25,7 +25,8 @@ Lardon3D ne vise pas simplement "dossier de photos → objet 3D", mais "ensemble
- **Import Task** : wrapper asynchrone avec états et progression
- **Image Catalog** : indexage des métadonnées d'images
- **Image View** : vues triées et filtrées pour la TUI
- **Task** : moteur de tâches avec pause/reprise, annulation, checkpoints
- **Task** : moteur de tâches avec pause/reprise, annulation et séquences
- **Task Checkpoint v1** : snapshot durable, fichier atomique et reprise sûre
- **Task Queue** : file FIFO avec sélection adaptative et backpressure
- **Hardware Profile** : détection des capacités matérielles
- **Resource Snapshot** : capture instantanée des ressources
@ -38,7 +39,7 @@ Lardon3D ne vise pas simplement "dossier de photos → objet 3D", mais "ensemble
### Briques prévues (PLANNED)
- Persistance des tâches et checkpoints
- Branchement des checkpoints aux pipelines et reprise globale
- DAG de dépendances
- Pools de workers multiples (CPU/GPU/IO)
- Publication live validée

67
docs/architecture/persistence.md
View file

@ -52,4 +52,69 @@ Les éléments suivants sont des concepts de domaine, PAS des tables SQL imposé
- Les artefacts partiels ne sont jamais considérés comme valides
- La reprise commence à la dernière frontière connue
## Statut : PLANNED (direction architecturale)
## Checkpoint durable de tâche v1
### État durable
Le modèle durable versionné contient uniquement l'identifiant stable, le nom,
l'estimation immuable, l'état observé, l'état de reprise, la progression, le
message, les horodatages et le compteur de séquences. Il ne contient aucun gros
artefact numérique. Une future version pourra référencer des identifiants
d'artefacts publiés et validés sans incorporer leur contenu.
Les mutex, conditions, callbacks, userdata, workers, gouverneur, réservations et
contrats d'exécution sont transitoires et ne sont jamais sérialisés.
### Normalisation après arrêt de processus
| État observé | État restauré |
|---|---|
| `TASK_PENDING` | `TASK_PENDING` |
| `TASK_RUNNING` | `TASK_PENDING` |
| `TASK_PAUSED` | `TASK_PENDING` |
| `TASK_COMPLETED` | `TASK_COMPLETED` |
| `TASK_FAILED` | `TASK_FAILED` |
| `TASK_CANCELLED` | `TASK_CANCELLED` |
Une rupture de séquence n'est pas un état : elle est observée comme
`TASK_RUNNING`. Son `sequence_count` est durable, mais la reprise revient à
`TASK_PENDING` et exige une nouvelle admission.
### Stockage minimal
Le codec v1 est indépendant de la future Project Database. Le fichier est de
taille fixe et bornée, encodé champ par champ, avec magie, version, taille et
checksum de payload.
La publication écrit un fichier temporaire unique dans le même répertoire,
effectue `fsync`, renomme atomiquement puis synchronise le répertoire parent.
La lecture distingue absence, corruption, version inconnue et erreur d'I/O.
La sauvegarde distingue trois frontières :
- avant `rename`, toute erreur retourne `IO_ERROR`, supprime le temporaire et
laisse l'ancien checkpoint publié inchangé ;
- après un `rename` réussi, le nouveau checkpoint est publié et visible et
n'est jamais présenté comme rollbackable ;
- si le `fsync` du répertoire échoue après ce `rename`, le résultat est
`PUBLISHED_NOT_DURABLE` : le fichier visible est valide, mais sa présence sous
ce nom après un crash ou une coupure n'est pas garantie. `OK` garantit que le
contenu et l'entrée de répertoire ont tous deux été synchronisés avec succès,
sous réserve des garanties fournies par le système de fichiers et le stockage.
Les tailles persistantes sont refusées avant conversion lorsqu'elles dépassent
`SIZE_MAX`. Les secondes sont des entiers non signés v1 : les timestamps
négatifs ne sont pas sérialisables et une valeur lue doit être représentable
par le `time_t` local avant conversion. Le format reste donc lisible entre
plateformes uniquement pour les valeurs communes à leurs domaines `size_t` et
`time_t`.
## Statut
**IMPLEMENTED** — modèle durable, codec v1, lecture validée, publication
atomique et restauration sûre d'une tâche isolée.
**NOT_YET_WIRED** — les pipelines et le scheduler ne déclenchent pas encore les
sauvegardes et ne rechargent pas globalement les tâches au démarrage.
**PLANNED** — Project Database SQLite, catalogue d'artefacts réels, transactions
entre métadonnées et artefacts, migration de formats et reprise globale.

15
docs/architecture/project_database.md
View file

@ -154,4 +154,17 @@ La base de données projet stocke les métadonnées de reconstruction et les rel
- Les artefacts partiels ne sont jamais considérés comme valides
- La reprise commence à la dernière frontière connue
## Statut : PLANNED (direction architecturale)
## Frontière avec les checkpoints de tâche
Le modèle durable v1 et son codec fichier sont implémentés indépendamment du
stockage. La future base réutilisera les mêmes règles de normalisation et de
validation ; elle ne stockera jamais les objets pthread, callbacks, pointeurs,
contrats ou réservations. Le fichier par tâche est une fondation, pas une
Project Database miniature.
## Statut
**NOT_YET_WIRED** — le modèle de checkpoint est prêt à être consommé.
**PLANNED** — schéma SQLite, migrations, transactions, inventaire d'artefacts,
chargement global et coordination avec le scheduler.

23
docs/architecture/runtime.md
View file

@ -57,7 +57,28 @@
- Worker unique (pas de pools multiples)
- Pas de parallélisme inter-tâches
- Pas de persistance des états
- Checkpoints isolés disponibles mais pas encore orchestrés au démarrage
## Reprise durable
Un snapshot ne conserve que l'état logique d'une tâche. `RUNNING` et `PAUSED`
sont normalisés vers `PENDING`; aucun worker, callback brut, pointeur, contrat
ou réservation n'est restauré. Le propriétaire fournit un nouveau callback et
resoumet la tâche. Les états terminaux sont conservés.
`started_at` désigne le début de la tentative d'exécution courante, pas le
premier démarrage historique. Un checkpoint `RUNNING` restauré en `PENDING`
conserve temporairement l'horodatage de la tentative interrompue pour
l'observation ; lors de `lardon3d_task_start()`, `started_at` est remplacé par le
nouveau démarrage et `finished_at` est remis à zéro. `finished_at` n'est fixé
qu'à la terminaison de cette tentative.
**IMPLEMENTED** — snapshot, codec v1 et restauration isolée.
**NOT_YET_WIRED** — sauvegarde périodique, chargement de projet et resoumission
automatique.
**PLANNED** — reprise globale du scheduler via la Project Database.
## Invariants

78
docs/architecture/task_system.md
View file

@ -16,55 +16,50 @@ coûts, exécution sous réservation et publication d'un résultat validé.
## Types principaux
```c
typedef enum {
TASK_STATE_IDLE,
TASK_STATE_QUEUED,
TASK_STATE_RUNNING,
TASK_STATE_PAUSED,
TASK_STATE_CANCELLED,
TASK_STATE_DONE,
TASK_STATE_FAILED
} task_state_t;
typedef struct {
uint64_t ram_bytes;
uint64_t gpu_bytes;
uint32_t cpu_weight;
uint32_t io_weight;
uint32_t batch_size;
uint32_t batch_max;
} task_estimate_t;
```
Les états réels sont `TASK_PENDING`, `TASK_RUNNING`, `TASK_PAUSED`,
`TASK_CANCELLED`, `TASK_FAILED` et `TASK_COMPLETED`. Une rupture de séquence est
une opération de réadmission, pas un état supplémentaire.
## API publique
| Fonction | Description |
|---|---|
| `task_create()` | Alloue et initialise une tâche avec son estimate |
| `task_destroy()` | Libère toutes les ressources de la tâche |
| `task_get_state()` | Retourne l'état courant (thread-safe en lecture) |
| `task_set_state()` | Met à jour l'état avec transitions validées |
| `task_get_estimate()` | Retourne l'estimation immuable des coûts |
| `task_advance_progress()` | Avance la progression d'un pas validé |
| `task_request_pause()` | Demande une pause coopérative |
| `task_request_cancel()` | Demande une annulation coopérative |
| `task_should_pause()` | Vérifie si la tâche doit se mettre en pause |
| `task_should_cancel()` | Vérifie si la tâche doit s'annuler |
| `lardon3d_task_create()` | Alloue une tâche et copie son estimation |
| `lardon3d_task_destroy()` | Annule, attend puis libère la tâche |
| `lardon3d_task_start()` | Exécute sous réservation active |
| `lardon3d_task_pause()` / `resume()` | Contrôle la pause coopérative |
| `lardon3d_task_request_cancel()` | Demande l'annulation coopérative |
| `lardon3d_task_checkpoint()` | Frontière coopérative en mémoire |
| `lardon3d_task_sequence_break()` | Libère puis renouvelle la réservation |
| `lardon3d_task_snapshot()` | Copie l'état d'observation runtime |
### API durable
| Fonction | Description |
|---|---|
| `lardon3d_task_durable_snapshot()` | Copie les champs durables sous mutex |
| `lardon3d_task_restore()` | Reconstruit une tâche sans état d'exécution vivant |
| `lardon3d_task_checkpoint_save()` | Publie atomiquement un snapshot v1 |
| `lardon3d_task_checkpoint_load()` | Lit et valide un checkpoint borné |
Après `rename`, un échec du `fsync` du répertoire retourne
`LARDON3D_TASK_CHECKPOINT_PUBLISHED_NOT_DURABLE` : la publication est visible,
mais sa durabilité après crash n'est pas confirmée.
## Invariants
1. **Estimation immuable** : une fois créée, l'estimation d'une tâche ne change
jamais. Elle est copiée en lecture seule lors de la réservation.
2. **Transitions d'état validées** : seules certaines transitions sont
autorisées (IDLE → QUEUED → RUNNING → DONE/FAILED).
3. **Pause et annulation coopératives** : le worker vérifie périodiquement
`task_should_pause()` et `task_should_cancel()`. Le callback ne force jamais
l'arrêt.
2. **Transitions d'état validées** : le cycle nominal est
`PENDING → RUNNING → COMPLETED/FAILED/CANCELLED`, avec pause coopérative.
3. **Pause et annulation coopératives** : le callback appelle périodiquement
`lardon3d_task_checkpoint()`. Aucun autre thread ne force son arrêt.
4. **Progression bornée** : la progression ne peut jamais dépasser la valeur
maximale définie par l'estimation.
5. **Callback unique** : chaque tâche possède un seul callback invoqué une
seule fois, quelle que soit l'issue.
5. **Reprise réadmise** : une tâche restaurée non terminale repasse par la file,
le scheduler et le Resource Governor avec une nouvelle réservation.
6. **Snapshot court** : seuls les champs durables sont copiés sous le mutex ;
la sérialisation et les I/O ont lieu après déverrouillage.
## Interactions
@ -75,11 +70,14 @@ typedef struct {
## Statut
**IMPLÉMENTÉ** — cycle de vie complet, pause et annulation coopératives.
**IMPLEMENTED** — cycle de vie, pause/annulation coopératives, séquences
adaptatives et fondation de checkpoints persistants isolés.
**NOT_YET_WIRED** — sauvegarde automatique et restauration par la file.
## Limites
- Aucune priorité interne : l'ordre est uniquement FIFO.
- Aucune persistance : les tâches disparaissent à l'arrêt du programme.
- Pas encore de reprise globale au démarrage du projet.
- Aucune dépendance inter-tâches (pas de DAG).
- La progression est linéaire : pas de séquençage adaptatif interne.
- Pas encore de références d'artefacts métier validés.

3
docs/roadmap/roadmap.md
View file

@ -28,7 +28,8 @@ Lardon3D suit une feuille de route ordonnée qui privilégie la stabilité et la
## Prochaines étapes décidées (NEXT)
### Phase 3 : Persistance
- 📋 Persistance des tâches et checkpoints
- ✅ Fondation versionnée des checkpoints de tâches
- 📋 Branchement des checkpoints aux pipelines et reprise globale
- 📋 Project Database v1 (SQLite)
- 📋 ScanSet et Image Catalog persistants

22
include/lardon3d/task.h
View file

@ -43,6 +43,19 @@ typedef struct {
unsigned int io_slots;
} Lardon3DTaskExecutionContract;
typedef struct {
uint64_t id;
char name[LARDON3D_TASK_NAME_CAPACITY];
Lardon3DResourceEstimate estimate;
unsigned int progress;
Lardon3DTaskState saved_state;
Lardon3DTaskState recovery_state;
char message[LARDON3D_TASK_MESSAGE_CAPACITY];
struct timespec started_at;
struct timespec finished_at;
unsigned int sequence_count;
} Lardon3DTaskDurableSnapshot;
Lardon3DTask *lardon3d_task_create(
const char *name,
const Lardon3DResourceEstimate *estimate,
@ -71,6 +84,15 @@ bool lardon3d_task_snapshot(
const Lardon3DTask *task,
Lardon3DTaskSnapshot *snapshot
);
bool lardon3d_task_durable_snapshot(
const Lardon3DTask *task,
Lardon3DTaskDurableSnapshot *snapshot
);
Lardon3DTask *lardon3d_task_restore(
const Lardon3DTaskDurableSnapshot *snapshot,
Lardon3DTaskCallback callback,
void *userdata
);
uint64_t lardon3d_task_id(const Lardon3DTask *task);
bool lardon3d_task_assign_id(Lardon3DTask *task, uint64_t id);
bool lardon3d_task_resource_estimate(

31
include/lardon3d/task_checkpoint.h Normal file
View file

@ -0,0 +1,31 @@
#ifndef LARDON3D_TASK_CHECKPOINT_H
#define LARDON3D_TASK_CHECKPOINT_H
#include <stdint.h>
#include <lardon3d/task.h>
enum {
LARDON3D_TASK_CHECKPOINT_VERSION = 1,
};
typedef enum {
LARDON3D_TASK_CHECKPOINT_OK = 0,
LARDON3D_TASK_CHECKPOINT_NOT_FOUND,
LARDON3D_TASK_CHECKPOINT_INVALID,
LARDON3D_TASK_CHECKPOINT_UNSUPPORTED_VERSION,
LARDON3D_TASK_CHECKPOINT_IO_ERROR,
LARDON3D_TASK_CHECKPOINT_PUBLISHED_NOT_DURABLE
} Lardon3DTaskCheckpointResult;
Lardon3DTaskCheckpointResult lardon3d_task_checkpoint_save(
const char *path,
const Lardon3DTaskDurableSnapshot *snapshot
);
Lardon3DTaskCheckpointResult lardon3d_task_checkpoint_load(
const char *path,
Lardon3DTaskDurableSnapshot *snapshot,
uint32_t *format_version
);
#endif

17
meson.build
View file

@ -35,6 +35,7 @@ executable(
'src/image_view.c',
'src/project.c',
'src/task.c',
'src/task_checkpoint.c',
'src/task_queue.c',
'src/resource_governor.c',
'src/resource_snapshot.c',
@ -109,6 +110,22 @@ task_test = executable(
test('task', task_test, timeout: 30)
task_checkpoint_test = executable(
'test-task-checkpoint',
sources: [
'tests/test_task_checkpoint.c',
'src/task.c',
'src/task_checkpoint.c',
'src/resource_governor.c',
'src/resource_snapshot.c',
],
c_args: ['-DLARDON3D_CHECKPOINT_TESTING'],
include_directories: include_directories('include'),
dependencies: [threads],
)
test('task-checkpoint', task_checkpoint_test, timeout: 30)
sequential_task_test = executable(
'test-sequential-task',
sources: [

88
src/task.c
View file

@ -2,6 +2,7 @@
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <lardon3d/task.h>
@ -35,6 +36,18 @@ is_terminal(Lardon3DTaskState state)
|| state == TASK_COMPLETED;
}
static bool
valid_state(Lardon3DTaskState state)
{
return state >= TASK_PENDING && state <= TASK_COMPLETED;
}
static Lardon3DTaskState
recovery_state(Lardon3DTaskState state)
{
return state == TASK_RUNNING || state == TASK_PAUSED ? TASK_PENDING : state;
}
static void
copy_text(char *destination, size_t capacity, const char *text)
{
@ -140,6 +153,7 @@ lardon3d_task_start(
};
task->has_contract = true;
(void)clock_gettime(CLOCK_REALTIME, &task->started_at);
task->finished_at = (struct timespec) {0};
if (task->cancel_requested) {
Lardon3DResourceReservation *reservation_copy = task->current_reservation;
task->current_reservation = NULL;
@ -558,6 +572,80 @@ lardon3d_task_snapshot(
return true;
}
bool
lardon3d_task_durable_snapshot(
const Lardon3DTask *task,
Lardon3DTaskDurableSnapshot *snapshot
)
{
if (!task || !snapshot) {
return false;
}
Lardon3DTask *mutable_task = (Lardon3DTask *)task;
(void)pthread_mutex_lock(&mutable_task->mutex);
*snapshot = (Lardon3DTaskDurableSnapshot) {
.id = task->id,
.estimate = task->estimate,
.progress = task->progress,
.saved_state = task->state,
.recovery_state = recovery_state(task->state),
.started_at = task->started_at,
.finished_at = task->finished_at,
.sequence_count = task->sequence_count,
};
copy_text(snapshot->name, sizeof(snapshot->name), task->name);
copy_text(snapshot->message, sizeof(snapshot->message), task->message);
(void)pthread_mutex_unlock(&mutable_task->mutex);
return true;
}
Lardon3DTask *
lardon3d_task_restore(
const Lardon3DTaskDurableSnapshot *snapshot,
Lardon3DTaskCallback callback,
void *userdata
)
{
if (!snapshot || snapshot->id == 0 || !snapshot->name[0] || !callback
|| memchr(snapshot->name, '\0', sizeof(snapshot->name)) == NULL
|| memchr(snapshot->message, '\0', sizeof(snapshot->message)) == NULL
|| snapshot->progress > 100 || !valid_state(snapshot->saved_state)
|| !valid_state(snapshot->recovery_state)
|| snapshot->recovery_state != recovery_state(snapshot->saved_state)
|| (snapshot->recovery_state == TASK_COMPLETED
&& snapshot->progress != 100)
|| snapshot->estimate.minimum_batch_size == 0
|| snapshot->estimate.maximum_batch_size
< snapshot->estimate.minimum_batch_size
|| snapshot->estimate.desired_cpu_threads == 0
|| snapshot->estimate.task_class < LARDON3D_RESOURCE_TASK_GENERAL
|| snapshot->estimate.task_class > LARDON3D_RESOURCE_TASK_MIXED
|| ((snapshot->estimate.gpu_memory_fixed_bytes != 0
|| snapshot->estimate.gpu_memory_bytes_per_item != 0)
&& snapshot->estimate.desired_gpu_slots == 0)) {
return NULL;
}
Lardon3DTask *task = lardon3d_task_create(
snapshot->name,
&snapshot->estimate,
callback,
userdata
);
if (!task) {
return NULL;
}
(void)pthread_mutex_lock(&task->mutex);
task->id = snapshot->id;
task->progress = snapshot->progress;
task->state = snapshot->recovery_state;
copy_text(task->message, sizeof(task->message), snapshot->message);
task->started_at = snapshot->started_at;
task->finished_at = snapshot->finished_at;
task->sequence_count = snapshot->sequence_count;
(void)pthread_mutex_unlock(&task->mutex);
return task;
}
uint64_t
lardon3d_task_id(const Lardon3DTask *task)
{

376
src/task_checkpoint.c Normal file
View file

@ -0,0 +1,376 @@
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>
#include <lardon3d/task_checkpoint.h>
enum {
CHECKPOINT_SIZE = 516,
CHECKPOINT_HEADER_SIZE = 20,
};
static const unsigned char checkpoint_magic[8] = {
'L', '3', 'D', 'T', 'A', 'S', 'K', '\0'
};
static void
put_u32(unsigned char *output, uint32_t value)
{
for (size_t index = 0; index < 4; ++index) {
output[index] = (unsigned char)(value >> (index * 8));
}
}
static void
put_u64(unsigned char *output, uint64_t value)
{
for (size_t index = 0; index < 8; ++index) {
output[index] = (unsigned char)(value >> (index * 8));
}
}
static uint32_t
get_u32(const unsigned char *input)
{
uint32_t value = 0;
for (size_t index = 0; index < 4; ++index) {
value |= (uint32_t)input[index] << (index * 8);
}
return value;
}
static uint64_t
get_u64(const unsigned char *input)
{
uint64_t value = 0;
for (size_t index = 0; index < 8; ++index) {
value |= (uint64_t)input[index] << (index * 8);
}
return value;
}
static uint32_t
payload_checksum(const unsigned char *data, size_t size)
{
uint32_t hash = UINT32_C(2166136261);
for (size_t index = 0; index < size; ++index) {
hash ^= data[index];
hash *= UINT32_C(16777619);
}
return hash;
}
static bool
valid_text(const char *text, size_t capacity)
{
return memchr(text, '\0', capacity) != NULL;
}
static bool
valid_snapshot(const Lardon3DTaskDurableSnapshot *snapshot)
{
return snapshot && snapshot->id != 0 && snapshot->name[0]
&& valid_text(snapshot->name, sizeof(snapshot->name))
&& valid_text(snapshot->message, sizeof(snapshot->message))
&& snapshot->progress <= 100
&& snapshot->saved_state >= TASK_PENDING
&& snapshot->saved_state <= TASK_COMPLETED
&& snapshot->recovery_state >= TASK_PENDING
&& snapshot->recovery_state <= TASK_COMPLETED
&& snapshot->recovery_state
== ((snapshot->saved_state == TASK_RUNNING
|| snapshot->saved_state == TASK_PAUSED)
? TASK_PENDING : snapshot->saved_state)
&& (snapshot->recovery_state != TASK_COMPLETED
|| snapshot->progress == 100)
&& snapshot->estimate.minimum_batch_size > 0
&& snapshot->estimate.maximum_batch_size
>= snapshot->estimate.minimum_batch_size
&& snapshot->estimate.desired_cpu_threads > 0
&& snapshot->estimate.task_class >= LARDON3D_RESOURCE_TASK_GENERAL
&& snapshot->estimate.task_class <= LARDON3D_RESOURCE_TASK_MIXED
&& ((snapshot->estimate.gpu_memory_fixed_bytes == 0
&& snapshot->estimate.gpu_memory_bytes_per_item == 0)
|| snapshot->estimate.desired_gpu_slots > 0)
&& snapshot->started_at.tv_nsec >= 0
&& snapshot->started_at.tv_nsec < 1000000000L
&& snapshot->started_at.tv_sec >= 0
&& snapshot->finished_at.tv_nsec >= 0
&& snapshot->finished_at.tv_nsec < 1000000000L
&& snapshot->finished_at.tv_sec >= 0;
}
static bool
decode_size(uint64_t value, size_t *decoded)
{
if (value > SIZE_MAX) {
return false;
}
*decoded = (size_t)value;
return true;
}
static uint64_t
maximum_time_value(void)
{
unsigned int bits = (unsigned int)(sizeof(time_t) * CHAR_BIT);
bool signed_time = (time_t)-1 < (time_t)0;
if (bits > 64 || !signed_time) {
return UINT64_MAX;
}
if (bits == 64) {
return UINT64_MAX >> 1;
}
return (UINT64_C(1) << (bits - 1)) - 1;
}
static bool
decode_time(uint64_t value, time_t *decoded)
{
if (value > maximum_time_value()) {
return false;
}
*decoded = (time_t)value;
return true;
}
static void
encode(unsigned char data[CHECKPOINT_SIZE], const Lardon3DTaskDurableSnapshot *s)
{
memset(data, 0, CHECKPOINT_SIZE);
memcpy(data, checkpoint_magic, sizeof(checkpoint_magic));
put_u32(data + 8, LARDON3D_TASK_CHECKPOINT_VERSION);
put_u32(data + 12, CHECKPOINT_SIZE);
size_t at = CHECKPOINT_HEADER_SIZE;
#define PUT64(value) do { put_u64(data + at, (uint64_t)(value)); at += 8; } while (0)
#define PUT32(value) do { put_u32(data + at, (uint32_t)(value)); at += 4; } while (0)
PUT64(s->id);
memcpy(data + at, s->name, sizeof(s->name)); at += sizeof(s->name);
PUT64(s->estimate.memory_fixed_bytes);
PUT64(s->estimate.gpu_memory_fixed_bytes);
PUT64(s->estimate.memory_bytes_per_item);
PUT64(s->estimate.gpu_memory_bytes_per_item);
PUT64(s->estimate.minimum_batch_size);
PUT64(s->estimate.maximum_batch_size);
PUT32(s->estimate.desired_cpu_threads);
PUT32(s->estimate.desired_gpu_slots);
PUT32(s->estimate.desired_io_slots);
PUT32(s->estimate.task_class);
PUT32(s->progress);
PUT32(s->saved_state);
PUT32(s->recovery_state);
memcpy(data + at, s->message, sizeof(s->message)); at += sizeof(s->message);
PUT64(s->started_at.tv_sec); PUT32(s->started_at.tv_nsec);
PUT64(s->finished_at.tv_sec); PUT32(s->finished_at.tv_nsec);
PUT32(s->sequence_count);
#undef PUT64
#undef PUT32
put_u32(
data + 16,
payload_checksum(data + CHECKPOINT_HEADER_SIZE,
CHECKPOINT_SIZE - CHECKPOINT_HEADER_SIZE)
);
}
static bool
decode(const unsigned char data[CHECKPOINT_SIZE], Lardon3DTaskDurableSnapshot *s)
{
*s = (Lardon3DTaskDurableSnapshot) {0};
size_t at = CHECKPOINT_HEADER_SIZE;
#define GET64(target) do { (target) = get_u64(data + at); at += 8; } while (0)
#define GET32(target) do { (target) = get_u32(data + at); at += 4; } while (0)
GET64(s->id);
memcpy(s->name, data + at, sizeof(s->name)); at += sizeof(s->name);
GET64(s->estimate.memory_fixed_bytes);
GET64(s->estimate.gpu_memory_fixed_bytes);
GET64(s->estimate.memory_bytes_per_item);
GET64(s->estimate.gpu_memory_bytes_per_item);
uint64_t size_value;
GET64(size_value);
if (!decode_size(size_value, &s->estimate.minimum_batch_size)) {
return false;
}
GET64(size_value);
if (!decode_size(size_value, &s->estimate.maximum_batch_size)) {
return false;
}
GET32(s->estimate.desired_cpu_threads);
GET32(s->estimate.desired_gpu_slots);
GET32(s->estimate.desired_io_slots);
uint32_t task_class;
GET32(task_class);
s->estimate.task_class = (Lardon3DResourceTaskClass)task_class;
GET32(s->progress);
uint32_t state;
GET32(state); s->saved_state = (Lardon3DTaskState)state;
GET32(state); s->recovery_state = (Lardon3DTaskState)state;
memcpy(s->message, data + at, sizeof(s->message)); at += sizeof(s->message);
uint64_t seconds;
uint32_t nanoseconds;
GET64(seconds);
if (!decode_time(seconds, &s->started_at.tv_sec)) {
return false;
}
GET32(nanoseconds); s->started_at.tv_nsec = (long)nanoseconds;
GET64(seconds);
if (!decode_time(seconds, &s->finished_at.tv_sec)) {
return false;
}
GET32(nanoseconds); s->finished_at.tv_nsec = (long)nanoseconds;
GET32(s->sequence_count);
#undef GET64
#undef GET32
return true;
}
static bool
write_all(int descriptor, const unsigned char *data, size_t size)
{
while (size > 0) {
ssize_t written = write(descriptor, data, size);
if (written < 0 && errno == EINTR) {
continue;
}
if (written <= 0) {
return false;
}
data += (size_t)written;
size -= (size_t)written;
}
return true;
}
static bool
sync_parent_directory(const char *path)
{
#ifdef LARDON3D_CHECKPOINT_TESTING
const char *forced_failure = getenv(
"LARDON3D_TEST_CHECKPOINT_SYNC_DIRECTORY_FAILURE"
);
if (forced_failure && strcmp(forced_failure, "1") == 0) {
errno = EIO;
return false;
}
#endif
char parent[4096];
int length = snprintf(parent, sizeof(parent), "%s", path);
if (length < 0 || (size_t)length >= sizeof(parent)) {
return false;
}
char *separator = strrchr(parent, '/');
if (separator) {
*separator = '\0';
if (!parent[0]) {
parent[0] = '/';
parent[1] = '\0';
}
} else {
(void)snprintf(parent, sizeof(parent), ".");
}
int descriptor = open(parent, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
if (descriptor < 0) {
return false;
}
bool synced = fsync(descriptor) == 0;
return close(descriptor) == 0 && synced;
}
Lardon3DTaskCheckpointResult
lardon3d_task_checkpoint_save(
const char *path,
const Lardon3DTaskDurableSnapshot *snapshot
)
{
if (!path || !path[0] || !valid_snapshot(snapshot)) {
return LARDON3D_TASK_CHECKPOINT_INVALID;
}
char temporary[4096];
int length = snprintf(temporary, sizeof(temporary), "%s.tmp.XXXXXX", path);
if (length < 0 || (size_t)length >= sizeof(temporary)) {
return LARDON3D_TASK_CHECKPOINT_INVALID;
}
unsigned char data[CHECKPOINT_SIZE];
encode(data, snapshot);
int descriptor = mkstemp(temporary);
if (descriptor < 0) {
return LARDON3D_TASK_CHECKPOINT_IO_ERROR;
}
bool ready_to_publish = write_all(descriptor, data, sizeof(data))
&& fsync(descriptor) == 0;
if (close(descriptor) != 0) {
ready_to_publish = false;
}
descriptor = -1;
if (!ready_to_publish || rename(temporary, path) != 0) {
int saved_errno = errno;
(void)unlink(temporary);
errno = saved_errno;
return LARDON3D_TASK_CHECKPOINT_IO_ERROR;
}
return sync_parent_directory(path) ? LARDON3D_TASK_CHECKPOINT_OK
: LARDON3D_TASK_CHECKPOINT_PUBLISHED_NOT_DURABLE;
}
Lardon3DTaskCheckpointResult
lardon3d_task_checkpoint_load(
const char *path,
Lardon3DTaskDurableSnapshot *snapshot,
uint32_t *format_version
)
{
if (!path || !path[0] || !snapshot) {
return LARDON3D_TASK_CHECKPOINT_INVALID;
}
int descriptor = open(path, O_RDONLY | O_CLOEXEC);
if (descriptor < 0) {
return errno == ENOENT ? LARDON3D_TASK_CHECKPOINT_NOT_FOUND
: LARDON3D_TASK_CHECKPOINT_IO_ERROR;
}
unsigned char data[CHECKPOINT_SIZE + 1];
size_t used = 0;
while (used < sizeof(data)) {
ssize_t amount = read(descriptor, data + used, sizeof(data) - used);
if (amount < 0 && errno == EINTR) {
continue;
}
if (amount < 0) {
(void)close(descriptor);
return LARDON3D_TASK_CHECKPOINT_IO_ERROR;
}
if (amount == 0) {
break;
}
used += (size_t)amount;
}
if (close(descriptor) != 0) {
return LARDON3D_TASK_CHECKPOINT_IO_ERROR;
}
if (used < CHECKPOINT_HEADER_SIZE
|| memcmp(data, checkpoint_magic, sizeof(checkpoint_magic)) != 0) {
return LARDON3D_TASK_CHECKPOINT_INVALID;
}
uint32_t version = get_u32(data + 8);
if (format_version) {
*format_version = version;
}
if (version != LARDON3D_TASK_CHECKPOINT_VERSION) {
return LARDON3D_TASK_CHECKPOINT_UNSUPPORTED_VERSION;
}
if (get_u32(data + 12) != CHECKPOINT_SIZE || used != CHECKPOINT_SIZE
|| get_u32(data + 16) != payload_checksum(
data + CHECKPOINT_HEADER_SIZE,
CHECKPOINT_SIZE - CHECKPOINT_HEADER_SIZE
)) {
return LARDON3D_TASK_CHECKPOINT_INVALID;
}
return decode(data, snapshot) && valid_snapshot(snapshot)
? LARDON3D_TASK_CHECKPOINT_OK
: LARDON3D_TASK_CHECKPOINT_INVALID;
}

430
tests/test_task_checkpoint.c Normal file
View file

@ -0,0 +1,430 @@
#include <fcntl.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>
#include <lardon3d/task_checkpoint.h>
#define CHECK(condition) \
do { \
if (!(condition)) { \
(void)fprintf(stderr, "Échec ligne %d : %s\n", __LINE__, #condition); \
return false; \
} \
} while (0)
enum {
TEST_CHECKPOINT_SIZE = 516,
TEST_CHECKPOINT_HEADER_SIZE = 20,
TEST_MINIMUM_BATCH_OFFSET = 188,
TEST_STARTED_SECONDS_OFFSET = 488,
};
static uint32_t
test_checksum(const unsigned char *data, size_t size)
{
uint32_t hash = UINT32_C(2166136261);
for (size_t index = 0; index < size; ++index) {
hash ^= data[index];
hash *= UINT32_C(16777619);
}
return hash;
}
static void
test_put_u32(unsigned char *output, uint32_t value)
{
for (size_t index = 0; index < 4; ++index) {
output[index] = (unsigned char)(value >> (index * 8));
}
}
static void
test_put_u64(unsigned char *output, uint64_t value)
{
for (size_t index = 0; index < 8; ++index) {
output[index] = (unsigned char)(value >> (index * 8));
}
}
static bool
rewrite_u64(const char *path, size_t offset, uint64_t value)
{
unsigned char data[TEST_CHECKPOINT_SIZE];
int descriptor = open(path, O_RDWR);
if (descriptor < 0
|| read(descriptor, data, sizeof(data)) != (ssize_t)sizeof(data)) {
if (descriptor >= 0) {
(void)close(descriptor);
}
return false;
}
test_put_u64(data + offset, value);
test_put_u32(
data + 16,
test_checksum(
data + TEST_CHECKPOINT_HEADER_SIZE,
TEST_CHECKPOINT_SIZE - TEST_CHECKPOINT_HEADER_SIZE
)
);
bool written = pwrite(descriptor, data, sizeof(data), 0)
== (ssize_t)sizeof(data);
return close(descriptor) == 0 && written;
}
static bool
unused_callback(Lardon3DTask *task, void *userdata)
{
(void)task;
(void)userdata;
return true;
}
typedef struct {
Lardon3DTaskDurableSnapshot snapshot;
bool captured;
} CaptureContext;
static bool
capture_running_callback(Lardon3DTask *task, void *userdata)
{
CaptureContext *context = userdata;
context->captured = lardon3d_task_durable_snapshot(task, &context->snapshot);
return context->captured;
}
typedef struct {
Lardon3DResourceGovernor *governor;
Lardon3DTaskDurableSnapshot snapshot;
bool captured;
} SequenceContext;
static bool
capture_sequence_callback(Lardon3DTask *task, void *userdata)
{
SequenceContext *context = userdata;
Lardon3DResourceReservation *reservation = NULL;
Lardon3DTaskExecutionContract contract;
if (!lardon3d_task_sequence_break(
task,
context->governor,
&reservation,
&contract
)) {
return false;
}
context->captured = lardon3d_task_durable_snapshot(task, &context->snapshot);
return context->captured;
}
static Lardon3DTaskDurableSnapshot
snapshot_for(Lardon3DTaskState state)
{
Lardon3DTaskDurableSnapshot snapshot = {
.id = 41,
.estimate = {
.minimum_batch_size = 1,
.maximum_batch_size = 8,
.desired_cpu_threads = 1,
.task_class = LARDON3D_RESOURCE_TASK_GENERAL,
},
.progress = state == TASK_COMPLETED ? 100 : 37,
.saved_state = state,
.recovery_state = state == TASK_RUNNING || state == TASK_PAUSED
? TASK_PENDING : state,
.started_at = {.tv_sec = 10, .tv_nsec = 20},
.finished_at = {.tv_sec = 30, .tv_nsec = 40},
.sequence_count = 3,
};
(void)snprintf(snapshot.name, sizeof(snapshot.name), "Tâche durable");
(void)snprintf(snapshot.message, sizeof(snapshot.message), "Frontière validée");
return snapshot;
}
static bool
write_bytes(const char *path, const void *data, size_t size)
{
int descriptor = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
if (descriptor < 0) {
return false;
}
ssize_t written = write(descriptor, data, size);
bool ok = written >= 0 && (size_t)written == size && close(descriptor) == 0;
if (!ok) {
(void)close(descriptor);
}
return ok;
}
static bool
check_restored_state(Lardon3DTaskState saved, Lardon3DTaskState expected)
{
Lardon3DTaskDurableSnapshot durable = snapshot_for(saved);
Lardon3DTask *task = lardon3d_task_restore(
&durable,
unused_callback,
NULL
);
CHECK(task);
Lardon3DTaskSnapshot runtime;
CHECK(lardon3d_task_snapshot(task, &runtime));
CHECK(runtime.state == expected);
CHECK(lardon3d_task_sequence_count(task) == durable.sequence_count);
Lardon3DTaskExecutionContract contract;
CHECK(!lardon3d_task_execution_contract(task, &contract));
lardon3d_task_destroy(task);
return true;
}
static bool
run_test(void)
{
char directory[] = "/tmp/lardon3d-checkpoint-XXXXXX";
CHECK(mkdtemp(directory));
char path[512];
char temporary[512];
char corrupt[512];
char uncertain_directory[512];
char uncertain_path[512];
CHECK(snprintf(path, sizeof(path), "%s/task.chk", directory) > 0);
CHECK(snprintf(temporary, sizeof(temporary), "%s/task.chk.tmp", directory) > 0);
CHECK(snprintf(corrupt, sizeof(corrupt), "%s/corrupt.chk", directory) > 0);
CHECK(snprintf(
uncertain_directory,
sizeof(uncertain_directory),
"%s/no-read",
directory
) > 0);
CHECK(snprintf(
uncertain_path,
sizeof(uncertain_path),
"%s/task.chk",
uncertain_directory
) > 0);
const Lardon3DResourceEstimate estimate = {
.minimum_batch_size = 1,
.maximum_batch_size = 4,
.desired_cpu_threads = 1,
.task_class = LARDON3D_RESOURCE_TASK_GENERAL,
};
Lardon3DTask *pending = lardon3d_task_create(
"Pending",
&estimate,
unused_callback,
NULL
);
CHECK(pending && lardon3d_task_assign_id(pending, 7));
Lardon3DTaskDurableSnapshot durable;
CHECK(lardon3d_task_durable_snapshot(pending, &durable));
CHECK(durable.saved_state == TASK_PENDING);
CHECK(durable.recovery_state == TASK_PENDING);
CHECK(durable.sequence_count == 0);
lardon3d_task_destroy(pending);
CHECK(lardon3d_task_checkpoint_save(path, &durable)
== LARDON3D_TASK_CHECKPOINT_OK);
Lardon3DTaskDurableSnapshot loaded;
uint32_t version = 0;
CHECK(lardon3d_task_checkpoint_load(path, &loaded, &version)
== LARDON3D_TASK_CHECKPOINT_OK);
CHECK(version == LARDON3D_TASK_CHECKPOINT_VERSION);
CHECK(loaded.id == durable.id && loaded.saved_state == TASK_PENDING);
durable = snapshot_for(TASK_COMPLETED);
durable.id = 99;
CHECK(lardon3d_task_checkpoint_save(path, &durable)
== LARDON3D_TASK_CHECKPOINT_OK);
CHECK(lardon3d_task_checkpoint_load(path, &loaded, NULL)
== LARDON3D_TASK_CHECKPOINT_OK);
CHECK(loaded.id == 99 && loaded.saved_state == TASK_COMPLETED);
CHECK(mkdir(uncertain_directory, 0700) == 0);
Lardon3DTaskDurableSnapshot previous = durable;
previous.id = 98;
CHECK(lardon3d_task_checkpoint_save(uncertain_path, &previous)
== LARDON3D_TASK_CHECKPOINT_OK);
durable.id = 100;
CHECK(setenv(
"LARDON3D_TEST_CHECKPOINT_SYNC_DIRECTORY_FAILURE",
"1",
1
) == 0);
Lardon3DTaskCheckpointResult uncertain = lardon3d_task_checkpoint_save(
uncertain_path,
&durable
);
CHECK(unsetenv("LARDON3D_TEST_CHECKPOINT_SYNC_DIRECTORY_FAILURE") == 0);
CHECK(uncertain == LARDON3D_TASK_CHECKPOINT_PUBLISHED_NOT_DURABLE);
CHECK(lardon3d_task_checkpoint_load(uncertain_path, &loaded, NULL)
== LARDON3D_TASK_CHECKPOINT_OK);
CHECK(loaded.id == 100);
unsigned char short_data[12] = {'L', '3', 'D', 'T', 'A', 'S', 'K', 0};
CHECK(write_bytes(corrupt, short_data, sizeof(short_data)));
CHECK(lardon3d_task_checkpoint_load(corrupt, &loaded, NULL)
== LARDON3D_TASK_CHECKPOINT_INVALID);
int descriptor = open(path, O_RDWR);
CHECK(descriptor >= 0);
unsigned char unknown_version[4] = {2, 0, 0, 0};
CHECK(pwrite(descriptor, unknown_version, sizeof(unknown_version), 8)
== (ssize_t)sizeof(unknown_version));
CHECK(close(descriptor) == 0);
CHECK(lardon3d_task_checkpoint_load(path, &loaded, &version)
== LARDON3D_TASK_CHECKPOINT_UNSUPPORTED_VERSION);
CHECK(version == 2);
durable = snapshot_for(TASK_PENDING);
CHECK(lardon3d_task_checkpoint_save(path, &durable)
== LARDON3D_TASK_CHECKPOINT_OK);
descriptor = open(path, O_RDWR);
CHECK(descriptor >= 0);
unsigned char changed = 'X';
CHECK(pwrite(descriptor, &changed, 1, 100) == 1);
CHECK(close(descriptor) == 0);
CHECK(lardon3d_task_checkpoint_load(path, &loaded, NULL)
== LARDON3D_TASK_CHECKPOINT_INVALID);
durable = snapshot_for(TASK_PENDING);
CHECK(lardon3d_task_checkpoint_save(path, &durable)
== LARDON3D_TASK_CHECKPOINT_OK);
CHECK(rewrite_u64(path, TEST_STARTED_SECONDS_OFFSET, UINT64_MAX));
CHECK(lardon3d_task_checkpoint_load(path, &loaded, NULL)
== LARDON3D_TASK_CHECKPOINT_INVALID);
if (sizeof(size_t) < sizeof(uint64_t)) {
CHECK(lardon3d_task_checkpoint_save(path, &durable)
== LARDON3D_TASK_CHECKPOINT_OK);
CHECK(rewrite_u64(path, TEST_MINIMUM_BATCH_OFFSET, UINT64_MAX));
CHECK(lardon3d_task_checkpoint_load(path, &loaded, NULL)
== LARDON3D_TASK_CHECKPOINT_INVALID);
}
const unsigned char incomplete[] = "incomplet";
CHECK(write_bytes(temporary, incomplete, sizeof(incomplete)));
CHECK(unlink(path) == 0);
CHECK(lardon3d_task_checkpoint_load(path, &loaded, NULL)
== LARDON3D_TASK_CHECKPOINT_NOT_FOUND);
durable = snapshot_for(TASK_PENDING);
durable.estimate.maximum_batch_size = 0;
CHECK(lardon3d_task_checkpoint_save(path, &durable)
== LARDON3D_TASK_CHECKPOINT_INVALID);
durable = snapshot_for(TASK_PENDING);
memset(durable.name, 'x', sizeof(durable.name));
CHECK(lardon3d_task_checkpoint_save(path, &durable)
== LARDON3D_TASK_CHECKPOINT_INVALID);
CHECK(check_restored_state(TASK_COMPLETED, TASK_COMPLETED));
CHECK(check_restored_state(TASK_FAILED, TASK_FAILED));
CHECK(check_restored_state(TASK_CANCELLED, TASK_CANCELLED));
CHECK(check_restored_state(TASK_RUNNING, TASK_PENDING));
CHECK(check_restored_state(TASK_PAUSED, TASK_PENDING));
Lardon3DHardwareProfile profile = {
.logical_cpu_count = 2,
.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 resources = {
.memory_available_bytes = UINT64_MAX,
.cpu_load_1m = 0.0,
};
Lardon3DResourceDecision decision;
Lardon3DResourceReservation *reservation = NULL;
CaptureContext running = {0};
Lardon3DTask *running_task = lardon3d_task_create(
"Running",
&estimate,
capture_running_callback,
&running
);
CHECK(governor && running_task && lardon3d_task_assign_id(running_task, 51));
CHECK(lardon3d_resource_governor_reserve(
governor, &resources, &estimate, &decision, &reservation
));
CHECK(lardon3d_task_start(running_task, governor, reservation));
CHECK(running.captured && running.snapshot.saved_state == TASK_RUNNING);
CHECK(running.snapshot.recovery_state == TASK_PENDING);
lardon3d_task_destroy(running_task);
Lardon3DTaskDurableSnapshot restarted_snapshot = snapshot_for(TASK_RUNNING);
restarted_snapshot.started_at = (struct timespec) {.tv_sec = 10, .tv_nsec = 20};
restarted_snapshot.finished_at = (struct timespec) {.tv_sec = 30, .tv_nsec = 40};
CaptureContext restarted = {0};
Lardon3DTask *restarted_task = lardon3d_task_restore(
&restarted_snapshot,
capture_running_callback,
&restarted
);
CHECK(restarted_task);
reservation = NULL;
CHECK(lardon3d_resource_governor_reserve(
governor, &resources, &estimate, &decision, &reservation
));
CHECK(lardon3d_task_start(restarted_task, governor, reservation));
CHECK(restarted.captured);
CHECK(restarted.snapshot.started_at.tv_sec != 10);
CHECK(restarted.snapshot.finished_at.tv_sec == 0);
CHECK(restarted.snapshot.finished_at.tv_nsec == 0);
lardon3d_task_destroy(restarted_task);
SequenceContext sequence = {.governor = governor};
Lardon3DTask *sequence_task = lardon3d_task_create(
"Sequence",
&estimate,
capture_sequence_callback,
&sequence
);
CHECK(sequence_task && lardon3d_task_assign_id(sequence_task, 52));
reservation = NULL;
CHECK(lardon3d_resource_governor_reserve(
governor, &resources, &estimate, &decision, &reservation
));
CHECK(lardon3d_task_start(sequence_task, governor, reservation));
CHECK(sequence.captured && sequence.snapshot.saved_state == TASK_RUNNING);
CHECK(sequence.snapshot.recovery_state == TASK_PENDING);
CHECK(sequence.snapshot.sequence_count == 1);
Lardon3DTask *restored_sequence = lardon3d_task_restore(
&sequence.snapshot,
unused_callback,
NULL
);
CHECK(restored_sequence);
CHECK(lardon3d_task_sequence_count(restored_sequence) == 1);
Lardon3DTaskExecutionContract restored_contract;
CHECK(!lardon3d_task_execution_contract(
restored_sequence,
&restored_contract
));
lardon3d_task_destroy(restored_sequence);
lardon3d_task_destroy(sequence_task);
lardon3d_resource_governor_destroy(governor);
CHECK(unlink(temporary) == 0);
CHECK(unlink(corrupt) == 0);
CHECK(unlink(uncertain_path) == 0);
CHECK(rmdir(uncertain_directory) == 0);
CHECK(rmdir(directory) == 0);
return true;
}
int
main(void)
{
return run_test() ? EXIT_SUCCESS : EXIT_FAILURE;
}