Concurrent recursive function memoization

vavr/src/main/java/io/vavr/concurrent/ConcurrentTrampoliningMemoizer.java

@@ -0,0 +1,42 @@
+package io.vavr.concurrent;
+
+import io.vavr.Function1;
+
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+
+/**
+ * A utility class that creates a memoizing version of an arbitrary concurrent function. Recursive calls are trampolined
+ * so as to avoid stack overflows.
+ * @see "http://sebastian-millies.blogspot.de/2016/05/concurrent-recursive-function.html"
+ */
+class ConcurrentTrampoliningMemoizer<T, R> {
+    private final ConcurrentMap<T, Promise<R>> memo;
+
+    ConcurrentTrampoliningMemoizer(ConcurrentMap<T, Promise<R>> cache) {
+        this.memo = cache;
+    }
+
+    Function1<T, Future<R>> memoize(Function1<T, Future<R>> f) {
+        return t -> {
+            Promise<R> r = memo.get(t);
+            if (r == null) {
+                // value not yet memoized: put a container in the map that will come to hold the value
+                final Promise<R> compute = Promise.make();
+                r = memo.putIfAbsent(t, compute);
+                if (r == null) {
+                    // only the thread that first has a cache miss calls the underlying function.
+                    // recursive asynchronous calls are bounced off the task queue inside the default executor, avoiding stack overflows.
+                    // the computed value is placed in the container.
+                    Future<R> futureValue = Future.ofSupplier(() -> f.apply(t)).flatMap(Function1.identity());  // unwrap nested Future
+                    r = compute.completeWith(futureValue);
+                }
+            }
+            return r.future();
+        };
+    }
+}

vavr/src/main/java/io/vavr/concurrent/MemoizedConcurrently.java

@@ -0,0 +1,47 @@
+package io.vavr.concurrent;
+
+import io.vavr.Function1;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Marker interface for concurrently memoized functions.
+ */
+public interface MemoizedConcurrently {
+
+    /**
+     * Delegates to {@link #of(Function1, ConcurrentMap)}} using a <tt>ConcurrentHashMap</tt> as cache implementation.
+     */
+    static <T, R> Function1<T, Future<R>> of(Function1<T, Future<R>> function) {
+        if (function instanceof MemoizedConcurrently) {
+            return function; // make this method idempotent
+        }
+        return MemoizedConcurrently.of(function, new ConcurrentHashMap<>());
+    }
+
+    /**
+     * Lift the given function to a thread-safe, concurrently memoizing version. The returned function computes the
+     * return value for a given argument only once. On subsequent calls given the same argument the memoized value
+     * is returned. The returned function has a few interesting properties:
+     * <ul>
+     * <li>The function does not permit {@code null} values.
+     * <li>Different threads will always wind up using the same value instances, so the function may compute values
+     * that are supposed to be singletons.
+     * <li>Concurrent callers won't block each other.
+     * </ul>
+     * This method is idempotent, i. e. applying it to an already concurrently memoized function will return the
+     * function unchanged.
+     * @param function a possibly recursive (asynchronous) function
+     * @param cache a structure that holds the memoized values
+     * @return the memoizing equivalent
+     */
+    static <T, R> Function1<T, Future<R>> of(Function1<T, Future<R>> function, ConcurrentMap<T, Promise<R>> cache) {
+        if (function instanceof MemoizedConcurrently) {
+            return function; // make this method idempotent
+        }
+        ConcurrentTrampoliningMemoizer<T, R> memoizer = new ConcurrentTrampoliningMemoizer<>(cache);
+        Function1<T, Future<R>> memoized = memoizer.memoize(function);
+        return (Function1<T, Future<R>> & MemoizedConcurrently) memoized::apply; // mark as memoized using intersection type
+    }
+}