pascal/scim
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[add] multiple commits merged

Browse Source 
# Conflicts:
#	docs/wiki/config/config.yml
#	src/main/scala/scim/Main.scala
#	src/main/scala/scim/archive/Elements.scala
#	src/main/scala/scim/archive/GroundTruth.scala
#	src/main/scala/scim/datastruct/Configuration.scala
#	src/main/scala/scim/lib/Concurrency.scala
This commit is contained in:
Pascal
2020-01-29 01:19:10 -06:00
parent 00669fa9d3
commit 32438ef5b8
12 changed files with 845 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/wiki/config/config.md
View File

@@ -3,7 +3,7 @@
## Simulation
* `duration`:`int` - number of simulation ticks, after which everything will be reset to the initial configuration
* `tickfactor`:`int` - used to artificially add ticks by repeating the result of a simulation tick by the given value
* `tickfactor`:`int` - used to artificially add ticks by repeating the calculation of a simulation tick by the given value
* `loop`:`bool` - loop the simulation
* `loopcount`:`int` - amount of loops, `0` for infinite
@@ -59,7 +59,7 @@ Elements:
* `end`:`int` _[optional]_ - last simulation tick before this function is inactive
Alternative condensed of `timeline` notation as `array`: `[start, end]`
Alternative condensed of `value` notation as `array`: `[start, value or None, ...]`
* `delta`:`array` _[optional]_ - function to calculate `value` of the next simulation tick, only called if no value is defined for the next tick. Result is saved as new `value` and passed again to the function for the next tick. If no function is given, the `value` remains
* `delta`:`array` _[optional]_ - function to calculate `value` of the next simulation tick, overwrites any existing value. Result is saved as new `value` and passed again to the function for the next tick. If no function is given, the `value` remains
* `name`:`String` - name of the function. Needs to exist at compile time in a map: `["name" -> (value: double, parentValue: double, args: ...) => double]`
* `args`:`array` - arguments passed to the function in addition to the `value`
* `timeline`:`array` - sets simulation tick when this function is active and turns inactive.

63
docs/wiki/config/config.yml
View File

@@ -1,8 +1,14 @@
simulation:
duration: 500
tickfactor: 10
loop: true
loopcount: 20
runs: 20
parCell: 0
parCellThreads: 5
parGroup: 0
parGroupThreads: 5
parSensor: 0
parSensorThreads: 5
sensors:
- group: temperature
@@ -16,27 +22,27 @@ sensors:
end: [3, 3]
count: 30
groundtruth:
- group: faultyTemp
parent: temperature
cells:
- [1, 2, 10]
groundtruths:
- group: temperature
truth:
- value: 11
- value: 11.0
timeline:
- start: 0
end: 120
- [240, 360]
delta:
- name: factor
args:
- 1.1
args: {factor: 1.1}
timeline:
- start: 60
end: 120
cells:
- [0, 2, 10]
blocks:
- start: [8, 1]
end: [3, 3]
count: 50
- start: 20
end: 95
- [170, randomValue, {arg: true}]
- group: humidity
parent: temperature
truth:
@@ -44,27 +50,18 @@ groundtruth:
- [30, 0.6]
- [50, None]
delta:
- [10, None, []]
- [30, addValuePercent, [0.03, 0.8]]
- [50, humidFromTemp, []]
blocks:
- start: [15, 15]
end: [17, 17]
percent: 0.8
- [10, None]
- [30, addValuePercent, {jitterA: 0.03, jitterB: 0.8}]
- [50, humidFromTemp]
behaviour:
behaviours:
- group: temperature
sharedObject: objectTemp
functions:
- name: randomValue
args:
- 40
- 80
duration:
- start: 60
- [100, normalValue, [2]]
- [120, None, []]
blocks:
- start: [5,6]
end: [8,9]
percent: 1.0
args: {a: 40, b: 80}
timeline:
- start: 44
end: 166
- [100, normalValue, {v: 2}]
- [120, addValuePercent, {}]

43
src/main/scala/scim/Library.scala
View File

