pat-lego commented on code in PR #2978:
URL: https://github.com/apache/jackrabbit-oak/pull/2978#discussion_r3536889477
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/spi/persistence/persistentcache/AbstractPersistentCache.java:
##########
@@ -31,27 +31,68 @@
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.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
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", THREADS * 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;
+ /**
+ * Name of the feature toggle for the OAK-12282 bounded write-queue fix.
+ * See {@link #FT_OAK_12282_BOUNDED_WRITE_QUEUE_ENABLED}.
+ */
+ public static final String FT_OAK_12282 = "FT_OAK-12282";
+
+ /**
+ * Whether the bounded write queue introduced in OAK-12282 is active.
+ * <p>
+ * When {@code true} (default), the executor uses a queue bounded to
+ * {@link #WRITE_QUEUE_SIZE} and silently discards write tasks when full,
+ * preventing OOM under high write load. This is safe because the disk
+ * cache is an optimisation only — a dropped write means the segment is
+ * fetched from remote storage on the next read.
+ * <p>
+ * Set to {@code false} via the {@link
org.apache.jackrabbit.oak.spi.toggle.FeatureToggle}
+ * registered with the Whiteboard to revert to the pre-fix unbounded queue.
+ * <strong>Note:</strong> changing this flag requires a process restart, as
+ * the executor is created at startup.
+ */
+ public static final AtomicBoolean FT_OAK_12282_BOUNDED_WRITE_QUEUE_ENABLED
= new AtomicBoolean(true);
+
public AbstractPersistentCache() {
- executor = Executors.newFixedThreadPool(THREADS);
+ // Formerly Executors.newFixedThreadPool() was used here, which
creates an unbounded
+ // LinkedBlockingQueue — allowing unlimited segment buffers to pile up
in memory under
+ // high write load. The bounded queue (gated by FT_OAK_12282) prevents
OOM by dropping
+ // write tasks when full; this is safe because the disk cache is an
optimisation only.
+ BlockingQueue<Runnable> writeQueue =
FT_OAK_12282_BOUNDED_WRITE_QUEUE_ENABLED.get()
+ ? new LinkedBlockingQueue<>(WRITE_QUEUE_SIZE)
+ : new LinkedBlockingQueue<>();
+ executor = new ThreadPoolExecutor(
Review Comment:
Gave the thread factory a proper name for easier debugging.
--
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]
