[SPARK-26604][CORE] Clean up channel registration for StreamManager #23521

viirya · 2019-01-11T15:30:38Z

What changes were proposed in this pull request?

Now in TransportRequestHandler.processStreamRequest, when a stream request is processed, the stream id is not registered with the current channel in stream manager. It should do that so in case of that the channel gets terminated we can remove associated streams of stream requests too.

This also cleans up channel registration in StreamManager. Since StreamManager doesn't register channel but only OneForOneStreamManager does it, this removes registerChannel from StreamManager. When OneForOneStreamManager goes to register stream, it will also register channel for the stream.

How was this patch tested?

Existing tests.

viirya · 2019-01-11T15:30:47Z

cc @cloud-fan

SparkQA · 2019-01-11T15:36:21Z

Test build #101089 has finished for PR 23521 at commit 6e35249.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-01-11T15:46:46Z

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

@@ -127,6 +127,7 @@ private void processStreamRequest(final StreamRequest req) {
    ManagedBuffer buf;
    try {
      buf = streamManager.openStream(req.streamId);
+      streamManager.registerChannel(channel, req.streamId);


I'm wondering when we should do this. There are many kinds of requests, and currently only chunk fetch request does it.

For stream request and get chunk request, they are recorded in stream manager. Registering channel for them is to make sure they are removed from stream manager when the channel is inactive. For other types of requests, I don't find they are recorded like that.

SparkQA · 2019-01-11T20:25:27Z

Test build #101090 has finished for PR 23521 at commit ba9c27e.

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

cloud-fan · 2019-01-14T05:28:43Z

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

+   * of the stream. This is similar to {@link #registerChannel(Channel, long)} method, but the
+   * <code>streamId</code> argument is for the stream in response to a stream() request.
+   */
+  public void registerChannel(Channel channel, String streamId) { }


Now things get tricky here. There are 2 different kind of stream requests:

to download jar and other files

to fetch data blocks (introduced at [SPARK-19659] Fetch big blocks to disk when shuffle-read. #16989)

Which one do we need to register channel?

For stream requests from fetching data blocks, the streams will be registered by RPC handler. So registering the channels helps to remove registered streams when the channels get inactive.

For stream requests from downloading jar and files, there is no such stream registration.

Are you saying TransportRequestHandler.processStreamRequest is only used to deal with stream request to fetch blocks?

TransportRequestHandler.processStreamRequest is used for both. But the streams are not registered there. It is registered by NettyBlockRpcServer when processing OpenBlocks message.

registerStream is only in OneForOneStreamManager, it's super weird that registerChannel needs to be called after registerStream, but registerChannel is in the parent StreamManager.

Hmm, I agree. I think it is also good to clean this up a bit. I will do a change later.

viirya · 2019-01-14T14:51:43Z

@cloud-fan Since StreamManager doesn't register channel but only OneForOneStreamManager does it, I remove registerChannel from StreamManager. When OneForOneStreamManager goes to serve chunk or stream request, it will register channel for the stream.

SparkQA · 2019-01-14T14:55:22Z

Test build #101187 has finished for PR 23521 at commit 9082f01.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-01-14T15:08:46Z

Test build #101188 has started for PR 23521 at commit 6b028a9.

shaneknapp · 2019-01-14T19:00:21Z

test this please

shaneknapp · 2019-01-14T19:08:51Z

test this please

SparkQA · 2019-01-14T19:23:39Z

Test build #101198 has finished for PR 23521 at commit 6b028a9.

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

cloud-fan · 2019-01-15T05:12:38Z

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

+   * Associates a stream with a single client connection, which is guaranteed to be the only reader
+   * of the stream. Once the connection is closed, the stream will never be used again, enabling
+   * cleanup by `connectionTerminated`.
+   */
  public void registerChannel(Channel channel, long streamId) {


this can be private?

cloud-fan · 2019-01-15T05:16:03Z

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

@@ -42,9 +42,10 @@
   * The returned ManagedBuffer will be release()'d after being written to the network.
   *
   * @param streamId id of a stream that has been previously registered with the StreamManager.
+   * @param channel The connection used to serve chunk request.


let's say more about this parameter, especially how it should be used. IIUC we need to track the channel states, and do some cleanup if the channel is inactive.

SparkQA · 2019-01-15T05:22:18Z

Test build #101211 has finished for PR 23521 at commit 53e9c6e.

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

viirya · 2019-01-15T09:16:14Z

@cloud-fan Moved channel registration to where we register the stream. Few tests are modified.

SparkQA · 2019-01-15T09:18:50Z

Test build #101243 has finished for PR 23521 at commit be01666.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-01-15T09:33:48Z

Test build #101244 has finished for PR 23521 at commit 7c1e13d.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2019-01-15T09:50:03Z

Oops...

cloud-fan · 2019-01-15T10:37:45Z

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

      this.appId = appId;
      this.buffers = Preconditions.checkNotNull(buffers);
+      this.associatedChannel = channel;


associatedChannel can be final now.

cloud-fan · 2019-01-15T10:40:00Z

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

    return myStreamId;
  }

+  @VisibleForTesting
+  public int streamStateSize() {


nit: numStreamStates

SparkQA · 2019-01-15T14:31:29Z

Test build #101245 has finished for PR 23521 at commit 8cd705e.

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

SparkQA · 2019-01-15T15:15:16Z

Test build #101251 has finished for PR 23521 at commit 7eed779.

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

viirya · 2019-01-15T15:16:58Z

retest this please.

SparkQA · 2019-01-15T19:53:49Z

Test build #101269 has finished for PR 23521 at commit 7eed779.

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

cloud-fan · 2019-01-16T02:58:18Z

thanks, merging to master!

## What changes were proposed in this pull request? Now in `TransportRequestHandler.processStreamRequest`, when a stream request is processed, the stream id is not registered with the current channel in stream manager. It should do that so in case of that the channel gets terminated we can remove associated streams of stream requests too. This also cleans up channel registration in `StreamManager`. Since `StreamManager` doesn't register channel but only `OneForOneStreamManager` does it, this removes `registerChannel` from `StreamManager`. When `OneForOneStreamManager` goes to register stream, it will also register channel for the stream. ## How was this patch tested? Existing tests. Closes apache#23521 from viirya/SPARK-26604. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

felixcheung · 2019-03-07T05:33:58Z

I think we need to backport this into branch-2.4

viirya · 2019-03-07T05:41:28Z

If it can't be directly merged to branch-2.4, I can make a PR for it.

abellina · 2019-03-07T15:53:17Z

I am taking a look, but it may not be until later today that I'd have a PR up. It's mostly clean but the 2.4 branch doesn't have the ChunkFetchRequestHandler, and then there's one extra registerChannel that needs to go away + testing. @viirya if you want to do PR sooner, please feel free.

Now in `TransportRequestHandler.processStreamRequest`, when a stream request is processed, the stream id is not registered with the current channel in stream manager. It should do that so in case of that the channel gets terminated we can remove associated streams of stream requests too. This also cleans up channel registration in `StreamManager`. Since `StreamManager` doesn't register channel but only `OneForOneStreamManager` does it, this removes `registerChannel` from `StreamManager`. When `OneForOneStreamManager` goes to register stream, it will also register channel for the stream. Existing tests. Closes apache#23521 from viirya/SPARK-26604. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

abellina · 2019-03-07T20:47:34Z

#24013

…treamManager ## What changes were proposed in this pull request? This is mostly a clean backport of #23521 to branch-2.4 ## How was this patch tested? I've tested this with a hack in `TransportRequestHandler` to force `ChunkFetchRequest` to get dropped. Then making a number of `ExternalShuffleClient.fetchChunk` requests (which `OpenBlocks` then `ChunkFetchRequest`) and closing out of my test harness. A heap dump later reveals that the `StreamState` references are unreachable. I haven't run this through the unit test suite, but doing that now. Wanted to get this up as I think folks are waiting for it for 2.4.1 Closes #24013 from abellina/SPARK-26604_cherry_pick_2_4. Lead-authored-by: Liang-Chi Hsieh <viirya@gmail.com> Co-authored-by: Alessandro Bellina <abellina@yahoo-inc.com> Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com>

…treamManager ## What changes were proposed in this pull request? This is mostly a clean backport of #23521 to branch-2.4 ## How was this patch tested? I've tested this with a hack in `TransportRequestHandler` to force `ChunkFetchRequest` to get dropped. Then making a number of `ExternalShuffleClient.fetchChunk` requests (which `OpenBlocks` then `ChunkFetchRequest`) and closing out of my test harness. A heap dump later reveals that the `StreamState` references are unreachable. I haven't run this through the unit test suite, but doing that now. Wanted to get this up as I think folks are waiting for it for 2.4.1 Closes #24013 from abellina/SPARK-26604_cherry_pick_2_4. Lead-authored-by: Liang-Chi Hsieh <viirya@gmail.com> Co-authored-by: Alessandro Bellina <abellina@yahoo-inc.com> Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com> (cherry picked from commit 216eeec) Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com>

…treamManager ## What changes were proposed in this pull request? This is mostly a clean backport of apache#23521 to branch-2.4 ## How was this patch tested? I've tested this with a hack in `TransportRequestHandler` to force `ChunkFetchRequest` to get dropped. Then making a number of `ExternalShuffleClient.fetchChunk` requests (which `OpenBlocks` then `ChunkFetchRequest`) and closing out of my test harness. A heap dump later reveals that the `StreamState` references are unreachable. I haven't run this through the unit test suite, but doing that now. Wanted to get this up as I think folks are waiting for it for 2.4.1 Closes apache#24013 from abellina/SPARK-26604_cherry_pick_2_4. Lead-authored-by: Liang-Chi Hsieh <viirya@gmail.com> Co-authored-by: Alessandro Bellina <abellina@yahoo-inc.com> Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com>

…treamManager ## What changes were proposed in this pull request? This is mostly a clean backport of apache/spark#23521 to branch-2.4 ## How was this patch tested? I've tested this with a hack in `TransportRequestHandler` to force `ChunkFetchRequest` to get dropped. Then making a number of `ExternalShuffleClient.fetchChunk` requests (which `OpenBlocks` then `ChunkFetchRequest`) and closing out of my test harness. A heap dump later reveals that the `StreamState` references are unreachable. I haven't run this through the unit test suite, but doing that now. Wanted to get this up as I think folks are waiting for it for 2.4.1 Closes #24013 from abellina/SPARK-26604_cherry_pick_2_4. Lead-authored-by: Liang-Chi Hsieh <viirya@gmail.com> Co-authored-by: Alessandro Bellina <abellina@yahoo-inc.com> Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com> (cherry picked from commit 216eeec)

## What changes were proposed in this pull request? Now in `TransportRequestHandler.processStreamRequest`, when a stream request is processed, the stream id is not registered with the current channel in stream manager. It should do that so in case of that the channel gets terminated we can remove associated streams of stream requests too. This also cleans up channel registration in `StreamManager`. Since `StreamManager` doesn't register channel but only `OneForOneStreamManager` does it, this removes `registerChannel` from `StreamManager`. When `OneForOneStreamManager` goes to register stream, it will also register channel for the stream. ## How was this patch tested? Existing tests. Closes apache#23521 from viirya/SPARK-26604. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit cf133e6)

Now in `TransportRequestHandler.processStreamRequest`, when a stream request is processed, the stream id is not registered with the current channel in stream manager. It should do that so in case of that the channel gets terminated we can remove associated streams of stream requests too. This also cleans up channel registration in `StreamManager`. Since `StreamManager` doesn't register channel but only `OneForOneStreamManager` does it, this removes `registerChannel` from `StreamManager`. When `OneForOneStreamManager` goes to register stream, it will also register channel for the stream. Existing tests. Closes apache#23521 from viirya/SPARK-26604. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit cf133e6) RB=1586979 BUG=LIHADOOP-44658 G=superfriends-reviewers R=mshen,yezhou,fli,edlu A=mshen

cloud-fan reviewed Jan 11, 2019

View reviewed changes

Register channel for stream request.

ba9c27e

viirya force-pushed the SPARK-26604 branch from 6e35249 to ba9c27e Compare January 11, 2019 15:56

cloud-fan reviewed Jan 14, 2019

View reviewed changes

viirya changed the title ~~[SPARK-26604][CORE] Register channel for stream request~~ [SPARK-26604][CORE] Clean up channel registration for StreamManager Jan 14, 2019

Remove registerChannel from StreamManager.

6b028a9

viirya force-pushed the SPARK-26604 branch from 9082f01 to 6b028a9 Compare January 14, 2019 15:05

Fix java style.

53e9c6e

cloud-fan reviewed Jan 15, 2019

View reviewed changes

viirya force-pushed the SPARK-26604 branch from be01666 to 7c1e13d Compare January 15, 2019 09:21

viirya force-pushed the SPARK-26604 branch from 7c1e13d to 8cd705e Compare January 15, 2019 09:50

cloud-fan reviewed Jan 15, 2019

View reviewed changes

cloud-fan approved these changes Jan 15, 2019

View reviewed changes

Register channel when registering stream.

7eed779

viirya force-pushed the SPARK-26604 branch from 8cd705e to 7eed779 Compare January 15, 2019 11:14

cloud-fan mentioned this pull request Jan 15, 2019

[SPARK-26418][SHUFFLE] Only OpenBlocks without any ChunkFetch for one stream will cause memory leak in ExternalShuffleService #23355

Closed

cloud-fan closed this in cf133e6 Jan 16, 2019

GulajavaMinistudio mentioned this pull request Jan 16, 2019

destroy injector when module is destroyed GulajavaMinistudio/spark#498

Merged

abellina mentioned this pull request Mar 7, 2019

[SPARK-26604][CORE][BACKPORT-2.4] Clean up channel registration for StreamManager #24013

Closed

This was referenced Dec 23, 2019

[SPARK-26604][CORE][BACKPORT-1.6] Clean up channel registration for S… #26988

Closed

[SPARK-26604][CORE][BACKPORT-1.6] Clean up channel registration for S… #26989

Closed

viirya deleted the SPARK-26604 branch December 27, 2023 18:36

[SPARK-26604][CORE] Clean up channel registration for StreamManager #23521

[SPARK-26604][CORE] Clean up channel registration for StreamManager #23521

Conversation

viirya commented Jan 11, 2019 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

viirya commented Jan 11, 2019

SparkQA commented Jan 11, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 11, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya commented Jan 14, 2019

SparkQA commented Jan 14, 2019

SparkQA commented Jan 14, 2019

shaneknapp commented Jan 14, 2019

shaneknapp commented Jan 14, 2019

SparkQA commented Jan 14, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 15, 2019

viirya commented Jan 15, 2019

SparkQA commented Jan 15, 2019

SparkQA commented Jan 15, 2019

viirya commented Jan 15, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 15, 2019

SparkQA commented Jan 15, 2019

viirya commented Jan 15, 2019

SparkQA commented Jan 15, 2019

cloud-fan commented Jan 16, 2019

felixcheung commented Mar 7, 2019

viirya commented Mar 7, 2019

abellina commented Mar 7, 2019

abellina commented Mar 7, 2019

viirya commented Jan 11, 2019 •

edited

Loading