From 699338a4e7b226ae7acf0f4aef1b0b28d90eb5b3 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Wed, 14 Feb 2018 12:12:57 +0100
Subject: feat(#12): Implement Instrumentation API in Omega kernel

These changes implement the Instrumentation API described in issue #11
into the Omega simulation kernel.
---
 .../com/atlarge/opendc/omega/OmegaSimulation.kt    | 23 ++++++++++++++++++++++
 1 file changed, 23 insertions(+)

(limited to 'opendc-kernel-omega/src/main')

diff --git a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaSimulation.kt b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaSimulation.kt
index 532a033a..11217e8d 100644
--- a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaSimulation.kt
+++ b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaSimulation.kt
@@ -25,7 +25,12 @@
 package com.atlarge.opendc.omega
 
 import com.atlarge.opendc.simulator.*
+import com.atlarge.opendc.simulator.instrumentation.Instrument
+import com.atlarge.opendc.simulator.instrumentation.InstrumentScope
 import com.atlarge.opendc.simulator.kernel.Simulation
+import kotlinx.coroutines.experimental.channels.Channel
+import kotlinx.coroutines.experimental.channels.ReceiveChannel
+import kotlinx.coroutines.experimental.channels.SendChannel
 import mu.KotlinLogging
 import java.util.*
 import kotlin.coroutines.experimental.*
@@ -127,6 +132,24 @@ internal class OmegaSimulation<M>(bootstrap: Bootstrap<M>) : Simulation<M>, Boot
         schedule(prepare(message, destination, sender, delay))
 
     // Simulation implementation
+    override fun <T> install(capacity: Int, instrument: Instrument<T, M>): ReceiveChannel<T> {
+        val channel = Channel<T>(capacity)
+        val process = object : Process<Unit, M> {
+            override val initialState = Unit
+            override suspend fun Context<Unit, M>.run() {
+                val builder = object : InstrumentScope<T, M>, SendChannel<T> by channel, Context<Unit, M> by this {}
+                try {
+                    instrument(builder)
+                    channel.close()
+                } catch (cause: Throwable) {
+                    channel.close(cause)
+                }
+            }
+        }
+        register(process)
+        return channel
+    }
+
     override fun step() {
         while (true) {
             val envelope = queue.peek() ?: return
-- 
cgit v1.2.3


From 2c980f4627309b55a160dbe7fc17156201d9dde4 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Tue, 27 Feb 2018 10:17:46 +0100
Subject: feat(#12): Allow closing of instrument streams

This change in Instrumentation API allows the user to close the data
stream of an instrument by introducing a new concept: Port. A user can
open a `Port` for a `Simulation` object and attach an arbitrary amount of
instruments to this port. The data streams are closed by calling
`Port#close()`.
---
 .../com/atlarge/opendc/omega/OmegaSimulation.kt    | 37 ++++++++++++++--------
 1 file changed, 24 insertions(+), 13 deletions(-)

(limited to 'opendc-kernel-omega/src/main')

diff --git a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaSimulation.kt b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaSimulation.kt
index 11217e8d..4e3aa16c 100644
--- a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaSimulation.kt
+++ b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaSimulation.kt
@@ -27,11 +27,13 @@ package com.atlarge.opendc.omega
 import com.atlarge.opendc.simulator.*
 import com.atlarge.opendc.simulator.instrumentation.Instrument
 import com.atlarge.opendc.simulator.instrumentation.InstrumentScope
+import com.atlarge.opendc.simulator.instrumentation.Port
 import com.atlarge.opendc.simulator.kernel.Simulation
 import kotlinx.coroutines.experimental.channels.Channel
 import kotlinx.coroutines.experimental.channels.ReceiveChannel
 import kotlinx.coroutines.experimental.channels.SendChannel
 import mu.KotlinLogging
+import java.lang.ref.WeakReference
 import java.util.*
 import kotlin.coroutines.experimental.*
 
@@ -132,22 +134,31 @@ internal class OmegaSimulation<M>(bootstrap: Bootstrap<M>) : Simulation<M>, Boot
         schedule(prepare(message, destination, sender, delay))
 
     // Simulation implementation
-    override fun <T> install(capacity: Int, instrument: Instrument<T, M>): ReceiveChannel<T> {
-        val channel = Channel<T>(capacity)
-        val process = object : Process<Unit, M> {
-            override val initialState = Unit
-            override suspend fun Context<Unit, M>.run() {
-                val builder = object : InstrumentScope<T, M>, SendChannel<T> by channel, Context<Unit, M> by this {}
-                try {
-                    instrument(builder)
-                    channel.close()
-                } catch (cause: Throwable) {
-                    channel.close(cause)
+    override fun openPort(): Port<M> = object : Port<M> {
+        val channels: MutableSet<WeakReference<Channel<*>>> = mutableSetOf()
+
+        override fun <T> install(capacity: Int, instrument: Instrument<T, M>): ReceiveChannel<T> {
+            val channel = Channel<T>(capacity)
+            val process = object : Process<Unit, M> {
+                override val initialState = Unit
+                override suspend fun Context<Unit, M>.run() {
+                    val builder = object : InstrumentScope<T, M>, SendChannel<T> by channel, Context<Unit, M> by this {}
+                    try {
+                        instrument(builder)
+                        channel.close()
+                    } catch (cause: Throwable) {
+                        channel.close(cause)
+                    }
                 }
             }
+            channels.add(WeakReference(channel))
+            register(process)
+            return channel
         }
-        register(process)
-        return channel
+
+        override fun close(cause: Throwable?): Boolean = channels
+            .map { it.get()?.close(cause) ?: false }
+            .any()
     }
 
     override fun step() {
-- 
cgit v1.2.3