From 75988aa5217e067fbf97682fcfbdb6bc5a8b3311 Mon Sep 17 00:00:00 2001 From: Pascal Date: Wed, 29 Jan 2020 01:19:11 -0600 Subject: [PATCH] [ref][fix] refactored Publisher trait - now uses watchFrame instead of vector - all BroadcastObject uses are moved to interfaces - [fix] fixed missing writeLock in BroadcastObject --- src/main/scala/scim/components/Parser.scala | 3 +- .../scala/scim/components/Simulation.scala | 2 +- .../scala/scim/datastruct/Configuration.scala | 12 +- src/main/scala/scim/lib/Concurrency.scala | 108 +----------------- src/main/scala/scim/lib/Interfaces.scala | 20 +++- 5 files changed, 27 insertions(+), 118 deletions(-) diff --git a/src/main/scala/scim/components/Parser.scala b/src/main/scala/scim/components/Parser.scala index cd56c49..bc87405 100644 --- a/src/main/scala/scim/components/Parser.scala +++ b/src/main/scala/scim/components/Parser.scala @@ -1,7 +1,6 @@ package scim.components import scim.datastruct.{Behaviour, BehaviourVector, Cell, Configuration, FunctionArgs, GroundTruth, Group, Sensor, SensorCount, SensorResult, SharedMemory, SimulationBehaviours, SimulationConfig, Time, Truth} -import scim.lib.ConcurrentBroadcast.BroadcastObject import scim.lib.debugUtil.{DebugType, debugf} import scim.lib.mapUtil._ import scim.lib.simInterfaces.SimulationBehaviour @@ -68,7 +67,7 @@ class SimulationConfigParser(input: Map[String, Any], simulationBehaviours: Simu val sharedMemoryTruthMap = traversableToMap[Time, Map[Cell, Map[Group, Truth]]](sharedMemoryTruth) - val sharedMemory = SharedMemory(sharedMemorySensorMap, sharedMemoryTruthMap, groundTruthValues.map(kv => kv._1 -> kv._2.toMap).toMap, config, groups.toMap, Vector.tabulate(config.duration)(_ => new BroadcastObject[Boolean]())) + val sharedMemory = SharedMemory(sharedMemorySensorMap, sharedMemoryTruthMap, groundTruthValues.map(kv => kv._1 -> kv._2.toMap).toMap, config, groups.toMap) Configuration(configurationMapping, sharedMemory) } diff --git a/src/main/scala/scim/components/Simulation.scala b/src/main/scala/scim/components/Simulation.scala index 0e39c3e..05f31ba 100644 --- a/src/main/scala/scim/components/Simulation.scala +++ b/src/main/scala/scim/components/Simulation.scala @@ -46,7 +46,7 @@ class Simulation(config: Configuration) { cells.foreach(simCell) } // broadcast the results - sharedMemory.publish(time) + sharedMemory.publish() } private def simGroups(time: Time, cell: Cell, groups: Map[Group, List[(Group, Sensor, Option[GroundTruth])]], configTruths: Option[Map[Group, Truth]], prevTruths: Map[Group, Truth], resultTruths: Map[Group, Truth], resultSensors: Map[Group, Vector[SensorResult]]): Unit = { diff --git a/src/main/scala/scim/datastruct/Configuration.scala b/src/main/scala/scim/datastruct/Configuration.scala index a19d139..3b02056 100644 --- a/src/main/scala/scim/datastruct/Configuration.scala +++ b/src/main/scala/scim/datastruct/Configuration.scala @@ -1,6 +1,5 @@ package scim.datastruct -import scim.lib.ConcurrentBroadcast._ import scim.lib.simInterfaces._ import scim.lib.configInterfaces._ import scim.lib.mapUtil._ @@ -31,8 +30,7 @@ case class SharedMemory(sensor: Map[Time, Map[Cell, Map[Group, Vector[SensorResu groundTruth: Map[Time, Map[Cell, Map[Group, Truth]]], groundTruthConfig: Map[Time, Map[Group, Truth]], config: SimulationConfig, - groups: Map[String, Group], - private val output: Vector[BroadcastObject[Boolean]]) extends Publisher { + groups: Map[String, Group]) extends Publisher { // memory optimization: remove sensor and throw output into output // only map Truths from config into groundtruth, do a if groundTruth.contains(time) groundTruth(time) else prevTruth in case of time.start Truth(None) (or better map it to time.start anyways in parser) // can drop cell from ground truth @@ -55,19 +53,17 @@ case class SharedMemory(sensor: Map[Time, Map[Cell, Map[Group, Vector[SensorResu def getResult(time: Time): Map[Cell, Map[Group, Vector[SensorResult]]] = { sensor(time) } + def outputReset(): Unit = { sensor.foreach(kv => kv._2.foreach(kv2 => kv2._2.foreach(kv3 => kv3._2.foreach(res => res.reset())))) groundTruth.foreach(kv => kv._2.foreach(kv2 => kv2._2.foreach(kv3 => kv3._2.reset()))) - output.foreach(e => e.broadcastReset()) + reset() } def subscribe(time: Time): SimulationOutput = { - output(time.time).watch() + waitForOutput(time) (getResult(time), getTruth(time)) } - def publish(time: Time): Unit = { - output(time.time).broadcast(true) - } } case class SimulationBehaviours(functions: Map[String, SimulationBehaviour]) { diff --git a/src/main/scala/scim/lib/Concurrency.scala b/src/main/scala/scim/lib/Concurrency.scala index 75938fe..43e561e 100644 --- a/src/main/scala/scim/lib/Concurrency.scala +++ b/src/main/scala/scim/lib/Concurrency.scala @@ -83,7 +83,7 @@ object ConcurrentBroadcast { private var broadcastEnded = false private var value: Option[A] = default - // NON-BLOCKING: check if value has been set, write new value and return if it was overwritten + // NON-BLOCKING: write new value def broadcast(newValue: A): Unit = { writeLock(() => { value = Some(newValue) @@ -118,9 +118,11 @@ object ConcurrentBroadcast { } def broadcastReset(): Unit = { - broadcastEnded = false - frame = 0 - value = default + writeLock(() => { + broadcastEnded = false + frame = 0 + value = default + }) } // BLOCKING: watch value, block until it is set, then return it @@ -174,101 +176,3 @@ object ConcurrentBroadcast { } } - - -class Test2 { - import scim.lib.ConcurrentBroadcast._ - - private val stream: mutable.ArraySeq[BroadcastObject[Int]] = new mutable.ArraySeq[Int](5).map(_ => new BroadcastObject[Int]) - - class Producer(stream: mutable.ArraySeq[BroadcastObject[Int]]) extends Broadcast[Int](stream) with Broadcaster[Int] { - def a(): Unit = { - stream.foreach(element => {element.broadcast(1); Thread.sleep(1000)}) - } - - override def broadcast(stream: mutable.ArraySeq[BroadcastObject[Int]]): Unit = { - val f: (Int, Int) => Int = (x, y) => x + y - var previous: Int = 0 - stream.foreach(element => { - previous = f(previous, 2) - element.broadcast(previous) - }) - } - } - - class Watcher(id: Int, stream: mutable.ArraySeq[BroadcastObject[Int]]) extends Runnable { - def run(): Unit = { - stream.foreach(element => { Thread.sleep(id); println(id + ": " + element.watch() + " - " + Calendar.getInstance().getTime()) }) - } - } - - def run(): Unit = { - val threadPool = Executors.newFixedThreadPool(4) - val producer = new Producer(stream) - val watcher = new Watcher(0,stream) - val watcher2 = new Watcher(3000,stream) - - val p = new Thread(producer) - - val t = new Thread(watcher) - val t2 = new Thread(watcher2) - println("start producer") - p.start() - println("start watcher") - t.start() - t2.start() - println("join producer") - p.join() - println("join watcher") - t.join() - t2.join() - println("done") - - - } - -} - -class Test { - import scim.lib.ConcurrentBroadcast.BroadcastObject - - private val stream: mutable.ArraySeq[BroadcastObject[Int]] = new mutable.ArraySeq[Int](5).map(_ => new BroadcastObject[Int]) - - class Producer(stream: mutable.ArraySeq[BroadcastObject[Int]]) extends Runnable { - def run(): Unit = { - stream.foreach(element => {element.broadcast(1); Thread.sleep(1000)}) - } - } - - class Watcher(id: Int, stream: mutable.ArraySeq[BroadcastObject[Int]]) extends Runnable { - def run(): Unit = { - stream.foreach(element => { Thread.sleep(id); println(id + ": " + element.watch() + " - " + Calendar.getInstance().getTime()) }) - } - } - - def run(): Unit = { - val threadPool = Executors.newFixedThreadPool(4) - val producer = new Producer(stream) - val watcher = new Watcher(0,stream) - val watcher2 = new Watcher(3000,stream) - - val p = new Thread(producer) - - val t = new Thread(watcher) - val t2 = new Thread(watcher2) - println("start producer") - p.start() - println("start watcher") - t.start() - t2.start() - println("join producer") - p.join() - println("join watcher") - t.join() - t2.join() - println("done") - - - } - -} diff --git a/src/main/scala/scim/lib/Interfaces.scala b/src/main/scala/scim/lib/Interfaces.scala index 42c7b30..fec04a6 100644 --- a/src/main/scala/scim/lib/Interfaces.scala +++ b/src/main/scala/scim/lib/Interfaces.scala @@ -1,23 +1,33 @@ package scim.lib -import scim.datastruct.{Behaviour, Cell, Group, SensorCount, SensorResult, Time, Truth} +import scim.datastruct.{Behaviour, Cell, FunctionArgs, Group, SensorCount, SensorResult, Time, Truth} import scim.lib.simInterfaces.SimulationOutput import scala.collection.parallel.ForkJoinTaskSupport package object simInterfaces { - import scim.datastruct._ // maybe change to by-name parameters? have to check performance gain type SimulationBehaviour = (Cell, Time, Option[Double], Option[FunctionArgs]) => Option[Double] type SimulationOutput = (Map[Cell, Map[Group, Vector[SensorResult]]], Map[Cell, Map[Group, Truth]]) } package object configInterfaces { + // have to import here, get compiler error otherwise + import scim.lib.ConcurrentBroadcast.BroadcastObject trait Publisher { + private val output = new BroadcastObject[Boolean]() + protected def waitForOutput(time: Time): Unit = { + // frames start at 1, simulation ticks at 0 + output.watchFrame(time.time + 1) + } def subscribe(time: Time): SimulationOutput - def publish(time: Time): Unit - def outputReset(): Unit + def publish(): Unit = { + output.broadcast(true) + } + def reset(): Unit = { + output.broadcastReset() + } } trait SimConfig { @@ -30,7 +40,7 @@ package object configInterfaces { } trait Result[A] { - val result = new scim.lib.ConcurrentBroadcast.BroadcastObject[A]() + val result = new BroadcastObject[A]() def get: A = { result.watch() }