#5784

Committed 15 Sep 2026 02:57AM UTC coverage: 99.848% (-0.02%) from 99.869%

Build # #5784

Build Type

push

github

Committed by

ben-manes

Commit Message

keep jcache access extensions across expiry races

A JCache read extends an entry's access expiry in two lock-free writes,
to the native timer and to the wrapper, and "Reschedule jcache access
expiry outside the compute" had put the native timer first. A read that
judged its captured wrapper expired removed it by identity alone, so it
could discard an entry another read had just extended and publish
EXPIRED for it, and any core read between the two writes re-derived the
native deadline from the older wrapper, letting the entry expire early.
The wrapper is written first again, and the expired-read removals in
containsKey, get, getAll and the loading get now require the captured
wrapper to still be expired.

Coverage Stats

4742 of 4802 branches covered (98.75%)

13 of 13 new or added lines in 2 files covered. (100.0%)

6 existing lines in 4 files now uncovered.

9211 of 9225 relevant lines covered (99.85%)

1.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.88

/jcache/src/main/java/com/github/benmanes/caffeine/jcache/CacheProxy.java

/*
 * Copyright 2015 Ben Manes. All Rights Reserved.
 *
 * 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 com.github.benmanes.caffeine.jcache;

import static java.util.Objects.requireNonNull;
import static java.util.Objects.requireNonNullElse;
import static java.util.stream.Collectors.toMap;
import static java.util.stream.Collectors.toSet;
import static java.util.stream.Collectors.toUnmodifiableList;

import java.lang.System.Logger;
import java.lang.System.Logger.Level;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Supplier;

import javax.cache.Cache;
import javax.cache.CacheException;
import javax.cache.CacheManager;
import javax.cache.configuration.CacheEntryListenerConfiguration;
import javax.cache.configuration.Configuration;
import javax.cache.event.CacheEntryListenerException;
import javax.cache.expiry.Duration;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.integration.CacheLoader;
import javax.cache.integration.CacheLoaderException;
import javax.cache.integration.CacheWriter;
import javax.cache.integration.CacheWriterException;
import javax.cache.integration.CompletionListener;
import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.EntryProcessorResult;

import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;

import com.github.benmanes.caffeine.cache.Ticker;
import com.github.benmanes.caffeine.jcache.configuration.CaffeineConfiguration;
import com.github.benmanes.caffeine.jcache.copy.Copier;
import com.github.benmanes.caffeine.jcache.event.EventDispatcher;
import com.github.benmanes.caffeine.jcache.event.Registration;
import com.github.benmanes.caffeine.jcache.integration.DisabledCacheWriter;
import com.github.benmanes.caffeine.jcache.management.JCacheMXBean;
import com.github.benmanes.caffeine.jcache.management.JCacheStatisticsMXBean;
import com.github.benmanes.caffeine.jcache.management.JmxRegistration;
import com.github.benmanes.caffeine.jcache.management.JmxRegistration.MBeanType;
import com.github.benmanes.caffeine.jcache.processor.Action;
import com.github.benmanes.caffeine.jcache.processor.EntryProcessorEntry;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
import com.google.errorprone.annotations.Var;

/**
 * An implementation of JSR-107 {@link Cache} backed by a Caffeine cache.
 *
 * @author ben.manes@gmail.com (Ben Manes)
 */
@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
public class CacheProxy<K, V> implements Cache<K, V> {
  private static final Logger logger = System.getLogger(CacheProxy.class.getName());

  protected final com.github.benmanes.caffeine.cache.Cache<K, @Nullable Expirable<V>> cache;
  protected final Optional<CacheLoader<K, V>> cacheLoader;
  protected final Set<CompletableFuture<?>> inFlight;
  protected final JCacheStatisticsMXBean statistics;
  protected final EventDispatcher<K, V> dispatcher;
  protected final Executor executor;
  protected final Ticker ticker;

  private final CaffeineConfiguration<K, V> configuration;
  private final CacheManagerImpl cacheManager;
  private final CacheWriter<K, V> writer;
  private final JCacheMXBean cacheMxBean;
  private final ExpiryPolicy expiry;
  private final Copier copier;
  private final String name;

  private volatile boolean closed;

  @SuppressWarnings({"PMD.ExcessiveParameterList", "this-escape", "TooManyParameters"})
  public CacheProxy(String name, Executor executor, CacheManagerImpl cacheManager,
      CaffeineConfiguration<K, V> configuration,
      com.github.benmanes.caffeine.cache.Cache<K, @Nullable Expirable<V>> cache,
      EventDispatcher<K, V> dispatcher, Optional<CacheLoader<K, V>> cacheLoader,
      ExpiryPolicy expiry, Ticker ticker, JCacheStatisticsMXBean statistics, Copier copier) {
    var cacheWriter = configuration.getCacheWriter();
    this.writer = configuration.isWriteThrough()
        ? requireNonNull(cacheWriter, "The CacheWriter factory returned null")
        : requireNonNullElse(cacheWriter, DisabledCacheWriter.get());
    this.configuration = requireNonNull(configuration);
    this.cacheManager = requireNonNull(cacheManager);
    this.cacheLoader = requireNonNull(cacheLoader);
    this.inFlight = ConcurrentHashMap.newKeySet();
    this.dispatcher = requireNonNull(dispatcher);
    this.statistics = requireNonNull(statistics);
    this.cacheMxBean = new JCacheMXBean(this);
    this.executor = requireNonNull(executor);
    this.expiry = requireNonNull(expiry);
    this.ticker = requireNonNull(ticker);
    this.cache = requireNonNull(cache);
    this.copier = requireNonNull(copier);
    this.name = requireNonNull(name);
  }

  @Override
  public boolean containsKey(K key) {
    requireOperable();

    Expirable<V> expirable = cache.getIfPresent(key);
    if (expirable == null) {
      return false;
    }
    if (expirable.isEternal()) {
      return true;
    }
    long millis = currentTimeMillis();
    if (expirable.hasExpired(millis)) {
      dispatcher.beginComputation();
      try {
        cache.asMap().computeIfPresent(key, (k, e) -> {
          if ((e == expirable) && expirable.hasExpired(millis)) {
            dispatcher.publishExpired(this, key, expirable.get());
            statistics.recordEvictions(1L);
            return null;
          }
          return e;
        });
      } finally {
        dispatcher.endComputation();
      }
      dispatcher.awaitSynchronous();
      return false;
    }
    return true;
  }

