[SPARK-10983] Unified memory manager #9084

andrewor14 · 2015-10-13T01:08:15Z

This patch unifies the memory management of the storage and execution regions such that either side can borrow memory from each other. When memory pressure arises, storage will be evicted in favor of execution. To avoid regressions in cases where storage is crucial, we dynamically allocate a fraction of space for storage that execution cannot evict. Several configurations are introduced:

spark.memory.fraction (default 0.75): fraction of the heap space used for execution and storage. The lower this is, the more frequently spills and cached data eviction occur. The purpose of this config is to set aside memory for internal metadata, user data structures, and imprecise size estimation in the case of sparse, unusually large records.
spark.memory.storageFraction (default 0.5): size of the storage region within the space set aside by spark.memory.fraction. Cached data may only be evicted if total storage exceeds this region.
spark.memory.useLegacyMode (default false): whether to use the memory management that existed in Spark 1.5 and before. This is mainly for backward compatibility.

For a detailed description of the design, see SPARK-10000. This patch builds on top of the MemoryManager interface introduced in #9000.

As of this commit, acquiring execution memory may cause cached blocks to be evicted. There are still a couple of TODOs, notably figuring out what the `maxExecutionMemory` and `maxStorageMemory` should actually be. They may not be fixed since either side can now borrow from the other.

Without this, we cannot both avoid deadlocks and race conditions, because each individual component ShuffleMemoryManager, MemoryStore and MemoryManager all have their own respective locks. This commit allows us to simplify several unintuitive control flows that were introduced to avoid acquiring locks in different orders. Since we have only one lock now, these code blocks can be significantly simplified. A forseeable downside to this is parallelism is reduced, but memory acquisitions and releases don't actually occur that frequently, so performance should not suffer noticeably. Additional investigations about this should ensue.

Previously, ShuffleMemoryManager will allow each task to acquire up to 1/N of the entire storage + execution region. What we want is more like 1/N of the space not occupied by storage, since the "max" now varies over time.

This happened because we were still calling `this.notifyAll()` in ShuffleMemoryManager when we were holding the `memoryManager` lock. Easy fix.

Tests are passing in this commit, but there are follow-ups that need to be done (see TODOs added in this commit). More tests will be added in the future.

As of this commit all *MemoryManagerSuite's are documented and pass tests.

TaskContext was not stubbed correctly in ShuffleMemoryManagerSuite. In particular, this patch added some code that does some things to the active TaskMetrics, but this was not part of the mocked TaskContext object.

…manager

SparkQA · 2015-10-13T03:46:15Z

Test build #43602 has finished for PR 9084 at commit 01ff533.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

JoshRosen · 2015-10-13T04:08:45Z

core/src/main/scala/org/apache/spark/SparkConf.scala

+  // Deprecation message for memory fraction configs used in the old memory management model
+  private val deprecatedMemoryFractionMessage =
+    "As of Spark 1.6, execution and storage memory management are unified. " +
+      "All memory fractions used in the old model are now deprecated and no longer read."


The old configurations will still be respected in legacy mode, so this is slightly ambiguous / confusing. Is there an easy way to avoid the warning if the legacy mode configuration is turned on? If not, I suppose we could just expand the deprecation message to mention this corner case, perhaps by just appending a "(unless spark.XX.YY is enabled)" at the end.

I can print a different warning if legacy mode is on

SparkQA · 2015-10-13T04:11:22Z

Test build #43607 has finished for PR 9084 at commit 059ae0d.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

JoshRosen · 2015-10-13T04:12:29Z

core/src/main/scala/org/apache/spark/memory/MemoryManager.scala

