Renamed input files and internally server is changed to task (#246)

* Updated SimTrace to use a single ArrayDeque instead of three separate lists for deadline, cpuUsage, and coreCount

* Renamed input files to tasks.parquet and fragments.parquet. Renamed server to task. OpenDC nows exports tasks.parquet instead of server.parquet

8 files changed, 54 insertions, 39 deletions

diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt
index 8fb4a938..8c1fa803 100644
--- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt
+++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt
@@ -29,9 +29,9 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.yield
-import org.opendc.compute.api.Server
-import org.opendc.compute.api.ServerState
-import org.opendc.compute.api.ServerWatcher
+import org.opendc.compute.api.Task
+import org.opendc.compute.api.TaskState
+import org.opendc.compute.api.TaskWatcher
 import org.opendc.compute.failure.models.FailureModel
 import org.opendc.compute.service.ComputeService
 import org.opendc.compute.workload.VirtualMachine
@@ -44,13 +44,13 @@ import kotlin.coroutines.coroutineContext
 import kotlin.math.max
 
 /**
- * A watcher that is locked and waits for a change in the server state to unlock
- * @param unlockStates determine which [ServerState] triggers an unlock.
+ * A watcher that is locked and waits for a change in the task state to unlock
+ * @param unlockStates determine which [TaskState] triggers an unlock.
  *                     Default values are TERMINATED, ERROR, and DELETED.
  */
-public class RunningServerWatcher : ServerWatcher {
+public class RunningTaskWatcher : TaskWatcher {
     // TODO: make this changeable
-    private val unlockStates: List<ServerState> = listOf(ServerState.DELETED, ServerState.TERMINATED)
+    private val unlockStates: List<TaskState> = listOf(TaskState.DELETED, TaskState.TERMINATED)
 
     private val mutex: Mutex = Mutex()
 
@@ -63,8 +63,8 @@ public class RunningServerWatcher : ServerWatcher {
     }
 
     override fun onStateChanged(
-        server: Server,
-        newState: ServerState,
+        task: Task,
+        newState: TaskState,
     ) {
         if (unlockStates.contains(newState)) {
             mutex.unlock()
@@ -78,7 +78,7 @@ public class RunningServerWatcher : ServerWatcher {
  * @param clock The simulation clock.
  * @param trace The trace to simulate.
  * @param seed The seed to use for randomness.
- * @param submitImmediately A flag to indicate that the servers are scheduled immediately (so not at their start time).
+ * @param submitImmediately A flag to indicate that the tasks are scheduled immediately (so not at their start time).
  * @param failureModelSpec A failure model to use for injecting failures.
  */
 public suspend fun ComputeService.replay(
@@ -111,12 +111,12 @@ public suspend fun ComputeService.replay(
                 val now = clock.millis()
                 val start = entry.startTime.toEpochMilli()
 
-                // Set the simulationOffset based on the starting time of the first server
+                // Set the simulationOffset based on the starting time of the first task
                 if (simulationOffset == Long.MIN_VALUE) {
                     simulationOffset = start - now
                 }
 
-                // Delay the server based on the startTime given by the trace.
+                // Delay the task based on the startTime given by the trace.
                 if (!submitImmediately) {
                     delay(max(0, (start - now - simulationOffset)))
                 }
@@ -135,8 +135,8 @@ public suspend fun ComputeService.replay(
                 val meta = mutableMapOf<String, Any>("workload" to workload)
 
                 launch {
-                    val server =
-                        client.newServer(
+                    val task =
+                        client.newTask(
                             entry.name,
                             image,
                             client.newFlavor(
@@ -148,15 +148,15 @@ public suspend fun ComputeService.replay(
                             meta = meta,
                         )
 
-                    val serverWatcher = RunningServerWatcher()
-                    serverWatcher.lock()
-                    server.watch(serverWatcher)
+                    val taskWatcher = RunningTaskWatcher()
+                    taskWatcher.lock()
+                    task.watch(taskWatcher)
 
-                    // Wait until the server is terminated
-                    serverWatcher.wait()
+                    // Wait until the task is terminated
+                    taskWatcher.wait()
 
-                    // Stop the server after reaching the end-time of the virtual machine
-                    server.delete()
+                    // Stop the task after reaching the end-time of the virtual machine
+                    task.delete()
                 }
             }
         }
diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
index 0f76d580..abd359ff 100644
--- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
+++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
@@ -122,14 +122,14 @@ public fun runScenario(
             )
 
             val workloadLoader = ComputeWorkloadLoader(File(scenario.workloadSpec.pathToFile))
-            val vms = getWorkloadType(scenario.workloadSpec.type).resolve(workloadLoader, Random(seed))
+            val tasks = getWorkloadType(scenario.workloadSpec.type).resolve(workloadLoader, Random(seed))
 
             val carbonTrace = getCarbonTrace(scenario.carbonTracePath)
-            val startTime = Duration.ofMillis(vms.minOf { it.startTime }.toEpochMilli())
+            val startTime = Duration.ofMillis(tasks.minOf { it.startTime }.toEpochMilli())
             addExportModel(provisioner, serviceDomain, scenario, seed, startTime, carbonTrace, scenario.id)
 
             val service = provisioner.registry.resolve(serviceDomain, ComputeService::class.java)!!
-            service.replay(timeSource, vms, failureModelSpec = scenario.failureModelSpec, seed = seed)
+            service.replay(timeSource, tasks, failureModelSpec = scenario.failureModelSpec, seed = seed)
         }
     }
 
diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/scenario/specs/WorkloadSpec.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/scenario/specs/WorkloadSpec.kt
index c104d759..956e97f1 100644
--- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/scenario/specs/WorkloadSpec.kt
+++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/scenario/specs/WorkloadSpec.kt
@@ -67,6 +67,6 @@ public enum class WorkloadTypes {
  */
 public fun getWorkloadType(type: WorkloadTypes): ComputeWorkload {
     return when (type) {
-        WorkloadTypes.ComputeWorkload -> trace("trace").sampleByLoad(1.0)
+        WorkloadTypes.ComputeWorkload -> trace().sampleByLoad(1.0)
     }
 }
diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioIntegrationTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioIntegrationTest.kt
index 5ad1ecde..301d507b 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioIntegrationTest.kt
+++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioIntegrationTest.kt
@@ -110,16 +110,16 @@ class ScenarioIntegrationTest {
                     "Success=${monitor.attemptsSuccess} " +
                     "Failure=${monitor.attemptsFailure} " +
                     "Error=${monitor.attemptsError} " +
-                    "Pending=${monitor.serversPending} " +
-                    "Active=${monitor.serversActive}",
+                    "Pending=${monitor.tasksPending} " +
+                    "Active=${monitor.tasksActive}",
             )
 
             // Note that these values have been verified beforehand
             assertAll(
                 { assertEquals(50, monitor.attemptsSuccess, "The scheduler should schedule 50 VMs") },
-                { assertEquals(0, monitor.serversActive, "All VMs should finish after a run") },
+                { assertEquals(0, monitor.tasksActive, "All VMs should finish after a run") },
                 { assertEquals(0, monitor.attemptsFailure, "No VM should be unscheduled") },
-                { assertEquals(0, monitor.serversPending, "No VM should not be in the queue") },
+                { assertEquals(0, monitor.tasksPending, "No VM should not be in the queue") },
                 { assertEquals(43795971955, monitor.idleTime) { "Incorrect idle time" } },
                 { assertEquals(2864995687, monitor.activeTime) { "Incorrect active time" } },
                 { assertEquals(148, monitor.stealTime) { "Incorrect steal time" } },
@@ -156,8 +156,8 @@ class ScenarioIntegrationTest {
                     "Success=${monitor.attemptsSuccess} " +
                     "Failure=${monitor.attemptsFailure} " +
                     "Error=${monitor.attemptsError} " +
-                    "Pending=${monitor.serversPending} " +
-                    "Active=${monitor.serversActive}",
+                    "Pending=${monitor.tasksPending} " +
+                    "Active=${monitor.tasksActive}",
             )
 
             // Note that these values have been verified beforehand
@@ -197,8 +197,8 @@ class ScenarioIntegrationTest {
                     "Success=${monitor.attemptsSuccess} " +
                     "Failure=${monitor.attemptsFailure} " +
                     "Error=${monitor.attemptsError} " +
-                    "Pending=${monitor.serversPending} " +
-                    "Active=${monitor.serversActive}",
+                    "Pending=${monitor.tasksPending} " +
+                    "Active=${monitor.tasksActive}",
             )
 
             // Note that these values have been verified beforehand
@@ -265,15 +265,15 @@ class ScenarioIntegrationTest {
         var attemptsSuccess = 0
         var attemptsFailure = 0
         var attemptsError = 0
-        var serversPending = 0
-        var serversActive = 0
+        var tasksPending = 0
+        var tasksActive = 0
 
         override fun record(reader: ServiceTableReader) {
             attemptsSuccess = reader.attemptsSuccess
             attemptsFailure = reader.attemptsFailure
             attemptsError = reader.attemptsError
-            serversPending = reader.serversPending
-            serversActive = reader.serversActive
+            tasksPending = reader.tasksPending
+            tasksActive = reader.tasksActive
         }
 
         var idleTime = 0L
diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioRunnerTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioRunnerTest.kt
index f10ab310..0b32b8f6 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioRunnerTest.kt
+++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioRunnerTest.kt
@@ -51,7 +51,7 @@ class ScenarioRunnerTest {
      Topology("topology"),
      Workload("bitbrains-small", trace("bitbrains-small")),
      OperationalPhenomena(failureFrequency = 24.0 * 7, hasInterference = true),
-     "active-servers"
+     "active-tasks"
      )
 
      assertDoesNotThrow { runner.runScenario(scenario, seed = 0L) }
@@ -70,7 +70,7 @@ class ScenarioRunnerTest {
      Topology("topology"),
      Workload("bitbrains-small", trace("bitbrains-small")),
      OperationalPhenomena(failureFrequency = 24.0 * 7, hasInterference = true),
-     "active-servers"
+     "active-tasks"
      )
 
      assertDoesNotThrow { runner.runScenario(scenario, seed = 0L) }
diff --git a/opendc-experiments/opendc-experiments-base/src/test/resources/model.json b/opendc-experiments/opendc-experiments-base/src/test/resources/model.json
new file mode 100644
index 00000000..91e2657f
--- /dev/null
+++ b/opendc-experiments/opendc-experiments-base/src/test/resources/model.json
@@ -0,0 +1,15 @@
+{
+    "version": "1.0",
+    "defaultSchema": "trace",
+    "schemas": [
+        {
+            "name": "trace",
+            "type": "custom",
+            "factory": "org.opendc.trace.calcite.TraceSchemaFactory",
+            "operand": {
+                "path": "trace",
+                "format": "opendc-vm"
+            }
+        }
+    ]
+}
diff --git a/opendc-experiments/opendc-experiments-base/src/test/resources/trace/bitbrains-small/trace.parquet b/opendc-experiments/opendc-experiments-base/src/test/resources/trace/bitbrains-small/fragments.parquet
index 9d953956..9d953956 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/resources/trace/bitbrains-small/trace.parquet
+++ b/opendc-experiments/opendc-experiments-base/src/test/resources/trace/bitbrains-small/fragments.parquet
diff --git a/opendc-experiments/opendc-experiments-base/src/test/resources/trace/bitbrains-small/meta.parquet b/opendc-experiments/opendc-experiments-base/src/test/resources/trace/bitbrains-small/tasks.parquet
index 9cded35f..9cded35f 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/resources/trace/bitbrains-small/meta.parquet
+++ b/opendc-experiments/opendc-experiments-base/src/test/resources/trace/bitbrains-small/tasks.parquet