  @Override
  public @Nullable V get(K key) {
    requireOperable();

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    @Var Expirable<V> expirable = cache.getIfPresent(key);
    if (expirable == null) {
      statistics.recordMisses(1L);
      if (statsEnabled) {
        statistics.recordGetTime(ticker.read() - start);
      }
      return null;
    }

    long millis;
    if (expirable.isEternal()) {
      millis = 0L;
    } else {
      long now = ticker.read();
      millis = nanosToMillis(now);
      if (expirable.hasExpired(millis)) {
        Expirable<V> current;
        var expired = expirable;
        dispatcher.beginComputation();
        try {
          current = cache.asMap().computeIfPresent(key, (k, e) -> {
            if ((e == expired) && expired.hasExpired(millis)) {
              dispatcher.publishExpired(this, key, expired.get());
              statistics.recordEvictions(1L);
              return null;
            }
            return e;
          });
        } finally {
          dispatcher.endComputation();
        }
        if ((current == null) || current.hasExpired(millis)) {
          var listenerFailure = awaitSynchronousFailure();
          statistics.recordMisses(1L);
          if (statsEnabled) {
            statistics.recordGetTime(ticker.read() - start);
          }
          rethrowListenerFailure(listenerFailure);
          return null;
        }
        expirable = current;
      }
    }

    var duration = getAccessExpireTime();
    setAccessExpireTime(expirable, duration, millis);
    setVariableExpiration(key, duration);
    V value = copyOf(expirable.get());
    if (statsEnabled) {
      statistics.recordHits(1L);
      statistics.recordGetTime(ticker.read() - start);
    }
    return value;
  }

  @Override
  public Map<K, V> getAll(Set<? extends K> keys) {
    requireOperable();

    boolean statsEnabled = statistics.isEnabled();
    long now = statsEnabled ? ticker.read() : 0L;
    Map<K, V> result;
    try {
      Map<K, Expirable<V>> entries = getAndFilterExpiredEntries(keys);
      result = copyMap(entries);
      if (statsEnabled) {
        statistics.recordGetTime(ticker.read() - now);
      }
    } catch (Throwable t) {
      awaitAndSuppressFailure(t);
      throw t;
    }
    rethrowListenerFailure(awaitSynchronousFailure());
    return result;
  }

  /**
   * Returns all of the mappings present, expiring as required, and updates their access expiry
   * time.
   */
  protected Map<K, Expirable<V>> getAndFilterExpiredEntries(Set<? extends K> keys) {
    int[] expired = { 0 };
    long[] millis = { 0L };
    int requested = keys.size();
    var result = new HashMap<K, @NonNull Expirable<V>>(cache.getAllPresent(keys));
    result.entrySet().removeIf(entry -> {
      if (!entry.getValue().isEternal() && (millis[0] == 0L)) {
        millis[0] = currentTimeMillis();
      }
      if (!entry.getValue().isEternal() && entry.getValue().hasExpired(millis[0])) {
        dispatcher.beginComputation();
        Expirable<V> current;
        try {
          current = cache.asMap().computeIfPresent(entry.getKey(), (k, expirable) -> {
            if ((expirable == entry.getValue()) && expirable.hasExpired(millis[0])) {
              dispatcher.publishExpired(this, entry.getKey(), entry.getValue().get());
              expired[0]++;
              return null;
            }
            return expirable;
          });
        } finally {
          dispatcher.endComputation();
        }
        if ((current == null) || current.hasExpired(millis[0])) {
          return true;
        }
        entry.setValue(current);
      }
      var duration = getAccessExpireTime();
      setAccessExpireTime(entry.getValue(), duration, millis[0]);
      setVariableExpiration(entry.getKey(), duration);
      return false;
    });

    statistics.recordHits(result.size());
    statistics.recordMisses(requested - result.size());
    statistics.recordEvictions(expired[0]);
    return result;
  }

  @Override
  @SuppressWarnings({"CollectionUndefinedEquality", "FutureReturnValueIgnored"})
  public void loadAll(Set<? extends K> keys, boolean replaceExistingValues,
      @Nullable CompletionListener completionListener) {
    requireOperable();
    keys.forEach(Objects::requireNonNull);
    CompletionListener listener = (completionListener == null)
        ? NullCompletionListener.INSTANCE
        : completionListener;
    if (cacheLoader.isEmpty()) {
      listener.onCompletion();
      return;
    }

    var future = new CompletableFuture<@Nullable Void>();
    synchronized (configuration) {
      requireNotClosed();
      inFlight.add(future);
    }
    try {
      // The tracked future spans the notification so that close() awaits the listener's callback
      CompletableFuture
          .supplyAsync(() -> loadAllAndNotify(keys, replaceExistingValues, listener), executor)
          .thenCompose(chain -> chain)
          .whenComplete((r, e) -> {
            inFlight.remove(future);
            future.complete(null);
          });
    } catch (RuntimeException e) {
      inFlight.remove(future);
      future.complete(null);
      listener.onException(new CacheLoaderException(e));
    }
  }

  /** Performs the bulk load and returns a future that includes its completion notification. */
  private CompletableFuture<@Nullable Void> loadAllAndNotify(Set<? extends K> keys,
      boolean replaceExistingValues, CompletionListener listener) {
    @Var CacheLoaderException failure = null;
    try {
      if (replaceExistingValues) {
        loadAllAndReplaceExisting(keys);
      } else {
        loadAllAndKeepExisting(keys);
      }
    } catch (CacheLoaderException e) {
      failure = e;
    } catch (Throwable t) {
      restoreInterrupt(t);
      failure = new CacheLoaderException(t);
    }
    var loadFailure = failure;
    return dispatcher.chainSynchronous().<@Nullable Void>handle((listenerFailure, error) -> {
      var dispatchFailure = (error == null) ? listenerFailure : new CacheLoaderException(error);
      var outcome = suppress(loadFailure, dispatchFailure);
      if (outcome == null) {
        listener.onCompletion();
      } else {
        listener.onException(outcome);
      }
      return null;
    });
  }

  /** Performs the bulk load where the existing entries are replaced. */
  protected void loadAllAndReplaceExisting(Set<? extends K> keys) {
    Map<K, V> loaded = cacheLoader.orElseThrow().loadAll(keys);
    for (var entry : loaded.entrySet()) {
      if ((entry.getKey() != null) && (entry.getValue() != null)) {
        putNoCopyOrAwait(entry.getKey(), entry.getValue(), /* publishToWriter= */ false);
      }
    }
  }

