dask / dask/distributed

Profiler triggers `BufferError: cannot close exported pointers exist`

Open
#6,776 4 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

**What happened**:

Running the [` ncempy` DM file reader](https://openncem.readthedocs.io/en/latest/ncempy.io.html#module-ncempy.io.dm) with ` on_memory=True` in ` distributed` futures causes `BufferError: cannot close exported pointers exist` errors, while running it with other methods doesn't. Both a short wait between the actual workload and closing the reader object, and disabling the `distributed` profiler fix the issue.

A minimal reproducer that mimics the behavior is included below.

**What you expected to happen**:

If code runs natively, it will also run in ` dask.distributed` futures.

**Minimal Complete Verifiable Example**:

```python
import tempfile
import mmap

import dask
# Set the profiling interval so short that the bug is very likely to be triggered.
# It also happens with the default interval, just more rarely.
dask.config.set({"distributed.worker.profile.interval": "1 ms"})

# The bug doesn't happen when the profiler is disabled
# dask.config.set({"distributed.worker.profile.enabled": False})

import numpy as np
import distributed

def recurse(arr, index):
'''
Create many references into the array on the call stack.
'''
if index >= len(arr):
return
# This slice references memory from the memory map
data = arr[index:index+1]
return recurse(arr, index + 1)

def work_on_mmap(mm):
'''
Create a NumPy array backed by the memory map
and do some work on it.
'''
aa = np.frombuffer(mm, dtype=np.uint8)
recurse(aa, 0)

def do_map():
'''
Entry point, function to run on distributed cluster
'''
with tempfile.NamedTemporaryFile() as f:
f.write(b"abc"*100)
f.seek(0)
mm = mmap.mmap(f.fileno(), 0)
work_on_mmap(mm)
mm.close()

if __name__ == '__main__':
for i in range(1000):
do_map() # works

with distributed.Client() as client:
for i in range(1000):
# breaks
future = client.submit(do_map, priority=1, pure=False)
future.result()

```

**Anything else we need to know?**:

The profiler seems to hold references to temporary objects within futures.

Debugged in collaboration with @sk1p

CC @ercius FYI

**Environment**:

- Dask version: 2022.7.0
- Python version: 3.9.12
- Operating System: CentOS 7.9
- Install method (conda, pip, source): pip

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.