Hi- I'm trying to understand how ResourceManagers work. In your example you have something like:

@resource_manager
class DaskCluster:
    def init(self, n_workers):
        self.n_workers = n_workers

    def setup(self):
        return Client(n_workers=self.n_workers)

    def cleanup(self, client):
       client.close()

In this case how does the

Client

get passed to tasks? Is is pickled and sent around? Moreover if the client held some mutable state that could be updated by the tasks, what would happen?

Hi @Chris Martin - the client is passed to tasks directly using

with/as

syntax, like this:

with Flow("My Flow") as flow:
  with DaskCluster(n_workers=2) as client:
    my_task(client)

To learn more about how the Client works, I'd encourage you to check out the code in the core tasks: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/tasks/core/resource_manager.py

Hi- yes understood. I'm more asking what goes on behind the scenes to make this happen and whether I can (ab)use a resourcemanager (in some limited scenarios involving local executors) to share state between tasks.