[SPARK-30246][CORE] OneForOneStreamManager might leak memory in connectionTerminated #27064

hensg · 2020-01-02T01:42:20Z

What changes were proposed in this pull request?

Ensure that all StreamStates are removed from OneForOneStreamManager memory map even if there's an error trying to release buffers

Why are the changes needed?

OneForOneStreamManager may not remove all StreamStates from memory map when a connection is terminated. A RuntimeException might be thrown in StreamState$buffers.next() by one of ExternalShuffleBlockResolver$getBlockData... breaking the loop through streams.entrySet(), keeping StreamStates in memory forever leaking memory.
That may happen when an application is terminated abruptly and executors removed before the connection is terminated or if shuffleIndexCache fails to get ShuffleIndexInformation

References:
https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java#L319

https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java#L357

https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java#L195

https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java#L208

https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java#L330

Does this PR introduce any user-facing change?

No

How was this patch tested?

Unit test added

OneForOneStreamManager may not remove all StreamStates from memory map when a connection is terminated. A RuntimeException might be thrown in StreamState$buffers.next() by one of ExternalShuffleBlockResolver$getBlockData/getRddBlock... breaking the loop through streams.entrySet() This commit removes all StreamStates from memory map first and, after that, releases all buffers from each removed StreamState

common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java

…re's any exception on releasing any state.

HeartSaVioR · 2020-01-02T21:35:21Z

cc. @viirya

viirya · 2020-01-02T22:02:30Z

common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java

