implement issue 4971

Author: re-thc

vertx-core/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java

@@ -31,6 +31,8 @@
 import io.vertx.core.streams.impl.InboundBuffer;
 
 import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
 import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousFileChannel;
 import java.nio.channels.CompletionHandler;
@@ -43,14 +45,15 @@
 import java.nio.file.attribute.PosixFilePermissions;
 import java.util.HashSet;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author <a href="http://tfox.org">Tim Fox</a>
  */
 public class AsyncFileImpl implements AsyncFile {
 
+  private static final VarHandle SEND_FAILURE = MethodHandles.arrayElementVarHandle(boolean[].class);
+
   private static final Logger log = LoggerFactory.getLogger(AsyncFile.class);
 
   public static final int DEFAULT_READ_BUFFER_SIZE = 8192;
@@ -332,7 +335,7 @@ private void handleException(Throwable t) {
 
   private synchronized void doWrite(ByteBuffer[] buffers, long position, Handler<AsyncResult<Void>> handler) {
     AtomicInteger cnt = new AtomicInteger();
-    AtomicBoolean sentFailure = new AtomicBoolean();
+    boolean[] sentFailure = new boolean[1];
     for (ByteBuffer b: buffers) {
       int limit = b.limit();
       doWrite(b, position, limit, ar -> {
@@ -341,7 +344,7 @@ private synchronized void doWrite(ByteBuffer[] buffers, long position, Handler<A
             handler.handle(ar);
           }
         } else {
-          if (sentFailure.compareAndSet(false, true)) {
+          if (SEND_FAILURE.compareAndSet(sentFailure, 0, false, true)) {
             handler.handle(ar);
           }
         }

vertx-core/src/main/java/io/vertx/core/impl/VertxImpl.java

@@ -73,14 +73,15 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
 import java.lang.ref.Cleaner;
 import java.lang.ref.WeakReference;
 import java.lang.reflect.Method;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
 import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
@@ -107,6 +108,16 @@ public class VertxImpl implements VertxInternal, MetricsProvider {
   // https://github.com/eclipse-vertx/vert.x/issues/4611
 
   private static final Logger log = LoggerFactory.getLogger(VertxImpl.class);
+  private static final VarHandle INTERNAL_TIMER_HANDLER_DISPOSED;
+
+  static {
+    try {
+      INTERNAL_TIMER_HANDLER_DISPOSED = MethodHandles.lookup()
+        .findVarHandle(InternalTimerHandler.class, "disposed", boolean.class);
+    } catch (NoSuchFieldException | IllegalAccessException e) {
+      throw new IllegalArgumentException(e);
+    }
+  }
 
   static final Object[] EMPTY_CONTEXT_LOCALS = new Object[0];
   private static final String CLUSTER_MAP_NAME = "__vertx.haInfo";
@@ -1063,7 +1074,7 @@ class InternalTimerHandler implements Handler<Void>, Closeable, Runnable {
     private final boolean periodic;
     private final long id;
     private final ContextInternal context;
-    private final AtomicBoolean disposed = new AtomicBoolean();
+    private volatile boolean disposed;
     private volatile java.util.concurrent.Future<?> future;
 
     InternalTimerHandler(long id, Handler<Long> runnable, boolean periodic, ContextInternal context) {
@@ -1084,10 +1095,10 @@ public void run() {
 
     public void handle(Void v) {
       if (periodic) {
-        if (!disposed.get()) {
+        if (!disposed) {
           handler.handle(id);
         }
-      } else if (disposed.compareAndSet(false, true)) {
+      } else if (compareAndSetDisposed()) {
         timeouts.remove(id);
         try {
           handler.handle(id);
@@ -1109,7 +1120,7 @@ private boolean cancel() {
     }
 
     private boolean tryCancel() {
-      if  (disposed.compareAndSet(false, true)) {
+      if  (compareAndSetDisposed()) {
         timeouts.remove(id);
         future.cancel(false);
         return true;
@@ -1123,6 +1134,10 @@ public void close(Completable<Void> completion) {
       tryCancel();
       completion.succeed();
     }
+
+    private boolean compareAndSetDisposed() {
+      return INTERNAL_TIMER_HANDLER_DISPOSED.compareAndSet(this, false, true);
+    }
   }
 
   @Override

vertx-core/src/main/java/io/vertx/core/net/endpoint/impl/EndpointResolverImpl.java

@@ -128,7 +128,7 @@ private void close() {
   private class ManagedEndpoint extends ManagedResource {
 
     private final Future<EndpointImpl> endpoint;
-    private final AtomicBoolean disposed = new AtomicBoolean();
+    private volatile boolean disposed;
 
     public ManagedEndpoint(Future<EndpointImpl> endpoint) {
       super();
@@ -145,12 +145,20 @@ protected void cleanup() {
     @Override
     protected void checkExpired() {
       if (endpoint.succeeded() && keepAliveMillis > 0 && System.currentTimeMillis() - endpoint.result().lastAccessed.get() >= keepAliveMillis) {
-        if (disposed.compareAndSet(false, true)) {
+        if (markDisposed()) {
           decRefCount();
         }
       }
     }
 
+    private synchronized boolean markDisposed() {
+      if (disposed) {
+        return false;
+      }
+      disposed = true;
+      return true;
+    }
+
     @Override
     public boolean incRefCount() {
       return super.incRefCount();
@@ -215,7 +223,7 @@ private ManagedEndpoint resolveAddress(A address) {
     }, fn);
     if (sFuture.created) {
       sFuture.fut.onFailure(err -> {
-        if (sFuture.endpoint.disposed.compareAndSet(false, true)) {
+        if (sFuture.endpoint.markDisposed()) {
           // We need to call decRefCount outside the withEndpoint method, hence we need
           // the Result class workaround
           sFuture.endpoint.decRefCount();

vertx-core/src/main/java/io/vertx/core/shareddata/impl/LocalAsyncLocks.java

@@ -16,17 +16,28 @@
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.shareddata.Lock;
 
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * @author Thomas Segismont
  */
 public class LocalAsyncLocks {
 
+  private static final VarHandle ASYNC_LOCK_INVOKED;
+
+  static {
+    try {
+      ASYNC_LOCK_INVOKED = MethodHandles.lookup().findVarHandle(AsyncLock.class, "invoked", boolean.class);
+    } catch (NoSuchFieldException | IllegalAccessException e) {
+      throw new IllegalArgumentException(e);
+    }
+  }
+
   private class LockWaiter {
 
     final ContextInternal context;
@@ -77,15 +88,15 @@ void acquireLock() {
   private class AsyncLock implements LockInternal {
 
     final String lockName;
-    final AtomicBoolean invoked = new AtomicBoolean();
+    private volatile boolean invoked;
 
     AsyncLock(String lockName) {
       this.lockName = lockName;
     }
 
     @Override
     public void release() {
-      if (invoked.compareAndSet(false, true)) {
+      if (ASYNC_LOCK_INVOKED.compareAndSet(this, false, true)) {
         nextWaiter(lockName);
       }
     }

vertx-core/src/main/java/io/vertx/core/shareddata/impl/LocalAsyncMapImpl.java

@@ -24,7 +24,6 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.util.concurrent.TimeUnit.*;
 import static java.util.stream.Collectors.*;
@@ -103,18 +102,18 @@ public Future<V> putIfAbsent(K k, V v, long ttl) {
   @Override
   public Future<Boolean> removeIfPresent(K k, V v) {
     ContextInternal ctx = vertx.getOrCreateContext();
-    AtomicBoolean result = new AtomicBoolean();
+    boolean[] result = new boolean[1];
     map.computeIfPresent(k, (key, holder) -> {
       if (holder.value.equals(v)) {
-        result.compareAndSet(false, true);
+        result[0] = true;
         if (holder.expires()) {
           vertx.cancelTimer(holder.timerId);
         }
         return null;
       }
       return holder;
     });
-    return ctx.succeededFuture(result.get());
+    return ctx.succeededFuture(result[0]);
   }
 
   @Override