GeneratingTimeLine prototype is working

settings.gradle.kts

@@ -52,6 +52,7 @@ dependencyResolutionManagement {
 }
 
 include(
+    ":simulation-kt",
     ":controls-core",
     ":controls-ports-ktor",
     ":controls-serial",

simulation-kt/build.gradle.kts

@@ -0,0 +1,32 @@
+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
+}

simulation-kt/src/commonMain/kotlin/GeneratingTimeline.kt

@@ -0,0 +1,132 @@
+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
+    }
+}

simulation-kt/src/commonMain/kotlin/MergedTimeline.kt

@@ -0,0 +1,51 @@
+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
+    }
+}

simulation-kt/src/commonMain/kotlin/SharedTimeline.kt

@@ -0,0 +1,88 @@
+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
+    }
+}

simulation-kt/src/commonMain/kotlin/Timeline.kt

@@ -0,0 +1,87 @@
+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
+}

simulation-kt/src/commonMain/kotlin/notNullUtils.kt

@@ -0,0 +1,35 @@
+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
+}

simulation-kt/src/commonTest/kotlin/TimelineTests.kt

@@ -0,0 +1,34 @@
+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()
+    }
+}