From 336accf447280551abdf4e3bb72ca6db84cd926b Mon Sep 17 00:00:00 2001 From: Noemi Szemenyei Date: Thu, 10 Mar 2022 16:57:56 +0100 Subject: [PATCH] Provide custom BackgroundExecutorWrapper to pass thread locals --- .../config/BackgroundExecutorWrapper.java | 23 ++++ .../java/io/ebean/config/DatabaseConfig.java | 16 +++ .../config/MdcBackgroundExecutorWrapper.java | 50 ++++++++ .../server/core/DefaultContainer.java | 4 +- .../executor/DefaultBackgroundExecutor.java | 98 +++++++++------ .../executor/DaemonExecutorServiceTest.java | 2 +- .../DefaultBackgroundExecutorTest.java | 51 +++++--- .../org/tests/cache/TestBeanCacheAsync.java | 113 ++++++++++++++++++ 8 files changed, 307 insertions(+), 50 deletions(-) create mode 100644 ebean-api/src/main/java/io/ebean/config/BackgroundExecutorWrapper.java create mode 100644 ebean-api/src/main/java/io/ebean/config/MdcBackgroundExecutorWrapper.java create mode 100644 ebean-test/src/test/java/org/tests/cache/TestBeanCacheAsync.java diff --git a/ebean-api/src/main/java/io/ebean/config/BackgroundExecutorWrapper.java b/ebean-api/src/main/java/io/ebean/config/BackgroundExecutorWrapper.java new file mode 100644 index 000000000..289b7327a --- /dev/null +++ b/ebean-api/src/main/java/io/ebean/config/BackgroundExecutorWrapper.java @@ -0,0 +1,23 @@ +package io.ebean.config; + +import java.util.concurrent.Callable; + +/** + * BackgroundExecutorWrapper that can be used to wrap tasks that are sent to background (i.e. an other thread). + * It should copy all neccessary thread-local variables. See {@link MdcBackgroundExecutorWrapper} for implementation details. + * + * @author Roland Praml, FOCONIS AG + */ +public interface BackgroundExecutorWrapper { + + /** + * Wrap the task with MDC context if defined. + */ + Callable wrap(Callable task); + + /** + * Wrap the task with MDC context if defined. + */ + Runnable wrap(Runnable task); + +} diff --git a/ebean-api/src/main/java/io/ebean/config/DatabaseConfig.java b/ebean-api/src/main/java/io/ebean/config/DatabaseConfig.java index 2e61d2755..fdadfdc0f 100644 --- a/ebean-api/src/main/java/io/ebean/config/DatabaseConfig.java +++ b/ebean-api/src/main/java/io/ebean/config/DatabaseConfig.java @@ -454,6 +454,7 @@ public class DatabaseConfig { private int backgroundExecutorSchedulePoolSize = 1; private int backgroundExecutorShutdownSecs = 30; + private BackgroundExecutorWrapper backgroundExecutorWrapper = new MdcBackgroundExecutorWrapper(); // defaults for the L2 bean caching @@ -1480,6 +1481,20 @@ public class DatabaseConfig { this.backgroundExecutorShutdownSecs = backgroundExecutorShutdownSecs; } + /** + * Return the background executor wrapper. + */ + public BackgroundExecutorWrapper getBackgroundExecutorWrapper() { + return backgroundExecutorWrapper; + } + + /** + * Sets the background executor wrapper. The wrapper is used when a task is sent to background and should copy the thread-locals. + */ + public void setBackgroundExecutorWrapper(BackgroundExecutorWrapper backgroundExecutorWrapper) { + this.backgroundExecutorWrapper = backgroundExecutorWrapper; + } + /** * Return the L2 cache default max size. */ @@ -2926,6 +2941,7 @@ public class DatabaseConfig { backgroundExecutorSchedulePoolSize = p.getInt("backgroundExecutorSchedulePoolSize", backgroundExecutorSchedulePoolSize); backgroundExecutorShutdownSecs = p.getInt("backgroundExecutorShutdownSecs", backgroundExecutorShutdownSecs); + backgroundExecutorWrapper = p.createInstance(BackgroundExecutorWrapper.class, "backgroundExecutorWrapper", backgroundExecutorWrapper); disableClasspathSearch = p.getBoolean("disableClasspathSearch", disableClasspathSearch); currentUserProvider = p.createInstance(CurrentUserProvider.class, "currentUserProvider", currentUserProvider); databasePlatform = p.createInstance(DatabasePlatform.class, "databasePlatform", databasePlatform); diff --git a/ebean-api/src/main/java/io/ebean/config/MdcBackgroundExecutorWrapper.java b/ebean-api/src/main/java/io/ebean/config/MdcBackgroundExecutorWrapper.java new file mode 100644 index 000000000..5a0fa7980 --- /dev/null +++ b/ebean-api/src/main/java/io/ebean/config/MdcBackgroundExecutorWrapper.java @@ -0,0 +1,50 @@ +package io.ebean.config; + +import java.util.Map; +import java.util.concurrent.Callable; + +import org.slf4j.MDC; + +public class MdcBackgroundExecutorWrapper implements BackgroundExecutorWrapper { + + + /** + * Wrap the task with MDC context if defined. + */ + @Override + public Callable wrap(Callable task) { + final Map map = MDC.getCopyOfContextMap(); + if (map == null) { + return task; + } else { + return () -> { + MDC.setContextMap(map); + try { + return task.call(); + } finally { + MDC.clear(); + } + }; + } + } + + /** + * Wrap the task with MDC context if defined. + */ + @Override + public Runnable wrap(Runnable task) { + final Map map = MDC.getCopyOfContextMap(); + if (map == null) { + return task; + } else { + return () -> { + MDC.setContextMap(map); + try { + task.run(); + } finally { + MDC.clear(); + } + }; + } + } +} diff --git a/ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java b/ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java index b1626036c..2001b043c 100644 --- a/ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java +++ b/ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java @@ -1,5 +1,6 @@ package io.ebeaninternal.server.core; +import io.ebean.config.BackgroundExecutorWrapper; import io.ebean.config.ContainerConfig; import io.ebean.config.DatabaseConfig; import io.ebean.config.DatabaseConfigProvider; @@ -67,7 +68,8 @@ public final class DefaultContainer implements SpiContainer { String namePrefix = "ebean-" + config.getName(); int schedulePoolSize = config.getBackgroundExecutorSchedulePoolSize(); int shutdownSecs = config.getBackgroundExecutorShutdownSecs(); - return new DefaultBackgroundExecutor(schedulePoolSize, shutdownSecs, namePrefix); + BackgroundExecutorWrapper wrapper = config.getBackgroundExecutorWrapper(); + return new DefaultBackgroundExecutor(schedulePoolSize, shutdownSecs, namePrefix, wrapper); } /** diff --git a/ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java b/ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java index d04348ba8..ca1f18947 100644 --- a/ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java +++ b/ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java @@ -1,70 +1,92 @@ package io.ebeaninternal.server.executor; import io.avaje.lang.NonNullApi; +import io.ebean.config.BackgroundExecutorWrapper; import io.ebeaninternal.api.SpiBackgroundExecutor; -import org.slf4j.MDC; -import java.util.Map; import java.util.concurrent.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + /** * The default implementation of the BackgroundExecutor. */ @NonNullApi public final class DefaultBackgroundExecutor implements SpiBackgroundExecutor { + protected static final Logger logger = LoggerFactory.getLogger("io.ebean.BackgroundExecutor"); + private final ScheduledExecutorService schedulePool; private final DaemonExecutorService pool; + private final BackgroundExecutorWrapper wrapper; /** * Construct the default implementation of BackgroundExecutor. */ - public DefaultBackgroundExecutor(int schedulePoolSize, int shutdownWaitSeconds, String namePrefix) { + public DefaultBackgroundExecutor(int schedulePoolSize, int shutdownWaitSeconds, String namePrefix, BackgroundExecutorWrapper wrapper) { this.schedulePool = new DaemonScheduleThreadPool(schedulePoolSize, shutdownWaitSeconds, namePrefix + "-periodic-"); this.pool = new DaemonExecutorService(shutdownWaitSeconds, namePrefix); + this.wrapper = wrapper; + logger.debug("Created backgroundExecutor {} (schedulePoolSize={}, shutdownWaitSeconds={})", namePrefix, schedulePoolSize, shutdownWaitSeconds); } /** * Wrap the task with MDC context if defined. */ - Callable wrapMDC(Callable task) { - final Map map = MDC.getCopyOfContextMap(); - if (map == null) { - return task; + Callable wrap(Callable task) { + if (wrapper == null) { + return clock(task); } else { - return () -> { - MDC.setContextMap(map); - try { - return task.call(); - } finally { - MDC.clear(); - } - }; + return wrapper.wrap(clock(task)); } } /** * Wrap the task with MDC context if defined. */ - Runnable wrapMDC(Runnable task) { - final Map map = MDC.getCopyOfContextMap(); - if (map == null) { - return task; + Runnable wrap(Runnable task) { + if (wrapper == null) { + return clock(task); } else { + return wrapper.wrap(clock(task)); + } + } + + private Callable clock(Callable task) { + if (logger.isTraceEnabled()) { + long queued = System.nanoTime(); + logger.trace("Queued {}", task); return () -> { - MDC.setContextMap(map); - try { - task.run(); - } finally { - MDC.clear(); - } + long start = System.nanoTime(); + logger.trace("Start {} (delay time {} us)", task, (start - queued) / 1000L); + T ret = task.call(); + logger.trace("Stop {} (exec time {} us)", task, (System.nanoTime() - start) / 1000L); + return ret; }; + } else { + return task; + } + } + + private Runnable clock(Runnable task) { + if (logger.isTraceEnabled()) { + long queued = System.nanoTime(); + logger.trace("Queued {}", task); + return () -> { + long start = System.nanoTime(); + logger.trace("Start {} (delay time {} us)", task, (start - queued) / 1000L); + task.run(); + logger.trace("Stop {} (exec time {} us)", task, (System.nanoTime() - start) / 1000L); + }; + } else { + return task; } } @Override public Future submit(Callable task) { - return pool.submit(wrapMDC(task)); + return pool.submit(wrap(task)); } /** @@ -72,48 +94,56 @@ public final class DefaultBackgroundExecutor implements SpiBackgroundExecutor { */ @Override public Future submit(Runnable task) { - return pool.submit(wrapMDC(task)); + return pool.submit(wrap(task)); } @Override public void execute(Runnable task) { - submit(task); + submit(() -> { + try { + task.run(); + } catch (Throwable t) { + logger.error("Error while executing the task {}", task, t); + } + }); } @Override public void executePeriodically(Runnable task, long delay, TimeUnit unit) { - schedulePool.scheduleWithFixedDelay(wrapMDC(task), delay, delay, unit); + schedulePool.scheduleWithFixedDelay(wrap(task), delay, delay, unit); } @Override public void executePeriodically(Runnable task, long initialDelay, long delay, TimeUnit unit) { - schedulePool.scheduleWithFixedDelay(wrapMDC(task), initialDelay, delay, unit); + schedulePool.scheduleWithFixedDelay(wrap(task), initialDelay, delay, unit); } @Override public ScheduledFuture scheduleWithFixedDelay(Runnable task, long initialDelay, long delay, TimeUnit unit) { - return schedulePool.scheduleWithFixedDelay(wrapMDC(task), initialDelay, delay, unit); + return schedulePool.scheduleWithFixedDelay(wrap(task), initialDelay, delay, unit); } @Override public ScheduledFuture scheduleAtFixedRate(Runnable task, long initialDelay, long delay, TimeUnit unit) { - return schedulePool.scheduleAtFixedRate(wrapMDC(task), initialDelay, delay, unit); + return schedulePool.scheduleAtFixedRate(wrap(task), initialDelay, delay, unit); } @Override public ScheduledFuture schedule(Runnable task, long delay, TimeUnit unit) { - return schedulePool.schedule(wrapMDC(task), delay, unit); + return schedulePool.schedule(wrap(task), delay, unit); } @Override public ScheduledFuture schedule(Callable task, long delay, TimeUnit unit) { - return schedulePool.schedule(wrapMDC(task), delay, unit); + return schedulePool.schedule(wrap(task), delay, unit); } @Override public void shutdown() { + logger.trace("Shutting down backgroundExecutor"); schedulePool.shutdown(); pool.shutdown(); + logger.debug("BackgroundExecutor stopped"); } } diff --git a/ebean-core/src/test/java/io/ebeaninternal/server/executor/DaemonExecutorServiceTest.java b/ebean-core/src/test/java/io/ebeaninternal/server/executor/DaemonExecutorServiceTest.java index 72ad74223..cf117e018 100644 --- a/ebean-core/src/test/java/io/ebeaninternal/server/executor/DaemonExecutorServiceTest.java +++ b/ebean-core/src/test/java/io/ebeaninternal/server/executor/DaemonExecutorServiceTest.java @@ -31,7 +31,7 @@ class DaemonExecutorServiceTest { @Test void submit_via_DefaultBackgroundExecutor() throws Exception { - DefaultBackgroundExecutor des = new DefaultBackgroundExecutor(1, 5, "junk"); + DefaultBackgroundExecutor des = new DefaultBackgroundExecutor(1, 5, "junk", null); long start = System.currentTimeMillis(); List> futures = new ArrayList<>(); for (int i = 0; i < count; i++) { diff --git a/ebean-test/src/test/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutorTest.java b/ebean-test/src/test/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutorTest.java index 08bc88640..cc3a3d481 100644 --- a/ebean-test/src/test/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutorTest.java +++ b/ebean-test/src/test/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutorTest.java @@ -4,17 +4,21 @@ import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.slf4j.MDC; +import io.ebean.config.MdcBackgroundExecutorWrapper; + +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertTrue; public class DefaultBackgroundExecutorTest { @Test public void submit_callable() throws Exception { - DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 2, "test"); + DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 2, "test", null); final Future future0 = es.submit(() -> "Hello"); final Future future1 = es.submit(() -> "There"); @@ -39,7 +43,7 @@ public class DefaultBackgroundExecutorTest { public void shutdown_slowCallable_expect_interrupted() throws Exception { int shutdownWaitSecs = 1; - DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, shutdownWaitSecs, "test"); + DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, shutdownWaitSecs, "test", null); final Future future2 = es.submit(() -> { try { @@ -61,7 +65,7 @@ public class DefaultBackgroundExecutorTest { @Disabled("test takes long time") public void shutdown_when_running_expect_waitAndNiceShutdown() { - DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 20, "test"); + DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 20, "test", null); es.execute(new RunFor(3000, "a")); es.execute(new RunFor(3000, "b")); @@ -74,7 +78,7 @@ public class DefaultBackgroundExecutorTest { @Disabled("test takes long time") public void shutdown_when_rougeRunnable_expect_InterruptedException() { - DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test"); + DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test", null); es.execute(new RunFor(300000, "a")); es.execute(new RunFor(3000, "b")); @@ -85,12 +89,12 @@ public class DefaultBackgroundExecutorTest { @Test public void wrapWithNoMDC() { - DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test"); + DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test", null); assertThat(MDC.getCopyOfContextMap()).isNull(); - es.wrapMDC(() -> { + es.wrap(() -> { assertThat(MDC.getCopyOfContextMap()).isNull(); }); - es.wrapMDC(() -> { + es.wrap(() -> { assertThat(MDC.getCopyOfContextMap()).isNull(); return "Callable"; }); @@ -98,25 +102,44 @@ public class DefaultBackgroundExecutorTest { } @Test - public void wrapWithMDC_expect_() { - DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test"); + public void wrapWithMDC_expect_() throws Exception { + DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test", new MdcBackgroundExecutorWrapper()); + // MDC has a copyOnThread map. So we must pass different values to check if the test will work MDC.clear(); + es.submit(()->{ + assertThat(MDC.get("hello")).isNull(); + }).get(); + MDC.put("hello", "there"); - es.wrapMDC(() -> { + es.wrap(() -> { assertThat(MDC.get("hello")).isEqualTo("there"); - }); - es.wrapMDC(() -> { + }).run(); // will clear the MDC. But this should be OK + + MDC.put("hello", "there"); + es.wrap(() -> { assertThat(MDC.get("hello")).isEqualTo("there"); return "Callable"; - }); + }).call(); // will clear the MDC. But this should be OK + + MDC.put("hello", "there"); + + CountDownLatch latch = new CountDownLatch(1); es.execute(() -> { + // the assertion is executed async, so it will only logged on console assertThat(MDC.get("hello")).isEqualTo("there"); + latch.countDown(); }); + assertTrue(latch.await(5, TimeUnit.SECONDS)); + es.submit(() -> { assertThat(MDC.get("hello")).isEqualTo("there"); return "Callable"; - }); + }).get(); MDC.clear(); + + es.execute(()->{ + assertThat(MDC.get("hello")).isNull(); + }); es.shutdown(); } diff --git a/ebean-test/src/test/java/org/tests/cache/TestBeanCacheAsync.java b/ebean-test/src/test/java/org/tests/cache/TestBeanCacheAsync.java new file mode 100644 index 000000000..41430f246 --- /dev/null +++ b/ebean-test/src/test/java/org/tests/cache/TestBeanCacheAsync.java @@ -0,0 +1,113 @@ +package org.tests.cache; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.util.concurrent.Callable; + +import org.junit.jupiter.api.Test; +import org.tests.model.basic.OCachedBean; + +import io.ebean.BaseTestCase; +import io.ebean.DB; +import io.ebean.Database; +import io.ebean.DatabaseFactory; +import io.ebean.config.CurrentTenantProvider; +import io.ebean.config.DatabaseConfig; +import io.ebean.config.MdcBackgroundExecutorWrapper; +import io.ebeaninternal.server.cache.DefaultServerCachePlugin; + +/** + * Test class testing async/background cache updates in a multi-tenant environment. + */ +public class TestBeanCacheAsync extends BaseTestCase { + + private final ThreadLocal tenantId = new ThreadLocal<>(); + + class ThreadLocalTenantProvider implements CurrentTenantProvider { + + @Override + public Object currentId() { + return tenantId.get(); + } + } + /** + * Copy tenant info to the background thread. + */ + class TenantCopyBackgroundExecutorWrapper extends MdcBackgroundExecutorWrapper { + @Override + public Callable wrap(Callable task) { + String tenant = tenantId.get(); + if (tenant == null) { + return super.wrap(task); + } else { + return () -> { + tenantId.set(tenant); + try { + return super.wrap(task).call(); + } finally { + tenantId.remove(); + } + }; + } + } + + @Override + public Runnable wrap(Runnable task) { + String tenant = tenantId.get(); + if (tenant == null) { + return super.wrap(task); + } else { + return () -> { + tenantId.set(tenant); + try { + super.wrap(task).run(); + } finally { + tenantId.remove(); + } + }; + } + } + } + + @Test + public void findById_with_tenant() throws InterruptedException { + DatabaseConfig config = new DatabaseConfig(); + config.setName(DB.getDefault().name()); + config.loadFromProperties(); + config.setDataSource(DB.getDefault().dataSource()); + config.setReadOnlyDataSource(DB.getDefault().readOnlyDataSource()); + config.setDdlExtra(false); + config.setDdlGenerate(false); + config.setDdlRun(false); + config.setDefaultServer(false); + config.setRegister(false); + config.setServerCachePlugin(new DefaultServerCachePlugin()); // disables foreground local caching (as it is done in Hz/Ignite) + config.setCurrentTenantProvider(new ThreadLocalTenantProvider()); + config.setBackgroundExecutorWrapper(new TenantCopyBackgroundExecutorWrapper()); + tenantId.set("4711"); + + Database db = DatabaseFactory.create(config); + try { + OCachedBean bean = new OCachedBean(); + bean.setName("findById"); + db.save(bean); + + OCachedBean bean0 = db.find(OCachedBean.class, bean.getId()); + assertNotNull(bean0); + assertThat(bean0.getName()).isEqualTo("findById"); + bean0.setName("findById2"); + db.save(bean0); + + Thread.sleep(100); // TODO: can we block finds on that ID if a pending cache update is present? + + bean0 = db.find(OCachedBean.class, bean.getId()); + assertNotNull(bean0); + assertThat(bean0.getName()).isEqualTo("findById2"); + + } finally { + db.shutdown(); + } + } + +}