如何在Python中防止代码块被KeyboardInterrupt打断？

Question

在执行数据保存操作时，按下 ctrl+c 会导致文件损坏（即仅写入部分内容，因此无法再次 load）。有没有办法让 dump 或者一般的代码块不可被中断？

---

我目前的解决方法类似于：

try:
    file = open(path, 'w')
    dump(obj, file)
    file.close()
except KeyboardInterrupt:
    file.close()
    file.open(path,'w')
    dump(obj, file)
    file.close()
    raise

如果操作被中断后重新开始似乎很傻，那么如何推迟中断呢？

- saffsd

Answer 1

以下是一个上下文管理器，它会为 SIGINT 附加一个信号处理程序。如果该上下文管理器的信号处理程序被调用，那么当上下文管理器退出时，只有将信号传递给原始处理程序才能延迟信号。

import signal
import logging

class DelayedKeyboardInterrupt:

    def __enter__(self):
        self.signal_received = False
        self.old_handler = signal.signal(signal.SIGINT, self.handler)
                
    def handler(self, sig, frame):
        self.signal_received = (sig, frame)
        logging.debug('SIGINT received. Delaying KeyboardInterrupt.')
    
    def __exit__(self, type, value, traceback):
        signal.signal(signal.SIGINT, self.old_handler)
        if self.signal_received:
            self.old_handler(*self.signal_received)

with DelayedKeyboardInterrupt():
    # stuff here will not be interrupted by SIGINT
    critical_code()

- Gary van der Merwe

---

虽然一开始可能看起来有些令人望而却步，但我认为这是最清晰、最可重复使用的解决方案。毕竟，你只需要定义上下文管理器一次（如果你喜欢，可以轻松地将其放在自己的模块中），然后在任何想要使用它的地方只需要一个“with”语句，这对于你代码的可读性来说是一个很大的优点。 - blubberdiblub

@Justin：这是因为信号处理程序只能在 Python 解释器的“原子”指令之间发生。（来自 https://docs.python.org/library/signal.html 的第三点） - Gary van der Merwe

非常好的课程，谢谢。我对其进行了扩展以支持同时处理多个信号 - 有时你还希望除了 SIGINT 之外也能响应 SIGTERM ：https://gist.github.com/tcwalther/ae058c64d5d9078a9f333913718bba95 - Thomas Walther

这段代码有漏洞，请勿使用。可能的漏洞列表：1.如果在调用signal后但在__enter__返回之前引发异常，则信号将永久地被阻塞；2.该代码可能会在非主线程中调用第三方异常处理程序，CPython从不这样做；3.如果signal返回一个非可调用值，则__exit__将崩溃。@ThomasWalther的版本部分修复了第三个漏洞，但增加了至少一个新漏洞。在Gist上有许多类似的类；所有这些都至少存在第1个漏洞。我建议不要尝试修复它们 - 这太难了，很难完全正确。 - benrg

哇，那是一种相当悲观的态度。你所描述的错误只会在非常偏僻的情况下遇到，可以轻松避免。仅因为这可能不适用于所有情况，并不意味着它对于没有任何情况都不合适。我认为你的评论很不建设性。 - Gary van der Merwe

Answer 2

将函数放入线程中，等待线程完成。
Python线程除非使用特殊的C API，否则无法被中断。

import time
from threading import Thread

def noInterrupt():
    for i in xrange(4):
        print i
        time.sleep(1)

a = Thread(target=noInterrupt)
a.start()
a.join()
print "done"


0
1
2
3
Traceback (most recent call last):
  File "C:\Users\Admin\Desktop\test.py", line 11, in <module>
    a.join()
  File "C:\Python26\lib\threading.py", line 634, in join
    self.__block.wait()
  File "C:\Python26\lib\threading.py", line 237, in wait
    waiter.acquire()
KeyboardInterrupt

看到了中断是如何被推迟直到线程完成的吗？

这里将其适应于您的使用：

import time
from threading import Thread

def noInterrupt(path, obj):
    try:
        file = open(path, 'w')
        dump(obj, file)
    finally:
        file.close()

a = Thread(target=noInterrupt, args=(path,obj))
a.start()
a.join()

- Unknown

---

