Understanding the Need for a Custom Scheduler

Why default scheduling mechanisms sometimes fall short and when a custom solution is warranted.

There are many situations where we need to schedule tasks to trigger after a delay or run periodically. We could simply make the current thread wait, but how can we do this for multiple actions without creating an excessive number of threads? In such cases, we might turn to third-party libraries, but these often come with unnecessary functionality.

Therefore, we sometimes need to implement our own solution tailored to our specific needs, avoiding unnecessary features. Additionally, this allows us to manage CPU and memory usage more effectively.

Core Principles of Task Scheduling

When building a custom scheduler for handling multiple tasks, understanding the core principles of task scheduling is essential. These principles ensure that your scheduler manages resources efficiently while maintaining high performance and reliability. Here’s a detailed look at these principles:

1. Task Prioritization and Dependencies

• Prioritization: Each task is assigned a priority level. The scheduler uses these priorities to decide the order of task execution. High-priority tasks are executed before lower-priority ones to ensure critical operations complete sooner.
• Dependencies: Some tasks cannot start until others finish. Identifying and managing these dependencies is crucial to prevent deadlocks and ensure smooth execution. A common approach is to use a directed acyclic graph (DAG) where each node represents a task, and edges denote dependency constraints.

2. Concurrency Management

• Parallelism: This involves running multiple tasks simultaneously, especially on systems with multiple processing cores. Proper management of parallelism can significantly improve the performance of your application.
• Throttling: Limiting the number of tasks that run concurrently can prevent resource exhaustion. Throttling is vital in environments with limited CPU, memory, or I/O capacity, ensuring the system remains stable and responsive.
• Thread Pooling: Using a pool of pre-initialized threads can reduce the overhead of creating and destroying threads. A scheduler can manage a thread pool to serve incoming tasks efficiently.

3. Resource Allocation

• Load Balancing: Distributing tasks across available resources to optimize utilization. Effective load balancing prevents certain nodes or processors from being overwhelmed, thereby enhancing overall efficiency.
• Resource Awareness: The scheduler should consider the resource needs of each task. For example, memory-intensive tasks might need to be spaced out or run on specific nodes with enough RAM.

4. Scheduling Algorithms

• Round-Robin: This simple algorithm cycles through tasks, assigning a fixed time slice to each. It’s fair but doesn’t account for task priority or resource needs.
• Shortest Job Next (SJN): This algorithm prefers tasks that are estimated to complete quickly, reducing the average time to complete tasks.
• Earliest Deadline First (EDF): Prioritizes tasks based on their deadlines. This is particularly useful for real-time applications where meeting deadline constraints is crucial.

5. Fault Tolerance and Error Handling

• Retry Mechanisms: The scheduler should be capable of retrying tasks that fail due to transient errors.
• Error Propagation: Proper mechanisms should be in place to log and notify developers of task failures, without letting these failures crash the entire scheduling system.
• Backup and Recovery: In critical systems, the scheduler should include strategies for task checkpointing and recovery, enabling it to restart from a known good state in case of failure.

6. Scalability and Adaptability

• Scalability: The scheduler should perform well as the number of tasks grows. This may involve dynamically adjusting resources or parallelism levels based on the workload.
• Adaptability: In changing conditions, such as varying network latency or CPU load, the scheduler might need to adjust its behavior dynamically. Techniques like feedback control can be used to adjust scheduling parameters in real-time.

7. Monitoring and Maintenance

• Logging and Auditing: Keeping detailed logs helps in debugging and understanding the behavior of the scheduler over time.
• Performance Metrics: Implementing monitoring of key performance indicators (KPIs) like task latency, throughput, and resource utilization helps in tuning the scheduler.

By understanding and implementing these core principles, developers can create robust, efficient, and flexible schedulers tailored to their specific application needs. This ensures that tasks are managed in a way that optimizes performance while adhering to operational constraints and requirements.

Designing the Scheduler Architecture

When designing a custom scheduler for handling multiple tasks, the architecture plays a crucial role in determining how effectively and efficiently the scheduler operates. A well-designed scheduler architecture balances load, manages resources wisely, and adapts dynamically to varying workloads. Below, I break down the key components of a scheduler’s architecture and how they interact.

