fix(capelin): Keep trace order after sampling

This change fixes an issue with the workload sampler where the resulting
workload entries would not be ordered properly according to their
submission time.

2 files changed, 3 insertions, 2 deletions

diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/ExperimentHelpers.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/ExperimentHelpers.kt
index fa9fa2fc..b5090119 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/ExperimentHelpers.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/ExperimentHelpers.kt
@@ -255,6 +255,8 @@ suspend fun processTrace(
                     offset = entry.start - clock.millis()
                 }
 
+                // Make sure the trace entries are ordered by submission time
+                assert(entry.start - offset >= 0) { "Invalid trace order" }
                 delay(max(0, (entry.start - offset) - clock.millis()))
                 launch {
                     chan.send(Unit)
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/trace/ParquetTraceReader.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/trace/ParquetTraceReader.kt
index 5ad75565..0f49ecd2 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/trace/ParquetTraceReader.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/trace/ParquetTraceReader.kt
@@ -53,8 +53,7 @@ public class ParquetTraceReader(
                     this.zip(listOf(workload))
                 }
             }
-            .map { sampleWorkload(it.first, workload, it.second, seed) }
-            .flatten()
+            .flatMap { sampleWorkload(it.first, workload, it.second, seed).sortedBy(TraceEntry<SimWorkload>::start) }
             .iterator()
 
     override fun hasNext(): Boolean = iterator.hasNext()