Allow actors to call actors on the same worker

[martindurant]

Fixes #4224

[martindurant]

I tested manually with the following code, to show that blocking, async methods and properties work:

import dask.distributed
client = dask.distributed.Client(n_workers=1)

class Minimal:
    def __init__(self):
        self.value = 0

    def inc(self):
        self.value += 1
        return self.value

    async def ainc(self):
        self.value += 1
        return self.value

    @property
    def val(self):
        return self.value

class UsesMinimal:

    def do_inc(self, ac, N):
        return ac.inc().result()

    async def ado_inc(self, ac, N):
        return await ac.ainc()

    def do_val(self, ac):
        return ac.val

ac = client.submit(Minimal, actor=True).result()
ac2 = client.submit(UsesMinimal, actor=True, workers=[ac._address]).result()

[mrocklin]

Cool. I'm glad to see this come together so fast. Do you think it would be doable to include your manual test as a pytest test?

[martindurant]

Yes of course - I wasn't certain whether what i wrote made sense or not.

…

On November 6, 2020 8:38:03 PM EST, Matthew Rocklin ***@***.***> wrote: Cool. I'm glad to see this come together so fast. Do you think it would be doable to include your manual test as a pytest test? -- You are receiving this because you authored the thread. Reply to this email directly or view it on GitHub: #4225 (comment)

-- Sent from my Android device with K-9 Mail. Please excuse my brevity.

[mrocklin]

I'm confused by this. It seems like a no-op? Would return attr be equivalent?

[martindurant]

That seems ... likely :). It must have been a little more complex as I tried&errored here.

[mrocklin]

Yes of course - I wasn't certain whether what i wrote made sense or not.

I don't see anything that rings any warning bells, but all of my experience here is somewhat dated. And hey, "if it passes tests ... " :)

[martindurant]

This took a little more work, because the new code path I made only applies to workers calling workers, not normal submitted tasks calling workers (because they go via the scheduler and get executed in the normal pool).

[martindurant]

This passed on Travis at https://travis-ci.com/github/martindurant/distributed/builds/199375854 but not updating here.

[martindurant]

The following also fails - in fact, I think any exception in an actor might kill the worker. I'll put the fix in a new PR as soon as I have it

def test_exception():
    class MyException(Exception):
        pass

    class Broken:
        def method(self):
            raise MyException

    with cluster(nworkers=2) as (cl, w):
        client = Client(cl["address"])
        ac = client.submit(Broken, actor=True).result()
        acfut = ac.method()
        with pytest.raises(MyException):
            acfut.result()

[mrocklin]

Typically we use threadlocals for this. You might want to look into the use of the thread_state variable in worker.py

[mrocklin]

Maybe we should set this to False just after calling the function?

[martindurant]

The intent, I think, is to state "yes, this is an actor-specific thread" rather than "we happen to be running an actor function right now". This use is more similar to checking the thread name.