ScheduledDataLoaderRegistry ticker mode schedules one task per data loader on every dispatchAll(), so pending tasks grow with loaders × dispatch calls

Summary

In ticker mode, each call to dispatchAll() starts a new self-rescheduling task for every registered data loader (dispatchOrReschedule → reschedule), and the chains already running from earlier calls keep going. After k calls with n loaders, the executor holds n × k pending tasks, and each one re-runs every schedule until close().

graphql-java's per-level dispatch calls dispatchAll() many times per request, so with ~200 loaders a single request leaves thousands of tasks on the shared scheduler thread for its whole lifetime.

The ticker contract only needs one periodic task per registry: "after the first dispatchAll(), dispatch every schedule until closed".

Version: java-dataloader 6.0.0, the latest release; ScheduledDataLoaderRegistry on master is unchanged since then. We hit this through Netflix DGS 12 with dgs.graphql.dataloader.ticker-mode-enabled=true and graphql-java 25.0.

Where

ScheduledDataLoaderRegistry (6.0.0):

private void reschedule(String key, DataLoader<?, ?> dataLoader) {
    if (!closed) {
        Runnable runThis = () -> dispatchOrReschedule(key, dataLoader);
        scheduledExecutorService.schedule(runThis, schedule.toMillis(), TimeUnit.MILLISECONDS);
    }
}

private int dispatchOrReschedule(String key, DataLoader<?, ?> dataLoader) {
    ...
    if (tickerMode || !shouldDispatch) {
        reschedule(key, dataLoader);   // a new chain per loader on every dispatchAll(); existing chains continue too
    }
    return sum;
}

rescheduleNow() has the same shape: one more chain per loader per call.

Reproducer

This needs only java-dataloader 6.0.0 (plus jspecify and reactive-streams) on the classpath and uses the default dispatch predicate. Run it with java -cp 'lib/*' Repro.java.

Repro.java

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.dataloader.BatchLoader;
import org.dataloader.DataLoaderFactory;
import org.dataloader.registries.ScheduledDataLoaderRegistry;

/**
 * Ticker-mode ScheduledDataLoaderRegistry: pending executor tasks grow with
 * (registered data loaders) x (dispatchAll() calls) and stay until close().
 *
 * Only java-dataloader on the classpath, default dispatch predicate:
 *   java -cp 'lib/*' Repro.java
 */
public class Repro {
    static final int LOADERS = 200;
    static final Duration TICK = Duration.ofMillis(15);

    public static void main(String[] args) throws Exception {
        System.out.println(ScheduledDataLoaderRegistry.class.getProtectionDomain().getCodeSource().getLocation().getPath()
                + ", " + LOADERS + " data loaders, ticker mode, schedule " + TICK.toMillis() + " ms");
        System.out.printf("%-18s %16s %22s %22s%n", "dispatchAll calls", "pending tasks", "tasks run / second", "pending 1 tick after close");
        for (int calls : new int[] {1, 2, 10, 50}) {
            run(calls);
        }
    }

    static void run(int calls) throws Exception {
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
        ScheduledDataLoaderRegistry registry = ScheduledDataLoaderRegistry.newScheduledRegistry()
                .scheduledExecutorService(executor)
                .tickerMode(true)
                .schedule(TICK)
                .build();
        for (int i = 0; i < LOADERS; i++) {
            BatchLoader<String, String> echo = keys -> CompletableFuture.completedFuture(keys);
            registry.register("loader-" + i, DataLoaderFactory.newDataLoader("loader-" + i, echo));
        }

        // graphql-java calls dispatchAll() several times per request (once per dispatch point / level).
        for (int i = 0; i < calls; i++) {
            registry.dispatchAll();
        }

        int pending = executor.getQueue().size();
        long before = executor.getCompletedTaskCount();
        TimeUnit.SECONDS.sleep(1);
        long perSecond = executor.getCompletedTaskCount() - before;

        registry.close();
        TimeUnit.MILLISECONDS.sleep(TICK.toMillis() * 3);
        int afterClose = executor.getQueue().size();

        System.out.printf("%-18d %16d %22d %22d%n", calls, pending, perSecond, afterClose);
        executor.shutdownNow();
    }
}

Output: 200 loaders, 15 ms schedule, JDK 26, Apple M-series.

`dispatchAll()` calls	pending tasks, 6.0.0	task runs/s, 6.0.0	pending tasks, patched	task runs/s, patched
1	200	11,400	1	56
2	400	22,800	1	56
10	2,000	114,000	1	57
50	10,000	590,000	1	58

close() drains both variants within one tick, so a registry that is closed stops costing anything. The cost is incurred while requests are in flight, and it is permanent for a registry that is never closed.

End-to-end with graphql-java

