memory leak with min/max aggregation of huge array

[phildong]

I'm running a single-machine cluster and try to load and process a movie larger than memory. However I seem to have memory leaking issue at various steps, especially when doing some aggregation. So I create the following simple script to track it down. Surprisingly even the first step in my workflow dask.array.image.imread seem to be enough to trigger some memory leak:

import os
import dask
import numpy as np
from dask.distributed import LocalCluster, Client, fire_and_forget
from dask.array.image import imread
cluster = LocalCluster(diagnostics_port=8989, memory_limit="200MB")
client = Client(cluster)

/opt/miniconda3/envs/dask/lib/python3.7/importlib/_bootstrap.py:219: RuntimeWarning: numpy.dtype size changed, may indicate binary incompatibility. Expected 96, got 88
  return f(*args, **kwds)
/opt/miniconda3/envs/dask/lib/python3.7/importlib/_bootstrap.py:219: RuntimeWarning: numpy.dtype size changed, may indicate binary incompatibility. Expected 96, got 88
  return f(*args, **kwds)

array = dask.array.zeros((20000, 500, 800), chunks=(1, -1, -1))
array

dask.array<zeros, shape=(20000, 500, 800), dtype=float64, chunksize=(1, 500, 800)>

#this is perfectly fine despite that the array is larger than my real movie
arr_sum = client.compute(array.sum())
arr_sum.result()

0.0

#the folder contains ~18000 tiff files, but to isolate the issue I'm not actually reading them
dpath = "/home/phild/Documents/test_data/"
def dummy_read(im):
    return np.zeros((480, 752))
array = imread(os.path.join(dpath, "*.tiff"), dummy_read)
array

dask.array<imread, shape=(17991, 480, 752), dtype=float64, chunksize=(1, 480, 752)>

arr_sum = client.compute(array.sum())
arr_sum.result()

distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker process 12465 was killed by signal 15
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker process 12463 was killed by signal 15
distributed.nanny - WARNING - Worker process 12484 was killed by signal 15
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker process 12474 was killed by signal 15
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker process 12479 was killed by signal 15
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker process 12477 was killed by signal 15
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Worker process 12468 was killed by signal 15
distributed.nanny - WARNING - Worker process 12472 was killed by signal 15
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker



---------------------------------------------------------------------------

KilledWorker                              Traceback (most recent call last)

<ipython-input-5-12b03eff355b> in <module>
      1 arr_sum = client.compute(array.sum())
----> 2 arr_sum.result()


/opt/miniconda3/envs/dask/lib/python3.7/site-packages/distributed/client.py in result(self, timeout)
    193                                   raiseit=False)
    194         if self.status == 'error':
--> 195             six.reraise(*result)
    196         elif self.status == 'cancelled':
    197             raise result


/opt/miniconda3/envs/dask/lib/python3.7/site-packages/six.py in reraise(tp, value, tb)
    691             if value.__traceback__ is not tb:
    692                 raise value.with_traceback(tb)
--> 693             raise value
    694         finally:
    695             value = None


KilledWorker: ("('imread-sum-22b1dc8c8dbc99d865bdc52557ca4d52', 5415, 0, 0)", 'tcp://127.0.0.1:44369')

The last sum produce memory leak and killed all my workers despite it's smaller than the array created by dask.array.zeros, and I have something like the following in my worker logs:

distributed.worker - DEBUG - Calling gc.collect(). 3.435s elapsed since previous call.

distributed.worker - DEBUG - gc.collect() took 0.089s

distributed.worker - WARNING - Worker is at 80% memory usage. Pausing worker. Process memory: 161.94 MB -- Worker memory limit: 200.00 MB

distributed.worker - WARNING - Memory use is high but worker has no data to store to disk. Perhaps some other process is leaking memory? Process memory: 161.94 MB -- Worker memory limit: 200.00 MB

distributed.worker - DEBUG - Heartbeat skipped: channel busy

distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:32873

distributed.worker - DEBUG - gc.collect() lasts 0.089s but only 0.137s elapsed since last call: throttling.

distributed.worker - WARNING - Memory use is high but worker has no data to store to disk. Perhaps some other process is leaking memory? Process memory: 162.17 MB -- Worker memory limit: 200.00 MB

