feat: adding emission time to events and a timer to measure event lat… (#114)

Author: AlexZangerle

src/main/kotlin/at/ac/uibk/dps/cirrina/execution/object/Event.kt

@@ -3,7 +3,6 @@ package at.ac.uibk.dps.cirrina.execution.`object`
 import at.ac.uibk.dps.cirrina.csm.Csml.EventChannel
 import at.ac.uibk.dps.cirrina.csm.Csml.EventDescription
 import at.ac.uibk.dps.cirrina.util.getInsecureUuid
-import kotlin.time.Clock
 
 data class Event(
   val topic: String,
@@ -12,7 +11,7 @@ data class Event(
   val target: String = "",
   val source: String = "",
   val id: String = getInsecureUuid().toString(),
-  val createdTime: Long = Clock.System.now().epochSeconds,
+  val emittedTime: Long = 0L,
 ) {
   fun evaluateData(extent: Extent): Event = copy(data = data.map { it.evaluate(extent) })
 

src/main/kotlin/at/ac/uibk/dps/cirrina/execution/object/StateMachine.kt

@@ -6,9 +6,11 @@ import at.ac.uibk.dps.cirrina.csm.Csml.EventChannel
 import at.ac.uibk.dps.cirrina.execution.`object`.StateMachine.Factory
 import at.ac.uibk.dps.cirrina.spec.Instance
 import at.ac.uibk.dps.cirrina.spec.StateMachine as StateMachineSpec
+import com.codahale.metrics.Timer
 import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
+import java.util.concurrent.TimeUnit
 import kotlin.properties.Delegates
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
@@ -78,6 +80,8 @@ internal constructor(
 
   override val extent: Extent
 
+  private val eventTimer: Timer = runtime.metricRegistry.timer("event.latency")
+
   init {
     val transientContext = Context.from(specification.transient)
     val instanceData = Context.from(instance.data).getAll()
@@ -132,6 +136,8 @@ internal constructor(
   private fun processEvent(event: Event) {
     if (isTerminated()) return
 
+    eventTimer.update((System.nanoTime() - event.emittedTime) / 1_000, TimeUnit.MICROSECONDS)
+
     handleEvent(event)?.let { transition -> step(transition) }
 
     if (event.channel == EventChannel.INTERNAL) stateMachineEventHandler.propagateToNested(event)
@@ -255,7 +261,8 @@ internal constructor(
   override fun toString() = "StateMachine(name='$name')"
 
   inner class StateMachineEventHandler(val eventHandler: EventHandler) {
-    fun emit(event: Event) = eventHandler.emit(event.copy(source = name))
+    fun emit(event: Event) =
+      eventHandler.emit(event.copy(source = name, emittedTime = System.nanoTime()))
 
     fun propagateToParent(event: Event) {
       parent?.stateMachineEventHandler?.propagateToParent(event) ?: pushEvent(event)

src/main/kotlin/at/ac/uibk/dps/cirrina/execution/util/EventExchange.kt

@@ -28,7 +28,7 @@ object EventExchange {
         .setTarget(event.target)
         .setSource(event.source)
         .setId(event.id)
-        .setCreatedTime(event.createdTime)
+        .setEmittedTime(event.emittedTime)
 
     event.data.forEach { variable -> builder.addData(ContextVariableExchange.toProto(variable)) }
 
@@ -66,7 +66,7 @@ object EventExchange {
       proto.target,
       proto.source,
       proto.id,
-      proto.createdTime,
+      proto.emittedTime,
     )
   }
 }

src/main/proto/Event.proto

@@ -22,5 +22,5 @@ message Event {
   string target = 4;
   string source = 5;
   string id = 6;
-  int64 createdTime = 7;
+  int64 emittedTime = 7;
 }

src/test/kotlin/at/ac/uibk/dps/cirrina/CompleteTest.kt

@@ -98,7 +98,7 @@ class CompleteTest {
         .addData(contextVariable)
         .setSource("source")
         .setId("someId")
-        .setCreatedTime(1)
+        .setEmittedTime(1)
         .build()
 
     assertThrows<NullPointerException> {

src/test/kotlin/at/ac/uibk/dps/cirrina/execution/util/EventExchangeTest.kt

@@ -30,7 +30,7 @@ class EventExchangeTest {
     assertEquals(originalEvent.data[0].value, "some string")
     assertEquals(originalEvent.source, "source")
     assertEquals(originalEvent.id, "id")
-    assertEquals(originalEvent.createdTime, 1)
+    assertEquals(originalEvent.emittedTime, 1)
 
     // Perform round-trip and verify the event properties
     originalEvent.roundTrip().let { receivedEvent ->
@@ -40,7 +40,7 @@ class EventExchangeTest {
       assertEquals(originalEvent.target, receivedEvent.target)
       assertEquals(originalEvent.source, receivedEvent.source)
       assertEquals(originalEvent.id, receivedEvent.id)
-      assertEquals(originalEvent.createdTime, receivedEvent.createdTime)
+      assertEquals(originalEvent.emittedTime, receivedEvent.emittedTime)
     }
   }