вызов функции с задержкой

Question

Я получил этот вопрос об интервью.

Реализация планировщика заданий, который принимает функцию f и целое число n и вызывает f через n миллисекунд.

У меня очень простое решение:

import time

def schedulerX(f,n):
    time.sleep(0.001*n)
    f

Тем не менее, предлагаемое решение является гораздо более сложным, как показано ниже. Я не понимаю, какова цель всего этого дополнительного кода. Пожалуйста, просветите меня.

from time import sleep
import threading

class Scheduler:
    def __init__(self):
        self.fns = [] # tuple of (fn, time)
        t = threading.Thread(target=self.poll)
        t.start()

    def poll(self):
        while True:
            now = time() * 1000
            for fn, due in self.fns:
                if now > due:
                    fn()
            self.fns = [(fn, due) for (fn, due) in self.fns if due > now]
            sleep(0.01)

    def delay(self, f, n):
        self.fns.append((f, time() * 1000 + n))

Answer 1

Как уже отмечали другие, ваше решение «блокирует»: оно предотвращает что-либо еще, пока оно ожидает своего запуска. Цель предлагаемого решения - позволить вам запланировать работу, а затем заняться другими делами.

Что касается объяснения того, что делает предложенный код:

Сначала вы создадите Scheduler, который запустит собственный поток, который эффективно работает в фоновом режиме и который будет запускать задания.

scheduler = Scheduler()

В своем коде вы можете запланировать любые задания, не ожидая их запуска:

def my_recurring_job():
    # Do some stuff in the background, then re-run this job again
    # in one second.

    ### Do some stuff ###

    scheduler.delay(my_recurring_job, 1000)

scheduler.delay(lambda: print("5 seconds passed!"), 5 * 1000)
scheduler.delay(lambda: print("2 hours passed!"), 2 * 60 * 60 * 1000)
scheduler.delay(my_recurring_job, 1000)

# You can keep doing other stuff without waiting

Поток планировщика просто зацикливается навсегда в своем методе poll, выполняя любые задания, для которых пришло время, затем спит 0,01 секунды и проверяет снова. В коде есть небольшая ошибка, из-за которой, если сейчас == должное, задание не будет запущено, но оно также не будет сохранено на потом. Это должно быть if now >= due: вместо.

Более продвинутый планировщик может использовать threading.Condition вместо опроса 100 раз в секунду:

import threading
from time import time

class Scheduler:
    def __init__(self):
        self.fns = [] # tuple of (fn, time)

        # The lock prevents 2 threads from messing with fns at the same time;
        # also lets us use Condition
        self.lock = threading.RLock()

        # The condition lets one thread wait, optionally with a timeout,
        # and lets other threads wake it up
        self.condition = threading.Condition(self.lock)

        t = threading.Thread(target=self.poll)
        t.start()

    def poll(self):
        while True:
            now = time() * 1000

            with self.lock:
                # Prevent the other thread from adding to fns while we're sorting
                # out the jobs to run now, and the jobs to keep for later

                to_run = [fn for fn, due in self.fns if due <= now]
                self.fns = [(fn, due) for (fn, due) in self.fns if due > now]

            # Run all the ready jobs outside the lock, so we don't keep it
            # locked longer than we have to
            for fn in to_run:
                fn()

            with self.lock:
                if not self.fns:
                    # If there are no more jobs, wait forever until a new job is 
                    # added in delay(), and notify_all() wakes us up again
                    self.condition.wait()
                else:
                    # Wait only until the soonest next job's due time.
                    ms_remaining = min(due for fn, due in self.fns) - time()*1000
                    if ms_remaining > 0:
                        self.condition.wait(ms_remaining / 1000)

    def delay(self, f, n):
        with self.lock:
            self.fns.append((f, time() * 1000 + n))

            # If the scheduler thread is currently waiting on the condition,
            # notify_all() will wake it up, so that it can consider the new job's
            # due time.
            self.condition.notify_all()

Answer 2

Есть несколько отличий (теоретически).

Первое, и я думаю, самое важное, что ваше решение может эффективно планировать только одну функцию за раз.Например, предположим, что вы хотите запустить функцию f1 10 миллисекунд с этого момента, а другую функцию f2 10 миллисекунд после этого.

Вы не сможете сделать это легко, поскольку что-то вроде schedulerX(f1, 10); schedulerX(f2, 10) будет ждать окончания работы f1, прежде чем начинать ожидание f2.Если f1 занимает час, ваше планирование f2 будет полностью неверным.

Цель второй версии, очевидно, состоит в том, чтобы таймер и каждая функция работали в отдельном потоке, так чтоодин вызов функции не блокирует другой.

Однако, как отмечают другие в комментариях, импорт выполняется неправильно, для выполнения функции требуется list, даже если в спецификации проблемы указано a функции, и она на самом деле не работает так, как я ее описал, так что более или менее нет никакой разницы.