Add actors for stateful operation

[mrocklin]

Often we want to maintain some state within a task during a computation, and have various other tasks interact with that state.

Currenlty our approach to handling this is to have long-running tasks that open up connections to other workers and communicate over queues, pub-sub, etc.. This works, but can be awkward, and is commonly a source of confusion even among advanced users.

I was playing with Ray recently and really enjoyed their new Actor model. I believe that we should shamelessly steal ideas from it :) (cc @robertnishihara)

User API

This involves creating a class

class Counter:
    def __init__(self):
        self.n = 0

    def increment(self):
        self.n += 1

    def get_value(self):
        return n

And then submitting this somehow to be an actor. Many API approaches here might be good. I'll do something dumb for now.

counter = client.submit_actor(Counter)

Functions on this class trigger something slightly lighter weight than a task

>>> counter.increment()
<ActorFuture ...>
>>> counter.get_value().result()
1

And this thing can be passed around to other tasks

def f(n, counter=None):
    for i in n:
        counter.increment()

futures = client.map(f, range(100), counter=counter)

Limitations

The introduction of state on a worker is powerful, but also limiting. Functionality would probably differ from tasks in the following way:

1. Resilience would not be guaranteed. Eventually we might support checkpointing in various ways, but probably not to start.
2. We wouldn't check in with the central scheduler, so these tasks wouldn't be recorded on the dashboard, used diagnosics, or load balanced in any way

Implementation

We might first enable workers to directly ask each other to execute tasks for them. No dependencies here, just "run this function for me please and hold onto the result until I or someone else asks for it".

ActorFutures would be like normal futures except that the scheduler would have no record of them, and they would also include the address of the worker on which they live and the ID of the actor to whom they belong.

We'll need to relax various validation checks that we use in testing to allow for this unexpected data to exist.

[martindurant]

Strongly behind this idea, I think it should enable new classes of algorithms. I'll get back to you on the questions you've raised here.

[martindurant]

Quick question: for the case of a counter, one obvious model would be to hold it on the scheduler, like a spark "accumulator". That would involve either executing code in the scheduler (bad), or the scheduler making tasks every time the object is invoked - which is fine. Is it the data structure that's the most attractive here, or the potential for non-scheduler data-flow?

[mrocklin]

These would live on workers. This would be to support workloads that need non-trivial distributed state, and that want to operate at lower latencies / higher bandwidths than the centralized scheduler could manage.

…

On Thu, Jul 12, 2018 at 9:40 AM, Martin Durant ***@***.***> wrote: Quick question: for the case of a counter, one obvious model would be to hold it on the scheduler, like a spark "accumulator". That would involve either executing code in the scheduler (bad), or the scheduler making tasks every time the object is invoked - which is fine. Is it the data structure that's the most attractive here, or the potential for non-scheduler data-flow? — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#2109 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszOaPoq9lJMhfJh7JCEvSLom61Qprks5uF1_xgaJpZM4VMw-5> .