  /** Performs the bulk load where the existing entries are retained. */
  @SuppressWarnings("ConstantValue")
  protected void loadAllAndKeepExisting(Set<? extends K> keys) {
    List<K> keysToLoad = keys.stream()
        .filter(key -> {
          var expirable = cache.policy().getIfPresentQuietly(key);
          return (expirable == null)
              || (!expirable.isEternal() && expirable.hasExpired(currentTimeMillis()));
        }).collect(toUnmodifiableList());
    Map<K, V> result = cacheLoader.orElseThrow().loadAll(keysToLoad);
    for (var entry : result.entrySet()) {
      if ((entry.getKey() != null) && (entry.getValue() != null)) {
        putIfAbsentNoAwait(entry.getKey(), entry.getValue(), /* publishToWriter= */ false);
      }
    }
  }

  @Override
  public void put(K key, V value) {
    requireOperable();

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    var result = putNoCopyOrAwait(key, value, /* publishToWriter= */ true);
    var listenerFailure = awaitSynchronousFailure();
    if (statsEnabled && result.written) {
      statistics.recordPuts(1);
      statistics.recordPutTime(ticker.read() - start);
    }
    rethrowListenerFailure(listenerFailure);
  }

  @Override
  public @Nullable V getAndPut(K key, V value) {
    requireOperable();

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    var result = putNoCopyOrAwait(key, value, /* publishToWriter= */ true);
    var listenerFailure = awaitSynchronousFailure();
    if (statsEnabled) {
      if (result.oldValue == null) {
        statistics.recordMisses(1L);
      } else {
        statistics.recordHits(1L);
      }
      long duration = ticker.read() - start;
      if (result.written) {
        statistics.recordPuts(1);
        statistics.recordPutTime(duration);
      }
      statistics.recordGetTime(duration);
    }
    rethrowListenerFailure(listenerFailure);
    return (result.oldValue == null) ? null : copyOf(result.oldValue);
  }

  /**
   * Associates the specified value with the specified key in the cache.
   *
   * @param key key with which the specified value is to be associated
   * @param value value to be associated with the specified key
   * @param publishToWriter if the writer should be notified
   * @return the oldValue and if the new value was stored
   */
  @CanIgnoreReturnValue
  protected PutResult<V> putNoCopyOrAwait(K key, V value, boolean publishToWriter) {
    requireNonNull(key);
    requireNonNull(value);
    var entry = new CopiedEntry<>(key, copyOf(key), value, copyOf(value));
    return putCopiedOrAwait(entry, publishToWriter);
  }

  /**
   * Associates the copied value with the copied key in the cache.
   *
   * @param entry the caller's entry alongside the copies to store
   * @param publishToWriter if the writer should be notified
   * @return the oldValue and if the new value was stored
   */
  @CanIgnoreReturnValue
  private PutResult<V> putCopiedOrAwait(CopiedEntry<K, V> entry, boolean publishToWriter) {
    K key = entry.key;
    V newValue = entry.copiedValue;
    var result = new PutResult<V>();
    BiFunction<K, @Nullable Expirable<V>, @Nullable Expirable<V>> remappingFunction =
        (K k, @Var @Nullable Expirable<V> expirable) -> {
      if (publishToWriter) {
        publishToCacheWriter(writer::write, () -> new EntryProxy<>(key, entry.value));
      }
      if ((expirable != null) && !expirable.isEternal()
          && expirable.hasExpired(currentTimeMillis())) {
        dispatcher.publishExpired(this, key, expirable.get());
        statistics.recordEvictions(1L);
        expirable = null;
      }
      @Var long expireTimeMillis = getWriteExpireTimeMillis((expirable == null));
      if ((expirable != null) && (expireTimeMillis == Long.MIN_VALUE)) {
        expireTimeMillis = expirable.getExpireTimeMillis();
      }
      if ((expireTimeMillis == 0) && (expirable == null)) {
        result.written = false;
        dispatcher.publishExpired(this, key, newValue);
        return null;
      } else if (expirable == null) {
        dispatcher.publishCreated(this, key, newValue);
      } else {
        result.oldValue = expirable.get();
        dispatcher.publishUpdated(this, key, expirable.get(), newValue);
      }
      result.written = true;
      return new Expirable<>(newValue, expireTimeMillis);
    };
    dispatcher.beginComputation();
    try {
      cache.asMap().compute(entry.copiedKey, remappingFunction);
    } finally {
      dispatcher.endComputation();
    }
    return result;
  }

  @Override
  @SuppressWarnings("CatchingUnchecked")
  public void putAll(Map<? extends K, ? extends V> map) {
    requireOperable();

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    var copies = new ArrayList<CopiedEntry<K, V>>(map.size());
    for (var entry : map.entrySet()) {
      K key = requireNonNull(entry.getKey());
      V value = requireNonNull(entry.getValue());
      copies.add(new CopiedEntry<>(key, copyOf(key), value, copyOf(value)));
    }

    @Var CacheWriterException error = null;
    @Var Set<? extends K> failedKeys = Set.of();
    if (configuration.isWriteThrough() && !copies.isEmpty()) {
      var entries = new ArrayList<Cache.Entry<? extends K, ? extends V>>(copies.size());
      for (var copy : copies) {
        entries.add(new EntryProxy<>(copy.key, copy.value));
      }
      try {
        writer.writeAll(entries);
      } catch (CacheWriterException e) {
        failedKeys = entries.stream().map(Cache.Entry::getKey).collect(toSet());
        error = e;
      } catch (Exception e) {
        restoreInterrupt(e);
        failedKeys = entries.stream().map(Cache.Entry::getKey).collect(toSet());
        error = new CacheWriterException("Exception in CacheWriter", e);
      }
    }

    @Var int puts = 0;
    try {
      for (var copy : copies) {
        if (!failedKeys.contains(copy.key)) {
          var result = putCopiedOrAwait(copy, /* publishToWriter= */ false);
          if (result.written) {
            puts++;
          }
        }
      }
    } catch (Throwable t) {
      if (error != null) {
        t.addSuppressed(error);
      }
      awaitAndSuppressFailure(t);
      throw t;
    }
    var listenerFailure = awaitSynchronousFailure();

    if (statsEnabled && (puts > 0)) {
      statistics.recordPuts(puts);
      statistics.recordPutTime(ticker.read() - start);
    }
    var failure = suppress(error, listenerFailure);
    if (failure != null) {
      throw failure;
    }
  }

  @Override
  public boolean putIfAbsent(K key, V value) {
    requireOperable();
    requireNonNull(value);

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    var result = putIfAbsentNoAwait(key, value, /* publishToWriter= */ true);
    var listenerFailure = awaitSynchronousFailure();
    if (statsEnabled) {
      if (result.oldValue != null) {
        statistics.recordHits(1L);
      } else {
        statistics.recordMisses(1L);
      }
      if (result.written) {
        statistics.recordPuts(1L);
        statistics.recordPutTime(ticker.read() - start);
      }
    }
    rethrowListenerFailure(listenerFailure);
    return result.written;
  }