这个解决方案比涉及 signal 模块的解决方案更好，因为它更容易正确使用。我不确定基于 signal 的解决方案是否可能编写得很健壮。 - benrg

好的，看起来在Python 3中，当你打断线程后，线程不会继续打印 - 中断立即出现，但线程仍在后台继续运行。 - Corvus

如果您想要从noInterrupt（）函数中返回某些内容怎么办？变量'a'会得到返回值吗？ - retro_coder

不要使用那段代码，因为它依赖于仅在Windows上存在的问题，所以在Linux或MacOS上无法工作：目前在Windows上，threading.Lock.acquire（因此其他同步原语如threading.Thread.join）无法通过Ctrl-C中断（参见https://bugs.python.org/issue29971）。 - Géry Ogam

我确认这在Linux上不起作用。 - undefined

Answer 3

使用signal模块，在进程运行期间禁用SIGINT：

s = signal.signal(signal.SIGINT, signal.SIG_IGN)
do_important_stuff()
signal.signal(signal.SIGINT, s)

- Ignacio Vazquez-Abrams

---

如果在类Unix系统中，我也会选择那个。 - Nadia Alramli

这确实适用于Windows操作系统。它是通过C运行时库模拟Posix信号来实现的。https://msdn.microsoft.com/en-us/library/xdkz3x12%28v=vs.90%29.aspx - Gary van der Merwe

比线程更简单易用，是个好的解决方案。谢谢。 - Josh Correia

如果在执行do_important_stuff()时发生信号，那么一旦信号被取消忽略，该信号会触发吗？ - Sean

我认为这通常是最清晰的解决方案。唯一的问题是当你不在主线程工作时：“信号只能在主线程中工作”。哼。 - mozzbozz

Answer 4

在我看来，使用线程完成这个任务有些大材小用。你可以通过简单地使用循环直到成功写入文件为止来确保文件被正确保存：

def saveToFile(obj, filename):
    file = open(filename, 'w')
    cPickle.dump(obj, file)
    file.close()
    return True

done = False
while not done:
    try:
        done = saveToFile(obj, 'file')
    except KeyboardInterrupt:
        print 'retry'
        continue

- Nadia Alramli

---

+1：这种方法比其他两种更符合Python风格，更易于理解。 - kquinn

这种方法不够好，因为你可以通过按住crtl+c来永远中断它，而我的线程方法从不被中断。另外需要注意的是，你必须拥有另一个变量 "isinterrupted" 和另一个条件语句来重新引发它。 - Unknown

这种方法每次都会重新启动转储，这正是我想避免的部分。 - saffsd

@Unknown，@Saffsd：你们两个都是对的。但是这个解决方案适用于简单的应用程序，您不希望出现恶意使用。这是一个解决用户无意中中断转储的非常不可能事件的方法。您可以选择最适合您的应用程序的解决方案。 - Nadia Alramli

不，@Saffsd 是不对的。他应该将 dump() 移出 saveToFile()。然后只调用一次 dump()，并根据需要多次调用 saveToFile()。 - Pavel Vlasov

实际上，对我来说禁用信号似乎是最自然的方法。在这种情况下，我们不会抑制它们，只需重复操作，直到我们的信号没有被抑制（如果我们很幸运的话）。这几乎完全与问题的基础作者所写的一样。这就是为什么我会投票反对这个答案的原因。 - Drachenfels

Answer 5

我一直在思考关于这个问题答案的批评，我相信我实现了一个更好的解决方案，可以这样使用：

with signal_fence(signal.SIGINT):
  file = open(path, 'w')
  dump(obj, file)
  file.close()

< p > signal_fence 上下文管理器如下所示，其改进了先前答案的方法。此函数的文档字符串记录了其接口和保证。 < /p >

import os
import signal
from contextlib import contextmanager
from types import FrameType
from typing import Callable, Iterator, Optional, Tuple
from typing_extensions import assert_never


