Skip to content

Commit

Permalink
Fix potential race during TcpTransport close
Browse files Browse the repository at this point in the history
Fixed two potential causes for leaked threads during tests:
1. When adding a channel to serverChannels, we add it under a monitor
that we do not use when reading from it. This is potentially unsafe if
there is no other happens-before relationship ensuring the safety of
this.
2. Long-shot but if the thread pool was shutdown before entering this
code, we would silently forget about closing server channels so added
assert.

Relates to CI failure issue: elastic#37543
  • Loading branch information
henningandersen committed Feb 18, 2019
1 parent 869de3f commit f61ee79
Showing 1 changed file with 9 additions and 2 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -553,13 +553,21 @@ protected final void doClose() {
protected final void doStop() {
final CountDownLatch latch = new CountDownLatch(1);
// make sure we run it on another thread than a possible IO handler thread
assert threadPool.generic().isShutdown() == false : "Must stop transport before terminating underlying threadpool";
threadPool.generic().execute(() -> {
closeLock.writeLock().lock();
try {
keepAlive.close();

// Copy map of server channels to avoid holding serverChannels monitor while closing channels.
Map<String, List<TcpServerChannel>> serverChannelsCopy;
synchronized (this.serverChannels) {
serverChannelsCopy = new HashMap<>(this.serverChannels);
this.serverChannels.clear();
}

// first stop to accept any incoming connections so nobody can connect to this transport
for (Map.Entry<String, List<TcpServerChannel>> entry : serverChannels.entrySet()) {
for (Map.Entry<String, List<TcpServerChannel>> entry : serverChannelsCopy.entrySet()) {
String profile = entry.getKey();
List<TcpServerChannel> channels = entry.getValue();
ActionListener<Void> closeFailLogger = ActionListener.wrap(c -> {
Expand All @@ -568,7 +576,6 @@ protected final void doStop() {
channels.forEach(c -> c.addCloseListener(closeFailLogger));
CloseableChannel.closeChannels(channels, true);
}
serverChannels.clear();

// close all of the incoming channels. The closeChannels method takes a list so we must convert the set.
CloseableChannel.closeChannels(new ArrayList<>(acceptedChannels), true);
Expand Down

0 comments on commit f61ee79

Please sign in to comment.