Bench.java (below) runs a graphql-java 25.0 workload with registries built the way DGS builds them in ticker mode:

one shared single-thread scheduler and a 15 ms schedule;
a registry predicate of DISPATCH_NEVER and a per-loader predicate of DISPATCH_ALWAYS;
200 loaders, 20 items × 3 nested levels, and one chained "late" load per item that only a tick can dispatch;
48 concurrent requests, with the registry closed when the request completes.

The patched numbers come from putting the patched class ahead of the jar on the classpath.

Metric	6.0.0, 3 ms I/O	patched, 3 ms I/O	6.0.0, 20 ms I/O	patched, 20 ms I/O
Throughput (rps)	327	594	296	283
Latency p50 / p99 (ms)	140 / 225	78 / 127	157 / 208	166 / 211
Scheduled tasks per request	2,381	4.0	3,956	9.0
Scheduler queue, peak / end	59,137 / 3,400	82 / 2	56,580 / 1,200	83 / 8
Scheduler-thread CPU per request (ms)	1.94	1.04	2.55	1.39

These are single laptop runs, so read the ratios rather than the absolute numbers.

When the scheduler thread is the bottleneck, the patched registry roughly doubles throughput. When the workload is I/O-bound there is a small p50 cost. With thousands of staggered chains, some chain happens to fire much sooner than one schedule after a late load is queued. With a single tick, that load waits up to one schedule. Lowering schedule recovers this cheaply once there is a single tick per registry.

Bench.java

import com.sun.management.ThreadMXBean;
import graphql.ExecutionInput;
import graphql.GraphQL;
import graphql.schema.DataFetcher;
import graphql.schema.DataFetchingEnvironment;
import graphql.schema.idl.RuntimeWiring;
import graphql.schema.idl.SchemaGenerator;
import graphql.schema.idl.SchemaParser;
import java.lang.management.ManagementFactory;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.dataloader.BatchLoader;
import org.dataloader.DataLoader;
import org.dataloader.DataLoaderFactory;
import org.dataloader.registries.DispatchPredicate;
import org.dataloader.registries.ScheduledDataLoaderRegistry;

/**
 * graphql-java workload over a ticker-mode ScheduledDataLoaderRegistry, built the way Netflix DGS builds
 * per-request registries when dgs.graphql.dataloader.ticker-mode-enabled=true: one shared single-thread
 * scheduler, 15 ms schedule, registry predicate DISPATCH_NEVER, per-loader DISPATCH_ALWAYS, registry closed
 * when the request completes.
 *
 * The query fans out to FIELD_LOADERS loaders per item over DEPTH nested levels; each item also issues one
 * "late" load chained on another load's result, which only a tick can dispatch.
 *
 *   java -Xms2g -Xmx2g -cp 'lib/*' Bench.java --io-ms=3
 *   java -Xms2g -Xmx2g -cp 'patched/classes:lib/*' Bench.java --io-ms=3
 */
public class Bench {
    static final int LOADERS = 200, FIELD_LOADERS = 10, ITEMS = 20, DEPTH = 3, WARMUP = 300;
    static final Duration TICK = Duration.ofMillis(15);

    public static void main(String[] args) throws Exception {
        int requests = 3000, concurrency = 48;
        long ioMs = 3;
        for (String a : args) {
            String[] kv = a.replaceFirst("^--", "").split("=", 2);
            switch (kv[0]) {
                case "requests" -> requests = Integer.parseInt(kv[1]);
                case "concurrency" -> concurrency = Integer.parseInt(kv[1]);
                case "io-ms" -> ioMs = Long.parseLong(kv[1]);
                default -> throw new IllegalArgumentException(a);
            }
        }
        String impl = ScheduledDataLoaderRegistry.class.getProtectionDomain().getCodeSource().getLocation().getPath();
        System.out.printf(Locale.ROOT, "registry class from %s%nrequests=%d concurrency=%d ioMs=%d loaders=%d items=%d depth=%d tick=%dms%n",
                impl, requests, concurrency, ioMs, LOADERS, ITEMS, DEPTH, TICK.toMillis());
        new Bench(ioMs).measure(requests, concurrency);
    }

