feat(#12): Implement Instrumentation API

These changes contain the specification of the new Instrumentation API for the simulator, in addition to the implementation for the Omega kernel. As an example, the API allows users to measure data from processes in simulation and interpolate data points between the measurements.

Closes #11, #12

3 files changed, 81 insertions, 1 deletions

diff --git a/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/instrumentation/Instrument.kt b/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/instrumentation/Instrument.kt
new file mode 100644
index 00000000..7633444e
--- /dev/null
+++ b/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/instrumentation/Instrument.kt
@@ -0,0 +1,29 @@
+package com.atlarge.opendc.simulator.instrumentation
+
+import com.atlarge.opendc.simulator.Context
+import com.atlarge.opendc.simulator.Entity
+import kotlinx.coroutines.experimental.channels.ReceiveChannel
+import kotlinx.coroutines.experimental.channels.SendChannel
+
+    /**
+ * A kernel instrumentation device that allows the observation and measurement of properties of interest within some
+ * model.
+ *
+ * An instrument is a [Process] that emits measurements from within some model in the form of a typed stream. An
+ * instrument is attached to a simulation using the [Port.install] method, which returns a [ReceiveChannel] from
+ * which the measurements can be extracted out of the simulation.
+ *
+ * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl)
+ */
+typealias Instrument<T, M> = suspend InstrumentScope<T, M>.() -> Unit
+
+/**
+ * This interface defines the scope in which an instrumentation device is built.
+ *
+ * An instrument is a [Process] without any observable state that is allowed to send messages to other [Entity]
+ * instances in the simulation. In addition, the instrument can emit measurements using the methods provided by the
+ * [SendChannel] interface.
+ *
+ * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl)
+ */
+interface InstrumentScope<in T, M>: SendChannel<T>, Context<Unit, M>
diff --git a/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/instrumentation/Port.kt b/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/instrumentation/Port.kt
new file mode 100644
index 00000000..880f9a15
--- /dev/null
+++ b/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/instrumentation/Port.kt
@@ -0,0 +1,42 @@
+package com.atlarge.opendc.simulator.instrumentation
+
+import com.atlarge.opendc.simulator.kernel.Simulation
+import kotlinx.coroutines.experimental.channels.Channel
+import kotlinx.coroutines.experimental.channels.ReceiveChannel
+
+/**
+ * A port allows users to install instrumentation devices to a [Simulation].
+ *
+ * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl)
+ */
+interface Port<M> {
+    /**
+     * Install the given instrumentation device to produce a stream of measurements of type <code>T</code>.
+     *
+     * The [ReceiveChannel] returned by this channel is by default backed by an unlimited buffer
+     * (using [Channel.UNLIMITED]), which may induce unnecessary overhead.
+     *
+     * @param instrument The instrumentation device to install.
+     * @return A [ReceiveChannel] to which the of measurements produced by the instrument are published.
+     */
+    fun <T> install(instrument: Instrument<T, M>): ReceiveChannel<T> = install(Channel.UNLIMITED, instrument)
+
+    /**
+     * Install the given instrumentation device to produce a stream of measurements of type code>T</code>.
+     *
+     * @param capacity The capacity of the buffer of the channel.
+     * @param instrument The instrumentation device to install.
+     * @return A [ReceiveChannel] to which the of measurements produced by the instrument are published.
+     */
+    fun <T> install(capacity: Int, instrument: Instrument<T, M>): ReceiveChannel<T>
+
+    /**
+     * Close this port and stop the instruments from producing more measurements.
+     * This is an idempotent operation – repeated invocations of this function have no effect and return false.
+     *
+     * @param cause An optional cause that is thrown when trying to receive more elements from the installed
+     * instruments.
+     * @return `true` if the port was closed, `false` if it was already closed.
+     */
+    fun close(cause: Throwable? = null): Boolean
+}
diff --git a/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/kernel/Simulation.kt b/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/kernel/Simulation.kt
index bb2ef818..0954868a 100644
--- a/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/kernel/Simulation.kt
+++ b/opendc-core/src/main/kotlin/com/atlarge/opendc/simulator/kernel/Simulation.kt
@@ -26,6 +26,8 @@ package com.atlarge.opendc.simulator.kernel
 
 import com.atlarge.opendc.simulator.Entity
 import com.atlarge.opendc.simulator.Instant
+import com.atlarge.opendc.simulator.instrumentation.Instrument
+import com.atlarge.opendc.simulator.instrumentation.Port
 
 /**
  * A message based discrete event simulation over some model `M`. This interface provides direct control over the
@@ -35,7 +37,7 @@ import com.atlarge.opendc.simulator.Instant
  * @param M The shape of the model over which the simulation runs.
  * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl)
  */
-interface Simulation<out M> {
+interface Simulation<M> {
     /**
      * The model in which the simulation runs.
      */
@@ -52,6 +54,13 @@ interface Simulation<out M> {
     val <E : Entity<S, *>, S> E.state: S
 
     /**
+     * Open a new [Port] to manage [Instrument]s.
+     *
+     * @return A new [Port] instance to install [Instrument]s to.
+     */
+    fun openPort(): Port<M>
+
+    /**
      * Step through one cycle in the simulation. This method will process all events in a single tick, update the
      * internal clock and then return the control to the user.
      */