MetaCache.java
// 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.datasource.metacache;
import org.apache.doris.common.CacheFactory;
import org.apache.doris.common.Pair;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.CacheLoader;
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.github.benmanes.caffeine.cache.RemovalListener;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Collectors;
public class MetaCache<T> {
private static final Logger LOG = LogManager.getLogger(MetaCache.class);
private Cache<String, NamesCacheValue> namesCache;
private final CacheLoader<String, List<Pair<String, String>>> namesCacheLoader;
private final Consumer<List<Pair<String, String>>> namesCacheUpdateAction;
private final BiConsumer<String, String> nameUpdateAction;
private final Consumer<String> nameInvalidationAction;
private final ExecutorService namesRefreshExecutor;
private final long namesRefreshAfterWriteNanos;
private final AtomicBoolean namesRefreshRunning = new AtomicBoolean();
private final Object namesDedupLock = new Object();
// Order explicit mutations with validation and publication of a loaded names snapshot.
private final Object namesMutationLock = new Object();
private final AtomicLong namesGeneration = new AtomicLong();
private final AtomicLong activeLoadGeneration = new AtomicLong(-1);
private long minimumLoadGeneration;
//Pair<String, String> : <Remote name, Local name>
private Map<Long, String> idToName = Maps.newConcurrentMap();
private LoadingCache<String, Optional<T>> metaObjCache;
private String name;
public MetaCache(String name,
ExecutorService executor,
OptionalLong expireAfterAccessSec,
OptionalLong refreshAfterWriteSec,
long maxSize,
CacheLoader<String, List<Pair<String, String>>> namesCacheLoader,
CacheLoader<String, Optional<T>> metaObjCacheLoader,
RemovalListener<String, Optional<T>> removalListener) {
this(name, executor, expireAfterAccessSec, refreshAfterWriteSec, maxSize,
namesCacheLoader, ignored -> { }, (remoteName, localName) -> { }, ignored -> { },
metaObjCacheLoader, removalListener);
}
public MetaCache(String name,
ExecutorService executor,
OptionalLong expireAfterAccessSec,
OptionalLong refreshAfterWriteSec,
long maxSize,
CacheLoader<String, List<Pair<String, String>>> namesCacheLoader,
Consumer<List<Pair<String, String>>> namesCacheUpdateAction,
CacheLoader<String, Optional<T>> metaObjCacheLoader,
RemovalListener<String, Optional<T>> removalListener) {
this(name, executor, expireAfterAccessSec, refreshAfterWriteSec, maxSize,
namesCacheLoader, namesCacheUpdateAction, (remoteName, localName) -> { }, ignored -> { },
metaObjCacheLoader, removalListener);
}
public MetaCache(String name,
ExecutorService executor,
OptionalLong expireAfterAccessSec,
OptionalLong refreshAfterWriteSec,
long maxSize,
CacheLoader<String, List<Pair<String, String>>> namesCacheLoader,
Consumer<List<Pair<String, String>>> namesCacheUpdateAction,
BiConsumer<String, String> nameUpdateAction,
Consumer<String> nameInvalidationAction,
CacheLoader<String, Optional<T>> metaObjCacheLoader,
RemovalListener<String, Optional<T>> removalListener) {
this.name = name;
this.namesCacheLoader = namesCacheLoader;
this.namesCacheUpdateAction = namesCacheUpdateAction;
this.nameUpdateAction = nameUpdateAction;
this.nameInvalidationAction = nameInvalidationAction;
this.namesRefreshExecutor = executor;
this.namesRefreshAfterWriteNanos = refreshAfterWriteSec.isPresent()
? TimeUnit.SECONDS.toNanos(refreshAfterWriteSec.getAsLong()) : Long.MAX_VALUE;
// ATTN:
// The refreshAfterWriteSec is only used for metaObjCache, not for namesCache.
// Because namesCache need to be refreshed at interval so that user can get the latest meta list.
// But metaObjCache does not need to be refreshed at interval, because the object is actually not
// from remote datasource, it is just a local generated object to represent the meta info.
// So it only need to be expired after specified duration.
CacheFactory namesCacheFactory = new CacheFactory(
expireAfterAccessSec,
OptionalLong.empty(),
1, // names cache has one and only one entry
true,
null);
CacheFactory objCacheFactory = new CacheFactory(
expireAfterAccessSec,
OptionalLong.empty(),
maxSize,
true,
null);
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);
}
public List<String> listNames() {
return getNames().stream().map(Pair::value).collect(Collectors.toList());
}
private List<Pair<String, String>> getNames() {
int maxRetries = 3;
for (int attempt = 0; attempt < maxRetries + 1; attempt++) {
NamesCacheValue value = namesCache.getIfPresent("");
if (value == null || !value.complete) {
value = loadNames(false);
if (value == null) {
continue;
}
}
boolean current;
synchronized (namesMutationLock) {
current = value.generation == namesGeneration.get();
}
if (current) {
scheduleNamesRefresh(value);
return value.names;
}
if (attempt == maxRetries - 1) {
return value.names;
}
}
NamesCacheValue value = namesCache.getIfPresent("");
return value != null ? value.names : Lists.newArrayList();
}
private NamesCacheValue loadNames(boolean forceRefresh) {
long loadGeneration;
List<Pair<String, String>> incompleteNames;
synchronized (namesDedupLock) {
synchronized (namesMutationLock) {
NamesCacheValue cached = namesCache.getIfPresent("");
if (!forceRefresh && cached != null && cached.complete
&& cached.generation == namesGeneration.get()) {
return cached;
}
loadGeneration = namesGeneration.get();
if (!forceRefresh && activeLoadGeneration.get() == loadGeneration) {
return null;
}
incompleteNames = cached != null && !cached.complete
? Lists.newArrayList(cached.names) : Lists.newArrayList();
activeLoadGeneration.set(loadGeneration);
}
}
List<Pair<String, String>> loadedNames;
try {
loadedNames = Objects.requireNonNull(namesCacheLoader.load(""));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new CompletionException(e);
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
throw new CompletionException(e);
} finally {
synchronized (namesDedupLock) {
activeLoadGeneration.compareAndSet(loadGeneration, -1);
}
}
synchronized (namesDedupLock) {
if (loadGeneration != namesGeneration.get()) {
return null;
}
}
synchronized (namesMutationLock) {
if (loadGeneration != namesGeneration.get() || loadGeneration < minimumLoadGeneration) {
return null;
}
List<Pair<String, String>> names = Lists.newArrayList(loadedNames);
incompleteNames.forEach(pair -> NameMutation.update(pair.key(), pair.value()).apply(names));
NamesCacheValue value = new NamesCacheValue(namesGeneration.get(), names, true);
namesCache.put("", value);
publishNames(value);
return value;
}
}
private void scheduleNamesRefresh(NamesCacheValue value) {
if (System.nanoTime() - value.writeNanos < namesRefreshAfterWriteNanos
|| !namesRefreshRunning.compareAndSet(false, true)) {
return;
}
startNamesRefresh();
}
private void startNamesRefresh() {
try {
namesRefreshExecutor.execute(() -> {
try {
loadNames(true);
} catch (Exception e) {
LOG.warn("Failed to refresh names cache for {}", name, e);
} finally {
namesRefreshRunning.set(false);
}
});
} catch (RuntimeException e) {
namesRefreshRunning.set(false);
LOG.warn("Failed to schedule names cache refresh for {}", name, e);
}
}
public String getRemoteName(String localName) {
return getNames().stream()
.filter(pair -> pair.value().equals(localName))
.map(Pair::key)
.findFirst()
.orElse(null);
}
private void publishNames(NamesCacheValue value) {
namesCacheUpdateAction.accept(value.names);
}
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;
}
if (LOG.isDebugEnabled()) {
LOG.debug("trigger getMetaObj in metacache {}, obj name: {}, id: {}",
this.name, name, id, new Exception());
}
metaObjCache.invalidate(name);
val = metaObjCache.get(name);
idToName.put(id, name);
}
}
return val;
}
public Optional<T> tryGetMetaObj(String name) {
Optional<T> val = metaObjCache.getIfPresent(name);
if (val == null || !val.isPresent()) {
return Optional.empty();
}
return val;
}
public Optional<T> getMetaObjById(long id) {
String name = idToName.get(id);
return name == null ? Optional.empty() : getMetaObj(name, id);
}
public void updateCache(String remoteName, String localName, T obj, long id) {
metaObjCache.put(localName, Optional.of(obj));
synchronized (namesMutationLock) {
long generation = namesGeneration.incrementAndGet();
NameMutation mutation = NameMutation.update(remoteName, localName);
NamesCacheValue current = namesCache.getIfPresent("");
if (current != null) {
List<Pair<String, String>> names = Lists.newArrayList(current.names);
mutation.apply(names);
NamesCacheValue updated = new NamesCacheValue(generation, names, current.complete);
namesCache.put("", updated);
publishNames(updated);
} else {
namesCache.put("", new NamesCacheValue(
generation, Lists.newArrayList(Pair.of(remoteName, localName)), false));
nameUpdateAction.accept(remoteName, localName);
}
}
idToName.put(id, localName);
}
public void invalidate(String localName, long id) {
synchronized (namesMutationLock) {
long generation = namesGeneration.incrementAndGet();
NameMutation mutation = NameMutation.invalidate(localName);
NamesCacheValue current = namesCache.getIfPresent("");
if (current != null) {
List<Pair<String, String>> names = Lists.newArrayList(current.names);
mutation.apply(names);
NamesCacheValue updated = new NamesCacheValue(generation, names, current.complete);
namesCache.put("", updated);
publishNames(updated);
} else {
nameInvalidationAction.accept(localName);
}
}
if (LOG.isDebugEnabled()) {
LOG.debug("invalidate obj in metacache {}, obj name: {}, id: {}",
name, localName, id, new Exception());
}
metaObjCache.invalidate(localName);
idToName.remove(id);
}
public void invalidateAll() {
synchronized (namesMutationLock) {
minimumLoadGeneration = namesGeneration.incrementAndGet();
namesCache.invalidateAll();
namesCacheUpdateAction.accept(Lists.newArrayList());
}
if (LOG.isDebugEnabled()) {
LOG.debug("invalidate all in metacache {}", name, new Exception());
}
metaObjCache.invalidateAll();
idToName.clear();
}
@VisibleForTesting
public LoadingCache<String, Optional<T>> getMetaObjCache() {
return metaObjCache;
}
@VisibleForTesting
public void refreshNamesForTest() {
if (namesRefreshRunning.compareAndSet(false, true)) {
startNamesRefresh();
}
}
@VisibleForTesting
public void addObjForTest(long id, String name, T db) {
idToName.put(id, name);
metaObjCache.put(name, Optional.of(db));
}
/**
* Reset the names cache.
* Should only be used after creating new database/table
*/
public void resetNames() {
synchronized (namesMutationLock) {
minimumLoadGeneration = namesGeneration.incrementAndGet();
namesCache.invalidateAll();
namesCacheUpdateAction.accept(Lists.newArrayList());
}
}
private static class NamesCacheValue {
private final long generation;
private final List<Pair<String, String>> names;
private final boolean complete;
private final long writeNanos = System.nanoTime();
private NamesCacheValue(long generation, List<Pair<String, String>> names, boolean complete) {
this.generation = generation;
this.names = names;
this.complete = complete;
}
}
private static class NameMutation {
private final String remoteName;
private final String localName;
private NameMutation(String remoteName, String localName) {
this.remoteName = remoteName;
this.localName = localName;
}
private static NameMutation update(String remoteName, String localName) {
return new NameMutation(remoteName, localName);
}
private static NameMutation invalidate(String localName) {
return new NameMutation(null, localName);
}
private void apply(List<Pair<String, String>> names) {
names.removeIf(pair -> pair.value().equals(localName));
if (remoteName != null) {
names.add(Pair.of(remoteName, localName));
}
}
}
}