diff --git a/src/jmh/java/performance/ScheduledDataLoaderRegistryDispatchPerformance.java b/src/jmh/java/performance/ScheduledDataLoaderRegistryDispatchPerformance.java
new file mode 100644
index 00000000..a1a0a3bf
--- /dev/null
+++ b/src/jmh/java/performance/ScheduledDataLoaderRegistryDispatchPerformance.java
@@ -0,0 +1,91 @@
+package performance;
+
+import org.dataloader.BatchLoader;
+import org.dataloader.DataLoader;
+import org.dataloader.DataLoaderFactory;
+import org.dataloader.DataLoaderOptions;
+import org.dataloader.registries.DispatchPredicate;
+import org.dataloader.registries.ScheduledDataLoaderRegistry;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A ticker-mode {@link ScheduledDataLoaderRegistry} set up the way a GraphQL server builds one per request: every
+ * data loader the application defines is registered, but a request only queues keys on a few of them, while
+ * graphql-java's per-level dispatch calls {@link ScheduledDataLoaderRegistry#dispatchAll()} many times.
+ *
+ * The tick is an hour so that the scheduled ticker never runs during a measurement.
+ */
+@State(Scope.Benchmark)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 2)
+@Fork(2)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+public class ScheduledDataLoaderRegistryDispatchPerformance {
+
+ @Param({"20", "200"})
+ int registeredLoaders;
+
+ @Param({"0", "5"})
+ int loadersWithQueuedKeys;
+
+ ScheduledThreadPoolExecutor executor;
+ ScheduledDataLoaderRegistry registry;
+ List> queuedLoaders;
+
+ @Setup
+ public void setup() {
+ executor = new ScheduledThreadPoolExecutor(1);
+ ScheduledDataLoaderRegistry.Builder builder = ScheduledDataLoaderRegistry.newScheduledRegistry()
+ .scheduledExecutorService(executor)
+ .tickerMode(true)
+ .schedule(Duration.ofHours(1))
+ .dispatchPredicate(DispatchPredicate.DISPATCH_NEVER);
+ BatchLoader echo = CompletableFuture::completedFuture;
+ DataLoaderOptions uncached = DataLoaderOptions.newOptions().setCachingEnabled(false).build();
+ List> loaders = new ArrayList<>();
+ for (int i = 0; i < registeredLoaders; i++) {
+ DataLoader loader = DataLoaderFactory.newDataLoader("loader-" + i, echo, uncached);
+ loaders.add(loader);
+ builder.register("loader-" + i, loader, DispatchPredicate.DISPATCH_ALWAYS);
+ }
+ registry = builder.build();
+ // spread across the registry rather than the first few entries
+ queuedLoaders = new ArrayList<>();
+ for (int i = 0; i < loadersWithQueuedKeys; i++) {
+ queuedLoaders.add(loaders.get(i * registeredLoaders / loadersWithQueuedKeys));
+ }
+ }
+
+ @TearDown
+ public void tearDown() {
+ registry.close();
+ executor.shutdownNow();
+ }
+
+ @Benchmark
+ public int dispatchAll() {
+ for (DataLoader loader : queuedLoaders) {
+ loader.load(1);
+ }
+ return registry.dispatchAllWithCount();
+ }
+}
diff --git a/src/main/java/org/dataloader/registries/ScheduledDataLoaderRegistry.java b/src/main/java/org/dataloader/registries/ScheduledDataLoaderRegistry.java
index 4b6cd7b7..ca4e24aa 100644
--- a/src/main/java/org/dataloader/registries/ScheduledDataLoaderRegistry.java
+++ b/src/main/java/org/dataloader/registries/ScheduledDataLoaderRegistry.java
@@ -182,6 +182,9 @@ public void dispatchAll() {
@Override
public int dispatchAllWithCount() {
+ if (tickerMode) {
+ return dispatchQueuedAndTick();
+ }
int sum = 0;
for (Map.Entry> entry : dataLoaders.entrySet()) {
DataLoader, ?> dataLoader = entry.getValue();
@@ -191,6 +194,26 @@ public int dispatchAllWithCount() {
return sum;
}
+ /**
+ * Ticker mode only dispatches the {@link DataLoader}s that have keys queued. Dispatching an empty
+ * {@link DataLoader} only refreshes its last dispatch time and calls the dispatch instrumentation, and a
+ * registry typically holds many more {@link DataLoader}s than a single dispatch has work for. A key queued
+ * after the {@link DataLoader#dispatchDepth()} check is dispatched by the next tick, as is a key queued just
+ * after a dispatch.
+ */
+ private int dispatchQueuedAndTick() {
+ // armed before dispatching so that a data loader or predicate throwing cannot stop the ticker
+ scheduleTick();
+ int sum = 0;
+ for (Map.Entry> entry : dataLoaders.entrySet()) {
+ DataLoader, ?> dataLoader = entry.getValue();
+ if (dataLoader.dispatchDepth() > 0 && shouldDispatch(entry.getKey(), dataLoader)) {
+ sum += dataLoader.dispatchWithCounts().getKeysCount();
+ }
+ }
+ return sum;
+ }
+
/**
* This will immediately dispatch the {@link DataLoader}s in the registry
diff --git a/src/test/java/org/dataloader/registries/ScheduledDataLoaderRegistryTest.java b/src/test/java/org/dataloader/registries/ScheduledDataLoaderRegistryTest.java
index 72577ca1..59f46d96 100644
--- a/src/test/java/org/dataloader/registries/ScheduledDataLoaderRegistryTest.java
+++ b/src/test/java/org/dataloader/registries/ScheduledDataLoaderRegistryTest.java
@@ -383,6 +383,28 @@ public void ticker_mode_reschedule_now_keeps_one_pending_task() {
executor.shutdownNow();
}
+ @Test
+ public void ticker_mode_dispatch_all_only_visits_data_loaders_with_queued_keys() {
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
+ List visited = new ArrayList<>();
+ DispatchPredicate recordingPredicate = (key, dl) -> {
+ visited.add(key);
+ return true;
+ };
+ ScheduledDataLoaderRegistry registry = tickerRegistry(executor, Duration.ofHours(1), 20, recordingPredicate);
+ DataLoader loader = registry.getDataLoader("loader-7");
+ CompletableFuture queued = loader.load("queued");
+
+ int dispatched = registry.dispatchAllWithCount();
+
+ assertThat(dispatched, equalTo(1));
+ assertThat(queued.getNow(null), equalTo("queued"));
+ assertThat(visited, equalTo(singletonList("loader-7")));
+ assertThat(executor.getQueue().size(), equalTo(1));
+ registry.close();
+ executor.shutdownNow();
+ }
+
@Test
public void ticker_mode_keeps_ticking_after_a_dispatch_predicate_throws() throws Exception {
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);