Clarify instructions to set MALLOC_TRIM_THRESHOLD_ in nanny config

[jbusecke]

I am experiencing some machine specific (e.g. I mostly see this happening on this machine vs others) memory issues, and wanted to set the memory trimming to a very aggressive setting as suggested here.

It is not quite clear to me how to carry out the following step:

You can modify this behavior using the distributed.nanny.environ configuration value.

I tried

import dask
dask.config.set({'distributed.nanny.environ.MALLOC_TRIM_THRESHOLD_':10})

# set up dask
from dask.distributed import LocalCluster, Client
cluster = LocalCluster()
client = Client(cluster)

# check environment variables on workers
def check():
    import os
    return os.environ["MALLOC_TRIM_THRESHOLD_"]
    
# print(f'On client: {check()}')
client.run(check)

This gives the following error trace

distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-uqmljo3j', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-xca4noxo', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-hjtzgrbs', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-0gfexb3i', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-izzzhnbt', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-givixhos', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-camyfg38', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-_yl2kk0k', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-3b8l0fg6', purging
distributed.diskutils - INFO - Found stale lock file and directory '/projects/GEOCLIM/LRGROUP/jbusecke/projects/cmip6_omz/notebooks/processing/dask-worker-space/worker-cl5f0fk3', purging
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
distributed.worker - WARNING - Run Failed
Function: check
args:     ()
kwargs:   {}
Traceback (most recent call last):
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/worker.py", line 3990, in run
    result = function(*args, **kwargs)
  File "/tmp/ipykernel_185371/2848085022.py", line 12, in check
  File "/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py", line 679, in __getitem__
    raise KeyError(key) from None
KeyError: 'MALLOC_TRIM_THRESHOLD_'
---------------------------------------------------------------------------
KeyError                                  Traceback (most recent call last)
/tmp/ipykernel_185371/2848085022.py in <module>
     13 
     14 # print(f'On client: {check()}')
---> 15 client.run(check)