  /**
   * Associates the specified value with the specified key in the cache if there is no existing
   * mapping.
   *
   * @param key key with which the specified value is to be associated
   * @param value value to be associated with the specified key
   * @param publishToWriter if the writer should be notified
   * @return the oldValue and if the new value was stored
   */
  @CanIgnoreReturnValue
  private PutResult<V> putIfAbsentNoAwait(K key, V value, boolean publishToWriter) {
    var result = new PutResult<V>();
    BiFunction<K, @Nullable Expirable<V>, @Nullable Expirable<V>> remappingFunction =
        (K k, @Nullable Expirable<V> expirable) -> {
      if ((expirable != null)
          && (expirable.isEternal() || !expirable.hasExpired(currentTimeMillis()))) {
        result.oldValue = expirable.get();
        return expirable;
      }

      V copy = copyOf(value);
      if (publishToWriter) {
        publishToCacheWriter(writer::write, () -> new EntryProxy<>(key, value));
      }
      if (expirable != null) {
        dispatcher.publishExpired(this, key, expirable.get());
        statistics.recordEvictions(1L);
      }

      long expireTimeMillis = getWriteExpireTimeMillis(/* created= */ true);
      if (expireTimeMillis == 0) {
        // A zero creation expiry means the entry is already expired and is not added
        dispatcher.publishExpired(this, key, copy);
        return null;
      } else {
        result.written = true;
        dispatcher.publishCreated(this, key, copy);
        return new Expirable<>(copy, expireTimeMillis);
      }
    };
    dispatcher.beginComputation();
    try {
      cache.asMap().compute(copyOf(key), remappingFunction);
    } finally {
      dispatcher.endComputation();
    }
    return result;
  }

  @Override
  public boolean remove(K key) {
    requireOperable();
    requireNonNull(key);

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    V value = removeNoCopyOrAwait(key, /* publishToWriter= */ true);
    var listenerFailure = awaitSynchronousFailure();
    if (value != null) {
      statistics.recordRemovals(1L);
      if (statsEnabled) {
        statistics.recordRemoveTime(ticker.read() - start);
      }
    }
    rethrowListenerFailure(listenerFailure);
    return (value != null);
  }

  /**
   * Removes the mapping from the cache without store-by-value copying nor waiting for synchronous
   * listeners to complete.
   *
   * @param key key whose mapping is to be removed from the cache
   * @param publishToWriter if the writer should be notified
   * @return the old value
   */
  private @Nullable V removeNoCopyOrAwait(K key, boolean publishToWriter) {
    @SuppressWarnings("unchecked")
    var removed = (V[]) new Object[1];
    BiFunction<K, @Nullable Expirable<V>, @Nullable Expirable<V>> remappingFunction =
        (K k, @Nullable Expirable<V> expirable) -> {
      if (publishToWriter) {
        publishToCacheWriter(writer::delete, () -> key);
      }
      if (expirable != null) {
        if (!expirable.isEternal() && expirable.hasExpired(currentTimeMillis())) {
          dispatcher.publishExpired(this, key, expirable.get());
          statistics.recordEvictions(1L);
        } else {
          dispatcher.publishRemoved(this, key, expirable.get());
          removed[0] = expirable.get();
        }
      }
      return null;
    };
    dispatcher.beginComputation();
    try {
      cache.asMap().compute(key, remappingFunction);
    } finally {
      dispatcher.endComputation();
    }
    return removed[0];
  }

  @Override
  @CanIgnoreReturnValue
  public boolean remove(K key, V oldValue) {
    requireOperable();
    requireNonNull(key);
    requireNonNull(oldValue);

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    boolean[] removed = { false };
    boolean[] found = { false };
    dispatcher.beginComputation();
    try {
      cache.asMap().computeIfPresent(key, (k, expirable) -> {
        long millis = expirable.isEternal()
            ? 0L
            : nanosToMillis((start == 0L) ? ticker.read() : start);
        if (expirable.hasExpired(millis)) {
          dispatcher.publishExpired(this, key, expirable.get());
          statistics.recordEvictions(1L);
          return null;
        }

        found[0] = true;
        if (oldValue.equals(expirable.get())) {
          publishToCacheWriter(writer::delete, () -> key);
          dispatcher.publishRemoved(this, key, expirable.get());
          removed[0] = true;
          return null;
        }
        setAccessExpireTime(expirable, getAccessExpireTime(), millis);
        return expirable;
      });
    } finally {
      dispatcher.endComputation();
    }
    var listenerFailure = awaitSynchronousFailure();
    if (statsEnabled) {
      if (removed[0]) {
        statistics.recordRemovals(1L);
        statistics.recordHits(1L);
        statistics.recordRemoveTime(ticker.read() - start);
      } else if (found[0]) {
        statistics.recordHits(1L);
      } else {
        statistics.recordMisses(1L);
      }
    }
    rethrowListenerFailure(listenerFailure);
    return removed[0];
  }

  @Override
  public @Nullable V getAndRemove(K key) {
    requireOperable();
    requireNonNull(key);

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    V value = removeNoCopyOrAwait(key, /* publishToWriter= */ true);
    var listenerFailure = awaitSynchronousFailure();
    if (statsEnabled) {
      long duration = ticker.read() - start;
      if (value == null) {
        statistics.recordMisses(1L);
      } else {
        statistics.recordHits(1L);
        statistics.recordRemovals(1L);
        statistics.recordRemoveTime(duration);
      }
      statistics.recordGetTime(duration);
    }
    rethrowListenerFailure(listenerFailure);
    return (value == null) ? null : copyOf(value);
  }

  @Override
  public boolean replace(K key, V oldValue, V newValue) {
    requireOperable();
    requireNonNull(oldValue);
    requireNonNull(newValue);

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    boolean[] replaced = { false };
    boolean[] found = { false };
    dispatcher.beginComputation();
    try {
      cache.asMap().computeIfPresent(key, (k, expirable) -> {
        long millis = expirable.isEternal()
            ? 0L
            : nanosToMillis((start == 0L) ? ticker.read() : start);
        if (expirable.hasExpired(millis)) {
          dispatcher.publishExpired(this, key, expirable.get());
          statistics.recordEvictions(1L);
          return null;
        }

        found[0] = true;
        Expirable<V> result;
        if (oldValue.equals(expirable.get())) {
          V copy = copyOf(newValue);
          publishToCacheWriter(writer::write, () -> new EntryProxy<>(key, newValue));
          dispatcher.publishUpdated(this, key, expirable.get(), copy);
          @Var long expireTimeMillis = getWriteExpireTimeMillis(/* created= */ false);
          if (expireTimeMillis == Long.MIN_VALUE) {
            expireTimeMillis = expirable.getExpireTimeMillis();
          }
          result = new Expirable<>(copy, expireTimeMillis);
          replaced[0] = true;
        } else {
          result = expirable;
          setAccessExpireTime(expirable, getAccessExpireTime(), millis);
        }
        return result;
      });
    } finally {
      dispatcher.endComputation();
    }
    var listenerFailure = awaitSynchronousFailure();

    if (statsEnabled) {
      statistics.recordMisses(found[0] ? 0L : 1L);
      statistics.recordHits(found[0] ? 1L : 0L);
      long duration = ticker.read() - start;
      if (replaced[0]) {
        statistics.recordPuts(1L);
        statistics.recordPutTime(duration);
      }
      statistics.recordGetTime(duration);
    }

    rethrowListenerFailure(listenerFailure);
    return replaced[0];
  }

  @Override
  public boolean replace(K key, V value) {
    requireOperable();

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    var oldValue = replaceNoCopyOrAwait(key, value);
    var listenerFailure = awaitSynchronousFailure();
    if (oldValue == null) {
      statistics.recordMisses(1L);
      if (statsEnabled) {
        statistics.recordGetTime(ticker.read() - start);
      }
    } else if (statsEnabled) {
      statistics.recordHits(1L);
      statistics.recordPuts(1L);
      long duration = ticker.read() - start;
      statistics.recordGetTime(duration);
      statistics.recordPutTime(duration);
    }
    rethrowListenerFailure(listenerFailure);
    return (oldValue != null);
  }

  @Override
  public @Nullable V getAndReplace(K key, V value) {
    requireOperable();

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    V oldValue = replaceNoCopyOrAwait(key, value);
    var listenerFailure = awaitSynchronousFailure();
    if (statsEnabled) {
      long duration = ticker.read() - start;
      if (oldValue == null) {
        statistics.recordMisses(1L);
      } else {
        statistics.recordHits(1L);
        statistics.recordPuts(1L);
        statistics.recordPutTime(duration);
      }
      statistics.recordGetTime(duration);
    }
    rethrowListenerFailure(listenerFailure);
    return (oldValue == null) ? null : copyOf(oldValue);
  }

  /**
   * Replaces the entry for the specified key only if it is currently mapped to some value. The
   * entry is not store-by-value copied nor does the method wait for synchronous listeners to
   * complete.
   *
   * @param key key with which the specified value is associated
   * @param value value to be associated with the specified key
   * @return the old value
   */
  private @Nullable V replaceNoCopyOrAwait(K key, V value) {
    requireNonNull(value);
    @SuppressWarnings("unchecked")
    var replaced = (V[]) new Object[1];
    dispatcher.beginComputation();
    try {
      cache.asMap().computeIfPresent(key, (k, expirable) -> {
        if (!expirable.isEternal() && expirable.hasExpired(currentTimeMillis())) {
          dispatcher.publishExpired(this, key, expirable.get());
          statistics.recordEvictions(1L);
          return null;
        }

        V copy = copyOf(value);
        publishToCacheWriter(writer::write, () -> new EntryProxy<>(key, value));
        @Var long expireTimeMillis = getWriteExpireTimeMillis(/* created= */ false);
        if (expireTimeMillis == Long.MIN_VALUE) {
          expireTimeMillis = expirable.getExpireTimeMillis();
        }
        dispatcher.publishUpdated(this, key, expirable.get(), copy);
        replaced[0] = expirable.get();
        return new Expirable<>(copy, expireTimeMillis);
      });
    } finally {
      dispatcher.endComputation();
    }
    return replaced[0];
  }

  @Override
  @SuppressWarnings("CatchingUnchecked")
  public void removeAll(Set<? extends K> keys) {
    requireOperable();
    var keysToRemove = new LinkedHashSet<>(keys);
    keysToRemove.forEach(Objects::requireNonNull);

    @Var CacheWriterException error = null;
    @Var Set<? extends K> failedKeys = Set.of();
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    if (configuration.isWriteThrough() && !keysToRemove.isEmpty()) {
      var keysToWrite = new LinkedHashSet<>(keysToRemove);
      try {
        writer.deleteAll(keysToWrite);
      } catch (CacheWriterException e) {
        error = e;
        failedKeys = keysToWrite;
      } catch (Exception e) {
        restoreInterrupt(e);
        error = new CacheWriterException("Exception in CacheWriter", e);
        failedKeys = keysToWrite;
      }
    }

    @Var int removed = 0;
    try {
      for (var key : keysToRemove) {
        if (!failedKeys.contains(key)
            && (removeNoCopyOrAwait(key, /* publishToWriter= */ false) != null)) {
          removed++;
        }
      }
    } catch (Throwable t) {
      if (error != null) {
        t.addSuppressed(error);
      }
      awaitAndSuppressFailure(t);
      throw t;
    }
    var listenerFailure = awaitSynchronousFailure();

    if (statsEnabled && (removed > 0)) {
      statistics.recordRemovals(removed);
      statistics.recordRemoveTime(ticker.read() - start);
    }
    var failure = suppress(error, listenerFailure);
    if (failure != null) {
      throw failure;
    }
  }

  @Override
  public void removeAll() {
    removeAll(cache.asMap().keySet());
  }

  @Override
  public void clear() {
    requireOperable();
    dispatcher.beginComputation();
    try {
      cache.invalidateAll();
    } finally {
      dispatcher.endComputation();
    }
  }

  @Override
  public <C extends Configuration<K, V>> C getConfiguration(Class<C> clazz) {
    if (clazz.isInstance(configuration)) {
      synchronized (configuration) {
        return clazz.cast(configuration.immutableCopy());
      }
    }
    throw new IllegalArgumentException("The configuration class " + clazz
        + " is not supported by this implementation");
  }