distributed.worker - DEBUG - gc.collect() lasts 0.089s but only 0.144s elapsed since last call: throttling.

I understand that 200MB is probably not a reasonable memory limit for data of this size, but I suppose dask should handle it even under such extreme cases by spilling everything to disk (in fact I think that was what happened when we create array with dask.array.zeros), instead of complaining about memory leak?

If this is a expected behavior, however, I'd like to understand what is a rule of thumb for estimating the minimal required memory when performing such tasks?

[mrocklin]

I wouldn't be surprised to learn that the majority of 200MB might be filled just by importing various scientific Python libraries.

In [1]: import psutil                                                           

In [2]: psutil.Process().memory_info().rss / 1e6                                
Out[2]: 44.490752

In [3]: import dask.array                                                       

In [4]: psutil.Process().memory_info().rss / 1e6                                
Out[4]: 64.192512

In [5]: import dask.array.image                                                 

In [6]: psutil.Process().memory_info().rss / 1e6                                
Out[6]: 113.864704

[phildong]

I wouldn't be surprised to learn that the majority of 200MB might be filled just by importing various scientific Python libraries.

In [1]: import psutil                                                           

In [2]: psutil.Process().memory_info().rss / 1e6                                
Out[2]: 44.490752

In [3]: import dask.array                                                       

In [4]: psutil.Process().memory_info().rss / 1e6                                
Out[4]: 64.192512

In [5]: import dask.array.image                                                 

In [6]: psutil.Process().memory_info().rss / 1e6                                
Out[6]: 113.864704

Hah! Thanks for the fast reply! I was silly in thinking that I can push memory_limit to extreme so that I could run into the leaking issue sooner with smaller size of data.

After I set memory_limit to 1GB (which I assume should be reasonable) the sum returns fine!

However I ran into memory leak again with the following attempt to normalize my entire array to the range of (0, 1):

arr_max = array.max()
arr_min = array.min()
arr_norm = (array - arr_min) / (arr_max - arr_min)

and a following call to client.persist(arr_norm) will give me killed worker and memory leak warnings in logs, no matter array was generated with dask.array.random or loaded from real tiffs (so I guess it's no longer an imread issue per se now). I can also confirm that if I don't do the normalization and just call client.persist(array), the array can sit in the cluster with probably part of it spilled to disk. So I assume I'm doing something stupid with the normalization?

As a side note, I also noticed just after I started the cluster, even if I do nothing, the "Bytes stored" under dashboard will very slowly increase, as if accumulating logs or something, even though I virtually turned off logging by setting "distributed: warning" in the configuration. Do you have any suggestion in how to track this down?

[mrocklin]

It gives you a warning about a leak, or just that it's low on memory? You're asking for the entire array to be in memory at once, so it will start to push data onto disk, and warn you that it's doing so.

…

On Tue, Nov 27, 2018 at 2:35 PM phildong ***@***.***> wrote: I wouldn't be surprised to learn that the majority of 200MB might be filled just by importing various scientific Python libraries. In [1]: import psutil In [2]: psutil.Process().memory_info().rss / 1e6 Out[2]: 44.490752 In [3]: import dask.array In [4]: psutil.Process().memory_info().rss / 1e6 Out[4]: 64.192512 In [5]: import dask.array.image In [6]: psutil.Process().memory_info().rss / 1e6 Out[6]: 113.864704 Hah! Thanks for the fast reply! I was silly in thinking that I can push memory_limit to extreme so that I could run into the leaking issue sooner with smaller size of data. After I set memory_limit to 1GB (which I assume should be reasonable) the sum returns fine! However I ran into memory leak again with the following attempt to normalize my entire array to the range of (0, 1): arr_max = array.max() arr_min = array.min() arr_norm = (array - arr_min) / (arr_max - arr_min) and a following call to client.persist(arr_norm) will give me killed worker and memory leak warnings in logs, no matter array was generated with dask.array.random or loaded from real tiffs (so I guess it's no longer an imread issue per se now). I can also confirm that if I don't do the normalization and just call client.persist(array), the array can sit in the cluster with probably part of it spilled to disk. So I assume I'm doing something stupid with the normalization? As a side note, I also noticed just after I started the cluster, even if I do nothing, the "Bytes stored" under dashboard will very slowly increase, as if accumulating logs or something, even though I virtually turned off logging by setting "distributed: warning" in the configuration. Do you have any suggestion in how to track this down? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#2377 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszJguIuVwh4COdRAdAV10ex19kdRzks5uzZP5gaJpZM4Y2Bba> .