    final AtomicReference<Thread> schedulerThread = new AtomicReference<>();
    final ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1, r -> {
        Thread t = new Thread(r, "dataloader-scheduler");
        schedulerThread.set(t);
        return t;
    });
    final ExecutorService io = Executors.newVirtualThreadPerTaskExecutor();
    final Executor ioDelay;
    final GraphQL graphql = buildGraphQL();
    final String query = "{ items { " + itemFields(DEPTH) + " } }";

    Bench(long ioMs) {
        ioDelay = CompletableFuture.delayedExecutor(ioMs, TimeUnit.MILLISECONDS, io);
    }

    void measure(int requests, int concurrency) throws Exception {
        for (int i = 0; i < WARMUP; i++) runOne().get();

        ThreadMXBean threads = (ThreadMXBean) ManagementFactory.getThreadMXBean();
        long tickerCpu0 = threads.getThreadCpuTime(schedulerThread.get().threadId());
        long tasks0 = scheduler.getTaskCount();
        AtomicInteger peakQueue = new AtomicInteger();
        ScheduledExecutorService sampler = Executors.newSingleThreadScheduledExecutor();
        sampler.scheduleAtFixedRate(() -> peakQueue.accumulateAndGet(scheduler.getQueue().size(), Math::max), 0, 5, TimeUnit.MILLISECONDS);

        Semaphore permits = new Semaphore(concurrency);
        ConcurrentLinkedQueue<Long> latencies = new ConcurrentLinkedQueue<>();
        long t0 = System.nanoTime();
        for (int i = 0; i < requests; i++) {
            permits.acquire();
            runOne().whenComplete((ns, err) -> {
                if (err == null) latencies.add(ns);
                permits.release();
            });
        }
        permits.acquire(concurrency);
        double wallMs = (System.nanoTime() - t0) / 1e6;
        int queueAtEnd = scheduler.getQueue().size();
        sampler.shutdownNow();

        double tickerCpuMs = (threads.getThreadCpuTime(schedulerThread.get().threadId()) - tickerCpu0) / 1e6;
        List<Long> sorted = new ArrayList<>(latencies);
        sorted.sort(null);
        System.out.printf(Locale.ROOT, "| throughput (rps) | %.0f |%n", requests / (wallMs / 1000));
        System.out.printf(Locale.ROOT, "| latency p50 / p99 (ms) | %.0f / %.0f |%n", pct(sorted, 0.5), pct(sorted, 0.99));
        System.out.printf(Locale.ROOT, "| scheduled tasks / request | %.1f |%n", (scheduler.getTaskCount() - tasks0) / (double) requests);
        System.out.printf(Locale.ROOT, "| scheduler queue peak / end | %d / %d |%n", peakQueue.get(), queueAtEnd);
        System.out.printf(Locale.ROOT, "| scheduler-thread CPU / request (ms) | %.2f |%n", tickerCpuMs / requests);
        scheduler.shutdownNow();
        io.shutdownNow();
    }

    static double pct(List<Long> sorted, double q) {
        return sorted.get((int) ((sorted.size() - 1) * q)) / 1e6;
    }

    CompletableFuture<Long> runOne() {
        long start = System.nanoTime();
        ScheduledDataLoaderRegistry registry = registry();
        ExecutionInput input = ExecutionInput.newExecutionInput(query).dataLoaderRegistry(registry).build();
        return graphql.executeAsync(input).thenApply(result -> {
            if (!result.getErrors().isEmpty()) throw new IllegalStateException(result.getErrors().toString());
            registry.close();
            return System.nanoTime() - start;
        });
    }

    ScheduledDataLoaderRegistry registry() {
        ScheduledDataLoaderRegistry registry = ScheduledDataLoaderRegistry.newScheduledRegistry()
                .scheduledExecutorService(scheduler)
                .tickerMode(true)
                .schedule(TICK)
                .dispatchPredicate(DispatchPredicate.DISPATCH_NEVER)
                .build();
        for (int i = 0; i < LOADERS; i++) {
            String name = "loader-" + i;
            BatchLoader<Integer, Integer> loader = keys -> CompletableFuture.supplyAsync(() -> keys, ioDelay);
            registry.register(name, DataLoaderFactory.newDataLoader(name, loader), DispatchPredicate.DISPATCH_ALWAYS);
        }
        return registry;
    }

    static String itemFields(int depth) {
        String scalars = IntStream.range(0, FIELD_LOADERS).mapToObj(i -> "f" + i).collect(Collectors.joining(" ")) + " id late";
        return depth == 0 ? scalars : scalars + " children { " + itemFields(depth - 1) + " }";
    }

    static DataLoader<Integer, Integer> loader(DataFetchingEnvironment env, String name) {
        return env.getDataLoader(name);
    }

    GraphQL buildGraphQL() {
        String sdl = "type Query { items: [Item] }\ntype Item { id: Int late: Int children: [Item] "
                + IntStream.range(0, FIELD_LOADERS).mapToObj(i -> "f" + i + ": Int").collect(Collectors.joining(" ")) + " }";
        RuntimeWiring wiring = RuntimeWiring.newRuntimeWiring()
                .type("Query", t -> t.dataFetcher("items", env -> IntStream.range(0, ITEMS).boxed().toList()))
                .type("Item", t -> {
                    t.dataFetcher("id", env -> env.getSource());
                    for (int i = 0; i < FIELD_LOADERS; i++) {
                        String name = "loader-" + i;
                        t.dataFetcher("f" + i, (DataFetcher<?>) env -> loader(env, name).load(env.<Integer>getSource()));
                    }
                    t.dataFetcher("late", env -> loader(env, "loader-100").load(env.<Integer>getSource())
                            .thenCompose(v -> loader(env, "loader-101").load(v)));
                    t.dataFetcher("children", env -> loader(env, "loader-150").load(env.<Integer>getSource())
                            .thenApply(id -> List.of(id * 10, id * 10 + 1, id * 10 + 2)));
                    return t;
                })
                .build();
        return GraphQL.newGraphQL(new SchemaGenerator().makeExecutableSchema(new SchemaParser().parse(sdl), wiring)).build();
    }
}

