[Java] Don't wait for receiver to close channel endpoint in the drive…

…r conductor as this is no longer required and can cause he driver to lockup. Issue #338.
aeron-io · Apr 12, 2017 · 2fef3e4 · 2fef3e4
1 parent ee409cc
commit 2fef3e4
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 17 deletions.
diff --git a/aeron-driver/src/main/java/io/aeron/driver/DriverConductor.java b/aeron-driver/src/main/java/io/aeron/driver/DriverConductor.java
@@ -132,11 +132,11 @@ public DriverConductor(final Context ctx)
 
     public void onClose()
     {
+        sendChannelEndpointByChannelMap.values().forEach(SendChannelEndpoint::close);
+        receiveChannelEndpointByChannelMap.values().forEach(ReceiveChannelEndpoint::close);
         networkPublications.forEach(NetworkPublication::close);
         publicationImages.forEach(PublicationImage::close);
         ipcPublications.forEach(IpcPublication::close);
-        sendChannelEndpointByChannelMap.values().forEach(SendChannelEndpoint::close);
-        receiveChannelEndpointByChannelMap.values().forEach(ReceiveChannelEndpoint::close);
     }
 
     public String roleName()
@@ -702,11 +702,6 @@ void onRemoveSubscription(final long registrationId, final long correlationId)
                 channelEndpoint.closeStatusIndicator();
                 receiveChannelEndpointByChannelMap.remove(channelEndpoint.udpChannel().canonicalForm());
                 receiverProxy.closeReceiveChannelEndpoint(channelEndpoint);
-
-                while (!channelEndpoint.isClosed())
-                {
-                    Thread.yield();
-                }
             }
         }
 
@@ -1240,7 +1235,7 @@ private PublicationParams getPublicationParams(
             {
                 if (count < 3)
                 {
-                    throw new IllegalStateException("Params must be used as a set: " +
+                    throw new IllegalStateException("Params must be used as a complete set: " +
                         INITIAL_TERM_ID_PARAM_NAME + " " + TERM_ID_PARAM_NAME + " " + TERM_OFFSET_PARAM_NAME);
                 }
 
@@ -1252,7 +1247,7 @@ private PublicationParams getPublicationParams(
                 {
                     throw new IllegalStateException(
                         TERM_OFFSET_PARAM_NAME + "=" + params.termOffset + " > " +
-                            TERM_LENGTH_PARAM_NAME + "=" + params.termLength);
+                        TERM_LENGTH_PARAM_NAME + "=" + params.termLength);
                 }
 
                 params.isReplay = true;

diff --git a/aeron-driver/src/main/java/io/aeron/driver/Receiver.java b/aeron-driver/src/main/java/io/aeron/driver/Receiver.java
@@ -163,7 +163,7 @@ private void checkPendingSetupMessages(final long nowNs)
                 else if (pending.shouldElicitSetupMessage())
                 {
                     pending.channelEndpoint().sendSetupElicitingStatusMessage(
-                            pending.controlAddress(), pending.sessionId(), pending.streamId());
+                        pending.controlAddress(), pending.sessionId(), pending.streamId());
                     pending.timeOfStatusMessage(nowNs);
                 }
             }

diff --git a/aeron-driver/src/main/java/io/aeron/driver/media/ReceiveChannelEndpoint.java b/aeron-driver/src/main/java/io/aeron/driver/media/ReceiveChannelEndpoint.java
@@ -54,7 +54,7 @@ public class ReceiveChannelEndpoint extends UdpChannelTransport
     private final Int2IntCounterMap refCountByStreamIdMap = new Int2IntCounterMap(0);
 
     private final long receiverId;
-    private volatile boolean isClosed = false;
+    private boolean isClosed = false;
 
     public ReceiveChannelEndpoint(
         final UdpChannel udpChannel,
@@ -140,11 +140,6 @@ public void close()
         isClosed = true;
     }
 
-    public boolean isClosed()
-    {
-        return isClosed;
-    }
-
     public void openChannel()
     {
         openDatagramChannel(statusIndicator);
@@ -190,7 +185,7 @@ public boolean hasExplicitControl()
 
     public InetSocketAddress explicitControlAddress()
     {
-        return (udpChannel.hasExplicitControl()) ? udpChannel.localControl() : null;
+        return udpChannel.hasExplicitControl() ? udpChannel.localControl() : null;
     }
 
     public int onDataPacket(