+      // Release all remaining buffers.
+      try {
        while (state.buffers.hasNext()) {
          ManagedBuffer buffer = state.buffers.next();


Once we got RuntimeException, don't we just fail? Is there memory leak?

spark/common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java

Lines 87 to 97 in 1b0570c

@Override

public void channelInactive() {

if (streamManager != null) {

try {

streamManager.connectionTerminated(channel);

} catch (RuntimeException e) {

logger.error("StreamManager connectionTerminated() callback failed.", e);

}

}

rpcHandler.channelInactive(reverseClient);

}

TransportRequestHandler.channelInactive will log error message and swallow the exception, so no, we don't fail because of exception thrown here.

oh I see. Then this change may also swallow the RuntimeException so channelInactive cannot log as before? Should we rethrow the same exception?

So I think there're two options here with achieving goal;

store any exception between exceptions (as we'll try to release from all the streams) and rethrow, then channelInactive will log it. other exceptions should be logged as well but maybe at here (or a new exception containing all exceptions and throw it instead).

catch them, and log and swallow at here.

Technically the change from logging side would be logger name which may not a big deal, but it's also valid concern if we think channelInactive is the one to decide how to handle the exception.

yea, I think it might make the caller (channelInactive or others if any) thinks everything is fine inside connectionTerminated. Maybe not big deal for now, but sounds a potential concern.

We might have multiple exception during releasing buffers. We can rethrow a RuntimeException if any exception happens during releasing buffers.

OK thanks for the input. @hensg Could you follow up the comment above? Thanks!

viirya · 2020-01-02T22:28:53Z

ok to test.

SparkQA · 2020-01-03T00:52:50Z

Test build #116058 has finished for PR 27064 at commit a9ebe4f.

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

012huang

there exists an case, if the application has already finished(failed), executors' info of the app has been cleaned, but the connectionTerminated be called delay, when execute this code, it will look up the buffer with appId and execId , so the buffer cannot be got and released. this happend in spark 2.4.3 as I report and I also make a pr, pls help review(#27060), thanks

HeartSaVioR · 2020-01-06T03:53:15Z

@012huang I feel it's orthogonal to this. Could you please cc. to some committers modified around the code having bug in your PR?

HeartSaVioR

LGTM. cc. @viirya Could you please take a look again? Thanks in advance.

SparkQA · 2020-01-09T10:03:14Z

Test build #116383 has finished for PR 27064 at commit 410651d.

This patch fails Java style tests.
This patch merges cleanly.
This patch adds no public classes.

[ERROR] src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java:[123] (sizes) LineLength: Line is longer than 100 characters (found 106).

SparkQA · 2020-01-09T12:40:17Z

Test build #116384 has finished for PR 27064 at commit ac58798.

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

HeartSaVioR · 2020-01-09T14:57:15Z

retest this, please

SparkQA · 2020-01-09T17:00:53Z

Test build #116398 has finished for PR 27064 at commit ac58798.

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

viirya · 2020-01-09T19:16:16Z

retest this please

SparkQA · 2020-01-09T21:14:32Z

Test build #116414 has finished for PR 27064 at commit ac58798.

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

viirya · 2020-01-09T21:22:22Z

looks like a valid failure test_memory_limit (pyspark.tests.test_worker.WorkerMemoryTest)? It continues to fail on this.

HeartSaVioR · 2020-01-09T22:56:54Z

@viirya The test fails from other PRs as well - please refer #26201 (comment)

Looks like other PRs succeed to build because the pyspark test seems run conditionally.

viirya · 2020-01-09T23:19:50Z

@HeartSaVioR hmm, I just ran the test locally and it passed.

HeartSaVioR · 2020-01-09T23:23:01Z

Yeah same for me as well, but I don't have pypy. Installing now.

EDIT: and no luck. It works with my local. Maybe there's something with Jenkins worker machine?

HeartSaVioR · 2020-01-09T23:29:02Z

EDIT: It seems to fail intermittently, so not able to reproduce the original failure.

~~Oh wait I'm seeing failure "after" installing "resource", but the test fails elsewhere which is odd.~~

Running PySpark tests. Output is in /Users/jlim/WorkArea/ScalaProjects/spark/python/unit-tests.log
Will test against the following Python executables: ['/usr/local/bin/python3', 'python2.7', 'pypy']
Will test the following Python tests: ['pyspark.tests.test_worker']
Starting test(/usr/local/bin/python3): pyspark.tests.test_worker
Starting test(pypy): pyspark.tests.test_worker
Starting test(python2.7): pyspark.tests.test_worker
test_memory_limit (pyspark.tests.test_worker.WorkerMemoryTest) ... Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
/Users/jlim/WorkArea/ScalaProjects/spark/python/pyspark/context.py:219: DeprecationWarning: Support for Python 2 and Python 3 prior to version 3.6 is deprecated as of Spark 3.0. See also the plan for dropping Python 2 support at https://spark.apache.org/news/plan-for-dropping-python-2-support.html.
  DeprecationWarning)
[Stage 0:>                                                          (0 + 1) / 1]Current mem limits: 9223372036854775807 of max 9223372036854775807

Setting mem limits to 1048576 of max 1048576

/Users/jlim/WorkArea/ScalaProjects/spark/python/pyspark/tests/test_worker.py:190: RuntimeWarning: Parent module 'pyspark.tests' not found while handling absolute import
  import resource
ok
test_reuse_worker_of_parallelize_xrange (pyspark.tests.test_worker.WorkerReuseTest) ... FAIL
test_accumulator_when_reuse_worker (pyspark.tests.test_worker.WorkerTests) ... ok
test_after_exception (pyspark.tests.test_worker.WorkerTests) ... ok
test_after_jvm_exception (pyspark.tests.test_worker.WorkerTests) ... ok
test_cancel_task (pyspark.tests.test_worker.WorkerTests) ... /Users/jlim/WorkArea/ScalaProjects/spark/python/pyspark/tests/test_worker.py:45: RuntimeWarning: Parent module 'pyspark.tests' not found while handling absolute import
  import os
/Users/jlim/WorkArea/ScalaProjects/spark/python/pyspark/tests/test_worker.py:46: RuntimeWarning: Parent module 'pyspark.tests' not found while handling absolute import
  import time
ok
test_python_exception_non_hanging (pyspark.tests.test_worker.WorkerTests) ... ok
test_reuse_worker_after_take (pyspark.tests.test_worker.WorkerTests) ... ok
test_with_different_versions_of_python (pyspark.tests.test_worker.WorkerTests) ... ok

======================================================================
Finished test(/usr/local/bin/python3): pyspark.tests.test_worker (15s)
FAIL: test_reuse_worker_of_parallelize_xrange (pyspark.tests.test_worker.WorkerReuseTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/jlim/WorkArea/ScalaProjects/spark/python/pyspark/tests/test_worker.py", line 176, in test_reuse_worker_of_parallelize_xrange
    self.assertTrue(pid in previous_pids)
AssertionError: False is not true

----------------------------------------------------------------------
Ran 9 tests in 15.360s

FAILED (failures=1)

### What changes were proposed in this pull request? This patch increases the memory limit in the test 'test_memory_limit' from 1m to 8m. Credit to srowen and HyukjinKwon to provide the idea of suspicion and guide how to fix. ### Why are the changes needed? We observed consistent Pyspark test failures on multiple PRs (apache#26955, apache#26201, apache#27064) which block the PR builds whenever the test is included. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Jenkins builds passed in WIP PR (apache#27159) Closes apache#27162 from HeartSaVioR/SPARK-30480. Authored-by: Jungtaek Lim (HeartSaVioR) <[email protected]> Signed-off-by: HyukjinKwon <[email protected]>

HyukjinKwon · 2020-01-10T06:32:39Z

retest this please

SparkQA · 2020-01-10T08:05:01Z

Test build #116453 has finished for PR 27064 at commit ac58798.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-01-10T08:16:06Z

retest this, please

SparkQA · 2020-01-10T10:23:52Z

Test build #116472 has finished for PR 27064 at commit ac58798.

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

SparkQA · 2020-01-12T14:54:52Z

Test build #116556 has finished for PR 27064 at commit d76483f.

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

SparkQA · 2020-01-14T00:41:54Z

Test build #116663 has finished for PR 27064 at commit b02e48a.

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

viirya

looks good. cc @cloud-fan

viirya · 2020-01-14T00:50:06Z

common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java

+              buffer.release();
+            }
          }
+        } catch (RuntimeException e) {


Is this the only exception type we need to deal with here?

Other kind of exception would be Error which TransportRequestHandler.channelInactive is also not catching and let the process fail.

common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java

SparkQA · 2020-01-14T23:31:19Z

Test build #116727 has finished for PR 27064 at commit 0fde4c9.

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

dongjoon-hyun · 2020-01-15T02:36:32Z

cc @vanzin , too.

vanzin · 2020-01-15T21:26:51Z

Merging to master / 2.4.

…ctionTerminated Ensure that all StreamStates are removed from OneForOneStreamManager memory map even if there's an error trying to release buffers OneForOneStreamManager may not remove all StreamStates from memory map when a connection is terminated. A RuntimeException might be thrown in StreamState$buffers.next() by one of ExternalShuffleBlockResolver$getBlockData... **breaking the loop through streams.entrySet(), keeping StreamStates in memory forever leaking memory.** That may happen when an application is terminated abruptly and executors removed before the connection is terminated or if shuffleIndexCache fails to get ShuffleIndexInformation References: https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java#L319 https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java#L357 https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java#L195 https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java#L208 https://github.com/apache/spark/blob/ee050ddbc6eb6bc08c7751a0eb00e7a05b011b52/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java#L330 No Unit test added Closes #27064 from hensg/SPARK-30246. Lead-authored-by: Henrique Goulart <[email protected]> Co-authored-by: Henrique Goulart <[email protected]> Signed-off-by: Marcelo Vanzin <[email protected]> (cherry picked from commit d42cf45) Signed-off-by: Marcelo Vanzin <[email protected]>

vanzin · 2020-01-15T21:37:24Z

FYI: fixed a trivial conflict in 2.4, and had to add a couple of things to the test:

import the Assert class
add a couple of @SuppressWarnings otherwise compilation was failing. Not sure why master did not complain...

dongjoon-hyun · 2020-01-15T23:19:59Z

Thank you all!

HeartSaVioR reviewed Jan 2, 2020

View reviewed changes

common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java Outdated Show resolved Hide resolved

Add try-catch block to release buffers for all removedStates even the…

a9ebe4f

…re's any exception on releasing any state.

hensg changed the title ~~[SPARK-30246]OneForOneStreamManager might leak memory in connectionTerminated~~ [SPARK-30246]OneForOneStreamManager may leak memory in connectionTerminated Jan 2, 2020

hensg changed the title ~~[SPARK-30246]OneForOneStreamManager may leak memory in connectionTerminated~~ [SPARK-30246]OneForOneStreamManager might leak memory in connectionTerminated Jan 2, 2020

viirya reviewed Jan 2, 2020

View reviewed changes

012huang reviewed Jan 3, 2020

View reviewed changes

HeartSaVioR approved these changes Jan 8, 2020

View reviewed changes

Rethrow exception when fails to release all StreamState buffers

410651d

Fix linter linelength error

ac58798

[ERROR] src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java:[123] (sizes) LineLength: Line is longer than 100 characters (found 106).

This was referenced Jan 10, 2020

[MINOR][BUILD][PYSPARK] Disable test_memory_limit test temporarily #27158

Closed

[SPARK-30480][PYSPARK][TESTS] Fix 'test_memory_limit' on pyspark test #27162

Closed

Remove extra blank line

b02e48a

viirya approved these changes Jan 14, 2020

View reviewed changes

cloud-fan reviewed Jan 14, 2020

View reviewed changes

common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java Outdated Show resolved Hide resolved

Store first exception and re-throw it

0fde4c9

hensg requested a review from cloud-fan January 14, 2020 23:32

dongjoon-hyun changed the title ~~[SPARK-30246]OneForOneStreamManager might leak memory in connectionTerminated~~ [SPARK-30246][CORE] OneForOneStreamManager might leak memory in connectionTerminated Jan 15, 2020

dongjoon-hyun added the SPARK CORE label Jan 15, 2020

cloud-fan approved these changes Jan 15, 2020

View reviewed changes

vanzin closed this in d42cf45 Jan 15, 2020

vanzin mentioned this pull request Jan 15, 2020

[SPARK-30246][CORE][SHUFFLE] fix spark external shuffle service memory leak #27201

Closed

	@Override
	public void channelInactive() {
	if (streamManager != null) {
	try {
	streamManager.connectionTerminated(channel);
	} catch (RuntimeException e) {
	logger.error("StreamManager connectionTerminated() callback failed.", e);
	}
	}
	rpcHandler.channelInactive(reverseClient);
	}

[SPARK-30246][CORE] OneForOneStreamManager might leak memory in connectionTerminated #27064

[SPARK-30246][CORE] OneForOneStreamManager might leak memory in connectionTerminated #27064

Uh oh!

Conversation

hensg commented Jan 2, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

Uh oh!

HeartSaVioR commented Jan 2, 2020

Uh oh!

viirya Jan 2, 2020

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Jan 2, 2020

Choose a reason for hiding this comment

Uh oh!

viirya Jan 2, 2020

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Jan 2, 2020

Choose a reason for hiding this comment

Uh oh!

viirya Jan 9, 2020

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Jan 9, 2020

Choose a reason for hiding this comment

Uh oh!

viirya commented Jan 2, 2020

Uh oh!

SparkQA commented Jan 3, 2020

Uh oh!

012huang left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Jan 6, 2020

Uh oh!

HeartSaVioR left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 9, 2020

Uh oh!

SparkQA commented Jan 9, 2020

Uh oh!

HeartSaVioR commented Jan 9, 2020

Uh oh!

SparkQA commented Jan 9, 2020

Uh oh!

viirya commented Jan 9, 2020

Uh oh!

SparkQA commented Jan 9, 2020

Uh oh!

viirya commented Jan 9, 2020

Uh oh!

HeartSaVioR commented Jan 9, 2020

Uh oh!

viirya commented Jan 9, 2020

Uh oh!

HeartSaVioR commented Jan 9, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

HeartSaVioR commented Jan 9, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

HyukjinKwon commented Jan 10, 2020

Uh oh!

SparkQA commented Jan 10, 2020

Uh oh!

HeartSaVioR commented Jan 10, 2020

Uh oh!

SparkQA commented Jan 10, 2020

Uh oh!

SparkQA commented Jan 12, 2020

Uh oh!

SparkQA commented Jan 14, 2020

hensg commented Jan 2, 2020 •

edited

Loading

012huang left a comment •

edited

Loading

HeartSaVioR commented Jan 9, 2020 •

edited

Loading

HeartSaVioR commented Jan 9, 2020 •

edited

Loading

HeartSaVioR Jan 14, 2020 •

edited

Loading