Своевременная очередь потребителя/производителя

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

Во-первых, вы можете реализовать это с помощью threading.Timer вместо явного фонового потока. Но, хотя это может быть проще, на самом деле это не решит проблему, связанную с тем, что вы создаете поток за спиной пользователя, хотят они того или нет. Кроме того, с threading.Timer вы фактически запускаете новый поток каждый раз, когда перезапускаете таймер, что может быть проблемой производительности. (У вас есть только один за раз, но тем не менее запуск и остановка потоков не бесплатны.)

Если вы посмотрите на модули PyPI, рецепты ActiveState и различные фреймворки, то увидите множество реализаций, позволяющих запускать несколько таймеров в одном фоновом потоке. Это решило бы вашу проблему.

Но это все еще не идеальное решение. Например, предположим, что моему приложению требуется 20 объектов TimelyQueue или TimelyQueue плюс еще 19 объектов, для которых нужны таймеры. Я бы все равно закончил с 20 потоками. Или, скажем, я создаю сервер сокетов или приложение с графическим интерфейсом (два наиболее очевидных варианта использования вашего TimelyQueue; я могу реализовать таймер поверх моего цикла событий (или, скорее всего, просто использовать таймер, который поставляется с фреймворк), так зачем мне вообще нужен поток?

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

def __init__(self, timerfactory = threading.Timer):
    self.timerfactory = timerfactory
    ...

Теперь, когда вам нужно настроить таймер:

if when < self.waketime:
    self.timer.cancel()
    self.timer = self.timerfactory(when - now(), self.timercallback)
    self.waketime = when

Для быстрых и грязных случаев использования этого было бы достаточно из коробки. Но если я, например, использую twisted, я могу просто используйте TimelyQueue(twisted.reactor.callLater), и теперь таймеры очереди проходят через цикл событий twisted. Или, если у меня есть реализация с несколькими таймерами и одним потоком, которую я использую в другом месте, TimelyQueue(multiTimer.add), и теперь таймеры очереди работают в том же потоке, что и все другие мои таймеры.

Если вы хотите, вы можете указать лучшее значение по умолчанию, чем threading.Timer, но на самом деле, я думаю, что большинство людей, которым нужно что-то лучше, чем threading.Timer, смогут предоставить что-то, что лучше для их конкретного приложения, чем то, что вы предоставляете.

Конечно, не каждая реализация таймера имеет тот же API, что и threading.Timer, хотя вы будете удивлены, узнав, сколько из них это делает. Но не так сложно написать адаптер, если у вас есть таймер, который вы хотите использовать с TimelyQueue, но у него неправильный интерфейс. Например, если я создаю приложение PyQt4/PySide, QTimer не имеет метода cancel и принимает мс вместо секунд, поэтому мне пришлось бы сделать что-то вроде этого:

class AdaptedQTimer(object):
    def __init__(self, timeout, callback):
        self.timer = QTimer.singleShot(timeout * 1000, callback)
    def cancel(self):
        self.timer.stop()

q = TimelyQueue(AdaptedQTimer)

if self.timer is None:
    self.timer = self.timerfactory(when)
elif when < self.timer.waketime:
    self.timer.adjust(when)

def timerFactoryAdapter(threadingStyleTimerFactory):
    class TimerFactory(object):
        def __init__(self, timestamp, callback):
            self.timer = threadingStyleTimerFactory(timestamp - now(), callback)
            self.callback = callback
        def cancel(self):
            return self.timer.cancel()
        def adjust(self, timestamp):
            self.timer.cancel()
            self.timer = threadingStyleTimerFactory(timestamp - now(), self.callback)

Для справки, я реализовал то, что вы предложили, используя фабрику таймеров. Я провел небольшой тест, используя версию выше и новую версию, используя класс threading.Timer:

1. Первая реализация

   • При разрешении по умолчанию (5 мс, то есть все, что находится в окне 5 мс, запускается вместе), достигается около 88k put()/сек и 69k get()/сек.

   • При разрешении, установленном на 0 мс (без оптимизации), достигается около 88 тыс. put()/сек и 55 тыс. get()/сек.

2. Вторая реализация

   • При разрешении по умолчанию (5 мс) достигается около 88k put()/сек и 65k get()/сек.

   • При разрешении, установленном на 0 мс, достигается около 88k put()/сек и 62k get()/сек.

Признаюсь, я удивлен, что вторая реализация работает быстрее без оптимизации разрешения. Сейчас слишком поздно для расследования.

import collections
import heapq
import threading
import time

class TimelyQueue:
    """
    Implements a similar but stripped down interface of Queue which
    delivers items on time only.
    """

    def __init__(self, resolution=5, timerfactory=threading.Timer):
        """
        `resolution' is an optimization to avoid wasting CPU cycles when
        something is about to happen in less than X ms.
        """
        self.resolution = float(resolution) / 1000
        self.timerfactory = timerfactory
        self.queue = []
        self.triggered = collections.deque()
        self.putcond = threading.Condition()
        self.getcond = threading.Condition()
        # Optimization to avoid waking the thread uselessly.
        self.putwaketime = 0
        self.timer = None
        self.terminating = False

    def __arm(self):
        """
        Arm the next timer; putcond must be acquired!
        """
        curtime = time.time()
        when, item = self.queue[0]
        interval = when - curtime
        self.putwaketime = when
        self.timer = self.timerfactory(interval, self.__fire)
        self.timer.start()

    def __fire(self):
        with self.putcond:
            curtime = time.time()
            debug = 0
            while True:
                # Don't dequeue now, we are not sure to use it yet.
                try:
                    when, item = self.queue[0]
                except IndexError:
                    break
                if when > curtime + self.resolution:
                    break

                debug += 1
                self.triggered.append(heapq.heappop(self.queue))
            if len(self.triggered) > 0:
                with self.getcond:
                    self.getcond.notify(len(self.triggered))
            if self.terminating:
                return
            if len(self.queue) > 0:
                self.__arm()

    def put(self, when, item):
        """
        `when' is a Unix time from Epoch.
        """
        with self.putcond:
            heapq.heappush(self.queue, (when, item))
            if when >= self.putwaketime and self.putwaketime != 0:
                return
            # Arm next timer.
            if self.timer is not None:
                self.timer.cancel()
            self.__arm()

    def get(self, timeout=None):
        """
        Timely return the next object on the queue.
        """
        with self.getcond:
            if len(self.triggered) > 0:
                when, item = self.triggered.popleft()
                return item
            self.getcond.wait(timeout)
            try:
                when, item = self.triggered.popleft()
            except IndexError:
                return None
            return item

    def qsize(self):
        """
        Self explanatory.
        """
        with self.putcond:
            return len(self.queue)

    def terminate(self):
        """
        Request the embedded thread to terminate.
        """
        with self.putcond:
            self.terminating = True
            if self.timer is not None:
                self.timer.cancel()
            self.putcond.notifyAll()


if __name__ == "__main__":
    q = TimelyQueue(0)
    N = 100000
    t0 = time.time()
    for i in range(N):
        q.put(time.time() + 2, i)
    dt = time.time() - t0
    print "put done in %.3fs (%.2f put/sec)" % (dt, N / dt)
    t0 = time.time()
    i = 0
    while i < N:
        a = q.get(3)
        if i == 0:
            dt = time.time() - t0
            print "start get after %.3fs" % dt
            t0 = time.time()
        i += 1
    dt = time.time() - t0
    print "get done in %.3fs (%.2f get/sec)" % (dt, N / dt)
    q.terminate()
    # Give change to the thread to exit properly, otherwise we may get
    # a stray interpreter exception.
    time.sleep(0.1)