Key Components of Scheduler Architecture

1. Task Queue
   • Purpose: This is where tasks are held before they are processed. It manages the incoming flow of tasks and serves as a buffer to decouple task submission from execution.
   • Characteristics: The task queue can prioritize tasks based on various criteria (e.g., priority level, deadlines). It might use different data structures like priority queues, heaps, or custom lists depending on the scheduling strategy.
2. Scheduler Core
   • Purpose: This component is the heart of the scheduler. It decides which task to execute next based on the scheduling algorithms and the state of the system.
   • Characteristics: The scheduler core uses algorithms like Round-Robin, Shortest Job Next, or Earliest Deadline First to make decisions. It also handles dependency resolution and task prioritization.
3. Worker Threads / Execution Units
   • Purpose: These are the threads or processes that actually execute the tasks. In a multi-threaded environment, these might be a fixed pool of worker threads that pick up tasks from the queue.
   • Characteristics: Worker threads manage task execution and report back on task completion or failure. They are often organized in a thread pool to minimize the overhead of creating and destroying threads.
4. Resource Manager
   • Purpose: Manages the allocation and monitoring of resources like CPU, memory, and I/O. It ensures that tasks have the necessary resources to run efficiently.
   • Characteristics: The resource manager tracks resource usage and might implement throttling or load balancing to prevent resource exhaustion and ensure fair use.
5. Dispatcher
   • Purpose: Acts as a coordinator between the task queue, the scheduler core, and the worker threads. It dispatches tasks to available workers based on the strategy defined by the scheduler core.
   • Characteristics: The dispatcher handles the logistics of task assignment, ensuring that resource constraints and task dependencies are respected.
6. Error Handler
   • Purpose: Manages errors and exceptions from task executions. It decides whether to retry a task, escalate the issue, or log the error based on predefined policies.
   • Characteristics: This component enhances the fault tolerance of the scheduler by providing robust mechanisms for dealing with failed tasks.
7. Monitor and Logger
   • Purpose: Tracks the performance and activities of the scheduler. It logs events and gathers statistics for analysis.
   • Characteristics: This might include real-time monitoring tools, logging systems, and performance metrics tracking. It’s essential for maintaining visibility into the scheduler’s operation and for debugging issues.

Building custom scheduler on C#

For now, let’s skip task priority and use a simple algorithm to build the scheduler. I will use the Round-Robin algorithm to iterate through all tasks, which will be grouped by their trigger interval in seconds.

First, since we will be using our scheduler from multiple threads, let’s create a concurrent dictionary to store all tasks. The key of the dictionary will be an integer representing seconds, which will allow us to group actions by their scheduled times. Additionally, we need a counter that increments while the scheduler is running to calculate the exact time when each task should run.

private static int _lifeTimeCounter;
private readonly static ConcurrentDictionary<int, List<Action>> _dictionary = new ConcurrentDictionary<int, List<Action>>();

This is a simple representation of all actions, grouped by key, and we made it static to make usage even easier. Let’s start with this and optimize it later.

Now, let’s create a method to schedule each action one by one.

public static void ScheduleJob(int seconds, Action action)
{
    if (!_dictionary.ContainsKey(seconds))
        _dictionary.TryAdd(seconds, new List<Action>());

    _dictionary[seconds].Add(action);
}

What are we doing here? Since we are grouping by seconds, we check if our dictionary contains the incoming seconds as a key. If not, we add that key and the first action for those seconds.

To handle our tasks based on their scheduled seconds, we need an infinite loop that iterates through our hashtable and checks if it is time to trigger the collection of actions related to a particular key. Additionally, let’s create a public property to allow us to stop the scheduler at any moment.

public static bool IsCanceled { get; set; }

private static void RunScheduler()
{
    while (!IsCanceled)
    {
        if (!_dictionary.IsEmpty)
        {
            foreach (var pair in _dictionary)
            {
                if (_lifeTimeCounter % pair.Key == 0)
                    pair.Value.ForEach(a => a());
            }
        }

        Thread.Sleep(1000);
        _lifeTimeCounter++;
    }
}

