New implementation of LazyValue (#2738)

* New implementation of LazyValue that lazily initializes a Semaphore instead of eagerly creating a ReentrantLock. Makes use of volatile guarantees and atomicity of VarHandle updates.

Signed-off-by: Santiago Pericasgeertsen <santiago.pericasgeertsen@oracle.com>

* New test for LazyValueImpl.

Signed-off-by: Santiago Pericasgeertsen <santiago.pericasgeertsen@oracle.com>

* Reduced sleep time in test.

Signed-off-by: Santiago Pericasgeertsen <santiago.pericasgeertsen@oracle.com>

common/common/src/main/java/io/helidon/common/LazyValueImpl.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019, 2020 Oracle and/or its affiliates.
+ * Copyright (c) 2019, 2021 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,21 +16,57 @@
 
 package io.helidon.common;
 
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.util.concurrent.Semaphore;
 import java.util.function.Supplier;
 
 class LazyValueImpl<T> implements LazyValue<T> {
-    private final Lock theLock = new ReentrantLock();
 
+    /**
+     * VarHandle's for atomic access to {@code theLock} and {@code loaded}
+     * instance variables.
+     */
+    private static final VarHandle THE_LOCK;
+    private static final VarHandle LOADED;
+    static {
+        try {
+            THE_LOCK = MethodHandles.lookup().findVarHandle(LazyValueImpl.class, "theLock", Semaphore.class);
+            LOADED = MethodHandles.lookup().findVarHandle(LazyValueImpl.class, "loaded", int.class);
+        } catch (Exception e) {
+            throw new Error("Unable to obtain VarHandle's", e);
+        }
+    }
+
+    /**
+     * Cached value returned by supplier or passed directly to constructor.
+     */
     private T value;
 
+    /**
+     * Wrapped delegate or {@code null} if using direct value instead.
+     */
     private Supplier<T> delegate;
-    private volatile boolean loaded;
+
+    /**
+     * Semaphore to prevent concurrent update of internal state. Updated
+     * only via {@code THE_LOCK}.
+     */
+    private volatile Semaphore theLock;
+
+    /**
+     * Boolean indicating value if readily available without calling
+     * a supplier.
+     */
+    private volatile int loaded;
+
+    private static final int DONE = -1;
+    private static final int INIT = 0;
+    private static final int WORKING = INIT + 1;
 
     LazyValueImpl(T value) {
         this.value = value;
-        this.loaded = true;
+        this.loaded = DONE;
     }
 
     LazyValueImpl(Supplier<T> supplier) {
@@ -39,27 +75,62 @@ class LazyValueImpl<T> implements LazyValue<T> {
 
     @Override
     public boolean isLoaded() {
-        return loaded;
+        return loaded == DONE;
     }
 
+    /**
+     * Ensure only a single thread calls the delegate if the value is not yet loaded.
+     * Note that {@code loadedCopy} and {@code theLockCopy} represent thread copies
+     * of the corresponding volatile variables, while {@code LOADED} and {@code THE_LOCK}
+     * are var references to those volatile variables.
+     *
+     * @return the value
+     */
     @Override
     public T get() {
-        if (loaded) {
+        int loadedCopy = loaded;
+        if (loadedCopy == DONE) {
             return value;
         }
 
-        // not loaded (probably)
-        theLock.lock();
+        Semaphore theLockCopy = theLock;
+
+        // Race winner that sets 'loaded' to WORKING skips this loop, losers enter it
+        while (loadedCopy != DONE && !LOADED.compareAndSet(this, INIT, WORKING)) {
+            // One of the losers initializes 'theLock'
+            if (theLockCopy == null) {
+                THE_LOCK.compareAndSet(this, null, new Semaphore(0));
+                theLockCopy = theLock;
+            }
+
+            loadedCopy = loaded;
+            if (loadedCopy == WORKING) {
+                theLockCopy.acquireUninterruptibly();
+                loadedCopy = loaded;
+            }
+        }
 
         try {
-            if (loaded) {
+            if (loadedCopy == DONE) {
                 return value;
             }
+            loadedCopy = INIT;
             value = delegate.get();
-            loaded = true;
             delegate = null;
+            loadedCopy = DONE;
+            loaded = DONE;
         } finally {
-            theLock.unlock();
+            // If condition holds, delegate threw exception
+            if (loadedCopy == INIT) {
+                loaded = INIT;
+            }
+            // Assert: if theLock is null, the successful compare-and-set of THE_LOCK is
+            // in the future; but after such compare-and-set there will be a check of
+            // loaded as not WORKING, resulting in no attempt to acquire the semaphore
+            theLockCopy = theLock;
+            if (theLockCopy != null) {
+                theLockCopy.release();
+            }
         }
 
         return value;

common/common/src/test/java/io/helidon/common/LazyValueTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2019, 2021 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 
 package io.helidon.common;
 
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -103,4 +104,36 @@ void testSupplierParallel() {
         assertThat(called.get(), is(1));
         assertThat(threadsStarted.get(), is(threadCount));
     }
+
+    @Test
+    void testSemaphoreRelease() throws Exception {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        LazyValue<String> value = LazyValue.create(() -> {
+            try {
+                future.complete(null);
+                Thread.sleep(500);
+            } catch (InterruptedException e) {
+                // falls through
+            }
+            return "DONE";
+        });
+
+        Thread[] threads = new Thread[3];
+        threads[0] = new Thread(value::get);
+        threads[0].start();
+        future.get();           // wait for supplier to be called
+
+        threads[1] = new Thread(value::get);
+        threads[2] = new Thread(value::get);
+        threads[1].start();     // blocked, then released by thread[0]
+        threads[2].start();     // blocked, then released by thread[1]
+
+        for (Thread t : threads) {
+            try {
+                t.join();
+            } catch (InterruptedException e) {
+                // falls through
+            }
+        }
+    }
 }