fix: fixing broken runtime to runtime communication (#131)

Author: Frnd-me

src/main/kotlin/at/ac/uibk/dps/cirrina/Runtime.kt

@@ -10,11 +10,12 @@ import at.ac.uibk.dps.cirrina.spec.ContextVariable
 import at.ac.uibk.dps.cirrina.spec.Csml as CsmlSpec
 import at.ac.uibk.dps.cirrina.spec.Event
 import at.ac.uibk.dps.cirrina.spec.Instance
-import at.ac.uibk.dps.cirrina.spec.graph.EventGraph
+import at.ac.uibk.dps.cirrina.spec.Instantiate
 import com.codahale.metrics.MetricRegistry
 import com.codahale.metrics.Timer
 import jakarta.inject.Inject
 import java.net.URI
+import java.util.concurrent.ConcurrentHashMap
 import kotlin.collections.component1
 import kotlin.collections.component2
 import kotlin.time.measureTime
@@ -56,43 +57,50 @@ constructor(
     var version = 0L
       private set
 
-    private val graph = EventGraph()
+    private val map = ConcurrentHashMap<String, StateMachine>()
 
     val instances: List<StateMachine>
-      get() = graph.instances
+      get() = map.values.toList()
+
+    init {
+      eventHandler.addSubscribers(csml.instances.map { it.name })
+
+      eventHandler.addDynamicSubscribers(
+        csml.collaborativeStateMachine
+          .getAllActions()
+          .filterIsInstance<Instantiate>()
+          .flatMap { it.instances }
+          .map { it.prefix }
+      )
+    }
 
     fun instantiate(
       instance: Instance,
       instanceData: List<ContextVariable> =
         instance.data.map { (k, v) -> ContextVariable(k, v.evaluate()) },
     ) {
-      val hierarchy =
-        stateMachineFactory.createHierarchy(
+      stateMachineFactory
+        .createHierarchy(
           name = instance.name,
           specification = instance.stateMachine,
-          instanceData = instanceData,
-          instanceSubscription = instance.subscription,
+          subscription = instance.subscription,
           instanceRegistry = this,
+          data = instanceData,
           parent = null,
           runtimeExtent = extent,
           eventHandler = eventHandler,
           serviceImplementationSelector = serviceImplementationSelector,
         )
+        .forEach {
+          map.put(it.name, it)
+          eventHandler.addPublishers(it.outputEvents)
+          ++version
 
-      hierarchy.forEach { machine -> graph.addInstance(machine) }
-
-      ++version
-
-      val instances = graph.instances
-
-      eventHandler.addPublishers(graph.getOutgoing(instances).map { it.event })
-      eventHandler.addSubscribers(instances.flatMap { it.subscriptions })
-
-      hierarchy.forEach { machine -> machine.start(parentContext = runtimeJob) }
+          it.start(runtimeJob)
+        }
     }
 
-    fun findStateMachineInstance(name: String): StateMachine? =
-      graph.instances.filter { it.name == name }.firstOrNull()
+    fun findStateMachineInstance(name: String): StateMachine? = map.get(name)
   }
 
   private val instanceRegistry = InstanceRegistry(stateMachineFactory)
@@ -104,6 +112,7 @@ constructor(
           .onFailure { logger.warn { "variable '${k}' already exists or failed to create" } }
       }
     }
+
     csml.instances.filter { it.name in run }.forEach { instanceRegistry.instantiate(it) }
   }
 

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