We are iterating through all items in our dictionary and using the modulus operator to check if it is time to trigger the current action. For example, if _lifeTimeCounter = 24 and pair.Key = 6, then 24 % 6 will be 0, and the actions will trigger. However, when _lifeTimeCounter becomes 25, the result will not equal to 0, so none of the actions will trigger.

Now we have the necessary minimum to schedule our tasks, but how do we run it? Let’s do this in a static constructor. It should run in a parallel thread; otherwise, it will block the caller thread, which could be problematic, especially if the caller thread is the main thread.

static GlobalScheduler()
{
    Task.Factory.StartNew(() => RunScheduler());
}

We have now completed the minimum setup needed to work with the scheduler.

Since everything is created as static, using the scheduler is quite simple. Let’s add a simple action to run every 4 seconds.

GlobalScheduler.ScheduleJob(4, () => Console.WriteLine(DateTime.Now.ToString("HH:mm:ss")));

The result here is predictable and straightforward.

23:33:05
23:33:09
23:33:13
23:33:17
23:33:21
23:33:25
23:33:29
23:33:33

Optimization Strategies

We have implemented a simple strategy based on cycling through all tasks, but what kind of problems might we face now? Let’s consider some “What if’s…”:

• What if we schedule different tasks at different times while our global counter has a non-zero value?
• What if some tasks throw an exception?
• What if some of the scheduled tasks take a long time to process?
• What if we need to schedule a job for a one-time use, such as “triggering in advance”?
• What if there is a need to pass parameters to the action?

These questions represent just the tip of the iceberg in a vast ocean of potential issues.

Taking into account that we have only one counter for all tasks, we will face issues when multiple tasks are scheduled at different times. For example, if we schedule task 1 to trigger every 3 seconds, and 6 seconds later we schedule task 2 to trigger every 7 seconds, our global counter will be at 6. This means in the next second, it will trigger task 2, which breaks the initial scheduling logic.

To fix this issue, we need to either adjust task scheduling to account for the current counter value or have a separate counter for each job. I personally prefer the second approach. Therefore, let’s create an object to handle each job’s counter separately and modify our collection to store Job object, instead of Action

private readonly static ConcurrentDictionary<int, List<Job>> _dictionary = new ConcurrentDictionary<int, List<Job>>();

private class Job
{
    public Action Action { get; set; }
    public int Counter { get; set; }
}

I’ve created a private Job object and encapsulated it within our scheduler to prevent global usage.

Now that we have a separate object for each job, we should increment its counter each time and handle it properly in the scheduler runner.

Let’s modify the ScheduleJob and RunScheduler methods:

_dictionary[seconds].Add(new Job { Action = action });

foreach (var pair in _dictionary)
{
    for (int i = pair.Value.Count - 1; i >= 0; i--)
    {
        Job value = pair.Value[i];
        value.Counter++;

        if (value.Counter % pair.Key == 0)
            value.Action();
    }
}

Now we can rely on the scheduled time for each job, regardless of when it was scheduled. Let’s move to the next point.

To prevent our scheduler from crashing, we can do one of the following:

1. Put the job action call inside a try...catch block and handle all unhandled exceptions in the catch statement, such as logging the error.
2. Wrap every action in its own Action during the job scheduling process.
3. Run each job’s action inside its own Task, where any unhandled exceptions will be handled by the parallel thread.

If we have a logging system, handling all unhandled exceptions is a good idea. I would definitely put the action call inside a try...catch block. However, this is not enough when we have long-running tasks that can block the scheduler, leading to delays in other tasks. To handle this, we can:

• Implement timeouts for tasks to ensure they don’t exceed a predefined execution time.
• Use separate queues for long-running and short-running tasks to prevent blockage.

We can implement both approaches, but it is very difficult to predict in advance which scheduled jobs will be long-running. For instance, a short-running job can become long-running, such as an I/O operation that initially reads/writes to an almost empty file but takes much longer as the file size increases. So, why not go with a simpler solution?

Let’s run each action in its own Task and catch any unhandled exceptions inside it.

