Writing complex actors

[fvasco]

I need to create a complex actor using some private methods to update the internal state, but I cannot find any useful base class to implement ActorJob.

What is the best practice in such case?

[elizarov]

I'd do something along these lines:

class MyActor {
    // your private state here
    suspend fun onReceive(msg: MyMsg) {
        // ... your code here ...
    }
}

fun myActorJob(): ActorJob<MyMsg> = actor(CommonPool) {
    with(MyActor()) {
        for (msg in channel) onReceive(msg)
    }
}

You can also pass actor's channel to MyActor constructor if it needs access to it.

[fvasco]

Hi @elizarov,
hi solved this issue in the same way, but an helper class looks like a miss of abstraction.

I want propose to introduce a dedicated interface for actor, something like:

interface Actor : ActorScope {
    fun invoke() : Unit
}

and the new function

fun launch(context: CoroutineContext, ..., actorBuild: (CoroutineContext) -> Actor) : ActorJob

I want to propose a draft implementation in the next week, what do you think about it?

[elizarov]

The key design consideration here is how a typical actor's code is going to look like. It is not clear in your example. The goal should be to minimize the amount of boilerplate that users of this library code are forced to write. I was thinking about something along the following lines:

abstract class Actor<T>(scope: ActorScope<T>): ActorScope<T> by scope {
    abstract suspend fun onReceive(msg: T)
}

An abstract class lets us share some common state among all actor implementations (in this case, the fact that ActorScope is provided to all the functions defined inside an actor impl). It also makes sense to share "for messages" loop, as every typical actor is going to have one, and leave only onReceive for override.

With this bases class, the specific actor implementation looks like this:

class MyActor(scope: ActorScope<MyMsg>): Actor<MyMsg>(scope) {
    // your private state here
    override suspend fun onReceive(msg: MyMsg) {
        // ... your code here ...
    }
}

There is still some boiler-plate (MyMsg and scope are mentioned twice), but it is not clear how to further reduce it.

Now, you can define a generic actorOf function to start any kind of actor:

fun <T> actorOf(
    context: CoroutineContext,
    init: (ActorScope<T>) -> Actor<T>
): ActorJob<T> = actor<T>(context) {
        val instance = init(this@actor)
        for (msg in channel) instance.onReceive(msg)
    }

Using this function you can start your custom actor with a natural-looking and easy-to-read invocation actorOf(CommonPool, ::MyActor) with ability to further customize construction of your actor impl by providing an explicit lambda for an init block.

[fvasco]

onReceive method is comfortable to use, but I want to expose you my concerns.

The actor function uses the iterable pattern, instead the Actor class use visitor pattern. Personally I consider the iterator pattern a better choice becouse it is possible to detect actor termination (end of iteration).

Using the explicit itaration in Actor makes it cut-and-paste compatible with the actor function.

I suggest to mantain Actor as public interface and to provide a comfortable implementation

public abstract class ActorBase<E>(actorScope: ActorScope<E>) : Actor<E>, ActorScope<E> by actorScope {

    override suspend fun invoke() {
        for (message in channel)
            onReceive(message)
    }

    public abstract suspend fun onReceive(message: E)
}

Finally actorOf misses of some parameters, we should keep this one, an extended version:

public fun <E> actorOf(
        context: CoroutineContext,
        capacity: Int = 0,
        start: CoroutineStart = CoroutineStart.DEFAULT,
        builder: (ActorScope<E>) -> Actor<E>
): ActorJob<E> = actor(context, capacity, start) {
    val actor = builder(this)
    actor.invoke()
}

or both?

[elizarov]

Makes sense. I'd keep an extended version. Maybe rename actor.invoke() to actor.run() (similar to Thread's run method). I'm not sure that an actor base class should be called ActorBase. I'd either keep it an Actor to be short or name it AbstractActor.

[elizarov]

Let me voice my concern that adding such functionality to the core is a start of a slippery road of turning it in into an actor-based programming framework. There are lots of issues to be addressed in a large-scale actor-based programming and the corresponding support libraries are only bound to grow over time. Maybe we should think-through what other things needs to be added, beyond an actor base-class and a construction function, and have a separate module to support all of that.

[elizarov]