  @Override
  public <T extends @Nullable Object> T invoke(K key,
      EntryProcessor<K, V, T> entryProcessor, Object @Nullable ... arguments) {
    requireNonNull(entryProcessor);
    requireOperable();

    var result = new Object[1];
    var failure = new Throwable[1];
    BiFunction<K, @Nullable Expirable<V>, @Nullable Expirable<V>> remappingFunction =
        (k, expirable) -> {
      // Publish a lazily-expired prior's expiration before the processor observes it as absent,
      // so listeners see a linearizable sequence and the expiration is committed exactly once
      boolean expired;
      Expirable<V> prior;
      if ((expirable != null) && !expirable.isEternal()
          && expirable.hasExpired(currentTimeMillis())) {
        dispatcher.publishExpired(this, key, expirable.get());
        statistics.recordEvictions(1L);
        expired = true;
        prior = null;
      } else {
        prior = expirable;
        expired = false;
      }

      V value;
      if (prior == null) {
        statistics.recordMisses(1L);
        value = null;
      } else {
        value = copyOf(prior.get());
        statistics.recordHits(1L);
      }
      var entry = new EntryProcessorEntry<>(key, value,
          configuration.isReadThrough() ? cacheLoader : Optional.empty());
      try {
        result[0] = entryProcessor.process(entry, arguments);
        return postProcess(prior, entry);
      } catch (Throwable e) {
        if (!expired) {
          throw processorFailure(e);
        }
        failure[0] = e;
        return null;
      }
    };
    try {
      dispatcher.beginComputation();
      try {
        cache.asMap().compute(copyOf(key), remappingFunction);
      } finally {
        dispatcher.endComputation();
      }
    } catch (Throwable t) {
      dispatcher.ignoreSynchronous();
      throw t;
    }
    var listenerFailure = awaitSynchronousFailure();
    if (failure[0] != null) {
      var error = (failure[0] instanceof Error) ? failure[0] : processorFailure(failure[0]);
      if (listenerFailure != null) {
        error.addSuppressed(listenerFailure);
      }
      throw processorFailure(error);
    }
    rethrowListenerFailure(listenerFailure);

    @SuppressWarnings("unchecked")
    var castedResult = (T) result[0];
    return castedResult;
  }

  /**
   * Waits for the synchronous listeners and returns their failure instead of throwing it. An
   * {@code Error} is not captured and propagates.
   */
  protected final @Nullable CacheEntryListenerException awaitSynchronousFailure() {
    try {
      dispatcher.awaitSynchronous();
      return null;
    } catch (CacheEntryListenerException e) {
      return e;
    }
  }

  /** Waits for the synchronous listeners and suppresses any failures onto the existing error. */
  @SuppressWarnings("CheckReturnValue")
  protected final void awaitAndSuppressFailure(Throwable error) {
    suppress(error, awaitSynchronousFailure());
  }

  /** Rethrows a captured synchronous listener failure after the operation is accounted for. */
  protected static void rethrowListenerFailure(@Nullable CacheEntryListenerException failure) {
    if (failure != null) {
      throw failure;
    }
  }

  /** Returns the exception to rethrow on an entry processor failure. */
  private static RuntimeException processorFailure(Throwable e) {
    if (e instanceof Error) {
      throw (Error) e;
    } else if (e instanceof EntryProcessorException) {
      return (EntryProcessorException) e;
    }
    return new EntryProcessorException(e);
  }

  /** Restores the thread's interrupt status if the failure is an interruption. */
  private static void restoreInterrupt(Throwable failure) {
    if (failure instanceof InterruptedException) {
      Thread.currentThread().interrupt();
    }
  }

  /**
   * Returns the updated expirable value after performing the post-processing actions. A null
   * {@code expirable} means the entry was absent and READ/UPDATED (which require a live prior)
   * never observe one.
   */
  @Nullable Expirable<V> postProcess(
      @Nullable Expirable<V> expirable, EntryProcessorEntry<K, V> entry) {
    switch (entry.getAction()) {
      case NONE:
        return expirable;
      case READ: {
        setAccessExpireTime(requireNonNull(expirable), getAccessExpireTime(), 0L);
        return expirable;
      }
      case CREATED:
      case LOADED: {
        V value = requireNonNull(entry.getValue());
        V copy = copyOf(value);
        if (entry.getAction() == Action.CREATED) {
          publishToCacheWriter(writer::write, () -> new EntryProxy<>(entry.getKey(), value));
        }
        long expireTimeMillis = getWriteExpireTimeMillis(/* created= */ true);
        if (expireTimeMillis == 0) {
          // A zero creation expiry means the entry is already expired and is not added, so the
          // create is neither counted nor published
          dispatcher.publishExpired(this, entry.getKey(), copy);
          return null;
        }
        if (entry.getAction() == Action.CREATED) {
          statistics.recordPuts(1L);
        }
        dispatcher.publishCreated(this, entry.getKey(), copy);
        return new Expirable<>(copy, expireTimeMillis);
      }
      case UPDATED: {
        requireNonNull(expirable, "Expected a previous value but was null");
        V value = requireNonNull(entry.getValue(), "Expected a new value but was null");
        V copy = copyOf(value);
        publishToCacheWriter(writer::write, () -> new EntryProxy<>(entry.getKey(), value));
        statistics.recordPuts(1L);
        dispatcher.publishUpdated(this, entry.getKey(), expirable.get(), copy);
        @Var long expireTimeMillis = getWriteExpireTimeMillis(/* created= */ false);
        if (expireTimeMillis == Long.MIN_VALUE) {
          expireTimeMillis = expirable.getExpireTimeMillis();
        }
        return new Expirable<>(copy, expireTimeMillis);
      }
      case DELETED:
        publishToCacheWriter(writer::delete, entry::getKey);
        if (expirable != null) {
          statistics.recordRemovals(1L);
          dispatcher.publishRemoved(this, entry.getKey(), expirable.get());
        }
        return null;
    }
    throw new IllegalStateException("Unknown state: " + entry.getAction());
  }

  @Override
  public <T> Map<K, EntryProcessorResult<T>> invokeAll(Set<? extends K> keys,
      EntryProcessor<K, V, T> entryProcessor, Object @Nullable ... arguments) {
    requireOperable();
    requireNonNull(keys);
    requireNonNull(entryProcessor);
    keys.forEach(Objects::requireNonNull);

    var results = new HashMap<K, EntryProcessorResult<T>>(keys.size(), 1.0f);
    for (K key : keys) {
      try {
        T result = invoke(key, entryProcessor, arguments);
        if (result != null) {
          results.put(key, () -> result);
        }
      } catch (EntryProcessorException e) {
        results.put(key, () -> { throw e; });
      } catch (RuntimeException e) {
        results.put(key, () -> { throw new EntryProcessorException(e); });
      }
    }
    return results;
  }

  @Override
  public String getName() {
    return name;
  }

  @Override
  public CacheManager getCacheManager() {
    return cacheManager;
  }

  @Override
  public boolean isClosed() {
    return closed;
  }