@contextmanager
def signal_fence(
    signum: signal.Signals,
    *,
    on_deferred_signal: Callable[[int, Optional[FrameType]], None] = None,
) -> Iterator[None]:
    """
    A `signal_fence` creates an uninterruptible "fence" around a block of code. The
    fence defers a specific signal received inside of the fence until the fence is
    destroyed, at which point the original signal handler is called with the deferred
    signal. Multiple deferred signals will result in a single call to the original
    handler. An optional callback `on_deferred_signal` may be specified which will be
    called each time a signal is handled while the fence is active, and can be used
    to print a message or record the signal.

    A `signal_fence` guarantees the following with regards to exception-safety:

    1. If an exception occurs prior to creating the fence (installing a custom signal
    handler), the exception will bubble up as normal. The code inside of the fence will
    not run.
    2. If an exception occurs after creating the fence, including in the fenced code,
    the original signal handler will always be restored before the exception bubbles up.
    3. If an exception occurs while the fence is calling the original signal handler on
    destruction, the original handler may not be called, but the original handler will
    be restored. The exception will bubble up and can be detected by calling code.
    4. If an exception occurs while the fence is restoring the original signal handler
    (exceedingly rare), the original signal handler will be restored regardless.
    5. No guarantees about the fence's behavior are made if exceptions occur while
    exceptions are being handled.

    A `signal_fence` can only be used on the main thread, or else a `ValueError` will
    raise when entering the fence.
    """
    handled: Optional[Tuple[int, Optional[FrameType]]] = None

    def handler(signum: int, frame: Optional[FrameType]) -> None:
        nonlocal handled
        if handled is None:
            handled = (signum, frame)
        if on_deferred_signal is not None:
            try:
                on_deferred_signal(signum, frame)
            except:
                pass

    # https://docs.python.org/3/library/signal.html#signal.getsignal
    original_handler = signal.getsignal(signum)
    if original_handler is None:
        raise TypeError(
            "signal_fence cannot be used with signal handlers that were not installed"
            " from Python"
        )
    if isinstance(original_handler, int) and not isinstance(
        original_handler, signal.Handlers
    ):
        raise NotImplementedError(
            "Your Python interpreter's signal module is using raw integers to"
            " represent SIG_IGN and SIG_DFL, which shouldn't be possible!"
        )

    # N.B. to best guarantee the original handler is restored, the @contextmanager
    #      decorator is used rather than a class with __enter__/__exit__ methods so
    #      that the installation of the new handler can be done inside of a try block,
    #      whereas per [PEP 343](https://www.python.org/dev/peps/pep-0343/) the
    #      __enter__ call is not guaranteed to have a corresponding __exit__ call if an
    #      exception interleaves
    try:
        try:
            signal.signal(signum, handler)
            yield
        finally:
            if handled is not None:
                if isinstance(original_handler, signal.Handlers):
                    if original_handler is signal.Handlers.SIG_IGN:
                        pass
                    elif original_handler is signal.Handlers.SIG_DFL:
                        signal.signal(signum, signal.SIG_DFL)
                        os.kill(os.getpid(), signum)
                    else:
                        assert_never(original_handler)
                elif callable(original_handler):
                    original_handler(*handled)
                else:
                    assert_never(original_handler)
            signal.signal(signum, original_handler)
    except:
        signal.signal(signum, original_handler)
        raise

首先，为什么不使用线程(被接受的答案)?
在非守护线程中运行代码确保该线程会在解释器关闭时被加入，但是主线程上的任何异常(例如KeyboardInterrupt)都不能阻止主线程继续执行。

考虑一下如果线程方法正在使用主线程在一个finally块中变异的一些数据，而此时主线程发生了KeyboardInterrupt会发生什么。

其次，针对@benrg对使用上下文管理器的最高投票答案的反馈:

1. 如果在调用信号前抛出异常，但在__enter__返回之前，信号将永久性地被阻塞;

我的解决方案通过使用生成器上下文管理器及@contextmanager装饰器避免了这个漏洞。有关详细信息，请参见上面的代码中的完整注释。

2. 此代码可能在除主线程之外的其他线程中调用第三方异常处理程序，CPython从不这样做;

我不认为这个漏洞是真实存在的。必须从主线程中调用signal.signal，否则会引发ValueError。这些上下文管理器只能在主线程上运行，因此只会从主线程调用第三方异常处理程序。

3. 如果信号返回一个非可调用值，则__exit__将崩溃

我的解决方案处理了信号处理程序的所有可能值，并适当地调用它们。此外，我使用assert_never从静态分析器中受益。

---