Also, if we are starting on a path of generic actor-based programming, then we are inevitably going to be compared with other actor-based frameworks, including comparisons on the basis of performance. It does not mean that we have to worry about performance up-front, but we have to keep our designs, at least, optimizable in the future.

That is my concern for overridable invoke/run method and other potential design decisions. Let me elaborate. A generic actor can, for example, start multiple concurrent actors to process a single mailbox concurrently if the actor's logic is stateless by itself. However, a stateful actor is must typically process its mailbox in serial fashion, e.g. it is a single consumer of its mailbox.

Current implementation of Channel is multi-consumer/multi-producer queue which makes it widely-applicable (it can do both use-cases), but it inevitably costs some performance. One might implement an optimized version of channel for single-producer case to support stateful-actor use-case with better performance, but then we have to ensure that it is not possible for an actor to accidentally launch additional coroutines to read from its mailbox (to make the design less error-prone and more fool-proof). It means that we should actually hide actor's ReceiveChannel from the actor itself.

[fvasco]

At this time of development, a stateless concurrent actor can be implemented using a BroadcastChannel to create a different ActorScope for each worker actor.

However your concern sounds right to me, but I suggest to mantain an uniformity of actor function and Actor class

public interface Actor<in E> : CoroutineScope {
    public suspend fun onReceive(message: E)
}

public fun <E> actor(
        context: CoroutineContext,
        capacity: Int = 0,
        start: CoroutineStart = CoroutineStart.DEFAULT,
        onReceive: suspend CoroutineScope.(E) -> Unit
): ActorJob<E>

In such case we deprecate ActorScope in favour of Actor.

[elizarov]

The idea of concurrent stateless actor is to have multiple instance working on the same mailbox in round-robin fashion as opposed to broadcasting messages to all actors. I actually considered adding an additional concurrency: Int = 1 optional parameter to actor builder to cover this use-case, so that you can launch multiple actors under the umbrella of a single ActorJob. However, it is not that really straight-forward as it does require us to start dabbling into supervision strategies, because multiple running coroutines do make it urgent to figure out what you are going to do when one of them crashes.

[fvasco]

Yes obviously, I was wrong.

As you proposed above I consider preferable a different module for actor supervisor and so on.
In my opinion this module should offer a common pool of interfaces for asynchronous programming, an "actor" module should implement more complex use cases, but I hope that the "core" module can interoperate well with existent solutions like Quasar or Akka.

So in my limited scope of view this "core" module shoud propose an "Actor" as a "Producer" dual, all other implementation should extends core's interfaces and provides its own builders.

[elizarov]

Let me record here a draft design that I current have with respect to complex actors. The idea is to radically reduce boiler-plate that is required for writing actors that accept multiple message types by providing a handy facility that totally avoids the need to write sealed class for messages by following this pattern:

class MyActor : Actor() {
    // actor state is here, must be private

    init {
        // optional state initialization code
    }

    // public actor operation without result
    suspend fun operationA(someParams...) = act {
        // logic here
    }

    // public actor operation with result
    suspend fun operationB(otherParams...): Result = actAndReply {
        // logic here
    }

    private fun helper() { /* regular code, can access state */ }
}

The idea is that all public functions defined on actor must be suspend and must invoke a special function (tentatively called act here) that captures "message parameters" (function parameters) and stores the message in the actor's inbox for processing. When response to the caller is needed, then a separate function (tentatively called actAndReply here) is used.

We can have IDE inspections to help avoid "sharing" pitfalls to verify that all of the actor's state is private and all the public functions are properly written using act pattern. IDE inspection can also help to ensure that only immutable types are passed in/out of actor or the proper defensive copies are made on mutable data (the latter is hard to ensure, though, without complicated reference capability annotations akin to the ones used in Pony language).

The reference to MyActor instance serves as a type-safe reference to an actor and should implement a Job interface. The actor is started lazily on the first invocation of any of its public (acting) functions or by explicitly invoking start(). Lazy start feature prevents races when systems of communicating actors are written like this:

class MySystem { 
    private val actorA = object : Actor() {
         // somewhere sends message to actorB
    }

    private val actorB = object : Actor() {
        // somewhere sends message to actorA
    }

    suspend fun facadeOperation() { actorA.doSomethig() }
}

We shall also consider changing default to CoroutineStart.LAZY for a simple actor { ... } builder, too.

[fvasco]

