Repository: kagkarlsson/db-scheduler (java)

Issue:
Flag to keep scheduler busy when there are many small one-time tasks?
Hi!

I'm running into an issue with low throughput of tasks when we have many one-time tasks.

My use-case is as follows: every morning we want to send a certain set of users an e-mail. As an experiment with this library, I now schedule a new one-time task instance with the user id. This is a nice design as it makes retries possible in case of temporary failure (e.g. transactional e-mail provider has temporary outage), also across restarts of the application.

In testing we now learned that (with 2 threads and a 10 second polling interval), we can send 2 emails every 10 seconds.

What I expected was that the scheduler would keep processing tasks and only start polling at 10 second intervals once there was no job to process right away.
Now it essentially runs two one-time tasks within a second and then sits there waiting for 9+ seconds.

It would make sense (to me) from a throughput perspective to poll the queue for new work when we complete, and if there is none, then we wait for the configured poll duration.
E.g. a flag like ```.pollAfterCompletion(boolean)```.

Alternatively I can lower the polling duration (e.g. to 250ms), but that means we are polling very frequently when there is nothing to be done.
Increasing the number of threads is not a solution for us as this leads to other issues (max number of simultaneous connections we can make to third party providers).

What are your thoughts on this?


Discussion:
Will have to check the code, will get back to you
Yes you are correct. I agree that the ideal behavior is for the scheduler to keep executing until there are no executions left. The current implementation is a bit naive.

Probably no quick-fix, but needs to be fixed..
Well, of course it depends on how others are using this library (including you).

We are testing it as a simple work queue, as we don't have workloads which warrant external work queues and this library is nice and lightweight 👍 

I just had another look at the source code myself and a first idea for doing this is as follows:

In Scheduler.releaseExecutor:
```
if(flag pollAfterCompletion was set on builder) {
    if(currentlyProcessing.isEmpty()) {
        triggerCheckForDueExecutions();
    }
}
```

This way the queue is drained of work as fast as possible, and the polling interval is used when there is no work (at this time).
PS Another topic when having a queue full of one-time tasks is that it's a bit wasteful to pull down up to 10.000 tasks from the database every time we poll (JdbcTaskRepository.MAX_RESULTS).
For our workload a much lower number would work as well (due to limited concurrency/competition between schedulers). Are you open to making this configurable with a builder method as well? E.g. ```pollingLimit``` (with a default of ```10000```).
If you are open to it, I can create pull requests.
Please do :)

tir. 18. des. 2018 kl. 16:56 skrev Sebastian Gooren <
notifications@github.com>:

> If you are open to it, I can create pull requests.
>
> —
> You are receiving this because you commented.
>
>
> Reply to this email directly, view it on GitHub
> <https://github.com/kagkarlsson/db-scheduler/issues/46#issuecomment-448269834>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AAKiHJI-Sc0dulrzIq3UsxsP9WOwjORzks5u6RAdgaJpZM4ZYaCB>
> .
>