@@ -4,6 +4,7 @@ package lib
// java imports
import java.io.{File, FileInputStream}
// scala imports
import scala.collection.convert.wrapAll._
import scala.collection.mutable
@@ -44,6 +45,37 @@ package object printUtil {
}
}
package object evalUtil {
def time[A](name: String, f: => A): A = {
val t0 = System.nanoTime()
val result = f
val t1 = System.nanoTime()
println(name + " Elapsed time: " + (t1 - t0) + "ns")
result
}
}
package object debugUtil {
sealed case class DebugType(name: String, level: Int)
object DebugType {
object ERROR extends DebugType("ERROR", 0)
object WARNING extends DebugType("WARNING", 1)
object MESSAGE extends DebugType("MESSAGE", 2)
val values = Seq(ERROR, WARNING, MESSAGE)
}
val level: Int = DebugType.ERROR.level
def currentMethodName: String = Thread.currentThread.getStackTrace()(2).getMethodName
def currentMethodCallerName: String = Thread.currentThread.getStackTrace()(3).getMethodName
private def debugPrefix(dbg: DebugType): String = {
dbg.name + " [" + Thread.currentThread.getStackTrace()(4).getMethodName + "]: "
}
def debug(dbg: DebugType, x: Any): Unit = if(level <= dbg.level) print(debugPrefix(dbg) + x)
def debugf(dbg: DebugType, text: String, xs: Any*): Unit = if(level <= dbg.level) printf(debugPrefix(dbg) + text, xs: _*)
}
package object convertUtil {
def javaArrayListToSeq(javaArray: Any): Option[Seq[Any]] = {
javaArray match {
@@ -61,6 +93,12 @@ package object convertUtil {
package object mapUtil {
def traversableToMap[A, B](traversable: Traversable[Any], getTuple: (Any) => (A, B)): Map[A, B] = {
// Traversable[Tuple[A, B]] can be converted via .map
// .foldLeft used in favor of .map().toMap as the latter needs two traversals
// traversable.map(e => getKey(e) -> getValue(e)).toMap
traversable.foldLeft(Map[A, B]()) { (m, e) => m + getTuple(e) }
}
def javaHashMapToScalaMap[K, V](javaMap: java.util.LinkedHashMap[_, _]): Map[K, V] = {
mapAsScalaMap(javaMap).toMap.asInstanceOf[Map[K, V]]
}
@@ -118,3 +156,8 @@ package object yml {
}
}
package object simulation {
import scim.datastruct._
// maybe change to by-name parameters? have to check performance gain
type SimulationBehaviour = (Cell, Time, Option[Double], Option[FunctionArgs]) => Option[Double]
}

40
src/main/scala/scim/Main.scala
View File

@@ -1,12 +1,48 @@
package scim
import scim.components.{Parser, SimulationConfigParser, Output, Simulation}
import scim.datastruct.{Cell, FunctionArgs, SharedMemory, SimulationBehaviours, Time}
import scim.lib._
import scim.lib.simulation.SimulationBehaviour
//package scim.components._
object Main {
val func: (Cell, Time, Option[Double], Option[FunctionArgs]) => Option[Double] = (cell, time, truth, args) => { truth }
val behaviours: Map[String, SimulationBehaviour] = Map[String, SimulationBehaviour](
"factor" -> func,
"None" -> func, "addValuePercent" -> func,
"humidFromTemp" -> func,
"normalValue" -> func,
"randomValue" -> func)
val simBehavirous = SimulationBehaviours(behaviours)
def main(args: Array[String]): Unit = {
val config = "docs/wiki/config/config.yml"
val list: Map[String, Any] = yml.parse(config)
println(printUtil.indentCollection(list.toString()).toString)
new SimulationConfigParser(list, simBehavirous).parse()
// println(printUtil.indentCollection(list.toString()).toString)
//new Test2().run()
val parlist: Map[String, Int] = Map("a" -> 1, "b" -> 1, "c" -> 1, "d" -> 1)
val v: Vector[Int] = Vector(1,2,3,4,5,6,7)
//lib.evalUtil.time("par", parlist.par.map({_ => Thread.sleep(10); 2}))
//lib.evalUtil.time("ser", parlist.map({_ => Thread.sleep(10); 2}))
}
def startSimulation(): Unit = {
val config = new Parser().parse("docs/wiki/config/config.yml", simBehavirous).get
val sharedMemory = config.sharedMemory
val output = new Output(sharedMemory)
val outputThread = new Thread(output)
val simulator = new Simulation(config)
for(_ <- 0 to sharedMemory.config.runs) {
outputThread.start()
simulator.start()
outputThread.join()
}
}
}

19
src/main/scala/scim/archive/Elements.scala Normal file
View File

@@ -0,0 +1,19 @@
package scim.archive
// child elements level 2
case class Coordinate(x: Int, y: Int)
case class FunctionArgs(args: Map[String, Any])
case class FunctionId(id: Int, name: String)
case class TimeLine(start: Int, end: Option[Int])
// child elements level 1
case class Group(id: Int, name: String)
case class Block(start: Coordinate, end: Coordinate, count: Int)
case class Truth(value: Option[Double], timeline: TimeLine)
case class Function(function: FunctionId, args: Option[FunctionArgs], timeLine: TimeLine)
// root elements
case class Simulation(duration: Int, tickfactor: Int, loop: Boolean, loopcount: Int)
case class Sensor(group: Group, blocks: Vector[Block])
case class GroundTruth(group: Group, parent: Option[GroundTruth], truths: Vector[Truth], deltas: Option[Vector[Function]])
case class Behaviour(group: Group, parent: Option[Behaviour], functions: Option[Vector[Function]])

27
src/main/scala/scim/archive/GroundTruth.scala Normal file
View File

@@ -0,0 +1,27 @@
package scim.archive
case class Handler(config: Configuration, data: Data)
case class BufferHandler() {
}
case class Data() {
//val buffer = Map[Position, (BufferHandler, ArraySeq[(Time, Truth)])]()
}
//object SimBuffer {
// def apply[A](v: A*): SimBuffer[A] = new SimBuffer(v)
// def apply[A](): SimBuffer[A] = new SimBuffer[A]()
//}
//class SimBuffer[A] (buffer: ArraySeq[A], handler: BufferHandler) {
// var bufferd: Int = 1
// def get(time: Time): A = {
//
// }
//}
case class Configuration()
case class GroundTruth(handler: Handler, config: Configuration, data: Data)
case class GroundTruths(groundTruths: Map[Group, GroundTruth])

21
src/main/scala/scim/archive/Simulation.scala Normal file
View File

@@ -0,0 +1,21 @@
package scim.datastruct
import scala.collection.mutable
//case class Coordinate(x: Int, y: Int)
//case class World(size: Int) // implements trait def getPosition(coord: Coordinate): Position; def getCoordinate(pos: Position): Coordinate
//case class Position(pos: Int)
//case class Time(tick: Int)
//case class Truth(truth: Option[Double])
//case class FunctionArgs(args: Map[String, Any])
//case class Function(function: (Truth, Position, Time, FunctionArgs) => Option[Double], args: FunctionArgs)
//case class Group(id: Int, name: String)
//
//case class GroundTruths(truth: Map[Group, mutable.ArraySeq[(Time, Truth)]]) // implements trait def getTruth(group: Group, pos: Position): Option[Double]
//case class Result(truth: Map[Group, List[(Time, Truth)]]) // implements trait def getTruth(group: Group, pos: Position): Option[Double]
//case class Behaviours(function: Map[Group, List[(Time, Option[Function])]]) // implement trait def run(truth: Truth, pos: Position, time: Time): Option[Double]
//case class Sensor(group: Group, count: Int)
//case class Cells(cells: Map[Position, Vector[Sensor]])
//case class Sensors(sensors: Map[Group, Vector[(Position, Int)]]) // implement trait def sim(time: Time): Option[Double]
//case class Simulation(groundTruths: GroundTruths, behaviours: Behaviours, cells: Cells)

13
src/main/scala/scim/components/Output.scala Normal file
View File

@@ -0,0 +1,13 @@
package scim.components
import scim.datastruct.SharedMemory
class Output(sharedMemory: SharedMemory) extends Runnable {
def run(): Unit = {
outputData()
}
def outputData(): Unit = {
}
}

226
src/main/scala/scim/components/Parser.scala Normal file
View File

@@ -0,0 +1,226 @@
package scim.components
import scim.datastruct.{Behaviour, BehaviourVector, Cell, Configuration, FunctionArgs, Group, SensorCount, SharedMemory, SimulationBehaviours, SimulationConfig, Time, Truth}
import scim.lib.ConcurrentBroadcast.BroadcastObject
import scim.lib.debugUtil.{DebugType, debugf}
import scim.lib.mapUtil.traversableToMap
import scim.lib.simulation.SimulationBehaviour
import scim.lib.yml
import scala.collection.mutable
class Parser() {
def parse(file: String, simulationBehaviours: SimulationBehaviours): Option[Configuration] = {
val parser = new SimulationConfigParser(yml.parse(file), simulationBehaviours)
parser.parse()
None // FIXME: this should return something
}
}
// when creating (Group, Sensor, Option[GroundTruth]), keep ref in Map[Group, Sensor, Option[GroundTruth]] for all, to find
// root group via group.parent (while !isempty get next) (maybe tree instead)
// for cells s-> create Map[Group, Map[Cell, Count]]
// for gt g-> create Map[Group, GroundTruth]
// for behaviour b-> create Map[Group, Behaviour]
// sensors = s.map(kv => (kv._1, kv._2.map(kv2 => (kv2._1, SENSOR, TRUTH)))
// sensors.foreach(kv => kv._1
class SimulationConfigParser(input: Map[String, Any], simulationBehaviours: SimulationBehaviours) {
private val config: SimulationConfig = parseSimulationConfig(input("simulation").asInstanceOf[Map[String, Any]])
private val groups = mutable.Map[String, Group]().empty
private val sensorList = mutable.Map[Group, Map[Cell, SensorCount]]().empty
private val behaviourListSensor = mutable.Map[Group, Behaviour]().empty
private val behaviourListGroundTruth = mutable.Map[Group, Behaviour]().empty
private val groundTruthValues = mutable.Map[Time, mutable.Map[Group, Truth]]() // FIXME: are not parsed correctly
// this will throw if the map doesn't contain what it should
@throws(classOf[Exception])
def parse(): Unit = {
val sensors = input("sensors").asInstanceOf[Vector[Map[String, Any]]]
val groundTruths = input("groundtruths").asInstanceOf[Vector[Map[String, Any]]]
val behaviours = input("behaviours").asInstanceOf[Vector[Map[String, Any]]]
sensors.foreach(sensor => parseSensor(sensor))
behaviours.foreach(behaviour => parseBehaviour(behaviour))
groundTruths.foreach(groundTruth => parseGroundTruth(groundTruth))
// TODO: insert dummy sensors if none exists at cells of the parent group where this group exists
Unit
}
private def parseSimulationConfig(configInput: Map[String, Any]): SimulationConfig = {
val config = configInput.asInstanceOf[Map[String, Int]]
val duration = config("duration")
val runs = config("runs")
val tickfactor = if(config.contains("tickfactor")) config("tickfactor") else 1
val parCell = if(config.contains("parCell")) config("parCell") else 0
val parCellThreads = if(config.contains("parCellThreads")) config("parCellThreads") else 5
val parGroup = if(config.contains("parGroup")) config("parGroup") else 0
val parGroupThreads = if(config.contains("parGroupThreads")) config("parGroupThreads") else 5
val parSensor = if(config.contains("parSensor")) config("parSensor") else 0
val parSensorThreads = if(config.contains("parSensorThreads")) config("parSensorThreads") else 5
SimulationConfig(duration, runs, tickfactor, parCell, parCellThreads, parGroup, parGroupThreads, parSensor, parSensorThreads)
}
private def parseSensor(sensor: Map[String, Any]): Unit = {
val groupString = sensor("group").asInstanceOf[String]
val parentString: Option[String] = if(sensor.contains("parent")) Some(sensor("parent").asInstanceOf[String]) else None
val cells: Map[Cell, SensorCount] = parseCells(sensor("cells").asInstanceOf[Vector[Any]])
val group = Group(groupString, parentString)
if(!groups.contains(groupString)) {
groups(groupString) = group
sensorList(group) = cells
} else {
debugf(DebugType.WARNING, "Duplicate Sensors in Group [%s]. Ignoring Sensor: %s", groupString, sensor.toString())
}
}
private def parseCells(cells: Vector[Any]): Map[Cell, SensorCount] = {
val parseCell = (cell: Any) => cell match {
case cell: Vector[_] => val cellVector = cell.asInstanceOf[Vector[Int]]
(Cell(cellVector(0), cellVector(1)), SensorCount(cellVector(2)))
case cell: Map[_, _] => val cellMap = cell.asInstanceOf[Map[String, Int]]
(Cell(cellMap("x"), cellMap("y")), SensorCount(cellMap("count")))
case _ => throw new Exception("Invalid Cell: " + cell.toString)
}
traversableToMap[Cell, SensorCount](cells, parseCell)
}
private def parseTimeLineElement(timeLineElement: Any): (Int, Option[Int]) = {
timeLineElement match {
case timeLineElement: Vector[_] => val elementVector = timeLineElement.asInstanceOf[Vector[Int]]
(elementVector(1), if(timeLineElement.contains(2)) Some(elementVector(2)) else None)
case timeLineElement: Map[_, _] => val elementMap = timeLineElement.asInstanceOf[Map[String, Int]]
(elementMap("start"), if(elementMap.contains("end")) Some(elementMap("end")) else None)
}
}
private def parseFunction(functions: Vector[Any]): Behaviour = {
val parseFunctionHelper = (function: Any) => {
var funcPointer: Option[(SimulationBehaviour, Option[FunctionArgs])] = None
function match {
case function: Vector[_] => {
val start: Int = function(0).asInstanceOf[Int]
// val end: Option[Int] = None // Not used
val name: String = function(1).asInstanceOf[String]
val args: Option[FunctionArgs] = if (function.contains(3)) Some(FunctionArgs(function(2).asInstanceOf[Map[String, Any]])) else None
if (!simulationBehaviours.get(name).isEmpty) funcPointer = Some((simulationBehaviours.get(name).get, args))
Map[Int, Option[(SimulationBehaviour, Option[FunctionArgs])]](start -> funcPointer)
}
case function: Map[_, _] => {
val functionMap = function.asInstanceOf[Map[String, Any]]
val timeline: Vector[Any] = functionMap("timeline").asInstanceOf[Vector[Any]]
val timeLineVector: Vector[(Int, Option[Int])] = timeline.map(e => parseTimeLineElement(e))
val name: String = functionMap("name").asInstanceOf[String]
val args: Option[FunctionArgs] = if (functionMap.contains("args")) Some(FunctionArgs(functionMap("args").asInstanceOf[Map[String, Any]])) else None
if (!simulationBehaviours.get(name).isEmpty) funcPointer = Some((simulationBehaviours.get(name).get, args))
val output: mutable.Map[Int, Option[(SimulationBehaviour, Option[FunctionArgs])]] = mutable.Map()
timeLineVector.map(tuple => {
output(tuple._1) = funcPointer
if (!tuple._2.isEmpty) output(tuple._2.get) = None
})
output.toMap
}
case _ => throw new Exception("Invalid Function: " + function.toString)
}
}
val parsedFunction: mutable.Map[Int, Option[(SimulationBehaviour, Option[FunctionArgs])]] = mutable.Map().empty
// TODO: This will currently overwrite already assigned values; output warning
functions.map(function => parseFunctionHelper(function)).foreach(e => e.foreach(kv => parsedFunction(kv._1) = kv._2))
val timeVector: mutable.ArraySeq[Option[(SimulationBehaviour, Option[FunctionArgs])]] = new mutable.ArraySeq[Option[(SimulationBehaviour, Option[FunctionArgs])]](config.duration).map(_ => None)
val it = parsedFunction.toList.sortBy(_._1).iterator
println(parsedFunction.toList.sortBy(_._1))
var current = if(it.hasNext) Some(it.next) else None
while(!current.isEmpty) {
val next = if(it.hasNext) Some(it.next) else None
for(time <- current.get._1 to (if(!next.isEmpty) next.get._1 else config.duration - 1)) {
timeVector(time) = current.get._2
}
current = next
}
Behaviour(BehaviourVector(timeVector.toVector))
}
private def parseBehaviour(behaviourMap: Map[String, Any]): Unit = {
val groupString = behaviourMap("group").asInstanceOf[String]
// only continue parsing if sensors in this group exist
if(groups.contains(groupString)) {
val group: Group = groups(groupString)
if(!behaviourListSensor.contains(group)) {
behaviourListSensor(group) = parseFunction(behaviourMap("functions").asInstanceOf[Vector[Any]])
} else {
debugf(DebugType.WARNING, "Duplicate Sensor Behaviour in Group [%s]. Ignoring Behaviour: %s", groupString, behaviourMap.toString())
}
} else {
debugf(DebugType.WARNING, "Unused Sensor Behaviour, no sensor for group [%s] exists. Ignoring Behaviour: %s", groupString, behaviourMap.toString())
}
}
private def parseGroundTruth(groundTruthMap: Map[String, Any]): Unit = {
val groupString = groundTruthMap("group").asInstanceOf[String]
// only continue parsing if sensors in this group exist
if(groups.contains(groupString)) {
val group: Group = groups(groupString)
if(group.parent.isEmpty) {
if(!behaviourListGroundTruth.contains(group)) {
group.parent = if(groundTruthMap.contains("parent")) Some(groundTruthMap("parent").asInstanceOf[String]) else None
behaviourListGroundTruth(group) = parseFunction(groundTruthMap("delta").asInstanceOf[Vector[Any]])
} else {
debugf(DebugType.WARNING, "Duplicate Ground Truth in Group [%s]. Ignoring Ground Truth: %s", groupString, groundTruthMap.toString())
}
parseTruth(groundTruthMap("truth").asInstanceOf[Vector[Any]]).map(kv => {
if(groundTruthValues.contains(kv._1)) {
groundTruthValues(kv._1)(group) = kv._2
} else {
groundTruthValues(kv._1) = mutable.Map[Group, Truth](group -> kv._2)
}
})
} else {
debugf(DebugType.WARNING, "Unused Ground Truth in Group [%s] because the Sensor has Parent [%s]. Ignoring Ground Truth: %s", groupString, group.parent.get, groundTruthMap.toString())
}
} else {
debugf(DebugType.WARNING, "Unused Ground Truth, no sensor for group [%s] exists. Ignoring Ground Truth: %s", groupString, groundTruthMap.toString())
}
}
private def parseTruth(truthElements: Vector[Any]): Map[Time, Truth] = {
val parseTruthHelper = (truthElement: Any) => {
truthElement match {
case truthObject: Vector[_] => {
val truth: Truth = Truth(new BroadcastObject())
println(truthElement)
val start: Int = truthObject(0).asInstanceOf[Int]
// val end: Option[Int] = None // Not used
truth.set(Some(truthObject(1).asInstanceOf[Double]))
Map[Time, Truth](Time(start) -> truth)
}
case truthObject: Map[_, _] => {
val truth: Truth = Truth(new BroadcastObject())
lazy val endTruth: Truth = Truth(new BroadcastObject())
val truthObjectMap = truthObject.asInstanceOf[Map[String, Any]]
val timeline: Vector[Any] = truthObjectMap("timeline").asInstanceOf[Vector[Any]]
val timeLineVector: Vector[(Int, Option[Int])] = timeline.map(e => parseTimeLineElement(e))
truth.set(Some(truthObjectMap("value").asInstanceOf[Double]))
val output: mutable.Map[Time, Truth] = mutable.Map()
timeLineVector.map(tuple => {
output(Time(tuple._1)) = truth
if (!tuple._2.isEmpty) {
endTruth.set(None)
output(Time(tuple._2.get)) = endTruth
}
})
output.toMap
}
case _ => throw new Exception("Invalid Truth: " + truthElement.toString)
}
}
val parsedTruth: mutable.Map[Time,Truth] = mutable.Map().empty
// TODO: This will currently overwrite already assigned values; output warning
truthElements.map(truth => parseTruthHelper(truth)).foreach(e => e.foreach(kv => parsedTruth(kv._1) = kv._2))
parsedTruth.toMap
}
}

56
src/main/scala/scim/components/Simulation.scala Normal file
View File

@@ -0,0 +1,56 @@
package scim.components
import scim.datastruct.{Behaviour, BehaviourVector, Cell, Configuration, FunctionArgs, GroundTruth, Group, Sensor, Time}
import scim.lib.simulation.SimulationBehaviour
class Simulation(config: Configuration) {
private val sharedMemory = config.sharedMemory
private val cells = config.cells
val timeVector = Vector.tabulate(sharedMemory.config.duration)(t => Time(t))
def start(): Unit = {
if(sharedMemory.config.parCell != 0 && cells.size >= sharedMemory.config.parCell ) {
val cellList = cells.par
cellList.tasksupport = sharedMemory.config.cellThreadPool
timeVector.foreach(time => cellList.foreach(cell => simGroups(time, cell._1, cell._2)))
} else {
timeVector.foreach(time => cells.foreach(cell => simGroups(time, cell._1, cell._2)))
}
}
private def simGroups(time: Time, cell: Cell, groups: Map[Group, List[(Group, Sensor, Option[GroundTruth])]]): Unit = {
// TODO: possible optimisation (see text below)
// this can be further optimised as a child is only dependent on its parents ground truth and could run
// at the same time as the parent in a different thread. This would allow the child to start running as soon
// as the ground truth is calculated in the parent and doesn't have have to wait for the sensors to finish.
// Any subsequent children can start in the same manner, when their parent is done with the ground truth.
// --
// other possibility:
// run simulateGT foreach first, then run simulateSensor foreach (dont know if there is much benefit tho)
if(sharedMemory.config.parGroup != 0 && cells.size >= sharedMemory.config.parGroup ) {
val groupList = groups.par
groupList.tasksupport = sharedMemory.config.groupThreadPool
groupList.foreach(group => group._2.foreach(element => simulate(time, cell, element)))
} else {
groups.foreach(group => group._2.foreach(element => simulate(time, cell, element)))
}
}
private def simulate(time: Time, cell: Cell, element: (Group, Sensor, Option[GroundTruth])): Unit = {
val group = element._1
val sensor = element._2
val groundTruth = element._3
// simulate the ground truth if required
if(!groundTruth.isEmpty) {
groundTruth.get.simulate(cell,group, time, sharedMemory)
} else if (group.parent.isEmpty) {
// calc default ground truth if none was defined and no parent given which is None
// this ensures a value when the sensor accesses the gt and prevents a deadlock
val v: Vector[Option[(SimulationBehaviour, Option[FunctionArgs])]] = Vector.fill(sharedMemory.config.duration)(None)
GroundTruth(Some(Behaviour(BehaviourVector(v)))).simulate(cell, group, time, sharedMemory)
}
// simulate the sensor
sensor.simulate(cell, group, time, sharedMemory)
}
}

132
src/main/scala/scim/datastruct/Configuration.scala Normal file
View File

@@ -0,0 +1,132 @@
package scim.datastruct
import scim.lib.ConcurrentBroadcast.BroadcastObject
import scim.lib.simulation._
import scala.collection.parallel.ForkJoinTaskSupport
// configuration
//case class Configuration(simulation: Simulation, sensors: Vector[Sensor], groundTruths: Vector[GroundTruth], behaviours: Vector[Behaviour])
case class Time(time: Int){
def prev: Time = { Time(time - 1) }
def start: Boolean = { time == 0 }
def get: Int = { time }
}
case class SharedMemory( sensor: Map[Time, Map[SensorID, SensorResult]],
groundTruth: Map[Cell, Map[Time, Map[Group, Truth]]],
config: SimulationConfig) {
def getTruth(cell: Cell, time: Time, group: Group): Truth = {
groundTruth(cell)(time)(group)
}
def getResult(time: Time): Map[SensorID, SensorResult] = {
sensor(time)
}
}
abstract class BroadcastHandler[A](broadcastObject: BroadcastObject[A]) {
def get: A = {
broadcastObject.watch()
}
def set(newValue: A): Unit = {
broadcastObject.broadcast(newValue)
}
def exists: Boolean = {
!broadcastObject.peak().isEmpty
}
}
case class SimulationBehaviours(functions: Map[String, SimulationBehaviour]) {
def get(name: String): Option[SimulationBehaviour] = {
if(functions.contains(name)) Some(functions(name)) else None
}
}
case class SimulationConfig(duration: Int, runs: Int, tickfactor: Int = 1,
parCell: Int = 0, parCellThreads: Int = 5,
parGroup: Int = 0, parGroupThreads: Int = 5,
parSensor: Int = 0, parSensorThreads: Int = 5) {
val cellThreadPool: ForkJoinTaskSupport = new ForkJoinTaskSupport(new scala.concurrent.forkjoin.ForkJoinPool(parCellThreads))
val groupThreadPool: ForkJoinTaskSupport = new ForkJoinTaskSupport(new scala.concurrent.forkjoin.ForkJoinPool(parGroupThreads))
val sensorThreadPool: ForkJoinTaskSupport = new ForkJoinTaskSupport(new scala.concurrent.forkjoin.ForkJoinPool(parSensorThreads))
// TODO: give truth their own threadpool and parallize the calculation
}
case class SensorResult(result: BroadcastObject[(Group, Cell, Option[Double])]) extends BroadcastHandler(result)
case class SensorID(group: Group, Id: Int)
case class Duration(simDuration: Int)
case class TickFactor(factor: Int) // processed when parsing configuration, artifically place timings
case class Cell(x: Int, y: Int)
case class Group(name: String, parentIn: Option[String] = None){
// FIXME: use parent as get function and parent.set as setter to make var private
var parent: Option[String] = parentIn // have to make this mutable to reassign it during parsing
def getParentOrThis: Group = {
if(parent.isEmpty) {
this
} else {
Group(parent.get)
}
}
}
case class Truth(value: BroadcastObject[Option[Double]]) extends BroadcastHandler(value)
case class FunctionArgs(args: Map[String, Any])
// passing the group and therefore the behaviour during the simulation adds additional processing costs during runtime.
// Benefit would be fewer sensor objects.
case class BehaviourVector(function: Vector[Option[(SimulationBehaviour, Option[FunctionArgs])]],
defaultFunc: SimulationBehaviour = (_: Cell, _: Time, v: Option[Double], _: Option[FunctionArgs]) => v,
defaultArgs: Option[FunctionArgs] = None){
def get(t: Time): (SimulationBehaviour, Option[FunctionArgs]) = {
function(t.time).getOrElse((defaultFunc, defaultArgs))
}
}
case class Behaviour(behaviour: BehaviourVector){
def run(time: Time, cell: Cell, truth: Truth): Option[Double] = {
val functionTuple = behaviour.get(time)
val function = functionTuple._1
val args = functionTuple._2
// TODO: add try catch, to prevent crash in case the args are wrong
function(cell, time, truth.get, args)
}
}
case class GroundTruth(behaviour: Option[Behaviour]) {
def simulate(cell: Cell, group: Group, time: Time, sharedMemory: SharedMemory): Unit = {
val result = sharedMemory.getTruth(cell, time, group.getParentOrThis)
// only calculate a new ground truth if a behaviour is defined
if(!behaviour.isEmpty) {
val prevTruth = sharedMemory.getTruth(cell, time.prev, group)
result.set(behaviour.get.run(time, cell, prevTruth))
} else {
result.set(None)
}
}
}
case class Sensor(behaviour: Behaviour, count: SensorCount) {
def simulate(cell: Cell, group: Group, time: Time, sharedMemory: SharedMemory): Unit = {
val truth = sharedMemory.getTruth(cell, time, group.getParentOrThis)
val result = sharedMemory.getResult(time)
if(sharedMemory.config.parSensor != 0 && count.count >= sharedMemory.config.parSensor ) {
val sensorList = count.getVector.par
sensorList.tasksupport = sharedMemory.config.sensorThreadPool
sensorList.foreach(i => result(SensorID(group, i)).set((group, cell, behaviour.run(time, cell, truth))))
} else {
count.getVector.foreach(i => result(SensorID(group, i)).set((group, cell, behaviour.run(time, cell, truth))))
}
}
}
// calculate 1 full tick at a time
// a group may depend on others within a cell
// a cell does not depend on others within a group
// a sensor and ground truth are unique within a group in a cell
case class SensorCount(count: Int) {
def getVector: Vector[Int] = {
Vector.tabulate(count)(e => e)
}
}
case class Configuration(cells: Map[Cell, Map[Group, List[(Group, Sensor, Option[GroundTruth])]]], sharedMemory: SharedMemory)

229
src/main/scala/scim/lib/Concurrency.scala Normal file
View File

@@ -0,0 +1,229 @@
package scim.lib
import java.util.Calendar
import java.util.concurrent.Executors
import java.util.concurrent.locks.{Condition, Lock, ReentrantReadWriteLock}
import scala.collection.mutable
object ScalaLocks {
class ScalaReadLock(lock: Lock) {
def apply[A](read: => A): A = {
lock.lock()
try read finally {
lock.unlock()
}
}
}
class ScalaWriteLock(lock: Lock) {
def apply[A](write: () => A): Unit = {
lock.lock()
try write() finally {
lock.unlock()
}
}
}
class ScalaReadWriteLock(lock: Lock) {
def apply[A](readWrite: () => A): A = {
lock.lock()
try readWrite() finally {
lock.unlock()
}
}
}
class ScalaConditionalReadLock(readLock: ReentrantReadWriteLock.ReadLock, writeLock: ReentrantReadWriteLock.WriteLock, signal: Condition) {
private val scalaWriteLock = new ScalaWriteLock(writeLock)
// condition and read should not change any mutable state
def apply[A](condition: => Boolean, read: => A): A = {
readLock.lock() // block writer
if(!condition) {
readLock.unlock() // unblock writer
// acquire write lock, await signal and release it
scalaWriteLock (() => {
while(!condition) {
signal.await()
}
// block the writer before releasing write lock
readLock.lock()
})
}
try read finally {
readLock.unlock()
}
}
}
}
object ConcurrentBroadcast {
class BroadcastObject[A]() {
import ScalaLocks._
private val javaLock = new ReentrantReadWriteLock()
private val javaReadLock = javaLock.readLock()
private val javaWriteLock = javaLock.writeLock()
private val signal = javaWriteLock.newCondition()
private val conditionalReadLock = new ScalaConditionalReadLock(javaReadLock, javaWriteLock, signal)
private val readWriteLock = new ScalaReadWriteLock(javaWriteLock)
private val writeLock = new ScalaWriteLock(javaWriteLock)
private val readLock = new ScalaReadLock(javaReadLock)
private var value: Option[A] = None
// NON-BLOCKING: check if value has been set, write new value and return if it was overwritten
def broadcast(newValue: A): Unit = {
writeLock(() => {
value = Some(newValue)
signal.signalAll()
})
}
// NON-BLOCKING: check if value has been set, write new value and return if it was overwritten
def broadcastOverwrite(newValue: A): Boolean = {
readWriteLock(() => {
val result = value.isEmpty
value = Some(newValue)
signal.signalAll()
result
})
}
// BLOCKING: watch value, block until it is set, then return it
def watch(): A = {
conditionalReadLock(!value.isEmpty, value.get)
}
// NON-BLOCKING: returns value as Option
def peak(): Option[A] = {
readLock {
value
}
}
}
abstract class Broadcast[A](stream: mutable.ArraySeq[BroadcastObject[A]]) extends Runnable {
override def run(): Unit = {
access(stream)
}
def access(stream: mutable.ArraySeq[BroadcastObject[A]])
}
// Trait is just used to rename the function (maybe a bit redundant)
trait Watcher[A] extends Broadcast[A] {
override def access(stream: mutable.ArraySeq[BroadcastObject[A]]): Unit = { watch(stream) }
def watch(stream: mutable.ArraySeq[BroadcastObject[A]])
}
trait Broadcaster[A] extends Broadcast[A] {
override def access(stream: mutable.ArraySeq[BroadcastObject[A]]): Unit = { broadcast(stream) }
def broadcast(stream: mutable.ArraySeq[BroadcastObject[A]])
}
}
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")
}
}