[Java] Check all contexts for reuse by multiple conclusion and improve on solution provided in PR #666, increase minor version as a result of change.

Author: mjpt777

aeron-archive/src/main/java/io/aeron/archive/Archive.java

@@ -20,6 +20,7 @@
 import io.aeron.Image;
 import io.aeron.archive.client.AeronArchive;
 import io.aeron.archive.client.ArchiveException;
+import io.aeron.exceptions.ConfigurationException;
 import io.aeron.logbuffer.LogBufferDescriptor;
 import org.agrona.BitUtil;
 import org.agrona.CloseHelper;
@@ -35,12 +36,14 @@
 import java.util.Objects;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.Supplier;
 
 import static io.aeron.driver.status.SystemCounterDescriptor.SYSTEM_COUNTER_TYPE_ID;
 import static io.aeron.logbuffer.LogBufferDescriptor.TERM_MAX_LENGTH;
 import static io.aeron.logbuffer.LogBufferDescriptor.TERM_MIN_LENGTH;
 import static java.lang.System.getProperty;
+import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
 import static org.agrona.SystemUtil.getDurationInNanos;
 import static org.agrona.SystemUtil.getSizeAsInt;
 import static org.agrona.SystemUtil.loadPropertiesFiles;
@@ -351,6 +354,13 @@ public static boolean deleteArchiveOnStart()
      */
     public static class Context implements Cloneable
     {
+        /**
+         * Using an integer because there is no support for boolean. 1 is concluded, 0 is not concluded.
+         */
+        private static final AtomicIntegerFieldUpdater<Context> IS_CONCLUDED_UPDATER = newUpdater(
+            Context.class, "isConcluded");
+        private volatile int isConcluded;
+
         private boolean deleteArchiveOnStart = Configuration.deleteArchiveOnStart();
         private boolean ownsAeronClient = false;
         private String aeronDirectoryName = CommonContext.getAeronDirectoryName();
@@ -414,6 +424,11 @@ public Context clone()
          */
         public void conclude()
         {
+            if (0 != IS_CONCLUDED_UPDATER.getAndSet(this, 1))
+            {
+                throw new ConfigurationException("Context already concluded");
+            }
+
             Objects.requireNonNull(errorHandler, "Error handler must be supplied");
 
             if (null == epochClock)

aeron-archive/src/main/java/io/aeron/archive/client/AeronArchive.java

@@ -19,6 +19,7 @@
 import io.aeron.archive.codecs.ControlResponseCode;
 import io.aeron.archive.codecs.ControlResponseDecoder;
 import io.aeron.archive.codecs.SourceLocation;
+import io.aeron.exceptions.ConfigurationException;
 import io.aeron.exceptions.TimeoutException;
 import org.agrona.CloseHelper;
 import org.agrona.ErrorHandler;
@@ -27,11 +28,13 @@
 import org.agrona.concurrent.*;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 import static io.aeron.archive.client.ArchiveProxy.DEFAULT_RETRY_ATTEMPTS;
 import static io.aeron.driver.Configuration.*;
+import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
 import static org.agrona.SystemUtil.getDurationInNanos;
 import static org.agrona.SystemUtil.getSizeAsInt;
 
@@ -1532,6 +1535,13 @@ public static int recordingEventsStreamId()
      */
     public static class Context implements Cloneable
     {
+        /**
+         * Using an integer because there is no support for boolean. 1 is concluded, 0 is not concluded.
+         */
+        private static final AtomicIntegerFieldUpdater<Context> IS_CONCLUDED_UPDATER = newUpdater(
+            Context.class, "isConcluded");
+        private volatile int isConcluded;
+
         private long messageTimeoutNs = Configuration.messageTimeoutNs();
         private String recordingEventsChannel = AeronArchive.Configuration.recordingEventsChannel();
         private int recordingEventsStreamId = AeronArchive.Configuration.recordingEventsStreamId();
@@ -1571,6 +1581,11 @@ public Context clone()
          */
         public void conclude()
         {
+            if (0 != IS_CONCLUDED_UPDATER.getAndSet(this, 1))
+            {
+                throw new ConfigurationException("Context already concluded");
+            }
+
             if (null == aeron)
             {
                 aeron = Aeron.connect(new Aeron.Context().aeronDirectoryName(aeronDirectoryName));

aeron-client/src/main/java/io/aeron/Aeron.java

@@ -123,8 +123,6 @@ public static Aeron connect()
      */
     public static Aeron connect(final Context ctx)
     {
-        ctx.verifyNotAlreadyInUse();
-
         try
         {
             final Aeron aeron = new Aeron(ctx);

aeron-client/src/main/java/io/aeron/CommonContext.java

@@ -16,6 +16,7 @@
 package io.aeron;
 
 import io.aeron.exceptions.AeronException;
+import io.aeron.exceptions.ConfigurationException;
 import io.aeron.exceptions.DriverTimeoutException;
 import org.agrona.DirectBuffer;
 import org.agrona.IoUtil;
@@ -33,13 +34,14 @@
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.Consumer;
 
 import static io.aeron.Aeron.sleep;
 import static io.aeron.CncFileDescriptor.*;
 import static java.lang.Long.getLong;
 import static java.lang.System.getProperty;
+import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
 
 /**
  * This class provides the Media Driver and client with common configuration for the Aeron directory.
@@ -215,7 +217,12 @@ public class CommonContext implements Cloneable
      */
     public static final String TETHER_PARAM_NAME = "tether";
 
-    private final AtomicBoolean inUse = new AtomicBoolean(false);
+    /**
+     * Using an integer because there is no support for boolean. 1 is concluded, 0 is not concluded.
+     */
+    private static final AtomicIntegerFieldUpdater<CommonContext> IS_CONCLUDED_UPDATER = newUpdater(
+        CommonContext.class, "isConcluded");
+    private volatile int isConcluded;
 
     private long driverTimeoutMs = DRIVER_TIMEOUT_MS;
     private String aeronDirectoryName = getAeronDirectoryName();
@@ -291,6 +298,11 @@ public static String generateRandomDirName()
      */
     public CommonContext conclude()
     {
+        if (0 != IS_CONCLUDED_UPDATER.getAndSet(this, 1))
+        {
+            throw new ConfigurationException("Context already concluded");
+        }
+
         concludeAeronDirectory();
 
         cncFile = new File(aeronDirectory, CncFileDescriptor.CNC_FILE);
@@ -313,21 +325,6 @@ public CommonContext concludeAeronDirectory()
         return this;
     }
 
-    /**
-     * Checks to see if another instance of the Aeron Client or Media Driver already is already using this
-     * context.
-     *
-     * @throws IllegalStateException if the context is already in use.
-     */
-    public void verifyNotAlreadyInUse()
-    {
-        if (!inUse.compareAndSet(false, true))
-        {
-            throw new IllegalStateException(
-                "Context instances may not be reused, create a new one for each Aeron/MediaDriver instance");
-        }
-    }
-
     /**
      * Get the top level Aeron directory used for communication between the client and Media Driver, and
      * the location of the data buffers.

aeron-client/src/test/java/io/aeron/CommonContextTest.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2014-2019 Real Logic Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.aeron;
+
+import io.aeron.exceptions.ConfigurationException;
+import org.junit.Test;
+
+public class CommonContextTest
+{
+    @Test(expected = ConfigurationException.class)
+    public void shouldNotAllowConcludeMoreThanOnce()
+    {
+        final CommonContext ctx = new CommonContext();
+
+        ctx.conclude();
+        ctx.conclude();
+    }
+}

aeron-cluster/src/main/java/io/aeron/cluster/ConsensusModule.java

@@ -23,6 +23,7 @@
 import io.aeron.cluster.client.ClusterException;
 import io.aeron.cluster.codecs.mark.ClusterComponentType;
 import io.aeron.cluster.service.*;
+import io.aeron.exceptions.ConfigurationException;
 import io.aeron.security.Authenticator;
 import io.aeron.security.AuthenticatorSupplier;
 import org.agrona.*;
@@ -35,12 +36,14 @@
 import java.util.Random;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.Supplier;
 
 import static io.aeron.cluster.ConsensusModule.Configuration.*;
 import static io.aeron.cluster.service.ClusteredServiceContainer.Configuration.SNAPSHOT_CHANNEL_PROP_NAME;
 import static io.aeron.cluster.service.ClusteredServiceContainer.Configuration.SNAPSHOT_STREAM_ID_PROP_NAME;
 import static io.aeron.driver.status.SystemCounterDescriptor.SYSTEM_COUNTER_TYPE_ID;
+import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
 import static org.agrona.SystemUtil.*;
 import static org.agrona.concurrent.status.CountersReader.METADATA_LENGTH;
 
@@ -885,6 +888,13 @@ public static int memberStatusStreamId()
      */
     public static class Context implements Cloneable
     {
+        /**
+         * Using an integer because there is no support for boolean. 1 is concluded, 0 is not concluded.
+         */
+        private static final AtomicIntegerFieldUpdater<Context> IS_CONCLUDED_UPDATER = newUpdater(
+            Context.class, "isConcluded");
+        private volatile int isConcluded;
+
         private boolean ownsAeronClient = false;
         private String aeronDirectoryName = CommonContext.getAeronDirectoryName();
         private Aeron aeron;
@@ -975,6 +985,11 @@ public Context clone()
         @SuppressWarnings("MethodLength")
         public void conclude()
         {
+            if (0 != IS_CONCLUDED_UPDATER.getAndSet(this, 1))
+            {
+                throw new ConfigurationException("Context already concluded");
+            }
+
             if (null == clusterDir)
             {
                 clusterDir = new File(clusterDirectoryName);

aeron-cluster/src/main/java/io/aeron/cluster/client/AeronCluster.java

@@ -30,7 +30,9 @@
 import org.agrona.concurrent.*;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
+import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
 import static org.agrona.SystemUtil.getDurationInNanos;
 
 /**
@@ -886,6 +888,13 @@ public static int egressStreamId()
      */
     public static class Context implements Cloneable
     {
+        /**
+         * Using an integer because there is no support for boolean. 1 is concluded, 0 is not concluded.
+         */
+        private static final AtomicIntegerFieldUpdater<Context> IS_CONCLUDED_UPDATER = newUpdater(
+            Context.class, "isConcluded");
+        private volatile int isConcluded;
+
         private long messageTimeoutNs = Configuration.messageTimeoutNs();
         private String clusterMemberEndpoints = Configuration.clusterMemberEndpoints();
         private String ingressChannel = Configuration.ingressChannel();
@@ -922,6 +931,11 @@ public Context clone()
 
         public void conclude()
         {
+            if (0 != IS_CONCLUDED_UPDATER.getAndSet(this, 1))
+            {
+                throw new ConfigurationException("Context already concluded");
+            }
+
             if (null == aeron)
             {
                 aeron = Aeron.connect(

aeron-cluster/src/main/java/io/aeron/cluster/service/ClusteredServiceAgent.java

@@ -696,7 +696,7 @@ private AtomicCounter awaitHeartbeatCounter(final CountersReader counters)
 
     private void loadSnapshot(final long recordingId)
     {
-        try (AeronArchive archive = AeronArchive.connect(archiveCtx))
+        try (AeronArchive archive = AeronArchive.connect(archiveCtx.clone()))
         {
             final String channel = ctx.replayChannel();
             final int streamId = ctx.replayStreamId();
@@ -741,7 +741,7 @@ private long onTakeSnapshot(final long logPosition, final long leadershipTermId)
     {
         final long recordingId;
 
-        try (AeronArchive archive = AeronArchive.connect(archiveCtx);
+        try (AeronArchive archive = AeronArchive.connect(archiveCtx.clone());
             Publication publication = aeron.addExclusivePublication(ctx.snapshotChannel(), ctx.snapshotStreamId()))
         {
             final String channel = ChannelUri.addSessionId(ctx.snapshotChannel(), publication.sessionId());

aeron-cluster/src/main/java/io/aeron/cluster/service/ClusteredServiceContainer.java

@@ -34,9 +34,11 @@
 
 import java.io.File;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.Supplier;
 
 import static io.aeron.driver.status.SystemCounterDescriptor.SYSTEM_COUNTER_TYPE_ID;
+import static java.util.concurrent.atomic.AtomicIntegerFieldUpdater.newUpdater;
 import static org.agrona.SystemUtil.getSizeAsInt;
 import static org.agrona.SystemUtil.loadPropertiesFiles;
 
@@ -430,6 +432,13 @@ public static boolean isRespondingService()
      */
     public static class Context implements Cloneable
     {
+        /**
+         * Using an integer because there is no support for boolean. 1 is concluded, 0 is not concluded.
+         */
+        private static final AtomicIntegerFieldUpdater<Context> IS_CONCLUDED_UPDATER = newUpdater(
+            Context.class, "isConcluded");
+        private volatile int isConcluded;
+
         private int serviceId = Configuration.serviceId();
         private String serviceName = Configuration.serviceName();
         private String replayChannel = Configuration.replayChannel();
@@ -481,6 +490,11 @@ public Context clone()
         @SuppressWarnings("MethodLength")
         public void conclude()
         {
+            if (0 != IS_CONCLUDED_UPDATER.getAndSet(this, 1))
+            {
+                throw new ConfigurationException("Context already concluded");
+            }
+
             if (serviceId < 0)
             {
                 throw new ConfigurationException("service id cannot be negative: " + serviceId);

aeron-driver/src/main/java/io/aeron/driver/MediaDriver.java

@@ -222,7 +222,6 @@ public static MediaDriver launch()
      */
     public static MediaDriver launch(final Context ctx)
     {
-        ctx.verifyNotAlreadyInUse();
         return new MediaDriver(ctx).start();
     }