如何获得使用系统正常运行时间的 Python threading.Timer 的等价物？

Question

TL;DR threading.Timer 使用系统时间，但是在我使用它时时间会发生变化，我怎样才能让它使用系统正常运行时间？

我有一个 Python 脚本，它做了很多事情，其中​​之一就是设置系统时间。当这个脚本启动的时候是错误的。此脚本还需要有 30 秒的全局超时。

我一直在使用以下超时类：

class Timeout(object):
    def __init__(self, seconds=1, signum=signal.SIGUSR1, exception=TimeoutException):
        self.exception = exception
        self.pid = os.getpid()
        self.signum = signum
        self.timer = threading.Timer(seconds, self.exit_function)

    def exit_function(self):
        os.kill(self.pid, self.signum)

    def handle_timeout(self, signum, frame):
        raise self.exception()

    def __enter__(self):
        signal.signal(self.signum, self.handle_timeout)
        self.timer.start()

    def __exit__(self, type, value, traceback):
        self.timer.cancel()

这包含了我的整个脚本：

with Timeout(seconds=30):
    main()

有时脚本会很快失败，或者在 30 秒后永远不会被杀死。我相信这是因为threading.Timer 使用了在脚本运行时更改的系统时间。无论如何我可以让它使用系统正常运行时间吗？

Answer 1

更新

我现在正在做的是使用 PyPI 上的 monotonic 包中的单调函数对 threading._time 进行猴子补丁。

import threading
import monotonic

threading._time = monotonic.monotonic

原答案

我最终扩展了threading.Timer 以使用系统正常运行时间。

class Timer(threading._Timer):

    def __init__(self, *args, **kwargs):
        super(Timer, self).__init__(*args, **kwargs)

        # only works on Linux
        self._libc = ctypes.CDLL('libc.so.6')
        self._buf = ctypes.create_string_buffer(128)

    def uptime(self):
        self._libc.sysinfo(self._buf)
        return struct.unpack_from('@l', self._buf.raw)[0]

    def run(self):
        start_time = self.uptime()
        while not self.finished.is_set():
            time.sleep(0.1)
            if self.uptime() - start_time > self.interval:
                self.function(*self.args, **self.kwargs)
                break
        self.finished.set()

Answer 2

看来您使用的是 Python monotonic 将是标准库中time.monotonic 的别名。 然后time.monotonic 也用于threading-库。如docs中所说：

返回单调时钟的值（以秒为单位），即不能倒退的时钟。 时钟不受系统时钟更新的影响。

因此，使用 Python >=3.3 threading.Timer 将是独立的。

---

试图解决您的问题： Windows

我看到了一个选项，可以使用 gettickcount 从 kernel32.dll 使用 ctypes 获得所需的系统正常运行时间：

import ctypes
kernel_32 = ctypes.cdll.LoadLibrary("Kernel32.dll")
kernel_32.GetTickCount()  # overflow after 49.7 days
kernel_32.GetTickCount64()

有了这个，您可以创建自己的计时器，但性能不佳：

def sleep(time):
    start = kernel_32.GetTickCount64()
    end = start + time
    while kernel_32.GetTickCount64() < end:
        pass
    print('done')

我真的希望这种方法可以帮助您解决问题 - 祝您好运

---

编辑： Linux

基于monotonic：你可以试试这个作为getTickCount的替代方法₁：

try:
    clock_gettime = ctypes.CDLL(ctypes.util.find_library('c'), use_errno=True).clock_gettime
except Exception:
    clock_gettime = ctypes.CDLL(ctypes.util.find_library('rt'), use_errno=True).clock_gettime

class timespec(ctypes.Structure):
    """Time specification, as described in clock_gettime(3)."""
    _fields_ = (('tv_sec', ctypes.c_long), ('tv_nsec', ctypes.c_long))

if sys.platform.startswith('linux'):
    CLOCK_MONOTONIC = 1
elif sys.platform.startswith('freebsd'):
    CLOCK_MONOTONIC = 4
elif sys.platform.startswith('sunos5'):
    CLOCK_MONOTONIC = 4
elif 'bsd' in sys.platform:
    CLOCK_MONOTONIC = 3
elif sys.platform.startswith('aix'):
    CLOCK_MONOTONIC = ctypes.c_longlong(10)

def monotonic():
    """Monotonic clock, cannot go backward."""
    ts = timespec()
    if clock_gettime(CLOCK_MONOTONIC, ctypes.pointer(ts)):
        errno = ctypes.get_errno()
        raise OSError(errno, os.strerror(errno))
    return ts.tv_sec + ts.tv_nsec / 1.0e9

---

_{1: 版权所有 2014, 2015, 2016 Ori Livneh 根据 Apache 许可证 2.0 版（“许可证”）获得许可； 除非遵守许可，否则您不得使用此文件。 您可以在以下网址获取许可证的副本 http://www.apache.org/licenses/LICENSE-2.0 除非适用法律要求或书面同意，否则软件 根据许可分发是在“原样”基础上分发的， 没有任何明示或暗示的保证或条件。}