Make ScheduledEventExecutor task scheduler pluggable

[injae-kim]

Motivation:

Netty currently only provides a priority queue based task scheduler.
We could add a small abstraction layer between EventExecutor and a task scheduler implementation so that
a user can choose his or her favorite scheduler algorithm, most notably, hashed wheel timer, which is more scalable.

• On Pluggable event loop task scheduler #8164, we want to add abstraction layer to make ScheduledEvenExecutor's task scheduler pluggable

PRs Overview

1. ✅ (this PR) Make ScheduledEventExecutor task scheduler pluggable by TaskScheduler interface
2. (next PR) Provide a way to plug user's TaskScheduler on ScheduledEventExecutor
3. Provide HashedWheelTimer task scheduler and make it pluggable to ScheduledEventExecutor for user who want to use it

Modification:

• Make ScheduledEventExecutor task scheduler pluggable by TaskScheduler interface

Result:

• Now user can use their custom task scheduler algorithm on ScheduledEventExecutor
• Close Pluggable event loop task scheduler #8164 issue

[normanmaurer]

/cc @trustin

[injae-kim]

• ✅ (this PR) Make ScheduledEventExecutor task scheduler pluggable by ScheduledTaskQueue interface
• (next PR) Provide a way to plug user's ScheduledTaskQueue on ScheduledEventExecutor
• Provide HashedWheelTimer task scheduler and make it pluggable to ScheduledEventExecutor

Hi @He-Pin , @trustin , can you review this PR? 🙇
Note that I'll make 2 more PRs after this PR is merged to close issue~!

[He-Pin]

How about make it a factory?

[injae-kim]

oh it's nice! ✅ I introduced TaskSchedulerFactory.

[He-Pin]

@injae-kim I would expect a Scheduler interface which defines methods like scheduleAtFixedRate., and the ScheduledTaskQueue may be just an implementation detail?

[injae-kim]

interface TaskScheduler
    |- AbstractTaskScheduler
        |- DefaultTaskScheduler

interface TaskSchedulerFactory
    |- DefaultTaskSchedulerFactory  

aha I understood your point. ✅ I introduced TaskScheduler interface!

[injae-kim]

Gentle ping @He-Pin , could you review this PR? I need this feature too! thanks a lot for your help 🙇

[BitoAgent]

Code Review Agent Run Status

• AI Based Review: Successful

Code Review Overview

• Summary: The PR introduces a significant refactor in the task scheduling mechanism of the AbstractScheduledEventExecutor, transitioning from a priority queue-based approach to a pluggable TaskScheduler interface. This change allows for more flexible scheduling strategies, including the potential for more efficient implementations. The addition of the TaskSchedulerFactory interface further enhances this flexibility by enabling the dynamic creation of TaskScheduler instances. Tests have been updated or added to validate the new functionality, ensuring backward compatibility and adherence to the scheduling contract.
• Code change type: Refactoring, Feature Addition, Tests
• Unit tests added: True
• Estimated effort to review (1-5, lower is better): 2, due to the comprehensive tests provided and clear separation of concerns, making it easier to review the changes.

>>See detailed code suggestions<<
The Bito AI Code Review Agent successfully reviewed 7 files and discovered 10 issues. Please review these issues along with suggested fixes in the Changed Files.

High-level Feedback

The PR is a well-thought-out enhancement to the Netty scheduling mechanism, introducing flexibility and the potential for performance optimizations with custom TaskScheduler implementations. The use of a factory pattern for creating TaskScheduler instances is a good design choice, allowing for future extensions and customizations. The addition of tests is commendable, ensuring that the new functionality does not break existing behavior.

[BitoAgent]

Suggestion: Consider documenting the behavior when both taskScheduler and taskSchedulerFactory are null. It might be beneficial to clarify the expected behavior or fallback mechanism in such cases, ensuring that users of the API have a clear understanding of how task scheduling is handled.
Code Suggestion:

+    /**
+     * Task scheduler instance used for scheduling tasks. If null, the taskSchedulerFactory is used
+     * to create a new instance. If both are null, document the expected behavior or fallback mechanism.
+     */
+    private TaskScheduler taskScheduler;
+    private TaskSchedulerFactory taskSchedulerFactory;

[BitoAgent]

Suggestion: For methods that schedule tasks (e.g., schedule, scheduleAtFixedRate, scheduleWithFixedDelay), consider specifying the behavior or exceptions when scheduling tasks in a shutdown state. This clarification can help ensure consistent usage and error handling.
Code Suggestion:

+    /**
+     * Schedules the specified task for execution after the specified delay.
+     * @param task the task to schedule
+     * @param delay the time from now to delay execution
+     * @param unit the time unit of the delay parameter
+     * @return a Future representing pending completion of the task
+     * @throws RejectedExecutionException if the task cannot be scheduled for execution
+     * @throws IllegalStateException if the scheduler is in a shutdown state
+     */
+    Future<Void> schedule(Runnable command, long delay, TimeUnit unit);

