如何获得使用系统正常运行时间的 Python threading.Timer 的等价物?
How can I get an equivalent of Python threading.Timer that uses system uptime?
TL;DR threading.Timer 使用系统时间,但是我使用的时候时间在变化,我怎样才能让它使用系统正常运行时间?
我有一个 Python 脚本可以做很多事情,其中之一是设置系统时间。当这个脚本启动时,时间是错误的。此脚本还需要有一个 30 秒的全局超时。
我一直在使用以下超时 class:
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
使用的系统时间在脚本为 运行 时发生了变化。无论如何我可以让它使用系统正常运行时间吗?
您似乎使用的是 Python < 3.3。在 Python 3.3 或更新版本中,monotonic
将是标准库中 time.monotonic
的别名。
time.monotonic
然后也用于 threading
-libary。正如 docs 中所说:
Return the value (in fractional seconds) of a monotonic clock, i.e. a clock that cannot go backwards. The clock is not affected by system clock updates.
所以,与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:您可以尝试将此作为 getTickCount1:
的替代方法
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 版("License")获得许可;
除非遵守许可证,否则您不得使用此文件。
您可以在以下位置获得许可证的副本
http://www.apache.org/licenses/LICENSE-2.0
除非适用法律要求或书面同意,软件
根据许可证分发是在 "AS IS" 基础上分发的,
没有任何明示或暗示的保证或条件。
更新
我现在正在做的是使用 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()
TL;DR threading.Timer 使用系统时间,但是我使用的时候时间在变化,我怎样才能让它使用系统正常运行时间?
我有一个 Python 脚本可以做很多事情,其中之一是设置系统时间。当这个脚本启动时,时间是错误的。此脚本还需要有一个 30 秒的全局超时。
我一直在使用以下超时 class:
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
使用的系统时间在脚本为 运行 时发生了变化。无论如何我可以让它使用系统正常运行时间吗?
您似乎使用的是 Python < 3.3。在 Python 3.3 或更新版本中,monotonic
将是标准库中 time.monotonic
的别名。
time.monotonic
然后也用于 threading
-libary。正如 docs 中所说:
Return the value (in fractional seconds) of a monotonic clock, i.e. a clock that cannot go backwards. The clock is not affected by system clock updates.
所以,与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:您可以尝试将此作为 getTickCount1:
的替代方法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 版("License")获得许可; 除非遵守许可证,否则您不得使用此文件。 您可以在以下位置获得许可证的副本 http://www.apache.org/licenses/LICENSE-2.0 除非适用法律要求或书面同意,软件 根据许可证分发是在 "AS IS" 基础上分发的, 没有任何明示或暗示的保证或条件。
更新
我现在正在做的是使用 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()