Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Implement DeviceHostFile.evict() #2

Merged
merged 2 commits into from
Mar 15, 2022
Merged

Implement DeviceHostFile.evict() #2

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cfac32e
Implement `DeviceHostFile.evict()`
pentschev Mar 10, 2022
7c9d6fc
Add timeouts and prevent test_spill from pausing
pentschev Mar 14, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions dask_cuda/device_host_file.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,16 @@ def __delitem__(self, key):
self.device_keys.discard(key)
del self.device_buffer[key]

def evict(self):
"""Evicts least recently used host buffer (aka, CPU or system memory)

Implements distributed.spill.ManualEvictProto interface"""
try:
_, _, weight = self.host_buffer.fast.evict()
return weight
except Exception: # We catch all `Exception`s, just like zict.LRU
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would strongly advise to log it. Otherwise your users will have no clue why they're running out of memory.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently we can't log appropriately, we may consider this in the future. Could you point me to what Dask currently logs? I'm not sure we even intended to log it this way but could you point to where Dask is logging this currently?

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://github.com/dask/distributed/blob/main/distributed/spill.py
In SpillBuffer.evict()
https://github.com/dask/distributed/blob/7a69b5e24c115245005cb83129b4b4b46f05f151/distributed/spill.py#L168-L180

and its context manager SpillBuffer.handle_errors()
https://github.com/dask/distributed/blob/7a69b5e24c115245005cb83129b4b4b46f05f151/distributed/spill.py#L81-L134

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @crusaderky for the pointers, for now I opened rapidsai#873 so we can consider doing the same, we'll also have to think how to add proper logging in Dask-CUDA without just hacking into Distributed's.

return -1

def set_address(self, addr):
if isinstance(self.host_buffer, LoggedBuffer):
self.host_buffer.set_address(addr)
Expand Down
24 changes: 22 additions & 2 deletions dask_cuda/tests/test_spill.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ def delayed_worker_assert(total_size, device_chunk_overhead, serialized_chunk_ov
[
{
"device_memory_limit": int(200e6),
"memory_limit": int(800e6),
"memory_limit": int(2000e6),
"host_target": False,
"host_spill": False,
"host_pause": False,
Expand All @@ -98,6 +98,16 @@ def delayed_worker_assert(total_size, device_chunk_overhead, serialized_chunk_ov
"host_pause": False,
"spills_to_disk": True,
},
{
# This test setup differs from the one above as Distributed worker
# pausing is enabled and thus triggers `DeviceHostFile.evict()`
"device_memory_limit": int(200e6),
"memory_limit": int(1000e6),
"host_target": None,
"host_spill": None,
"host_pause": None,
"spills_to_disk": True,
},
{
"device_memory_limit": int(200e6),
"memory_limit": None,
Expand Down Expand Up @@ -159,7 +169,7 @@ async def test_cupy_cluster_device_spill(params):
[
{
"device_memory_limit": int(200e6),
"memory_limit": int(800e6),
"memory_limit": int(4000e6),
"host_target": False,
"host_spill": False,
"host_pause": False,
Expand All @@ -173,6 +183,16 @@ async def test_cupy_cluster_device_spill(params):
"host_pause": False,
"spills_to_disk": True,
},
{
# This test setup differs from the one above as Distributed worker
# pausing is enabled and thus triggers `DeviceHostFile.evict()`
"device_memory_limit": int(200e6),
"memory_limit": int(2000e6),
"host_target": None,
"host_spill": None,
"host_pause": None,
"spills_to_disk": True,
},
{
"device_memory_limit": int(200e6),
"memory_limit": None,
Expand Down