[BitoAgent]

Suggestion: Enhance the test coverage by adding cases that verify the behavior when a TaskScheduler fails to schedule a task, e.g., throwing an exception. This will ensure the robustness of the scheduling mechanism in error scenarios.
Code Suggestion:

+    @Test
+    public void testTaskScheduler_scheduleFailure() {
+        TestScheduledEventExecutor executor = new TestScheduledEventExecutor(
+                executor -> new FailingTaskScheduler(executor));
+        assertThrows(SchedulingException.class, () -> executor.schedule(TEST_RUNNABLE, 0, TimeUnit.NANOSECONDS));
+    }

[BitoAgent]

Scalability Issue: The introduction of TaskScheduler and TaskSchedulerFactory as fields increases the memory footprint of each AbstractScheduledEventExecutor instance. In a highly scalable system, where potentially millions of these executors could be instantiated, this additional memory usage could become significant. Moreover, the dynamic selection of TaskScheduler based on a factory pattern, while providing flexibility, introduces additional overhead during executor instantiation, which could impact the startup time of systems with a large number of executors.
Fix: Consider using a static, shared TaskScheduler instance for common scheduling scenarios, reducing the memory footprint per executor instance. This shared instance could be a highly efficient, multi-tenant scheduler designed to handle tasks from multiple executors concurrently. For use cases requiring specialized scheduling, continue to allow the injection of a custom TaskScheduler, but ensure this is an exceptional scenario rather than the norm.
Code Suggestion:

-    private TaskScheduler taskScheduler;
-    private TaskSchedulerFactory taskSchedulerFactory;
+    private static final TaskScheduler sharedTaskScheduler = DefaultTaskSchedulerFactory.INSTANCE.newTaskScheduler(this);

     protected AbstractScheduledEventExecutor(TaskSchedulerFactory taskSchedulerFactory) {
-        this.taskSchedulerFactory = requireNonNull(taskSchedulerFactory, "taskSchedulerFactory");
+        // Use sharedTaskScheduler for common cases
     }

+    TaskScheduler taskScheduler() {
+        return sharedTaskScheduler;
+    }

[BitoAgent]

Scalability Issue: The DefaultTaskSchedulerFactory creates a new instance of DefaultTaskScheduler for each call to newTaskScheduler. This could lead to unnecessary object creation if many executors share the same scheduling logic, impacting the scalability of the system.
Fix: Implement a caching mechanism within DefaultTaskSchedulerFactory to reuse DefaultTaskScheduler instances where possible. This could be based on identifying common configurations or parameters that lead to equivalent scheduling behavior.
Code Suggestion:

Implement a caching mechanism within DefaultTaskSchedulerFactory to reuse DefaultTaskScheduler instances where possible. This could be based on identifying common configurations or parameters that lead to equivalent scheduling behavior.

protected AbstractScheduledEventExecutor(TaskSchedulerFactory taskSchedulerFactory) {
    this.taskSchedulerFactory = requireNonNull(taskSchedulerFactory, "taskSchedulerFactory");
}

TaskScheduler taskScheduler() {
    if (taskScheduler == null && taskSchedulerFactory != null) {
        taskScheduler = taskSchedulerFactory.newTaskScheduler(this);
    }
    if (taskScheduler == null) {
        taskScheduler = DefaultTaskSchedulerFactory.ISTANCE.newTaskScheduler(this);
    }
    return taskScheduler;
}

[BitoAgent]

Scalability Issue: The modification to use taskScheduler().schedule directly for quietPeriodTask in GlobalEventExecutor could introduce a delay in the execution of this task due to the overhead of task scheduling. In a highly scalable system, even minor delays introduced in core system tasks can amplify, impacting overall system responsiveness and throughput.
Fix: Review the scheduling requirements for quietPeriodTask to determine if it can be executed immediately or if a more efficient scheduling strategy can be applied. For tasks critical to system performance, consider bypassing the task scheduler and executing directly or using a more lightweight scheduling mechanism.
Code Suggestion:

protected AbstractScheduledEventExecutor(TaskSchedulerFactory taskSchedulerFactory) {
    this.taskSchedulerFactory = requireNonNull(taskSchedulerFactory, "taskSchedulerFactory");
}

protected TaskScheduler taskScheduler() {
    if (taskScheduler == null && taskSchedulerFactory != null) {
        taskScheduler = taskSchedulerFactory.newTaskScheduler(this);
    }
    if (taskScheduler == null) {
        taskScheduler = DefaultTaskSchedulerFactory.INSTANCE.newTaskScheduler(this);
    }
    return taskScheduler;
}

