Dask-cuda: [FEA] Allow communicating spilled data

[beckernick]

In some workflows, users rely on spilling to avoid going out of memory. My understanding is that, currently, if another worker needs spilled data, it must be brought back into GPU memory before it can be sent over the wire. In this scenario, there is a conceptually unnecessary GPU memory roundtrip (the data will be sent right back to the CPU) and additional GPU memory pressure as a result.

It would be very nice if we could directly communicate already spilled data, avoiding this additional memory pressure and performance hit. This topic has been discussed in the past, and @madsbk has done some exploratory work in https://github.com/dask/distributed/pull/3624 . As @jakirkham has also noted, the data could remain compressed which could improve bandwidth, too. There has been some discussion of this related topic, including https://github.com/dask/distributed/issues/3656 and the associated PR https://github.com/dask/distributed/pull/3702

Filing this issue to serve as a discussion board for the idea of communicating already-spilled data.

Note that if this issue is better suited for dask/distributed, I will raise it there.

[madsbk]

My exploratory work did two things:

• Added API for retrieving data from DeviceHostFile without the data being moved to device memory implicitly: https://github.com/rapidsai/dask-cuda/pull/259
• Made the Dask workers use this API when communicating: https://github.com/dask/distributed/pull/3624

It all works great but as @mrocklin pointed out, it might be too specialized a problem to introduce a new API for self.data beyond that of a MutableMapping.
This is where autocompression get in the picture. If we have something like https://github.com/dask/distributed/pull/3702, the motivation to support communication of spilled and/or compressed data might be strong enough to introduce a new API for self.data .

[pentschev]

It all works great but as @mrocklin pointed out, it might be too specialized a problem to introduce a new API for self.data beyond that of a MutableMapping.

An alternative to this is to reimplement that in dask-cuda itself. It will significantly increase the complexity of current dask-cuda codebase, but the potential benefits seem worth that cost.

[madsbk]

Thinking about this issue some more, I think I have an even better solution, which doesn't require any modification to Dask/Distributed either:

We introduce a new DeviceHostFile that returns a transparent proxy object of the data when Dask’s worker calls self.data[x]. The proxied data can then continue being compressed and/or spilled even after it's returned from self.data. Only when a task accesses the data, will we decompress and/or un-spill.

In order to support communication of spilled data, we introduce a new Dask serializer that recognize proxy objects and can avoid decompress and/or un-spill the data. It can even look at the serialization context and decides where it is best to have the data located (disk, main memory, device, etc) for the communication.

A positive consequence of this approach is that Dask tasks also gets the proxied data as inputs, which enables all sorts of important optimizations. First off, compressed and/or spilled data are only decompressed and/or un-spilled when the task accesses it and even more importantly, the task can re-compress and/or re-spill the data again as soon as possible. The task can even delete the data completely as soon as possible.

The last point is very useful for long running tasks like explicit-comms's dataframe shuffle, which cannot rely on the scheduler to release data early. In order to maintain resilience, the task can notify the scheduler that the data is lost, which makes the schedule include the data producer in a future recovery.

I really think this approach can make a huge difference both in terms of performance and memory usage and I am happy to work on it after my summer vacation 😊

[pentschev]

This sounds like a great idea @madsbk , thanks for writing this down. Generally speaking, I quite like that we can control this kind of behavior within dask-cuda, it allows us to experiment quickly and keeps dask/distributed clear of domain-specific code, which is a win-win.

It can even look at the serialization context and decides where it is best to have the data located (disk, main memory, device, etc) for the communication.

This is another good idea, probably a feature of its own. Today we have no such information as to where data is located or about the topology, likely something that would have to be solved before.

Maybe a question for @beckernick , how urgent is that request? Would you be fine waiting until Mads gets back or is this something that we should prioritize and try to get in ASAP?

[beckernick]

Very cool idea @madsbk ! Super excited about the discussion. I agree with Mads that this could be very impactful.

Vacations should be sacred. This can wait from my perspective 😄

[pentschev]

Vacations should be sacred. This can wait from my perspective 😄

Haha, sure! I wasn't suggesting Mads couldn't go, I'm more asking if this is more urgent and someone else should start working on it.

Although Mads is one of the most productive people I know, so maybe we should consider making him stay. 🤔 😈

[mrocklin]

I'm curious, how do tasks know when to unpack the object?

https://github.com/dask/distributed/blob/a228e69d42031a84f74a221f815dc1cef2c906b7/distributed/worker.py#L2515-L2550

On Fri, Jul 10, 2020 at 6:28 AM Peter Andreas Entschev <
[email protected]> wrote:

Vacations should be sacred. This can wait from my perspective 😄

Haha, sure! I wasn't suggesting Mads couldn't go, I'm more asking if this
is more urgent and someone else should start working on it.

Although Mads is one of the most productive people I know, so maybe we
should consider making him stay. 🤔 😈

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
https://github.com/rapidsai/dask-cuda/issues/342#issuecomment-656676786,
or unsubscribe
https://github.com/notifications/unsubscribe-auth/AACKZTFSTFDAH4GVI5ORIALR24JPPANCNFSM4OV5JCNQ
.

[madsbk]

I'm curious, how do tasks know when to unpack the object?

The idea is that data[k] = self.data[k] returns a _proxy object_ that mimics the underlying object much like Delayed. Only when the task uses the proxy object, e.g. by calling obj[42] or obj.merge(), will DeviceHostFile unpack the underlying object. Also keep in mind that DeviceHostFile can be selective when deciding which type of objects it want to return as a proxy. As a beginning we can limit DeviceHostFile to only return proxy objects of arrays and DataFrames and then return everything else _as is_.

In order to implement the proxy object I was thinking of something like: https://github.com/GrahamDumpleton/wrapt, which also mimics type checks like isinstance()

[mrocklin]

Ah. I see. Sounds like an interesting idea.

On Fri, Jul 10, 2020 at 1:45 PM Mads R. B. Kristensen <
[email protected]> wrote:

'm curious, how do tasks know when to unpack the object?

The idea is that data[k] = self.data[k] returns a proxy object that
mimics the underlying object much like Delayed. Only when the task uses
the proxy object, e.g. by calling obj[42] or obj.merge(), will
DeviceHostFile unpack the underlying object. Also keep in mind that
DeviceHostFile can be selective when deciding which type of objects it
want to return as a proxy. As a beginning we can limit DeviceHostFile to
only return proxy objects of arrays and DataFrames and then return
everything else as is.

In order to implement the proxy object I was thinking of something like:
https://github.com/GrahamDumpleton/wrapt, which also mimics type checks
like isinstance()

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
https://github.com/rapidsai/dask-cuda/issues/342#issuecomment-656881995,
or unsubscribe
https://github.com/notifications/unsubscribe-auth/AACKZTFCKYT7FOR4EH6VPGLR254WPANCNFSM4OV5JCNQ
.

[jakirkham]

FWIW there is some related work being pursued upstream in PR ( https://github.com/dask/distributed/pull/3998 ).