请注意，signal_fence设计用于处理主线程上的一个中断，例如KeyboardInterrupt。如果用户在信号处理程序被恢复时不断按ctrl+c，那么没有什么可以拯救你。鉴于需要执行的操作码相对较少以恢复处理程序，这种情况不太可能发生。（为最大的健壮性，该解决方案需要以C语言重写）

- Brendan Batliner

---

不能在Windows/其他平台上运行，但线程版本可以。 - Erik Aronesty

我刚刚在Windows 10 / Python 3.10上测试了一下，似乎可以正常工作。signal_fence(signal.SIGINT)正确地推迟了我的命令提示符中的KeyboardInterrupt。这个答案中还有一些其他评论表明Python在Windows上模拟这些Unix信号，使用CTRL_C_EVENT/CTRL_BREAK_EVENT。 - Brendan Batliner

Answer 6

这个问题是关于阻止KeyboardInterrupt，但我发现原子文件写入更加干净并提供额外的保护。

使用原子写入，要么整个文件被正确地写入，要么什么也不会发生。Stackoverflow有各种解决方案，但我个人喜欢只使用atomicwrites库。

运行pip install atomicwrites后，只需像这样使用它：

from atomicwrites import atomic_write

with atomic_write(path, overwrite=True) as file:
    dump(obj, file)

- Chris

Answer 7

一种通用的方法是使用一个上下文管理器，接受一组要挂起的信号：

import signal

from contextlib import contextmanager


@contextmanager
def suspended_signals(*signals):
    """
    Suspends signal handling execution
    """
    signal.pthread_sigmask(signal.SIG_BLOCK, set(signals))
    try:
        yield None
    finally:
        signal.pthread_sigmask(signal.SIG_UNBLOCK, set(signals))

- Maxim Kirilov

Answer 8

这不可中断（可以试试），但同时保持良好的接口，让您的函数按照您期望的方式运行。

import concurrent.futures
import time

def do_task(func):
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as run:
        fut = run.submit(func)
        return fut.result()


def task():
    print("danger will robinson")
    time.sleep(5)
    print("all ok")

do_task(task)

这里有一种简单的方法可以创建一个不需要信号处理的不间断睡眠：

def uninterruptible_sleep(secs):
    fut = concurrent.futures.Future()
    with contextlib.suppress(concurrent.futures.TimeoutError):
        fut.result(secs)

- Erik Aronesty

Answer 9

这是我尝试创建的一个上下文管理器，可以根据https://dev59.com/l3RA5IYBdhLWcg3wvgob#71330357中的答案来延迟信号处理。

import signal
from contextlib import contextmanager

@contextmanager
def defer_signal(signum):
    # Based on https://dev59.com/l3RA5IYBdhLWcg3wvgob#71330357

    original_handler = None
    defer_handle_args = None

    def defer_handle(*args):
        nonlocal defer_handle_args
        defer_handle_args = args

    # Do nothing if
    # - we don't have a registered handler in Python to defer
    # - or the handler is not callable, so either SIG_DFL where the system
    #   takes some default action, or SIG_IGN to ignore the signal
    # - or we're not in the main thread that doesn't get signals anyway
    original_handler = signal.getsignal(signum)
    if (
            original_handler is None
            or not callable(original_handler)
            or threading.current_thread() is not threading.main_thread()
    ):
        yield
        return

    try:
        signal.signal(signum, defer_handle)
        yield
    finally:
        signal.signal(signum, original_handler)
        if defer_handle_args is not None:
            original_handler(*defer_handle_args)

可以用作以下用途：

with defer_signal(signal.SIGINT):
   # code to not get interrupted by SIGINT

主要的区别：

• 不会尝试推迟非Python处理程序，例如，如果系统默认设置了处理程序，或者信号被忽略
• 在调用期间发生延迟期间的信号之前，恢复原始的单个处理程序
• 它是“最后一个信号获胜”，而不是“第一个信号获胜”。不确定这是否是一个有意义的区别，但它更简单一些。

但仍然存在一些情况下可能永远无法恢复原始处理程序...比如说，如果另一个信号有另一个处理程序，在finally:之后引发异常。由于异常可以在任何地方被信号处理程序“引发”，在纯Python中可能是不可能的。

但是-如果您不添加这样的处理程序，只关心SIGINT/KeyboardInterrupt，那么我认为它是健壮的(?)

- Michal Charemza