[phildong]

It gives you a warning about a leak, or just that it's low on memory? You're asking for the entire array to be in memory at once, so it will start to push data onto disk, and warn you that it's doing so.
…
On Tue, Nov 27, 2018 at 2:35 PM phildong @.***> wrote: I wouldn't be surprised to learn that the majority of 200MB might be filled just by importing various scientific Python libraries. In [1]: import psutil In [2]: psutil.Process().memory_info().rss / 1e6 Out[2]: 44.490752 In [3]: import dask.array In [4]: psutil.Process().memory_info().rss / 1e6 Out[4]: 64.192512 In [5]: import dask.array.image In [6]: psutil.Process().memory_info().rss / 1e6 Out[6]: 113.864704 Hah! Thanks for the fast reply! I was silly in thinking that I can push memory_limit to extreme so that I could run into the leaking issue sooner with smaller size of data. After I set memory_limit to 1GB (which I assume should be reasonable) the sum returns fine! However I ran into memory leak again with the following attempt to normalize my entire array to the range of (0, 1): arr_max = array.max() arr_min = array.min() arr_norm = (array - arr_min) / (arr_max - arr_min) and a following call to client.persist(arr_norm) will give me killed worker and memory leak warnings in logs, no matter array was generated with dask.array.random or loaded from real tiffs (so I guess it's no longer an imread issue per se now). I can also confirm that if I don't do the normalization and just call client.persist(array), the array can sit in the cluster with probably part of it spilled to disk. So I assume I'm doing something stupid with the normalization? As a side note, I also noticed just after I started the cluster, even if I do nothing, the "Bytes stored" under dashboard will very slowly increase, as if accumulating logs or something, even though I virtually turned off logging by setting "distributed: warning" in the configuration. Do you have any suggestion in how to track this down? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#2377 (comment)>, or mute the thread https://github.com/notifications/unsubscribe-auth/AASszJguIuVwh4COdRAdAV10ex19kdRzks5uzZP5gaJpZM4Y2Bba .

I believe it is a warning about leak:

distributed.worker - WARNING - Memory use is high but worker has no data to store to disk. Perhaps some other process is leaking memory? Process memory: 711.71 MB -- Worker memory limit: 1000.00 MB

I'm perfectly fine with the worker pushing data to disk -- that's exactly what I want them to do. But I feel like they are unable to do so some times.

I also noticed that when I just call client.persist(array) without doing the normalization, I still get the memory leak warning, even though it eventually finished and no worker gets killed.

[mrocklin]

Yeah, to be clear it's not saying "there is a leak" it's saying "we're low on memory and don't know why". Unfortunately it's not clear what is taking up memory in your processes. None of the data that Dask is tracking seems to be reponsible.

…

