Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit d6865b29 authored by Steve Elliott's avatar Steve Elliott
Browse files

[kairos] remove most internal usage of `suspend fun` 

And by proxy, all concurrency from internal graph evaluation.

This produces a large performance improvement, mostly due to observed overhead with `suspend fun`.

Flag: EXEMPT unused
Test: atest kairos-tests
Change-Id: I72d3a0a15ae4d9a143eca8d587f177fdee171a44
parent 02946ae2

packages/SystemUI/utils/kairos/src/com/android/systemui/kairos/Combinators.kt

+5 −5
Original line number Diff line number Diff line
@@ -35,14 +35,14 @@ fun <A> TFlow<Transactional<A>>.sampleTransactionals(): TFlow<A> = map { it.samp 
@ExperimentalFrpApi

fun <A, B, C> TFlow<A>.sample(

    state: TState<B>,

    transform: suspend FrpTransactionScope.(A, B) -> C,

    transform: FrpTransactionScope.(A, B) -> C,

): TFlow<C> = map { transform(it, state.sample()) }



/** @see FrpTransactionScope.sample */

@ExperimentalFrpApi

fun <A, B, C> TFlow<A>.sample(

    transactional: Transactional<B>,

    transform: suspend FrpTransactionScope.(A, B) -> C,

    transform: FrpTransactionScope.(A, B) -> C,

): TFlow<C> = map { transform(it, transactional.sample()) }



/**
@@ -57,7 +57,7 @@ fun <A, B, C> TFlow<A>.sample( 
@ExperimentalFrpApi

fun <A, B, C> TFlow<A>.samplePromptly(

    state: TState<B>,

    transform: suspend FrpTransactionScope.(A, B) -> C,

    transform: FrpTransactionScope.(A, B) -> C,

): TFlow<C> =

    sample(state) { a, b -> These.thiz<Pair<A, B>, B>(a to b) }

        .mergeWith(state.stateChanges.map { These.that(it) }) { thiz, that ->
@@ -189,7 +189,7 @@ fun interface FrpBuildMode<out A> { 
     * Invoked when this mode is enabled. Returns a value and a [TFlow] that signals a switch to a

     * new mode.

     */

    suspend fun FrpBuildScope.enableMode(): Pair<A, TFlow<FrpBuildMode<A>>>

    fun FrpBuildScope.enableMode(): Pair<A, TFlow<FrpBuildMode<A>>>

}



/**
@@ -229,7 +229,7 @@ fun interface FrpStatefulMode<out A> { 
     * Invoked when this mode is enabled. Returns a value and a [TFlow] that signals a switch to a

     * new mode.

     */

    suspend fun FrpStateScope.enableMode(): Pair<A, TFlow<FrpStatefulMode<A>>>

    fun FrpStateScope.enableMode(): Pair<A, TFlow<FrpStatefulMode<A>>>

}



/**

packages/SystemUI/utils/kairos/src/com/android/systemui/kairos/FrpBuildScope.kt

+51 −49
Original line number Diff line number Diff line
@@ -42,7 +42,7 @@ import kotlinx.coroutines.flow.scan 
import kotlinx.coroutines.launch



/** A function that modifies the FrpNetwork. */

typealias FrpSpec<A> = suspend FrpBuildScope.() -> A

typealias FrpSpec<A> = FrpBuildScope.() -> A



/**

 * Constructs an [FrpSpec]. The passed [block] will be invoked with an [FrpBuildScope] that can be
@@ -51,7 +51,7 @@ typealias FrpSpec<A> = suspend FrpBuildScope.() -> A 
 */

@ExperimentalFrpApi

@Suppress("NOTHING_TO_INLINE")

inline fun <A> frpSpec(noinline block: suspend FrpBuildScope.() -> A): FrpSpec<A> = block

inline fun <A> frpSpec(noinline block: FrpBuildScope.() -> A): FrpSpec<A> = block



/** Applies the [FrpSpec] within this [FrpBuildScope]. */

@ExperimentalFrpApi
@@ -62,12 +62,15 @@ inline operator fun <A> FrpBuildScope.invoke(block: FrpBuildScope.() -> A) = run 
@RestrictsSuspension

interface FrpBuildScope : FrpStateScope {



    /** TODO: Javadoc */

    val frpNetwork: FrpNetwork



    /** TODO: Javadoc */

    @ExperimentalFrpApi

    fun <R> deferredBuildScope(block: suspend FrpBuildScope.() -> R): FrpDeferredValue<R>

    fun <R> deferredBuildScope(block: FrpBuildScope.() -> R): FrpDeferredValue<R>



    /** TODO: Javadoc */

    @ExperimentalFrpApi fun deferredBuildScopeAction(block: suspend FrpBuildScope.() -> Unit)

    @ExperimentalFrpApi fun deferredBuildScopeAction(block: FrpBuildScope.() -> Unit)



    /**

     * Returns a [TFlow] containing the results of applying [transform] to each value of the
@@ -81,8 +84,7 @@ interface FrpBuildScope : FrpStateScope { 
     * (or a downstream) [TFlow] is observed separately, [transform] will not be invoked, and no

     * internal side-effects will occur.

     */

    @ExperimentalFrpApi

    fun <A, B> TFlow<A>.mapBuild(transform: suspend FrpBuildScope.(A) -> B): TFlow<B>

    @ExperimentalFrpApi fun <A, B> TFlow<A>.mapBuild(transform: FrpBuildScope.(A) -> B): TFlow<B>



    /**

     * Invokes [block] whenever this [TFlow] emits a value, allowing side-effects to be safely
@@ -100,7 +102,7 @@ interface FrpBuildScope : FrpStateScope { 
    @ExperimentalFrpApi

    fun <A> TFlow<A>.observe(

        coroutineContext: CoroutineContext = EmptyCoroutineContext,

        block: suspend FrpEffectScope.(A) -> Unit = {},

        block: FrpEffectScope.(A) -> Unit = {},

    ): Job



    /**
@@ -133,7 +135,8 @@ interface FrpBuildScope : FrpStateScope { 
     * tFlow { ... }.apply { observe() }

     * ```

     */

    @ExperimentalFrpApi fun <T> tFlow(builder: suspend FrpProducerScope<T>.() -> Unit): TFlow<T>

    @ExperimentalFrpApi

    fun <T> tFlow(name: String? = null, builder: suspend FrpProducerScope<T>.() -> Unit): TFlow<T>



    /**

     * Creates an instance of a [TFlow] with elements that are emitted from [builder].
@@ -197,7 +200,7 @@ interface FrpBuildScope : FrpStateScope { 
     * @see observe

     */

    @ExperimentalFrpApi

    fun <A> TFlow<A>.observeBuild(block: suspend FrpBuildScope.(A) -> Unit = {}): Job =

    fun <A> TFlow<A>.observeBuild(block: FrpBuildScope.(A) -> Unit = {}): Job =

        mapBuild(block).observe()



    /**
@@ -320,9 +323,8 @@ interface FrpBuildScope : FrpStateScope { 
     * [observers][observe] are unregistered, and any pending [effects][effect] are cancelled).

     */

    @ExperimentalFrpApi

    fun <A, B> TFlow<A>.flatMapLatestBuild(

        transform: suspend FrpBuildScope.(A) -> TFlow<B>

    ): TFlow<B> = mapCheap { frpSpec { transform(it) } }.applyLatestSpec().flatten()

    fun <A, B> TFlow<A>.flatMapLatestBuild(transform: FrpBuildScope.(A) -> TFlow<B>): TFlow<B> =

        mapCheap { frpSpec { transform(it) } }.applyLatestSpec().flatten()



    /**

     * Returns a [TState] by applying [transform] to the value held by the original [TState].
@@ -333,9 +335,8 @@ interface FrpBuildScope : FrpStateScope { 
     * cancelled).

     */

    @ExperimentalFrpApi

    fun <A, B> TState<A>.flatMapLatestBuild(

        transform: suspend FrpBuildScope.(A) -> TState<B>

    ): TState<B> = mapLatestBuild { transform(it) }.flatten()

    fun <A, B> TState<A>.flatMapLatestBuild(transform: FrpBuildScope.(A) -> TState<B>): TState<B> =

        mapLatestBuild { transform(it) }.flatten()



    /**

     * Returns a [TState] that transforms the value held inside this [TState] by applying it to the
@@ -347,7 +348,7 @@ interface FrpBuildScope : FrpStateScope { 
     * cancelled).

     */

    @ExperimentalFrpApi

    fun <A, B> TState<A>.mapLatestBuild(transform: suspend FrpBuildScope.(A) -> B): TState<B> =

    fun <A, B> TState<A>.mapLatestBuild(transform: FrpBuildScope.(A) -> B): TState<B> =

        mapCheapUnsafe { frpSpec { transform(it) } }.applyLatestSpec()



    /**
@@ -391,7 +392,7 @@ interface FrpBuildScope : FrpStateScope { 
     * cancelled).

     */

    @ExperimentalFrpApi

    fun <A, B> TFlow<A>.mapLatestBuild(transform: suspend FrpBuildScope.(A) -> B): TFlow<B> =

    fun <A, B> TFlow<A>.mapLatestBuild(transform: FrpBuildScope.(A) -> B): TFlow<B> =

        mapCheap { frpSpec { transform(it) } }.applyLatestSpec()



    /**
@@ -407,7 +408,7 @@ interface FrpBuildScope : FrpStateScope { 
    @ExperimentalFrpApi

    fun <A, B> TFlow<A>.mapLatestBuild(

        initialValue: A,

        transform: suspend FrpBuildScope.(A) -> B,

        transform: FrpBuildScope.(A) -> B,

    ): Pair<TFlow<B>, FrpDeferredValue<B>> =

        mapLatestBuildDeferred(deferredOf(initialValue), transform)
@@ -424,7 +425,7 @@ interface FrpBuildScope : FrpStateScope { 
    @ExperimentalFrpApi

    fun <A, B> TFlow<A>.mapLatestBuildDeferred(

        initialValue: FrpDeferredValue<A>,

        transform: suspend FrpBuildScope.(A) -> B,

        transform: FrpBuildScope.(A) -> B,

    ): Pair<TFlow<B>, FrpDeferredValue<B>> =

        mapCheap { frpSpec { transform(it) } }

            .applyLatestSpec(initialSpec = frpSpec { transform(initialValue.get()) })
@@ -519,12 +520,12 @@ interface FrpBuildScope : FrpStateScope { 
    fun <K, A, B> TFlow<Map<K, Maybe<A>>>.mapLatestBuildForKey(

        initialValues: FrpDeferredValue<Map<K, A>>,

        numKeys: Int? = null,

        transform: suspend FrpBuildScope.(A) -> B,

        transform: FrpBuildScope.(K, A) -> B,

    ): Pair<TFlow<Map<K, Maybe<B>>>, FrpDeferredValue<Map<K, B>>> =

        map { patch -> patch.mapValues { (_, v) -> v.map { frpSpec { transform(it) } } } }

        map { patch -> patch.mapValues { (k, v) -> v.map { frpSpec { transform(k, it) } } } }

            .applyLatestSpecForKey(

                deferredBuildScope {

                    initialValues.get().mapValues { (_, v) -> frpSpec { transform(v) } }

                    initialValues.get().mapValues { (k, v) -> frpSpec { transform(k, v) } }

                },

                numKeys = numKeys,

            )
@@ -546,7 +547,7 @@ interface FrpBuildScope : FrpStateScope { 
    fun <K, A, B> TFlow<Map<K, Maybe<A>>>.mapLatestBuildForKey(

        initialValues: Map<K, A>,

        numKeys: Int? = null,

        transform: suspend FrpBuildScope.(A) -> B,

        transform: FrpBuildScope.(K, A) -> B,

    ): Pair<TFlow<Map<K, Maybe<B>>>, FrpDeferredValue<Map<K, B>>> =

        mapLatestBuildForKey(deferredOf(initialValues), numKeys, transform)
@@ -565,7 +566,7 @@ interface FrpBuildScope : FrpStateScope { 
    @ExperimentalFrpApi

    fun <K, A, B> TFlow<Map<K, Maybe<A>>>.mapLatestBuildForKey(

        numKeys: Int? = null,

        transform: suspend FrpBuildScope.(A) -> B,

        transform: FrpBuildScope.(K, A) -> B,

    ): TFlow<Map<K, Maybe<B>>> = mapLatestBuildForKey(emptyMap(), numKeys, transform).first



    /** Returns a [Deferred] containing the next value to be emitted from this [TFlow]. */
@@ -585,7 +586,8 @@ interface FrpBuildScope : FrpStateScope { 
    }



    /** Returns a [TFlow] that emits whenever this [Flow] emits. */

    @ExperimentalFrpApi fun <A> Flow<A>.toTFlow(): TFlow<A> = tFlow { collect { emit(it) } }

    @ExperimentalFrpApi

    fun <A> Flow<A>.toTFlow(name: String? = null): TFlow<A> = tFlow(name) { collect { emit(it) } }



    /**

     * Shorthand for:
@@ -626,7 +628,7 @@ interface FrpBuildScope : FrpStateScope { 
     * cancelled).

     */

    @ExperimentalFrpApi

    fun <A> TFlow<A>.observeLatestBuild(block: suspend FrpBuildScope.(A) -> Unit = {}): Job =

    fun <A> TFlow<A>.observeLatestBuild(block: FrpBuildScope.(A) -> Unit = {}): Job =

        mapLatestBuild { block(it) }.observe()