  @Override
  public void close() {
    if (isClosed()) {
      return;
    }
    @Var Throwable thrown = null;
    synchronized (configuration) {
      if (isClosed()) {
        return;
      }
      thrown = tryClose((AutoCloseable) () -> enableManagement(false), thrown);
      thrown = tryClose((AutoCloseable) () -> enableStatistics(false), thrown);

      closed = true;
      try {
        cacheManager.destroyCache(name, this);
      } catch (IllegalStateException ignored) { /* manager already closed */ }
    }

    thrown = shutdownExecutor(thrown);
    thrown = tryClose(expiry, thrown);
    thrown = tryClose(writer, thrown);
    thrown = tryClose(cacheLoader.orElse(null), thrown);
    for (Registration<K, V> registration : dispatcher.registrations()) {
      thrown = tryClose(registration.getCacheEntryListener(), thrown);
    }
    if (thrown != null) {
      logger.log(Level.WARNING, "Failure when closing cache resources", thrown);
    }
    dispatcher.beginComputation();
    try {
      cache.invalidateAll();
    } finally {
      dispatcher.endComputation();
    }
  }

  @SuppressWarnings("FutureReturnValueIgnored")
  private @Nullable Throwable shutdownExecutor(@Var @Nullable Throwable thrown) {
    if (executor instanceof ExecutorService) {
      @SuppressWarnings("PMD.CloseResource")
      var es = (ExecutorService) executor;
      thrown = tryClose((AutoCloseable) es::shutdown, thrown);
    }

    try {
      CompletableFuture
          .allOf(inFlight.toArray(CompletableFuture[]::new))
          .get(10, TimeUnit.SECONDS);
    } catch (ExecutionException | TimeoutException e) {
      thrown = suppress(thrown, e);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      thrown = suppress(thrown, e);
    }
    inFlight.clear();

    if (!(executor instanceof ExecutorService)) {
      thrown = tryClose(executor, thrown);
    }
    return thrown;
  }

  /**
   * Attempts to close the resource. If an error occurs and an outermost exception is set, then adds
   * the error to the suppression list.
   *
   * @param o the resource to close if Closeable
   * @param outer the outermost error, or null if unset
   * @return the outermost error, or null if unset and successful
   */
  private static @Nullable Throwable tryClose(@Nullable Object o, @Nullable Throwable outer) {
    if (o instanceof AutoCloseable) {
      try {
        ((AutoCloseable) o).close();
      } catch (Throwable t) {
        return suppress(outer, t);
      }
    }
    return outer;
  }

  /** Returns the outermost error, retaining the error as suppressed if one is already set. */
  private static <T extends Throwable> @Nullable T suppress(@Nullable T outer, @Nullable T t) {
    if (t == null) {
      return outer;
    }
    if (outer == null) {
      return t;
    }
    if (outer != t) {
      outer.addSuppressed(t);
    }
    return outer;
  }

  @Override
  public <T> T unwrap(Class<T> clazz) {
    if (clazz.isInstance(cache)) {
      return clazz.cast(cache);
    } else if (clazz.isInstance(this)) {
      return clazz.cast(this);
    }
    throw new IllegalArgumentException("Unwrapping to " + clazz
        + " is not supported by this implementation");
  }

  @Override
  public void registerCacheEntryListener(
      CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration) {
    synchronized (configuration) {
      requireOperable();
      configuration.addCacheEntryListenerConfiguration(cacheEntryListenerConfiguration);
      try {
        dispatcher.register(cacheEntryListenerConfiguration);
      } catch (Throwable t) {
        configuration.removeCacheEntryListenerConfiguration(cacheEntryListenerConfiguration);
        throw t;
      }
    }
  }

  @Override
  public void deregisterCacheEntryListener(
      CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration) {
    synchronized (configuration) {
      requireOperable();
      configuration.removeCacheEntryListenerConfiguration(cacheEntryListenerConfiguration);
      dispatcher.deregister(cacheEntryListenerConfiguration);
    }
  }

  @Override
  public Iterator<Cache.Entry<K, V>> iterator() {
    requireOperable();
    return new EntryIterator();
  }

  /** Enables or disables the configuration management JMX bean. */
  void enableManagement(boolean enabled) {
    synchronized (configuration) {
      requireNotClosed();
      if (enabled) {
        JmxRegistration.registerMxBean(this, cacheMxBean, MBeanType.CONFIGURATION);
      } else if (configuration.isManagementEnabled()) {
        JmxRegistration.unregisterMxBean(this, MBeanType.CONFIGURATION);
      }
      configuration.setManagementEnabled(enabled);
    }
  }

  /** Enables or disables the statistics JMX bean. */
  void enableStatistics(boolean enabled) {
    synchronized (configuration) {
      requireNotClosed();
      if (enabled) {
        JmxRegistration.registerMxBean(this, statistics, MBeanType.STATISTICS);
      } else if (configuration.isStatisticsEnabled()) {
        JmxRegistration.unregisterMxBean(this, MBeanType.STATISTICS);
      }
      statistics.enable(enabled);
      configuration.setStatisticsEnabled(enabled);
    }
  }

  /** Performs the action with the cache writer if write-through is enabled. */
  @SuppressWarnings("CatchingUnchecked")
  private <T> void publishToCacheWriter(Consumer<T> action, Supplier<T> data) {
    if (!configuration.isWriteThrough()) {
      return;
    }
    try {
      action.accept(data.get());
    } catch (CacheWriterException e) {
      throw e;
    } catch (Exception e) {
      restoreInterrupt(e);
      throw new CacheWriterException("Exception in CacheWriter", e);
    }
  }

  /** Checks that the cache is not closed. */
  protected final void requireNotClosed() {
    if (isClosed()) {
      throw new IllegalStateException();
    }
  }

  /** Checks that the operation may begin. */
  protected final void requireOperable() {
    requireNotClosed();
    if (dispatcher.inCallback()) {
      throw new IllegalStateException("Recursive cache operation");
    }
  }

  /**
   * Returns a copy of the value if value-based caching is enabled.
   *
   * @param object the object to be copied
   * @param <T> the type of object being copied
   * @return a copy of the object if storing by value or the same instance if by reference
   */
  @SuppressWarnings("CatchingUnchecked")
  protected final <T> T copyOf(T object) {
    try {
      return requireNonNull(
          copier.copy(requireNonNull(object), requireNonNull(cacheManager.getClassLoader())));
    } catch (NullPointerException | IllegalStateException | ClassCastException | CacheException e) {
      throw e;
    } catch (Exception e) {
      restoreInterrupt(e);
      throw new CacheException(e);
    }
  }

