Skip data loaders with no queued keys in ticker-mode dispatchAll

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.
+ * <p>
+ * 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<DataLoader<Integer, Integer>> 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<Integer, Integer> echo = CompletableFuture::completedFuture;
+        DataLoaderOptions uncached = DataLoaderOptions.newOptions().setCachingEnabled(false).build();
+        List<DataLoader<Integer, Integer>> loaders = new ArrayList<>();
+        for (int i = 0; i < registeredLoaders; i++) {
+            DataLoader<Integer, Integer> 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<Integer, Integer> loader : queuedLoaders) {
+            loader.load(1);
+        }
+        return registry.dispatchAllWithCount();
+    }
+}

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<String, DataLoader<?, ?>> 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<String, DataLoader<?, ?>> 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

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<String> visited = new ArrayList<>();
+        DispatchPredicate recordingPredicate = (key, dl) -> {
+            visited.add(key);
+            return true;
+        };
+        ScheduledDataLoaderRegistry registry = tickerRegistry(executor, Duration.ofHours(1), 20, recordingPredicate);
+        DataLoader<String, String> loader = registry.getDataLoader("loader-7");
+        CompletableFuture<String> 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);