compute: Improvements to cloud compute model (v1)

This is the first of the pull requests in an attempt to improve the existing cloud compute model (see #86). This pull request restructures the compute API and splits the consumer and service interfaces into different modules:

- opendc-compute-api now defines the API interface for the OpenDC Compute module, which can be used by consumers of the OpenDC Compute service.
- opendc-compute-service hosts the service implementation for OpenDC Compute and contains all business logic regarding the IaaS platform (such as scheduling).
- opendc-compute-simulator implements a "compute driver" for the OpenDC Compute platform that simulates submitted workloads.
- Image is now a data-class and does not specify itself the workload to simulate. Instead, the workload should be passed via its tags currently (with key "workload"). In the future, the simulation backend will accept a mapper interface that maps Images to Workloads.

1 files changed, 22 insertions, 6 deletions

diff --git a/simulator/opendc-utils/src/main/kotlin/org/opendc/utils/flow/EventFlow.kt b/simulator/opendc-utils/src/main/kotlin/org/opendc/utils/flow/EventFlow.kt
index 948595b1..10f29f4e 100644
--- a/simulator/opendc-utils/src/main/kotlin/org/opendc/utils/flow/EventFlow.kt
+++ b/simulator/opendc-utils/src/main/kotlin/org/opendc/utils/flow/EventFlow.kt
@@ -58,12 +58,22 @@ public fun <T> EventFlow(): EventFlow<T> = EventFlowImpl()
 @OptIn(ExperimentalCoroutinesApi::class, FlowPreview::class)
 private class EventFlowImpl<T> : EventFlow<T> {
     private var closed: Boolean = false
-    private val subscribers = HashMap<SendChannel<T>, Unit>()
+    private val subscribers = mutableListOf<SendChannel<T>>()
 
     override fun emit(event: T) {
+        if (closed) {
+            return
+        }
+
+        val it = subscribers.iterator()
         synchronized(this) {
-            for ((chan, _) in subscribers) {
-                chan.offer(event)
+            while (it.hasNext()) {
+                val chan = it.next()
+                if (chan.isClosedForSend) {
+                    it.remove()
+                } else {
+                    chan.offer(event)
+                }
             }
         }
     }
@@ -72,9 +82,11 @@ private class EventFlowImpl<T> : EventFlow<T> {
         synchronized(this) {
             closed = true
 
-            for ((chan, _) in subscribers) {
+            for (chan in subscribers) {
                 chan.close()
             }
+
+            subscribers.clear()
         }
     }
 
@@ -87,9 +99,13 @@ private class EventFlowImpl<T> : EventFlow<T> {
             }
 
             channel = Channel(Channel.UNLIMITED)
-            subscribers[channel] = Unit
+            subscribers.add(channel)
+        }
+        try {
+            channel.consumeAsFlow().collect(collector)
+        } finally {
+            channel.close()
         }
-        channel.consumeAsFlow().collect(collector)
     }
 
     override fun toString(): String = "EventFlow"