Patch:
diff --git a/src/main/java/com/github/kagkarlsson/scheduler/Scheduler.java b/src/main/java/com/github/kagkarlsson/scheduler/Scheduler.java
--- a/src/main/java/com/github/kagkarlsson/scheduler/Scheduler.java
+++ b/src/main/java/com/github/kagkarlsson/scheduler/Scheduler.java
@@ -45,6 +45,8 @@ public class Scheduler implements SchedulerClient {
 	private final ExecutorService executorService;
 	private final Waiter executeDueWaiter;
 	private boolean enableImmediateExecution;
+	private final boolean pollAfterCompletion;
+	private boolean lastPollAcquiredAllAvailableExecutors;
 	protected final List<OnStartup> onStartup;
 	private final Waiter detectDeadWaiter;
 	private final Duration heartbeatInterval;
@@ -59,8 +61,8 @@ public class Scheduler implements SchedulerClient {
 	final Semaphore executorsSemaphore;
 
 	Scheduler(Clock clock, TaskRepository taskRepository, TaskResolver taskResolver, int maxThreads, SchedulerName schedulerName,
-			  Waiter executeDueWaiter, Duration updateHeartbeatWaiter, boolean enableImmediateExecution, StatsRegistry statsRegistry, int pollingLimit, List<OnStartup> onStartup) {
-		this(clock, taskRepository, taskResolver, maxThreads, defaultExecutorService(maxThreads), schedulerName, executeDueWaiter, updateHeartbeatWaiter, enableImmediateExecution, statsRegistry, pollingLimit, onStartup);
+			  Waiter executeDueWaiter, Duration updateHeartbeatWaiter, boolean enableImmediateExecution, boolean pollAfterCompletion, StatsRegistry statsRegistry, int pollingLimit, List<OnStartup> onStartup) {
+		this(clock, taskRepository, taskResolver, maxThreads, defaultExecutorService(maxThreads), schedulerName, executeDueWaiter, updateHeartbeatWaiter, enableImmediateExecution, pollAfterCompletion, statsRegistry, pollingLimit, onStartup);
 	}
 
 	private static ExecutorService defaultExecutorService(int maxThreads) {
@@ -68,13 +70,14 @@ private static ExecutorService defaultExecutorService(int maxThreads) {
 	}
 
 	protected Scheduler(Clock clock, TaskRepository taskRepository, TaskResolver taskResolver, int maxThreads, ExecutorService executorService, SchedulerName schedulerName,
-			  Waiter executeDueWaiter, Duration heartbeatInterval, boolean enableImmediateExecution, StatsRegistry statsRegistry, int pollingLimit, List<OnStartup> onStartup) {
+			  Waiter executeDueWaiter, Duration heartbeatInterval, boolean enableImmediateExecution, boolean pollAfterCompletion, StatsRegistry statsRegistry, int pollingLimit, List<OnStartup> onStartup) {
 		this.clock = clock;
 		this.taskRepository = taskRepository;
 		this.taskResolver = taskResolver;
 		this.executorService = executorService;
 		this.executeDueWaiter = executeDueWaiter;
 		this.enableImmediateExecution = enableImmediateExecution;
+		this.pollAfterCompletion = pollAfterCompletion;
 		this.onStartup = onStartup;
 		this.detectDeadWaiter = new Waiter(heartbeatInterval.multipliedBy(2), clock);
 		this.heartbeatInterval = heartbeatInterval;
@@ -185,11 +188,12 @@ protected void executeDue() {
 		Instant now = clock.now();
 		List<Execution> dueExecutions = taskRepository.getDue(now, pollingLimit);
 
-		int count = 0;
+		int pickedExecutionCount = 0;
+		lastPollAcquiredAllAvailableExecutors = false;
 		LOG.trace("Found {} taskinstances due for execution", dueExecutions.size());
 		for (Execution e : dueExecutions) {
 			if (schedulerState.isShuttingDown()) {
-				LOG.info("Scheduler has been shutdown. Skipping {} due executions.", dueExecutions.size() - count);
+				LOG.info("Scheduler has been shutdown. Skipping {} due executions.", dueExecutions.size() - pickedExecutionCount);
 				return;
 			}
 
@@ -197,7 +201,8 @@ protected void executeDue() {
 			try {
 				pickedExecution = aquireExecutorAndPickExecution(e);
 			} catch (NoAvailableExecutors ex) {
-				LOG.debug("No available executors. Skipping {} due executions.", dueExecutions.size() - count);
+				LOG.debug("No available executors. Skipping {} due executions.", dueExecutions.size() - pickedExecutionCount);
+				lastPollAcquiredAllAvailableExecutors = true;
 				return;
 			}
 
@@ -208,7 +213,7 @@ protected void executeDue() {
 			} else {
 				LOG.debug("Execution picked by another scheduler. Continuing to next due execution.");
 			}
-			count++;
+			pickedExecutionCount++;
 		}
 	}
 
@@ -239,6 +244,10 @@ private void releaseExecutor(Execution execution) {
 			LOG.error("Released execution was not found in collection of executions currently being processed. Should never happen.");
 			statsRegistry.registerUnexpectedError();
 		}
+
+		if (pollAfterCompletion && lastPollAcquiredAllAvailableExecutors) {
+			triggerCheckForDueExecutions();
+		}
 	}
 
 	@SuppressWarnings({"rawtypes","unchecked"})
@@ -410,6 +419,7 @@ public static class Builder {
 		protected Serializer serializer = Serializer.DEFAULT_JAVA_SERIALIZER;
 		protected String tableName = JdbcTaskRepository.DEFAULT_TABLE_NAME;
 		protected boolean enableImmediateExecution = false;
+		protected boolean pollAfterCompletion = true;
 
 		public Builder(DataSource dataSource, List<Task<?>> knownTasks) {
 			this.dataSource = dataSource;
@@ -485,11 +495,16 @@ public Builder enableImmediateExecution() {
 			return this;
 		}
 
+		public Builder pollAfterCompletion(boolean pollAfterCompletion) {
+			this.pollAfterCompletion = pollAfterCompletion;
+			return this;
+		}
+
 		public Scheduler build() {
 			final TaskResolver taskResolver = new TaskResolver(knownTasks);
 			final JdbcTaskRepository taskRepository = new JdbcTaskRepository(dataSource, tableName, taskResolver, schedulerName, serializer);
 
-			return new Scheduler(clock, taskRepository, taskResolver, executorThreads, schedulerName, waiter, heartbeatInterval, enableImmediateExecution, statsRegistry, pollingLimit, startTasks);
+			return new Scheduler(clock, taskRepository, taskResolver, executorThreads, schedulerName, waiter, heartbeatInterval, enableImmediateExecution, pollAfterCompletion, statsRegistry, pollingLimit, startTasks);
 		}
 	}
 
diff --git a/src/main/java/com/github/kagkarlsson/scheduler/testhelper/ManualScheduler.java b/src/main/java/com/github/kagkarlsson/scheduler/testhelper/ManualScheduler.java
--- a/src/main/java/com/github/kagkarlsson/scheduler/testhelper/ManualScheduler.java
+++ b/src/main/java/com/github/kagkarlsson/scheduler/testhelper/ManualScheduler.java
@@ -29,8 +29,8 @@ public class ManualScheduler extends Scheduler {
     private static final Logger LOG = LoggerFactory.getLogger(ManualScheduler.class);
     private final SettableClock clock;
 
-    ManualScheduler(SettableClock clock, TaskRepository taskRepository, TaskResolver taskResolver, int maxThreads, ExecutorService executorService, SchedulerName schedulerName, Waiter waiter, Duration heartbeatInterval, boolean executeImmediately, StatsRegistry statsRegistry, int pollingLimit, List<OnStartup> onStartup) {
-        super(clock, taskRepository, taskResolver, maxThreads, executorService, schedulerName, waiter, heartbeatInterval, executeImmediately, statsRegistry, pollingLimit, onStartup);
+    ManualScheduler(SettableClock clock, TaskRepository taskRepository, TaskResolver taskResolver, int maxThreads, ExecutorService executorService, SchedulerName schedulerName, Waiter waiter, Duration heartbeatInterval, boolean executeImmediately, boolean pollAfterCompletion, StatsRegistry statsRegistry, int pollingLimit, List<OnStartup> onStartup) {
+        super(clock, taskRepository, taskResolver, maxThreads, executorService, schedulerName, waiter, heartbeatInterval, executeImmediately, pollAfterCompletion, statsRegistry, pollingLimit, onStartup);
         this.clock = clock;
     }
 


Test patch:
diff --git a/src/main/java/com/github/kagkarlsson/scheduler/testhelper/TestHelper.java b/src/main/java/com/github/kagkarlsson/scheduler/testhelper/TestHelper.java
--- a/src/main/java/com/github/kagkarlsson/scheduler/testhelper/TestHelper.java
+++ b/src/main/java/com/github/kagkarlsson/scheduler/testhelper/TestHelper.java
@@ -54,7 +54,7 @@ public ManualScheduler build() {
             final TaskResolver taskResolver = new TaskResolver(knownTasks);
             final JdbcTaskRepository taskRepository = new JdbcTaskRepository(dataSource, tableName, taskResolver, schedulerName, serializer);
 
-            return new ManualScheduler(clock, taskRepository, taskResolver, executorThreads, new DirectExecutorService(), schedulerName, waiter, heartbeatInterval, enableImmediateExecution, statsRegistry, pollingLimit, startTasks);
+            return new ManualScheduler(clock, taskRepository, taskResolver, executorThreads, new DirectExecutorService(), schedulerName, waiter, heartbeatInterval, enableImmediateExecution, pollAfterCompletion, statsRegistry, pollingLimit, startTasks);
         }
 
         public ManualScheduler start() {
diff --git a/src/test/java/com/github/kagkarlsson/scheduler/DeadExecutionsTest.java b/src/test/java/com/github/kagkarlsson/scheduler/DeadExecutionsTest.java
--- a/src/test/java/com/github/kagkarlsson/scheduler/DeadExecutionsTest.java
+++ b/src/test/java/com/github/kagkarlsson/scheduler/DeadExecutionsTest.java
@@ -57,6 +57,7 @@ public void setUp() {
 				new Waiter(Duration.ZERO),
 				Duration.ofMinutes(1),
 				false,
+				false,
 				StatsRegistry.NOOP,
 				POLLING_LIMIT,
 				new ArrayList<>());
diff --git a/src/test/java/com/github/kagkarlsson/scheduler/SchedulerTest.java b/src/test/java/com/github/kagkarlsson/scheduler/SchedulerTest.java
--- a/src/test/java/com/github/kagkarlsson/scheduler/SchedulerTest.java
+++ b/src/test/java/com/github/kagkarlsson/scheduler/SchedulerTest.java
@@ -43,7 +43,7 @@ private Scheduler schedulerFor(Task<?> ... tasks) {
 	private Scheduler schedulerFor(ExecutorService executor, Task<?> ... tasks) {
 		TaskResolver taskResolver = new TaskResolver(tasks);
 		JdbcTaskRepository taskRepository = new JdbcTaskRepository(postgres.getDataSource(), DEFAULT_TABLE_NAME, taskResolver, new SchedulerName.Fixed("scheduler1"));
-		return new Scheduler(clock, taskRepository, taskResolver, 1, executor, new SchedulerName.Fixed("name"), new Waiter(Duration.ZERO), Duration.ofSeconds(1), false, StatsRegistry.NOOP, 10_000, new ArrayList<>());
+		return new Scheduler(clock, taskRepository, taskResolver, 1, executor, new SchedulerName.Fixed("name"), new Waiter(Duration.ZERO), Duration.ofSeconds(1), false, false, StatsRegistry.NOOP, 10_000, new ArrayList<>());
 	}
 
 	@Test