/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/client.py in run(self, function, *args, **kwargs)
   2543         >>> c.run(print_state, wait=False)  # doctest: +SKIP
   2544         """
-> 2545         return self.sync(self._run, function, *args, **kwargs)
   2546 
   2547     @_deprecated(use_instead="Client.run which detects async functions automatically")

/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
    859             return future
    860         else:
--> 861             return sync(
    862                 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
    863             )

/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
    324     if error[0]:
    325         typ, exc, tb = error[0]
--> 326         raise exc.with_traceback(tb)
    327     else:
    328         return result[0]

/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/utils.py in f()
    307             if callback_timeout is not None:
    308                 future = asyncio.wait_for(future, callback_timeout)
--> 309             result[0] = yield future
    310         except Exception:
    311             error[0] = sys.exc_info()

/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/tornado/gen.py in run(self)
    760 
    761                     try:
--> 762                         value = future.result()
    763                     except Exception:
    764                         exc_info = sys.exc_info()

/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/site-packages/distributed/client.py in _run(self, function, nanny, workers, wait, *args, **kwargs)
   2480             elif resp["status"] == "error":
   2481                 typ, exc, tb = clean_exception(**resp)
-> 2482                 raise exc.with_traceback(tb)
   2483         if wait:
   2484             return results

/tmp/ipykernel_185371/2848085022.py in check()
     10 def check():
     11     import os
---> 12     return os.environ["MALLOC_TRIM_THRESHOLD_"]
     13 
     14 # print(f'On client: {check()}')

/scratch/gpfs2/jbusecke/conda_tigressdata/envs/cmip6_omz/lib/python3.9/os.py in __getitem__()
    677         except KeyError:
    678             # raise KeyError with the original key value
--> 679             raise KeyError(key) from None
    680         return self.decodevalue(value)
    681 

KeyError: 'MALLOC_TRIM_THRESHOLD_'

which I interpret as the fact that the env variable is actually not set?

If I set MALLOC_TRIM_THRESHOLD_=0 (note the different value) before starting my notebook, I get this output.

On client: 0
{'tcp://127.0.0.1:33430': '0',
 'tcp://127.0.0.1:35192': '0',
 'tcp://127.0.0.1:37706': '0',
 'tcp://127.0.0.1:42139': '0',
 'tcp://127.0.0.1:43006': '0',
 'tcp://127.0.0.1:44153': '0',
 'tcp://127.0.0.1:44407': '0',
 'tcp://127.0.0.1:45369': '0',
 'tcp://127.0.0.1:45910': '0',
 'tcp://127.0.0.1:46315': '0'}

I mainly have two questions:

1. I assume I am making a mistake in setting the env variable in the first place? Or does the LocalCluster not use nannys?

2. Is the check I am carrying out sufficient, and is in my case setting MALLOC_TRIM_THRESHOLD_ before starting the notebook sufficient?

[jrbourbeau]

Hey @jbusecke, good to see you here! Could you try adding import distributed towards the top of your snippet right after import dask. This will ensure that all the distributed-related config options have been loaded into Dask's config system.

Separately, Dask started setting MALLOC_TRIM_THRESHOLD_=65536 by default for workers using a Nanny. I'm wondering if this isn't sufficient for your use case. What's prompted setting MALLOC_TRIM_THRESHOLD_=10?

[jrbourbeau]

Or does the LocalCluster not use nannys?

LocalCluster uses nannies

Is the check I am carrying out sufficient, and is in my case setting MALLOC_TRIM_THRESHOLD_ before starting the notebook sufficient?

Setting MALLOC_TRIM_THRESHOLD_ before starting up your notebook should be sufficient. The main thing you'll want to make sure is, from the documentation section you linked to, "The variable must be set before starting the dask-worker process".

[jbusecke]

Thanks for the quick reply @jrbourbeau!

Could you try adding import distributed

That did not change the outcome. I am getting:

{'tcp://127.0.0.1:36182': '0',
 'tcp://127.0.0.1:36494': '0',
 'tcp://127.0.0.1:39515': '0',
 'tcp://127.0.0.1:39968': '0',
 'tcp://127.0.0.1:40062': '0',
 'tcp://127.0.0.1:43085': '0',
 'tcp://127.0.0.1:44138': '0',
 'tcp://127.0.0.1:44305': '0',
 'tcp://127.0.0.1:45462': '0',
 'tcp://127.0.0.1:45856': '0'}

The value 0 is set before starting the notebook.

What's prompted setting MALLOC_TRIM_THRESHOLD_=10

In this example I mainly wanted to set two differing values to be able to differentiate what came from the 'setting before' or the dask.config call.

Generally the low number was meant to be a test following these instructions in the docs:

To aggressively and automatically trim the memory in a production environment, you should instead set the environment variable MALLOC_TRIM_THRESHOLD_ (note the final underscore) to 0 or a low number

Separately, Dask started setting MALLOC_TRIM_THRESHOLD_=65536 by default for workers using a Nanny

How could I confirm that this is the case in my setup (when I do nothing), actually? My first attempt above returned a bunch of errors.

To give some more context: I ran a large calculation that gave me a bunch of warnings pointing me to this part of the docs. I however cannot reproduce this behavior anymore, and so maybe this is just a case of a particular whacky machine and another reason to switch to the cloud haha.

[gjoseph92]

@jbusecke what version of distributed are you running? Could you make sure it's the latest? I'm wondering if you just aren't using a version that knows what to do with distributed.nanny.environ yet.

Could you try adding import distributed towards the top of your snippet right after import dask

This is actually already happening when you do from dask.distributed import LocalCluster, Client—and it's clobbering your dask.config.set with

distributed/distributed/distributed.yaml

Line 146 in e1488e5

MALLOC_TRIM_THRESHOLD_: 65536

Try moving your dask.config.set({'distributed.nanny.environ.MALLOC_TRIM_THRESHOLD_':10}) after any distributed-related imports.

[gjoseph92]

What's prompted setting MALLOC_TRIM_THRESHOLD_=10

Also I've sometimes noticed a difference between 65535 and 0 (0 releases pages whereas 65535 doesn't seem to do much), so I think trying different values is reasonable. I've never tried 10 vs 0—if you notice a difference, maybe 0 is a special case in glibc malloc? We talked internally a little about changing the default to 0 instead of 65535, but haven't done it.

[jrbourbeau]

Try moving your dask.config.set({'distributed.nanny.environ.MALLOC_TRIM_THRESHOLD_':10}) after any distributed-related imports.

Yep, that would also work too

@jbusecke there are a few different ways to go about setting environment variables and it looks like you're trying multiple approaches at once. I would recommend starting off with a single mechanism first. For example, if you don't set any environment variables ahead of time, then

    import dask
    from dask.distributed import LocalCluster, Client

    for i in [10, 100]:
        print(f"\n\n{i = }")
        dask.config.set({'distributed.nanny.environ.MALLOC_TRIM_THRESHOLD_': i})
        with LocalCluster() as cluster:
            with Client(cluster) as client:

                # check environment variables on workers
                def check():
                    import os
                    return os.environ["MALLOC_TRIM_THRESHOLD_"]
                    
                print(client.run(check))

will create two clusters, each with a different MALLOC_TRIM_THRESHOLD_ set on the workers

[jbusecke]

@jbusecke what version of distributed are you running? Could you make sure it's the latest? I'm wondering if you just aren't using a version that knows what to do with distributed.nanny.environ yet.

Ah good catch. I didnt realize there was a release during my vacation hehe. I am on 2021.07.1 currently. Let me update and check.

I did not see that setting the value to 0 brought any improvement and eventually was able to tracer this particular down to probably a chunking issue. So I guess in the end that warning was just a bit misleading. Thanks so much for the help already!

[jbusecke]

Ok I can confirm that after updating the version dask.config.set({'distributed.nanny.environ.MALLOC_TRIM_THRESHOLD_': i}) works as expected!

A final point: I actually looked at the version of the docs before and was pretty confused that it seems RTD uses a different versioning scheme?

But overall I consider this fixed! Thanks everyone.

[jrbourbeau]

Glad to hear it!

Also, thanks for pointing out those versions on readthedocs. Those are from super old releases. I've opened up an issue here #5282 about whether or not we should start publishing released versions of our docs again, or just removing the old versions