Production

We saw this on a production DGS 12 service with ~200 data loaders per request, a 15 ms schedule and a CPU-bound workload. We compared continuous-profiler samples, 8 pods × 60–120 s, before and after shipping the change below as a patched jar. Both sets were taken before an unrelated profiler-configuration change that affected heap and GC.

	6.0.0	patched
CPU samples in `ScheduledThreadPoolExecutor` frames	10.2%	0.98%
…on the scheduler thread (`DelayedWorkQueue.siftDown`/`compareTo` on a huge queue, plus tracer per-task overhead)	6.6%	0.95%
…on request threads (enqueueing from `dispatchAll()`)	3.5%	~0.01%
CPU samples with `ScheduledDataLoaderRegistry` frames	6.8%	2.1%
Live `ScheduledFutureTask`s after GC, per pod (JFR `ObjectCountAfterGC`)	420k–1.06M	below the histogram's reporting threshold

APM tracers that wrap every scheduled task, such as dd-trace's context propagation, multiply the per-task cost. That makes the effect larger in production than in the benchmark.

What remains on the scheduler thread after the change is the tick itself: iterating the loaders and evaluating shouldDispatch.

Proposed fix

In ticker mode, keep at most one pending tick per registry. Each tick re-arms itself, then calls dispatchAllWithCount(), which still applies every registry-level and per-loader predicate. Non-ticker mode is unchanged.

+import java.util.concurrent.atomic.AtomicBoolean;
 ...
     private volatile boolean closed;
+    private final AtomicBoolean tickScheduled = new AtomicBoolean();
 ...
     public void rescheduleNow() {
-        dataLoaders.forEach(this::reschedule);
+        if (tickerMode) {
+            scheduleTick();
+        } else {
+            dataLoaders.forEach(this::reschedule);
+        }
     }
 ...
     private int dispatchOrReschedule(String key, DataLoader<?, ?> dataLoader) {
         ...
-        if (tickerMode || !shouldDispatch) {
+        if (tickerMode) {
+            scheduleTick();
+        } else if (!shouldDispatch) {
             reschedule(key, dataLoader);
         }
         return sum;
     }
+
+    private void scheduleTick() {
+        if (!closed && tickScheduled.compareAndSet(false, true)) {
+            scheduledExecutorService.schedule(this::tick, schedule.toMillis(), TimeUnit.MILLISECONDS);
+        }
+    }
+
+    private void tick() {
+        tickScheduled.set(false);
+        // re-arm before dispatching so that a data loader throwing on dispatch cannot stop the ticker
+        scheduleTick();
+        dispatchAllWithCount();
+    }

Behaviour this preserves:

In ticker mode, after the first dispatchAll(), every loader whose predicate passes is dispatched at least every schedule until close().
close() stops further ticks. The pending tick runs once more, sees closed and does not re-arm. An explicit dispatchAll() after close still dispatches.
In non-ticker mode, per-loader rescheduling continues only while the predicate says no, exactly as today.

We have run this patched class in production since 2026-10-15. Tests cover:

one pending task after many dispatchAll() and rescheduleNow() calls;
a load issued after dispatchAll() completing on the next tick;
the ticker surviving a predicate that throws during a tick;
close() stopping the ticker while dispatchAll() still dispatches.

Happy to open a PR with these tests if this direction works for you.

Navigation Menu

Repository navigation

ScheduledDataLoaderRegistry ticker mode schedules one task per data loader on every dispatchAll(), so pending tasks grow with loaders × dispatch calls #290

Summary

Where

Reproducer

End-to-end with graphql-java

Production

Proposed fix

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

ScheduledDataLoaderRegistry ticker mode schedules one task per data loader on every dispatchAll(), so pending tasks grow with loaders × dispatch calls #290

Description

Summary

Where

Reproducer

End-to-end with graphql-java

Production

Proposed fix

Activity

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions