#5669

Committed 21 Jul 2026 02:11AM UTC coverage: 99.965% (-0.01%) from 99.976%

Build # #5669

Build Type

push

github

Committed by

ben-manes

Commit Message

Record a read-through get miss on a loader throw

LoadingCacheProxy.getOrLoad recorded the miss after cache.get runs the
read-through load, so a throwing loader propagated before the miss was
counted -- getCacheMisses stayed 0. The loading getAll, the RI
(increaseCacheMisses before cacheLoader.load), and core all count the
miss regardless of load outcome (the find already failed). Record it
before the load. Not TCK-observable (its stats tests use no loader).

Pin untested jcache close-gate and config paths

Add coverage for public-API paths the TCK never exercises: clear() and
register/deregisterCacheEntryListener on a closed cache throw ISE
(CG-1, CG-2); createCache of a name defined in application.conf throws
CacheException (CG-6); and createCache with a plain non-Complete
Configuration (types + store-by-value only) succeeds via the basic
resolveConfigurationFor branch (CG-7). Test-only; the code was already
correct.

Coverage Stats

4194 of 4203 branches covered (99.79%)

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

2 existing lines in 2 files now uncovered.

8472 of 8475 relevant lines covered (99.96%)

1.0 hits per line

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

99.87

/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.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) {
    this.writer = requireNonNullElse(configuration.getCacheWriter(), 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) {
    requireNotClosed();
    Expirable<V> expirable = cache.getIfPresent(key);
    if (expirable == null) {
      return false;
    }
    if (!expirable.isEternal() && expirable.hasExpired(currentTimeMillis())) {
      cache.asMap().computeIfPresent(key, (k, e) -> {
        if (e == expirable) {
          dispatcher.publishExpired(this, key, expirable.get());
          statistics.recordEvictions(1L);
          return null;
        }
        return e;
      });
      dispatcher.awaitSynchronous();
      return false;
    }
    return true;
  }

  @Override
  public @Nullable V get(K key) {
    requireNotClosed();
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    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)) {
        cache.asMap().computeIfPresent(key, (k, e) -> {
          if (e == expirable) {
            dispatcher.publishExpired(this, key, expirable.get());
            statistics.recordEvictions(1L);
            return null;
          }
          return e;
        });
        dispatcher.awaitSynchronous();
        statistics.recordMisses(1L);
        if (statsEnabled) {
          statistics.recordGetTime(ticker.read() - start);
        }
        return null;
      }
    }

    var duration = getAccessExpireTime();
    setVariableExpiration(key, duration);
    setAccessExpireTime(expirable, duration, millis);
    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) {
    requireNotClosed();

    boolean statsEnabled = statistics.isEnabled();
    long now = statsEnabled ? ticker.read() : 0L;
    try {
      Map<K, Expirable<V>> result = getAndFilterExpiredEntries(keys);
      if (statsEnabled) {
        statistics.recordGetTime(ticker.read() - now);
      }
      return copyMap(result);
    } finally {
      dispatcher.awaitSynchronous();
    }
  }

  /**
   * 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 };
    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().hasExpired(millis[0])) {
        cache.asMap().computeIfPresent(entry.getKey(), (k, expirable) -> {
          if (expirable == entry.getValue()) {
            dispatcher.publishExpired(this, entry.getKey(), entry.getValue().get());
            expired[0]++;
            return null;
          }
          return expirable;
        });
        return true;
      }
      var duration = getAccessExpireTime();
      setVariableExpiration(entry.getKey(), duration);
      setAccessExpireTime(entry.getValue(), duration, millis[0]);
      return false;
    });

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

  @Override
  @SuppressWarnings({"CollectionUndefinedEquality", "FutureReturnValueIgnored"})
  public void loadAll(Set<? extends K> keys, boolean replaceExistingValues,
      @Nullable CompletionListener completionListener) {
    requireNotClosed();
    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 {
      CompletableFuture.runAsync(() -> {
        @Var boolean success = false;
        try {
          if (replaceExistingValues) {
            loadAllAndReplaceExisting(keys);
          } else {
            loadAllAndKeepExisting(keys);
          }
          success = true;
        } catch (CacheLoaderException e) {
          dispatcher.ignoreSynchronous();
          listener.onException(e);
        } catch (RuntimeException e) {
          dispatcher.ignoreSynchronous();
          listener.onException(new CacheLoaderException(e));
        }
        if (success) {
          dispatcher.chainSynchronous().whenComplete((failure, error) -> {
            if (error != null) {
              listener.onException(new CacheLoaderException(error));
            } else if (failure != null) {
              listener.onException(failure);
            } else {
              listener.onCompletion();
            }
          });
        }
      }, executor).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 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) {
    requireNotClosed();
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    var result = putNoCopyOrAwait(key, value, /* publishToWriter= */ true);
    dispatcher.awaitSynchronous();

    if (statsEnabled && result.written) {
      statistics.recordPuts(1);
      statistics.recordPutTime(ticker.read() - start);
    }
  }

  @Override
  public @Nullable V getAndPut(K key, V value) {
    requireNotClosed();
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    var result = putNoCopyOrAwait(key, value, /* publishToWriter= */ true);
    dispatcher.awaitSynchronous();

    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);
    }
    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);

    V newValue = copyOf(value);
    var result = new PutResult<V>();
    cache.asMap().compute(copyOf(key), (K k, @Var Expirable<V> expirable) -> {
      if (publishToWriter) {
        publishToCacheWriter(writer::write, () -> new EntryProxy<>(key, 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);
    });
    return result;
  }

  @Override
  public void putAll(Map<? extends K, ? extends V> map) {
    requireNotClosed();
    for (var entry : map.entrySet()) {
      requireNonNull(entry.getKey());
      requireNonNull(entry.getValue());
    }

    @Var CacheWriterException error = null;
    @Var Set<? extends K> failedKeys = Set.of();
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    if (configuration.isWriteThrough() && !map.isEmpty()) {
      var entries = new ArrayList<Cache.Entry<? extends K, ? extends V>>(map.size());
      for (var entry : map.entrySet()) {
        entries.add(new EntryProxy<>(entry.getKey(), entry.getValue()));
      }
      try {
        writer.writeAll(entries);
      } catch (CacheWriterException e) {
        failedKeys = entries.stream().map(Cache.Entry::getKey).collect(toSet());
        error = e;
      } catch (RuntimeException e) {
        failedKeys = entries.stream().map(Cache.Entry::getKey).collect(toSet());
        error = new CacheWriterException("Exception in CacheWriter", e);
      }
    }

    @Var int puts = 0;
    try {
      for (var entry : map.entrySet()) {
        if (!failedKeys.contains(entry.getKey())) {
          var result = putNoCopyOrAwait(entry.getKey(),
              entry.getValue(), /* publishToWriter= */ false);
          if (result.written) {
            puts++;
          }
        }
      }
    } finally {
      dispatcher.awaitSynchronous();
    }

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

  @Override
  public boolean putIfAbsent(K key, V value) {
    requireNotClosed();
    requireNonNull(value);
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    var result = putIfAbsentNoAwait(key, value, /* publishToWriter= */ true);
    dispatcher.awaitSynchronous();

    if (statsEnabled) {
      if (result.oldValue != null) {
        statistics.recordHits(1L);
      } else {
        statistics.recordMisses(1L);
      }
      if (result.written) {
        statistics.recordPuts(1L);
        statistics.recordPutTime(ticker.read() - start);
      }
    }
    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>();
    cache.asMap().compute(copyOf(key), (K k, 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);
      }
    });
    return result;
  }

  @Override
  public boolean remove(K key) {
    requireNotClosed();
    requireNonNull(key);
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    V value = removeNoCopyOrAwait(key, /* publishToWriter= */ true);
    dispatcher.awaitSynchronous();

    if (value != null) {
      statistics.recordRemovals(1L);
      if (statsEnabled) {
        statistics.recordRemoveTime(ticker.read() - start);
      }
      return true;
    }
    return false;
  }

  /**
   * 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];
    cache.asMap().compute(key, (K k, 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;
    });
    return removed[0];
  }

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

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;
    boolean[] found = { false };

    boolean[] removed = { false };
    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;
    });
    dispatcher.awaitSynchronous();
    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);
      }
    }
    return removed[0];
  }

  @Override
  public @Nullable V getAndRemove(K key) {
    requireNotClosed();
    requireNonNull(key);
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    V value = removeNoCopyOrAwait(key, /* publishToWriter= */ true);
    dispatcher.awaitSynchronous();

    V copy = (value == null) ? null : copyOf(value);
    if (statsEnabled) {
      long duration = ticker.read() - start;
      if (copy == null) {
        statistics.recordMisses(1L);
      } else {
        statistics.recordHits(1L);
        statistics.recordRemovals(1L);
        statistics.recordRemoveTime(duration);
      }
      statistics.recordGetTime(duration);
    }
    return copy;
  }

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

    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    boolean[] found = { false };
    boolean[] replaced = { false };
    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;
    });
    dispatcher.awaitSynchronous();

    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);
    }

    return replaced[0];
  }

  @Override
  public boolean replace(K key, V value) {
    requireNotClosed();
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    @Nullable V oldValue = replaceNoCopyOrAwait(key, value);
    dispatcher.awaitSynchronous();
    if (oldValue == null) {
      statistics.recordMisses(1L);
      if (statsEnabled) {
        statistics.recordGetTime(ticker.read() - start);
      }
      return false;
    }

    if (statsEnabled) {
      statistics.recordHits(1L);
      statistics.recordPuts(1L);
      long duration = ticker.read() - start;
      statistics.recordGetTime(duration);
      statistics.recordPutTime(duration);
    }
    return true;
  }

  @Override
  public @Nullable V getAndReplace(K key, V value) {
    requireNotClosed();
    boolean statsEnabled = statistics.isEnabled();
    long start = statsEnabled ? ticker.read() : 0L;

    V oldValue = replaceNoCopyOrAwait(key, value);
    dispatcher.awaitSynchronous();

    V copy = (oldValue == null) ? null : copyOf(oldValue);
    if (statsEnabled) {
      long duration = ticker.read() - start;
      if (copy == null) {
        statistics.recordMisses(1L);
      } else {
        statistics.recordHits(1L);
        statistics.recordPuts(1L);
        statistics.recordPutTime(duration);
      }
      statistics.recordGetTime(duration);
    }
    return copy;
  }

  /**
   * 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) {
    V copy = copyOf(value);
    @SuppressWarnings("unchecked")
    var replaced = (V[]) new Object[1];
    cache.asMap().computeIfPresent(key, (k, expirable) -> {
      if (!expirable.isEternal() && expirable.hasExpired(currentTimeMillis())) {
        dispatcher.publishExpired(this, key, expirable.get());
        statistics.recordEvictions(1L);
        return null;
      }

      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);
    });
    return replaced[0];
  }

  @Override
  public void removeAll(Set<? extends K> keys) {
    requireNotClosed();
    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 (RuntimeException 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++;
        }
      }
    } finally {
      dispatcher.awaitSynchronous();
    }

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

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

  @Override
  public void clear() {
    requireNotClosed();
    cache.invalidateAll();
  }

  @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... arguments) {
    requireNonNull(entryProcessor);
    requireNonNull(arguments);
    requireNotClosed();

    var result = new Object[1];
    var failure = new Throwable[1];
    BiFunction<K, Expirable<V>, 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 {
      cache.asMap().compute(copyOf(key), remappingFunction);
    } catch (Throwable t) {
      dispatcher.ignoreSynchronous();
      throw t;
    }
    dispatcher.awaitSynchronous();
    if (failure[0] != null) {
      throw processorFailure(failure[0]);
    }

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

  /** 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);
  }

  /**
   * 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... arguments) {
    requireNotClosed();
    requireNonNull(keys);
    requireNonNull(arguments);
    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;
    }
    synchronized (configuration) {
      if (!isClosed()) {
        closed = true;
        try {
          cacheManager.destroyCache(name, this);
        } catch (IllegalStateException ignored) { /* manager already closed */ }

        @Var var thrown = shutdownExecutor();
        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);
        }
        thrown = tryClose((AutoCloseable) () ->
            JmxRegistration.unregisterMxBean(this, MBeanType.STATISTICS), thrown);
        thrown = tryClose((AutoCloseable) () ->
            JmxRegistration.unregisterMxBean(this, MBeanType.CONFIGURATION), thrown);
        if (thrown != null) {
          logger.log(Level.WARNING, "Failure when closing cache resources", thrown);
        }
      }
    }
    cache.invalidateAll();
  }

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

    @Var Throwable thrown = null;
    try {
      CompletableFuture
          .allOf(inFlight.toArray(CompletableFuture[]::new))
          .get(10, TimeUnit.SECONDS);
    } catch (ExecutionException | TimeoutException e) {
      thrown = e;
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      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) {
        if (outer == null) {
          return t;
        } else 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) {
      requireNotClosed();
      configuration.addCacheEntryListenerConfiguration(cacheEntryListenerConfiguration);
      dispatcher.register(cacheEntryListenerConfiguration);
    }
  }

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

  @Override
  public Iterator<Cache.Entry<K, V>> iterator() {
    requireNotClosed();
    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 {
        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 {
        JmxRegistration.unregisterMxBean(this, MBeanType.STATISTICS);
      }
      statistics.enable(enabled);
      configuration.setStatisticsEnabled(enabled);
    }
  }

  /** Performs the action with the cache writer if write-through is enabled. */
  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 (RuntimeException e) {
      throw new CacheWriterException("Exception in CacheWriter", e);
    }
  }

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

  /**
   * 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
   */
  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 (RuntimeException 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. */
  protected final @Nullable Duration getAccessExpireTime() {
    try {
      return expiry.getExpiryForAccess();
    } catch (RuntimeException 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
   */
  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 (RuntimeException 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();
          setVariableExpiration(entry.getKey(), duration);
          setAccessExpireTime(entry.getValue(), duration, millis);
          cursor = entry;
        }
      }
      return (cursor != null);
    }

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

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

  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 / #5669

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