Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16603#discussion_r96337114
--- Diff: core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java
---
@@ -144,23 +170,31 @@ public long acquireExecutionMemory(long required,
MemoryConsumer consumer) {
// spilling, avoid to have too many spilled files.
if (got < required) {
// Call spill() on other consumers to release memory
+ // Sort the consumers according their memory usage. So we avoid
spilling the same consumer
+ // which is just spilled in last few times and re-spilling on it
will produce many small
+ // spill files.
+ List<MemoryConsumer> sortedList = new ArrayList<>();
for (MemoryConsumer c: consumers) {
if (c != consumer && c.getUsed() > 0 && c.getMode() == mode) {
- try {
- long released = c.spill(required - got, consumer);
- if (released > 0) {
- logger.debug("Task {} released {} from {} for {}",
taskAttemptId,
- Utils.bytesToString(released), c, consumer);
- got += memoryManager.acquireExecutionMemory(required -
got, taskAttemptId, mode);
- if (got >= required) {
- break;
- }
+ sortedList.add(c);
+ }
+ }
+ Collections.sort(sortedList, new ConsumerComparator());
+ for (MemoryConsumer c: sortedList) {
+ try {
+ long released = c.spill(required - got, consumer);
+ if (released > 0) {
+ logger.debug("Task {} released {} from {} for {}",
taskAttemptId,
+ Utils.bytesToString(released), c, consumer);
+ got += memoryManager.acquireExecutionMemory(required - got,
taskAttemptId, mode);
+ if (got >= required) {
+ break;
}
- } catch (IOException e) {
- logger.error("error while calling spill() on " + c, e);
- throw new OutOfMemoryError("error while calling spill() on "
+ c + " : "
- + e.getMessage());
}
+ } catch (IOException e) {
+ logger.error("error while calling spill() on " + c, e);
+ throw new OutOfMemoryError("error while calling spill() on " +
c + " : "
+ + e.getMessage());
}
--- End diff --
As the memory usage of memory consumer is changing over time, not sure if
we use TreeSet/TreeMap for consumers, can we get the correctly sorted order
from the TreeSet/Map? In other words, the sorted order of TreeSet/Map is still
guaranteed if the elements are mutable and changing after insertion? I think it
is not.
If we are going to do sorting here anyway, a TreeMap/TreeSet might be
overkill than a list like that. Another concern is that the API of
TreeMap/TreeSet can let us find the tail set or ceiling element, but it
requires we give it an input element to compare. But we only have the required
memory number, not a memory consumer to compare.
Another concern is that TreeSet/TreeMap could return an empty set if all
elements have less memory than required size. In this case, we need to go back
to iterate all elements in the set/map to spill. It seems add more complexity.
Totally agreed that it is better to fetch the required size instead of
going from largest to smallest always. With the current list based approach, we
still can achieve that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
