[解決済み] Python time.sleep()とevent.wait()の比較

質問

マルチスレッドのPythonアプリケーションで、あるアクションを一定間隔で実行したいです。私はそれを行うための2つの異なる方法を見てきました

exit = False
def thread_func(): 
    while not exit:
       action()
       time.sleep(DELAY)

または

exit_flag = threading.Event()
def thread_func(): 
    while not exit_flag.wait(timeout=DELAY):
       action()

どちらか一方に優位性がありますか？ どちらがより少ないリソースを使用するか、または他のスレッドや GIL とよりうまく機能するか？ どちらが私のアプリの残りのスレッドをより反応しやすくしますか?

(ある外部イベントが exit または exit_flag で、シャットダウンしている間、完全に遅れても構わないと思っています)

どのように解決するのですか？

使用方法 exit_flag.wait(timeout=DELAY) を使うと、whileループから即座に抜け出せるので、よりレスポンスが良くなります。 exit_flag が設定されると、whileループから即座に抜け出せるからです。とは time.sleep では、イベントが設定された後でも、待ち時間に time.sleep の呼び出しで寝るまで DELAY を呼び出します。

実装面では、Python 2.xとPython 3.xでは、動作が大きく異なります。Python 2.xでは Event.wait は純粋な Python で実装され、たくさんの小さな time.sleep の呼び出しを使って実装されています。

from time import time as _time, sleep as _sleep

....
# This is inside the Condition class (Event.wait calls Condition.wait).
def wait(self, timeout=None):
    if not self._is_owned():
        raise RuntimeError("cannot wait on un-acquired lock")
    waiter = _allocate_lock()
    waiter.acquire()
    self.__waiters.append(waiter)
    saved_state = self._release_save()
    try:    # restore state no matter what (e.g., KeyboardInterrupt)
        if timeout is None:
            waiter.acquire()
            if __debug__:
                self._note("%s.wait(): got it", self)
        else:
            # Balancing act:  We can't afford a pure busy loop, so we
            # have to sleep; but if we sleep the whole timeout time,
            # we'll be unresponsive.  The scheme here sleeps very
            # little at first, longer as time goes on, but never longer
            # than 20 times per second (or the timeout time remaining).
            endtime = _time() + timeout
            delay = 0.0005 # 500 us -> initial delay of 1 ms
            while True:
                gotit = waiter.acquire(0)
                if gotit:
                    break
                remaining = endtime - _time()
                if remaining <= 0:
                    break
                delay = min(delay * 2, remaining, .05)
                _sleep(delay)
            if not gotit:
                if __debug__:
                    self._note("%s.wait(%s): timed out", self, timeout)
                try:
                    self.__waiters.remove(waiter)
                except ValueError:
                    pass
            else:
                if __debug__:
                    self._note("%s.wait(%s): got it", self, timeout)
    finally:
        self._acquire_restore(saved_state)

これは実際には wait を使うことは、単に完全な DELAY を無条件にスリープさせるよりも少し CPU を消費しますが、 (潜在的に、どの程度の時間 DELAY の長さにもよりますが）より反応しやすくなるという利点があります。また、GIL を頻繁に再取得する必要があるため、次のスリープをスケジューリングできる一方で time.sleep は GIL を解放して完全な DELAY . さて、GIL をより頻繁に取得することは、アプリケーションの他のスレッドに顕著な影響を与えるでしょうか？そうかもしれませんし、そうでないかもしれません。それは、他のスレッドが何本動いていて、どのような作業負荷がかかっているかによります。私の推測では、スレッド数が多いか、または CPU に拘束される作業を多く行う別のスレッドがない限り、特に目立つことはないでしょうが、両方の方法を試して見ることは十分に簡単です。

Python 3.xでは、実装の多くが純粋なCコードに移行しています。

import _thread # C-module
_allocate_lock = _thread.allocate_lock

class Condition:
    ...
    def wait(self, timeout=None):
        if not self._is_owned():
            raise RuntimeError("cannot wait on un-acquired lock")
        waiter = _allocate_lock()
        waiter.acquire()
        self._waiters.append(waiter)
        saved_state = self._release_save()
        gotit = False
        try:    # restore state no matter what (e.g., KeyboardInterrupt)
            if timeout is None:
                waiter.acquire()
                gotit = True
            else:
                if timeout > 0:
                    gotit = waiter.acquire(True, timeout)  # This calls C code
                else:
                    gotit = waiter.acquire(False)
            return gotit
        finally:
            self._acquire_restore(saved_state)
            if not gotit:
                try:
                    self._waiters.remove(waiter)
                except ValueError:
                    pass

class Event:
    def __init__(self):
        self._cond = Condition(Lock())
        self._flag = False

    def wait(self, timeout=None):
        self._cond.acquire()
        try:
            signaled = self._flag
            if not signaled:
                signaled = self._cond.wait(timeout)
            return signaled
        finally:
            self._cond.release()

そして、ロックを取得するCのコード。

/* Helper to acquire an interruptible lock with a timeout.  If the lock acquire
 * is interrupted, signal handlers are run, and if they raise an exception,
 * PY_LOCK_INTR is returned.  Otherwise, PY_LOCK_ACQUIRED or PY_LOCK_FAILURE
 * are returned, depending on whether the lock can be acquired withing the
 * timeout.
 */
static PyLockStatus
acquire_timed(PyThread_type_lock lock, PY_TIMEOUT_T microseconds)
{
    PyLockStatus r;
    _PyTime_timeval curtime;
    _PyTime_timeval endtime;


    if (microseconds > 0) {
        _PyTime_gettimeofday(&endtime);
        endtime.tv_sec += microseconds / (1000 * 1000);
        endtime.tv_usec += microseconds % (1000 * 1000);
    }


    do {
        /* first a simple non-blocking try without releasing the GIL */
        r = PyThread_acquire_lock_timed(lock, 0, 0);
        if (r == PY_LOCK_FAILURE && microseconds != 0) {
            Py_BEGIN_ALLOW_THREADS  // GIL is released here
            r = PyThread_acquire_lock_timed(lock, microseconds, 1);
            Py_END_ALLOW_THREADS
        }

        if (r == PY_LOCK_INTR) {
            /* Run signal handlers if we were interrupted.  Propagate
             * exceptions from signal handlers, such as KeyboardInterrupt, by
             * passing up PY_LOCK_INTR.  */
            if (Py_MakePendingCalls() < 0) {
                return PY_LOCK_INTR;
            }

            /* If we're using a timeout, recompute the timeout after processing
             * signals, since those can take time.  */
            if (microseconds > 0) {
                _PyTime_gettimeofday(&curtime);
                microseconds = ((endtime.tv_sec - curtime.tv_sec) * 1000000 +
                                (endtime.tv_usec - curtime.tv_usec));

                /* Check for negative values, since those mean block forever.
                 */
                if (microseconds <= 0) {
                    r = PY_LOCK_FAILURE;
                }
            }
        }
    } while (r == PY_LOCK_INTR);  /* Retry if we were interrupted. */

    return r;
}

この実装は応答性が良く、GILを再取得するための頻繁なウェイクアップを必要としません。