From 0d651019c94801091c4334fcc644d3779da1b411 Mon Sep 17 00:00:00 2001 From: Pascal Date: Wed, 29 Jan 2020 01:19:11 -0600 Subject: [PATCH] [ref] added Publisher and SimConfig trait - Publisher trait allows simulation to publish and Output to subscribe - Simconfig trait wraps simulation config settings --- src/main/scala/scim/components/Output.scala | 5 ++--- .../scala/scim/components/Simulation.scala | 20 ++++++++++--------- .../scala/scim/datastruct/Configuration.scala | 19 ++++++++++++++++-- src/main/scala/scim/lib/Interfaces.scala | 18 ++++++++++++++++- 4 files changed, 47 insertions(+), 15 deletions(-) diff --git a/src/main/scala/scim/components/Output.scala b/src/main/scala/scim/components/Output.scala index 8260412..1207c62 100644 --- a/src/main/scala/scim/components/Output.scala +++ b/src/main/scala/scim/components/Output.scala @@ -4,11 +4,10 @@ import scim.datastruct.{Cell, Group, SensorResult, SharedMemory, Time, Truth} class Output(sharedMemory: SharedMemory, printOutput: Boolean = true) extends Runnable { def run(): Unit = { - val outputBroadcast = sharedMemory.output val duration = sharedMemory.config.duration def outputHelper(time: Time): Unit = { - outputBroadcast(time.time).watch() - outputData(time, sharedMemory.getResult(time), sharedMemory.getTruth(time)) + val simOutput = sharedMemory.subscribe(time) + outputData(time, simOutput._1, simOutput._2) } Vector.tabulate(sharedMemory.config.duration)(t => Time(t, duration)).foreach(outputHelper) diff --git a/src/main/scala/scim/components/Simulation.scala b/src/main/scala/scim/components/Simulation.scala index ae293cd..a216441 100644 --- a/src/main/scala/scim/components/Simulation.scala +++ b/src/main/scala/scim/components/Simulation.scala @@ -6,13 +6,15 @@ import scim.lib.debugUtil.{DebugType, debug, debugf} class Simulation(config: Configuration) { private val sharedMemory = config.sharedMemory - private val sharedConfig = sharedMemory.config + //private val sharedConfig = sharedMemory.config private val cells = config.cells - val output = sharedMemory.output - + private val parallelizeCellMin = config.getCellThreadConfig._1 + private val parallelizeCellThreadPool = config.getCellThreadConfig._2 + private val parallelizeGroupMin = config.getGroupThreadConfig._1 + private val parallelizeGroupThreadPool = config.getGroupThreadConfig._2 def start(): Unit = { - val duration = sharedConfig.duration + val duration = config.getDuration val timeVector = Vector.tabulate(duration)(t => Time(t, duration)) def simTime(time: Time): Unit = { // FIXME: really shouldn't have to .map(kv => kv._1 -> kv._2) to get an immutable map, in fact it shouldn't even be mutable in the first place... @@ -36,15 +38,15 @@ class Simulation(config: Configuration) { val configTruth = if(sharedMemory.groundTruthConfig.contains(time)) Some(sharedMemory.groundTruthConfig(time)) else None simGroups(time, cell, groups, configTruth, prevTruths(cell), resultTruths(cell), sensorResults(cell)) } - if(sharedConfig.parCell != 0 && cells.size >= sharedConfig.parCell ) { + if(parallelizeCellMin != 0 && cells.size >= parallelizeCellMin ) { val cellList = cells.par - cellList.tasksupport = sharedConfig.cellThreadPool + cellList.tasksupport = parallelizeCellThreadPool cellList.foreach(simCell) } else { cells.foreach(simCell) } // broadcast the results - output(time.time).broadcast(true) + sharedMemory.publish(time) } 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 = { @@ -73,9 +75,9 @@ class Simulation(config: Configuration) { simulate(time, cell, prevTruth, resTruth, restSens, element) }) } - if(sharedConfig.parGroup != 0 && cells.size >= sharedConfig.parGroup ) { + if(parallelizeGroupMin != 0 && cells.size >= parallelizeGroupMin ) { val groupList = groups.par - groupList.tasksupport = sharedConfig.groupThreadPool + groupList.tasksupport = parallelizeGroupThreadPool groupList.foreach(simGroup) } else { groups.foreach(simGroup) diff --git a/src/main/scala/scim/datastruct/Configuration.scala b/src/main/scala/scim/datastruct/Configuration.scala index 8d8f852..2f33abc 100644 --- a/src/main/scala/scim/datastruct/Configuration.scala +++ b/src/main/scala/scim/datastruct/Configuration.scala @@ -29,7 +29,7 @@ case class SharedMemory(sensor: Map[Time, Map[Cell, Map[Group, Vector[SensorResu groundTruthConfig: Map[Time, Map[Group, Truth]], config: SimulationConfig, groups: Map[String, Group], - output: Vector[BroadcastObject[Boolean]]) { + private val output: Vector[BroadcastObject[Boolean]]) 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 @@ -57,6 +57,14 @@ case class SharedMemory(sensor: Map[Time, Map[Cell, Map[Group, Vector[SensorResu groundTruth.foreach(kv => kv._2.foreach(kv2 => kv2._2.foreach(kv3 => kv3._2.reset()))) output.foreach(e => e.broadcastReset()) } + + def subscribe(time: Time): SimulationOutput = { + output(time.time).watch() + (getResult(time), getTruth(time)) + } + def publish(time: Time): Unit = { + output(time.time).broadcast(true) + } } case class SimulationBehaviours(functions: Map[String, SimulationBehaviour]) { @@ -177,5 +185,12 @@ case class SensorCount(count: Int) { } // NOTE: The Group is actually obsolete in the Tuple3 as Sensor and GroundTruth hold theirs (unless removed from their case class) -case class Configuration(cells: Map[Cell, Map[Group, List[(Group, Sensor, Option[GroundTruth])]]], sharedMemory: SharedMemory) +case class Configuration(cells: Map[Cell, Map[Group, List[(Group, Sensor, Option[GroundTruth])]]], sharedMemory: SharedMemory) extends SimConfig { + def getDuration: Int = sharedMemory.config.duration + def getRuns: Int = sharedMemory.config.runs + def getTickFactor: Int = sharedMemory.config.tickfactor + def getCellThreadConfig: (Int, ForkJoinTaskSupport) = (sharedMemory.config.parCell, sharedMemory.config.cellThreadPool) + def getGroupThreadConfig: (Int, ForkJoinTaskSupport) = (sharedMemory.config.parGroup, sharedMemory.config.groupThreadPool) + def getSensorThreadConfig: (Int, ForkJoinTaskSupport) = (sharedMemory.config.parSensor, sharedMemory.config.sensorThreadPool) +} diff --git a/src/main/scala/scim/lib/Interfaces.scala b/src/main/scala/scim/lib/Interfaces.scala index 27163b2..42c7b30 100644 --- a/src/main/scala/scim/lib/Interfaces.scala +++ b/src/main/scala/scim/lib/Interfaces.scala @@ -1,6 +1,7 @@ package scim.lib import scim.datastruct.{Behaviour, Cell, Group, SensorCount, SensorResult, Time, Truth} +import scim.lib.simInterfaces.SimulationOutput import scala.collection.parallel.ForkJoinTaskSupport @@ -8,11 +9,26 @@ 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 { + trait Publisher { + def subscribe(time: Time): SimulationOutput + def publish(time: Time): Unit + def outputReset(): Unit + } + + trait SimConfig { + def getDuration: Int + def getRuns: Int + def getTickFactor: Int + def getCellThreadConfig: (Int, ForkJoinTaskSupport) + def getGroupThreadConfig: (Int, ForkJoinTaskSupport) + def getSensorThreadConfig: (Int, ForkJoinTaskSupport) + } + trait Result[A] { val result = new scim.lib.ConcurrentBroadcast.BroadcastObject[A]() def get: A = {