Is there any documentation around ResourceManagers and concurrency. Mostly just wondering about something like

with Flow(
    'resource_manager',
    executor=LocalDaskExecutor(),
) as flow:
    with ResourceManager() as manager:
        something_res = something(manager)
        get_res = another_thing.map(something_res, unmapped(manager))

Mostly wondering If I have the resource manager return “itself” how many copies are made via map

I believe 1 if unmapped but only 90% sure

For context, I am setting up a temporary directory in my manager, and returning itself from

setup

. The manager object has functions like

get_file

, etc to download a remote file to the temp dir. Its more or less a local file cache that will keep track of the disk space available, load/delete files as needed. I am wondering how this will work if I pass the manager to multiple tasks.

I think the issue there is potentially hitting the same file with concurrency right?

I think that is one issue yeah, the other is needing some kind of transaction, otherwise 2 mapped tasks can ask for files simultaneously and then depending on the order of the “available space” check, they both will get their files downloaded and then use up all the disk space.

If you use LocalDaskExecutor threads, memory can be shared. If you use process, I believe Dask copies it. It might even fail when you use processes unless that unmapped thing can be serialized which connections normally can’t be

seems like using a

threading.Lock

directly worked just fine. Roughly:

@resource_manager
from threading import Lock

class Manager:
    def do_thing(self, foo):
        with self._lock:
            return foo + 1

    def setup(self):
        self._lock = Lock()
        return self

    def cleanup(self, _):
        pass

Ah I see that’s nice

This just really helped me! Thank you!!