added locks to spawned servers

2 files changed, 39 insertions, 3 deletions

diff --git a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/ComputeSteps.kt b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/ComputeSteps.kt
index 690156e2..c79d9b37 100644
--- a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/ComputeSteps.kt
+++ b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/ComputeSteps.kt
@@ -59,7 +59,7 @@ public fun setupComputeService(
 public fun registerComputeMonitor(
     serviceDomain: String,
     monitor: ComputeMonitor,
-    exportInterval: Duration = Duration.ofMinutes(5)
+    exportInterval: Duration = Duration.ofMinutes(1)
 ): ProvisioningStep {
     return ComputeMonitorProvisioningStep(serviceDomain, monitor, exportInterval)
 }
diff --git a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/TraceHelpers.kt b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/TraceHelpers.kt
index 5d4c88cd..36daa756 100644
--- a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/TraceHelpers.kt
+++ b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/TraceHelpers.kt
@@ -27,13 +27,42 @@ package org.opendc.experiments.compute
 import kotlinx.coroutines.coroutineScope
 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.service.ComputeService
 import java.time.InstantSource
 import java.util.Random
 import kotlin.coroutines.coroutineContext
 import kotlin.math.max
 
+public class RunningServerWatcher: ServerWatcher {
+
+    private val _mutex: Mutex = Mutex();
+
+    public suspend fun lock () {
+        _mutex.lock()
+    }
+
+    override fun onStateChanged(server: Server, newState: ServerState) {
+        when (newState) {
+            ServerState.TERMINATED -> {
+                _mutex.unlock()
+            }
+            ServerState.ERROR -> {
+                _mutex.unlock()
+            }
+            ServerState.DELETED -> {
+                _mutex.unlock()
+            }
+            else -> {}
+        }
+    }
+
+}
+
 /**
  * Helper method to replay the specified list of [VirtualMachine] and suspend execution util all VMs have finished.
  *
@@ -102,9 +131,16 @@ public suspend fun ComputeService.replay(
                         meta = meta
                     )
 
+
                     // Wait for the server reach its end time
-                    val endTime = entry.stopTime.toEpochMilli()
-                    delay(endTime + workloadOffset - clock.millis() + (5 * 60 * 10000))
+//                    val endTime = entry.stopTime.toEpochMilli()
+//                    delay(endTime + workloadOffset - clock.millis() + (5 * 60 * 10000))
+
+                    val serverWatcher = RunningServerWatcher()
+
+                    serverWatcher.lock()
+                    server.watch(serverWatcher)
+                    serverWatcher.lock()
 
                     // Stop the server after reaching the end-time of the virtual machine
                     server.stop()