Python task scheduler

1. Review

1. The docstrings could be improved. It would be nice if the docstring for the Scheduler class explained (briefly) how to use it, and there should be docstrings for the schedule and cancel methods.

2. When using the logging module, it's not normally necessary to format the log messages yourself. Instead, pass the format string and arguments, and let the logger do the formatting if and when it needs to (depending on the log level, it might never need to format a particular message). So instead of:

   logger.info('waiting with timeout: {}'.format(self._timeout))
   

   write:

   logger.info("waiting with timeout %f", self._timeout)
   

3. If there are multiple tasks with the same name, then the cancel method cancels the one that appears closest to the start of the heap. This is not necessarily the one with the soonest start time (because heaps are arranged as a tree, not as a sorted list). This behaviour seems quite obscure to me and seems likely to lead to unreliable programs.

   I think it would be easier for the user to write reliable programs if you adopted one of these three approaches: (i) cancel all tasks with matching names so that the behaviour doesn't depend on the arrangement of the heap; or (ii) require all tasks in the queue to have unique names; or (iii) have the schedule method return some object representing the task, so that this can later be passed to cancel to uniquely identify the particular task to be cancelled.

4. Cancellation takes time proportional to the number of tasks in the heap. This could be improved to (amortized) \$O(\log n)\$ by (i) having the schedule method return some object representing the task as suggesting above; (ii) passing this task object to the cancel method; (iii) leaving the cancelled task in the heap but marking it as cancelled; (iv) discarding cancelled tasks when they are popped from the heap.

5. The attributes of a Task are start, name, fn, but the arguments to schedule are name, fn, start. This kind of inconsistency risks confusion or error. (You need start to be first so that tasks compare according to their start times, but there are other ways to achieve this.)

6. If you attempt to schedule two tasks with the same start time and the same name, then you get TypeError: '<' not supported between instances of 'function' and 'function'.

7. The name _minheap describes the data structure, but it is more important to describe the purpose (the data structure is an implementation detail). So I would use a name like _tasks.

8. The _timeout attribute is only used in the run function, so it could be a local variable there instead of an attribute on the Scheduler object.

9. The _get_next_timeout method is only used in the run function, so it could be a local function there instead of a method on the Scheduler class. (It must only be called with the lock held, so it is not suitable for a stand-alone method.)

10. The _start method is only called in the __init__ method, so could be inlined there.

11. The logic in run seems wrong to me. The following sequence of events is possible: (i) The timer thread calls _get_next_timeout which returns a timeout of 10 seconds, say. (ii) The timer thread released the condition variable. (iii) Another thread calls schedule with a task that should be run in 5 seconds time and notifies the condition variable (but nothing is waiting on the condition variable). (iv) The timer thread acquires the condition variable. (v) The timer thread waits on the old timeout of 10 seconds. But it should have computed a new timeout of 5 seconds and waited on that instead.

    To fix this, it is necessary to compute the timeout between acquiring the condition variable and waiting on it.

12. The use of acquire and release on the condition variable seems risky, as an exception at the wrong time would leave the lock held. Normally one uses the context manager interface to avoid this risk. I guess that you didn't do this because you wanted to launch the task with the lock not held. But it only takes a small rearrangement of code to make this work.

2. Revised code

This is a version with nameless tasks, but if you prefer your tasks to have names, then it will be straightfoward for you to add them.

class Scheduler:
    """A schedule of tasks to be run in background threads. Call the
    schedule method to schedule a task to run at a particular time.
    Call the task's cancel method to cancel it if it has not already
    started running.

    """

    @functools.total_ordering
    class _Task:
        "A scheduled task."

        def __init__(self, fn, start):
            "Create task that will run fn at or after the datetime start."
            self.fn = fn
            self.start = start
            self.cancelled = False

        def __le__(self, other):
            # Tasks compare according to their start time.
            return self.start <= other.start

        @property
        def timeout(self):
            "Return time remaining in seconds before task should start."
            return (self.start - datetime.now()).total_seconds()

        def cancel(self):
            "Cancel task if it has not already started running."
            self.cancelled = True
            logger.info("canceled %s", self)

    def __init__(self):
        cv = self._cv = threading.Condition(threading.Lock())
        tasks = self._tasks = []

        def run():
            while True:
                with cv:
                    while True:
                        timeout = None
                        while tasks and tasks[0].cancelled:
                            heapq.heappop(tasks)
                        if tasks:
                            timeout = tasks[0].timeout
                            if timeout <= 0:
                                task = heapq.heappop(tasks)
                                break
                        cv.wait(timeout=timeout)
                logger.info("starting task %s", task)
                threading.Thread(target=task.fn).start()

        threading.Thread(target=run, name='Scheduler').start()

    def schedule(self, fn, start):
        """Schedule a task that will run fn at or after start (which must be a
        datetime object) and return an object representing that task.

        """
        task = self._Task(fn, start)
        logger.info("scheduling task %s", task)
        with self._cv:
            heapq.heappush(self._tasks, task)
            self._cv.notify()
        logger.info("scheduled task %s", task)
        return task