编程语言中的 timeout 原理是什么？

Question

比如下面代码示例中的 timeout

import requests

requests.get(url = 'http://www.google.com.hk', timeout=5)

代码中不存在一个计时线程来实现 timeout，我盲猜，实现的原理就是：用户程序向操作系统注册一个 timeout 的 timer，时间到了，操作系统就给应用程序一个中断信号。是这样吗？

如果我想自己实现一个任意 timeout 的程序，应该怎么做？

不限于 HTTP 的 timeout

OS 提供了什么 API 让我做这件事情？

有什么资料可供参考，使得我了解更多吗？

我需要的是一个任意解，调用任何即便没有任何系统调用的函数，都可以任意设置 timeout 的解决方案

Answer 1

不一定，有的是定时器，有的是用 select 模拟的，有的是操作系统 API 本身就提供了 timeout 相关设置。

但要是具体到某一个接口上的话，我们看源码就知道它是如何实现的了：

# REF: https://github.com/psf/requests/blob/main/requests/adapters.py#L499

resp = conn.urlopen(
  method=request.method,
  url=url,
  body=request.body,
  headers=request.headers,
  redirect=False,
  assert_same_host=False,
  preload_content=False,
  decode_content=False,
  retries=self.max_retries,
  timeout=timeout,
)

timeout 参数最后传进了 conn.urlopen 方法里，而 conn 是 urllib3 库里提供的对象，那我们接着跟踪：

# REF: https://github.com/urllib3/urllib3/blob/main/src/urllib3/util/wait.py

def poll_wait_for_socket(
    sock: socket.socket,
    read: bool = False,
    write: bool = False,
    timeout: Optional[float] = None,
) -> bool:
    if not read and not write:
        raise RuntimeError("must specify at least one of read=True, write=True")
    mask = 0
    if read:
        mask |= select.POLLIN
    if write:
        mask |= select.POLLOUT
    poll_obj = select.poll()
    poll_obj.register(sock, mask)

    # For some reason, poll() takes timeout in milliseconds
    def do_poll(t: Optional[float]) -> List[Tuple[int, int]]:
        if t is not None:
            t *= 1000
        return poll_obj.poll(t)

    return bool(do_poll(timeout))


def _have_working_poll() -> bool:
    # Apparently some systems have a select.poll that fails as soon as you try
    # to use it, either due to strange configuration or broken monkeypatching
    # from libraries like eventlet/greenlet.
    try:
        poll_obj = select.poll()
        poll_obj.poll(0)
    except (AttributeError, OSError):
        return False
    else:
        return True


def wait_for_socket(
    sock: socket.socket,
    read: bool = False,
    write: bool = False,
    timeout: Optional[float] = None,
) -> bool:
    # We delay choosing which implementation to use until the first time we're
    # called. We could do it at import time, but then we might make the wrong
    # decision if someone goes wild with monkeypatching select.poll after
    # we're imported.
    global wait_for_socket
    if _have_working_poll():
        wait_for_socket = poll_wait_for_socket
    elif hasattr(select, "select"):
        wait_for_socket = select_wait_for_socket
    return wait_for_socket(sock, read, write, timeout)

发现是用 select 模拟的。

最后 Python 中如果你想用的比较通用的超时机制的话，可以用 eventlet.Timeout/gevent.Timeout 来实现。你也可以像你在题目里说的那样，用 signal 自己封装一下，不过由于 Windows 下不支持 signal，所以如果你有跨平台的需要的话就不能这么做了。

Answer 2

用signal可以在线程级别上timeout, 具体看看代码

import signal
from contextlib import contextmanager
from functools import wraps


@contextmanager
def timeout_signal(second):
    signal.signal(signal.SIGALRM, raise_timeout)
    signal.alarm(second)
    try:
        yield
    finally:
        signal.signal(signal.SIGALRM, signal.SIG_IGN)


def raise_timeout(_signum, _frame):
    raise TimeoutError


def timeout(second):
    def _timeout(fun):
        @wraps(fun)
        def _fun(*args, **kwargs):
            with timeout_signal(second):
                return fun(*args, **kwargs)

        return _fun

    return _timeout


@timeout(1)
def f():
    time.sleep(2)

f()