Compare commits
No commits in common. "becde944032c84074ec0b940680149988a2ac09b" and "89d78c43bbc148d67bb61025502a6a085f8a68c5" have entirely different histories.
becde94403
...
89d78c43bb
@ -52,7 +52,6 @@ dependencyResolutionManagement {
|
||||
}
|
||||
|
||||
include(
|
||||
":simulation-kt",
|
||||
":controls-core",
|
||||
":controls-ports-ktor",
|
||||
":controls-serial",
|
||||
|
||||
@ -1,32 +0,0 @@
|
||||
import space.kscience.gradle.Maturity
|
||||
|
||||
plugins {
|
||||
id("space.kscience.gradle.mpp")
|
||||
`maven-publish`
|
||||
}
|
||||
|
||||
description = """
|
||||
Core interfaces for building a device server
|
||||
""".trimIndent()
|
||||
|
||||
kscience {
|
||||
jvm()
|
||||
js()
|
||||
native()
|
||||
wasm()
|
||||
useCoroutines()
|
||||
useContextReceivers()
|
||||
|
||||
commonMain {
|
||||
api(spclibs.kotlinx.datetime)
|
||||
}
|
||||
|
||||
jvmTest{
|
||||
implementation(spclibs.logback.classic)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
readme{
|
||||
maturity = Maturity.EXPERIMENTAL
|
||||
}
|
||||
@ -1,132 +0,0 @@
|
||||
package space.kscience.simulation
|
||||
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.channels.BufferOverflow
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.*
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.datetime.Instant
|
||||
import kotlin.time.Duration
|
||||
|
||||
/**
|
||||
* @param lookaheadInterval an interval for generated events ahead of the last observed event.
|
||||
*/
|
||||
public class GeneratingTimeline<E : TimelineEvent>(
|
||||
private val generationScope: CoroutineScope,
|
||||
private val initialEvent: E,
|
||||
private val lookaheadInterval: Duration,
|
||||
private val generatorChain: suspend (E) -> E
|
||||
) : Timeline<E>, AutoCloseable {
|
||||
|
||||
// push to this channel to trigger event generation
|
||||
private val wakeupChannel = Channel<Unit>(onBufferOverflow = BufferOverflow.DROP_OLDEST)
|
||||
|
||||
private suspend fun kickGenerator() {
|
||||
wakeupChannel.send(Unit)
|
||||
}
|
||||
|
||||
private val mutex = Mutex()
|
||||
|
||||
private val history = ArrayDeque<E>()
|
||||
|
||||
private val lastEvent = MutableSharedFlow<E>(replay = Int.MAX_VALUE)
|
||||
|
||||
private val updateHistoryJob = generationScope.launch {
|
||||
lastEvent.onEach {
|
||||
mutex.withLock {
|
||||
history.add(it)
|
||||
//cleanup old events
|
||||
val threshold = observedTime ?: return@withLock
|
||||
while (history.isNotEmpty() && history.last().time > threshold) {
|
||||
history.removeFirst()
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private val observers: MutableSet<TimelineObserver> = mutableSetOf()
|
||||
|
||||
override val time: Instant
|
||||
get() = history.lastOrNull()?.time ?: initialEvent.time
|
||||
|
||||
override val observedTime: Instant?
|
||||
get() = observers.minOfNotNullOrNull { it.time }
|
||||
|
||||
override fun flowUnobservedEvents(): Flow<E> = flow {
|
||||
history.forEach { e ->
|
||||
emit(e)
|
||||
}
|
||||
emitAll(lastEvent)
|
||||
}
|
||||
|
||||
override suspend fun advance(toTime: Instant) {
|
||||
observers.forEach {
|
||||
it.collect(toTime)
|
||||
}
|
||||
}
|
||||
|
||||
private var generatorJob: Job = launchGenerator(initialEvent)
|
||||
|
||||
private fun launchGenerator(event: E): Job = generationScope.launch {
|
||||
kickGenerator()
|
||||
var currentEvent = event
|
||||
// for each wakeup generate all events in lookaheadInterval
|
||||
for (u in wakeupChannel) {
|
||||
while (currentEvent.time < (observedTime ?: event.time) + lookaheadInterval) {
|
||||
val nextEvent = generatorChain(currentEvent)
|
||||
lastEvent.emit(nextEvent)
|
||||
currentEvent = nextEvent
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public suspend fun interrupt(newStart: E) {
|
||||
check(newStart.time > (observedTime ?: Instant.DISTANT_FUTURE)) {
|
||||
"Can't interrupt generating timeline after observed event"
|
||||
}
|
||||
mutex.withLock {
|
||||
while (history.isNotEmpty() && history.last().time > newStart.time) {
|
||||
history.removeLast()
|
||||
}
|
||||
generatorJob.cancel()
|
||||
generatorJob = launchGenerator(newStart)
|
||||
|
||||
}
|
||||
kickGenerator()
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
updateHistoryJob.cancel()
|
||||
generatorJob.cancel()
|
||||
}
|
||||
|
||||
override suspend fun observe(collector: suspend Flow<E>.() -> Unit): TimelineObserver {
|
||||
val observer = object : TimelineObserver {
|
||||
override var time: Instant = this@GeneratingTimeline.time
|
||||
|
||||
override suspend fun collect(upTo: Instant) {
|
||||
flowUnobservedEvents().takeWhile {
|
||||
it.time <= upTo
|
||||
}.onEach {
|
||||
time = it.time
|
||||
kickGenerator()
|
||||
}.collector()
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
observers.remove(this)
|
||||
if(observers.isEmpty()){
|
||||
this@GeneratingTimeline.close()
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
observers.add(observer)
|
||||
return observer
|
||||
}
|
||||
}
|
||||
@ -1,51 +0,0 @@
|
||||
package space.kscience.simulation
|
||||
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.merge
|
||||
import kotlinx.coroutines.flow.onEach
|
||||
import kotlinx.coroutines.flow.takeWhile
|
||||
import kotlinx.datetime.Instant
|
||||
|
||||
|
||||
public class MergedTimeline<E : TimelineEvent>(
|
||||
private val timelines: List<Timeline<E>>
|
||||
) : Timeline<E> {
|
||||
override val time: Instant
|
||||
get() = timelines.minOfNotNullOrNull { it.time } ?: Instant.DISTANT_PAST
|
||||
|
||||
override val observedTime: Instant?
|
||||
get() = timelines.maxOfNotNullOrNull { it.observedTime }
|
||||
|
||||
override fun flowUnobservedEvents(): Flow<E> = timelines.map { flowUnobservedEvents() }.merge()
|
||||
|
||||
override suspend fun advance(toTime: Instant) {
|
||||
timelines.forEach { it.advance(toTime) }
|
||||
}
|
||||
|
||||
// override suspend fun interrupt(atTime: Instant) {
|
||||
// timelines.forEach { it.interrupt(atTime) }
|
||||
// }
|
||||
|
||||
private val observers: MutableSet<TimelineObserver> = mutableSetOf()
|
||||
|
||||
override suspend fun observe(collector: suspend Flow<E>.() -> Unit): TimelineObserver {
|
||||
val observer = object : TimelineObserver {
|
||||
override var time: Instant = this@MergedTimeline.time
|
||||
|
||||
override suspend fun collect(upTo: Instant) = timelines
|
||||
.map { flowUnobservedEvents() }
|
||||
.merge()
|
||||
.takeWhile { it.time <= upTo }.onEach {
|
||||
time = it.time
|
||||
}.collector()
|
||||
|
||||
|
||||
override fun close() {
|
||||
observers.remove(this)
|
||||
}
|
||||
|
||||
}
|
||||
observers.add(observer)
|
||||
return observer
|
||||
}
|
||||
}
|
||||
@ -1,88 +0,0 @@
|
||||
package space.kscience.simulation
|
||||
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.asFlow
|
||||
import kotlinx.coroutines.flow.onEach
|
||||
import kotlinx.coroutines.flow.takeWhile
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.datetime.Instant
|
||||
|
||||
/**
|
||||
* A manually mutable [Timeline] that could be modified via [emit] method by multiple
|
||||
*/
|
||||
public class SharedTimeline<E : TimelineEvent> : Timeline<E> {
|
||||
|
||||
private val mutex = Mutex()
|
||||
|
||||
private val events = ArrayDeque<E>()
|
||||
|
||||
private val observers: MutableSet<TimelineObserver> = mutableSetOf()
|
||||
|
||||
override val time: Instant
|
||||
get() = events.lastOrNull()?.time ?: Instant.DISTANT_PAST
|
||||
|
||||
override val observedTime: Instant?
|
||||
get() = observers.minOfNotNullOrNull { it.time }
|
||||
|
||||
override fun flowUnobservedEvents(): Flow<E> = events.asFlow()
|
||||
|
||||
/**
|
||||
* Emit new event to the timeline
|
||||
*/
|
||||
public suspend fun emit(event: E): Boolean = mutex.withLock {
|
||||
if (event.time < (observedTime ?: Instant.DISTANT_PAST)) {
|
||||
error("Can't emit event $event because there are observed events after $observedTime")
|
||||
}
|
||||
events.add(event)
|
||||
}
|
||||
|
||||
override suspend fun advance(toTime: Instant) {
|
||||
observers.forEach {
|
||||
it.collect(toTime)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Discard all events before [observedTime]
|
||||
*/
|
||||
private suspend fun cleanup(): Unit = mutex.withLock {
|
||||
val threshold = observedTime ?: return@withLock
|
||||
while (events.isNotEmpty() && events.last().time > threshold) {
|
||||
events.removeFirst()
|
||||
}
|
||||
}
|
||||
|
||||
// /**
|
||||
// * Discard unconsumed events after [atTime].
|
||||
// */
|
||||
// override suspend fun interrupt(atTime: Instant): Unit = mutex.withLock {
|
||||
// val threshold = observedTime
|
||||
// if (atTime < threshold)
|
||||
// error("Timeline interrupt at time $atTime is not possible because there are observed events before $threshold")
|
||||
// while (events.isNotEmpty() && events.last().time > atTime) {
|
||||
// events.removeLast()
|
||||
// }
|
||||
// }
|
||||
|
||||
override suspend fun observe(collector: suspend Flow<E>.() -> Unit): TimelineObserver {
|
||||
val observer = object : TimelineObserver {
|
||||
val observerMutex = Mutex()
|
||||
override var time: Instant = this@SharedTimeline.time
|
||||
|
||||
override suspend fun collect(upTo: Instant) = observerMutex.withLock {
|
||||
flowUnobservedEvents().takeWhile { it.time <= upTo }.onEach {
|
||||
time = it.time
|
||||
}.collector()
|
||||
cleanup()
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
observers.remove(this)
|
||||
}
|
||||
|
||||
}
|
||||
observers.add(observer)
|
||||
return observer
|
||||
}
|
||||
}
|
||||
@ -1,87 +0,0 @@
|
||||
package space.kscience.simulation
|
||||
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.datetime.Instant
|
||||
import kotlin.time.Duration
|
||||
|
||||
|
||||
public interface TimelineEvent {
|
||||
public val time: Instant
|
||||
}
|
||||
|
||||
public interface TimelineInterval : TimelineEvent {
|
||||
public val startTime: Instant
|
||||
public val duration: Duration
|
||||
|
||||
override val time: Instant
|
||||
get() = startTime + duration
|
||||
}
|
||||
|
||||
public data class SimpleTimelineEvent<T>(override val time: Instant, val value: T) : TimelineEvent
|
||||
|
||||
public interface TimelineObserver : AutoCloseable {
|
||||
/**
|
||||
* The subjective time of this observer
|
||||
*/
|
||||
public val time: Instant
|
||||
|
||||
/**
|
||||
* Collect all uncollected events from [time] to [upTo].
|
||||
*
|
||||
* By default, collects all events.
|
||||
*/
|
||||
public suspend fun collect(upTo: Instant = Instant.DISTANT_FUTURE)
|
||||
}
|
||||
|
||||
/**
|
||||
* A time-ordered sequence of events of type [E]. There time of events is strictly monotonic, meaning that the time of
|
||||
* the next event is greater than the previous event time.
|
||||
*
|
||||
* Timeline guarantees that all collectors could read all events when they need. Meaning that all unread events are cached.
|
||||
*
|
||||
* Timeline guarantees that already read events won't change, but unread events could change.
|
||||
*/
|
||||
public interface Timeline<E : TimelineEvent> {
|
||||
/**
|
||||
* A subjective time of this timeline. The time could advance without events being produced.
|
||||
*/
|
||||
public val time: Instant
|
||||
|
||||
/**
|
||||
* The time of the last event that was observed by all observers
|
||||
*/
|
||||
public val observedTime: Instant?
|
||||
|
||||
/**
|
||||
* Flow events from [observedTime] to [time].
|
||||
*
|
||||
* The resulting flow is finite and should not suspend.
|
||||
*
|
||||
* This method does not affect [observedTime].
|
||||
*/
|
||||
public fun flowUnobservedEvents(): Flow<E>
|
||||
|
||||
/**
|
||||
* Attach observer to this [Timeline]. The observer collection is not triggered right away, but only on demand.
|
||||
*
|
||||
* Each collection shifts [TimelineObserver.time] for this observer.
|
||||
* The value of [observedTime] is the least of all observers [TimelineObserver.time].
|
||||
*/
|
||||
public suspend fun observe(
|
||||
collector: suspend Flow<E>.() -> Unit
|
||||
): TimelineObserver
|
||||
|
||||
/**
|
||||
* Advance simulation time to [toTime]. This method forces all observers to collect all events in the given range.
|
||||
*
|
||||
* This method suspends until all advancement is done
|
||||
*/
|
||||
public suspend fun advance(toTime: Instant)
|
||||
|
||||
// /**
|
||||
// * Interrupt generation of this timeline and discard unconsumed events after [atTime].
|
||||
// *
|
||||
// * Throw exception if at least one observer advanced
|
||||
// */
|
||||
// public suspend fun interrupt(atTime: Instant): Unit
|
||||
}
|
||||
@ -1,35 +0,0 @@
|
||||
package space.kscience.simulation
|
||||
|
||||
internal inline fun <T, R : Comparable<R>> Iterable<T>.minOfNotNullOrNull(selector: (T) -> R?): R? {
|
||||
val iterator = iterator()
|
||||
if (!iterator.hasNext()) return null
|
||||
var minValue = selector(iterator.next())
|
||||
while (iterator.hasNext()) {
|
||||
val v = selector(iterator.next())
|
||||
when {
|
||||
minValue == null -> minValue = v
|
||||
v == null -> {/*do nothing*/}
|
||||
minValue > v -> {
|
||||
minValue = v
|
||||
}
|
||||
}
|
||||
}
|
||||
return minValue
|
||||
}
|
||||
|
||||
internal inline fun <T, R : Comparable<R>> Iterable<T>.maxOfNotNullOrNull(selector: (T) -> R?): R? {
|
||||
val iterator = iterator()
|
||||
if (!iterator.hasNext()) return null
|
||||
var maxValue = selector(iterator.next())
|
||||
while (iterator.hasNext()) {
|
||||
val v = selector(iterator.next())
|
||||
when {
|
||||
maxValue == null -> maxValue = v
|
||||
v == null -> {/*do nothing*/}
|
||||
maxValue < v -> {
|
||||
maxValue = v
|
||||
}
|
||||
}
|
||||
}
|
||||
return maxValue
|
||||
}
|
||||
@ -1,34 +0,0 @@
|
||||
package space.kscience.simulation
|
||||
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.datetime.Instant
|
||||
import kotlin.test.Test
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
||||
class TimelineTests {
|
||||
|
||||
|
||||
@Test
|
||||
fun testGeneration() = runTest(timeout = 5.seconds) {
|
||||
val startTime = Instant.parse("2020-01-01T00:00:00.000Z")
|
||||
|
||||
val generation = GeneratingTimeline<SimpleTimelineEvent<DoubleArray>>(
|
||||
this,
|
||||
initialEvent = SimpleTimelineEvent(startTime, List(10) { it.toDouble() }.toDoubleArray()),
|
||||
lookaheadInterval = 1.seconds
|
||||
) { event ->
|
||||
val time = event.time + 0.1.seconds
|
||||
println("Emit: $time")
|
||||
SimpleTimelineEvent(time, event.value.map { it + 1.0 }.toDoubleArray())
|
||||
}
|
||||
|
||||
val collector = generation.observe {
|
||||
collect {
|
||||
println("Consume: ${it.time}")
|
||||
}
|
||||
}
|
||||
|
||||
collector.collect(startTime + 2.seconds)
|
||||
collector.close()
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user