github-actions[bot] commented on code in PR #67417:
URL: https://github.com/apache/doris/pull/67417#discussion_r4056391214
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java:
##########
@@ -627,16 +675,31 @@ private List<Pair<String, String>>
getFilteredDatabaseNames() {
* and reloaded during the refresh process.
*/
public void resetToUninitialized(boolean invalidCache) {
- synchronized (this) {
- this.objectCreated = false;
- this.initialized = false;
- synchronized (this.confLock) {
- this.cachedConf = null;
+ MetaCache<ExternalDatabase<? extends ExternalTable>> cacheToInvalidate
= null;
+ try {
+ synchronized (this) {
+ metadataLoadEpoch.incrementAndGet();
+ this.objectCreated = false;
+ this.initialized = false;
+ synchronized (this.confLock) {
+ this.cachedConf = null;
+ }
+ this.lowerCaseToDatabaseName.clear();
+ cacheToInvalidate = metaCache;
+ if (cacheToInvalidate != null) {
+ cacheToInvalidate.invalidateNames();
+ }
+ onClose();
+ }
+ } finally {
+ if (cacheToInvalidate != null) {
+ cacheToInvalidate.invalidateObjects();
Review Comment:
[P1] Retire the object generation before reopening initialization
This `finally` runs only after the catalog monitor has been released. A
waiting caller can therefore enter `makeSureInitialized()`, initialize the
replacement connector, and return an existing database or publish a new one
into the still-current object cache before this older reset swaps it. The
subsequent removal callback then resets that database, leaving the caller with
a detached object and the current cache empty.
`ExternalDatabase.resetMetaToUninitialized()` has the analogous table race at
its line 140. Keeping callbacks outside the parent monitor is necessary, but
the short generation/cache swap must remain inside the initialization fence
(with the retired-cache callbacks drained afterward). Please add catalog- and
database-level races that let reinitialization publish between monitor release
and retirement.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java:
##########
@@ -64,122 +166,602 @@ public MetaCache(String name,
// So it only need to be expired after specified duration.
CacheFactory namesCacheFactory = new CacheFactory(
expireAfterAccessSec,
- refreshAfterWriteSec,
+ OptionalLong.empty(),
1, // names cache has one and only one entry
true,
null);
- CacheFactory objCacheFactory = new CacheFactory(
+ metaObjCacheFactory = new CacheFactory(
expireAfterAccessSec,
OptionalLong.empty(),
maxSize,
true,
null);
- namesCache = namesCacheFactory.buildCache(namesCacheLoader, executor);
+ namesCache = namesCacheFactory.buildCache();
// Use sync removal listener to prevent deadlock (removal listener
calls invalidateAll)
// NOTE: This cache should NOT use refreshAfterWrite, as it would
become synchronous
- metaObjCache =
objCacheFactory.buildCacheWithSyncRemovalListener(metaObjCacheLoader,
removalListener);
+ metaObjCache = buildMetaObjCache();
}
public List<String> listNames() {
- return
Objects.requireNonNull(namesCache.get("")).stream().map(Pair::value).collect(Collectors.toList());
+ return
getNames(false).stream().map(Pair::value).collect(Collectors.toList());
+ }
+
+ public List<String> refreshNames() {
+ throwIfInterrupted();
+ // Retire any active load so the forced refresh is not blocked behind
a stuck
+ // background refresh. Keep the retired physical owner accounted until
its loader
+ // exits, otherwise repeated forced refreshes can bypass
MAX_PHYSICAL_NAMES_LOADS.
+ // Only advance the generation when the active load is still running
(not done); a
+ // completed load has already been cleared by finishNamesLoad and
cannot publish stale results.
+ synchronized (namesMutationLock) {
+ if (activeNamesLoad != null && !activeNamesLoad.result.isDone()) {
+ NamesCacheValue current = namesCache.getIfPresent("");
+ long currentGeneration = namesGeneration.get();
+ long nextGeneration = advanceNamesGeneration();
+ if (current != null && current.complete && current.generation
== currentGeneration) {
+ namesCache.put("", current.withGeneration(nextGeneration));
+ }
+ }
+ activeNamesLoad = null;
+ }
+ throwIfInterrupted();
+ return
getNames(true).stream().map(Pair::value).collect(Collectors.toList());
+ }
+
+ private void throwIfInterrupted() {
+ if (Thread.currentThread().isInterrupted()) {
+ throw new CompletionException(new InterruptedException());
+ }
+ }
+
+ private List<Pair<String, String>> getNames(boolean forceRefresh) {
+ for (int attempt = 0; attempt < MAX_NAMES_LOAD_ATTEMPTS; attempt++) {
+ if (forceRefresh) {
+ throwIfInterrupted();
+ }
+ NamesCacheValue value = forceRefresh ? null :
namesCache.getIfPresent("");
+ List<Pair<String, String>> currentNames = null;
+ synchronized (namesMutationLock) {
+ if (value != null && value.complete && value.generation ==
namesGeneration.get()) {
+ currentNames = value.snapshot();
+ }
+ }
+ if (currentNames != null) {
+ scheduleNamesRefresh(value);
+ return currentNames;
+ }
+ value = loadNames(forceRefresh, true, null);
+ synchronized (namesMutationLock) {
+ if (value != null && value.complete && value.generation ==
namesGeneration.get()) {
+ return value.snapshot();
+ }
+ }
+ }
+ throw new IllegalStateException("Failed to load names for " + name
+ + " because metadata kept changing");
+ }
+
+ private NamesCacheValue loadNames(boolean forceRefresh, boolean
awaitActiveLoad, Long expectedGeneration) {
+ NamesLoad namesLoad = null;
+ boolean loadOwner = false;
+ long requestedGeneration;
+ synchronized (namesMutationLock) {
+ if (expectedGeneration != null && expectedGeneration !=
namesGeneration.get()) {
+ return null;
+ }
+ NamesCacheValue cached = namesCache.getIfPresent("");
+ if (!forceRefresh && cached != null && cached.complete
+ && cached.generation == namesGeneration.get()) {
+ return cached;
+ }
+ long loadGeneration = namesGeneration.get();
+ if (activeNamesLoad != null && activeNamesLoad.generation ==
loadGeneration) {
+ if (!awaitActiveLoad) {
+ return null;
+ }
+ namesLoad = activeNamesLoad;
+ }
+ requestedGeneration = loadGeneration;
+ }
+
+ if (namesLoad != null) {
+ return awaitNamesLoad(namesLoad);
+ }
+
+ // Lifecycle admission may acquire the catalog monitor. Keep it
outside the names
+ // mutation lock because catalog reset advances the names generation
under that monitor.
+ long loadEpoch = namesLoadEpochSupplier.getAsLong();
+ synchronized (namesMutationLock) {
+ if (requestedGeneration != namesGeneration.get()
+ || expectedGeneration != null && expectedGeneration !=
namesGeneration.get()) {
+ return null;
+ }
+ NamesCacheValue cached = namesCache.getIfPresent("");
+ if (!forceRefresh && cached != null && cached.complete
+ && cached.generation == requestedGeneration) {
+ return cached;
+ }
+ if (activeNamesLoad != null && activeNamesLoad.generation ==
requestedGeneration) {
+ if (!awaitActiveLoad) {
+ return null;
+ }
+ namesLoad = activeNamesLoad;
+ } else {
+ if (physicalNamesLoads.size() >= MAX_PHYSICAL_NAMES_LOADS) {
+ return null;
+ }
+ Map<String, Pair<String, String>> incompleteNames = cached !=
null && !cached.complete
+ ? Maps.newLinkedHashMap(cached.names) :
Maps.newLinkedHashMap();
+ namesLoad = new NamesLoad(requestedGeneration, loadEpoch,
incompleteNames);
+ activeNamesLoad = namesLoad;
+ physicalNamesLoads.add(namesLoad);
+ loadOwner = true;
+ }
+ }
+
+ if (!loadOwner) {
+ return awaitNamesLoad(namesLoad);
+ }
+
+ try {
+ List<Pair<String, String>> loadedNames =
Objects.requireNonNull(namesCacheLoader.load(""));
+ NamesCacheValue value = null;
+ synchronized (namesMutationLock) {
+ if (namesLoad.generation == namesGeneration.get()
+ && namesLoad.generation >= minimumLoadGeneration
+ && namesLoadEpochValidator.test(namesLoad.loadEpoch)) {
+ Map<String, Pair<String, String>> names =
toNamesMap(loadedNames);
+ names.putAll(namesLoad.incompleteNames);
+ value = new NamesCacheValue(namesGeneration.get(), names,
true);
+ namesCache.put("", value);
+ publishNames(value);
+ }
+ }
+ finishNamesLoad(namesLoad);
+ namesLoad.result.complete(value);
+ return value;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ CompletionException failure = new CompletionException(e);
+ finishNamesLoad(namesLoad);
+ namesLoad.result.completeExceptionally(failure);
+ throw failure;
+ } catch (RuntimeException e) {
+ finishNamesLoad(namesLoad);
+ if (!namesLoad.result.completeExceptionally(e)) {
+ return null;
+ }
+ throw e;
+ } catch (Error e) {
+ finishNamesLoad(namesLoad);
+ namesLoad.result.completeExceptionally(e);
+ throw e;
+ } catch (Exception e) {
+ CompletionException failure = new CompletionException(e);
+ finishNamesLoad(namesLoad);
+ if (!namesLoad.result.completeExceptionally(failure)) {
+ return null;
+ }
+ throw failure;
+ } finally {
+ finishNamesLoad(namesLoad);
+ }
+ }
+
+ private void finishNamesLoad(NamesLoad namesLoad) {
+ synchronized (namesMutationLock) {
+ if (activeNamesLoad == namesLoad) {
+ activeNamesLoad = null;
+ }
+ physicalNamesLoads.remove(namesLoad);
+ }
+ }
+
+ private NamesCacheValue awaitNamesLoad(NamesLoad namesLoad) {
+ try {
+ return namesLoad.result.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new CompletionException(cause);
+ }
+ }
+
+ private Map<String, Pair<String, String>> toNamesMap(List<Pair<String,
String>> names) {
+ Map<String, Pair<String, String>> namesMap = Maps.newLinkedHashMap();
+ for (Pair<String, String> pair : names) {
+ namesMap.putIfAbsent(pair.value(), pair);
+ }
+ return namesMap;
+ }
+
+ private void scheduleNamesRefresh(NamesCacheValue value) {
+ if (System.nanoTime() - value.writeNanos <
namesRefreshAfterWriteNanos) {
+ return;
+ }
+ scheduleNamesRefresh(value.generation);
+ }
+
+ private void scheduleNamesRefresh(long generation) {
+ NamesRefresh refresh;
+ synchronized (namesMutationLock) {
+ if (generation != namesGeneration.get()
+ || activeNamesRefreshes.containsKey(generation)) {
+ return;
+ }
+ if (activeNamesRefreshes.size() >= MAX_NAMES_REFRESH_FLIGHTS) {
+ pendingNamesRefreshGeneration = generation;
+ return;
+ }
+ refresh = new NamesRefresh(generation);
+ activeNamesRefreshes.put(generation, refresh);
+ if (pendingNamesRefreshGeneration != null &&
pendingNamesRefreshGeneration == generation) {
+ pendingNamesRefreshGeneration = null;
+ }
+ }
+ try {
+ namesRefreshExecutor.execute(() -> {
+ try {
+ loadNames(true, false, refresh.generation);
+ } catch (Exception e) {
+ LOG.warn("Failed to refresh names cache for {}", name, e);
+ } finally {
+ clearNamesRefresh(refresh);
+ }
+ });
+ } catch (RuntimeException e) {
+ clearNamesRefresh(refresh);
+ LOG.warn("Failed to schedule names cache refresh for {}", name, e);
+ }
+ }
+
+ private void clearNamesRefresh(NamesRefresh refresh) {
+ Long pendingGeneration = null;
+ synchronized (namesMutationLock) {
+ if (activeNamesRefreshes.get(refresh.generation) == refresh) {
+ activeNamesRefreshes.remove(refresh.generation);
+ if (pendingNamesRefreshGeneration != null) {
+ pendingGeneration = pendingNamesRefreshGeneration;
+ pendingNamesRefreshGeneration = null;
+ }
+ }
+ }
+ if (pendingGeneration != null) {
+ scheduleNamesRefresh(pendingGeneration);
+ }
}
public String getRemoteName(String localName) {
- return Objects.requireNonNull(namesCache.getIfPresent("")).stream()
+ NamesCacheValue value = namesCache.getIfPresent("");
+ synchronized (namesMutationLock) {
+ if (value != null && value.generation == namesGeneration.get()) {
+ Pair<String, String> pair = value.names.get(localName);
+ if (pair != null) {
+ return pair.key();
+ }
+ if (value.complete) {
+ return null;
+ }
+ }
+ }
+ return getNames(false).stream()
.filter(pair -> pair.value().equals(localName))
.map(Pair::key)
.findFirst()
.orElse(null);
}
+ private void publishNames(NamesCacheValue value) {
+ namesCacheUpdateAction.accept(value.snapshot());
+ }
+
public Optional<T> getMetaObj(String name, long id) {
- Optional<T> val = metaObjCache.getIfPresent(name);
- if (val == null || !val.isPresent()) {
- synchronized (metaObjCache) {
- val = metaObjCache.getIfPresent(name);
- if (val != null && val.isPresent()) {
- return val;
+ Optional<T> val = withMetaObjLifecycleReadLock(() ->
metaObjCache.getIfPresent(name));
+ if (val != null && val.isPresent()) {
+ return val;
+ }
+ return withMetaObjKeyLock(name, () -> {
Review Comment:
[P1] Let the replacement generation bypass a retired object loader
`withMetaObjKeyLock` owns this same lock across the whole physical load and
retry loop, but `invalidateObjects()` only swaps `metaObjCache`/increments
`metaObjGeneration`; it does not retire the lock owner. If a G database/table
loader is stuck in its names/connector path, reset can now finish, yet every
G+1 lookup or HMS event for that name still blocks behind G and cannot use the
replacement cache/connector. If G eventually returns, it retries from line 491
while still holding the gate, so repeated scheduled invalidations can keep all
current callers queued and issue unbounded reloads. This is distinct from the
same-generation serialization and old Caffeine-reset threads: key ownership
itself needs to be generation-scoped (with a physical-load bound), and a
regression should keep G blocked across `invalidateObjects()` while proving a
G+1 same-key lookup completes.
--
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]
