Fix variable expiration with async cache (fixes #159)

An in-flight future was mistakenly given the maximum expiry allowed,
causing it to not honor an expire-after-create setting. Instead it was
supposed to be beyond the maximum to signal adaption on the completion
update.

The calculations for fixed expiration was made more robust to the time
rolling over. This now complies with System.nanoTime() warnings.

Strengthened the remove and replace operations to be more predictably
linearizable. This removed optimizations to avoid unnecessary work by
checking if the entry was present in a lock-free manner. Since the hash
table supresses loads until complete, that might mean that a call to
remove a loading entry was not performed. The contract allows either,
so the optimization is left to user code and gives preference to those
who need the linearizable behavior. (See #156)

Author: ben-manes

caffeine/src/main/java/com/github/benmanes/caffeine/cache/BoundedLocalCache.java

@@ -678,21 +678,20 @@ void expireAfterAccessEntries(long now) {
       return;
     }
 
-    long expirationTime = (now - expiresAfterAccessNanos());
-    expireAfterAccessEntries(accessOrderEdenDeque(), expirationTime, now);
+    expireAfterAccessEntries(accessOrderEdenDeque(), now);
     if (evicts()) {
-      expireAfterAccessEntries(accessOrderProbationDeque(), expirationTime, now);
-      expireAfterAccessEntries(accessOrderProtectedDeque(), expirationTime, now);
+      expireAfterAccessEntries(accessOrderProbationDeque(), now);
+      expireAfterAccessEntries(accessOrderProtectedDeque(), now);
     }
   }
 
   /** Expires entries in an access-order queue. */
   @GuardedBy("evictionLock")
-  void expireAfterAccessEntries(AccessOrderDeque<Node<K, V>> accessOrderDeque,
-      long expirationTime, long now) {
+  void expireAfterAccessEntries(AccessOrderDeque<Node<K, V>> accessOrderDeque, long now) {
+    long duration = expiresAfterAccessNanos();
     for (;;) {
       Node<K, V> node = accessOrderDeque.peekFirst();
-      if ((node == null) || (node.getAccessTime() > expirationTime)) {
+      if ((node == null) || ((now - node.getAccessTime()) < duration)) {
         return;
       }
       evictEntry(node, RemovalCause.EXPIRED, now);
@@ -705,10 +704,10 @@ void expireAfterWriteEntries(long now) {
     if (!expiresAfterWrite()) {
       return;
     }
-    long expirationTime = now - expiresAfterWriteNanos();
+    long duration = expiresAfterWriteNanos();
     for (;;) {
       final Node<K, V> node = writeOrderDeque().peekFirst();
-      if ((node == null) || (node.getWriteTime() > expirationTime)) {
+      if ((node == null) || ((now - node.getWriteTime()) < duration)) {
         break;
       }
       evictEntry(node, RemovalCause.EXPIRED, now);
@@ -762,12 +761,10 @@ boolean evictEntry(Node<K, V> node, RemovalCause cause, long now) {
         if (actualCause[0] == RemovalCause.EXPIRED) {
           boolean expired = false;
           if (expiresAfterAccess()) {
-            long expirationTime = now - expiresAfterAccessNanos();
-            expired |= (n.getAccessTime() <= expirationTime);
+            expired |= ((now - n.getAccessTime()) >= expiresAfterAccessNanos());
           }
           if (expiresAfterWrite()) {
-            long expirationTime = now - expiresAfterWriteNanos();
-            expired |= (n.getWriteTime() <= expirationTime);
+            expired |= ((now - n.getWriteTime()) >= expiresAfterWriteNanos());
           }
           if (expiresVariable()) {
             expired |= (n.getVariableTime() <= now);
@@ -1333,10 +1330,10 @@ public void run() {
       if (isComputingAsync(node)) {
         synchronized (node) {
           if (!Async.isReady((CompletableFuture<?>) node.getValue())) {
-            long expirationTime = expirationTicker().read() + Async.MAXIMUM_EXPIRY;
-            setWriteTime(node, expirationTime);
-            setAccessTime(node, expirationTime);
+            long expirationTime = expirationTicker().read() + Long.MAX_VALUE;
             setVariableTime(node, expirationTime);
+            setAccessTime(node, expirationTime);
+            setWriteTime(node, expirationTime);
           }
         }
       }
@@ -1745,9 +1742,8 @@ public V remove(Object key) {
    * @return the removed value or null if no mapping was found
    */
   V removeNoWriter(Object key) {
-    Node<K, V> node;
-    Object lookupKey = nodeFactory.newLookupKey(key);
-    if (!data.containsKey(lookupKey) || ((node = data.remove(lookupKey)) == null)) {
+    Node<K, V> node = data.remove(nodeFactory.newLookupKey(key));
+    if (node == null) {
       return null;
     }
 
@@ -1822,9 +1818,7 @@ V removeWithWriter(Object key) {
   @Override
   public boolean remove(Object key, Object value) {
     requireNonNull(key);
-
-    Object lookupKey = nodeFactory.newLookupKey(key);
-    if ((value == null) || !data.containsKey(lookupKey)) {
+    if (value == null) {
       return false;
     }
 
@@ -1837,7 +1831,7 @@ public boolean remove(Object key, Object value) {
     RemovalCause[] cause = new RemovalCause[1];
 
     long now = expirationTicker().read();
-    data.computeIfPresent(lookupKey, (kR, node) -> {
+    data.computeIfPresent(nodeFactory.newLookupKey(key), (kR, node) -> {
       synchronized (node) {
         oldKey[0] = node.getKey();
         oldValue[0] = node.getValue();

caffeine/src/main/java/com/github/benmanes/caffeine/cache/LocalAsyncLoadingCache.java

@@ -559,6 +559,11 @@ public int size() {
       return delegate.size();
     }
 
+    @Override
+    public void clear() {
+      delegate.clear();
+    }
+
     @Override
     public boolean containsKey(Object key) {
       return delegate.containsKey(key);
@@ -603,6 +608,28 @@ public V remove(Object key) {
       return Async.getWhenSuccessful(oldValueFuture);
     }
 
+    @Override
+    public boolean remove(Object key, Object value) {
+      requireNonNull(key);
+      if (value == null) {
+        return false;
+      }
+      CompletableFuture<V> oldValueFuture = delegate.get(key);
+      if ((oldValueFuture != null) && !value.equals(Async.getWhenSuccessful(oldValueFuture))) {
+        // Optimistically check if the current value is equal, but don't skip if it may be loading
+        return false;
+      }
+
+      @SuppressWarnings("unchecked")
+      K castedKey = (K) key;
+      boolean[] removed = { false };
+      delegate.compute(castedKey, (k, oldValue) -> {
+        removed[0] = value.equals(Async.getWhenSuccessful(oldValue));
+        return removed[0] ? null : oldValue;
+      }, /* recordStats */ false, /* recordLoad */ false);
+      return removed[0];
+    }
+
     @Override
     public V replace(K key, V value) {
       requireNonNull(value);
@@ -616,24 +643,19 @@ public boolean replace(K key, V oldValue, V newValue) {
       requireNonNull(oldValue);
       requireNonNull(newValue);
       CompletableFuture<V> oldValueFuture = delegate.get(key);
-      return oldValue.equals(Async.getIfReady(oldValueFuture))
-          && delegate.replace(key, oldValueFuture, CompletableFuture.completedFuture(newValue));
-    }
-
-    @Override
-    public boolean remove(Object key, Object value) {
-      requireNonNull(key);
-      if (value == null) {
+      if ((oldValueFuture != null) && !oldValue.equals(Async.getWhenSuccessful(oldValueFuture))) {
+        // Optimistically check if the current value is equal, but don't skip if it may be loading
         return false;
       }
-      CompletableFuture<V> oldValueFuture = delegate.get(key);
-      return value.equals(Async.getIfReady(oldValueFuture))
-          && delegate.remove(key, oldValueFuture);
-    }
 
-    @Override
-    public void clear() {
-      delegate.clear();
+      @SuppressWarnings("unchecked")
+      K castedKey = key;
+      boolean[] replaced = { false };
+      delegate.compute(castedKey, (k, value) -> {
+        replaced[0] = oldValue.equals(Async.getWhenSuccessful(value));
+        return replaced[0] ? CompletableFuture.completedFuture(newValue) : value;
+      }, /* recordStats */ false, /* recordLoad */ false);
+      return replaced[0];
     }
 
     @Override

caffeine/src/main/java/com/github/benmanes/caffeine/cache/UnboundedLocalCache.java

@@ -327,6 +327,11 @@ public boolean isEmpty() {
     return data.isEmpty();
   }
 
+  @Override
+  public int size() {
+    return data.size();
+  }
+
   @Override
   public void clear() {
     if (!hasRemovalListener() && (writer == CacheWriter.disabledWriter())) {
@@ -338,11 +343,6 @@ public void clear() {
     }
   }
 
-  @Override
-  public int size() {
-    return data.size();
-  }
-
   @Override
   public boolean containsKey(Object key) {
     return data.containsKey(key);