FIFO Task Scheduler with Queues

Hello again! Welcome back to our module on "Building Systems Incrementally."

In the last two lessons, you built and extended an in-memory key-value store, adding Time-To-Live (TTL) functionality and eviction strategies. This focused on building a stateful, data-centric component.

Now, we shift our focus to a logic-centric component: a task scheduler. Your work on the Rudi inference scheduler at AWS gives you a deep background in this domain, especially concerning concurrency, priority queuing, and load management in Rust. This lesson will leverage that expertise as we build the foundational version of a scheduler in Python. We'll start with the simplest model: one that processes tasks in a First-In, First-Out (FIFO) order using a queue.

By the end of this lesson, you will be able to build a simple task scheduler that processes tasks in FIFO order using a queue.

---

1. The Anatomy of a Task Scheduler

At its core, a task scheduler is a system designed to run jobs asynchronously. A client submits a task, and one or more workers pick it up and execute it at some point in the future. This decouples the task submission from its execution, which is fundamental for building responsive and scalable applications.

This architecture can be visualized as a classic producer-consumer system, a pattern you are likely very familiar with from your work with Tokio channels.

To begin, let's explore the fundamental concepts and requirements of a job scheduler.

As the video suggests, we need to define these key components:

1. A Task/Job: A representation of the work to be done (e.g., a function and its arguments).
2. An Interface: A way for clients to submit tasks to the system (e.g., a schedule method).
3. A Queue: A data structure to store pending tasks.
4. A Worker: A process or thread that retrieves tasks from the queue and executes them.

---

2. The FIFO Principle and Python's Queues

For a scheduler that processes tasks in the order they were submitted, the natural choice for a data store is a First-In, First-Out (FIFO) queue.

In Python, the standard library offers two primary queue implementations suitable for this:

• collections.deque: A highly efficient double-ended queue. It's great for single-threaded applications where you manage the execution loop yourself.
• queue.Queue: A thread-safe, blocking queue. It's designed for concurrent producer-consumer scenarios.

The queue.Queue is a perfect Python analogue to the message-passing channels (like MPSC) you use in Rust with Tokio. When a worker thread calls get() on an empty queue.Queue, the thread sleeps efficiently until another thread put()s an item into the queue. This avoids busy-waiting (inefficiently polling in a loop) and is a cornerstone of the producer-consumer pattern.

Given its elegance and direct mapping to robust concurrency patterns, we will use queue.Queue as the backbone of our scheduler. This choice also sets us up perfectly for future lessons on concurrency.

---

3. Implementing the Scheduler

Let's translate the architecture into Python code. We'll follow the incremental approach of defining our data structures and then building the logic around them. The blog post "Building a task queue, Part 1" provides an excellent narrative for this process, even though it uses Ruby and a database. We'll adapt its logic for our in-memory Python implementation.

Step 1: Defining the Task

Just as we created a CacheEntry in the last lesson, we'll create a Task dataclass. It will store a callable (the function to execute) and its arguments.

from dataclasses import dataclass, field
from typing import Callable, Any

@dataclass
class Task:
    """Represents a unit of work to be executed."""
    func: Callable
    args: tuple = field(default_factory=tuple)
    kwargs: dict = field(default_factory=dict)

Step 2: Designing the Scheduler and Worker

Our FIFOScheduler class will manage the queue and the worker thread.

• schedule(func, *args, **kwargs): The "producer" side. It creates a Task and adds it to the queue.Queue.
• _worker_loop(): The "consumer" side. It runs in an infinite loop, blocking on self.tasks.get() until a task is available, then executes it.
• start() and stop(): Methods to manage the lifecycle of the worker thread.

Running the worker in a separate thread is crucial. It ensures that the main part of your application (the producer) doesn't block while waiting for tasks to be executed.

---

4. Your Task: Implement the FIFO Scheduler

Now it's your turn to build the scheduler. Below is the complete class structure. Your job is to fill in the schedule and _worker_loop methods. I've also added start and stop methods to manage the worker thread cleanly.

import queue
import threading
import time
from dataclasses import dataclass, field
from typing import Callable, Any

@dataclass
class Task:
    """Represents a unit of work to be executed by the scheduler."""
    func: Callable
    args: tuple = field(default_factory=tuple)
    kwargs: dict = field(default_factory=dict)

