Actor not released from worker memory, inducing memory leak

[oarcher]

I have a class WorkWithActor that instantiate an actor, and work with it.
The actor survive to de deletion of the WorkWithActor instance, and it induce some memory leaks on the workers

I expect that the actor is removed from the worker when the future no longer exists

Minimal Complete Verifiable Example:

from dask.distributed import Client, worker_client


class Actor:

    def __init__(self, n):
        self.n = n
        self.memory = bytearray(32000000)  # consume some memory

    def do_something(self):
        return -self.n

    def __del__(self):
        print('__del__ Actor')  # not reached


class WorkWithActor:
    def __init__(self, n):
        with worker_client() as client:
            self._future = client.submit(Actor, n, actor=True)
            self._actor = self._future.result()

    def work(self):
        return self._actor.do_something()

    def __del__(self):
        print('__del__ WorkWithActor') # reached


def batch(n):
    work_with_actor = WorkWithActor(n)
    return work_with_actor.work()

if __name__ == "__main__":
    client = Client()
    print(client.dashboard_link)

    results = client.map(batch, range(1000))

Environment:

• Dask version: 2021.12.0
• Python version: 3.9.7
• Operating System: linux
• Install method (conda, pip, source): conda-forge

[oarcher]

It might be related to #4937 ?

@martindurant in #4937 (comment) says:

Is there a test showing that an actor is not currently cleaned up?

[holdenk]

Interestingly re-ran this on Dask 2022.5.1 with Python 3.8 and it seems that the inner actor is also being cleaned up.