Task Scheduler

medium · fake clock, priority by due time

Brief-reported interview attribution (unverified): Amazon, Microsoft

Design a one-shot task scheduler. schedule(task_id, fn, delay_seconds) runs fn after delay_seconds from now (a callable with no arguments). Time comes from a clock callable passed to the constructor (default: time.time). run_due() runs every task whose time has come, in due-time order (ties in insertion order), and returns the list of task ids it ran. Scheduling an id that is already pending replaces it (new function and new time).

class TaskScheduler:
    def __init__(self, clock=None):
    def schedule(self, task_id, fn, delay_seconds):
    def cancel(self, task_id):
        # -> True if a pending task was cancelled, False otherwise
    def pending(self):
        # -> [(task_id, run_at), ...] sorted by run time (insertion order on ties)
    def run_due(self):
        # -> [task_id, ...] of the tasks that just ran, in run order

Rules. A task runs when clock() >= run_at. run_due() removes a task before calling its function (a task that re-schedules itself is a new task). pending times are absolute clock values.

Tests use explicit inputs and injected clocks/callbacks. No network, real threads or application-server execution is involved.

Public demo

# Public demo: run after implementing the class API above.
now = [1000.0]
log = []
s = TaskScheduler(clock=lambda: now[0])
s.schedule('a', lambda: log.append('a'), 10)
s.schedule('b', lambda: log.append('b'), 5)
s.schedule('c', lambda: log.append('c'), 5)
print(s.run_due())
print(log)

Starter code

class TaskScheduler:

    def __init__(self, clock=None):
        """Initialize the state described in the problem."""
        pass

    def schedule(self, task_id, fn, delay_seconds):
        """Implement the return value and rules in the API contract."""
        pass

    def cancel(self, task_id):
        """-> True if a pending task was cancelled, False otherwise"""
        pass

    def pending(self):
        """-> [(task_id, run_at), ...] sorted by run time (insertion order on ties)"""
        pass

    def run_due(self):
        """-> [task_id, ...] of the tasks that just ran, in run order"""
        pass

Loading editor…

Each test has a five-second execution limit. Your practice drafts and solved progress stay in this browser.

Return to the Python compiler