fix: aligning dining philosopher implementation with csm version (#29)

* fix: aligning dining philosopher implementation with CSM version

* style: fixing formatting for dining philosophers

* style: switching dining philosophers to correct style

* style: fixing imports for dining philosophers

Author: AlexZangerle

diningPhilosophers/src/main/kotlin/ac/at/uibk/dps/dapr/philosophers/DiningPhilosophers.kt

@@ -9,8 +9,6 @@ import io.dapr.actors.ActorId
 import io.dapr.actors.client.ActorClient
 import io.dapr.actors.client.ActorProxyBuilder
 import io.dapr.actors.runtime.ActorRuntime
-import io.dapr.client.DaprClient
-import io.dapr.client.DaprClientBuilder
 import io.micrometer.core.instrument.Clock
 import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics
 import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics
@@ -21,9 +19,6 @@ import io.micrometer.core.instrument.util.HierarchicalNameMapper
 import java.nio.file.Files
 import java.nio.file.Paths
 import java.util.concurrent.TimeUnit
-import kotlin.apply
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
 import org.springframework.boot.ApplicationArguments
 import org.springframework.boot.ApplicationRunner
 import org.springframework.boot.autoconfigure.SpringBootApplication
@@ -33,8 +28,6 @@ import org.springframework.stereotype.Component
 @SpringBootApplication
 class DiningPhilosophers {
   companion object {
-    val logger: Logger = LoggerFactory.getLogger(DiningPhilosophers::class.java)
-    val daprClient: DaprClient = DaprClientBuilder().build()
     val metricsDirectory = System.getenv("METRICS_DIRECTORY") ?: "metrics"
     val metricsPeriod = System.getenv("METRICS_PERIOD")?.toLong() ?: 1L
 
@@ -70,10 +63,7 @@ class DiningPhilosophers {
 fun main(args: Array<String>) {
   val role = System.getenv("ROLE") ?: "philosopher"
   if (role == "arbitrator") {
-    val numberOfPhilosophers = System.getenv("NUMBER_OF_PHILOSOPHERS").toInt()
-    ActorRuntime.getInstance().registerActor(ArbitratorActorImpl::class.java) { runtime, id ->
-      ArbitratorActorImpl(runtime, id, numberOfPhilosophers)
-    }
+    ActorRuntime.getInstance().registerActor(ArbitratorActorImpl::class.java)
   }
   if (role == "philosopher")
     ActorRuntime.getInstance().registerActor(PhilosopherActorImpl::class.java)
@@ -88,7 +78,6 @@ class AutoStarter : ApplicationRunner {
     if (role == "arbitrator") return
     val id = System.getenv("PHILOSOPHER_ID")
     val proxy = ActorProxyBuilder(PhilosopherActor::class.java, ActorClient()).build(ActorId(id))
-    DiningPhilosophers.logger.info("philosopher requesting initial forks")
-    proxy.start().subscribe()
+    proxy.starting()
   }
 }

diningPhilosophers/src/main/kotlin/ac/at/uibk/dps/dapr/philosophers/arbitrator/ArbitratorActor.kt

@@ -2,12 +2,10 @@ package ac.at.uibk.dps.dapr.philosophers.arbitrator
 
 import io.dapr.actors.ActorMethod
 import io.dapr.actors.ActorType
-import reactor.core.publisher.Mono
 
 @ActorType(name = "ArbitratorActor")
 interface ArbitratorActor {
+  @ActorMethod(name = "hungry") fun hungry(data: Map<String, Any>)
 
-  @ActorMethod(name = "requestForks") fun requestForks(data: Map<String, Any>): Mono<Void>
-
-  @ActorMethod(name = "doneEating") fun doneEating(data: Map<String, Any>): Mono<Void>
+  @ActorMethod(name = "release") fun release(data: Map<String, Any>)
 }

diningPhilosophers/src/main/kotlin/ac/at/uibk/dps/dapr/philosophers/arbitrator/ArbitratorActorImpl.kt

@@ -1,71 +1,57 @@
 package ac.at.uibk.dps.dapr.philosophers.arbitrator
 
 import ac.at.uibk.dps.dapr.philosophers.DiningPhilosophers
-import ac.at.uibk.dps.dapr.philosophers.philosopher.PhilosopherPubSub
 import io.dapr.actors.ActorId
 import io.dapr.actors.runtime.AbstractActor
 import io.dapr.actors.runtime.ActorRuntimeContext
+import io.dapr.client.DaprClient
+import io.dapr.client.DaprClientBuilder
 import java.util.concurrent.TimeUnit
 import kotlin.time.Clock
-import kotlin.time.measureTime
-import kotlin.time.toJavaDuration
-import reactor.core.publisher.Mono
-
-class ArbitratorActorImpl(
-  runtimeContext: ActorRuntimeContext<ArbitratorActorImpl>,
-  id: ActorId,
-  val numberOfPhilosophers: Int,
-) : AbstractActor(runtimeContext, id), ArbitratorActor {
-
-  companion object {
-    const val EVENT_TIMER_NAME = "event.latency"
-    const val REQUEST_DURATION = "request.duration"
-    const val DONE_DURATION = "done.duration"
-  }
 
-  private val forks = BooleanArray(numberOfPhilosophers) { true }
+class ArbitratorActorImpl(runtimeContext: ActorRuntimeContext<ArbitratorActorImpl>, id: ActorId) :
+  AbstractActor(runtimeContext, id), ArbitratorActor {
+  private val client: DaprClient = DaprClientBuilder().build()
+
+  private val forks = BooleanArray(6) { false }
 
-  private val waiting = BooleanArray(numberOfPhilosophers) { false }
+  private val waiting = BooleanArray(6) { false }
 
   val metricsRegistry = DiningPhilosophers.provideMetricRegistry()
 
-  private fun tryAssign(pos: Int) {
-    if (forks[pos] && forks[next(pos)]) {
-      forks[pos] = false
-      forks[next(pos)] = false
-      waiting[pos] = false
-      PhilosopherPubSub.eat(DiningPhilosophers.daprClient, getMap(pos)).subscribe()
+  override fun hungry(data: Map<String, Any>) {
+    measureEventTime(data)
+    val id = data["id"].toString().toInt()
+
+    if (!(forks[id] || forks[(id + 1) % 6])) {
+      forks[id] = true
+      forks[(id + 1) % 6] = true
+      client.publishEvent("pubsub", "acquire", getMap(id)).subscribe()
     } else {
-      waiting[pos] = true
+      waiting[id] = true
     }
   }
 
-  override fun requestForks(data: Map<String, Any>): Mono<Void> {
+  override fun release(data: Map<String, Any>) {
     measureEventTime(data)
-    val duration = measureTime {
-      val pos = data["id"].toString().toInt()
-      if (pos in 0 until numberOfPhilosophers) tryAssign(pos)
-      else DiningPhilosophers.logger.info("Invalid philosopher position: $pos")
-    }
-    metricsRegistry.timer(REQUEST_DURATION).update(duration.toJavaDuration())
-    return Mono.empty()
-  }
+    val id = data["id"].toString().toInt()
 
-  override fun doneEating(data: Map<String, Any>): Mono<Void> {
-    measureEventTime(data)
-    val duration = measureTime {
-      val pos = data["id"].toString().toInt()
-      forks[pos] = true
-      forks[next(pos)] = true
-      listOf(next(pos), prev(pos), pos).filter { waiting[it] }.forEach { tryAssign(it) }
-    }
-    metricsRegistry.timer(DONE_DURATION).update(duration.toJavaDuration())
-    return Mono.empty()
+    forks[id] = false
+    forks[(id + 1) % 6] = false
+
+    test((id + 6 - 1) % 6)
+    test((id + 1) % 6)
   }
 
-  private fun next(i: Int) = (i + 1) % numberOfPhilosophers
+  private fun test(nid: Int) {
+    if (waiting[nid] && !(forks[nid] || forks[(nid + 1) % 6])) {
+      waiting[nid] = false
+      forks[nid] = true
+      forks[(nid + 1) % 6] = true
 
-  private fun prev(i: Int) = (i - 1 + numberOfPhilosophers) % numberOfPhilosophers
+      client.publishEvent("pubsub", "acquire", getMap(nid)).subscribe()
+    }
+  }
 
   private fun getMap(i: Int): Map<String, Any> {
     val now = Clock.System.now()
@@ -80,6 +66,6 @@ class ArbitratorActorImpl(
 
     val deltaNanos = (nowNanos - data["time"] as Long).coerceAtLeast(0L)
 
-    metricsRegistry.timer(EVENT_TIMER_NAME)!!.update((deltaNanos), TimeUnit.NANOSECONDS)
+    metricsRegistry.timer("event.latency")!!.update((deltaNanos), TimeUnit.NANOSECONDS)
   }
 }

diningPhilosophers/src/main/kotlin/ac/at/uibk/dps/dapr/philosophers/arbitrator/ArbitratorPubSub.kt

@@ -4,45 +4,27 @@ import io.dapr.Topic
 import io.dapr.actors.ActorId
 import io.dapr.actors.client.ActorClient
 import io.dapr.actors.client.ActorProxyBuilder
-import io.dapr.client.DaprClient
 import io.dapr.client.domain.CloudEvent
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.web.bind.annotation.PostMapping
 import org.springframework.web.bind.annotation.RequestBody
 import org.springframework.web.bind.annotation.RestController
-import reactor.core.publisher.Mono
 
 @RestController
 @ConditionalOnProperty("app.role", havingValue = "arbitrator")
 class ArbitratorPubSub {
-
-  companion object {
-    const val REQUEST_FORKS_TOPIC_NAME = "requestForks"
-    const val DONE_EATING_TOPIC_NAME = "doneEating"
-    const val PUB_SUB_NAME = "pubsub"
-    const val ARBITRATOR_NAME = "arbitrator"
-
-    fun requestForks(client: DaprClient, data: Map<String, Any>): Mono<Void> {
-      return client.publishEvent(PUB_SUB_NAME, REQUEST_FORKS_TOPIC_NAME, data)
-    }
-
-    fun doneEating(client: DaprClient, data: Map<String, Any>): Mono<Void> {
-      return client.publishEvent(PUB_SUB_NAME, DONE_EATING_TOPIC_NAME, data)
-    }
-  }
-
   val arbitratorProxy: ArbitratorActor =
-    ActorProxyBuilder(ArbitratorActor::class.java, ActorClient()).build(ActorId(ARBITRATOR_NAME))
+    ActorProxyBuilder(ArbitratorActor::class.java, ActorClient()).build(ActorId("arbitrator"))
 
-  @Topic(name = REQUEST_FORKS_TOPIC_NAME, pubsubName = PUB_SUB_NAME)
-  @PostMapping("/requestForks")
-  fun requestForksSubscriber(@RequestBody(required = true) event: CloudEvent<Map<String, Any>>) {
-    arbitratorProxy.requestForks(event.data).subscribe()
+  @Topic(name = "hungry", pubsubName = "pubsub")
+  @PostMapping("/hungry")
+  fun hungrySubscriber(@RequestBody(required = true) event: CloudEvent<Map<String, Any>>) {
+    arbitratorProxy.hungry(event.data)
   }
 
-  @Topic(name = DONE_EATING_TOPIC_NAME, pubsubName = PUB_SUB_NAME)
-  @PostMapping("/doneEating")
-  fun doneEatingSubscriber(@RequestBody(required = true) event: CloudEvent<Map<String, Any>>) {
-    arbitratorProxy.doneEating(event.data).subscribe()
+  @Topic(name = "release", pubsubName = "pubsub")
+  @PostMapping("/release")
+  fun releaseSubscriber(@RequestBody(required = true) event: CloudEvent<Map<String, Any>>) {
+    arbitratorProxy.release(event.data)
   }
 }

diningPhilosophers/src/main/kotlin/ac/at/uibk/dps/dapr/philosophers/philosopher/PhilosopherActor.kt

@@ -2,12 +2,10 @@ package ac.at.uibk.dps.dapr.philosophers.philosopher
 
 import io.dapr.actors.ActorMethod
 import io.dapr.actors.ActorType
-import reactor.core.publisher.Mono
 
 @ActorType(name = "PhilosopherActor")
 interface PhilosopherActor {
+  @ActorMethod(name = "starting") fun starting()
 
-  @ActorMethod(name = "eat") fun eat(data: Map<String, Any>): Mono<Void>
-
-  @ActorMethod(name = "start") fun start(): Mono<Void>
+  @ActorMethod(name = "eating") fun eating(data: Map<String, Any>)
 }

diningPhilosophers/src/main/kotlin/ac/at/uibk/dps/dapr/philosophers/philosopher/PhilosopherActorImpl.kt

@@ -1,66 +1,50 @@
 package ac.at.uibk.dps.dapr.philosophers.philosopher
 
 import ac.at.uibk.dps.dapr.philosophers.DiningPhilosophers
-import ac.at.uibk.dps.dapr.philosophers.arbitrator.ArbitratorPubSub
 import io.dapr.actors.ActorId
 import io.dapr.actors.runtime.AbstractActor
 import io.dapr.actors.runtime.ActorRuntimeContext
+import io.dapr.client.DaprClient
+import io.dapr.client.DaprClientBuilder
 import java.security.SecureRandom
-import java.time.Duration
 import java.util.concurrent.TimeUnit
 import kotlin.random.Random
 import kotlin.time.Clock
-import kotlin.time.measureTime
-import kotlin.time.toJavaDuration
-import reactor.core.publisher.Mono
 
 class PhilosopherActorImpl(runtimeContext: ActorRuntimeContext<PhilosopherActorImpl>, id: ActorId) :
   AbstractActor(runtimeContext, id), PhilosopherActor {
+  private val client: DaprClient = DaprClientBuilder().build()
 
-  companion object {
-    const val COUNTER_NAME = "philosopher.meals"
-    const val EVENT_TIMER_NAME = "event.latency"
-    const val EAT_DURATION_NAME = "eat.duration"
-  }
-
-  val seedGenerator = SecureRandom()
-
+  private val seedGenerator = SecureRandom()
   private val threadRng =
     object : ThreadLocal<Random>() {
       override fun initialValue(): Random {
         return Random(seedGenerator.nextLong())
       }
     }
 
-  var completedRounds: Int = 0
+  private val metricsRegistry = DiningPhilosophers.provideMetricRegistry()
 
-  val metricsRegistry = DiningPhilosophers.provideMetricRegistry()
+  private var meals: Int = 0
 
-  override fun start(): Mono<Void> {
-    return ArbitratorPubSub.requestForks(DiningPhilosophers.daprClient, getMap())
+  override fun starting() {
+    client.publishEvent("pubsub", "hungry", getMap()).subscribe()
   }
 
-  override fun eat(data: Map<String, Any>): Mono<Void> {
+  override fun eating(data: Map<String, Any>) {
     val now = Clock.System.now()
     val nowNanos = (now.epochSeconds * 1_000_000_000L) + now.nanosecondsOfSecond
-
     val deltaNanos = (nowNanos - data["time"] as Long).coerceAtLeast(0L)
 
-    metricsRegistry.timer(EVENT_TIMER_NAME)!!.update((deltaNanos), TimeUnit.NANOSECONDS)
+    metricsRegistry.timer("event.latency")!!.update((deltaNanos), TimeUnit.NANOSECONDS)
 
-    val delta = measureTime {
-      completedRounds++
-      metricsRegistry.counter(COUNTER_NAME).inc(1L)
+    Thread.sleep(randomAround(10, 2).toLong())
 
-      val delay =
-        Mono.delay(Duration.ofMillis(randomAround(10, 2).toLong())).flatMap {
-          ArbitratorPubSub.doneEating(DiningPhilosophers.daprClient, getMap())
-        }
-      delay.then(ArbitratorPubSub.requestForks(DiningPhilosophers.daprClient, getMap())).subscribe()
-    }
+    ++meals
+    metricsRegistry.counter("philosopher.meals").inc(1L)
 
-    metricsRegistry.timer(EAT_DURATION_NAME).update(delta.toJavaDuration())
-    return Mono.empty()
+    client.publishEvent("pubsub", "release", getMap()).subscribe()
+    client.publishEvent("pubsub", "hungry", getMap()).subscribe()
   }
 
   private fun getMap(): Map<String, Any> {
@@ -71,7 +55,6 @@ class PhilosopherActorImpl(runtimeContext: ActorRuntimeContext<PhilosopherActorI
   }
 
   fun randomAround(base: Int, delta: Int): Int {
-    val rng = threadRng.get()
-    return (base - delta..base + delta).random(rng)
+    return (base - delta..base + delta).random(threadRng.get())
   }
 }

diningPhilosophers/src/main/kotlin/ac/at/uibk/dps/dapr/philosophers/philosopher/PhilosopherPubSub.kt

@@ -4,37 +4,25 @@ import io.dapr.Topic
 import io.dapr.actors.ActorId
 import io.dapr.actors.client.ActorClient
 import io.dapr.actors.client.ActorProxyBuilder
-import io.dapr.client.DaprClient
 import io.dapr.client.domain.CloudEvent
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.web.bind.annotation.PostMapping
 import org.springframework.web.bind.annotation.RequestBody
 import org.springframework.web.bind.annotation.RestController
-import reactor.core.publisher.Mono
 
 @RestController
 @ConditionalOnProperty("app.role", havingValue = "philosopher")
 class PhilosopherPubSub {
-
-  companion object {
-    const val EAT_TOPIC_NAME = "eat"
-    const val PUB_SUB_NAME = "pubsub"
-
-    fun eat(client: DaprClient, data: Map<String, Any>): Mono<Void> {
-      return client.publishEvent(PUB_SUB_NAME, EAT_TOPIC_NAME, data)
-    }
-  }
-
   private val id = System.getenv("PHILOSOPHER_ID")
 
   private val philosopherProxy =
     ActorProxyBuilder(PhilosopherActor::class.java, ActorClient()).build(ActorId("$id"))
 
-  @Topic(name = EAT_TOPIC_NAME, pubsubName = PUB_SUB_NAME)
-  @PostMapping("/eat")
-  fun eatSubscriber(@RequestBody event: CloudEvent<Map<String, Any>>) {
+  @Topic(name = "acquire", pubsubName = "pubsub")
+  @PostMapping("/acquire")
+  fun acquireSubscriber(@RequestBody event: CloudEvent<Map<String, Any>>) {
     if (event.data["id"] as Int == id.toInt()) {
-      philosopherProxy.eat(event.data).subscribe()
+      philosopherProxy.eating(event.data)
     }
   }
 }