private static boolean isNullOrEmpty(TaskScheduler taskScheduler) {
    return taskScheduler == null || taskScheduler.isEmpty();
}

protected final void cancelScheduledTasks() {
    assert inEventLoop();
    TaskScheduler taskScheduler = this.taskScheduler;
    if (isNullOrEmpty(taskScheduler)) {
        return;
    }
    taskScheduler.cancelScheduledTasks();
}

[BitoAgent]

Scalability Issue: The schedule method in AbstractTaskScheduler directly creates a new instance of RunnableScheduledFuture for each scheduled task. This could lead to a high rate of object creation and garbage collection in systems with a high volume of task scheduling operations, impacting scalability and performance.
Fix: Utilize a pool of RunnableScheduledFuture objects instead of instantiating new ones for each task. This object pooling pattern can significantly reduce the rate of object creation and garbage collection, improving the system's scalability and performance.
Code Suggestion:

+    private final ObjectPool<RunnableScheduledFuture<?>> futurePool;
+    public AbstractTaskScheduler(ObjectPool<RunnableScheduledFuture<?>> futurePool) {
+        this.futurePool = futurePool;
+    }

+    @Override
+    public Future<Void> schedule(Runnable command, long delay, TimeUnit unit) {
+        RunnableScheduledFuture<Void> task = futurePool.get();
+        task.reset(command, delay, unit);
+        return schedule(task);
+    }

[BitoAgent]

Scalability Issue: The DefaultTaskScheduler uses a PriorityQueue to manage scheduled tasks, which can become inefficient for a very high number of concurrent tasks due to its O(log n) complexity for insertion and removal operations. This could become a scalability bottleneck in systems with a high volume of scheduled tasks.
Fix: Implement or utilize a more efficient data structure for managing scheduled tasks, such as a time-wheel or a hierarchical timing wheel, which can offer better performance characteristics (O(1) complexity for insertion and removal) for the scheduling use case.
Code Suggestion:

private TaskScheduler taskScheduler;

protected AbstractScheduledEventExecutor() {
}

protected AbstractScheduledEventExecutor(TaskSchedulerFactory taskSchedulerFactory) {
    this.taskSchedulerFactory = requireNonNull(taskSchedulerFactory, "taskSchedulerFactory");
    this.taskScheduler = taskSchedulerFactory.newTaskScheduler(this);
}

TaskScheduler taskScheduler() {
    return this.taskScheduler;
}

[BitoAgent]

Performance Issue: The use of Comparable::compareTo in the DefaultTaskScheduler may not be the most efficient way to compare RunnableScheduledFutureNode instances. This approach relies on the natural ordering of the elements, which might not be optimized for the specific use case of task scheduling, potentially leading to performance degradation when managing large numbers of scheduled tasks.
Fix: Implement a custom Comparator for RunnableScheduledFutureNode instances that is specifically optimized for the task scheduling use case. This custom Comparator should prioritize the most critical factors for task ordering (e.g., execution time) to ensure efficient task scheduling and execution.
Code Suggestion:

private static final Comparator<RunnableScheduledFuture<?>> TASK_SCHEDULER_COMPARATOR = new Comparator<RunnableScheduledFuture<?>>() {
    @Override
    public int compare(RunnableScheduledFuture<?> o1, RunnableScheduledFuture<?> o2) {
        return Long.compare(o1.getDelay(TimeUnit.NANOSECONDS), o2.getDelay(TimeUnit.NANOSECONDS));
    }
};

private PriorityQueue<RunnableScheduledFuture<?>> scheduledTaskQueue = new PriorityQueue<>(TASK_SCHEDULER_COMPARATOR);

[BitoAgent]

Performance Issue: The RunnableScheduledFutureAdapter instantiation within DefaultTaskScheduler's newScheduledTaskFor method doesn't consider the reuse of instances, which might lead to unnecessary object creation and increased garbage collection overhead.
Fix: Implement a mechanism to reuse RunnableScheduledFutureAdapter instances, such as an object pool. This approach will reduce the number of created objects and mitigate the garbage collection impact, especially under high load.
Code Suggestion:

private final ObjectPool<RunnableScheduledFutureAdapter<?>> runnableScheduledFutureAdapterPool = new DefaultObjectPool<>(new ObjectPool.ObjectCreator<RunnableScheduledFutureAdapter<?>>() {
    @Override
    public RunnableScheduledFutureAdapter<?> create() {
        return new RunnableScheduledFutureAdapter<>(DefaultTaskScheduler.this, newPromise(), callable, deadlineNanos, periodNanos);
    }
});

@Override
protected <V> RunnableScheduledFuture<V> newScheduledTaskFor(Callable<V> callable, long deadlineNanos, long periodNanos) {
    return (RunnableScheduledFuture<V>) runnableScheduledFutureAdapterPool.get().init(callable, deadlineNanos, periodNanos);
}