github-actions[bot] commented on code in PR #66633:
URL: https://github.com/apache/doris/pull/66633#discussion_r3754745909
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -135,189 +120,181 @@ public static <K, V> MetaCacheEntry<K, V>
withSyncRemovalListener(String name, F
public static <K, V> MetaCacheEntry<K, V> withSyncRemovalListener(String
name, Function<K, V> loader,
CacheSpec cacheSpec, ExecutorService refreshExecutor, int
stripeCount,
RemovalListener<K, V> removalListener) {
- return new MetaCacheEntry<>(
- name,
- loader,
- cacheSpec,
- refreshExecutor,
- false,
- false,
- stripeCount,
- Objects.requireNonNull(removalListener, "removalListener can
not be null"),
- true);
+ return new MetaCacheEntry<>(name, loader, cacheSpec, refreshExecutor,
false, false, stripeCount,
+ Objects.requireNonNull(removalListener, "removalListener can
not be null"));
}
private MetaCacheEntry(String name, @Nullable Function<K, V> loader,
CacheSpec cacheSpec,
- ExecutorService refreshExecutor, boolean autoRefresh, boolean
contextualOnly,
- int stripeCount, @Nullable RemovalListener<K, V> removalListener,
boolean syncRemovalListener) {
+ ExecutorService refreshExecutor, boolean autoRefresh, boolean
contextualOnly, int stripeCount,
+ @Nullable RemovalListener<K, V> removalListener) {
this.name = Objects.requireNonNull(name, "name can not be null");
- if (contextualOnly) {
- if (loader != null) {
- throw new IllegalArgumentException("contextual-only entry
loader must be null");
- }
- if (autoRefresh) {
- throw new IllegalArgumentException("contextual-only entry can
not enable auto refresh");
- }
- } else {
+ this.loader = loader;
+ this.cacheSpec = Objects.requireNonNull(cacheSpec, "cacheSpec can not
be null");
+ this.autoRefresh = autoRefresh;
+ Objects.requireNonNull(refreshExecutor, "refreshExecutor can not be
null");
+ if (contextualOnly && loader != null) {
+ throw new IllegalArgumentException("contextual-only entry loader
must be null");
+ }
+ if (contextualOnly && autoRefresh) {
+ throw new IllegalArgumentException("contextual-only entry can not
enable auto refresh");
+ }
+ if (!contextualOnly) {
Objects.requireNonNull(loader, "loader can not be null");
}
- if (syncRemovalListener && autoRefresh) {
+ if (removalListener != null && autoRefresh) {
throw new IllegalArgumentException("sync removal listener cache
can not enable refreshAfterWrite");
}
- if (removalListener != null && !syncRemovalListener) {
- throw new IllegalArgumentException("asynchronous removal listener
is not supported");
- }
- this.loader = loader;
- this.cacheSpec = Objects.requireNonNull(cacheSpec, "cacheSpec can not
be null");
- this.autoRefresh = autoRefresh;
if (stripeCount < 1) {
throw new IllegalArgumentException("stripeCount must be positive");
}
this.stripeCount = stripeCount;
- this.stripeStates = new AtomicReferenceArray<>(stripeCount);
+ stripeStates = new AtomicReferenceArray<>(stripeCount);
if (stripeCount == SINGLE_KEY_STRIPES) {
- // Names entries always use their sole stripe, so keep their
established allocation behavior.
stripeStates.set(0, new StripeState<>());
}
- Objects.requireNonNull(refreshExecutor, "refreshExecutor can not be
null");
- this.effectiveEnabled = CacheSpec.isCacheEnabled(
- this.cacheSpec.isEnable(), this.cacheSpec.getTtlSecond(),
this.cacheSpec.getCapacity());
- OptionalLong expireAfterAccessSec =
- effectiveEnabled ?
CacheSpec.toExpireAfterAccess(this.cacheSpec.getTtlSecond()) :
OptionalLong.empty();
- OptionalLong refreshAfterWriteSec =
- effectiveEnabled && autoRefresh
- ?
OptionalLong.of(Config.external_cache_refresh_time_minutes * 60)
- : OptionalLong.empty();
- long maxSize = effectiveEnabled ? this.cacheSpec.getCapacity() : 0L;
- CacheFactory cacheFactory = new CacheFactory(
- expireAfterAccessSec,
- refreshAfterWriteSec,
- maxSize,
- true,
- null);
- // Build through a dedicated loader so refresh results admitted under
an older generation are rejected.
- CacheLoader<K, V> cacheLoader = newCacheLoader();
- if (syncRemovalListener) {
- this.loadingData =
cacheFactory.buildCacheWithSyncRemovalListener(cacheLoader, removalListener);
- } else {
- this.loadingData = cacheFactory.buildCache(cacheLoader,
refreshExecutor);
+ effectiveEnabled = CacheSpec.isCacheEnabled(
+ cacheSpec.isEnable(), cacheSpec.getTtlSecond(),
cacheSpec.getCapacity());
+ MetaCacheDefinition.Builder<K, V> builder =
MetaCacheDefinition.builder(
+ name, cacheSpec, ignored -> ScopePath.catalog());
+ if (loader != null) {
+ builder.loader(key -> loadAndPause(key, loader));
+ }
+ if (removalListener != null) {
+ builder.removalListener((key, value, reason) ->
+ removalListener.onRemoval(key, value,
toCaffeineRemovalCause(reason)));
+ }
+ if (autoRefresh && Config.external_cache_refresh_time_minutes > 0) {
+ builder.refreshAfterWrite(
+
Duration.ofMinutes(Config.external_cache_refresh_time_minutes),
refreshExecutor);
}
- this.data = loadingData;
+ data = owner.create(builder.build());
}
public String name() {
return name;
}
public V get(K key) {
- return getWithManualLoad(key, this::applyDefaultLoader, null, null);
+ if (loader == null) {
+ throw new UnsupportedOperationException(String.format(
+ "Entry '%s' requires a contextual miss loader.", name));
+ }
+ return data.get(key);
}
public V get(K key, Function<K, V> missLoader) {
- Function<K, V> loadFunction = Objects.requireNonNull(missLoader,
"missLoader can not be null");
- return getWithManualLoad(key, loadFunction, null, null);
+ Function<K, V> nonNullLoader = Objects.requireNonNull(missLoader,
"missLoader can not be null");
+ return data.get(key, loadKey -> loadAndPause(loadKey, nonNullLoader));
}
- /**
- * Get the current value and run a short local action under the same
per-key publication protocol.
- *
- * <p>For an enabled entry, a hot-value action only runs while that value
remains current. A miss-load action uses
- * an exact-key fence, so an unrelated mutation in the same stripe may
reject object publication without
- * suppressing the action. For a disabled entry, the value is not cached,
but the exact-key action can still run.
- */
public V getAndRunIfCurrent(K key, BiConsumer<K, V> currentValueAction) {
return getAndRunIfCurrent(key, (ignored, value) -> true,
currentValueAction);
}
- /**
- * Get the current value and conditionally run a short local action under
the per-key publication protocol.
- *
- * <p>A hot value returns without entering the publication lock when
{@code actionRequired} is false. When the
- * action is required, both the value identity and the condition are
re-checked under the lock before running it.
- * Both callbacks must be short, deterministic, non-blocking local
operations. They must not perform remote I/O,
- * call back into this entry, mutate its object cache or active action
state, or acquire locks that reverse the
- * caller's lock order. The action is intended only for local auxiliary
state such as an ID-to-name index.
- */
public V getAndRunIfCurrent(K key, BiPredicate<K, V> actionRequired,
BiConsumer<K, V> currentValueAction) {
- BiPredicate<K, V> required = Objects.requireNonNull(
- actionRequired, "actionRequired can not be null");
- BiConsumer<K, V> action = Objects.requireNonNull(
- currentValueAction, "currentValueAction can not be null");
- return getWithManualLoad(key, this::applyDefaultLoader, required,
action);
+ BiPredicate<K, V> required = Objects.requireNonNull(actionRequired,
"actionRequired can not be null");
+ BiConsumer<K, V> action = Objects.requireNonNull(currentValueAction,
"currentValueAction can not be null");
+ V cached = data.getIfPresent(key);
+ if (cached != null && !required.test(key, cached)) {
+ return cached;
+ }
+ StripeState<K> stripe = stripeState(key);
+ ActionToken<K> token;
+ synchronized (stripe) {
+ token = beginAction(stripe, key);
+ }
+ try {
+ V value = cached == null ? get(key) : cached;
+ if (value == null) {
+ return null;
+ }
+ if (!required.test(key, value)) {
+ return value;
+ }
+ beforeCurrentValueActionForTest(key, value);
+ synchronized (stripe) {
+ boolean current = isCurrent(stripe, token)
+ && (!effectiveEnabled || data.getIfPresent(key) ==
value);
+ if (current && required.test(key, value) && isCurrent(stripe,
token)) {
+ try {
+ action.accept(key, value);
+ } catch (RuntimeException | Error throwable) {
+ data.invalidateKey(key);
+ throw throwable;
+ }
+ }
+ }
+ return value;
+ } finally {
+ synchronized (stripe) {
+ endAction(stripe, token);
+ }
+ }
}
public V getIfPresent(K key) {
- if (!effectiveEnabled) {
- return null;
- }
return data.getIfPresent(key);
}
@Nullable
public V findIfPresent(Predicate<K> keyPredicate) {
- if (!effectiveEnabled) {
- return null;
- }
- // Replay-only fallback needs a cache-only scan over current hot keys
without triggering load-through.
- for (java.util.Map.Entry<K, V> entry : data.asMap().entrySet()) {
- if (keyPredicate.test(entry.getKey())) {
- return entry.getValue();
+ Objects.requireNonNull(keyPredicate, "keyPredicate can not be null");
+ List<V> result = new ArrayList<>(1);
+ data.forEach((key, value) -> {
+ if (result.isEmpty() && keyPredicate.test(key)) {
+ result.add(value);
}
- }
- return null;
+ });
+ return result.isEmpty() ? null : result.get(0);
}
public void put(K key, V value) {
- // Public mutations advance the generation so loads admitted under an
older generation cannot overwrite them.
- Objects.requireNonNull(key, "key can not be null");
Objects.requireNonNull(value, "value can not be null");
- if (!effectiveEnabled) {
- return;
- }
- StripeState<K> state = stripeState(key);
- synchronized (state) {
- bumpGenerationLocked(state);
- bumpActiveActionGenerationLocked(state, key);
+ StripeState<K> stripe = stripeState(key);
+ synchronized (stripe) {
+ bumpAction(stripe, key);
beforePublicMutationWriteForTest(key);
data.put(key, value);
}
}
public V compute(K key, BiFunction<K, V, V> remappingFunction) {
- // Public compute must also advance the stripe generation before
mutating the cache state.
- Objects.requireNonNull(key, "key can not be null");
- Objects.requireNonNull(remappingFunction, "remappingFunction can not
be null");
- if (!effectiveEnabled) {
- return null;
- }
- StripeState<K> state = stripeState(key);
- synchronized (state) {
- bumpGenerationLocked(state);
- bumpActiveActionGenerationLocked(state, key);
- beforePublicMutationWriteForTest(key);
- return data.asMap().compute(key, remappingFunction);
- }
+ return computeAndRun(key, remappingFunction, () -> {
+ });
}
- /**
- * Compute the cached value and update related local state in one per-key
publication window.
- *
- * <p>The action always runs, including when the cache is disabled or the
remapping function keeps a cold key
- * absent. This allows callers to maintain lightweight auxiliary indexes
without warming the object entry.
- */
public V computeAndRun(K key, BiFunction<K, V, V> remappingFunction,
Runnable afterMutation) {
- Objects.requireNonNull(key, "key can not be null");
- Objects.requireNonNull(remappingFunction, "remappingFunction can not
be null");
+ BiFunction<K, V, V> remapper =
Objects.requireNonNull(remappingFunction, "remappingFunction can not be null");
Runnable action = Objects.requireNonNull(afterMutation, "afterMutation
can not be null");
- StripeState<K> state = stripeState(key);
- synchronized (state) {
- bumpGenerationLocked(state);
- bumpActiveActionGenerationLocked(state, key);
+ StripeState<K> stripe = stripeState(key);
+ synchronized (stripe) {
+ bumpAction(stripe, key);
beforePublicMutationWriteForTest(key);
- V value = effectiveEnabled ? data.asMap().compute(key,
remappingFunction) : null;
+ V updated = effectiveEnabled ? remapper.apply(key,
data.getIfPresent(key)) : null;
+ data.invalidateKey(key);
+ if (updated != null) {
+ data.put(key, updated);
+ }
action.run();
- return value;
+ return updated;
+ }
+ }
+
+ public V computeAfterValidation(K key, BiFunction<K, V, V>
remappingFunction, Runnable validationAction) {
+ BiFunction<K, V, V> remapper =
Objects.requireNonNull(remappingFunction, "remappingFunction can not be null");
+ Runnable validation = Objects.requireNonNull(validationAction,
"validationAction can not be null");
+ StripeState<K> stripe = stripeState(key);
+ synchronized (stripe) {
+ V updated = effectiveEnabled ? remapper.apply(key,
data.getIfPresent(key)) : null;
Review Comment:
[P1] Make name-snapshot remapping atomic with refresh
`updated` is derived from the current cached snapshot here, but the later
invalidate/put pair is protected only by the FE stripe. Shared-runtime refresh
does not take that stripe, so it can publish a newer snapshot N1 after this
read; the remapper then invalidates N1 and installs U derived from the older
N0, dropping any unrelated database/table names that arrived in the refresh
until another full reload. Both database- and table-name caches enable
auto-refresh, and their incremental add/drop paths use these remappers. Please
add an exact-key atomic remap/compare-and-retry primitive so a concurrent
refresh forces the function to re-evaluate the new value, with latch tests for
both name caches.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -135,189 +120,181 @@ public static <K, V> MetaCacheEntry<K, V>
withSyncRemovalListener(String name, F
public static <K, V> MetaCacheEntry<K, V> withSyncRemovalListener(String
name, Function<K, V> loader,
CacheSpec cacheSpec, ExecutorService refreshExecutor, int
stripeCount,
RemovalListener<K, V> removalListener) {
- return new MetaCacheEntry<>(
- name,
- loader,
- cacheSpec,
- refreshExecutor,
- false,
- false,
- stripeCount,
- Objects.requireNonNull(removalListener, "removalListener can
not be null"),
- true);
+ return new MetaCacheEntry<>(name, loader, cacheSpec, refreshExecutor,
false, false, stripeCount,
+ Objects.requireNonNull(removalListener, "removalListener can
not be null"));
}
private MetaCacheEntry(String name, @Nullable Function<K, V> loader,
CacheSpec cacheSpec,
- ExecutorService refreshExecutor, boolean autoRefresh, boolean
contextualOnly,
- int stripeCount, @Nullable RemovalListener<K, V> removalListener,
boolean syncRemovalListener) {
+ ExecutorService refreshExecutor, boolean autoRefresh, boolean
contextualOnly, int stripeCount,
+ @Nullable RemovalListener<K, V> removalListener) {
this.name = Objects.requireNonNull(name, "name can not be null");
- if (contextualOnly) {
- if (loader != null) {
- throw new IllegalArgumentException("contextual-only entry
loader must be null");
- }
- if (autoRefresh) {
- throw new IllegalArgumentException("contextual-only entry can
not enable auto refresh");
- }
- } else {
+ this.loader = loader;
+ this.cacheSpec = Objects.requireNonNull(cacheSpec, "cacheSpec can not
be null");
+ this.autoRefresh = autoRefresh;
+ Objects.requireNonNull(refreshExecutor, "refreshExecutor can not be
null");
+ if (contextualOnly && loader != null) {
+ throw new IllegalArgumentException("contextual-only entry loader
must be null");
+ }
+ if (contextualOnly && autoRefresh) {
+ throw new IllegalArgumentException("contextual-only entry can not
enable auto refresh");
+ }
+ if (!contextualOnly) {
Objects.requireNonNull(loader, "loader can not be null");
}
- if (syncRemovalListener && autoRefresh) {
+ if (removalListener != null && autoRefresh) {
throw new IllegalArgumentException("sync removal listener cache
can not enable refreshAfterWrite");
}
- if (removalListener != null && !syncRemovalListener) {
- throw new IllegalArgumentException("asynchronous removal listener
is not supported");
- }
- this.loader = loader;
- this.cacheSpec = Objects.requireNonNull(cacheSpec, "cacheSpec can not
be null");
- this.autoRefresh = autoRefresh;
if (stripeCount < 1) {
throw new IllegalArgumentException("stripeCount must be positive");
}
this.stripeCount = stripeCount;
- this.stripeStates = new AtomicReferenceArray<>(stripeCount);
+ stripeStates = new AtomicReferenceArray<>(stripeCount);
if (stripeCount == SINGLE_KEY_STRIPES) {
- // Names entries always use their sole stripe, so keep their
established allocation behavior.
stripeStates.set(0, new StripeState<>());
}
- Objects.requireNonNull(refreshExecutor, "refreshExecutor can not be
null");
- this.effectiveEnabled = CacheSpec.isCacheEnabled(
- this.cacheSpec.isEnable(), this.cacheSpec.getTtlSecond(),
this.cacheSpec.getCapacity());
- OptionalLong expireAfterAccessSec =
- effectiveEnabled ?
CacheSpec.toExpireAfterAccess(this.cacheSpec.getTtlSecond()) :
OptionalLong.empty();
- OptionalLong refreshAfterWriteSec =
- effectiveEnabled && autoRefresh
- ?
OptionalLong.of(Config.external_cache_refresh_time_minutes * 60)
- : OptionalLong.empty();
- long maxSize = effectiveEnabled ? this.cacheSpec.getCapacity() : 0L;
- CacheFactory cacheFactory = new CacheFactory(
- expireAfterAccessSec,
- refreshAfterWriteSec,
- maxSize,
- true,
- null);
- // Build through a dedicated loader so refresh results admitted under
an older generation are rejected.
- CacheLoader<K, V> cacheLoader = newCacheLoader();
- if (syncRemovalListener) {
- this.loadingData =
cacheFactory.buildCacheWithSyncRemovalListener(cacheLoader, removalListener);
- } else {
- this.loadingData = cacheFactory.buildCache(cacheLoader,
refreshExecutor);
+ effectiveEnabled = CacheSpec.isCacheEnabled(
+ cacheSpec.isEnable(), cacheSpec.getTtlSecond(),
cacheSpec.getCapacity());
+ MetaCacheDefinition.Builder<K, V> builder =
MetaCacheDefinition.builder(
+ name, cacheSpec, ignored -> ScopePath.catalog());
+ if (loader != null) {
+ builder.loader(key -> loadAndPause(key, loader));
+ }
+ if (removalListener != null) {
+ builder.removalListener((key, value, reason) ->
+ removalListener.onRemoval(key, value,
toCaffeineRemovalCause(reason)));
+ }
+ if (autoRefresh && Config.external_cache_refresh_time_minutes > 0) {
+ builder.refreshAfterWrite(
+
Duration.ofMinutes(Config.external_cache_refresh_time_minutes),
refreshExecutor);
}
- this.data = loadingData;
+ data = owner.create(builder.build());
}
public String name() {
return name;
}
public V get(K key) {
- return getWithManualLoad(key, this::applyDefaultLoader, null, null);
+ if (loader == null) {
+ throw new UnsupportedOperationException(String.format(
+ "Entry '%s' requires a contextual miss loader.", name));
+ }
+ return data.get(key);
}
public V get(K key, Function<K, V> missLoader) {
- Function<K, V> loadFunction = Objects.requireNonNull(missLoader,
"missLoader can not be null");
- return getWithManualLoad(key, loadFunction, null, null);
+ Function<K, V> nonNullLoader = Objects.requireNonNull(missLoader,
"missLoader can not be null");
+ return data.get(key, loadKey -> loadAndPause(loadKey, nonNullLoader));
}
- /**
- * Get the current value and run a short local action under the same
per-key publication protocol.
- *
- * <p>For an enabled entry, a hot-value action only runs while that value
remains current. A miss-load action uses
- * an exact-key fence, so an unrelated mutation in the same stripe may
reject object publication without
- * suppressing the action. For a disabled entry, the value is not cached,
but the exact-key action can still run.
- */
public V getAndRunIfCurrent(K key, BiConsumer<K, V> currentValueAction) {
return getAndRunIfCurrent(key, (ignored, value) -> true,
currentValueAction);
}
- /**
- * Get the current value and conditionally run a short local action under
the per-key publication protocol.
- *
- * <p>A hot value returns without entering the publication lock when
{@code actionRequired} is false. When the
- * action is required, both the value identity and the condition are
re-checked under the lock before running it.
- * Both callbacks must be short, deterministic, non-blocking local
operations. They must not perform remote I/O,
- * call back into this entry, mutate its object cache or active action
state, or acquire locks that reverse the
- * caller's lock order. The action is intended only for local auxiliary
state such as an ID-to-name index.
- */
public V getAndRunIfCurrent(K key, BiPredicate<K, V> actionRequired,
BiConsumer<K, V> currentValueAction) {
- BiPredicate<K, V> required = Objects.requireNonNull(
- actionRequired, "actionRequired can not be null");
- BiConsumer<K, V> action = Objects.requireNonNull(
- currentValueAction, "currentValueAction can not be null");
- return getWithManualLoad(key, this::applyDefaultLoader, required,
action);
+ BiPredicate<K, V> required = Objects.requireNonNull(actionRequired,
"actionRequired can not be null");
+ BiConsumer<K, V> action = Objects.requireNonNull(currentValueAction,
"currentValueAction can not be null");
+ V cached = data.getIfPresent(key);
+ if (cached != null && !required.test(key, cached)) {
+ return cached;
+ }
+ StripeState<K> stripe = stripeState(key);
+ ActionToken<K> token;
+ synchronized (stripe) {
+ token = beginAction(stripe, key);
+ }
+ try {
+ V value = cached == null ? get(key) : cached;
Review Comment:
[P1] Keep a cold object hidden until its ID mapping is published
On a miss, `get(key)` installs the object in the shared `MetaCache` before
this method reaches the pre-action hook and reacquires the FE stripe. If
another same-stripe mutation holds that monitor, cache-only/name paths can
already observe the object while `ExternalCatalog.getDbNullable(id)` or
`ExternalDatabase.getTableNullable(id)` still reads an empty `IdNameIndex` and
returns `null`. The base path also put immediately before the index action, so
unlocked readers had a very small interval, but it acquired the stripe (and
passed the test hook) before publishing either side; this refactor turns that
into an interval that can block behind stripe contention. Please keep the value
unpublished until the validated auxiliary action succeeds, and cover
database/table by-ID lookups with a latch-based test.
##########
fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java:
##########
@@ -0,0 +1,1036 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 org.apache.doris.connector.cache;
+
+import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.CacheAddress;
+import
org.apache.doris.connector.cache.ScopedMetaCacheRegistry.PublicationState;
+import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.ScopeLease;
+import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.ScopeSnapshot;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.RemovalCause;
+import com.github.benmanes.caffeine.cache.RemovalListener;
+import com.github.benmanes.caffeine.cache.Ticker;
+
+import java.math.BigInteger;
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Objects;
+import java.util.OptionalLong;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+/**
+ * One physical Caffeine cache participating in a {@link
ScopedMetaCacheRegistry}.
+ *
+ * <p>Every value is wrapped with both its hierarchical scope-state identities
and an exact-key state. Hierarchical
+ * invalidation can therefore detach a whole catalog/database/table/partition
subtree, while exact-key invalidation
+ * fences only one physical cache key. Removal listeners conditionally remove
the exact wrapper from its original
+ * scope bucket and key node, so delayed callbacks cannot delete a replacement.
+ */
+public final class ScopedMetaCache<K, V> implements AutoCloseable {
+ private static final System.Logger LOG =
System.getLogger(ScopedMetaCache.class.getName());
+ private static final Runnable NO_OP = () -> {
+ };
+
+ private final ScopedMetaCacheRegistry registry;
+ private final String name;
+ private final boolean effectiveEnabled;
+ private final Cache<K, VersionedValue<K, V>> data;
+ private final ConcurrentMap<K, KeyNode<K, V>> keyNodes = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<LoadAddress<K>, CompletableFuture<V>>
inFlightLoads = new ConcurrentHashMap<>();
+ private final ConcurrentMap<K, VersionedValue<K, V>> refreshing = new
ConcurrentHashMap<>();
+ private final StripedPhaseGate bulkInvalidationGate = new
StripedPhaseGate();
+ private final Map<K, BigInteger> exactInvalidations = new HashMap<>();
+ private final NavigableMap<BigInteger, Integer> activeBulkStarts = new
TreeMap<>();
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final LongAdder requestCount = new LongAdder();
+ private final LongAdder hitCount = new LongAdder();
+ private final LongAdder missCount = new LongAdder();
+ private final LongAdder loadSuccessCount = new LongAdder();
+ private final LongAdder loadFailureCount = new LongAdder();
+ private final LongAdder totalLoadTimeNanos = new LongAdder();
+ private final LongAdder evictionCount = new LongAdder();
+ private final LongAdder invalidateCount = new LongAdder();
+ private final AtomicReference<Long> lastLoadSuccessTimeMs = new
AtomicReference<>(-1L);
+ private final AtomicReference<Long> lastLoadFailureTimeMs = new
AtomicReference<>(-1L);
+ private final AtomicReference<String> lastError = new
AtomicReference<>("");
+ private final RemovalListener<K, V> beforeRemoval;
+ private final Ticker ticker;
+ private final long refreshAfterWriteNanos;
+ private final Executor refreshExecutor;
+ private final Runnable afterLoadElection;
+ private final Runnable afterBulkStage;
+ private final Runnable afterRefreshRegistration;
+ private final ThreadLocal<RemovalDeferral<K, V>> removalDeferrals =
+ ThreadLocal.withInitial(RemovalDeferral::new);
+ private BigInteger exactInvalidationSequence = BigInteger.ZERO;
+
+ ScopedMetaCache(
+ ScopedMetaCacheRegistry registry,
+ String name,
+ CacheSpec cacheSpec,
+ Ticker ticker,
+ RemovalListener<K, V> beforeRemoval,
+ Duration refreshAfterWrite,
+ Executor refreshExecutor,
+ Runnable afterLoadElection,
+ Runnable afterBulkStage,
+ Runnable afterRefreshRegistration) {
+ this.registry = Objects.requireNonNull(registry, "registry can not be
null");
+ this.name = Objects.requireNonNull(name, "name can not be null");
+ Objects.requireNonNull(cacheSpec, "cacheSpec can not be null");
+ this.beforeRemoval = beforeRemoval;
+ this.ticker = ticker == null ? Ticker.systemTicker() : ticker;
+ this.refreshAfterWriteNanos = refreshAfterWrite == null ? 0L :
refreshAfterWrite.toNanos();
+ this.refreshExecutor = refreshExecutor;
+ this.afterLoadElection =
+ Objects.requireNonNull(afterLoadElection, "afterLoadElection
can not be null");
+ this.afterBulkStage = Objects.requireNonNull(afterBulkStage,
"afterBulkStage can not be null");
+ this.afterRefreshRegistration = Objects.requireNonNull(
+ afterRefreshRegistration, "afterRefreshRegistration can not be
null");
+ this.effectiveEnabled = CacheSpec.isCacheEnabled(
+ cacheSpec.isEnable(), cacheSpec.getTtlSecond(),
cacheSpec.getCapacity());
+
+ Caffeine<Object, Object> builder = Caffeine.newBuilder()
+ .maximumSize(effectiveEnabled ? cacheSpec.getCapacity() : 0L)
+ .executor(Runnable::run)
+ .removalListener(this::onRemoval);
+ OptionalLong expiry = effectiveEnabled
+ ? CacheSpec.toExpireAfterAccess(cacheSpec.getTtlSecond())
+ : OptionalLong.empty();
+ if (expiry.isPresent()) {
+ builder.expireAfterAccess(Duration.ofSeconds(expiry.getAsLong()));
+ }
+ if (ticker != null) {
+ builder.ticker(this.ticker);
+ }
+ this.data = builder.build();
+ }
+
+ public String name() {
+ return name;
+ }
+
+ public V get(K key, ScopePath path, Function<K, V> loader) {
+ Objects.requireNonNull(key, "key can not be null");
+ Objects.requireNonNull(path, "path can not be null");
+ Function<K, V> loadFunction = Objects.requireNonNull(loader, "loader
can not be null");
+ checkOpen();
+ if (!effectiveEnabled) {
+ recordAccess(false);
+ return loadAndRecord(key, loadFunction);
+ }
+
+ VersionedValue<K, V> presentVersioned = currentVersionedValue(key,
path);
+ if (presentVersioned != null) {
+ recordAccess(true);
+ scheduleRefresh(key, path, loader, presentVersioned);
+ return presentVersioned.value;
+ }
+ recordAccess(false);
+ try (PublicationLease<K, V> lease = acquirePublicationLease(key, path,
true)) {
+ LoadAddress<K> loadAddress = new LoadAddress<>(key, path, lease);
+ CompletableFuture<V> ownLoad = new CompletableFuture<>();
+ CompletableFuture<V> existingLoad =
inFlightLoads.putIfAbsent(loadAddress, ownLoad);
+ if (existingLoad != null) {
+ return awaitLoad(existingLoad);
+ }
+ try {
+ afterLoadElection.run();
+ synchronized (lease.keyNode) {
+ VersionedValue<K, V> present = currentVersionedValue(key,
path);
+ if (present != null) {
+ ownLoad.complete(present.value);
+ return present.value;
+ }
+ }
+ V loaded = loadAndRecord(key, loadFunction);
+ if (loaded != null) {
+ synchronized (lease.keyNode) {
+ publishCommitted(lease, key, loaded);
+ }
+ }
+ ownLoad.complete(loaded);
+ return loaded;
+ } catch (RuntimeException | Error throwable) {
+ ownLoad.completeExceptionally(throwable);
+ throw throwable;
+ } finally {
+ inFlightLoads.remove(loadAddress, ownLoad);
+ }
+ }
+ }
+
+ public V getIfPresent(K key, ScopePath path) {
+ Objects.requireNonNull(key, "key can not be null");
+ Objects.requireNonNull(path, "path can not be null");
+ checkOpen();
+ if (!effectiveEnabled) {
+ recordAccess(false);
+ return null;
+ }
+ VersionedValue<K, V> versioned = currentVersionedValue(key, path);
+ recordAccess(versioned != null);
+ return versioned == null ? null : versioned.value;
+ }
+
+ private VersionedValue<K, V> currentVersionedValue(K key, ScopePath path) {
+ VersionedValue<K, V> versioned = data.getIfPresent(key);
+ if (versioned == null) {
+ return null;
+ }
+ if (!versioned.scopeSnapshot.path().equals(path)) {
+ return null;
+ }
+ if (!versioned.isCurrent(registry, keyNodes)) {
+ data.asMap().remove(key, versioned);
+ return null;
+ }
+ return versioned;
+ }
+
+ public void put(K key, ScopePath path, V value) {
+ Objects.requireNonNull(key, "key can not be null");
+ Objects.requireNonNull(path, "path can not be null");
+ Objects.requireNonNull(value, "value can not be null");
+ checkOpen();
+ if (!effectiveEnabled) {
+ return;
+ }
+ try (PublicationLease<K, V> lease = acquirePublicationLease(key, path,
false)) {
+ synchronized (lease.keyNode) {
+ lease.keyNode.loadPublicationState.set(new Object());
+ publishCommitted(lease, key, value);
+ }
+ }
+ }
+
+ public void invalidateKey(K key) {
+ invalidateKey(key, NO_OP, NO_OP);
+ }
+
+ void invalidateKey(K key, Runnable afterStateReplacement) {
+ invalidateKey(key, NO_OP, afterStateReplacement);
+ }
+
+ void invalidateKey(
+ K key, Runnable beforeInvalidationLock, Runnable
afterStateReplacement) {
+ Objects.requireNonNull(key, "key can not be null");
+ Objects.requireNonNull(beforeInvalidationLock, "beforeInvalidationLock
can not be null");
+ Objects.requireNonNull(afterStateReplacement, "afterStateReplacement
can not be null");
+ checkOpen();
+ beforeInvalidationLock.run();
+ InvalidatedKey<K, V> invalidated = bulkInvalidationGate.write(() -> {
+ if (closed.get()) {
+ return null;
+ }
+ exactInvalidationSequence =
exactInvalidationSequence.add(BigInteger.ONE);
+ if (!activeBulkStarts.isEmpty()) {
+ exactInvalidations.put(key, exactInvalidationSequence);
+ }
+ KeyNode<K, V> node = keyNodes.get(key);
+ KeyState invalidatedState = null;
+ if (node != null) {
+ invalidatedState = replaceKeyState(node);
+ }
+ return new InvalidatedKey<>(node, invalidatedState);
+ });
+ if (invalidated == null || invalidated.node == null) {
+ return;
+ }
+ afterStateReplacement.run();
+ VersionedValue<K, V> registered = invalidated.node.registration.get();
+ if (registered != null && registered.keyState == invalidated.keyState)
{
+ data.asMap().remove(key, registered);
+ invalidated.node.registration.compareAndSet(registered, null);
+ }
+ tryPruneKey(key, invalidated.node);
+ }
+
+ public BulkLoadHandle beginBulkLoad(ScopePath parentScope) {
+ Objects.requireNonNull(parentScope, "parentScope can not be null");
+ checkOpen();
+ if (!effectiveEnabled) {
+ return BulkLoadHandle.disabled(this, parentScope);
+ }
+ ScopeLease scopeLease = registry.acquire(parentScope);
+ BigInteger exactSequence = bulkInvalidationGate.write(() -> {
+ if (closed.get()) {
+ scopeLease.close();
+ throw new IllegalStateException("Scoped meta cache '" + name +
"' is closed");
+ }
+ BigInteger sequence = exactInvalidationSequence;
+ activeBulkStarts.merge(sequence, 1, Integer::sum);
+ return sequence;
+ });
+ return new BulkLoadHandle(
+ this,
+ parentScope,
+ scopeLease,
+ scopeLease.publicationState(),
+ exactSequence);
+ }
+
+ public boolean publish(
+ BulkLoadHandle handle, K key, ScopePath actualScope, V value) {
+ Objects.requireNonNull(handle, "handle can not be null");
+ Objects.requireNonNull(key, "key can not be null");
+ Objects.requireNonNull(actualScope, "actualScope can not be null");
+ Objects.requireNonNull(value, "value can not be null");
+ checkOpen();
+ handle.checkOwner(this);
+ if (!handle.parentScope.contains(actualScope)) {
+ throw new IllegalArgumentException(
+ "Actual scope " + actualScope + " is outside bulk-load
parent " + handle.parentScope);
+ }
+ if (!effectiveEnabled) {
+ return false;
+ }
+ try (PublicationLease<K, V> lease = acquirePublicationLease(key,
actualScope, false)) {
+ synchronized (lease.keyNode) {
+ VersionedValue<K, V> staged = newVersionedValue(lease, key,
value);
+ afterBulkStage.run();
+ if (handle.tryCommit(key, lease, staged)) {
+ return true;
+ }
+ return false;
+ }
+ }
+ }
+
+ public CacheMetrics metrics() {
+ return bulkInvalidationGate.read(() -> new CacheMetrics(
+ data.estimatedSize(),
+ keyNodes.size(),
+ inFlightLoads.size(),
+
activeBulkStarts.values().stream().mapToInt(Integer::intValue).sum(),
+ exactInvalidations.size(),
+ effectiveEnabled,
+ requestCount.sum(),
+ hitCount.sum(),
+ missCount.sum(),
+ loadSuccessCount.sum(),
+ loadFailureCount.sum(),
+ totalLoadTimeNanos.sum(),
+ evictionCount.sum(),
+ invalidateCount.sum(),
+ lastLoadSuccessTimeMs.get(),
+ lastLoadFailureTimeMs.get(),
+ lastError.get()));
+ }
+
+ int refreshingCountForTest() {
+ return refreshing.size();
+ }
+
+ public void forEach(BiConsumer<K, V> consumer) {
+ Objects.requireNonNull(consumer, "consumer can not be null");
+ data.asMap().forEach((key, versioned) -> {
+ if (versioned.isCurrent(registry, keyNodes)) {
+ consumer.accept(key, versioned.value);
+ }
+ });
+ }
+
+ private V loadAndRecord(K key, Function<K, V> loader) {
+ long startNanos = System.nanoTime();
+ try {
+ V loaded = loader.apply(key);
+ loadSuccessCount.increment();
+ lastLoadSuccessTimeMs.set(System.currentTimeMillis());
+ return loaded;
+ } catch (RuntimeException | Error throwable) {
+ loadFailureCount.increment();
+ lastLoadFailureTimeMs.set(System.currentTimeMillis());
+ lastError.set(throwable.toString());
+ throw throwable;
+ } finally {
+ totalLoadTimeNanos.add(System.nanoTime() - startNanos);
+ }
+ }
+
+ private void recordAccess(boolean hit) {
+ requestCount.increment();
+ if (hit) {
+ hitCount.increment();
+ } else {
+ missCount.increment();
+ }
+ }
+
+ public void cleanUp() {
+ data.cleanUp();
+ }
+
+ @Override
+ public void close() {
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+ registry.removeCache(this);
+ closePhysicalState();
+ }
+
+ void closeFromRegistry() {
+ if (closed.compareAndSet(false, true)) {
+ closePhysicalState();
+ }
+ }
+
+ void removeExpectedRaw(Object rawKey, Object expectedValue) {
+ @SuppressWarnings("unchecked")
+ K key = (K) rawKey;
+ @SuppressWarnings("unchecked")
+ VersionedValue<K, V> versionedValue = (VersionedValue<K, V>)
expectedValue;
+ data.asMap().remove(key, versionedValue);
+ }
+
+ private PublicationLease<K, V> acquirePublicationLease(
+ K key, ScopePath path, boolean fenceAgainstDirectPublication) {
+ while (true) {
+ ScopeLease scopeLease = registry.acquire(path);
+ KeyNode<K, V> keyNode = keyNodes.computeIfAbsent(key, ignored ->
new KeyNode<>());
+ keyNode.activeLoads.incrementAndGet();
+ KeyState keyState = keyNode.current.get();
+ Object loadPublicationState =
+ fenceAgainstDirectPublication ?
keyNode.loadPublicationState.get() : null;
+ if (keyNodes.get(key) == keyNode && scopeLease.isCurrent()) {
+ return new PublicationLease<>(
+ this, key, scopeLease, keyNode, keyState,
loadPublicationState);
+ }
+ releaseKey(key, keyNode);
+ scopeLease.close();
+ }
+ }
+
+ private VersionedValue<K, V> publish(PublicationLease<K, V> lease, K key,
V value) {
+ if (!lease.isCurrent()) {
+ return null;
+ }
+ VersionedValue<K, V> versioned = newVersionedValue(lease, key, value);
+ install(versioned, lease);
+ if (!lease.isCurrent()) {
+ data.asMap().remove(key, versioned);
+ return null;
+ }
+ return versioned;
+ }
+
+ private VersionedValue<K, V> publishCommitted(PublicationLease<K, V>
lease, K key, V value) {
+ return publish(lease, key, value);
+ }
+
+ private VersionedValue<K, V> newVersionedValue(
+ PublicationLease<K, V> lease, K key, V value) {
+ CacheAddress address = new CacheAddress(this, key);
+ return new VersionedValue<>(
+ key, value, address, lease.scopeLease.snapshot(),
lease.keyNode, lease.keyState, ticker.read());
+ }
+
+ private void scheduleRefresh(K key, ScopePath path, Function<K, V> loader,
VersionedValue<K, V> current) {
+ if (refreshAfterWriteNanos == 0L || ticker.read() -
current.writeTimeNanos < refreshAfterWriteNanos
+ || refreshing.putIfAbsent(key, current) != null) {
+ return;
+ }
+ PublicationLease<K, V> lease;
+ try {
+ afterRefreshRegistration.run();
+ lease = acquirePublicationLease(key, path, true);
+ } catch (RuntimeException | Error throwable) {
+ refreshing.remove(key, current);
+ throw throwable;
+ }
+ if (data.getIfPresent(key) != current || !current.isCurrent(registry,
keyNodes)) {
+ lease.close();
+ refreshing.remove(key, current);
+ return;
+ }
+ try {
+ refreshExecutor.execute(() -> {
+ try (PublicationLease<K, V> ignored = lease) {
+ if (closed.get()) {
+ return;
+ }
+ V refreshed = loadAndRecord(key, loader);
+ if (refreshed != null) {
+ synchronized (lease.keyNode) {
+ publishCommitted(lease, key, refreshed);
Review Comment:
[P1] Do not let an old refresh replace a newer miss result
This commit validates only the refresh lease's key/scope states, not that
the wrapper which triggered the refresh is still current. A concrete production
schedule is: refresh R starts from V1 and captures the old remote result;
capacity/expiry evicts V1 without changing `KeyState` or
`loadPublicationState`; a normal miss then loads and publishes V2 using those
same identities; finally R reaches this line and overwrites V2 with stale V1.
Bounded multi-key FE schema caches enable auto-refresh, so eviction makes this
reachable. Please require the exact `current` wrapper/registration to remain
current at refresh commit (or use expected-value conditional replacement), and
add a latch test for eviction plus a newer miss.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