if (value.Counter % pair.Key == 0)
{
    Task.Run(() =>
    {
        try
        {
            value.Action();
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.ToString());
        }
    });
}

With the current implementation, we can even pass a timeout and a CancellationToken to cancel the task at any time. But for now, let’s leave it as is.

There are many situations where we need to schedule a job to run only once or pass a parameter to the action that we are going to run. Our scheduler currently does not support one-time runs or parameters, as we are using a simple Action class. We can modify our Job class to support a generic Action<T>, but considering that we don’t always need a parameter, it is simpler to create separate handling for parameterized actions.

To store all jobs with parameters, we need a separate collection. Additionally, let’s add a new property to handle “advanced triggered” jobs.

private readonly static ConcurrentDictionary<int, List<JobWithParam>> _dictionaryWithParam = new ConcurrentDictionary<int, List<JobWithParam>>();

private class Job
{
    public Action Action { get; set; }
    public bool HandleOnce { get; set; }
    public int Counter { get; set; }
    public bool IsInProgress { get; set; }
}

private class JobWithParam
{
    public Action<object> Action { get; set; }
    public object Parameter { get; set; }
    public bool HandleOnce { get; set; }
    public int Counter { get; set; }
    public bool IsInProgress { get; set; }
}

We also need to overload our ScheduleJob method for this purpose and modify the existing one.

public void ScheduleJob(int seconds, Action action, bool handleOnce = false)
{
    var model = new Job
    {
        Action = action,
        HandleOnce = handleOnce
    };

    if (!_dictionary.ContainsKey(seconds))
        _dictionary.TryAdd(seconds, new List<Job>());

    _dictionary[seconds].Add(model);
}

public void ScheduleJob(int seconds, Action<object> action, object parameter, bool handleOnce = false)
{
    var model = new JobWithParam
    {
        Action = action,
        Parameter = parameter,
        HandleOnce = handleOnce
    };

    if (!_dictionaryWithParam.ContainsKey(seconds))
        _dictionaryWithParam.TryAdd(seconds, new List<JobWithParam>());

    _dictionaryWithParam[seconds].Add(model);
}

Finally, we need to make several changes to the RunScheduler method.

private void RunScheduler()
{
    while (true)
    {
        if (!IsCanceled)
        {
            if (!_dictionary.IsEmpty)
            {
                foreach (var pair in _dictionary)
                {
                    for (int i = pair.Value.Count - 1; i >= 0; i--)
                    {
                        Job value = pair.Value[i];
                        if (value.IsInProgress)
                            continue;
            
                        value.Counter++;
            
                        if (value.Counter % pair.Key == 0)
                        {
                            value.IsInProgress = true;
                            Task.Factory.StartNew(v =>
                            {
                                Job val = null;
                                try
                                {
                                    val = (Job)v;
                                    val.Action();
                                }
                                catch (Exception ex)
                                {
                                    Console.WriteLine(ex);
                                }
                                finally
                                {
                                    val.IsInProgress = false;
                                }
                            }, value);
            
                            if (value.HandleOnce)
                                pair.Value.RemoveAt(i);
                        }
                    }
            
                    if (pair.Value.Count == 0)
                        _dictionary.TryRemove(pair.Key, out _);
                }
            }
            
            if (!_dictionaryWithParam.IsEmpty)
            {
                foreach (var pair in _dictionaryWithParam)
                {
                    for (int i = pair.Value.Count - 1; i >= 0; i--)
                    {
                        JobWithParam value = pair.Value[i];
                        if (value.IsInProgress)
                            continue;
            
                        value.Counter++;
            
                        if (value.Counter % pair.Key == 0)
                        {
                            value.IsInProgress = true;
                            Task.Factory.StartNew(v =>
                            {
                                JobWithParam val = null;
                                try
                                {
                                    val = (JobWithParam)v;
                                    val.Action(val.Parameter);
                                }
                                catch (Exception ex)
                                {
                                    Console.WriteLine(ex);
                                }
                                finally
                                {
                                    val.IsInProgress = false;
                                }
                            }, value);
            
                            if (value.HandleOnce)
                                pair.Value.RemoveAt(i);
                        }
                    }
            
                    if (pair.Value.Count == 0)
                        _dictionaryWithParam.TryRemove(pair.Key, out _);
                }
            }
        }

        Thread.Sleep(1000);
    }
}