@@ -72,46 +92,62 @@ private[spark] abstract class MemoryManager {
  def acquireUnrollMemory(


Given that acquireUnrollMemory appears to act as a synonym for acquireStorageMemory in the current implementation, it might be worth adding a brief comment above this method to explain that this extra method exists in order to give us the future flexibility to account for unroll memory differently in the future.

actually, it's more than a synonym. In StaticMemoryManager it's required to preserve existing behavior where unrolling doesn't evict all the blocks.

JoshRosen · 2015-10-13T05:03:30Z

It looks like the test failures are occurring in the external sorter suites. Some of these tests exercise spilling-related logic (such as ensuring that spill files are cleaned up); in order to induce spilling, these tests configured the legacy memory fractions to be really small. In order to fix these tests, we can either enable legacy mode for those tests only or we can modify the tests to set the new configurations. Ideally these sorter unit tests wouldn't be relying on memory management behavior in order to test spill cleanup, but they were written a long time ago. We could refactor these tests, but I'd prefer a band-aid solution for now.

Many of these tests relied on setting a very low shuffle memory fraction. Since that config is now deprecated and not read, the tests now fail because things aren't spilling. This approach is, however, quite brittle as it depends on the heap size of the test JVM. Instead, we should explicitly set a limit on the memory used.

Unfortunately not all affected tests can be easily rewritten. Instead, this commit adds TODO's in places where the tests may not actually do anything so we can fix them in the future (SPARK-11078).

…manager

SparkQA · 2015-10-13T08:16:23Z

Test build #43622 has finished for PR 9084 at commit 24a391c.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-10-13T10:09:47Z

Test build #43631 has finished for PR 9084 at commit face129.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

andrewor14 · 2015-10-13T16:27:32Z

retest this please

SparkQA · 2015-10-13T16:54:08Z

Test build #43646 has finished for PR 9084 at commit face129.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-10-13T18:49:01Z

Test build #1890 has finished for PR 9084 at commit face129.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-10-13T18:54:06Z

Test build #1888 has finished for PR 9084 at commit face129.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-10-13T18:55:04Z

Test build #1889 has finished for PR 9084 at commit face129.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-10-13T19:10:49Z

Test build #1891 has finished for PR 9084 at commit face129.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

JoshRosen · 2015-10-13T20:49:33Z

LGTM. I'm going to merge this now. There's some followup tasks to do in terms of refactoring some old spilling tests, but there's followup tasks for those and we'll do them soon.

This patch unifies the memory management of the storage and execution regions such that either side can borrow memory from each other. When memory pressure arises, storage will be evicted in favor of execution. To avoid regressions in cases where storage is crucial, we dynamically allocate a fraction of space for storage that execution cannot evict. Several configurations are introduced: - **spark.memory.fraction (default 0.75)**: fraction of the heap space used for execution and storage. The lower this is, the more frequently spills and cached data eviction occur. The purpose of this config is to set aside memory for internal metadata, user data structures, and imprecise size estimation in the case of sparse, unusually large records. - **spark.memory.storageFraction (default 0.5)**: size of the storage region within the space set aside by `spark.memory.fraction`. Cached data may only be evicted if total storage exceeds this region. - **spark.memory.useLegacyMode (default false)**: whether to use the memory management that existed in Spark 1.5 and before. This is mainly for backward compatibility. For a detailed description of the design, see [SPARK-10000](https://issues.apache.org/jira/browse/SPARK-10000). This patch builds on top of the `MemoryManager` interface introduced in apache#9000. Author: Andrew Or <andrew@databricks.com> Closes apache#9084 from andrewor14/unified-memory-manager.

#9084 uncovered that many tests that test spilling don't actually spill. This is a follow-up patch to fix that to ensure our unit tests actually catch potential bugs in spilling. The size of this patch is inflated by the refactoring of `ExternalSorterSuite`, which had a lot of duplicate code and logic. Author: Andrew Or <andrew@databricks.com> Closes #9124 from andrewor14/spilling-tests.

pzz2011 · 2016-08-21T13:41:30Z

core/src/main/scala/org/apache/spark/memory/MemoryManager.scala

@@ -40,6 +41,10 @@ private[spark] abstract class MemoryManager {
    _memoryStore
  }

+  // Amount of execution/storage memory in use, accesses must be synchronized on `this`
+  protected var _executionMemoryUsed: Long = 0


why here is protected?

@pzz2011 you're making several comments on old PRs. Generally people won't see that and it's not the place for discussion anyway. If you can formulate a specific question beyond "why is the code this way?" ask on user@.

Andrew Or added 15 commits October 9, 2015 13:32

Add UnifiedMemoryManager skeleton

6481bc1

Use a dynamic max for execution and storage

cc5f64c

Previously, ShuffleMemoryManager will allow each task to acquire up to 1/N of the entire storage + execution region. What we want is more like 1/N of the space not occupied by storage, since the "max" now varies over time.

Fix notifyAll() IllegalMonitorStateException

ad8a6c4

This happened because we were still calling `this.notifyAll()` in ShuffleMemoryManager when we were holding the `memoryManager` lock. Easy fix.

Minor: update a few comments

0dc9a95

Register blocks evicted by execution

5a4ffb9

Minor: more comment updates

b519540

Add tests for UnifiedMemoryManager + TODOs

a65799e

Tests are passing in this commit, but there are follow-ups that need to be done (see TODOs added in this commit). More tests will be added in the future.

Clean up test code, resolve TODOs

6e913a5

As of this commit all *MemoryManagerSuite's are documented and pass tests.

Fix tests

3eef5a4

TaskContext was not stubbed correctly in ShuffleMemoryManagerSuite. In particular, this patch added some code that does some things to the active TaskMetrics, but this was not part of the mocked TaskContext object.

Minor: Add TODOs, fix a few comments

c56600b

Merge branch 'master' of github.com:apache/spark into unified-memory-…

01ff533

…manager

Documentation + enable by default

93c3cef

Log deprecation warning for old memory fraction configs

059ae0d

JoshRosen reviewed Oct 13, 2015
View reviewed changes

Andrew Or added 4 commits October 13, 2015 00:11

Remove more usages of old deprecated configs

13898b5

Unfortunately not all affected tests can be easily rewritten. Instead, this commit adds TODO's in places where the tests may not actually do anything so we can fix them in the future (SPARK-11078).

Warn against deprecated configs only in non-legacy mode

4b64846

Address comments

4f70806

Merge branch 'master' of github.com:apache/spark into unified-memory-…

face129

…manager

asfgit closed this in b3ffac5 Oct 13, 2015

andrewor14 deleted the unified-memory-manager branch October 13, 2015 20:53

andrewor14 mentioned this pull request Oct 14, 2015

[SPARK-11078] Ensure spilling tests actually spill #9124

Closed

pzz2011 reviewed Aug 21, 2016
View reviewed changes

		@@ -72,46 +92,62 @@ private[spark] abstract class MemoryManager {
		def acquireUnrollMemory(

[SPARK-10983] Unified memory manager #9084

[SPARK-10983] Unified memory manager #9084

Uh oh!

Conversation

andrewor14 commented Oct 13, 2015

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

JoshRosen Oct 13, 2015

Choose a reason for hiding this comment

Uh oh!

andrewor14 Oct 13, 2015

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

JoshRosen Oct 13, 2015

Choose a reason for hiding this comment

Uh oh!

andrewor14 Oct 13, 2015

Choose a reason for hiding this comment

Uh oh!

JoshRosen commented Oct 13, 2015

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

andrewor14 commented Oct 13, 2015

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

SparkQA commented Oct 13, 2015

Uh oh!

JoshRosen commented Oct 13, 2015

Uh oh!

pzz2011 Aug 21, 2016

Choose a reason for hiding this comment

Uh oh!

srowen Aug 21, 2016

Choose a reason for hiding this comment

Uh oh!

Uh oh!