class FIFOScheduler:
    """
    A simple task scheduler that executes tasks in a First-In, First-Out order.
    """
    def __init__(self):
        self.tasks = queue.Queue()
        self._worker_thread = None



        # A sentinel value to signal the worker thread to stop
        self._stop_signal = object()

    def schedule(self, func: Callable, *args: Any, **kwargs: Any):
        """
        Schedules a function to be executed by the worker.

        Args:
            func: The function to execute.
            *args: Positional arguments to pass to the function.
            **kwargs: Keyword arguments to pass to the function.
        """



        # 1. Create a Task object with the provided func, args, and kwargs.
        # 2. Put the Task object onto the self.tasks queue.
        pass # Your implementation here

    def _worker_loop(self):
        """The main loop for the worker thread."""
        print("Worker thread started.")
        while True:

```grasp
{
  "type": "exercise",
  "id": "44e8fa75-7022-450f-8a90-7e3030289b3d"
}

        # 1. Get a task from the queue. This call will block until a task is available.
        # 2. Check if the retrieved task is the stop signal. If it is, break the loop.
        # 3. If it's a regular task, execute it.
        #    - Use a try...except block to catch any exceptions during task execution.
        #    - Print a success or failure message.
        #    - Unpack the task's args and kwargs to call the function: task.func(*task.args, **task.kwargs)
        pass # Your implementation here
    print("Worker thread stopped.")

def start(self):
    """Starts the worker thread."""
    if self._worker_thread is not None:
        print("Worker is already running.")
        return

    self._worker_thread = threading.Thread(target=self._worker_loop)
    self._worker_thread.start()

def stop(self):
    """Stops the worker thread gracefully."""
    if self._worker_thread is None:
        print("Worker is not running.")
        return




    # Put the sentinel value on the queue to signal the worker to stop
    self.tasks.put(self._stop_signal)



    # Wait for the worker thread to finish
    self._worker_thread.join()
    self._worker_thread = None

--- Example Usage ---

def sample_task(name: str, duration: int):
"""A simple task that prints a message and sleeps."""
print(f"Starting task: '{name}' (will run for {duration}s)")
time.sleep(duration)
print(f"Finished task: '{name}'")

def failing_task():
"""A task that is designed to fail."""
print("Starting failing task...")
raise ValueError("This task was meant to fail!")

if name == 'main':
scheduler = FIFOScheduler()
scheduler.start()

# Schedule some tasks
print("Scheduling tasks...")
scheduler.schedule(sample_task, "Task A", 2)
scheduler.schedule(sample_task, name="Task B", duration=1)
scheduler.schedule(failing_task)
scheduler.schedule(sample_task, "Task C", 1)




# Give the tasks some time to run
# Notice the main thread is not blocked and can do other things.
print("Main thread is free while tasks run in the background.")
time.sleep(6)




# Stop the scheduler
print("Stopping scheduler...")
scheduler.stop()
print("Scheduler stopped.")

**Reflection Question:** Our current worker is single-threaded. It processes one task at a time. Drawing from your experience building Rudi, what would be the first step to allow for concurrent execution of multiple tasks? What new challenges would this introduce?

---


```grasp
{
  "type": "exercise",
  "id": "1d307d9f-6697-4e91-868e-286604dfd4f4"
}

Conclusion

Congratulations! You've successfully built a simple, yet robust, FIFO task scheduler. You've applied the classic producer-consumer pattern in Python, a direct parallel to the concurrent programming paradigms you're proficient with in Rust. This foundational component is the first step in our incremental journey toward a more complex scheduling system.

Key Takeaways:

• A task scheduler can be elegantly modeled as a producer-consumer system.
• For FIFO ordering, a queue is the central data structure.
• Python's queue.Queue is a thread-safe, blocking queue, which allows a worker thread to wait for tasks efficiently without polling, analogous to Rust's channels.
• Running the worker in a separate thread decouples execution from submission, making the main application responsive.
• Graceful shutdown using a sentinel value is a clean pattern for managing the worker's lifecycle.

Next Lesson Preview

Our current scheduler treats all tasks equally. In the real world, some tasks are more important than others. In the next lesson, we will evolve the scheduler to support priority levels. This will involve replacing our FIFO queue.Queue with a queue.PriorityQueue, which is internally implemented as a min-heap, and exploring the trade-offs of this more complex design.