Here we are taking into account long running tasks, so using IsInProgress flag to handle such cases and if that flag in true state, then we just go to the next job and don’t increment that job’s Counter.

In addition, we should handle the new HandleOnce property and clean up our global dictionaries when they contain empty collections.

As you can see, I am not addressing many other points that we might encounter in real-world applications, such as:

• Separating threads by the priority of the tasks we are running.
• Scheduler instance separation and management.
• Using a feedback control mechanism to dynamically adjust scheduling parameters based on real-time performance metrics about CPU usage, memory consumption, task latency, etc.
• Dynamic scaling, taking into account the number of worker threads or resources based on the current load.

Monitoring and Maintenance

Effective monitoring and maintenance are critical for ensuring the long-term performance, reliability, and stability of a scheduler. Implementing robust monitoring and maintenance strategies allows for real-time tracking, debugging, and optimization of the scheduling system.

Logging and Auditing

1. Comprehensive Logging:
   • Implement detailed logging for key events and actions within the scheduler. This includes task submissions, executions, completions, and failures.
   • Ensure logs capture critical information such as timestamps, task identifiers, execution durations, exceptions, and resource usage.
2. Audit Trails:
   • Maintain audit trails to track changes in the scheduler configuration, task modifications, and user actions.
   • Store audit logs in a secure, centralized location for easy retrieval and analysis.

Performance Metrics

1. Key Performance Indicators (KPIs):
   • Define and monitor key performance indicators such as task latency, throughput, CPU usage, memory consumption, and error rates.
   • Regularly review KPIs to identify performance trends and potential bottlenecks.
2. Real-Time Monitoring:
   • Implement real-time monitoring tools to visualize performance metrics and system health. Dashboards and alerting systems can provide immediate feedback and notifications.
   • Use tools like Prometheus and Grafana for monitoring and visualization.

Error Handling and Fault Tolerance

1. Error Logging and Notifications:
   • Ensure all errors and exceptions are logged with detailed context for debugging.
   • Set up automated notifications (e.g., email, SMS, Slack) for critical errors to alert the development and operations teams.
2. Retry and Recovery Mechanisms:
   • Implement retry logic for transient errors, with exponential backoff strategies to prevent overwhelming the system.
   • Incorporate task checkpointing and state-saving mechanisms to enable recovery and continuation after failures.

Scalability and Adaptability

1. Dynamic Scaling:
   • Monitor resource utilization and adjust the number of worker threads or task execution units dynamically based on the workload.
   • Use an adaptive thread pool that can grow or shrink in response to demand.
2. Load Balancing:
   • Implement load balancing strategies to evenly distribute tasks across available resources, preventing any single resource from becoming a bottleneck.
   • Use metrics to identify hotspots and redistribute tasks as needed.

Maintenance and Updates

1. Regular Maintenance:
   • Schedule regular maintenance windows for system updates, performance tuning, and resource cleanup.
   • Perform routine checks on logs, audit trails, and performance metrics to identify areas for improvement.
2. Version Control and Deployment:
   • Use version control systems (e.g., Git) to manage changes to the scheduler codebase.
   • Implement continuous integration and continuous deployment (CI/CD) pipelines to automate testing and deployment processes.
3. Documentation:
   • Maintain comprehensive documentation for the scheduler’s architecture, components, and operational procedures.
   • Ensure documentation is kept up-to-date with any changes or new features.

By incorporating these monitoring and maintenance strategies, you can ensure that your custom scheduler remains efficient, reliable, and adaptable to changing requirements and workloads. This proactive approach will help in early detection of issues, facilitate quick resolution, and enable continuous improvement of the scheduling system.

Real-world Applications

Custom schedulers play an important role in a number of application areas in which task management, resource optimization, and performance features are considered to be critical. A developed custom scheduler would further optimize the efficiency and reliability of the systems in the following application domains.

