Repository navigation
[Question] A new approach to memory spilling #4568
Description
Question: would the Dask/Distributed community be interested in an improved memory spilling model that fixes the shortcomings of the current one but make use of proxy object wrappers?
In Dask-CUDA we have introduced a new approach to memory spilling that handles object aliasing and JIT memory un-spilling: rapidsai/dask-cuda#451
The result is memory spilling that:
- Avoids double counting: Double Counting and Issues w/Spilling #4186
- Avoids spilling of the same object multiple times.
- Avoids memory spikes because of incorrect memory tally.
- Implement just-in-time un-spilling: Delay deserialization of Data in workers until actual usage. #3998
- Support communication of spilled data so that GPU data don’t have to be un-spilled just to be spilled again as part of communication: [FEA] Allow communicating spilled data rapidsai/dask-cuda#342
- First step to enable partial spilled objects such as the spilling of individual columns in a data frame: Ability to have output dask_cudf.DataFrame not necessarily all in memory BlazingDB/blazingsql#1128
The current implement in Dask-CUDA handles CUDA device objects but it is possible to generalize to also handle spilling to disk.
The disadvantage of this approach is the use of proxy objects that get exposed to the users. The inputs to a tasks might be wrapped in a proxy object, which doesn't mimic the proxied object perfectly. E.g.:
# Type checking using instance() works as expected but direct type checking doesn't:
>>> import numpy as np
>>> from dask_cuda.proxy_object import asproxy
>>> x = np.arange(3)
>>> isinstance(asproxy(x), type(x))
True
>>> type(asproxy(x)) is type(x)
FalseBecause of this, the approach shouldn't be enabled by default but do you think that the Dask community would be interested in a generalization of this approach? Or is the proxy object hurdle too much of an issue?
cc. @mrocklin, @jrbourbeau, @quasiben
Activity
quasiben commentedon Mar 10, 2021
I think having this machine optionally available would be a nice improvement. @eric-czech do many or any sgkit workloads rely on dask spilling ?
In the past, @rabernat described an example where spilling might be impactful -- though the bulk of the conversation is about scheduler decisions/improvements
jakirkham commentedon Mar 18, 2021
Yes I think this would be great to have in Dask. I recall at one point Matt may have mentioned interest in this being upstreamed (though I could be misremembering)
eric-czech commentedon Mar 19, 2021
Hm no in the sense that we prefer to do performance testing w/ spilling disabled but yes in the sense that we still want it to work for our users who may not know how to tune chunk sizes to avoid memory problems. We're still figuring out the latter as well.
Does this spilling typically occur when usage would exceed GPU memory or when it would exceed RAM? I can imagine that spilling may become more critical for us if it is necessary to keep workflows constrained by GPU memory << RAM from failing.
cc: @tomwhite @ravwojdyla
quasiben commentedon Mar 22, 2021
@alimanfoo recently wrote up some thoughts on changing memory storage/spilling . @madsbk do you think a pluggable backend would be possible with the proxy object you are proposing here ?
madsbk commentedon Mar 22, 2021
Yes, this should 100% compatible with a pluggable backend approach. Users will be able to define what should happen when spilling between layers.
BTW, the spilling issues @alimanfoo are reporting remains me of the spilling issues we were having in GPU-BDB. That was the motivation for implementing the new spilling approach :)
mrocklin commentedon Mar 22, 2021
I think that we should talk about the problems that we're trying to solve, and then assess different solutions to those problems. This seems like one possible solution, possibly out of a few.
This approach seems interesting, but also somewhat magical/complex (although possibly that is only because I don't know much about it yet). I think that we should maybe have a longer discussion first before adopting this particular path.
@madsbk I suspect the answer is no, but do you happen to have anything that you can share where different options here were discussed that shows pros/cons, or should we start that conversation fresh?
mrocklin commentedon Mar 22, 2021
My thoughts by the way are that we should break the MutableMapping abstraction and start explicitly including things like Disk storage in the worker. I'm also fine including device memory explicitly in the worker. Part of this is described here: #4424 I expect it to be more effective and more simple long term than proxy objects.
madsbk commentedon Mar 23, 2021
Right, breaking the
MutableMappingabstraction will make it fairly easy to fix the double counting issue (#4186) and avoid memory spikes because of incorrect memory tally. However, in order to avoid spilling of the same object multiple times, I think you need some kind of proxy object but maybe it is possible to avoid exposing the proxy objects to the task/user.In order to support spilling/un-spilling of objects already passed to a task, I don't see a way around proxy objects.
madsbk commentedon Mar 23, 2021
I think the only real disadvantage is the proxy objects. I cannot think of a more simple solution that addresses all of the mentioned issues. But if the proxy objects are too much of an issue, I can look at fixing the double counting issue by breaking the
MutableMappingabstraction.mrocklin commentedon Mar 23, 2021
Yeah, I think that when a task starts running we wouldn't try to spill it to disk. I'm not sure I understand this point.
More generally, I'm not sure exactly what is being proposed here. Can you say more about what a proxy object is, how it is used, and how it affects users? I get the sense that this is a topic that is already well understood to the RAPIDS team, but may not be well understood more generally.
Does this mean that we would change the object that is passed to user functions? If so, this seems like it might open up a lot of unexpected issues.
mrocklin commentedon Mar 23, 2021
This may also provide other benefits, like allowing for async reading/writing, and improved worker scheduling of tasks that is sensitive to data that is already in fast memory, or pre-fetching data from slow memory.
madsbk commentedon Mar 23, 2021
Yes, relevant objects are wrapped in a proxy object before passed to user functions.
madsbk commentedon Mar 23, 2021
Issues with the current implementation of memory spilling
Object duplication
The same memory object can live inside a task and in the worker's
self.datasimultaneously. If the worker decides to spill an objectxin itsself.data, it will not free any memory if the task holds a reference tox. Even worse, if the output of the task includesx, the worker'sself.datawill now contain two copies ofx, a spilled and an un-spilled copy.Spilling order
The order of which we spill objects can have a drastic effect on performance. The current implementation tracks access time of each key in
self.dataand not access time of objects inself.data. This means that agetitem(v, 42)task on a collectionvwill effectively update the access time of all items invand not only the 42th item.Aliases at the task level
Consider the following two tasks:
Running the two task on a worker we get something like:
Since the current implementation of spilling works on keys and handles each keys separately, it overestimate the device memory used:
sizeof(df)*3. But even worse, if it decides to spillk2no device memory is freed sincek1still holds a reference todf2!The new spilling approach fixes this issue by wrapping identical objects in a shared
ProxyObejctthus in this casedf2in bothk1andk2will refer to the sameProxyObejct.Sharing device buffers
Consider the following code snippet:
In this case
v1andv2are separate objects and are handled separately both in the current and the new spilling implementation. However, theshuffle_group()in cudf actually returns a single device memory buffer such thatv1andv2points to the same underlying memory buffer. Thus the current implement will again overestimate the memory use and spill one of the dataframes without any effect.The new implementation takes this into account when estimating memory usage and make sure that either both dataframes are spilled or none of them are.
mrocklin commentedon Mar 23, 2021
Thank you for the detailed explanation @madsbk. This is quite helpful to motivate the change. I'm curious, in practice does this mostly come up in dataframe shuffle situations when we're low on memory? Are there other common use cases?
This would be a very large change in the abstractions used for Dask workers (user data would no longer be unchanged), and the introduction of a substantial amount of magic. Personally, if there are not very many problems that this solves, then I would prefer to try to think of other possible solutions to this problem before going down this path. My concern is that a change like this would negatively impact a number of other workloads in hard-to-understand ways.
jakirkham commentedon Mar 23, 2021
Maybe this is worth discussing at the monthly meeting next week? 😉
beckernick commentedon Mar 23, 2021
This spilling workstream has made a huge impact on quite a few workloads, largely during shuffle as @mrocklin noted. GPU-BDB (as @madsbk mentioned) is a clear example, as it comprises 30 independent workloads designed to represent common ETL/Analytics workloads. With the default spilling, in several of the workloads we end up repeatedly spilling the same data, significantly reducing performance and largely eliminating the value-add of spilling itself. Workloads that take 350-400 seconds on a constrained system can finish in 80-100 seconds with this optional spilling configuration enabled.
One additional note is that shuffle in general is memory intensive, so we often end up in a "low-memory situation" when we may not have expected one. What perhaps initially feels like a reasonable setup of a 1:1 or 1:2 ratio of data size to total memory can quickly become a low-memory scenario when large datasets need to be merged.
Default spilling allows some of these workloads like GPU-BDB to succeed on lower-memory infrastructure but at the cost of a steep performance penalty. In the GPU-BDB example, the penalty of being able to run on lower-memory systems relative to the workload data requirements is simply not worth it. With the optional spilling configuration enabled the calculation changes dramatically.
mrocklin commentedon Mar 23, 2021
I wonder if there are other ways that we can improve the memory performance of shuffle that don't impact other workloads as much. @madsbk you've mentioned before wanting to affect task prioritization. Now that annotations are in maybe that is a way that you can help to improve things while using less magic? That might be a better use of time.
Just so we're clear, the reason why I'm pushing back on this is that I have a high degree of confidence that if we do this I'll get many complaints from different groups that their types have changed, and that some strange thing that they were doing that was type dependent no longer works. For example, my understanding is that we would be breaking any user code that looked like this:
Is my understanding correct? If so then I'm a solid -1 on this change. I would be surprised if others here wouldn't also be -1 on this. This seems like a pretty clear violation of the "Dask just runs your Python code" contract. The comment of "well users can use isinstance instead" assumes that all users will read whatever doc we write about this, which I think we know isn't true.
madsbk commentedon Mar 24, 2021
Agree, I don't think anyone suggest that this will be enabled by default. The idea is to have it as an option along side regular spilling and no spilling. Users that encounter spilling issues like degraded performance and unexpected out-of-memory issues, can then try this new spilling option. I suspect that user that have significant spilling problems are motivated enough to address the
typ is intissue.As @beckernick mentioned, the current Dask-CUDA solution works very well and doesn't require any change to Distributed but I figured if we are going to generalize the solution to support spilling to disk it might make sense to do it in Distributed instead of Dask-CUDA.
mrocklin commentedon Mar 24, 2021
I'm fully committed to improving memory performance. I think that there are some general purpose approaches that we can leave on by default and that would help everyone more generally. I'm personally much more excited about these approaches. Some of those include:
The double-counting issue I don't have a solution to. This seems hard to do in general. My guess is that there are situations that even the ProxyObject solution here doesn't solve, like views of numpy arrays or objects in more nested data structures. This particular solution seems useful (as you all have clearly demonstrated), but also narrow enough that it might be better maintained outside of the main codebase.
mrocklin commentedon Mar 25, 2021
If this is very effective for shuffle workloads then maybe it's something that we could implement just for that code path? That might be a tightly scoped place to try this out more broadly.
Support cuDF's built-in spilling (#984)