Repository navigation
Semaphore gets released too often #4147
Description
To avoid generating too much NFS traffic, I am using a semaphore like this to copy data from NFS to a local file storage to read from later:
cluster = SLURMCluster(cores=16, memory="64 GB", #processes=1,
local_directory="/var/scratch/me/dask_scheduler_spill",
interface='ib0', walltime='24:00:00')
# Create a client to submit to.
client = Client(cluster)
# Allocate 10 nodes in the cluster
cluster.scale(10)
print("Waiting for workers")
# Wait until they are ready
client.wait_for_workers(10)
print("Workers are ready!")
sem = Semaphore(max_leases=4, name='data_copy')
def copy_data(location, target, sem):
with sem:
copy_tree(location, target, update=True)
client.run(copy_data, root_dataset, local_dataset, sem)
sem.close() # Clean up the semaphore at the scheduler side
Now if I run it I get the error that
site-packages/distributed/semaphore.py", line 472, in release
raise RuntimeError("Released too often")
RuntimeError: Released too often
I tried debugging, I found out that setting processes=1, in the construction of SLURMCluster does **not ** fix this. So it looks like a threading issue to me. #4057 should've fixed that, but clearly it didn't? Creating the semaphore with sem = Semaphore(max_leases=4, name='data_copy', client=client, register=True) also did not work.
Activity
lfdversluis commentedon Oct 5, 2020
Probably important details:
and
mrocklin commentedon Oct 5, 2020
cc @fjetter @lr4d
fjetter commentedon Oct 6, 2020
Can you try defining the semaphore instance within the function instead of passing it via the function
This should help ruling out a (de-)serialization issue. I am pretty surprised to see this and have no idea what's beneath this. The first thing I would suspect is that for whatever reason two worker threads use the same semaphore instance and they simultaneously manipulate internal state non-threadsafe. However, according to the docs, the deque in question here is supposed to be thread safe for the operations we're performing (append and popleft)
On a different note, the semaphore module is quite verbose at logging. Can you maybe post the semaphore module logs in here? Maybe they give some insight into what's going on
lfdversluis commentedon Oct 6, 2020
No Semaphore output is ending up in the slurm-xxxxxx.out files. I ran my script with your suggested modification and it still happens.
Here is the output of one of the slurm-xxxxx.out files
I now noticed the line
coroutine 'Semaphore._acquire' was never awaited. I see the _acquire is async, yetreturn self.client.sync(self._acquire, timeout=timeout)in the source code should deal with this, no? I don't have enough knowledge of Dask internals to debug this fully.fjetter commentedon Oct 6, 2020
Well, yes I would think so. However, if this was never awaited, for whatever reason, it would explain why your code could reach the release before it actually acquired a lease. Let's just put a
assert this_cannot_happensomewhere in there 😅If this was never awaited you should effectively not seeing your computations to slow down. The acquire is the blocking call and without it all jobs will run simultaneously.
While I haven't understood the problem yet and believe this should be fixed, if this is the root cause you might work around it by using the following
by declaring it as a coroutine and awaiting explicitly you should be able to work around the issue.
Browsing the code a bit, the
client.syncmethod should take care of the future unless you're in an async context already. Haven't worked with dask jobque myself, yet. Is there anythign special about the setup there?fjetter commentedon Oct 6, 2020
Not really pretty, but to get to the bottom of it, there are two things which would cause the client.sync to not block and this is if the client is setup asynchronously or if it detects the thread to be in an async state.
You might want to look at the output of
lfdversluis commentedon Oct 7, 2020
If I run your snippet with
The script crashes before it can even print, seems like the run function does not block?
Running with
outputs nothing in the logs? Just
printing doesn't work either; I think as it runs on the node instead of scheduler?
fjetter commentedon Oct 7, 2020
I had another look at the issue and it turns out it can be rather easily reproduced with the local cluster even
The issue is that you're using https://distributed.dask.org/en/latest/api.html#distributed.Client.run to schedule an out-of-band function, is this intentional? Internally this works slightly differently compared to an ordinarily scheduled task/future.
The normal way would be to use https://distributed.dask.org/en/latest/api.html#distributed.Client.submit or https://distributed.dask.org/en/latest/api.html#distributed.Client.map
For example
The later works for me as expected. Can you confirm this on your end?
lfdversluis commentedon Oct 7, 2020
🤔 I was using run to ensure it runs on all nodes, but I guess submit works too? The reason I used run is because it's blocking. I need to ensure all workers have the dataset on local storage before I start the computation (otherwise the data might not be there). This shouldn't also impact the usage of the semaphore, being out-of-band or not (right?).
I will try to use the submit function with
.result()orwait(fut)fjetter commentedon Oct 7, 2020
Well, if you call
submitit will be scheduled as one task and will only be executed on one worker (unless you force it to be distributed among all workers).I believe when using
Client.runit is always an asynchronous context, whether you want it to be or not since it is running in the primary server thread (ordinary tasks/futures are computed in a different thread/threadpool). Therefore, if you want to go for the Client.run you probably need to use theasync defsyntax, otherwise the semaphore won't work. I'll think about whether or not we can check this inthe semaphore to at least raise an appropriate exceptionSide note: For your purpose you might want to consider https://distributed.dask.org/en/latest/api.html#distributed.Client.register_worker_callbacks which ensures that this function is called every time a new worker starts. Alternatively, you can implement a WorkerPlugin with a similar purpose. Haven't tested the semaphore with this approach, though. You might run into the same issue.