Skip to content

Commit 08cc5c5

Browse files
authored
Merge pull request #291 from Meemaw/ticker-mode-single-pending-task
Keep one pending ticker task per ScheduledDataLoaderRegistry
2 parents 747a7de + 779a6f8 commit 08cc5c5

2 files changed

Lines changed: 115 additions & 3 deletions

File tree

‎src/main/java/org/dataloader/registries/ScheduledDataLoaderRegistry.java‎

Lines changed: 31 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import java.util.concurrent.Executors;
1616
import java.util.concurrent.ScheduledExecutorService;
1717
import java.util.concurrent.TimeUnit;
18+
import java.util.concurrent.atomic.AtomicBoolean;
1819

1920
import static org.dataloader.impl.Assertions.nonNull;
2021

@@ -33,6 +34,8 @@
3334
* no rescheduling will occur, and you will need to call dispatch again to restart the process.
3435
* <p>
3536
* However, when {@link #tickerMode} is true, the registry will always reschedule continuously after the first ever call to {@link #dispatchAll()}.
37+
* A single task per registry re-runs the dispatch of all data loaders every {@link #getScheduleDuration()}, however
38+
* often {@link #dispatchAll()} or {@link #rescheduleNow()} is called.
3639
* <p>
3740
* This will allow you to chain together {@link DataLoader} load calls like this :
3841
* <pre>{@code
@@ -67,6 +70,7 @@ public class ScheduledDataLoaderRegistry extends DataLoaderRegistry implements A
6770
private final Duration schedule;
6871
private final boolean tickerMode;
6972
private volatile boolean closed;
73+
private final AtomicBoolean tickScheduled = new AtomicBoolean();
7074

7175
private ScheduledDataLoaderRegistry(Builder builder) {
7276
super(builder.dataLoaders, builder.instrumentation);
@@ -214,7 +218,11 @@ public int dispatchAllWithCountImmediately() {
214218
* a pre-check of the predicate like {@link #dispatchAll()} would
215219
*/
216220
public void rescheduleNow() {
217-
dataLoaders.forEach(this::reschedule);
221+
if (tickerMode) {
222+
scheduleTick();
223+
} else {
224+
dataLoaders.forEach(this::reschedule);
225+
}
218226
}
219227

220228
/**
@@ -246,12 +254,32 @@ private int dispatchOrReschedule(String key, DataLoader<?, ?> dataLoader) {
246254
if (shouldDispatch) {
247255
sum = dataLoader.dispatchWithCounts().getKeysCount();
248256
}
249-
if (tickerMode || !shouldDispatch) {
257+
if (tickerMode) {
258+
scheduleTick();
259+
} else if (!shouldDispatch) {
250260
reschedule(key, dataLoader);
251261
}
252262
return sum;
253263
}
254264

265+
/**
266+
* In ticker mode a single task per registry re-runs {@link #dispatchAllWithCount()} every {@link #schedule},
267+
* however often {@link #dispatchAll()} is called. Scheduling a task per data loader on every
268+
* {@link #dispatchAll()} call would grow the executor queue by the number of registered data loaders on each call.
269+
*/
270+
private void scheduleTick() {
271+
if (!closed && tickScheduled.compareAndSet(false, true)) {
272+
scheduledExecutorService.schedule(this::tick, schedule.toMillis(), TimeUnit.MILLISECONDS);
273+
}
274+
}
275+
276+
private void tick() {
277+
tickScheduled.set(false);
278+
// re-arm before dispatching so that a data loader throwing on dispatch cannot stop the ticker
279+
scheduleTick();
280+
dispatchAllWithCount();
281+
}
282+
255283
/**
256284
* By default, this will create use a {@link Executors#newSingleThreadScheduledExecutor()}
257285
* and a schedule duration of 10 milliseconds.
@@ -349,7 +377,7 @@ public Builder dispatchPredicate(DispatchPredicate dispatchPredicate) {
349377
/**
350378
* This sets ticker mode on the registry. When ticker mode is true the registry will
351379
* continuously reschedule the data loaders for possible dispatching after the first call
352-
* to dispatchAll.
380+
* to dispatchAll, using a single scheduled task per registry.
353381
*
354382
* @param tickerMode true or false
355383
* @return this builder for a fluent pattern

‎src/test/java/org/dataloader/registries/ScheduledDataLoaderRegistryTest.java‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,16 @@
1515
import java.util.concurrent.CompletableFuture;
1616
import java.util.concurrent.Executors;
1717
import java.util.concurrent.ScheduledExecutorService;
18+
import java.util.concurrent.ScheduledThreadPoolExecutor;
19+
import java.util.concurrent.TimeUnit;
1820
import java.util.concurrent.atomic.AtomicBoolean;
1921
import java.util.concurrent.atomic.AtomicInteger;
2022

2123
import static java.util.Arrays.asList;
2224
import static java.util.Collections.singletonList;
2325
import static org.awaitility.Awaitility.await;
2426
import static org.awaitility.Duration.TWO_SECONDS;
27+
import static org.dataloader.fixtures.TestKit.idLoader;
2528
import static org.dataloader.fixtures.TestKit.snooze;
2629
import static org.hamcrest.MatcherAssert.assertThat;
2730
import static org.hamcrest.Matchers.equalTo;
@@ -350,6 +353,87 @@ public void chain_data_loaders_will_hang_if_not_in_ticker_mode(TestDataLoaderFac
350353
registry.close();
351354
}
352355

356+
@Test
357+
public void ticker_mode_keeps_one_pending_task_however_often_dispatch_all_is_called() {
358+
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
359+
// a tick that never fires during the test, so the queue is only inspected between ticks
360+
ScheduledDataLoaderRegistry registry = tickerRegistry(executor, Duration.ofHours(1), 20, alwaysDispatch);
361+
362+
for (int i = 0; i < 50; i++) {
363+
registry.dispatchAll();
364+
}
365+
366+
assertThat(executor.getQueue().size(), equalTo(1));
367+
registry.close();
368+
executor.shutdownNow();
369+
}
370+
371+
@Test
372+
public void ticker_mode_reschedule_now_keeps_one_pending_task() {
373+
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
374+
ScheduledDataLoaderRegistry registry = tickerRegistry(executor, Duration.ofHours(1), 20, alwaysDispatch);
375+
registry.dispatchAll();
376+
377+
for (int i = 0; i < 50; i++) {
378+
registry.rescheduleNow();
379+
}
380+
381+
assertThat(executor.getQueue().size(), equalTo(1));
382+
registry.close();
383+
executor.shutdownNow();
384+
}
385+
386+
@Test
387+
public void ticker_mode_keeps_ticking_after_a_dispatch_predicate_throws() throws Exception {
388+
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
389+
AtomicBoolean throwOnNextCheck = new AtomicBoolean();
390+
DispatchPredicate throwingOnce = (key, dl) -> {
391+
if (throwOnNextCheck.getAndSet(false)) {
392+
throw new IllegalStateException("predicate failure");
393+
}
394+
return true;
395+
};
396+
ScheduledDataLoaderRegistry registry = tickerRegistry(executor, Duration.ofMillis(10), 1, throwingOnce);
397+
registry.dispatchAll();
398+
throwOnNextCheck.set(true);
399+
400+
DataLoader<String, String> loader = registry.getDataLoader("loader-0");
401+
CompletableFuture<String> late = loader.load("late");
402+
403+
assertThat(late.get(2, TimeUnit.SECONDS), equalTo("late"));
404+
registry.close();
405+
executor.shutdownNow();
406+
}
407+
408+
@Test
409+
public void close_stops_the_ticker_but_dispatch_all_still_dispatches() throws Exception {
410+
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
411+
ScheduledDataLoaderRegistry registry = tickerRegistry(executor, Duration.ofMillis(10), 1, alwaysDispatch);
412+
registry.dispatchAll();
413+
registry.close();
414+
await().atMost(TWO_SECONDS).until(() -> executor.getQueue().isEmpty());
415+
416+
DataLoader<String, String> loader = registry.getDataLoader("loader-0");
417+
CompletableFuture<String> afterClose = loader.load("after-close");
418+
registry.dispatchAll();
419+
420+
assertThat(afterClose.getNow(null), equalTo("after-close"));
421+
assertThat(executor.getQueue().size(), equalTo(0));
422+
executor.shutdownNow();
423+
}
424+
425+
private ScheduledDataLoaderRegistry tickerRegistry(ScheduledExecutorService executor, Duration schedule, int loaderCount, DispatchPredicate loaderPredicate) {
426+
ScheduledDataLoaderRegistry.Builder builder = ScheduledDataLoaderRegistry.newScheduledRegistry()
427+
.scheduledExecutorService(executor)
428+
.tickerMode(true)
429+
.schedule(schedule)
430+
.dispatchPredicate(neverDispatch);
431+
for (int i = 0; i < loaderCount; i++) {
432+
builder.register("loader-" + i, idLoader("loader-" + i), loaderPredicate);
433+
}
434+
return builder.build();
435+
}
436+
353437
@Test
354438
public void executors_are_shutdown() {
355439
ScheduledDataLoaderRegistry registry = ScheduledDataLoaderRegistry.newScheduledRegistry().build();

0 commit comments

Comments
 (0)