Hi @elizarov
I wish expose some consideration about your -interesting- proposal.

Actor, as intended until now, is the dual of Producer, so your design does not help to write complex actor.

Your proposal looks like a fully synchronized, non blocking class, which is equally interesting.
I wrote a dummy implementation using your requirement plus one: avoid abstract superclass; so I wrote an task synchronization helper (TaskQueue here) and a case of use.

class TaskQueue(
        val context: CoroutineContext = DefaultDispatcher,
        val mutex: Mutex = Mutex(),
        lazyInit: (suspend CoroutineScope.() -> Unit)? = null
) {

    private var lazyInit: Deferred<Unit>?

    init {
        this.lazyInit = lazyInit?.let {
            async(context, start = CoroutineStart.LAZY, block = it)
        }
    }

    /**
     * Force lazy initialization
     */
    suspend fun init() {
        lazyInit?.run {
            await()
            lazyInit = null
        }
    }

    suspend operator fun <T> invoke(block: () -> T): T {
        init()
        return mutex.withLock(this) {
            withContext(context) {
                block()
            }
        }
    }
}

class HttpSession {

    val start = Instant.now()

    private lateinit var state: MutableMap<String, String>

    private val taskQueue = TaskQueue {
        state = mutableMapOf()
    }

    suspend fun get(key: String) = taskQueue {
        state[key]
    }

    suspend fun set(key: String, value: String) {
        taskQueue {
            state[key] = value
        }
    }
}

Plus: using this implementation and issue #94 makes easy to implement a read task queue and a write task queue.

[elizarov]

@fvasco Indeed, it does look like a "like a fully synchronized, non blocking class", but it is not one. There are lots of similarities between monitor-based synchronization (like synchronized methods in Java) and actor-based programming model (like behavior functions in Pony). But there are important differences, too. Let me cite Pony documentation here:

A behaviour is like a function, except that functions are synchronous and behaviours are asynchronous.

Let's take a look at it in the context of Kotlin. First of all, we don't need new primitive for a "fully synchronized class". We already have Mutex. So, a fully synchronized class can be written like this:

class HttpSessionSync {
    private val mutex = Mutex()
    // state initialization does not have to happen under lock
    private var state: MutableMap<String, String> = mutableMapOf()
   
    suspend fun set(key: String, value: String) = mutex.withLock {
        state[key] = value
    }

    // etc
}

Notice the boilerplate here. We have to define private val mutex = Mutex() every time we use this pattern, so some kind of out-of-the box Mutex abstract base class might help. Shall we make Mutex open? Anyway, we don't want to promote this pattern, so we will not further discuss it in this thread.

You've made an interesting observation that requiring base class for complex actors is not good idea, so while, IMHO, we should give an option of using one, it should not be a requirement. Let's sketch implementation of a complex actor without having to use a base class:

class HttpSessionActor {
    private val actor = Actor()
    // state initialization does not have to happen under lock
    private var state: MutableMap<String, String> = mutableMapOf()
   
    suspend fun set(key: String, value: String) = actor.act {
        state[key] = value
    }

    // etc
}

What is the difference here as compared to Mutex-based version? The difference is that an Actor has an inbox channel and sending messaged to an actor can be decoupled from their execution. When HttpSessionActor.set is invoked, the invoker can go on with it own work while HttpSessionActor is busy, unless actor's mailbox capacity is exhausted. In the latter case, the invoker will have to wait until mailbox has capacity to store a message, but not longer. This starts to be important for scalability when actors perform long-running asynchronous activities.

[fvasco]

@elizarov thanks for explanation,
in some sense a Mutex act like an actor with capacity = 0 and without coroutine context.

I fix my draft, but I sure that it is possible to implement a better one.

class TaskChannel(
        context: CoroutineContext = DefaultDispatcher,
        capacity: Int = 0,
        lazyInit: (suspend CoroutineScope.() -> Unit)? = null
) {

    private val tasks = Channel<Task<*>>(capacity)

    private var lazyInit: Deferred<*>? = async(context, start = CoroutineStart.LAZY) {
        lazyInit?.invoke(this)

        launch(coroutineContext) {
            tasks.consumeEach { it() }
        }
    }

    /**
     * Force lazy initialization
     */
    suspend fun init() {
        lazyInit?.run {
            await()
            lazyInit = null
        }
    }

    suspend fun <T> act(block: suspend () -> T): Deferred<T> {
        init()
        val task = Task(block)
        tasks.send(task)
        return task
    }

    suspend fun <T> actAndReply(block: suspend () -> T): T = act(block).await()

    private class Task<T>(block: suspend () -> T) : CompletableDeferred<T> by CompletableDeferred() {
        private var block: (suspend () -> T)? = block

        suspend operator fun invoke() {
            try {
                complete(block!!())
            } catch (t: Throwable) {
                completeExceptionally(t)
            } finally {
                block = null
            }
        }
    }
}

[elizarov]

Even with capacity = 0 the Actor is asynchronous. You can send a message to an Actor and continue working on your code, while actor processes your message concurrently with your code. The Mutex, on the other hand, is always synchronous. No concurrency. That is, conceptually, why solutions based on Mutex/synchronized do not scale well.

[fvasco]

Even with capacity = 0 the Actor is asynchronous
The Mutex ... is always synchronous.

@elizarov can you confirm the follow code snippet?

suspend fun operationB(otherParams...): Result = actAndReply {
    // logic here
}

Is the functions's return typeResult and not Deferred<Result>?

Accordly with #261 it is pretty easy write the act function on top of a single private executor.

[elizarov]

@fvasco Yes, when you ask and actor and want a result back the proper design would be to have a suspend fun with a normal (non-deferred) Result. However, please note that this whole ask & wait pattern is an anti-pattern in actor-based systems, since it limits scalability. Well-designed actor-based system do not work that way.

Internally, actors are still implemented on top of channels.

[ZakTaccardi]

Something I’m thinking about is how send(..) on the current actor behaves (as it implements SendChannel<T>).

I believe it might be useful to suspend until the actor has completed its processing of the message.

So then there would be 3 ways to communicate into the actor:

1. non-suspending offer(..)
   2.send(..) that suspends until the actor can guarantee the message will be delivered
2. send(..) that suspends until the actor has processed the message

Under what conditions is number 2 useful?

[elizarov]

Waiting until the actor has processed the message is sometimes useful, but is always dangerous and easily leads to deadlocks. I would not recommend to support it.

[tristancaron]

What is the benefit of using Actors instead of StateFlow? Here two implementations of CounterModel:

fun CoroutineScope.counterActor() = actor<CounterMsg> {
    var counter = 0 // actor state
    for (msg in channel) { // iterate over incoming messages
        when (msg) {
            is IncCounter -> counter++
            is GetCounter -> msg.response.complete(counter)
        }
    }
}

and

class CounterModel {
    private val _counter = MutableStateFlow(0) // private mutable state flow
    val counter: StateFlow<Int> get() = _counter // publicly exposed as read-only state flow

    fun inc() {
        _counter.value++
    }
}

[fvasco]

Hi, @tristancaron,
pay attention, MutableStateFlow does not provide a CAS mutator (yet), _counter.value++ is not atomic.

[tristancaron]

Thanks for taking the time to answer @fvasco

So MutableStateFlow is not atomic, thus we can have a data race. Am I understanding it correctly?

If so, the documentation seems to imply that data race is not possible. https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-state-flow/index.html

Concurrency
All methods of data flow are thread-safe and can be safely invoked from concurrent coroutines without external synchronization.

Unless I misunderstood it.

[fvasco]

get and set value is thread safe, however _counter.value = _counter.value + 1 is not atomic, so you can successfully write a wrong value.

[handstandsam]

@fvasco - Since the get and set methods of StateFlow are thread-safe, isn't there no need for _counter.value to be atomic? All get and set requests will be processed in order meaning that there would be no race conditions. If you ended up doing background work inside the model to modify _counter, I could see an issue, but as long as you are doing computations within that context you should be fine, right?

[fvasco]

Hi, @handstandsam,
I exposed above some classic data race issues.
A single writer does not need synchronization (with other writers).

Obviously multiple concurrent writers, even in the same context, may fail (https://kotlinlang.org/docs/reference/coroutines/shared-mutable-state-and-concurrency.html#mutual-exclusion).

[adam-arold]

What should I use in the meantime if I don't want to use a Mutex (because I'm afraid of deadlocks)? What I have right now is a super primitive wrapper class that can get/transform its state:

class Atom<T : Any>(initialValue: T) {

    @Volatile
    private var value: T = initialValue

    override fun get(): T = value

    @Synchronized
    override fun transform(transformer: (T) -> T): T {
        value = transformer(value)
        return value
    }

}

but with this I can still shoot myself in the foot if two of these Atoms cause a deadlock. What i'm looking for is a way to calculate a new value for shared mutable state (transform here) that's relatively fast. I scrapped my other implementation that was using actor when it became obsolete, but I'm not happy with the current solution.

[totemcaf]

We are starting a new project and Actors model fits perfectly. I wonder that using Kotlin Actors requires to mark all code with @ObsoleteCoroutinesApi.

Do we have any progress on the actors implementation replacement?

[LifeIsStrange]

Note that Swift is currently working on actor model support, I wonder if some inspiration could be taken:
Food for thoughts:
https://news.ycombinator.com/item?id=26480922

Edit:
Active review:
https://forums.swift.org/t/se-0306-actors/45734
Review by Chris latner:
https://forums.swift.org/t/se-0306-actors/45734/4

[dimsuz]

Hmm, baking this framework into the language seems as a bit weird approach to take.

[penn5]

Hmm, baking this framework into the language seems as a bit weird approach to take.

I agree, it should be a library based approach (like it is now)

[LifeIsStrange]

@dimsuz I don't know but at least the pros and cons of doing this should be outlined in their RFC

[asad-awadia]

Regardless of how the actor piece is brought in- it would be good to have the ability to add supervision strategies like being able to restart on exception

[pigbayspy]

So will actor api change in future?

[drinkthestars]

what is the latest on the timeline/plans for actor api updates?

[chriswu000]

what is the latest on the timeline/plans for actor api updates?

Reddit AMA they said possibly start work in 2022? #485 (comment)

[nikiJava]

Hi :) Are there any updates about new actors API?

[Kshitij09-sc]

Based on the description of Stateful Actor and the idea of reducing boilerplate, I would like to propose following abstraction (inspired by orbit-mvi's Container). Let me know if anything might go wrong with this implementation.
cc: @elizarov @fvasco

interface ActorContext

abstract class StatefulActor(
    private val parentScope: CoroutineScope,
    bufferCapacity: Int = 64
) {
    private val dispatchChannel = Channel<suspend ActorContext.() -> Unit>(bufferCapacity)
    private val initialized = AtomicBoolean(false)

    private fun initializeIfNeeded() {
        if(initialized.compareAndSet(false, true)) {
            parentScope.launch {
                for (action in dispatchChannel) {
                    launch { context.action() }
                }
            }
        }
    }

    fun shutdown() {
        dispatchChannel.close()
    }

    protected fun act(action: suspend ActorContext.() -> Unit) {
        initializeIfNeeded()
        dispatchChannel.trySendBlocking(action)
    }

    abstract val context: ActorContext
}

Here's one actor implementation using this API

class MyActor(scope: CoroutineScope) : StatefulActor(scope), ActorContext {
    private var clickCounter = 0

    override val context: ActorContext get() = this

    fun increment() = act { clickCounter++ }

    fun printCount() = act { println(clickCounter) }
}

I found it surprisingly fast as well (Unless I'm making some mistake measuring the performance).

Playground with above snippets and measuring setup: https://pl.kotl.in/RHPpCbcUT

[fvasco]

@Kshitij09-sc
You are mixing blocking (dispatchChannel.trySendBlocking(action)) and asynchronous (action: suspend ActorContext.() -> Unit) code, this is a delicate choice and may lead to errors.
The Job in initializeIfNeeded executes actions concurrently (two nested launch), so printCount will print a number concurrently with other tasks.
You are using an AtomicBoolean instead of a lazy launch.

[Kshitij09-sc]

@fvasco Agree with latter 2 points. Regarding mixing trySendBlocking(action) with suspending action: suspend ActorContext.() -> Unit, I considered this to be able to detect deadlocks. trySendBlocking will return failure once queue is full, queue might get full due to less buffer capacity or some action went into infinite loop (deadlock) which could be thrown as exception in debug environment.

[nowiwr01w]

Hi @elizarov, it's 2024. What about the new API?

[renatomrcosta]

Hi @elizarov, it's 2024. What about the new API?

Dude doesn't even work for Jetbrains anymore mate