Here are some real-world examples:

1. Financial Services
   • Automated Trading Systems:
     • Custom-built schedulers will manage trade scheduling to ensure that trading algorithms are executed at the right time.
     • Trading algorithms can be run at different times of the day for maximal performance.
   • Batch Processing and Reporting:
     • Schedulers can support EOD and EOM report automation where financial reports are generated using large datasets.
     • Dependencies between the tasks ensure that data processing is finished before report generation begins.
2. Telecommunications
   • Network Maintenance and Monitoring:
     • Custom schedulers manage the scheduling of routine network maintenance tasks.
     • Key tasks, managing network health and performance metrics, are dynamically scheduled by priority.
     • Real-time monitoring tasks are scheduled according to priority so that network health and performance metrics have real-time updates always.
   • Load Balancing and Resource Allocation:
     • Proper network resource allocation requires appropriate bandwidth allocation to handle the peak times in the most efficient way.
3. Healthcare
   • Medical Imaging and Analysis:
     • Scheduling processing tasks for medical imaging and running diagnostic-related algorithms reduce waiting time for important results.
     • Immediate results ensure that cases of urgency get timely responses and hence better care for the patients.
   • Appointment and Management of Resources:
     • Schedulers would manage appointments for patients and best utilize medical equipment and personnel.
4. Manufacturing
   • Production Line Scheduling:
     • Custom schedulers can schedule the sequence of tasks on the production lines so that the throughput would be maximized and downtime would be minimized.
     • It will do real-time work order rescheduling in machine centers, depending on available and demand, to raise the efficiency.
   • Inventory Management:
     • The scheduling of tasks for checking inventory and restocking ensures that materials are available as and when required, thus minimizing delays in production.
5. Information Technology
   • Windows Service:
     • Taking into account the nature of windows services, there are a bunch of places to use scheduler to track information from/to Api time to time
   • CI/CD:
     • The scheduler will there coordinate the execution of build, test, and deploy tasks in a CI/CD pipeline, to have predictable and timely release of the software products.
     • To ensure that one never deploys without doing a test, task dependencies are made to make sure that the build is tested.
     • Thus, custom schedulers will run automation for recurring maintenance activities and system upgrades to decrease the user disturbances.
6. E-Commerce
   • Order Processing and Fulfillment:
     • The schedulers will, therefore, coordinate task sequences that involve requisition processing right from payment confirmation through packaging and on-time delivery.
     • Task prioritization can thus facilitate acceleration on delivery of orders of high priority.
   • Inventory and price updates can be scheduled, and thus, the availability of products and their prices can be updated at all times.
7. Transportation and Logistics
   • Fleet Management:
     • Schedulers can, therefore, manage the schedule of the maintenance of an entire fleet of vehicles, while at the same time ensuring that all services are done in the least possible time.
     • Task scheduling can further ensure that the number of vehicles on the field will be as an appearance of the optimum amount needed for the deliveries.
     • Route optimization is done so as to save fuel and time.
   • Warehouse Operations:
     • Custom schedulers will manage warehouse operations like picking, packing, and shipping tasks so that workflow is optimized and processing time is minimized.
8. Research and Development
   • Simulation and Modeling:
     • The schedulers will therefore coordinate and run complex systems and models of simulation in such a way that computational resources are used to the fullest.
     • Task dependencies ensure that intermediate results are run correctly before their final analysis.
   • Analysis and Reporting:
     • Schedulers can run analysis and data reporting tasks so that research findings are observed and documented in time.

Conclusion

Various systems in industries have custom schedulers that help foster efficiency and reliability. An organizational approach through this procedure meant for minimizing the response time of the schedulers, in an application-specific way, sets forth a means through which an organization can conduct resource optimization, reduce the cost of operation, and scale in terms of performance. Appropriate principles and implementations for monitoring, maintaining, and carrying out optimization strategies for the right knowledge can be of aid in making highly effective custom schedulers for the varying demands of real-life systems.

Codebase

All the code used here, along with additional optimizations, can be found on my GitHub at the following link:
https://github.com/mustf4/MultitaskScheduler