    /**
@@ -636,7 +638,7 @@ interface FrpBuildScope : FrpStateScope { 
     * With each invocation of [block], running effects from the previous invocation are cancelled.

     */

    @ExperimentalFrpApi

    fun <A> TFlow<A>.observeLatest(block: suspend FrpEffectScope.(A) -> Unit = {}): Job {

    fun <A> TFlow<A>.observeLatest(block: FrpEffectScope.(A) -> Unit = {}): Job {

        var innerJob: Job? = null

        return observeBuild {

            innerJob?.cancel()
@@ -651,8 +653,7 @@ interface FrpBuildScope : FrpStateScope { 
     * With each invocation of [block], running effects from the previous invocation are cancelled.

     */

    @ExperimentalFrpApi

    fun <A> TState<A>.observeLatest(block: suspend FrpEffectScope.(A) -> Unit = {}): Job =

        launchScope {

    fun <A> TState<A>.observeLatest(block: FrpEffectScope.(A) -> Unit = {}): Job = launchScope {

        var innerJob = effect { block(sample()) }

        stateChanges.observeBuild {

            innerJob.cancel()
@@ -670,8 +671,7 @@ interface FrpBuildScope : FrpStateScope { 
     * [observers][observe] are unregistered, and any pending [side-effects][effect] are cancelled).

     */

    @ExperimentalFrpApi

    fun <A> TState<A>.observeLatestBuild(block: suspend FrpBuildScope.(A) -> Unit = {}): Job =

        launchScope {

    fun <A> TState<A>.observeLatestBuild(block: FrpBuildScope.(A) -> Unit = {}): Job = launchScope {

        var innerJob: Job = launchScope { block(sample()) }

        stateChanges.observeBuild {

            innerJob.cancel()
@@ -680,7 +680,7 @@ interface FrpBuildScope : FrpStateScope { 
    }



    /** Applies the [FrpSpec] within this [FrpBuildScope]. */

    @ExperimentalFrpApi suspend fun <A> FrpSpec<A>.applySpec(): A = this()

    @ExperimentalFrpApi fun <A> FrpSpec<A>.applySpec(): A = this()



    /**

     * Applies the [FrpSpec] within this [FrpBuildScope], returning the result as an
@@ -695,8 +695,7 @@ interface FrpBuildScope : FrpStateScope { 
     * [effect].

     */

    @ExperimentalFrpApi

    fun <A> TState<A>.observeBuild(block: suspend FrpBuildScope.(A) -> Unit = {}): Job =

        launchScope {

    fun <A> TState<A>.observeBuild(block: FrpBuildScope.(A) -> Unit = {}): Job = launchScope {

        block(sample())

        stateChanges.observeBuild(block)

    }
@@ -714,7 +713,7 @@ interface FrpBuildScope : FrpStateScope { 
     * otherwise, it will be invoked with the [current][sample] value.

     */

    @ExperimentalFrpApi

    fun <A> TState<A>.observe(block: suspend FrpEffectScope.(A) -> Unit = {}): Job =

    fun <A> TState<A>.observe(block: FrpEffectScope.(A) -> Unit = {}): Job =

        now.map { sample() }.mergeWith(stateChanges) { _, new -> new }.observe { block(it) }

}
@@ -753,7 +752,10 @@ fun <A> FrpBuildScope.asyncTFlow(block: suspend () -> A): TFlow<A> = 
 * ```

 */

@ExperimentalFrpApi

fun FrpBuildScope.effect(block: suspend FrpEffectScope.() -> Unit): Job = now.observe { block() }

fun FrpBuildScope.effect(

    context: CoroutineContext = EmptyCoroutineContext,

    block: FrpEffectScope.() -> Unit,

): Job = now.observe(context) { block() }



/**

 * Launches [block] in a new coroutine, returning a [Job] bound to the coroutine.

packages/SystemUI/utils/kairos/src/com/android/systemui/kairos/FrpNetwork.kt

+18 −5
Original line number Diff line number Diff line
@@ -23,7 +23,6 @@ import com.android.systemui.kairos.internal.util.awaitCancellationAndThen 
import com.android.systemui.kairos.internal.util.childScope

import kotlin.coroutines.CoroutineContext

import kotlin.coroutines.EmptyCoroutineContext

import kotlin.coroutines.coroutineContext

import kotlinx.coroutines.CoroutineName

import kotlinx.coroutines.CoroutineScope

import kotlinx.coroutines.Job
@@ -53,7 +52,7 @@ interface FrpNetwork { 
     * If the network is cancelled while the caller of [transact] is suspended, then the call will

     * be cancelled.

     */

    @ExperimentalFrpApi suspend fun <R> transact(block: suspend FrpTransactionScope.() -> R): R

    @ExperimentalFrpApi suspend fun <R> transact(block: FrpTransactionScope.() -> R): R



    /**

     * Activates [spec] in a transaction, suspending indefinitely. While suspended, all observers
@@ -133,22 +132,36 @@ internal class LocalFrpNetwork( 
    private val scope: CoroutineScope,

    private val endSignal: TFlow<Any>,

) : FrpNetwork {

    override suspend fun <R> transact(block: suspend FrpTransactionScope.() -> R): R =

    override suspend fun <R> transact(block: FrpTransactionScope.() -> R): R =

        network.transaction("FrpNetwork.transact") { runInTransactionScope { block() } }.await()



    override suspend fun activateSpec(spec: FrpSpec<*>) {

        val stopEmitter =

            CoalescingMutableTFlow(

                name = "activateSpec",

                coalesce = { _, _: Unit -> },

                network = network,

                getInitialValue = {},

            )

        val job =

            network

                .transaction("FrpNetwork.activateSpec") {

                    val buildScope =

                        BuildScopeImpl(

                            stateScope = StateScopeImpl(evalScope = this, endSignal = endSignal),

                            stateScope =

                                StateScopeImpl(

                                    evalScope = this,

                                    endSignal = mergeLeft(stopEmitter, endSignal),

                                ),

                            coroutineScope = scope,

                        )

                    buildScope.runInBuildScope { launchScope(spec) }

                }

                .await()

        awaitCancellationAndThen { job.cancel() }

        awaitCancellationAndThen {

            stopEmitter.emit(Unit)

            job.cancel()

        }

    }



    override fun <In, Out> coalescingMutableTFlow(

packages/SystemUI/utils/kairos/src/com/android/systemui/kairos/FrpScope.kt

+5 −18
Original line number Diff line number Diff line
@@ -16,13 +16,8 @@ 


package com.android.systemui.kairos



import com.android.systemui.kairos.internal.CompletableLazy

import kotlin.coroutines.RestrictsSuspension

import kotlin.coroutines.resume

import kotlin.coroutines.resumeWithException

import kotlinx.coroutines.CompletableDeferred

import kotlinx.coroutines.Deferred

import kotlinx.coroutines.ExperimentalCoroutinesApi

import kotlinx.coroutines.suspendCancellableCoroutine



/** Denotes [FrpScope] interfaces as [DSL markers][DslMarker]. */

@DslMarker annotation class FrpScopeMarker
@@ -37,13 +32,7 @@ interface FrpScope { 
    /**

     * Returns the value held by the [FrpDeferredValue], suspending until available if necessary.

     */

    @ExperimentalFrpApi

    @OptIn(ExperimentalCoroutinesApi::class)

    suspend fun <A> FrpDeferredValue<A>.get(): A = suspendCancellableCoroutine { k ->

        unwrapped.invokeOnCompletion { ex ->

            ex?.let { k.resumeWithException(ex) } ?: k.resume(unwrapped.getCompleted())

        }

    }

    @ExperimentalFrpApi fun <A> FrpDeferredValue<A>.get(): A = unwrapped.value

}



/**
@@ -53,7 +42,7 @@ interface FrpScope { 
 * @see FrpScope.get

 */

@ExperimentalFrpApi

class FrpDeferredValue<out A> internal constructor(internal val unwrapped: Deferred<A>)

class FrpDeferredValue<out A> internal constructor(internal val unwrapped: Lazy<A>)



/**

 * Returns the value held by this [FrpDeferredValue], or throws [IllegalStateException] if it is not
@@ -64,10 +53,8 @@ class FrpDeferredValue<out A> internal constructor(internal val unwrapped: Defer 
 *

 * @see FrpScope.get

 */

@ExperimentalFrpApi

@OptIn(ExperimentalCoroutinesApi::class)

fun <A> FrpDeferredValue<A>.getUnsafe(): A = unwrapped.getCompleted()

@ExperimentalFrpApi fun <A> FrpDeferredValue<A>.getUnsafe(): A = unwrapped.value



/** Returns an already-available [FrpDeferredValue] containing [value]. */

@ExperimentalFrpApi

fun <A> deferredOf(value: A): FrpDeferredValue<A> = FrpDeferredValue(CompletableDeferred(value))
 
fun <A> deferredOf(value: A): FrpDeferredValue<A> = FrpDeferredValue(CompletableLazy(value))

packages/SystemUI/utils/kairos/src/com/android/systemui/kairos/FrpStateScope.kt

+53 −34

File changed.

Preview size limit exceeded, changes collapsed.