diff --git a/oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/persistentcache/PersistentDiskCache.java b/oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/persistentcache/PersistentDiskCache.java index 25687239380..239cc9148f9 100644 --- a/oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/persistentcache/PersistentDiskCache.java +++ b/oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/persistentcache/PersistentDiskCache.java @@ -111,7 +111,8 @@ public PersistentDiskCache(File directory, int cacheMaxSizeMB, DiskCacheIOMonito () -> maxCacheSizeBytes, () -> Long.valueOf(directory.listFiles().length), () -> FileUtils.sizeOfDirectory(directory), - () -> evictionCount.get()); + () -> evictionCount.get(), + () -> discardCount.get()); } @Override diff --git a/oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/persistentcache/PersistentRedisCache.java b/oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/persistentcache/PersistentRedisCache.java index 204484e23c1..754e657fc56 100644 --- a/oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/persistentcache/PersistentRedisCache.java +++ b/oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/persistentcache/PersistentRedisCache.java @@ -73,7 +73,7 @@ public PersistentRedisCache(String redisHost, int redisPort, int redisExpireSeco this.redisPool = new JedisPool(jedisPoolConfig, redisHost, redisPort, redisConnectionTimeout, redisSocketTimeout, null, redisDBIndex, null); this.segmentCacheStats = new SegmentCacheStats(NAME, this::getRedisMaxMemory, this::getCacheElementCount, - this::getCurrentWeight, this::getNumberOfEvictedKeys); + this::getCurrentWeight, this::getNumberOfEvictedKeys, () -> discardCount.get()); } private long getCacheElementCount() { diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/AbstractPersistentCache.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/AbstractPersistentCache.java index e60ae43867c..f1641ffb531 100644 --- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/AbstractPersistentCache.java +++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/AbstractPersistentCache.java @@ -22,6 +22,7 @@ import org.apache.jackrabbit.oak.cache.AbstractCacheStats; import org.apache.jackrabbit.oak.commons.Buffer; +import org.apache.jackrabbit.oak.commons.log.LogSilencer; import org.apache.jackrabbit.oak.commons.time.Stopwatch; import org.apache.jackrabbit.oak.segment.spi.RepositoryNotReachableException; import org.jetbrains.annotations.NotNull; @@ -31,27 +32,53 @@ import java.io.Closeable; import java.util.Set; import java.util.UUID; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; public abstract class AbstractPersistentCache implements PersistentCache, Closeable { private static final Logger logger = LoggerFactory.getLogger(AbstractPersistentCache.class); public static final int THREADS = Integer.getInteger("oak.segment.cache.threads", 10); + public static final int WRITE_QUEUE_SIZE = Integer.getInteger("oak.segment.cache.writeQueueSize", 100); protected ExecutorService executor; protected AtomicLong cacheSize = new AtomicLong(0); protected PersistentCache nextCache; protected final Set writesPending; + protected AtomicLong discardCount = new AtomicLong(); protected SegmentCacheStats segmentCacheStats; + private final LogSilencer writeQueueFullSilencer = new LogSilencer(); + public AbstractPersistentCache() { - executor = Executors.newFixedThreadPool(THREADS); + AtomicInteger threadCount = new AtomicInteger(0); + ThreadFactory threadFactory = r -> { + Thread t = new Thread(r, "segment-cache-writer-" + threadCount.incrementAndGet()); + t.setDaemon(true); + return t; + }; + BlockingQueue writeQueue = new LinkedBlockingQueue<>(WRITE_QUEUE_SIZE); + executor = new ThreadPoolExecutor( + THREADS, THREADS, + 0L, TimeUnit.MILLISECONDS, + writeQueue, + threadFactory, + (r, e) -> { + discardCount.incrementAndGet(); + if (!writeQueueFullSilencer.silence("writeQueueFull")) { + logger.warn("Segment write task discarded: write queue full (capacity={}, totalDiscarded={}){}", + WRITE_QUEUE_SIZE, discardCount.get(), LogSilencer.SILENCING_POSTFIX); + } + }); writesPending = ConcurrentHashMap.newKeySet(); } @@ -148,4 +175,8 @@ public void close() { public int getWritesPending() { return writesPending.size(); } + + public long getWriteDiscardCount() { + return discardCount.get(); + } } diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/SegmentCacheStats.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/SegmentCacheStats.java index 541e31aa82c..a29148852ed 100644 --- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/SegmentCacheStats.java +++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/SegmentCacheStats.java @@ -54,16 +54,21 @@ public class SegmentCacheStats extends AbstractCacheStats { @NotNull final AtomicLong missCount = new AtomicLong(); + @NotNull + private final Supplier writeDiscardCountSupplier; + public SegmentCacheStats(@NotNull String name, @NotNull Supplier maximumWeight, @NotNull Supplier elementCount, @NotNull Supplier currentWeight, - @NotNull Supplier evictionCount) { + @NotNull Supplier evictionCount, + @NotNull Supplier writeDiscardCount) { super(name); this.maximumWeight = maximumWeight; this.elementCount = requireNonNull(elementCount); this.currentWeight = requireNonNull(currentWeight); this.evictionCount = evictionCount; + this.writeDiscardCountSupplier = requireNonNull(writeDiscardCount); } @Override @@ -78,6 +83,10 @@ protected CacheStats getCurrentStats() { ); } + public long getWriteDiscardCount() { + return writeDiscardCountSupplier.get(); + } + @Override public long getElementCount() { return elementCount.get(); diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/package-info.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/package-info.java index 0ad0171f1b9..96eb50916cf 100644 --- a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/package-info.java +++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/package-info.java @@ -15,7 +15,7 @@ * limitations under the License. */ @Internal(since = "1.0.0") -@Version("6.1.0") +@Version("7.0.0") package org.apache.jackrabbit.oak.segment.spi.persistence.persistentcache; import org.apache.jackrabbit.oak.commons.annotations.Internal; diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/MemoryPersistentCache.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/MemoryPersistentCache.java index 896a98c8e5b..db0135f476b 100644 --- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/MemoryPersistentCache.java +++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/MemoryPersistentCache.java @@ -38,7 +38,8 @@ public MemoryPersistentCache(boolean throwException) { () -> null, () -> null, () -> null, - () -> null); + () -> null, + () -> 0L); } @Override diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/PersistentCacheStatsTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/PersistentCacheStatsTest.java index df2013c26f1..7aa5bcedd9d 100644 --- a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/PersistentCacheStatsTest.java +++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/PersistentCacheStatsTest.java @@ -170,7 +170,7 @@ class PersistentCacheImpl extends AbstractPersistentCache { HashMap segments = new HashMap<>(); public PersistentCacheImpl() { - segmentCacheStats = new SegmentCacheStats("stats", () -> maximumWeight, () -> elementCount.get(), () -> currentWeight.get(), () -> evictionCount.get()); + segmentCacheStats = new SegmentCacheStats("stats", () -> maximumWeight, () -> elementCount.get(), () -> currentWeight.get(), () -> evictionCount.get(), () -> 0L); } long maximumWeight = Long.MAX_VALUE; @@ -179,7 +179,7 @@ public PersistentCacheImpl() { AtomicLong evictionCount = new AtomicLong(); void AbstractPersistentCache() { - segmentCacheStats = new SegmentCacheStats("stats", () -> maximumWeight, () -> elementCount.get(), () -> currentWeight.get(), () -> evictionCount.get()); + segmentCacheStats = new SegmentCacheStats("stats", () -> maximumWeight, () -> elementCount.get(), () -> currentWeight.get(), () -> evictionCount.get(), () -> 0L); } @Override