Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
hensg wants to merge 6 commits into from
+58 −5
Closed

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

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
22def77
OneForOneStreamManager might leak memory in connectionTerminated
hensg Jan 2, 2020
a9ebe4f
Add try-catch block to release buffers for all removedStates even the…
hensg Jan 2, 2020
410651d
Rethrow exception when fails to release all StreamState buffers
hensg Jan 9, 2020
ac58798
Fix linter linelength error
hensg Jan 9, 2020
b02e48a
Remove extra blank line
hensg Jan 12, 2020
0fde4c9
Store first exception and re-throw it
hensg Jan 14, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions .../network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.network.server;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
Expand Down Expand Up @@ -117,19 +118,26 @@ public static Pair<Long, Integer> parseStreamChunkId(String streamChunkId) {

@Override
public void connectionTerminated(Channel channel) {
LinkedList<StreamState> removedStates = new LinkedList<>();
// Close all streams which have been associated with the channel.
for (Map.Entry<Long, StreamState> entry: streams.entrySet()) {
StreamState state = entry.getValue();
if (state.associatedChannel == channel) {
streams.remove(entry.getKey());
removedStates.add(streams.remove(entry.getKey()));
}
}

// Release all remaining buffers.
for (StreamState state: removedStates) {
// Release all remaining buffers.
Comment thread
hensg marked this conversation as resolved.
Outdated
try {
while (state.buffers.hasNext()) {
ManagedBuffer buffer = state.buffers.next();

@viirya viirya Jan 2, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@HeartSaVioR HeartSaVioR Jan 2, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@viirya viirya Jan 2, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@HeartSaVioR HeartSaVioR Jan 2, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

  1. 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).

  2. 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.

@viirya viirya Jan 9, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@HeartSaVioR HeartSaVioR Jan 9, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

if (buffer != null) {
buffer.release();
}
}
} catch (RuntimeException e) {
logger.error("Exception trying to release remaining StreamState buffers", e);
}
}
}
Expand Down
35 changes: 35 additions & 0 deletions ...ork-common/src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.network.server;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import io.netty.channel.Channel;
Expand Down Expand Up @@ -96,4 +97,38 @@ public void managedBuffersAreFreedWhenConnectionIsClosed() {
Mockito.verify(buffer2, Mockito.times(1)).release();
Assert.assertEquals(0, manager.numStreamStates());
}

@Test
public void streamStatesAreFreedWhenConnectionIsClosedEvenIfBufferIteratorThrowsException() {
OneForOneStreamManager manager = new OneForOneStreamManager();

Iterator<ManagedBuffer> buffers = Mockito.mock(Iterator.class);
Mockito.when(buffers.hasNext()).thenReturn(true);
Mockito.when(buffers.next()).thenThrow(RuntimeException.class);

ManagedBuffer mockManagedBuffer = Mockito.mock(ManagedBuffer.class);

Iterator<ManagedBuffer> buffers2 = Mockito.mock(Iterator.class);
Mockito.when(buffers2.hasNext()).thenReturn(true).thenReturn(true);
Mockito.when(buffers2.next()).thenReturn(mockManagedBuffer).thenThrow(RuntimeException.class);

Channel dummyChannel = Mockito.mock(Channel.class, Mockito.RETURNS_SMART_NULLS);
manager.registerStream("appId", buffers, dummyChannel);
manager.registerStream("appId", buffers2, dummyChannel);

Assert.assertEquals(2, manager.numStreamStates());

manager.connectionTerminated(dummyChannel);

Mockito.verify(buffers, Mockito.times(1)).hasNext();
Mockito.verify(buffers, Mockito.times(1)).next();

Mockito.verify(buffers2, Mockito.times(2)).hasNext();
Mockito.verify(buffers2, Mockito.times(2)).next();

Mockito.verify(mockManagedBuffer, Mockito.times(1)).release();

Assert.assertEquals(0, manager.numStreamStates());
}

}