On Tue, Nov 27, 2018 at 3:07 PM phildong ***@***.***> wrote: It gives you a warning about a leak, or just that it's low on memory? You're asking for the entire array to be in memory at once, so it will start to push data onto disk, and warn you that it's doing so. … <#m_12034826415996572_> On Tue, Nov 27, 2018 at 2:35 PM phildong ***@***.***> wrote: I wouldn't be surprised to learn that the majority of 200MB might be filled just by importing various scientific Python libraries. In [1]: import psutil In [2]: psutil.Process().memory_info().rss / 1e6 Out[2]: 44.490752 In [3]: import dask.array In [4]: psutil.Process().memory_info().rss / 1e6 Out[4]: 64.192512 In [5]: import dask.array.image In [6]: psutil.Process().memory_info().rss / 1e6 Out[6]: 113.864704 Hah! Thanks for the fast reply! I was silly in thinking that I can push memory_limit to extreme so that I could run into the leaking issue sooner with smaller size of data. After I set memory_limit to 1GB (which I assume should be reasonable) the sum returns fine! However I ran into memory leak again with the following attempt to normalize my entire array to the range of (0, 1): arr_max = array.max() arr_min = array.min() arr_norm = (array - arr_min) / (arr_max - arr_min) and a following call to client.persist(arr_norm) will give me killed worker and memory leak warnings in logs, no matter array was generated with dask.array.random or loaded from real tiffs (so I guess it's no longer an imread issue per se now). I can also confirm that if I don't do the normalization and just call client.persist(array), the array can sit in the cluster with probably part of it spilled to disk. So I assume I'm doing something stupid with the normalization? As a side note, I also noticed just after I started the cluster, even if I do nothing, the "Bytes stored" under dashboard will very slowly increase, as if accumulating logs or something, even though I virtually turned off logging by setting "distributed: warning" in the configuration. Do you have any suggestion in how to track this down? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#2377 (comment) <#2377 (comment)>>, or mute the thread https://github.com/notifications/unsubscribe-auth/AASszJguIuVwh4COdRAdAV10ex19kdRzks5uzZP5gaJpZM4Y2Bba . I believe it is a warning about leak: distributed.worker - WARNING - Memory use is high but worker has no data to store to disk. Perhaps some other process is leaking memory? Process memory: 711.71 MB -- Worker memory limit: 1000.00 MB I'm perfectly fine with the worker pushing data to disk -- that's exactly what I want them to do. But I feel like they are unable to do so some times. I also noticed that when I just call client.persist(array) without doing the normalization, I still get the memory leak warning, even though it eventually finished and no worker gets killed. — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#2377 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszK_0hp0BAtompjZXQvAiqVSvnmLBks5uzZuJgaJpZM4Y2Bba> .

[phildong]

sorry I'm a bit confused -- does the warning just means "we are low on memory", which is perfectly fine, or it means "we are low on memory And something else is going on that is out of control by dask", which is very problematic?

If the later is true, here is a minimal example that will produce the warning on my machine, and I don't see why there would be any data that is not tracked by dask and cannot be spilled to disk:

import dask
import dask.array as da
from dask.distributed import LocalCluster, Client
cluster = LocalCluster(diagnostics_port=8989, memory_limit="1GB")
client = Client(cluster)
array = da.zeros((200000, 500, 800), chunks=(1, -1, -1))
client.persist(array)

Sorry if I seems intentionally trying to beat dask to death, but this warning is the only thing that seems abnormal during my computation and my workflow inevitably involve holding huge video arrays on the workers and I really want them to be able to hold arbitrary size with potentially spilling to disk. Any other suggestion on how to do or debug this would be super helpful!

Update: the code snippet not only produce warnings but also lead to worker killed with reaching 95% of memory limit. Please let me know if this is expected.

[mrocklin]

Beating Dask to death is a productive and educational experience. I tried your example and got the same result (thanks for the simple example). I then tried it but with a larger chunk size of (100, -1, -1) (you have 200k tasks here) and it ran fine. In this case I suspect that the metadata to store all of the tasks was getting up to your 1 GB memory range. Generally speaking I recommend having workers with larger memory than 1GB, and also having fewer than 200k tasks.

…

On Tue, Nov 27, 2018 at 3:59 PM phildong ***@***.***> wrote: sorry I'm a bit confused -- does the warning just means "we are low on memory", which is perfectly fine, or it means "we are low on memory And something else is going on that is out of control by dask", which is very problematic? If the later is true, here is a minimal example that will produce the warning on my machine, and I don't see why there would be any data that is not tracked by dask and cannot be spilled to disk: import daskimport dask.array as dafrom dask.distributed import LocalCluster, Client cluster = LocalCluster(diagnostics_port=8989, memory_limit="1GB") client = Client(cluster) array = da.zeros((200000, 500, 800), chunks=(1, -1, -1)) client.persist(array) Sorry if I seems intentionally trying to beat dask to death, but this warning is the only thing that seems abnormal during my computation and my workflow inevitably involve holding huge video arrays on the workers and I really want them to be able to hold arbitrary size with potentially spilling to disk. Any other suggestion on how to do or debug this would be super helpful! — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#2377 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszAc4Yyx3oj9qbgCQgbplZYY1laIeks5uzafBgaJpZM4Y2Bba> .

[phildong]

I see... yeah that make sense. Thanks for the insight! Just out of curiosity, is it possible to ask dask to not hold back and spill everything to disk, including the metadata?

Also sadly when I get back to my normalization attempt, I found something that seems cannot be explained by stored metadata alone --

I found the following to be working OK:

import dask
import dask.array as da
from dask.distributed import LocalCluster, Client
cluster = LocalCluster(diagnostics_port=8989, memory_limit="1GB")
client = Client(cluster)
array = dask.array.random.random((20000, 500, 800), chunks=(1, -1, -1))
arr_max = array.max().compute()
arr_min = array.min().compute()
arr_norm = (array - arr_min) / (arr_max - arr_min)
client.persist(arr_norm)

as expected, arr_max and arr_min are returned in that order with the blocking compute(). Up to then there was no warning in the workers and memory consumption barely increased, which I guess is also expected since we are not asking for array to be stored. Then the call to persist triggered some "memory high but no data to store to disk" warning, but not any "pausing worker" nor "killed worker", and it finished fine. The problem with this is that I believe dask.array.random gets called three times and array get generated three times. In reality if the array come from some expensive image loading and processing, executing the whole process three times would be very impractical.

And interestingly, when I use chunks=(100, -1, -1) with the previous code, arr_max and arr_min returned fine, if not faster, but call to persist(arr_norm) immediately lead to worker killed due to exceeding 95% memory budget. I guess this cannot be accounted for by overflowing metadata?

I also tried the following, which is only different from before in that I didn't ask arr_max and arr_min to be compute() before, and dask should be smart enough to realize that we should do everything in one shot:

arr_max = array.max()
arr_min = array.min()
arr_norm = (array - arr_min) / (arr_max - arr_min)
client.persist(arr_norm)

and dask did try to do the max() min() together with only one shot of random(). However, with chunks=(1, -1, -1), things begin to fall apart somewhere along the half way, where workers gets paused or killed repeatedly and the whole computation goes back and forth and can not progress effectively (along with the memory warning of course). With chunks=(100, -1, -1), again the call to persist lead to immediate all workers killed.

I feel like I must be doing something stupidly wrong, so I'm wondering what's the recommend way to normalize a huge array to the range of (0, 1) with dask.distributed?

[mrocklin]

Hrm, yeah, I'm also able to reproduce the same failure, even with more permissive settings

import dask
import dask.array as da
from dask.distributed import LocalCluster, Client
cluster = LocalCluster(memory_limit="2GB")
client = Client(cluster)
x = dask.array.random.random((20000, 500, 800), chunks=(100, -1, -1))
x = x - x.max()
z = client.persist(x)

This would be a good fail case if anyone wants to investigate debugging these sorts of issues further. cc @jcrist @TomAugspurger @martindurant ? I'd be happy to help someone if they want to dive in here.

[phildong]

I'd love to help in any way. I'm probably not knowledgeable enough to dive in the source of dask, but I'd be happy to run some tests if needed. Really love to see this working!

[TomAugspurger]

I should be able to take an initial look tomorrow afternoon if no one beats me to it.

…

On Wed, Nov 28, 2018 at 2:37 PM phildong ***@***.***> wrote: I'd love to help in any way. I'm probably not knowledgeable enough to dive in the source of dask, but I'd be happy to run some tests if needed. Really love to see this working! — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#2377 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ABQHIq-qn3iDThc5bFwydLO0NfhtU2Lyks5uzvP1gaJpZM4Y2Bba> .

[TomAugspurger]

I haven't had a chance to debug this yet.

[jakirkham]

Have just been reading up on this issue. I'm not sure I totally understand the question yet. So apologies if most of this is missing the point.

Naively this looks a lot like issue ( dask/dask#874 ). Namely there are some cases where we want to perform some operation where we require an aggregate result (e.g. mean, min, max) as intermediate to perform some global operation (e.g. recentering, rescaling, etc.). In these cases the scheduler struggles and it would be useful to do something better (i.e. the goal of the xref'd issue). Workarounds would be to optimize the graph with particular constraints or store the intermediate somehow. A solution might be to have a proper cache with an eviction rule.

Though this probably differs from the aforementioned issue in a way that I've missed. Pointers to how it differs would be very helpful in improving my understanding and thinking of a better way to address the need here. Thanks in advance for bearing with me. 🙂

[phildong]

@jakirkham I agree this should be the same as dask/dask#874. Let me know if this should be closed to redirect place of discussion