如何防止 Python 中的 KeyboardInterrupt 中断代码块？

当 dump操作正在保存数据时，按下 ctrl + c，中断将导致文件损坏(即仅部分写入，因此无法再次进行 loaded)。

有没有办法使 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

如果操作被中断，重新启动操作似乎很愚蠢，那么中断怎么可能被延迟呢？

27799 次浏览
小开
最佳答案

将函数放在一个线程中，然后等待线程完成。

除非使用特殊的 Capi，否则不能中断 Python 线程。

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()
小开

使用 信号模块在进程期间禁用 SIGINT:

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

在我看来，使用线程来完成这个任务是过分的。您可以通过简单的循环来确保文件保存正确，直到成功完成写操作:

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
小开

下面是附加 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()
小开

这个问题是关于阻塞 KeyboardInterrupt的，但是对于这种情况，我发现原子文件编写更加干净，并且提供了额外的保护。

使用原子写操作，要么正确地写入整个文件，要么什么都不写。Stackoverflow 有一个 不同的解决方案，但我个人喜欢使用 原子写作库。

在运行 pip install atomicwrites之后，像这样使用它: 

from atomicwrites import atomic_write


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

一个通用的方法是使用一个上下文管理器，它接受一组信号来挂起:

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))
小开

我一直在思考对这个问题答案的批评，我相信我已经实现了一个更好的解决方案，它是这样使用的:

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

下面是 signal_fence上下文管理器，然后解释了它对前面的答案所做的改进。这个函数的 docstring 记录了它的接口和保证。

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)上的任何异常都不会阻止主线程继续执行。< br > < br > 考虑一下，如果线程方法使用主线程在 KeyboardInterrupt之后的 finally块中变异的某些数据，会发生什么情况。

其次，使用上下文管理器来处理 @ benrg对最受欢迎答案的反馈:

  1. 如果在调用信号之后但在 __enter__返回之前引发异常，则信号将被永久阻塞;

我的解决方案在 @contextmanager装饰器的帮助下使用了生成器上下文管理器，从而避免了这个 bug。有关更多细节，请参见上面代码中的完整注释。

  1. 这段代码可以在主线程以外的线程中调用第三方异常处理程序，而 CPython 从不这样做;

我觉得这虫子不是真的。需要从主线程调用 signal.signal，否则引发 ValueError。这些上下文管理器只能在主线程上运行，因此只能从主线程调用第三方异常处理程序。

  1. 如果信号返回一个不可调用的值，__exit__将崩溃

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

请注意，signal_fence设计用于处理主线程(如 KeyboardInterrupt)上的 一中断。如果在恢复信号处理程序时，用户正在发送垃圾邮件 ctrl + c，那么没有多少能够拯救您。考虑到恢复处理程序所需执行的操作码相对较少，这不太可能，但是这是可能的。(为了获得最大的鲁棒性，这个解决方案需要是 用 C 重写的)

小开

这是不可中断的(尝试一下) ，但也维护了一个很好的接口，所以您的函数可以按照您预期的方式工作。

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)

提交答案