Merge pull request #168 from DanteNiewenhuis/greenifier-demo

Updated the simulation to let servers run until they are finished.

4 files changed, 121 insertions, 57 deletions

diff --git a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/ComputeWorkloadLoader.kt b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/ComputeWorkloadLoader.kt
index 9aa544cb..e8596218 100644
--- a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/ComputeWorkloadLoader.kt
+++ b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/ComputeWorkloadLoader.kt
@@ -45,6 +45,8 @@ import java.io.File
 import java.lang.ref.SoftReference
 import java.util.UUID
 import java.util.concurrent.ConcurrentHashMap
+import java.time.Duration
+import java.time.Instant
 import kotlin.math.max
 import kotlin.math.roundToLong
 
@@ -82,14 +84,12 @@ public class ComputeWorkloadLoader(private val baseDir: File) {
             while (reader.nextRow()) {
                 val id = reader.getString(idCol)!!
                 val time = reader.getInstant(timestampCol)!!
-                val duration = reader.getDuration(durationCol)!!
+                val durationMs = reader.getDuration(durationCol)!!
                 val cores = reader.getInt(coresCol)
                 val cpuUsage = reader.getDouble(usageCol)
 
-                val deadlineMs = time.toEpochMilli()
-                val timeMs = (time - duration).toEpochMilli()
                 val builder = fragments.computeIfAbsent(id) { Builder() }
-                builder.add(timeMs, deadlineMs, cpuUsage, cores)
+                builder.add(time, durationMs, cpuUsage, cores)
             }
 
             fragments
@@ -246,17 +246,17 @@ public class ComputeWorkloadLoader(private val baseDir: File) {
          * @param usage CPU usage of this fragment.
          * @param cores Number of cores used.
          */
-        fun add(timestamp: Long, deadline: Long, usage: Double, cores: Int) {
-            val duration = max(0, deadline - timestamp)
-            totalLoad += (usage * duration) / 1000.0 // avg MHz * duration = MFLOPs
+        fun add(deadline: Instant, duration: Duration, usage: Double, cores: Int) {
+            val startTimeMs = (deadline - duration).toEpochMilli()
+            totalLoad += (usage * duration.toMillis()) / 1000.0 // avg MHz * duration = MFLOPs
 
-            if (timestamp != previousDeadline) {
+            if ((startTimeMs != previousDeadline) && (previousDeadline != Long.MIN_VALUE)) {
                 // There is a gap between the previous and current fragment; fill the gap
-                builder.add(timestamp, 0.0, cores)
+                builder.add(startTimeMs, 0.0, cores)
             }
 
-            builder.add(deadline, usage, cores)
-            previousDeadline = deadline
+            builder.add(deadline.toEpochMilli(), usage, cores)
+            previousDeadline = deadline.toEpochMilli()
         }
 
         /**
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..059985a8 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,47 @@ 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()
+    }
+
+    public suspend fun wait () {
+        // TODO: look at the better way to wait for an unlock
+        this.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.
  *
@@ -63,25 +97,26 @@ public suspend fun ComputeService.replay(
             // Start the fault injector
             injector?.start()
 
-            var offset = Long.MIN_VALUE
+            var simulationOffset = Long.MIN_VALUE
 
             for (entry in trace.sortedBy { it.startTime }) {
                 val now = clock.millis()
                 val start = entry.startTime.toEpochMilli()
 
-                if (offset < 0) {
-                    offset = start - now
+                // Set the simulationOffset based on the starting time of the first server
+                if (simulationOffset == Long.MIN_VALUE) {
+                    simulationOffset = start - now
                 }
 
                 // Make sure the trace entries are ordered by submission time
-                assert(start - offset >= 0) { "Invalid trace order" }
+//                assert(start - simulationOffset >= 0) { "Invalid trace order" }
 
+                // Delay the server based on the startTime given by the trace.
                 if (!submitImmediately) {
-                    delay(max(0, (start - offset) - now))
+                    delay(max(0, (start - now - simulationOffset)));
                 }
 
-                val workloadOffset = -offset + 300001
-                val workload = entry.trace.createWorkload(workloadOffset)
+                val workload = entry.trace.createWorkload(start)
                 val meta = mutableMapOf<String, Any>("workload" to workload)
 
                 val interferenceProfile = entry.interferenceProfile
@@ -102,16 +137,18 @@ 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 serverWatcher = RunningServerWatcher()
+                    serverWatcher.lock()
+                    server.watch(serverWatcher)
+
+                    // Wait until the server is terminated
+                    serverWatcher.wait()
 
                     // Stop the server after reaching the end-time of the virtual machine
-                    server.stop()
+                    server.delete()
                 }
             }
         }
-
         yield()
     } finally {
         injector?.close()
diff --git a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/telemetry/ComputeMetricReader.kt b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/telemetry/ComputeMetricReader.kt
index de0f0b7c..e8cd3a55 100644
--- a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/telemetry/ComputeMetricReader.kt
+++ b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/telemetry/ComputeMetricReader.kt
@@ -79,46 +79,47 @@ public class ComputeMetricReader(
      */
     private val job = scope.launch {
         val intervalMs = exportInterval.toMillis()
-        val service = service
-        val monitor = monitor
-        val hostTableReaders = hostTableReaders
-        val serverTableReaders = serverTableReaders
-        val serviceTableReader = serviceTableReader
-
         try {
             while (isActive) {
                 delay(intervalMs)
 
-                try {
-                    val now = clock.instant()
-
-                    for (host in service.hosts) {
-                        val reader = hostTableReaders.computeIfAbsent(host) { HostTableReaderImpl(it) }
-                        reader.record(now)
-                        monitor.record(reader.copy())
-                        reader.reset()
-                    }
-
-                    for (server in service.servers) {
-                        val reader = serverTableReaders.computeIfAbsent(server) { ServerTableReaderImpl(service, it) }
-                        reader.record(now)
-                        monitor.record(reader)
-                        reader.reset()
-                    }
-
-                    serviceTableReader.record(now)
-                    monitor.record(serviceTableReader)
-                } catch (cause: Throwable) {
-                    logger.warn(cause) { "Exporter threw an Exception" }
-                }
+                loggState()
             }
+
         } finally {
+//            loggState()
+
             if (monitor is AutoCloseable) {
                 monitor.close()
             }
         }
     }
 
+    private fun loggState() {
+        try {
+            val now = this.clock.instant()
+
+            for (host in this.service.hosts) {
+                val reader = this.hostTableReaders.computeIfAbsent(host) { HostTableReaderImpl(it) }
+                reader.record(now)
+                this.monitor.record(reader.copy())
+                reader.reset()
+            }
+
+            for (server in this.service.servers) {
+                val reader = this.serverTableReaders.computeIfAbsent(server) { ServerTableReaderImpl(service, it) }
+                reader.record(now)
+                this.monitor.record(reader.copy())
+                reader.reset()
+            }
+
+            this.serviceTableReader.record(now)
+            monitor.record(this.serviceTableReader.copy())
+        } catch (cause: Throwable) {
+            this.logger.warn(cause) { "Exporter threw an Exception" }
+        }
+    }
+
     override fun close() {
         job.cancel()
     }
@@ -127,6 +128,27 @@ public class ComputeMetricReader(
      * An aggregator for service metrics before they are reported.
      */
     private class ServiceTableReaderImpl(private val service: ComputeService) : ServiceTableReader {
+
+        override fun copy(): ServiceTableReader {
+            val newServiceTable = ServiceTableReaderImpl(service)
+            newServiceTable.setValues(this)
+
+            return newServiceTable
+        }
+
+        override fun setValues(table: ServiceTableReader) {
+            _timestamp = table.timestamp
+
+            _hostsUp = table.hostsUp
+            _hostsDown = table.hostsDown
+            _serversTotal = table.serversTotal
+            _serversPending = table.serversPending
+            _serversActive = table.serversActive
+            _attemptsSuccess = table.attemptsSuccess
+            _attemptsFailure = table.attemptsFailure
+            _attemptsError = table.attemptsError
+        }
+
         private var _timestamp: Instant = Instant.MIN
         override val timestamp: Instant
             get() = _timestamp
@@ -362,18 +384,18 @@ public class ComputeMetricReader(
         }
 
         override fun setValues(table: ServerTableReader) {
+            host = table.host
+
             _timestamp = table.timestamp
-            _uptime = table.uptime
-            _downtime = table.downtime
-            _provisionTime = table.provisionTime
-            _bootTime = table.bootTime
             _cpuLimit = table.cpuLimit
             _cpuActiveTime = table.cpuActiveTime
             _cpuIdleTime = table.cpuIdleTime
             _cpuStealTime = table.cpuStealTime
             _cpuLostTime = table.cpuLostTime
-
-            host = table.host
+            _uptime = table.uptime
+            _downtime = table.downtime
+            _provisionTime = table.provisionTime
+            _bootTime = table.bootTime
         }
 
         private val _server = server
diff --git a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/telemetry/table/ServiceTableReader.kt b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/telemetry/table/ServiceTableReader.kt
index bb926298..a077a476 100644
--- a/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/telemetry/table/ServiceTableReader.kt
+++ b/opendc-experiments/opendc-experiments-compute/src/main/kotlin/org/opendc/experiments/compute/telemetry/table/ServiceTableReader.kt
@@ -28,6 +28,11 @@ import java.time.Instant
  * An interface that is used to read a row of a service trace entry.
  */
 public interface ServiceTableReader {
+
+    public fun copy(): ServiceTableReader
+
+    public fun setValues(table: ServiceTableReader)
+
     /**
      * The timestamp of the current entry of the reader.
      */