  /**
   * Returns a deep copy of the map if value-based caching is enabled.
   *
   * @param map the mapping of keys to expirable values
   * @return a deep or shallow copy of the mappings depending on the store by value setting
   */
  @SuppressWarnings("CollectorMutability")
  protected final Map<K, V> copyMap(Map<K, Expirable<V>> map) {
    return map.entrySet().stream().collect(toMap(
        entry -> copyOf(entry.getKey()),
        entry -> copyOf(entry.getValue().get())));
  }

  /** Returns the current time in milliseconds. */
  protected final long currentTimeMillis() {
    return nanosToMillis(ticker.read());
  }

  /** Returns the nanosecond time in milliseconds. */
  protected static long nanosToMillis(long nanos) {
    return TimeUnit.NANOSECONDS.toMillis(nanos);
  }

  /** Returns the duration to expire an accessed entry after, or {@code null} if unchanged. */
  @SuppressWarnings("CatchingUnchecked")
  protected final @Nullable Duration getAccessExpireTime() {
    try {
      return expiry.getExpiryForAccess();
    } catch (Exception e) {
      restoreInterrupt(e);
      logger.log(Level.WARNING, "Failed to get the policy's expiration time", e);
      return null;
    }
  }

  /**
   * Sets the JCache access expiration time.
   *
   * @param expirable the entry that was operated on
   * @param duration the access duration, or null if unchanged
   * @param currentTimeMillis the current time, or zero if not read yet
   */
  protected final void setAccessExpireTime(Expirable<?> expirable,
      @Nullable Duration duration, @Var long currentTimeMillis) {
    if (duration == null) {
      return;
    } else if (duration.isZero()) {
      expirable.setExpireTimeMillis(0L);
    } else if (duration.isEternal()) {
      expirable.setExpireTimeMillis(Long.MAX_VALUE);
    } else {
      if (currentTimeMillis == 0L) {
        currentTimeMillis = currentTimeMillis();
      }
      @Var long expireTimeMillis = duration.getAdjustedTime(currentTimeMillis);
      expireTimeMillis = ((expireTimeMillis == 0L) || (expireTimeMillis == Long.MAX_VALUE))
          ? (expireTimeMillis - 1)
          : expireTimeMillis;
      expirable.setExpireTimeMillis(expireTimeMillis);
    }
  }

  /**
   * Sets the native access expiration time.
   *
   * @param key the entry's key
   * @param duration the access duration, or null if unchanged
   */
  protected final void setVariableExpiration(K key, @Nullable Duration duration) {
    if (duration == null) {
      return;
    }
    cache.policy().expireVariably().ifPresent(policy -> {
      if (duration.isZero()) {
        policy.setExpiresAfter(key, 0L, TimeUnit.NANOSECONDS);
      } else if (duration.isEternal()) {
        policy.setExpiresAfter(key, Long.MAX_VALUE, TimeUnit.NANOSECONDS);
      } else {
        policy.setExpiresAfter(key, duration.getDurationAmount(), duration.getTimeUnit());
      }
    });
  }

  /**
   * Returns the time when the entry will expire.
   *
   * @param created if the write operation is an insert or an update
   * @return the time when the entry will expire, zero if it should expire immediately,
   *         Long.MIN_VALUE if it should not be changed, or Long.MAX_VALUE if eternal
   */
  @SuppressWarnings("CatchingUnchecked")
  protected final long getWriteExpireTimeMillis(boolean created) {
    try {
      Duration duration = created ? expiry.getExpiryForCreation() : expiry.getExpiryForUpdate();
      if (duration == null) {
        return created ? Long.MAX_VALUE : Long.MIN_VALUE;
      } else if (duration.isZero()) {
        return 0L;
      } else if (duration.isEternal()) {
        return Long.MAX_VALUE;
      }
      long expireTimeMillis = duration.getAdjustedTime(currentTimeMillis());
      return ((expireTimeMillis == 0L) || (expireTimeMillis == Long.MAX_VALUE))
          ? (expireTimeMillis - 1)
          : expireTimeMillis;
    } catch (Exception e) {
      restoreInterrupt(e);
      logger.log(Level.WARNING, "Failed to get the policy's expiration time", e);
      return created ? Long.MAX_VALUE : Long.MIN_VALUE;
    }
  }

  /** An iterator to safely expose the cache entries. */
  final class EntryIterator implements Iterator<Cache.Entry<K, V>> {
    // NullAway does not yet understand the @NonNull annotation in the return type of asMap.
    @SuppressWarnings("NullAway")
    final Iterator<Map.Entry<K, Expirable<V>>> delegate = cache.asMap().entrySet().iterator();

    Map.@Nullable Entry<K, Expirable<V>> current;
    Map.@Nullable Entry<K, Expirable<V>> cursor;

    @Override
    public boolean hasNext() {
      while ((cursor == null) && delegate.hasNext()) {
        Map.Entry<K, Expirable<V>> entry = delegate.next();
        long millis = entry.getValue().isEternal() ? 0L : currentTimeMillis();
        if (!entry.getValue().hasExpired(millis)) {
          var duration = getAccessExpireTime();
          setAccessExpireTime(entry.getValue(), duration, millis);
          setVariableExpiration(entry.getKey(), duration);
          cursor = entry;
        }
      }
      return (cursor != null);
    }

    @Override
    public Cache.Entry<K, V> next() {
      if (!hasNext()) {
        throw new NoSuchElementException();
      }
      var entry = requireNonNull(cursor);
      cursor = null;
      var copy = new EntryProxy<>(copyOf(entry.getKey()), copyOf(entry.getValue().get()));
      statistics.recordHits(1L);
      current = entry;
      return copy;
    }

    @Override
    @SuppressWarnings("CheckReturnValue")
    public void remove() {
      if (current == null) {
        throw new IllegalStateException();
      }
      CacheProxy.this.remove(current.getKey());
      current = null;
    }
  }

  /** An entry paired with the store-by-value copies to be stored in its place. */
  private static final class CopiedEntry<K, V> {
    final K key;
    final V value;
    final K copiedKey;
    final V copiedValue;

    CopiedEntry(K key, K copiedKey, V value, V copiedValue) {
      this.copiedValue = copiedValue;
      this.copiedKey = copiedKey;
      this.value = value;
      this.key = key;
    }
  }

  protected static final class PutResult<V> {
    @Nullable V oldValue;
    boolean written;
  }

  protected enum NullCompletionListener implements CompletionListener {
    INSTANCE;

    @Override
    public void onCompletion() {}

    @Override
    public void onException(Exception e) {}
  }
}

ben-manes / caffeine / #5784

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous