joerghoh commented on code in PR #3071:
URL: https://github.com/apache/jackrabbit-oak/pull/3071#discussion_r3766241392
##########
oak-core-spi/src/main/java/org/apache/jackrabbit/oak/cache/api/CacheBuilder.java:
##########
@@ -255,11 +285,19 @@ private LoadingCache<K, V> buildCaffeine(CacheLoader<K,
V> loader) {
@SuppressWarnings({"unchecked", "rawtypes"})
private Caffeine<K, V> configureCaffeineBuilder() {
Caffeine caffeineBuilder = Caffeine.newBuilder();
- if (refreshAfterWrite == null) {
- // Caffeine uses one executor for both maintenance and refresh
work.
- // Run maintenance on the caller thread unless refresh must stay
asynchronous.
- caffeineBuilder = caffeineBuilder.executor(Runnable::run);
- }
+ // Caffeine uses one executor for both maintenance and refresh. A
refresh loader may make a
+ // remote call; running it inline would block every caller thread that
triggers a refresh on
+ // that call, which defeats the point of refreshAfterWrite (return the
stale value, reload in
+ // the background). So refreshing caches always run on Oak's
maintenance executor, regardless
+ // of the toggle - the tradeoff is that a slow reload can occupy one
of the pool's threads for
+ // longer, which is preferable to blocking callers. Zero-capacity
caches are a "disable
+ // caching" idiom relied upon elsewhere for immediate eviction, so
they always run inline -
+ // otherwise a read immediately following a write could still observe
the entry before
+ // background maintenance evicts it.
+ boolean zeroCapacity = maximumWeight == 0 || maximumSize == 0;
+ boolean inline = zeroCapacity
Review Comment:
```suggestion
boolean inSameThread = zeroCapacity
```
##########
oak-core-spi/src/main/java/org/apache/jackrabbit/oak/cache/impl/CacheMaintenanceExecutor.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.jackrabbit.oak.cache.impl;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+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;
+
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The process-wide pool Caffeine-backed Oak caches run their maintenance
(eviction, removal
+ * notification, buffer drains) on, instead of {@link
java.util.concurrent.ForkJoinPool#commonPool()}.
+ */
+public final class CacheMaintenanceExecutor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CacheMaintenanceExecutor.class);
+
+ private static final Thread.UncaughtExceptionHandler
UNCAUGHT_EXCEPTION_HANDLER = (t, e) ->
+ LOG.warn("Uncaught exception in thread {}", t.getName(), e);
+
+ private static final String THREAD_PREFIX = "oak-cache-maintenance-";
Review Comment:
That feels a bit too generic, maybe more specific?
```suggestion
private static final String THREAD_PREFIX = "oak-caffeine-maintenance-";
```
##########
oak-core-spi/src/test/java/org/apache/jackrabbit/oak/cache/impl/CacheBuilderMaintenanceTest.java:
##########
@@ -0,0 +1,369 @@
+/*
+ * 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.jackrabbit.oak.cache.impl;
+
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.jackrabbit.oak.cache.api.Cache;
+import org.apache.jackrabbit.oak.cache.api.CacheBuilder;
+import org.apache.jackrabbit.oak.cache.api.EvictionCause;
+import org.apache.jackrabbit.oak.cache.api.LoadingCache;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests that Caffeine cache maintenance (eviction, removal notification) is
dispatched
+ * off the calling thread, per OAK-12290.
+ */
+public class CacheBuilderMaintenanceTest {
+
+ private static final long TIMEOUT_SECONDS = 10;
+
+ /**
+ * The toggle is process-wide static state, so reset it around every test
- a test that leaked
+ * inline maintenance would silently change the behaviour asserted by
every later test in the
+ * same JVM.
+ */
+ @Before
+ public void enableOak12290Toggle() {
+ CacheBuilder.FT_OAK_12290_ASYNC_CACHE_MAINTENANCE_ENABLED.set(true);
+ }
+
+ @After
+ public void resetOak12290Toggle() {
+ CacheBuilder.FT_OAK_12290_ASYNC_CACHE_MAINTENANCE_ENABLED.set(true);
+ }
+
+ /** Maintenance triggered by a write must not be executed by the writing
thread. */
+ @Test
+ public void evictionNotificationRunsOffCallerThread() throws
InterruptedException {
+ AtomicReference<Thread> evictionThread = new AtomicReference<>();
+ CountDownLatch evicted = new CountDownLatch(1);
+
+ Cache<String, String> cache = CacheBuilder.<String, String>newBuilder()
+ .maximumSize(1)
+ .evictionListener((k, v, cause) -> {
+ if (cause == EvictionCause.SIZE) {
+ evictionThread.set(Thread.currentThread());
+ evicted.countDown();
+ }
+ })
+ .build();
+
+ cache.put("k1", "v1");
+ cache.put("k2", "v2");
+
+ Assert.assertTrue("size-based eviction was never notified",
+ evicted.await(TIMEOUT_SECONDS, TimeUnit.SECONDS));
+ Assert.assertNotSame("cache maintenance must not run on the calling
thread",
+ Thread.currentThread(), evictionThread.get());
+ }
+
+ /**
+ * A slow maintenance callback must not stall the writer. With inline
maintenance the
+ * writer runs the callback itself while holding the eviction lock, so
{@code put()}
+ * cannot return until the callback finishes.
+ */
+ @Test(timeout = TIMEOUT_SECONDS * 1000)
+ public void slowMaintenanceDoesNotBlockCallerThread() throws
InterruptedException {
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch maintenanceDone = new CountDownLatch(1);
+
+ Cache<String, String> cache = CacheBuilder.<String, String>newBuilder()
+ .maximumSize(1)
+ .evictionListener((k, v, cause) -> {
+ if (cause == EvictionCause.SIZE) {
+ try {
+ release.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ maintenanceDone.countDown();
+ }
+ })
+ .build();
+
+ cache.put("k1", "v1");
+ // returns only if the blocked maintenance callback runs on another
thread
+ cache.put("k2", "v2");
+
+ release.countDown();
+ Assert.assertTrue("maintenance callback never completed",
+ maintenanceDone.await(TIMEOUT_SECONDS, TimeUnit.SECONDS));
+ }
+
+ /** Disabling the toggle restores the previous inline-maintenance
behaviour. */
+ @Test
+ public void toggleDisabledRunsMaintenanceInline() {
+ AtomicReference<Thread> evictionThread = new AtomicReference<>();
+
+ CacheBuilder.FT_OAK_12290_ASYNC_CACHE_MAINTENANCE_ENABLED.set(false);
+ Cache<String, String> cache = CacheBuilder.<String, String>newBuilder()
+ .maximumSize(1)
+ .evictionListener((k, v, cause) -> {
+ if (cause == EvictionCause.SIZE) {
+ evictionThread.set(Thread.currentThread());
+ }
+ })
+ .build();
+
+ cache.put("k1", "v1");
+ cache.put("k2", "v2");
+
+ Assert.assertSame("maintenance should run inline when the toggle is
off",
+ Thread.currentThread(), evictionThread.get());
+ }
+
+ /**
+ * Maintenance must run on Oak's own named pool, not on {@code
ForkJoinPool.commonPool()} -
+ * the common pool is shared with the hosting application and can be
configured with zero
+ * workers, in which case submitted tasks are queued and never run.
+ */
+ @Test
+ public void maintenanceRunsOnOakOwnedThread() throws InterruptedException {
+ AtomicReference<String> threadName = new AtomicReference<>();
+ CountDownLatch evicted = new CountDownLatch(1);
+
+ Cache<String, String> cache = CacheBuilder.<String, String>newBuilder()
+ .maximumSize(1)
+ .evictionListener((k, v, cause) -> {
+ if (cause == EvictionCause.SIZE) {
+ threadName.set(Thread.currentThread().getName());
+ evicted.countDown();
+ }
+ })
+ .build();
+
+ cache.put("k1", "v1");
+ cache.put("k2", "v2");
+
+ Assert.assertTrue("size-based eviction was never notified",
+ evicted.await(TIMEOUT_SECONDS, TimeUnit.SECONDS));
+ Assert.assertTrue("maintenance ran on an unexpected thread: " +
threadName.get(),
+ threadName.get().startsWith("oak-cache-maintenance-"));
Review Comment:
Isn't there a constant defined for the thread name prefix?
##########
oak-core-spi/src/main/java/org/apache/jackrabbit/oak/cache/impl/CacheMaintenanceExecutor.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.jackrabbit.oak.cache.impl;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+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;
+
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The process-wide pool Caffeine-backed Oak caches run their maintenance
(eviction, removal
+ * notification, buffer drains) on, instead of {@link
java.util.concurrent.ForkJoinPool#commonPool()}.
+ */
+public final class CacheMaintenanceExecutor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CacheMaintenanceExecutor.class);
+
+ private static final Thread.UncaughtExceptionHandler
UNCAUGHT_EXCEPTION_HANDLER = (t, e) ->
+ LOG.warn("Uncaught exception in thread {}", t.getName(), e);
+
+ private static final String THREAD_PREFIX = "oak-cache-maintenance-";
+
+ /**
+ * The pool is process-wide, so its workers must not inherit an OSGi
request thread's context
+ * class loader and keep a refreshed consumer bundle alive.
+ */
+ private static final ClassLoader THREAD_CONTEXT_CLASS_LOADER =
+ CacheMaintenanceExecutor.class.getClassLoader();
+
+ /**
+ * Number of maintenance threads shared by all Oak caches, between 2 and
8. Caffeine keeps at
+ * most one maintenance task per cache in flight, so this doesn't need to
scale with core count.
+ */
+ private static final int THREADS = Math.max(2, Math.min(8,
Runtime.getRuntime().availableProcessors() - 1));
+
+ /**
+ * Bound on queued maintenance tasks: deep enough to absorb a burst,
shallow enough that a
+ * wedged pool falls back to {@link ThreadPoolExecutor.CallerRunsPolicy}
instead of queueing
+ * without bound.
+ */
+ private static final int QUEUE_CAPACITY = 1024;
+
+ private CacheMaintenanceExecutor() {
+ }
+
+ /**
+ * The shared maintenance pool. Created on first call, so no threads exist
in a JVM that never
+ * builds a cache.
+ *
+ * @return the process-wide maintenance executor
+ */
+ @NotNull
+ public static Executor get() {
+ return Holder.EXECUTOR;
+ }
+
+ /**
+ * Lazy holder so the pool is only created once a cache is actually built.
+ */
+ private static final class Holder {
+
+ private static final Executor EXECUTOR = newExecutor();
+
+ private static Executor newExecutor() {
+ AtomicInteger threadCounter = new AtomicInteger();
+ ThreadPoolExecutor executor = new ThreadPoolExecutor(
+ THREADS, THREADS,
+ 60, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(QUEUE_CAPACITY),
+ newThreadFactory(threadCounter),
+ new LoggingCallerRunsPolicy());
+ executor.allowCoreThreadTimeOut(true);
+ return executor;
+ }
+ }
+
+ static ThreadFactory newThreadFactory(AtomicInteger threadCounter) {
+ return runnable -> {
+ Thread thread = new Thread(runnable, THREAD_PREFIX +
threadCounter.incrementAndGet());
+ // Do not keep a consumer bundle's class loader alive through this
process-wide worker.
+ thread.setContextClassLoader(THREAD_CONTEXT_CLASS_LOADER);
+ // Daemon: the pool is process-wide and never shut down, and no
maintenance task is
+ // required to complete for a clean exit.
+ thread.setDaemon(true);
+ thread.setUncaughtExceptionHandler(UNCAUGHT_EXCEPTION_HANDLER);
+ return thread;
+ };
+ }
+
+ /**
+ * {@link ThreadPoolExecutor.CallerRunsPolicy} that also logs, at most
once a minute, that the
+ * pool is saturated.
+ */
+ private static final class LoggingCallerRunsPolicy implements
RejectedExecutionHandler {
+
+ private static final long LOG_INTERVAL_NANOS =
TimeUnit.MINUTES.toNanos(1);
+
+ private final AtomicLong nextLogNanos = new AtomicLong();
+
+ @Override
+ public void rejectedExecution(@NotNull Runnable task, @NotNull
ThreadPoolExecutor executor) {
+ long now = System.nanoTime();
+ long next = nextLogNanos.get();
+ if (now >= next && nextLogNanos.compareAndSet(next, now +
LOG_INTERVAL_NANOS)) {
+ LOG.warn("Cache maintenance pool exhausted ({} threads,
{}-deep queue full); running "
+ + "maintenance inline on the calling thread instead.
Expected under a burst; "
+ + "sustained occurrence means the pool is undersized
for the load.",
Review Comment:
In this particular case I would naively search for a way to configure this
limit in the pool size, but it's not possible.
Should we make ``QUEUE_CAPACITY`` configurable? Now?
--
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]
