distributed: Unmanaged (Old) memory hanging

What happened:

I am using Xarray and Dask to analyse large datasets ~100s of Gb, to reduce their dimensions from f(x,y,z,t) to f(t) by doing averages or sums. Several times I’ve encountered that Unmanaged (Old) memory hangs in the Cluster memory until it kills the client.

MemorySampler for one of my analysis for a spatial average, as show in the graph, the cluster memory never decreases towards it’s pre-run state (It looks like a memory leak, but the computation is only dataset.mean()).

I’ve tried setting "MALLOC_TRIM_THRESHOLD_" to 16284 and 0 and the issue persists.

I’ve also tried manually trimming memory:

import ctypes

def trim_memory() -> int:
    libc = ctypes.CDLL("libc.so.6")
    return libc.malloc_trim(0)

client.run(trim_memory)

and although, this usually only frees <1Gb and the cluster unmanaged memory remains in the order of 10s of Gb.

In order to better report this issue, I’ve attached an example that leaves around 1GB of unmanaged (old) memory (I’m not sure if this is related to issue https://github.com/dask/dask/issues/3530)

https://gist.github.com/josuemtzmo/3620e01c9caf88c18109809c52d77180

After lazy loading a test dataset, the prices memory is ~159.01 MiB using psutil. After running the computation; a mean on space (i.e. f(t,x,y).mean((‘x’,‘y’)), the process memory is ~256.83 MiB using psutil. This increase in memory is expected, however the cluster memory remains in 1.27GiB, where each of the workers has around 180MiB of Unmanaged (Old) memory hanging.

Screenshot 2022-04-28 at 16 48 57

Running the manual trim memory, the cluster memory only decreases the cluster memory from 1.26GiB to 1.23GiB. If the same code is run several times, the unmanaged memory continues to increase.

What you expected to happen:

Release of unmanaged (old) memory.

Environment:

Dask version: dask 2022.4.1
Python version: Python 3.9.12
Operating System: SLES linux
Install method (conda, pip, source): conda

Thanks!

About this issue

Original URL
State: open
Created 2 years ago
Comments: 17

Most upvoted comments

we are facing this issue, tried the snippet shared earlier:

from dask.distributed import get_client
import dask.array as da
from distributed.diagnostics import MemorySampler

client = get_client('10.10.10.1:8786')

import ctypes

def trim_memory() -> int:
    libc = ctypes.CDLL("libc.so.6")
    return libc.malloc_trim(0)


client.amm.start()
a = da.random.random((20_000, 20_000), chunks=(2_000, 2_000))
b = (a @ a.T).mean()

ms = MemorySampler()
with ms.sample("With AMM"):
    b.compute()
    
client.run(trim_memory)
ms.plot(figsize=(15,10), grid=True)

The memory increases on workers from around 20GB to 50 GB and keeps growing steadily. It grows significantly if i make the following change:

a = da.random.random((200_000, 200_000), chunks=(20_000, 20_000))

with above, the memory usage crossed 80GB and stayed there, even with client.run(trim_memory) and other tricks.

I have tried setting MALLOC_TRIM_THRESHOLD_ to 0 for all dask workers, but that doesn’t help either.

This is an issue for us, because we run dags which do significantly more processing than the above snippet and the memory utilization of the cluster crosses 250 GB. We have a cluster capacity of around 2 TB(memory). After some time the memory utilization becomes an issue and the workers get killed in between task executions thereby delaying our pipelines.

I tried the suggestions in this issue and also tried basic jemalloc tuning but nothing has helped so far.

I would like to understand if we can quickly address this.

n3rV3 on Feb 1, 2023

@crusaderky’s health/not healthy list would be a pretty great addition to the docs!

dcherian on May 3, 2022