@@ -38,32 +38,32 @@ class ActionExecutor(
   private val eventHandler: StateMachine.StateMachineEventHandler,
   private val coroutineScope: CoroutineScope,
 ) {
-  fun execute(spec: Action, scope: Scope): List<Action> =
-    when (spec) {
-      is Invoke -> executeInvoke(spec, scope)
-      is Eval -> executeEval(spec, scope)
-      is Emit -> executeEmit(spec, scope)
-      is Timeout -> listOf(spec.triggers)
+  fun execute(specification: Action, scope: Scope): List<Action> =
+    when (specification) {
+      is Invoke -> executeInvoke(specification, scope)
+      is Eval -> executeEval(specification, scope)
+      is Emit -> executeEmit(specification, scope)
+      is Timeout -> listOf(specification.triggers)
       is Reset -> emptyList()
-      is Match -> executeMatch(spec, scope)
-      is Log -> executeLog(spec, scope)
-      is Instantiate -> executeInstantiate(spec, scope)
-      is Ctr -> executeCtr(spec, scope)
-      else -> error("unknown action type '${spec::class.simpleName}'")
+      is Match -> executeMatch(specification, scope)
+      is Log -> executeLog(specification, scope)
+      is Instantiate -> executeInstantiate(specification, scope)
+      is Ctr -> executeCtr(specification, scope)
+      else -> error("unknown action type '${specification::class.simpleName}'")
     }
 
-  private fun executeInvoke(spec: Invoke, scope: Scope): List<Action> {
+  private fun executeInvoke(specification: Invoke, scope: Scope): List<Action> {
     val service =
-      selector.select(spec.type, spec.mode)
-        ?: error("no service implementation found for type '${spec.type}'")
+      selector.select(specification.type, specification.mode)
+        ?: error("no service implementation found for type '${specification.type}'")
 
-    val input = spec.input.map { it.evaluate(scope.extent) }
+    val input = specification.input.map { it.evaluate(scope.extent) }
 
     coroutineScope.launch {
       val delta = measureTime {
         runCatching { service.invoke(input) }
           .onSuccess { output ->
-            spec.output.forEach { reference ->
+            specification.output.forEach { reference ->
               output
                 .firstOrNull { it.name == reference }
                 ?.let {
@@ -79,7 +79,7 @@ class ActionExecutor(
                 }
             }
 
-            spec.emits.forEach { conditionalEvent ->
+            specification.emits.forEach { conditionalEvent ->
               val shouldEmit = conditionalEvent.provided?.evaluate(scope.extent) ?: true
 
               if (shouldEmit != false) {
@@ -100,21 +100,22 @@ class ActionExecutor(
     return emptyList()
   }
 
-  private fun executeEval(spec: Eval, scope: Scope): List<Action> {
-    spec.expression.evaluate(scope.extent)
+  private fun executeEval(specification: Eval, scope: Scope): List<Action> {
+    specification.expression.evaluate(scope.extent)
     return emptyList()
   }
 
-  private fun executeEmit(spec: Emit, scope: Scope): List<Action> {
+  private fun executeEmit(specification: Emit, scope: Scope): List<Action> {
     val emittedEvent =
-      spec.event.run {
+      specification.event.run {
         val evaluatedData =
           data.map { item ->
             if (item is LazyContextVariable) item.evaluate(scope.extent) else item
           }
 
         copy(data = evaluatedData).let { eventWithData ->
-          val target = spec.target?.let { source -> source.evaluate(scope.extent) as? String }
+          val target =
+            specification.target?.let { source -> source.evaluate(scope.extent) as? String }
 
           if (target != null) eventWithData.copy(target = target) else eventWithData
         }
@@ -128,19 +129,19 @@ class ActionExecutor(
     return emptyList()
   }
 
-  private fun executeMatch(spec: Match, scope: Scope): List<Action> =
-    spec.cases.entries
+  private fun executeMatch(specification: Match, scope: Scope): List<Action> =
+    specification.cases.entries
       .filter { (expression, _) -> expression.evaluate(scope.extent) == true }
       .flatMap { it.value }
-      .ifEmpty { listOfNotNull(spec.default) }
+      .ifEmpty { listOfNotNull(specification.default) }
 
   private fun executeLog(spec: Log, scope: Scope): List<Action> {
     spec.message.evaluate(scope.extent).toString().also { logger.info(it) }
     return emptyList()
   }
 
-  private fun executeInstantiate(spec: Instantiate, scope: Scope): List<Action> {
-    val instances = spec.instances.map { it.evaluate(scope.extent).getOrThrow() }
+  private fun executeInstantiate(specification: Instantiate, scope: Scope): List<Action> {
+    val instances = specification.instances.map { it.evaluate(scope.extent).getOrThrow() }
 
     instances.forEach {
       val instanceData = it.data.map { (k, v) -> ContextVariable(k, v.evaluate(scope.extent)) }
@@ -151,20 +152,20 @@ class ActionExecutor(
     return emptyList()
   }
 
-  private fun executeCtr(spec: Ctr, scope: Scope): List<Action> {
+  private fun executeCtr(specification: Ctr, scope: Scope): List<Action> {
     val tags =
-      spec.tags?.entries?.joinToString(".") { (key, value) ->
+      specification.tags?.entries?.joinToString(".") { (key, value) ->
         "$key=${value.evaluate(scope.extent)}"
       } ?: ""
 
     val name =
       if (tags.isEmpty()) {
-        spec.counter
+        specification.counter
       } else {
-        "${spec.counter}.$tags"
+        "${specification.counter}.$tags"
       }
 
-    metricRegistry.counter(name).inc(spec.by)
+    metricRegistry.counter(name).inc(specification.by)
 
     return emptyList()
   }

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

@@ -61,6 +61,13 @@ class EventHandler(private val handler: PropagationHandler) : AutoCloseable {
     }
   }
 
+  fun addDynamicSubscribers(instancePrefixes: List<String>) {
+    instancePrefixes.forEach { prefix ->
+      val key = "events/$prefix\$*/**"
+      subscribers.computeIfAbsent(key) { createSubscriber(key, subscriberDetection = true) }
+    }
+  }
+
   fun emit(event: Event) {
     if (event.channel != Csml.EventChannel.EXTERNAL) return
 

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

@@ -14,9 +14,6 @@ import at.ac.uibk.dps.cirrina.spec.StateMachine as StateMachineSpecification
 import at.ac.uibk.dps.cirrina.spec.Timeout
 import com.codahale.metrics.MetricRegistry
 import com.codahale.metrics.Timer
-import com.google.common.cache.CacheBuilder
-import com.google.common.cache.CacheLoader
-import com.google.common.cache.LoadingCache
 import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
@@ -56,10 +53,10 @@ class StateMachine
 internal constructor(
   @Assisted val name: String,
   @Assisted val specification: StateMachineSpecification,
-  @Assisted val instanceData: List<ContextVariable>,
-  @Assisted val instanceSubscription: Regex,
+  @Assisted val subscription: Regex,
   @Assisted override val instanceRegistry: Runtime.InstanceRegistry,
   @Assisted private val parent: StateMachine?,
+  @Assisted data: List<ContextVariable>,
   @Assisted runtimeExtent: Extent,
   @Assisted eventHandler: EventHandler,
   @Assisted serviceImplementationSelector: ServiceImplementationSelector,
@@ -69,21 +66,7 @@ internal constructor(
 ) : Scope {
   override val extent: Extent
 
-  private val subscriptionCache: LoadingCache<Long, List<String>> =
-    CacheBuilder.newBuilder()
-      .maximumSize(1)
-      .build(
-        object : CacheLoader<Long, List<String>>() {
-          override fun load(key: Long): List<String> {
-            return instanceRegistry.instances
-              .filter { instanceSubscription.matches(it.name) }
-              .map { it.name }
-          }
-        }
-      )
-
-  val subscriptions: List<String>
-    get() = subscriptionCache.get(instanceRegistry.version)
+  var outputEvents = specification.outputEvents.map { it.copy(source = name) }
 
   var nested: List<String> by
     Delegates.vetoable(emptyList()) { _, old, _ ->
@@ -143,7 +126,7 @@ internal constructor(
     val transientContext =
       Context.empty().apply {
         specification.transient.forEach { (k, v) -> create(k, v.evaluate()) }
-        instanceData.forEach { create(it.name, it.value) }
+        data.forEach { create(it.name, it.value) }
       }
 
     val eventContext = Context.empty()
@@ -239,7 +222,7 @@ internal constructor(
 
   private fun Event.isValid(): Boolean {
     if (target.isNotEmpty() && target != name) return false
-    if (channel == EventChannel.EXTERNAL && source !in subscriptions) return false
+    if (channel == EventChannel.EXTERNAL && !subscription.matches(source)) return false
 
     return true
   }
@@ -370,9 +353,9 @@ internal constructor(
     fun create(
       name: String,
       specification: StateMachineSpecification,
-      instanceData: List<ContextVariable>,
-      instanceSubscription: Regex,
-      instanceRegistry: Runtime.InstanceRegistry,
+      subscription: Regex,
+      registry: Runtime.InstanceRegistry,
+      data: List<ContextVariable>,
       parent: StateMachine?,
       runtimeExtent: Extent,
       eventHandler: EventHandler,
@@ -384,9 +367,9 @@ internal constructor(
 fun Factory.createHierarchy(
   name: String,
   specification: StateMachineSpecification,
-  instanceData: List<ContextVariable>,
-  instanceSubscription: Regex,
+  subscription: Regex,
   instanceRegistry: Runtime.InstanceRegistry,
+  data: List<ContextVariable>,
   parent: StateMachine?,
   runtimeExtent: Extent,
   eventHandler: EventHandler,
@@ -395,9 +378,9 @@ fun Factory.createHierarchy(
   create(
       name,
       specification,
-      instanceData,
-      instanceSubscription,
+      subscription,
       instanceRegistry,
+      data,
       parent,
       runtimeExtent,
       eventHandler,
@@ -409,9 +392,9 @@ fun Factory.createHierarchy(
           createHierarchy(
             "${current.name}@${nested.name}",
             nested,
-            instanceData,
-            instanceSubscription,
+            subscription,
             instanceRegistry,
+            data,
             current,
             runtimeExtent,
             eventHandler,

src/main/kotlin/at/ac/uibk/dps/cirrina/spec/Action.kt

@@ -97,7 +97,9 @@ class Log internal constructor(description: LogDescription) : Action {
 class Instantiate internal constructor(csml: Csml, description: InstantiateDescription) : Action {
   val instances =
     description.instances.map {
-      DynamicInstance.create(csml, it.value, Expression(it.key)).getOrThrow()
+      val dynamicName = it.key
+      DynamicInstance.create(csml, it.value, dynamicName.first, Expression(dynamicName.second))
+        .getOrThrow()
     }
 }
 

src/main/kotlin/at/ac/uibk/dps/cirrina/spec/CollaborativeStateMachine.kt

@@ -17,6 +17,24 @@ private constructor(csml: Csml, description: CollaborativeStateMachineDescriptio
   fun getStateMachine(name: String) =
     stateMachines[name] ?: error("state machine class '${name}' not found")
 
+  fun getAllActions(): List<Action> {
+    fun StateMachine.allStateMachines(): List<StateMachine> =
+      listOf(this) + nested.flatMap { it.allStateMachines() }
+
+    return stateMachines.values
+      .flatMap { it.allStateMachines() }
+      .flatMap { sm ->
+        val stateActions =
+          sm.vertexSet().flatMap { state ->
+            listOf(state.entry, state.exit, state.during, state.after).flatMap { it.vertexSet() }
+          }
+
+        val transitionActions = sm.edgeSet().flatMap { transition -> transition.yields.vertexSet() }
+
+        stateActions + transitionActions
+      }
+  }
+
   companion object {
     fun create(csml: Csml, description: CollaborativeStateMachineDescription) = runCatching {
       CollaborativeStateMachine(csml, description)