OAK-12282 : defining a fixed bound for the AbstractDiskCache

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

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() {

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<String> 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<Runnable> 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();
+    }
 }

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<Long> writeDiscardCountSupplier;
+
     public SegmentCacheStats(@NotNull String name,
                              @NotNull Supplier<Long> maximumWeight,
                              @NotNull Supplier<Long> elementCount,
                              @NotNull Supplier<Long> currentWeight,
-                             @NotNull Supplier<Long> evictionCount) {
+                             @NotNull Supplier<Long> evictionCount,
+                             @NotNull Supplier<Long> 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();

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;

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

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<UUID, Buffer> 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