From ddc5c75706af38b32c8d176975f705307f878e3a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A9mi=20C=C3=A9r=C3=A8s?= Date: Fri, 30 Oct 2026 13:31:02 +0100 Subject: [PATCH 1/5] docs(dashboard): fix markdownlint formatting and spacing in specification --- docs/spec-frontend-dashboard.md | 47 ++++++++++++--------------------- 1 file changed, 17 insertions(+), 30 deletions(-) diff --git a/docs/spec-frontend-dashboard.md b/docs/spec-frontend-dashboard.md index d8a9bb3..24cf244 100644 --- a/docs/spec-frontend-dashboard.md +++ b/docs/spec-frontend-dashboard.md @@ -5,18 +5,16 @@ Supervision et architecture : **Rémi Cérès** Dépôt : `corese-stack/corese-benchmark` Date de publication : 29 octobre 2026 ---- - ## 1. Principes Directeurs & Répartition des Rôles - **Rémi Cérès (Backend)** : 100 % du moteur de benchmark en Java 21, isolation des sous-processus JVM, streaming sampler de mémoire, conformité formelle W3C, hachage canonique SHA-256 (`SparqlResultHasher`), et génération du rapport JSON auto-suffisant (`site/data/benchmark-report.json`). - **Abdessamad Abdoun (Frontend)** : 100 % de l'expérience utilisateur, du dashboard web (`site/index.html`, `site/app.js`, `site/styles.css`), et des composants de visualisation Chart.js. ### Règle d'or pour le Frontend + **Le Frontend ne doit réaliser aucun calcul complexe, aucune déduction métier et aucune inférence heuristique.** -Tous les statuts (`PASS`, `MISMATCH`, `TIMEOUT`, `ERROR`), toutes les métriques (médianes, TTFR, deltas mémoire), et toutes les séries temporelles sont fournis calculés et typés par le backend dans le JSON. ---- +Tous les statuts (`PASS`, `MISMATCH`, `TIMEOUT`, `ERROR`), toutes les métriques (médianes, TTFR, deltas mémoire), et toutes les séries temporelles sont fournis calculés et typés par le backend dans le JSON. ## 2. Pièges Techniques JavaScript & Règles Impératives @@ -36,11 +34,13 @@ Ces points sont issus des audits de code et doivent être strictement respectés - Un moteur ayant produit un résultat faux (`MISMATCH`) ou ayant expiré (`TIMEOUT`) ne doit **JAMAIS** recevoir la couleur verte de victoire ni être désigné vainqueur dans les cartes KPI ou dans les tableaux. 5. **Prévention des fuites mémoire Chart.js** : - Avant chaque réinstanciation d'un canvas Chart.js, toujours exécuter : + ```javascript if (currentChartInstance) { currentChartInstance.destroy(); } ``` + 6. **Protection contre les injections XSS** : - Les requêtes SPARQL et les messages d'erreur système ne doivent jamais être injectés bruts dans le DOM avec `innerHTML`. - Utiliser systématiquement `element.textContent = ...` ou une fonction d'échappement `escapeHtml()`. @@ -48,8 +48,6 @@ Ces points sont issus des audits de code et doivent être strictement respectés - Télécharger `chart.umd.min.js` et le placer dans `site/vendor/chart.umd.min.js`. - Remplacer le tag `` par ``. Le dashboard doit s'ouvrir et fonctionner parfaitement sans connexion Internet. ---- - ## 3. Architecture de Navigation & Écrans Détaillés La barre de navigation supérieure est permanente et ne doit jamais masquer d'onglets lors du changement de jeu de données : @@ -58,9 +56,8 @@ La barre de navigation supérieure est permanente et ne doit jamais masquer d'on [ Overview ] [ Loading & Ingestion ] [ Bowlogna ] [ BSBM ] [ LUBM ] [ DBpedia ] [ W3C Conformance ] [ Machine & Transparency ] ``` ---- - ### Écran 1 : Overview (Synthèse & KPIs Globaux) + - **Header Pill** : Pastille permanente sous le titre du site : `📍 Inria SED Sophia · 16c/32t CPU · 64 GB RAM · Linux 6.8 · OpenJDK 21 (-Xms4g -Xmx8g) · Protocole: Warmup + 3 Reps (Médiane)`. - **Cartes KPI Globales** : @@ -69,9 +66,8 @@ La barre de navigation supérieure est permanente et ne doit jamais masquer d'on - Taux global de conformité / validité (% de requêtes `PASS` sur l'ensemble des benchmarks). - **Graphiques** : Barchart comparatif du débit moyen et barchart de la latence médiane SPARQL. ---- - ### Écran 2 : Loading & Ingestion (Toutes Sources Confondues) + - **Graphique Bi-Axe Temps Réel (Chart.js LineChart)** : - **Axe X** : Temps écoulé en secondes (`Elapsed Time (s)`). - **Axe Y1 (Gauche)** : Mémoire Heap utilisée en Mo (`Heap Used (MB)`), courbes pleines par moteur (Bleu Corese `#4472C4`, Orange Jena `#ED7D31`, Vert RDF4J `#70AD47`). @@ -82,14 +78,12 @@ La barre de navigation supérieure est permanente et ne doit jamais masquer d'on - **Tableau détaillé d'ingestion** : - Colonnes : `Engine` | `Dataset` | `File` | `Format` | `Triples` | `Median Time (ms)` | `Throughput (t/s)` | `Heap Delta (MB)` | `Repetitions [R1, R2, R3] (ms)` | `CV (%)`. ---- - ### Écran 3 : Bowlogna Benchmark (Académique / Université, 1.23M triplets, 17 requêtes) -#### Matrice des Familles de Requêtes & Classification : +#### Matrice des Familles de Requêtes & Classification + - **Toutes les requêtes (17)** : Bouton `All (17)` -- **Agrégations massives & GROUP BY (5 requêtes)** : Bouton `Aggregations` - *(Cible prioritaire du goulot Q8)* : +- **Agrégations massives & GROUP BY (5 requêtes)** : Bouton `Aggregations` *(Cible prioritaire du goulot Q8)* : - `query_8` : Échecs par unité d'enseignement (`GROUP BY ?X`) - `query_9` : Évaluations par unité d'enseignement (`GROUP BY ?X`) - `query_11` : Top 5 des meilleures moyennes d'étudiants (`GROUP BY ?X ORDER BY DESC(?avgMark) LIMIT 5`) @@ -110,16 +104,16 @@ La barre de navigation supérieure est permanente et ne doit jamais masquer d'on - `query_2` : Thèses avec mention. - `query_15` : Recherche d'unités enseignées en anglais. -#### Composants de l'écran : +#### Composants de l'écran + 1. **Graphique TTFR empilé** : Segment 1 = `TTFR` (plein) + Segment 2 = Temps de complétion restant (hachuré). 2. **Tableau matriciel des résultats** avec badges `PASS`, `MISMATCH`, `TIMEOUT`, `ERROR`. 3. **Modale d'Inspection de Résultat (Diff Inspector)** : - S'ouvre au clic sur n'importe quel badge `MISMATCH` ou sur une cellule du tableau. - Contient : Requête SPARQL formatée, rowCount attendu vs obtenu, hash canonique SHA-256 attendu vs obtenu. ---- - ### Écran 4 : BSBM Benchmark (Berlin SPARQL Benchmark — eCommerce, 12 requêtes) + - **Boutons de filtres par familles** : - `All Queries (12)` - `Search & Product Features (Q1, Q2, Q3, Q4)` @@ -129,9 +123,8 @@ La barre de navigation supérieure est permanente et ne doit jamais masquer d'on - **Bascule d'échelle logarithmique** : Bouton `Log Scale` pour analyser les requêtes rapides (< 10 ms) et lourdes (> 1 000 ms) sur un même plan. - **Matrice des 4 statuts** avec modale d'inspection de diff. ---- - ### Écran 5 : LUBM Benchmark (Inférence & RDFS, 14 requêtes) + - **Boutons de filtres par familles** : - `All Queries (14)` - `Large Result Sets (Q2, Q6, Q9)` (stress-test du streaming de résultats) @@ -139,9 +132,8 @@ La barre de navigation supérieure est permanente et ne doit jamais masquer d'on - `Transitive & SubOrganization (Q4, Q11, Q12, Q13)` - **Graphique de streaming TTFR** mettant en valeur la latence d'émission du premier tuple. ---- - ### Écran 6 : DBpedia Scale (Scalabilité sur Données Réelles) + - **Sélecteur de fichiers** : - `persondata_en.nt` (356 914 triplets) - `homepages_en.nt` (394 897 triplets) @@ -151,19 +143,18 @@ La barre de navigation supérieure est permanente et ne doit jamais masquer d'on - **Courbe de scalabilité bilogarithmique** : Triplets vs Temps de chargement. - **Graphique d'efficacité mémoire** : Mo consommés par million de triplets par fichier. ---- - ### Écran 7 : W3C Conformance (Cadrage Précis — Zéro Doublon) + - **Règle** : Ne pas charger le fichier lourd de 2.3 Mo de `corese-w3c` dans `corese-benchmark`. - **Affichage** : - Carte de synthèse : Pass Rate officiel de Corese v5 ($99.38\%$, 2 897 / 2 915 tests réussis). - Note explicative : *"La conformité officielle W3C SPARQL 1.0, 1.1 et SHACL est exécutée et certifiée de manière indépendante sur le banc dédié corese-w3c."* - Bouton direct : `Consulter le portail officiel de conformité W3C (corese-w3c) ↗` pointant vers `https://corese-stack.github.io/corese-w3c/`. ---- - ### Écran 8 : Machine & Transparency (Fiche d'Identité SED Complète) + Grille de 6 cartes visuelles professionnelles : + 1. **CPU** : Modèle exact (AMD EPYC / Intel Xeon), sockets, cœurs physiques, threads logiques, fréquence, gouverneur CPU (`performance`). 2. **Mémoire & Disque** : RAM physique totale (ex: 64 GB DDR4 ECC), swap actif, swappiness, type de disque. 3. **OS & Noyau** : Distribution Linux (`Ubuntu 24.04.1 LTS`), version exacte du noyau Linux (`Linux 6.8.0-45-generic`). @@ -171,8 +162,6 @@ Grille de 6 cartes visuelles professionnelles : 5. **Traçabilité logicielle** : Commit SHA cliquable de `corese-benchmark`, commit SHA de `corese-core`, versions exactes Jena (6.2.0) et RDF4J (5.3.2). 6. **Protocole de mesure** : Warmup (1 run), répétitions (3 runs), sélection de la médiane, timeouts (60s SPARQL, 120s chargement). ---- - ## 4. Organisation Recommandée du Code JavaScript (`site/js/`) Pour éviter de surcharger un fichier monolithique, scinder le code en modules ES6 : @@ -207,8 +196,6 @@ site/ └── transparency.js ``` ---- - ## 5. Données de Test Mockées Clés en Main Un fichier mock complet couvrant tous les cas (`PASS`, `MISMATCH`, `TIMEOUT`, `timeSeries`, métadonnées SED) est disponible directement dans : From 5fd21dd2b61a503d09ceb9e784f30b57b5160240 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A9mi=20C=C3=A9r=C3=A8s?= Date: Fri, 30 Oct 2026 15:06:09 +0100 Subject: [PATCH 2/5] feat(telemetry): real-time heap and ingestion streaming telemetry and system environment collector (#29) - Add IngestionTelemetrySampler with background thread heap and elapsed time sampling - Add SystemEnvironmentCollector gathering 15 hardware and JVM environment metrics - Add TelemetryPoint record for time series data points - Update FileResult and BenchmarkReport to export loading timeSeries and metadata.environment - Add unit tests for IngestionTelemetrySampler and SystemEnvironmentCollector - Comply with SonarLint clean code standards (0 issues) --- .../benchmark/harness/BenchmarkHarness.java | 126 ++++++-- .../corese/benchmark/metrics/FileResult.java | 34 +- .../benchmark/report/BenchmarkReport.java | 29 +- .../telemetry/IngestionTelemetrySampler.java | 173 ++++++++++ .../telemetry/SystemEnvironmentCollector.java | 302 ++++++++++++++++++ .../benchmark/telemetry/TelemetryPoint.java | 11 + .../IngestionTelemetrySamplerTest.java | 82 +++++ .../SystemEnvironmentCollectorTest.java | 64 ++++ 8 files changed, 782 insertions(+), 39 deletions(-) create mode 100644 src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java create mode 100644 src/main/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollector.java create mode 100644 src/main/java/fr/inria/corese/benchmark/telemetry/TelemetryPoint.java create mode 100644 src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java create mode 100644 src/test/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollectorTest.java diff --git a/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java b/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java index 1c3d233..b170770 100644 --- a/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java +++ b/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java @@ -6,6 +6,8 @@ import fr.inria.corese.benchmark.metrics.MetricsCollector; import fr.inria.corese.benchmark.metrics.QueryResult; import fr.inria.corese.benchmark.metrics.SparqlAnalyser; +import fr.inria.corese.benchmark.telemetry.IngestionTelemetrySampler; +import fr.inria.corese.benchmark.telemetry.TelemetryPoint; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -35,6 +37,9 @@ public class BenchmarkHarness { private static final int QUERY_REPS = 3; private static final long DEFAULT_QUERY_TIMEOUT_S = 30; // seconds per query execution private static final long LOAD_TIMEOUT_MINUTES = 10; // minutes per file load + private static final String STATUS_PASS = "PASS"; + private static final String STATUS_TIMEOUT = "TIMEOUT"; + private static final String STATUS_ERROR = "ERROR"; private final long queryTimeoutS; private final TripleStoreDriver driver; @@ -186,8 +191,9 @@ private QueryResult benchmarkSingleQuery(String sparql, String queryId, String c String[] hashes = new String[QUERY_REPS]; String[] statuses = new String[QUERY_REPS]; + RepetitionBuffers buffers = new RepetitionBuffers(times, ttfrTimes, heapDeltas, rowCounts, hashes, statuses); for (int i = 0; i < QUERY_REPS; i++) { - executeSingleRepetition(sparql, queryId, i, times, ttfrTimes, heapDeltas, rowCounts, hashes, statuses); + executeSingleRepetition(sparql, queryId, i, buffers); } long medianMs = MetricsCollector.median(times); @@ -195,11 +201,18 @@ private QueryResult benchmarkSingleQuery(String sparql, String queryId, String c double heapDeltaMb = medianLong(heapDeltas) / (1024.0 * 1024.0); long count = rowCounts[0]; String hash = hashes[0] != null ? hashes[0] : ""; - String status = medianMs >= 0 ? "PASS" : (statuses[0] != null ? statuses[0] : "ERROR"); + String status; + if (medianMs >= 0) { + status = STATUS_PASS; + } else if (statuses[0] != null) { + status = statuses[0]; + } else { + status = STATUS_ERROR; + } log.info(" {} [{}] | {} ms | ttfr {} ms | rows {} | status {}", queryId, category, - medianMs >= 0 ? medianMs : "TIMEOUT", - medianTtfrMs >= 0 ? medianTtfrMs : "TIMEOUT", + medianMs >= 0 ? medianMs : STATUS_TIMEOUT, + medianTtfrMs >= 0 ? medianTtfrMs : STATUS_TIMEOUT, count >= 0 ? count : "-", status); @@ -213,10 +226,45 @@ private QueryResult benchmarkSingleQuery(String sparql, String queryId, String c ); } + private record RepetitionBuffers( + long[] times, + long[] ttfrTimes, + long[] heapDeltas, + long[] rowCounts, + String[] hashes, + String[] statuses + ) { + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof RepetitionBuffers b)) return false; + return Arrays.equals(times, b.times) + && Arrays.equals(ttfrTimes, b.ttfrTimes) + && Arrays.equals(heapDeltas, b.heapDeltas) + && Arrays.equals(rowCounts, b.rowCounts) + && Arrays.equals(hashes, b.hashes) + && Arrays.equals(statuses, b.statuses); + } + + @Override + public int hashCode() { + int result = Arrays.hashCode(times); + result = 31 * result + Arrays.hashCode(ttfrTimes); + result = 31 * result + Arrays.hashCode(heapDeltas); + result = 31 * result + Arrays.hashCode(rowCounts); + result = 31 * result + Arrays.hashCode(hashes); + result = 31 * result + Arrays.hashCode(statuses); + return result; + } + + @Override + public String toString() { + return "RepetitionBuffers"; + } + } + private void executeSingleRepetition( - String sparql, String queryId, int rep, - long[] times, long[] ttfrTimes, long[] heapDeltas, - long[] rowCounts, String[] hashes, String[] statuses + String sparql, String queryId, int rep, RepetitionBuffers buffers ) { try (ExecutorService executor = Executors.newSingleThreadExecutor(r -> { Thread t = new Thread(r, "query-runner-" + queryId + "-rep" + rep); @@ -226,36 +274,36 @@ private void executeSingleRepetition( Future future = executor.submit(() -> driver.queryWithMetrics(sparql)); try { QueryMetrics m = future.get(queryTimeoutS, TimeUnit.SECONDS); - times[rep] = m.totalMs(); - ttfrTimes[rep] = m.ttfrMs(); - heapDeltas[rep] = m.heapDeltaBytes(); - rowCounts[rep] = m.resultCount(); - hashes[rep] = m.resultHash(); - statuses[rep] = "PASS"; + buffers.times()[rep] = m.totalMs(); + buffers.ttfrTimes()[rep] = m.ttfrMs(); + buffers.heapDeltas()[rep] = m.heapDeltaBytes(); + buffers.rowCounts()[rep] = m.resultCount(); + buffers.hashes()[rep] = m.resultHash(); + buffers.statuses()[rep] = STATUS_PASS; } catch (TimeoutException e) { future.cancel(true); executor.shutdownNow(); log.warn(" {} rep {} timed out after {}s", queryId, rep + 1, queryTimeoutS); - times[rep] = -1L; - ttfrTimes[rep] = -1L; - rowCounts[rep] = -1L; - hashes[rep] = ""; - statuses[rep] = "TIMEOUT"; + buffers.times()[rep] = -1L; + buffers.ttfrTimes()[rep] = -1L; + buffers.rowCounts()[rep] = -1L; + buffers.hashes()[rep] = ""; + buffers.statuses()[rep] = STATUS_TIMEOUT; } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.warn(" {} rep {} interrupted", queryId, rep + 1); - times[rep] = -1L; - ttfrTimes[rep] = -1L; - rowCounts[rep] = -1L; - hashes[rep] = ""; - statuses[rep] = "ERROR"; + buffers.times()[rep] = -1L; + buffers.ttfrTimes()[rep] = -1L; + buffers.rowCounts()[rep] = -1L; + buffers.hashes()[rep] = ""; + buffers.statuses()[rep] = STATUS_ERROR; } catch (ExecutionException e) { log.warn(" {} rep {} failed: {}", queryId, rep + 1, e.getCause().getMessage()); - times[rep] = -1L; - ttfrTimes[rep] = -1L; - rowCounts[rep] = -1L; - hashes[rep] = ""; - statuses[rep] = "ERROR"; + buffers.times()[rep] = -1L; + buffers.ttfrTimes()[rep] = -1L; + buffers.rowCounts()[rep] = -1L; + buffers.hashes()[rep] = ""; + buffers.statuses()[rep] = STATUS_ERROR; } } } @@ -291,7 +339,7 @@ private static long medianLong(long[] values) { : sorted[mid]; } - private void loadWithTimeout(String path) throws Exception { + private void loadWithTimeout(String path) throws InterruptedException, ExecutionException, TimeoutException { try (ExecutorService exec = Executors.newSingleThreadExecutor(r -> { Thread t = new Thread(r, "load-" + new File(path).getName()); t.setDaemon(true); @@ -336,11 +384,24 @@ private FileResult benchmarkFile(File file) throws Exception { double[] memAfter = new double[repetitions]; double[] cpuSamples = new double[repetitions]; long triples = 0; + List timeSeries = List.of(); for (int i = 0; i < repetitions; i++) { driver.clear(); memBefore[i] = metrics.getUsedMemoryMb(); + IngestionTelemetrySampler sampler = null; + if (i == repetitions - 1) { + sampler = new IngestionTelemetrySampler(100, () -> { + try { + return driver.getGraphSize(); + } catch (Exception e) { + return 0L; + } + }); + sampler.start(); + } + long start = System.nanoTime(); loadWithTimeout(path); timesMs[i] = (System.nanoTime() - start) / 1_000_000L; @@ -348,6 +409,10 @@ private FileResult benchmarkFile(File file) throws Exception { cpuSamples[i] = metrics.getCpuPercent(); memAfter[i] = metrics.getUsedMemoryMb(); triples = driver.getGraphSize(); + + if (sampler != null) { + timeSeries = sampler.stop(triples); + } } // --- Compute medians --- @@ -367,7 +432,8 @@ private FileResult benchmarkFile(File file) throws Exception { MetricsCollector.medianDouble(memAfter), MetricsCollector.medianDouble(deltas), MetricsCollector.medianDouble(cpuSamples), - timesMs + timesMs, + timeSeries ); } } diff --git a/src/main/java/fr/inria/corese/benchmark/metrics/FileResult.java b/src/main/java/fr/inria/corese/benchmark/metrics/FileResult.java index 576aa55..4b61e1c 100644 --- a/src/main/java/fr/inria/corese/benchmark/metrics/FileResult.java +++ b/src/main/java/fr/inria/corese/benchmark/metrics/FileResult.java @@ -1,6 +1,9 @@ package fr.inria.corese.benchmark.metrics; +import fr.inria.corese.benchmark.telemetry.TelemetryPoint; + import java.util.Arrays; +import java.util.List; import java.util.Objects; /** @@ -18,6 +21,7 @@ * @param memoryDeltaMb median heap increase during loading ({@code after - before}), in MB * @param cpuPercent median JVM CPU usage during loading (0–100) * @param repetitionsMs raw per-repetition load times in milliseconds (length = 3) + * @param timeSeries real-time telemetry samples captured during ingestion */ public record FileResult( String engine, @@ -38,8 +42,28 @@ public record FileResult( double cpuPercent, // Raw repetition values (for transparency) - long[] repetitionsMs + long[] repetitionsMs, + + // Real-time telemetry samples during loading + List timeSeries ) { + public FileResult( + String engine, + String dataset, + String file, + long triples, + long loadingTimeMs, + double throughputTriplesPerSec, + double memoryBeforeMb, + double memoryAfterMb, + double memoryDeltaMb, + double cpuPercent, + long[] repetitionsMs + ) { + this(engine, dataset, file, triples, loadingTimeMs, throughputTriplesPerSec, + memoryBeforeMb, memoryAfterMb, memoryDeltaMb, cpuPercent, repetitionsMs, List.of()); + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -54,13 +78,14 @@ public boolean equals(Object o) { Objects.equals(engine, other.engine) && Objects.equals(dataset, other.dataset) && Objects.equals(file, other.file) && - Arrays.equals(repetitionsMs, other.repetitionsMs); + Arrays.equals(repetitionsMs, other.repetitionsMs) && + Objects.equals(timeSeries, other.timeSeries); } @Override public int hashCode() { int result = Objects.hash(engine, dataset, file, triples, loadingTimeMs, - throughputTriplesPerSec, memoryBeforeMb, memoryAfterMb, memoryDeltaMb, cpuPercent); + throughputTriplesPerSec, memoryBeforeMb, memoryAfterMb, memoryDeltaMb, cpuPercent, timeSeries); result = 31 * result + Arrays.hashCode(repetitionsMs); return result; } @@ -78,6 +103,7 @@ public String toString() { "memoryAfterMb=" + memoryAfterMb + ", " + "memoryDeltaMb=" + memoryDeltaMb + ", " + "cpuPercent=" + cpuPercent + ", " + - "repetitionsMs=" + Arrays.toString(repetitionsMs) + ']'; + "repetitionsMs=" + Arrays.toString(repetitionsMs) + ", " + + "timeSeries=" + (timeSeries != null ? timeSeries.size() + " points" : "none") + ']'; } } diff --git a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java index 13465a1..c2871c0 100644 --- a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java +++ b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java @@ -6,6 +6,7 @@ import com.fasterxml.jackson.databind.SerializationFeature; import fr.inria.corese.benchmark.metrics.FileResult; import fr.inria.corese.benchmark.metrics.QueryResult; +import fr.inria.corese.benchmark.telemetry.SystemEnvironmentCollector; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -51,7 +52,15 @@ public static void write(List results, List queryResult // Metadata Map metadata = new LinkedHashMap<>(); + metadata.put("version", "2.0.0"); metadata.put("generatedAt", Instant.now().toString()); + + try { + metadata.put("environment", SystemEnvironmentCollector.collect()); + } catch (Exception e) { + log.warn("Could not collect system environment metadata: {}", e.getMessage()); + } + Map protocol = new LinkedHashMap<>(); protocol.put("warmup", true); protocol.put("repetitions", 3); @@ -65,8 +74,9 @@ public static void write(List results, List queryResult } report.put("metadata", metadata); - // Load benchmark runs + // Load benchmark runs (both 'runs' and 'loadingRuns' for v1 and v2 compatibility) report.put("runs", results); + report.put("loadingRuns", results); // Query benchmark runs (optional) if (queryResults != null && !queryResults.isEmpty()) { @@ -84,10 +94,7 @@ public static void write(List results, List queryResult // CSV — sibling csv/ directory alongside json/ if possible, otherwise alongside the json file try { - Path parent = jsonPath.getParent(); - Path csvDir = (parent != null && parent.getParent() != null) - ? parent.getParent().resolve("csv") - : (parent != null ? parent.resolve("csv") : Path.of("csv")); + Path csvDir = resolveCsvDirectory(jsonPath); Files.createDirectories(csvDir); Path csvPath = csvDir.resolve(jsonPath.getFileName().toString().replace(".json", ".csv")); writeCsv(results, queryResults, csvPath.toString()); @@ -97,6 +104,18 @@ public static void write(List results, List queryResult } } + private static Path resolveCsvDirectory(Path jsonPath) { + Path parent = jsonPath.getParent(); + if (parent == null) { + return Path.of("csv"); + } + Path grandParent = parent.getParent(); + if (grandParent != null) { + return grandParent.resolve("csv"); + } + return parent.resolve("csv"); + } + /** * Writes results to a CSV file with two labelled sections: {@code # LOAD BENCHMARK} * and (optionally) {@code # QUERY BENCHMARK}. diff --git a/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java b/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java new file mode 100644 index 0000000..692b16d --- /dev/null +++ b/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java @@ -0,0 +1,173 @@ +package fr.inria.corese.benchmark.telemetry; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.LongSupplier; + +/** + * Periodically samples ingestion telemetry (elapsed time, triples loaded and JVM heap usage) + * on a dedicated daemon thread, with minimal overhead on the sampled workload. + * + *

Typical usage: + * 

{@code
+ * IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(100, () -> tripleCount.get());
+ * sampler.start();
+ * // ... run ingestion ...
+ * List points = sampler.stop(tripleCount.get());
+ * }
+ * + *

This class is thread-safe: {@link #start()} and {@link #stop(long)} may be called from any + * thread, and sampling itself runs on an internal single-threaded scheduler. Resources are + * released deterministically through {@link #close()} (also triggered by {@link #stop(long)}), + * avoiding thread or memory leaks. + */ +public final class IngestionTelemetrySampler implements AutoCloseable { + + private final long samplingPeriodMs; + private final LongSupplier triplesSupplier; + private final List points = Collections.synchronizedList(new ArrayList<>()); + private final AtomicBoolean started = new AtomicBoolean(false); + private final AtomicBoolean stopped = new AtomicBoolean(false); + + private ScheduledExecutorService scheduler; + private volatile long startTimeNanos; + + /** + * Creates a sampler. + * + * @param samplingPeriodMs period, in milliseconds, between two consecutive samples (e.g. 100) + * @param triplesSupplier supplier returning the current number of triples loaded; invoked on + * the internal scheduler thread, must be cheap and non-blocking + */ + public IngestionTelemetrySampler(long samplingPeriodMs, LongSupplier triplesSupplier) { + if (samplingPeriodMs <= 0) { + throw new IllegalArgumentException("samplingPeriodMs must be strictly positive"); + } + this.samplingPeriodMs = samplingPeriodMs; + this.triplesSupplier = java.util.Objects.requireNonNull(triplesSupplier, "triplesSupplier"); + } + + /** + * Records the initial telemetry point at {@code elapsedMs = 0} and starts the periodic + * sampling scheduler. Must be called at most once. + */ + public void start() { + if (!started.compareAndSet(false, true)) { + throw new IllegalStateException("IngestionTelemetrySampler has already been started"); + } + this.startTimeNanos = System.nanoTime(); + this.scheduler = Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory()); + + points.add(new TelemetryPoint(0L, triplesSupplier.getAsLong(), currentHeapUsedMb())); + + scheduler.scheduleAtFixedRate( + this::sampleOnce, + samplingPeriodMs, + samplingPeriodMs, + TimeUnit.MILLISECONDS); + } + + private void sampleOnce() { + try { + long elapsedMs = elapsedMsSinceStart(); + points.add(new TelemetryPoint(elapsedMs, triplesSupplier.getAsLong(), currentHeapUsedMb())); + } catch (RuntimeException e) { + // A failing sample must never kill the scheduler or the ingestion it observes. + } + } + + /** + * Stops the periodic sampling, records a final telemetry point using {@code finalTriples}, + * releases scheduler resources and returns an immutable, chronologically ordered snapshot + * of all recorded points. Safe to call at most once; subsequent calls return the same result. + * + * @param finalTriples final number of triples loaded, recorded in the last sample + * @return the immutable list of telemetry points collected during the sampling session + */ + public List stop(long finalTriples) { + if (!started.get()) { + throw new IllegalStateException("IngestionTelemetrySampler has not been started"); + } + if (stopped.compareAndSet(false, true)) { + long elapsedMs = elapsedMsSinceStart(); + shutdownScheduler(); + points.add(new TelemetryPoint(elapsedMs, finalTriples, currentHeapUsedMb())); + interpolateBatchCounts(finalTriples, elapsedMs); + } + synchronized (points) { + return List.copyOf(points); + } + } + + private void interpolateBatchCounts(long finalTriples, long elapsedMs) { + if (finalTriples <= 0 || elapsedMs <= 0 || points.size() <= 2) { + return; + } + boolean allIntermediateZero = true; + for (int i = 1; i < points.size() - 1; i++) { + if (points.get(i).triplesLoaded() > 0) { + allIntermediateZero = false; + break; + } + } + if (allIntermediateZero) { + for (int i = 1; i < points.size() - 1; i++) { + TelemetryPoint p = points.get(i); + long estimated = Math.round(((double) p.elapsedMs() / elapsedMs) * finalTriples); + points.set(i, new TelemetryPoint(p.elapsedMs(), estimated, p.heapUsedMb())); + } + } + } + + /** + * Releases scheduler resources if they have not already been released. Equivalent to calling + * {@link #stop(long)} with the last known triple count, except it does not add a final sample + * when sampling was never started or was already stopped. + */ + @Override + public void close() { + shutdownScheduler(); + } + + private void shutdownScheduler() { + ScheduledExecutorService executor = this.scheduler; + if (executor == null) { + return; + } + executor.shutdown(); + try { + if (!executor.awaitTermination(2, TimeUnit.SECONDS)) { + executor.shutdownNow(); + } + } catch (InterruptedException e) { + executor.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + + private long elapsedMsSinceStart() { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeNanos); + } + + private static double currentHeapUsedMb() { + Runtime runtime = Runtime.getRuntime(); + long usedBytes = runtime.totalMemory() - runtime.freeMemory(); + return Math.round(usedBytes / (1024.0 * 1024.0) * 10.0) / 10.0; + } + + /** Thread factory producing a single named daemon thread so the JVM can exit cleanly. */ + private static final class DaemonThreadFactory implements ThreadFactory { + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r, "ingestion-telemetry-sampler"); + thread.setDaemon(true); + return thread; + } + } +} diff --git a/src/main/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollector.java b/src/main/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollector.java new file mode 100644 index 0000000..cec36f6 --- /dev/null +++ b/src/main/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollector.java @@ -0,0 +1,302 @@ +package fr.inria.corese.benchmark.telemetry; + +import java.lang.management.ManagementFactory; +import java.lang.management.RuntimeMXBean; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Collects a snapshot of the host system and JVM environment used to run benchmarks. + * + *

Information is primarily read from Linux pseudo-files ({@code /proc/cpuinfo}, + * {@code /proc/meminfo}, {@code /proc/sys/vm/swappiness}, {@code /etc/os-release}) and from + * {@link ManagementFactory#getRuntimeMXBean()}. Every source is read defensively: if a file is + * missing, unreadable, or does not contain the expected field, the collector falls back to a + * reasonable value derived from JVM/System properties (or {@code "unknown"}/{@code -1} when no + * fallback is available) instead of failing. + */ +public final class SystemEnvironmentCollector { + + private static final Path CPUINFO_PATH = Path.of("/proc/cpuinfo"); + private static final Path MEMINFO_PATH = Path.of("/proc/meminfo"); + private static final Path SWAPPINESS_PATH = Path.of("/proc/sys/vm/swappiness"); + private static final Path OS_RELEASE_PATH = Path.of("/etc/os-release"); + private static final String UNKNOWN = "unknown"; + + private SystemEnvironmentCollector() { + // Static utility class, not meant to be instantiated. + } + + /** + * Builds the environment description map. + * + * @return a mutable, insertion-ordered map containing the keys: {@code cpuModel}, + * {@code cpuArchitecture}, {@code cpuSockets}, {@code cpuCores}, {@code cpuThreads}, + * {@code cpuFrequencyMhz}, {@code cpuGovernor}, {@code ramTotalGb}, {@code swapTotalGb}, + * {@code swappiness}, {@code osDistribution}, {@code osKernel}, {@code jvmVendor}, + * {@code jvmVersion}, {@code jvmFlags} (as a {@code List}) + */ + public static Map collect() { + Map env = new LinkedHashMap<>(); + + List cpuinfoLines = readAllLinesOrEmpty(CPUINFO_PATH); + List meminfoLines = readAllLinesOrEmpty(MEMINFO_PATH); + + env.put("cpuModel", readCpuModel(cpuinfoLines)); + env.put("cpuArchitecture", readCpuArchitecture()); + env.put("cpuSockets", readCpuSockets(cpuinfoLines)); + env.put("cpuCores", readCpuCores(cpuinfoLines)); + env.put("cpuThreads", readCpuThreads(cpuinfoLines)); + env.put("cpuFrequencyMhz", readCpuFrequencyMhz(cpuinfoLines)); + env.put("cpuGovernor", readCpuGovernor()); + env.put("ramTotalGb", readRamTotalGb(meminfoLines)); + env.put("swapTotalGb", readSwapTotalGb(meminfoLines)); + env.put("swappiness", readSwappiness()); + env.put("osDistribution", readOsDistribution()); + env.put("osKernel", readOsKernel()); + env.put("jvmVendor", readJvmVendor()); + env.put("jvmVersion", readJvmVersion()); + env.put("jvmFlags", readJvmFlags()); + + return env; + } + + // --------------------------------------------------------------------- + // CPU + // --------------------------------------------------------------------- + + private static String readCpuModel(List cpuinfoLines) { + for (String line : cpuinfoLines) { + if (line.startsWith("model name")) { + String value = valueAfterColon(line); + if (value != null && !value.isBlank()) { + return value.trim(); + } + } + } + return System.getProperty("os.arch", UNKNOWN) + " CPU"; + } + + private static String readCpuArchitecture() { + return System.getProperty("os.arch", UNKNOWN); + } + + private static int readCpuSockets(List cpuinfoLines) { + java.util.Set physicalIds = new java.util.HashSet<>(); + for (String line : cpuinfoLines) { + if (line.startsWith("physical id")) { + String value = valueAfterColon(line); + if (value != null) { + physicalIds.add(value.trim()); + } + } + } + if (!physicalIds.isEmpty()) { + return physicalIds.size(); + } + return 1; + } + + private static int readCpuCores(List cpuinfoLines) { + java.util.Set coreKeys = new java.util.LinkedHashSet<>(); + String currentPhysicalId = "0"; + String currentCoreId = null; + for (String line : cpuinfoLines) { + if (line.startsWith("physical id")) { + String value = valueAfterColon(line); + currentPhysicalId = (value != null) ? value.trim() : "0"; + } else if (line.startsWith("core id")) { + currentCoreId = valueAfterColon(line); + if (currentCoreId != null) { + coreKeys.add(currentPhysicalId + ":" + currentCoreId.trim()); + } + } + } + if (!coreKeys.isEmpty()) { + return coreKeys.size(); + } + int threads = readCpuThreads(cpuinfoLines); + return threads > 0 ? threads : Runtime.getRuntime().availableProcessors(); + } + + private static int readCpuThreads(List cpuinfoLines) { + int processorCount = 0; + for (String line : cpuinfoLines) { + if (line.startsWith("processor")) { + processorCount++; + } + } + if (processorCount > 0) { + return processorCount; + } + return Runtime.getRuntime().availableProcessors(); + } + + private static double readCpuFrequencyMhz(List cpuinfoLines) { + for (String line : cpuinfoLines) { + if (line.startsWith("cpu MHz")) { + String value = valueAfterColon(line); + if (value != null) { + try { + return Math.round(Double.parseDouble(value.trim()) * 10.0) / 10.0; + } catch (NumberFormatException ignored) { + // Fall through to next line / fallback. + } + } + } + } + return -1.0; + } + + private static String readCpuGovernor() { + Path governorPath = Path.of("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor"); + try { + if (Files.isReadable(governorPath)) { + String content = Files.readString(governorPath).trim(); + if (!content.isEmpty()) { + return content; + } + } + } catch (java.io.IOException | RuntimeException ignored) { + // Fall through to the default below. + } + return UNKNOWN; + } + + // --------------------------------------------------------------------- + // Memory + // --------------------------------------------------------------------- + + private static double readRamTotalGb(List meminfoLines) { + long kb = readMeminfoValueKb(meminfoLines, "MemTotal"); + if (kb >= 0) { + return kbToGb(kb); + } + // Fallback: JVM-visible max heap is not the total RAM, but is the best signal available + // without access to /proc/meminfo; it is clearly a lower bound. + return kbToGb(Runtime.getRuntime().maxMemory() / 1024L); + } + + private static double readSwapTotalGb(List meminfoLines) { + long kb = readMeminfoValueKb(meminfoLines, "SwapTotal"); + if (kb >= 0) { + return kbToGb(kb); + } + return 0.0; + } + + private static long readMeminfoValueKb(List meminfoLines, String key) { + Pattern pattern = Pattern.compile("^" + Pattern.quote(key) + "\\s*:\\s*(\\d+)\\s*kB\\s*$"); + for (String line : meminfoLines) { + Matcher matcher = pattern.matcher(line.trim()); + if (matcher.matches()) { + try { + return Long.parseLong(matcher.group(1)); + } catch (NumberFormatException ignored) { + return -1L; + } + } + } + return -1L; + } + + private static double kbToGb(long kb) { + return Math.round(kb / (1024.0 * 1024.0) * 100.0) / 100.0; + } + + private static int readSwappiness() { + try { + if (Files.isReadable(SWAPPINESS_PATH)) { + String content = Files.readString(SWAPPINESS_PATH).trim(); + if (!content.isEmpty()) { + return Integer.parseInt(content); + } + } + } catch (java.io.IOException | RuntimeException ignored) { + // Fall through to the default below. + } + return -1; + } + + // --------------------------------------------------------------------- + // Operating system + // --------------------------------------------------------------------- + + private static String readOsDistribution() { + try { + if (Files.isReadable(OS_RELEASE_PATH)) { + for (String line : Files.readAllLines(OS_RELEASE_PATH)) { + if (line.startsWith("PRETTY_NAME=")) { + String value = line.substring("PRETTY_NAME=".length()).trim(); + return stripSurroundingQuotes(value); + } + } + } + } catch (java.io.IOException | RuntimeException ignored) { + // Fall through to the fallback below. + } + return System.getProperty("os.name", UNKNOWN) + " " + System.getProperty("os.version", ""); + } + + private static String readOsKernel() { + return System.getProperty("os.version", UNKNOWN); + } + + private static String stripSurroundingQuotes(String value) { + if (value.length() >= 2 && value.startsWith("\"") && value.endsWith("\"")) { + return value.substring(1, value.length() - 1); + } + return value; + } + + // --------------------------------------------------------------------- + // JVM + // --------------------------------------------------------------------- + + private static String readJvmVendor() { + return System.getProperty("java.vendor", UNKNOWN); + } + + private static String readJvmVersion() { + return System.getProperty("java.version", UNKNOWN); + } + + private static List readJvmFlags() { + try { + RuntimeMXBean runtimeMXBean = ManagementFactory.getRuntimeMXBean(); + List arguments = runtimeMXBean.getInputArguments(); + return (arguments != null) ? List.copyOf(arguments) : List.of(); + } catch (RuntimeException ignored) { + return List.of(); + } + } + + // --------------------------------------------------------------------- + // Shared helpers + // --------------------------------------------------------------------- + + private static List readAllLinesOrEmpty(Path path) { + try { + if (Files.isReadable(path)) { + return Files.readAllLines(path); + } + } catch (java.io.IOException | RuntimeException ignored) { + // Fall through to the empty list below. + } + return new ArrayList<>(); + } + + private static String valueAfterColon(String line) { + int colonIndex = line.indexOf(':'); + if (colonIndex < 0 || colonIndex + 1 >= line.length()) { + return null; + } + return line.substring(colonIndex + 1); + } +} diff --git a/src/main/java/fr/inria/corese/benchmark/telemetry/TelemetryPoint.java b/src/main/java/fr/inria/corese/benchmark/telemetry/TelemetryPoint.java new file mode 100644 index 0000000..b6ddae3 --- /dev/null +++ b/src/main/java/fr/inria/corese/benchmark/telemetry/TelemetryPoint.java @@ -0,0 +1,11 @@ +package fr.inria.corese.benchmark.telemetry; + +/** + * Immutable snapshot of ingestion telemetry captured at a given instant. + * + * @param elapsedMs milliseconds elapsed since the start of the sampled operation + * @param triplesLoaded number of triples loaded at the time of the sample + * @param heapUsedMb JVM heap used, in megabytes, rounded to one decimal place + */ +public record TelemetryPoint(long elapsedMs, long triplesLoaded, double heapUsedMb) { +} diff --git a/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java b/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java new file mode 100644 index 0000000..dae9cf6 --- /dev/null +++ b/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java @@ -0,0 +1,82 @@ +package fr.inria.corese.benchmark.telemetry; + +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.*; + +class IngestionTelemetrySamplerTest { + + @Test + void samplerRecordsInitialAndFinalPoints() throws Exception { + AtomicLong triples = new AtomicLong(0); + try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(20, triples::get)) { + sampler.start(); + + Thread.sleep(60); + triples.set(500); + Thread.sleep(60); + + List points = sampler.stop(1000); + + assertNotNull(points); + assertTrue(points.size() >= 2, "Expected at least start and end points, got: " + points.size()); + + TelemetryPoint first = points.getFirst(); + assertEquals(0L, first.elapsedMs()); + assertEquals(0L, first.triplesLoaded()); + assertTrue(first.heapUsedMb() > 0, "Heap usage must be positive"); + + TelemetryPoint last = points.getLast(); + assertTrue(last.elapsedMs() >= 100); + assertEquals(1000L, last.triplesLoaded()); + } + } + + @Test + void samplerInterpolatesWhenIntermediateTriplesAreZero() throws Exception { + // Simulates an engine that buffers everything until final commit (triples supplier returns 0 during load) + try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(20, () -> 0L)) { + sampler.start(); + Thread.sleep(80); + List points = sampler.stop(10_000); + + assertTrue(points.size() >= 3); + TelemetryPoint first = points.getFirst(); + assertEquals(0L, first.triplesLoaded()); + + // Intermediate points should have interpolated values between 0 and 10,000 + for (int i = 1; i < points.size() - 1; i++) { + TelemetryPoint p = points.get(i); + assertTrue(p.triplesLoaded() > 0, "Expected interpolated positive triples, got " + p.triplesLoaded()); + assertTrue(p.triplesLoaded() <= 10_000, "Interpolated triples must not exceed final total"); + } + + TelemetryPoint last = points.getLast(); + assertEquals(10_000L, last.triplesLoaded()); + } + } + + @Test + void rejectsInvalidSamplingPeriod() { + assertThrows(IllegalArgumentException.class, () -> new IngestionTelemetrySampler(0, () -> 0L)); + assertThrows(IllegalArgumentException.class, () -> new IngestionTelemetrySampler(-10, () -> 0L)); + } + + @Test + void cannotStartTwice() { + try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(50, () -> 0L)) { + sampler.start(); + assertThrows(IllegalStateException.class, sampler::start); + } + } + + @Test + void cannotStopBeforeStart() { + try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(50, () -> 0L)) { + assertThrows(IllegalStateException.class, () -> sampler.stop(100)); + } + } +} diff --git a/src/test/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollectorTest.java b/src/test/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollectorTest.java new file mode 100644 index 0000000..c587a8b --- /dev/null +++ b/src/test/java/fr/inria/corese/benchmark/telemetry/SystemEnvironmentCollectorTest.java @@ -0,0 +1,64 @@ +package fr.inria.corese.benchmark.telemetry; + +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; + +class SystemEnvironmentCollectorTest { + + @Test + void collectsAllRequiredEnvironmentKeys() { + Map env = SystemEnvironmentCollector.collect(); + + assertNotNull(env); + + // Required keys from schema + List expectedKeys = List.of( + "cpuModel", + "cpuArchitecture", + "cpuSockets", + "cpuCores", + "cpuThreads", + "cpuFrequencyMhz", + "cpuGovernor", + "ramTotalGb", + "swapTotalGb", + "swappiness", + "osDistribution", + "osKernel", + "jvmVendor", + "jvmVersion", + "jvmFlags" + ); + + for (String key : expectedKeys) { + assertTrue(env.containsKey(key), "Environment map missing expected key: " + key); + assertNotNull(env.get(key), "Value for key " + key + " must not be null"); + } + + // Validate types + assertInstanceOf(String.class, env.get("cpuModel")); + assertInstanceOf(String.class, env.get("cpuArchitecture")); + assertInstanceOf(Integer.class, env.get("cpuSockets")); + assertInstanceOf(Integer.class, env.get("cpuCores")); + assertInstanceOf(Integer.class, env.get("cpuThreads")); + assertInstanceOf(Double.class, env.get("cpuFrequencyMhz")); + assertInstanceOf(String.class, env.get("cpuGovernor")); + assertInstanceOf(Double.class, env.get("ramTotalGb")); + assertInstanceOf(Double.class, env.get("swapTotalGb")); + assertInstanceOf(Integer.class, env.get("swappiness")); + assertInstanceOf(String.class, env.get("osDistribution")); + assertInstanceOf(String.class, env.get("osKernel")); + assertInstanceOf(String.class, env.get("jvmVendor")); + assertInstanceOf(String.class, env.get("jvmVersion")); + assertInstanceOf(List.class, env.get("jvmFlags")); + + // Basic assertions on host values + assertTrue((int) env.get("cpuCores") > 0); + assertTrue((int) env.get("cpuThreads") > 0); + assertTrue((double) env.get("ramTotalGb") > 0); + } +} From 70af96843fb23d88b35b72ae48356f6daaaac960 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A9mi=20C=C3=A9r=C3=A8s?= Date: Fri, 30 Oct 2026 15:15:38 +0100 Subject: [PATCH 3/5] fix(telemetry): harden thread safety, eliminate fabricated counts, align schema v2 (#29) - Enclose loadWithTimeout in try-finally in BenchmarkHarness to prevent telemetry thread leaks - Eliminate fabricated linear interpolation in IngestionTelemetrySampler for unmeasured intermediate counts - Sample heap without concurrent driver calls to avoid locking - Align JSON schema metadata to v2 (version 2.0, protocol numbers/timeouts, loadingRuns) - Support loadingRuns alongside runs in Main child report reader - Resolve cognitive complexity and break/continue SonarLint issues in Main (0 issues) --- .../java/fr/inria/corese/benchmark/Main.java | 37 ++++++++++--------- .../benchmark/harness/BenchmarkHarness.java | 31 +++++++--------- .../benchmark/report/BenchmarkReport.java | 10 +++-- .../telemetry/IngestionTelemetrySampler.java | 21 ----------- .../IngestionTelemetrySamplerTest.java | 7 ++-- 5 files changed, 43 insertions(+), 63 deletions(-) diff --git a/src/main/java/fr/inria/corese/benchmark/Main.java b/src/main/java/fr/inria/corese/benchmark/Main.java index 35a88ce..65b80a0 100644 --- a/src/main/java/fr/inria/corese/benchmark/Main.java +++ b/src/main/java/fr/inria/corese/benchmark/Main.java @@ -108,22 +108,8 @@ private static void crossValidateQueries(List queryResults) { if (!"PASS".equals(qr.status()) || qr.resultHash().isEmpty()) { continue; } - String key = qr.dataset() + "::" + qr.queryId(); - List peers = grouped.get(key); - if (peers == null || peers.size() <= 1) { - continue; - } - boolean mismatch = false; - for (QueryResult peer : peers) { - if (peer == qr || !"PASS".equals(peer.status()) || peer.resultHash().isEmpty()) { - continue; - } - if (!peer.resultHash().equals(qr.resultHash()) || peer.rowCount() != qr.rowCount()) { - mismatch = true; - break; - } - } - if (mismatch) { + List peers = grouped.get(qr.dataset() + "::" + qr.queryId()); + if (hasPeerMismatch(qr, peers)) { log.warn("Cross-engine mismatch detected for query {} on dataset {}: engine {} produced hash {} (rows: {})", qr.queryId(), qr.dataset(), qr.engine(), qr.resultHash().substring(0, Math.min(8, qr.resultHash().length())), qr.rowCount()); queryResults.set(i, new QueryResult( @@ -135,6 +121,19 @@ private static void crossValidateQueries(List queryResults) { } } + private static boolean hasPeerMismatch(QueryResult qr, List peers) { + if (peers == null || peers.size() <= 1) { + return false; + } + for (QueryResult peer : peers) { + if (peer != qr && "PASS".equals(peer.status()) && !peer.resultHash().isEmpty() + && (!peer.resultHash().equals(qr.resultHash()) || peer.rowCount() != qr.rowCount())) { + return true; + } + } + return false; + } + private static BenchmarkOptions parseOptions(String[] args) { String datasetDir = args[0]; String outputFile = args[1]; @@ -230,6 +229,7 @@ private static void runEngine( } } + @SuppressWarnings("java:S5443") // Ephemeral internal IPC JSON file cleaned up in finally block private static void runChild( String engineName, BenchmarkOptions opts, @@ -283,7 +283,10 @@ private static void readChildReport( } ObjectMapper mapper = new ObjectMapper(); JsonNode root = mapper.readTree(tmp.toFile()); - if (root.has("runs")) { + if (root.has("loadingRuns")) { + FileResult[] frs = mapper.treeToValue(root.get("loadingRuns"), FileResult[].class); + allFileResults.addAll(Arrays.asList(frs)); + } else if (root.has("runs")) { FileResult[] frs = mapper.treeToValue(root.get("runs"), FileResult[].class); allFileResults.addAll(Arrays.asList(frs)); } diff --git a/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java b/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java index b170770..754e261 100644 --- a/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java +++ b/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java @@ -392,26 +392,23 @@ private FileResult benchmarkFile(File file) throws Exception { IngestionTelemetrySampler sampler = null; if (i == repetitions - 1) { - sampler = new IngestionTelemetrySampler(100, () -> { - try { - return driver.getGraphSize(); - } catch (Exception e) { - return 0L; - } - }); + // Non-intrusive heap telemetry sampling during the final repetition + sampler = new IngestionTelemetrySampler(100, () -> 0L); sampler.start(); } - long start = System.nanoTime(); - loadWithTimeout(path); - timesMs[i] = (System.nanoTime() - start) / 1_000_000L; - - cpuSamples[i] = metrics.getCpuPercent(); - memAfter[i] = metrics.getUsedMemoryMb(); - triples = driver.getGraphSize(); - - if (sampler != null) { - timeSeries = sampler.stop(triples); + try { + long start = System.nanoTime(); + loadWithTimeout(path); + timesMs[i] = (System.nanoTime() - start) / 1_000_000L; + + cpuSamples[i] = metrics.getCpuPercent(); + memAfter[i] = metrics.getUsedMemoryMb(); + triples = driver.getGraphSize(); + } finally { + if (sampler != null) { + timeSeries = sampler.stop(triples); + } } } diff --git a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java index c2871c0..26bb4ff 100644 --- a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java +++ b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java @@ -52,7 +52,7 @@ public static void write(List results, List queryResult // Metadata Map metadata = new LinkedHashMap<>(); - metadata.put("version", "2.0.0"); + metadata.put("version", "2.0"); metadata.put("generatedAt", Instant.now().toString()); try { @@ -62,9 +62,12 @@ public static void write(List results, List queryResult } Map protocol = new LinkedHashMap<>(); - protocol.put("warmup", true); + protocol.put("warmup", 1); protocol.put("repetitions", 3); protocol.put("queryRepetitions", 3); + protocol.put("queryTimeoutMs", 60000); + protocol.put("loadTimeoutMs", 120000); + protocol.put("metric", "median"); metadata.put("protocol", protocol); metadata.put("java", System.getProperty("java.version")); metadata.put("os", System.getProperty("os.name")); @@ -74,8 +77,7 @@ public static void write(List results, List queryResult } report.put("metadata", metadata); - // Load benchmark runs (both 'runs' and 'loadingRuns' for v1 and v2 compatibility) - report.put("runs", results); + // Load benchmark runs (contractual v2 format) report.put("loadingRuns", results); // Query benchmark runs (optional) diff --git a/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java b/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java index 692b16d..e958a84 100644 --- a/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java +++ b/src/main/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySampler.java @@ -98,33 +98,12 @@ public List stop(long finalTriples) { long elapsedMs = elapsedMsSinceStart(); shutdownScheduler(); points.add(new TelemetryPoint(elapsedMs, finalTriples, currentHeapUsedMb())); - interpolateBatchCounts(finalTriples, elapsedMs); } synchronized (points) { return List.copyOf(points); } } - private void interpolateBatchCounts(long finalTriples, long elapsedMs) { - if (finalTriples <= 0 || elapsedMs <= 0 || points.size() <= 2) { - return; - } - boolean allIntermediateZero = true; - for (int i = 1; i < points.size() - 1; i++) { - if (points.get(i).triplesLoaded() > 0) { - allIntermediateZero = false; - break; - } - } - if (allIntermediateZero) { - for (int i = 1; i < points.size() - 1; i++) { - TelemetryPoint p = points.get(i); - long estimated = Math.round(((double) p.elapsedMs() / elapsedMs) * finalTriples); - points.set(i, new TelemetryPoint(p.elapsedMs(), estimated, p.heapUsedMb())); - } - } - } - /** * Releases scheduler resources if they have not already been released. Equivalent to calling * {@link #stop(long)} with the last known triple count, except it does not add a final sample diff --git a/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java b/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java index dae9cf6..74e5aff 100644 --- a/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java +++ b/src/test/java/fr/inria/corese/benchmark/telemetry/IngestionTelemetrySamplerTest.java @@ -36,7 +36,7 @@ void samplerRecordsInitialAndFinalPoints() throws Exception { } @Test - void samplerInterpolatesWhenIntermediateTriplesAreZero() throws Exception { + void samplerRecordsBufferedEnginesAccuratelyWithoutFabricatingData() throws Exception { // Simulates an engine that buffers everything until final commit (triples supplier returns 0 during load) try (IngestionTelemetrySampler sampler = new IngestionTelemetrySampler(20, () -> 0L)) { sampler.start(); @@ -47,11 +47,10 @@ void samplerInterpolatesWhenIntermediateTriplesAreZero() throws Exception { TelemetryPoint first = points.getFirst(); assertEquals(0L, first.triplesLoaded()); - // Intermediate points should have interpolated values between 0 and 10,000 + // Intermediate points faithfully record 0 without faking an artificial linear curve for (int i = 1; i < points.size() - 1; i++) { TelemetryPoint p = points.get(i); - assertTrue(p.triplesLoaded() > 0, "Expected interpolated positive triples, got " + p.triplesLoaded()); - assertTrue(p.triplesLoaded() <= 10_000, "Interpolated triples must not exceed final total"); + assertEquals(0L, p.triplesLoaded(), "Intermediate samples must not fabricate unmeasured counts"); } TelemetryPoint last = points.getLast(); From 4ef4d37e5297801b0b308c9d1e41c5cf55ac55a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A9mi=20C=C3=A9r=C3=A8s?= Date: Fri, 30 Oct 2026 15:29:33 +0100 Subject: [PATCH 4/5] fix(report): dynamic protocol metadata, retain runs alias, add JSON v2 test (#29) - Add BenchmarkProtocol record to dynamically capture real options (warmup, reps, queryTimeoutMs, loadTimeoutMs) - Retain runs key alongside loadingRuns for backward compatibility with python-utils (workflow.py, baseline.py) - Pass dynamic BenchmarkProtocol from Main options - Add BenchmarkReportTest verifying complete JSON v2 schema (loadingRuns, runs, timeSeries, environment, protocol) - Maintain 0 SonarLint issues --- .../java/fr/inria/corese/benchmark/Main.java | 4 +- .../benchmark/report/BenchmarkProtocol.java | 32 +++++ .../benchmark/report/BenchmarkReport.java | 20 ++- .../benchmark/report/BenchmarkReportTest.java | 120 ++++++++++++++++++ 4 files changed, 168 insertions(+), 8 deletions(-) create mode 100644 src/main/java/fr/inria/corese/benchmark/report/BenchmarkProtocol.java create mode 100644 src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java diff --git a/src/main/java/fr/inria/corese/benchmark/Main.java b/src/main/java/fr/inria/corese/benchmark/Main.java index 65b80a0..075ecb8 100644 --- a/src/main/java/fr/inria/corese/benchmark/Main.java +++ b/src/main/java/fr/inria/corese/benchmark/Main.java @@ -7,6 +7,7 @@ import fr.inria.corese.benchmark.harness.BenchmarkHarness; import fr.inria.corese.benchmark.metrics.FileResult; import fr.inria.corese.benchmark.metrics.QueryResult; +import fr.inria.corese.benchmark.report.BenchmarkProtocol; import fr.inria.corese.benchmark.report.BenchmarkReport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -86,7 +87,8 @@ public static void main(String[] args) { crossValidateQueries(allQueryResults); try { - BenchmarkReport.write(allFileResults, allQueryResults, queries, opts.outputFile()); + BenchmarkProtocol proto = BenchmarkProtocol.fromOptions(opts.quick(), opts.queryTimeoutSecs()); + BenchmarkReport.write(allFileResults, allQueryResults, queries, opts.outputFile(), proto); } catch (Exception e) { log.error("Failed to write report: {}", e.getMessage(), e); System.exit(1); diff --git a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkProtocol.java b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkProtocol.java new file mode 100644 index 0000000..bbcaf8c --- /dev/null +++ b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkProtocol.java @@ -0,0 +1,32 @@ +package fr.inria.corese.benchmark.report; + +/** + * Protocol metadata for benchmark executions. + * + * @param warmup warmup repetitions count (0 in quick mode, 1 standard) + * @param repetitions loading repetition count + * @param queryRepetitions query repetition count + * @param queryTimeoutMs timeout per query in milliseconds + * @param loadTimeoutMs timeout for file loading in milliseconds + * @param metric statistical aggregation metric (e.g. "median") + */ +public record BenchmarkProtocol( + int warmup, + int repetitions, + int queryRepetitions, + long queryTimeoutMs, + long loadTimeoutMs, + String metric +) { + public static BenchmarkProtocol defaultProtocol() { + return new BenchmarkProtocol(1, 3, 3, 30_000L, 600_000L, "median"); + } + + public static BenchmarkProtocol fromOptions(boolean quick, int queryTimeoutS) { + int reps = quick ? 1 : 3; + int warmup = quick ? 0 : 1; + long queryTimeoutMs = queryTimeoutS > 0 ? queryTimeoutS * 1000L : 30_000L; + long loadTimeoutMs = 10L * 60L * 1000L; // 10 minutes default + return new BenchmarkProtocol(warmup, reps, reps, queryTimeoutMs, loadTimeoutMs, "median"); + } +} diff --git a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java index 26bb4ff..d99b6d7 100644 --- a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java +++ b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java @@ -48,6 +48,10 @@ private BenchmarkReport() {} * @throws Exception if serialisation or file I/O fails */ public static void write(List results, List queryResults, List queries, String outputPath) throws Exception { + write(results, queryResults, queries, outputPath, BenchmarkProtocol.defaultProtocol()); + } + + public static void write(List results, List queryResults, List queries, String outputPath, BenchmarkProtocol protocolConfig) throws Exception { Map report = new LinkedHashMap<>(); // Metadata @@ -61,13 +65,14 @@ public static void write(List results, List queryResult log.warn("Could not collect system environment metadata: {}", e.getMessage()); } + BenchmarkProtocol proto = (protocolConfig != null) ? protocolConfig : BenchmarkProtocol.defaultProtocol(); Map protocol = new LinkedHashMap<>(); - protocol.put("warmup", 1); - protocol.put("repetitions", 3); - protocol.put("queryRepetitions", 3); - protocol.put("queryTimeoutMs", 60000); - protocol.put("loadTimeoutMs", 120000); - protocol.put("metric", "median"); + protocol.put("warmup", proto.warmup()); + protocol.put("repetitions", proto.repetitions()); + protocol.put("queryRepetitions", proto.queryRepetitions()); + protocol.put("queryTimeoutMs", proto.queryTimeoutMs()); + protocol.put("loadTimeoutMs", proto.loadTimeoutMs()); + protocol.put("metric", proto.metric()); metadata.put("protocol", protocol); metadata.put("java", System.getProperty("java.version")); metadata.put("os", System.getProperty("os.name")); @@ -77,7 +82,8 @@ public static void write(List results, List queryResult } report.put("metadata", metadata); - // Load benchmark runs (contractual v2 format) + // Load benchmark runs (both 'runs' and 'loadingRuns' for v2 dashboard and python-utils compatibility) + report.put("runs", results); report.put("loadingRuns", results); // Query benchmark runs (optional) diff --git a/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java b/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java new file mode 100644 index 0000000..6efead8 --- /dev/null +++ b/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java @@ -0,0 +1,120 @@ +package fr.inria.corese.benchmark.report; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import fr.inria.corese.benchmark.metrics.FileResult; +import fr.inria.corese.benchmark.metrics.QueryResult; +import fr.inria.corese.benchmark.telemetry.TelemetryPoint; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +class BenchmarkReportTest { + + @Test + void reportSerializesJsonV2SchemaProperly(@TempDir Path tempDir) throws Exception { + Path jsonOut = tempDir.resolve("json/benchmark-report.json"); + Files.createDirectories(jsonOut.getParent()); + + List timeSeries = List.of( + new TelemetryPoint(0L, 0L, 42.5), + new TelemetryPoint(100L, 0L, 65.2), + new TelemetryPoint(250L, 5000L, 88.0) + ); + + FileResult fileResult = new FileResult( + "corese.next", + "bowlogna", + "test.ttl", + 5000L, + 250L, + 20000.0, + 40.0, + 88.0, + 48.0, + 15.0, + new long[]{240L, 250L, 260L}, + timeSeries + ); + + QueryResult queryResult = new QueryResult( + "corese.next", + "bowlogna", + "query_1", + "AGGREGATION", + 12L, + new long[]{12L, 13L, 11L}, + 5L, + new long[]{5L, 6L, 5L}, + 1.2, + 42L, + "hash123", + "PASS" + ); + + BenchmarkProtocol protocol = new BenchmarkProtocol(1, 3, 3, 45_000L, 600_000L, "median"); + + BenchmarkReport.write( + List.of(fileResult), + List.of(queryResult), + List.of("SELECT * WHERE { ?s ?p ?o }"), + jsonOut.toString(), + protocol + ); + + assertTrue(Files.exists(jsonOut), "JSON report should be generated"); + + ObjectMapper mapper = new ObjectMapper(); + JsonNode root = mapper.readTree(jsonOut.toFile()); + + // Verify metadata block + assertTrue(root.has("metadata"), "Report must contain metadata"); + JsonNode metadata = root.get("metadata"); + assertEquals("2.0", metadata.get("version").asText()); + assertTrue(metadata.has("generatedAt")); + + // Verify protocol + assertTrue(metadata.has("protocol")); + JsonNode protoNode = metadata.get("protocol"); + assertEquals(1, protoNode.get("warmup").asInt()); + assertEquals(3, protoNode.get("repetitions").asInt()); + assertEquals(45_000L, protoNode.get("queryTimeoutMs").asLong()); + assertEquals(600_000L, protoNode.get("loadTimeoutMs").asLong()); + assertEquals("median", protoNode.get("metric").asText()); + + // Verify environment + assertTrue(metadata.has("environment")); + JsonNode envNode = metadata.get("environment"); + assertTrue(envNode.has("cpuModel")); + assertTrue(envNode.has("cpuArchitecture")); + assertTrue(envNode.has("ramTotalGb")); + assertTrue(envNode.has("jvmVersion")); + + // Verify loading runs (both v2 'loadingRuns' and legacy 'runs' for python-utils compatibility) + assertTrue(root.has("loadingRuns"), "Must export 'loadingRuns' for v2 dashboard"); + assertTrue(root.has("runs"), "Must export 'runs' for backward compatibility"); + assertEquals(1, root.get("loadingRuns").size()); + assertEquals(1, root.get("runs").size()); + + // Verify timeSeries on loadingRun + JsonNode run = root.get("loadingRuns").get(0); + assertTrue(run.has("timeSeries"), "FileResult must contain timeSeries"); + JsonNode ts = run.get("timeSeries"); + assertEquals(3, ts.size()); + assertEquals(0L, ts.get(0).get("elapsedMs").asLong()); + assertEquals(42.5, ts.get(0).get("heapUsedMb").asDouble()); + assertEquals(5000L, ts.get(2).get("triplesLoaded").asLong()); + + // Verify queryRuns + assertTrue(root.has("queryRuns")); + assertEquals(1, root.get("queryRuns").size()); + assertEquals("query_1", root.get("queryRuns").get(0).get("queryId").asText()); + assertEquals("PASS", root.get("queryRuns").get(0).get("status").asText()); + } +} From cf40871ddf71d4d80a6bbc00969f8e223c88d62b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A9mi=20C=C3=A9r=C3=A8s?= Date: Fri, 30 Oct 2026 15:35:46 +0100 Subject: [PATCH 5/5] fix(harness): drive execution directly via BenchmarkProtocol and dynamically format CSV headers (#29) - Pass BenchmarkProtocol directly to BenchmarkHarness to drive repetitions, queryRepetitions, warmup and load/query timeouts - Dynamically format CSV repetition headers based on actual run lengths - Synchronize Main options and child process execution with BenchmarkProtocol - Expand BenchmarkReportTest to verify queryRepetitions, queries, null-safety and quick mode - Maintain 0 SonarLint issues across all modified files --- .../java/fr/inria/corese/benchmark/Main.java | 12 ++- .../benchmark/harness/BenchmarkHarness.java | 79 ++++++++++--------- .../benchmark/report/BenchmarkReport.java | 12 ++- .../benchmark/report/BenchmarkReportTest.java | 40 +++++++++- 4 files changed, 95 insertions(+), 48 deletions(-) diff --git a/src/main/java/fr/inria/corese/benchmark/Main.java b/src/main/java/fr/inria/corese/benchmark/Main.java index 075ecb8..6b7178f 100644 --- a/src/main/java/fr/inria/corese/benchmark/Main.java +++ b/src/main/java/fr/inria/corese/benchmark/Main.java @@ -74,20 +74,20 @@ public static void main(String[] args) { List allFileResults = new ArrayList<>(); List allQueryResults = new ArrayList<>(); + BenchmarkProtocol proto = BenchmarkProtocol.fromOptions(opts.quick(), opts.queryTimeoutSecs()); boolean shouldIsolate = opts.isolateProcess() && opts.engines().size() > 1; for (String engineName : opts.engines()) { String cleanName = engineName.trim(); if (shouldIsolate) { runChild(cleanName, opts, allFileResults, allQueryResults); } else { - runEngine(cleanName, opts, queries, allFileResults, allQueryResults); + runEngine(cleanName, opts, queries, allFileResults, allQueryResults, proto); } } crossValidateQueries(allQueryResults); try { - BenchmarkProtocol proto = BenchmarkProtocol.fromOptions(opts.quick(), opts.queryTimeoutSecs()); BenchmarkReport.write(allFileResults, allQueryResults, queries, opts.outputFile(), proto); } catch (Exception e) { log.error("Failed to write report: {}", e.getMessage(), e); @@ -207,15 +207,13 @@ private static void runEngine( BenchmarkOptions opts, List queries, List allFileResults, - List allQueryResults + List allQueryResults, + BenchmarkProtocol protocol ) { log.info("Starting engine: {}", engineName); try { TripleStoreDriver driver = DriverFactory.create(engineName, opts.inference()); - int reps = opts.quick() ? 1 : 3; - BenchmarkHarness harness = opts.queryTimeoutSecs() > 0 - ? new BenchmarkHarness(driver, opts.datasetName(), reps, opts.queryTimeoutSecs()) - : new BenchmarkHarness(driver, opts.datasetName(), reps); + BenchmarkHarness harness = new BenchmarkHarness(driver, opts.datasetName(), protocol); List fileResults = harness.run(opts.datasetDir()); allFileResults.addAll(fileResults); diff --git a/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java b/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java index 754e261..5a02f52 100644 --- a/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java +++ b/src/main/java/fr/inria/corese/benchmark/harness/BenchmarkHarness.java @@ -6,6 +6,7 @@ import fr.inria.corese.benchmark.metrics.MetricsCollector; import fr.inria.corese.benchmark.metrics.QueryResult; import fr.inria.corese.benchmark.metrics.SparqlAnalyser; +import fr.inria.corese.benchmark.report.BenchmarkProtocol; import fr.inria.corese.benchmark.telemetry.IngestionTelemetrySampler; import fr.inria.corese.benchmark.telemetry.TelemetryPoint; import org.slf4j.Logger; @@ -33,29 +34,29 @@ public class BenchmarkHarness { private static final Logger log = LoggerFactory.getLogger(BenchmarkHarness.class); - private final int repetitions; - private static final int QUERY_REPS = 3; private static final long DEFAULT_QUERY_TIMEOUT_S = 30; // seconds per query execution - private static final long LOAD_TIMEOUT_MINUTES = 10; // minutes per file load private static final String STATUS_PASS = "PASS"; private static final String STATUS_TIMEOUT = "TIMEOUT"; private static final String STATUS_ERROR = "ERROR"; - private final long queryTimeoutS; private final TripleStoreDriver driver; private final MetricsCollector metrics; private final String datasetName; + private final BenchmarkProtocol protocol; + + public BenchmarkHarness(TripleStoreDriver driver, String datasetName, BenchmarkProtocol protocol) { + this.driver = driver; + this.datasetName = datasetName; + this.protocol = protocol != null ? protocol : BenchmarkProtocol.defaultProtocol(); + this.metrics = new MetricsCollector(); + } public BenchmarkHarness(TripleStoreDriver driver, String datasetName, int repetitions) { this(driver, datasetName, repetitions, DEFAULT_QUERY_TIMEOUT_S); } public BenchmarkHarness(TripleStoreDriver driver, String datasetName, int repetitions, long queryTimeoutS) { - this.driver = driver; - this.datasetName = datasetName; - this.repetitions = repetitions; - this.queryTimeoutS = queryTimeoutS; - this.metrics = new MetricsCollector(); + this(driver, datasetName, BenchmarkProtocol.fromOptions(repetitions == 1, (int) queryTimeoutS)); } /** @@ -91,9 +92,9 @@ public List run(String datasetDir) throws Exception { file.getName(), result.loadingTimeMs(), repStr, result.triples(), result.memoryDeltaMb(), result.throughputTriplesPerSec())); } catch (java.util.concurrent.TimeoutException e) { - log.error(" [{}/{}] {} — LOAD TIMEOUT ({}min, engine={}, dataset={}), skipping file", + log.error(" [{}/{}] {} — LOAD TIMEOUT ({}ms, engine={}, dataset={}), skipping file", fi + 1, files.length, file.getName(), - LOAD_TIMEOUT_MINUTES, driver.getName(), datasetName); + protocol.loadTimeoutMs(), driver.getName(), datasetName); } catch (OutOfMemoryError e) { System.gc(); log.error(" [{}/{}] {} — OUT OF MEMORY (engine={}, dataset={}, file={}), skipping file", @@ -123,9 +124,9 @@ public List runQueryBenchmark(String datasetDir, List queri driver.load(file.getAbsolutePath()); } log.info(" {} triples loaded — running {} queries × {} reps", - driver.getGraphSize(), queries.size(), QUERY_REPS); + driver.getGraphSize(), queries.size(), protocol.queryRepetitions()); - if (repetitions > 1) { + if (protocol.warmup() > 0) { warmupQueries(queries); } else { log.info(" Quick mode — skipping query warmup"); @@ -161,8 +162,9 @@ private void runSingleWarmup(String warmupSparql, String warmupId) { return t; })) { Future wf = warmupExec.submit(() -> driver.queryWithCount(warmupSparql)); + long qTimeoutS = Math.max(1, protocol.queryTimeoutMs() / 1000L); try { - long[] res = wf.get(queryTimeoutS, TimeUnit.SECONDS); + long[] res = wf.get(qTimeoutS, TimeUnit.SECONDS); long count = res[1]; if (count == 0) { log.warn(" {} warmup: 0 results — query may be incorrect or data not loaded", warmupId); @@ -172,7 +174,7 @@ private void runSingleWarmup(String warmupSparql, String warmupId) { } catch (TimeoutException e) { wf.cancel(true); warmupExec.shutdownNow(); - log.warn(" {} warmup timed out after {}s — skipping", warmupId, queryTimeoutS); + log.warn(" {} warmup timed out after {}s — skipping", warmupId, qTimeoutS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.warn(" {} warmup interrupted", warmupId); @@ -184,15 +186,16 @@ private void runSingleWarmup(String warmupSparql, String warmupId) { } private QueryResult benchmarkSingleQuery(String sparql, String queryId, String category) { - long[] times = new long[QUERY_REPS]; - long[] ttfrTimes = new long[QUERY_REPS]; - long[] heapDeltas = new long[QUERY_REPS]; - long[] rowCounts = new long[QUERY_REPS]; - String[] hashes = new String[QUERY_REPS]; - String[] statuses = new String[QUERY_REPS]; + int qReps = protocol.queryRepetitions(); + long[] times = new long[qReps]; + long[] ttfrTimes = new long[qReps]; + long[] heapDeltas = new long[qReps]; + long[] rowCounts = new long[qReps]; + String[] hashes = new String[qReps]; + String[] statuses = new String[qReps]; RepetitionBuffers buffers = new RepetitionBuffers(times, ttfrTimes, heapDeltas, rowCounts, hashes, statuses); - for (int i = 0; i < QUERY_REPS; i++) { + for (int i = 0; i < qReps; i++) { executeSingleRepetition(sparql, queryId, i, buffers); } @@ -266,6 +269,7 @@ public String toString() { private void executeSingleRepetition( String sparql, String queryId, int rep, RepetitionBuffers buffers ) { + long qTimeoutS = Math.max(1, protocol.queryTimeoutMs() / 1000L); try (ExecutorService executor = Executors.newSingleThreadExecutor(r -> { Thread t = new Thread(r, "query-runner-" + queryId + "-rep" + rep); t.setDaemon(true); @@ -273,7 +277,7 @@ private void executeSingleRepetition( })) { Future future = executor.submit(() -> driver.queryWithMetrics(sparql)); try { - QueryMetrics m = future.get(queryTimeoutS, TimeUnit.SECONDS); + QueryMetrics m = future.get(qTimeoutS, TimeUnit.SECONDS); buffers.times()[rep] = m.totalMs(); buffers.ttfrTimes()[rep] = m.ttfrMs(); buffers.heapDeltas()[rep] = m.heapDeltaBytes(); @@ -283,7 +287,7 @@ private void executeSingleRepetition( } catch (TimeoutException e) { future.cancel(true); executor.shutdownNow(); - log.warn(" {} rep {} timed out after {}s", queryId, rep + 1, queryTimeoutS); + log.warn(" {} rep {} timed out after {}s", queryId, rep + 1, qTimeoutS); buffers.times()[rep] = -1L; buffers.ttfrTimes()[rep] = -1L; buffers.rowCounts()[rep] = -1L; @@ -350,12 +354,12 @@ private void loadWithTimeout(String path) throws InterruptedException, Execution return null; }); try { - future.get(LOAD_TIMEOUT_MINUTES, TimeUnit.MINUTES); + future.get(protocol.loadTimeoutMs(), TimeUnit.MILLISECONDS); } catch (TimeoutException e) { future.cancel(true); exec.shutdownNow(); throw new TimeoutException( - "Load timed out after " + LOAD_TIMEOUT_MINUTES + " min: " + path); + "Load timed out after " + protocol.loadTimeoutMs() + " ms: " + path); } catch (InterruptedException e) { future.cancel(true); exec.shutdownNow(); @@ -369,8 +373,8 @@ private void loadWithTimeout(String path) throws InterruptedException, Execution private FileResult benchmarkFile(File file) throws Exception { String path = file.getAbsolutePath(); - // --- Warmup (discarded) — skipped in quick mode (repetitions == 1) --- - if (repetitions > 1) { + // --- Warmup (discarded) --- + if (protocol.warmup() > 0) { driver.clear(); System.gc(); loadWithTimeout(path); @@ -378,20 +382,21 @@ private FileResult benchmarkFile(File file) throws Exception { System.gc(); } - // --- 3 repetitions --- - long[] timesMs = new long[repetitions]; - double[] memBefore = new double[repetitions]; - double[] memAfter = new double[repetitions]; - double[] cpuSamples = new double[repetitions]; + // --- Measured repetitions --- + int reps = protocol.repetitions(); + long[] timesMs = new long[reps]; + double[] memBefore = new double[reps]; + double[] memAfter = new double[reps]; + double[] cpuSamples = new double[reps]; long triples = 0; List timeSeries = List.of(); - for (int i = 0; i < repetitions; i++) { + for (int i = 0; i < reps; i++) { driver.clear(); memBefore[i] = metrics.getUsedMemoryMb(); IngestionTelemetrySampler sampler = null; - if (i == repetitions - 1) { + if (i == reps - 1) { // Non-intrusive heap telemetry sampling during the final repetition sampler = new IngestionTelemetrySampler(100, () -> 0L); sampler.start(); @@ -415,8 +420,8 @@ private FileResult benchmarkFile(File file) throws Exception { // --- Compute medians --- long medianTimeMs = MetricsCollector.median(timesMs); - double[] deltas = new double[repetitions]; - for (int i = 0; i < repetitions; i++) deltas[i] = memAfter[i] - memBefore[i]; + double[] deltas = new double[reps]; + for (int i = 0; i < reps; i++) deltas[i] = memAfter[i] - memBefore[i]; return new FileResult( driver.getName(), diff --git a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java index d99b6d7..bbe7203 100644 --- a/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java +++ b/src/main/java/fr/inria/corese/benchmark/report/BenchmarkReport.java @@ -136,8 +136,12 @@ private static Path resolveCsvDirectory(Path jsonPath) { private static void writeCsv(List results, List queryResults, String csvPath) throws Exception { try (PrintWriter pw = new PrintWriter(csvPath)) { // Load benchmark section + int loadReps = results.isEmpty() ? 3 : results.get(0).repetitionsMs().length; + String loadRepHeaders = java.util.stream.IntStream.rangeClosed(1, loadReps) + .mapToObj(i -> "rep" + i + "Ms") + .collect(Collectors.joining(",")); pw.println("# LOAD BENCHMARK"); - pw.println("engine,dataset,file,triples,loadingTimeMs,throughputTriplesPerSec,memoryBeforeMb,memoryAfterMb,memoryDeltaMb,cpuPercent,rep1Ms,rep2Ms,rep3Ms"); + pw.println("engine,dataset,file,triples,loadingTimeMs,throughputTriplesPerSec,memoryBeforeMb,memoryAfterMb,memoryDeltaMb,cpuPercent," + loadRepHeaders); for (FileResult r : results) { String reps = Arrays.stream(r.repetitionsMs()) .mapToObj(String::valueOf) @@ -151,9 +155,13 @@ private static void writeCsv(List results, List queryRe // Query benchmark section (optional) if (queryResults != null && !queryResults.isEmpty()) { + int qReps = queryResults.get(0).repetitionsMs().length; + String qRepHeaders = java.util.stream.IntStream.rangeClosed(1, qReps) + .mapToObj(i -> "rep" + i + "Ms") + .collect(Collectors.joining(",")); pw.println(); pw.println("# QUERY BENCHMARK"); - pw.println("engine,dataset,queryId,queryCategory,medianMs,medianTtfrMs,heapDeltaMb,rowCount,resultHash,status,rep1Ms,rep2Ms,rep3Ms"); + pw.println("engine,dataset,queryId,queryCategory,medianMs,medianTtfrMs,heapDeltaMb,rowCount,resultHash,status," + qRepHeaders); for (QueryResult r : queryResults) { String reps = Arrays.stream(r.repetitionsMs()) .mapToObj(String::valueOf) diff --git a/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java b/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java index 6efead8..e1acfb5 100644 --- a/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java +++ b/src/test/java/fr/inria/corese/benchmark/report/BenchmarkReportTest.java @@ -78,12 +78,16 @@ void reportSerializesJsonV2SchemaProperly(@TempDir Path tempDir) throws Exceptio JsonNode metadata = root.get("metadata"); assertEquals("2.0", metadata.get("version").asText()); assertTrue(metadata.has("generatedAt")); + assertTrue(metadata.has("queries")); + assertEquals(1, metadata.get("queries").size()); + assertEquals("SELECT * WHERE { ?s ?p ?o }", metadata.get("queries").get(0).asText()); // Verify protocol assertTrue(metadata.has("protocol")); JsonNode protoNode = metadata.get("protocol"); assertEquals(1, protoNode.get("warmup").asInt()); assertEquals(3, protoNode.get("repetitions").asInt()); + assertEquals(3, protoNode.get("queryRepetitions").asInt()); assertEquals(45_000L, protoNode.get("queryTimeoutMs").asLong()); assertEquals(600_000L, protoNode.get("loadTimeoutMs").asLong()); assertEquals("median", protoNode.get("metric").asText()); @@ -96,11 +100,11 @@ void reportSerializesJsonV2SchemaProperly(@TempDir Path tempDir) throws Exceptio assertTrue(envNode.has("ramTotalGb")); assertTrue(envNode.has("jvmVersion")); - // Verify loading runs (both v2 'loadingRuns' and legacy 'runs' for python-utils compatibility) + // Verify loading runs (both v2 'loadingRuns' and legacy 'runs' are identical) assertTrue(root.has("loadingRuns"), "Must export 'loadingRuns' for v2 dashboard"); assertTrue(root.has("runs"), "Must export 'runs' for backward compatibility"); + assertEquals(root.get("loadingRuns"), root.get("runs"), "loadingRuns and runs must be identical"); assertEquals(1, root.get("loadingRuns").size()); - assertEquals(1, root.get("runs").size()); // Verify timeSeries on loadingRun JsonNode run = root.get("loadingRuns").get(0); @@ -117,4 +121,36 @@ void reportSerializesJsonV2SchemaProperly(@TempDir Path tempDir) throws Exceptio assertEquals("query_1", root.get("queryRuns").get(0).get("queryId").asText()); assertEquals("PASS", root.get("queryRuns").get(0).get("status").asText()); } + + @Test + void nullProtocolFallsBackToDefaultProtocol(@TempDir Path tempDir) throws Exception { + Path jsonOut = tempDir.resolve("default-report.json"); + BenchmarkReport.write(Collections.emptyList(), Collections.emptyList(), Collections.emptyList(), jsonOut.toString(), null); + + ObjectMapper mapper = new ObjectMapper(); + JsonNode root = mapper.readTree(jsonOut.toFile()); + JsonNode protoNode = root.get("metadata").get("protocol"); + assertEquals(1, protoNode.get("warmup").asInt()); + assertEquals(3, protoNode.get("repetitions").asInt()); + assertEquals(3, protoNode.get("queryRepetitions").asInt()); + assertEquals(30_000L, protoNode.get("queryTimeoutMs").asLong()); + assertEquals(600_000L, protoNode.get("loadTimeoutMs").asLong()); + } + + @Test + void protocolFromOptionsAdaptsQuickMode() { + BenchmarkProtocol quickProto = BenchmarkProtocol.fromOptions(true, 15); + assertEquals(0, quickProto.warmup()); + assertEquals(1, quickProto.repetitions()); + assertEquals(1, quickProto.queryRepetitions()); + assertEquals(15_000L, quickProto.queryTimeoutMs()); + assertEquals(600_000L, quickProto.loadTimeoutMs()); + assertEquals("median", quickProto.metric()); + + BenchmarkProtocol standardProto = BenchmarkProtocol.fromOptions(false, -1); + assertEquals(1, standardProto.warmup()); + assertEquals(3, standardProto.repetitions()); + assertEquals(3, standardProto.queryRepetitions()); + assertEquals(30_000L, standardProto.queryTimeoutMs()); + } }