使用 concurrent.futures 或多处理的文件下载 - 如何在 Ctrl-C 上停止？

Question

我已经阅读了很多关于 SO 和其他地方关于这个主题的问题，但无法让它发挥作用。可能是因为我用的是Windows，我不知道。

我要做的是并行下载一堆文件（其 URL 是从 CSV 文件中读取的）。我已经尝试过使用multiprocessing 和concurrent.futures，但没有成功。

主要问题是我无法在 Ctrl-C 上停止程序 - 它只是继续运行。这在进程而不是线程的情况下尤其糟糕（我为此使用了multiprocessing），因为我每次都必须手动终止每个进程。

这是我当前的代码：

import concurrent.futures
import signal
import sys
import urllib.request

class Download(object):
  def __init__(self, url, filename):
    self.url = url
    self.filename = filename

def perform_download(download):
  print('Downloading {} to {}'.format(download.url, download.filename))
  return urllib.request.urlretrieve(download.url, filename=download.filename)  

def main(argv):
  args = parse_args(argv)
  queue = []
  with open(args.results_file, 'r', encoding='utf8') as results_file:
    # Irrelevant CSV parsing...
    queue.append(Download(url, filename))

  def handle_interrupt():
    print('CAUGHT SIGINT!!!!!!!!!!!!!!!!!!!11111111')
    sys.exit(1)

  signal.signal(signal.SIGINT, handle_interrupt)

  with concurrent.futures.ThreadPoolExecutor(max_workers=args.num_jobs) as executor:
    futures = {executor.submit(perform_download, d): d for d in queue}
    try:
      concurrent.futures.wait(futures)
    except KeyboardInterrupt:
      print('Interrupted')
      sys.exit(1)

我在这里尝试以两种不同的方式捕捉 Ctrl-C，但它们都不起作用。后一个（except KeyboardInterrupt）实际上已运行，但调用sys.exit后进程不会退出。

在此之前，我像这样使用multiprocessing 模块：

try:      
    pool = multiprocessing.Pool(processes=args.num_jobs)
    pool.map_async(perform_download, queue).get(1000000)
  except Exception as e:
    pool.close()
    pool.terminate()
    sys.exit(0)

那么，在终端中按下 Ctrl-C 后，添加终止所有工作线程或进程的能力的正确方法是什么？

系统信息：

• Python 版本：3.6.1 32 位
• 操作系统：Windows 10

Answer 1

您在信号处理程序中捕获SIGINT 信号并将其重新路由为SystemExit 异常。这可以防止KeyboardInterrupt 异常到达您的主循环。

此外，如果SystemExit 没有在主线程中引发，它只会杀死引发它的子线程。

multiprocessing 库的作者 Jesse Noller 解释了如何在旧的 blog post 中处理 CTRL+C。

import signal
from multiprocessing import Pool


def initializer():
    """Ignore CTRL+C in the worker process."""
    signal.signal(SIGINT, SIG_IGN)


pool = Pool(initializer=initializer)

try:
    pool.map(perform_download, dowloads)
except KeyboardInterrupt:
    pool.terminate()
    pool.join()

Answer 2

我不相信接受的答案在 Windows 下有效，当然在当前版本的 Python 下无效（我正在运行 3.8.5）。事实上，它根本不会运行，因为SIGINT 和SIG_IGN 将是未定义的（需要的是signal.SIGINT 和signal.SIG_IGN）。

这是 Windows 下的一个已知问题。我提出的解决方案基本上与公认的解决方案相反：主进程必须忽略键盘中断，我们将进程池初始化为最初将全局标志ctrl_c_entered 设置为False 和如果输入 Ctrl-C，则将此标志设置为 True。然后，任何多处理工作函数（或方法）都用一个特殊的装饰器handle_ctrl_c 进行装饰，它首先测试ctrl_c_entered 标志，并且只有当False 在重新启用键盘中断并建立尝试后它才会运行工作函数/catch 键盘中断处理程序。如果ctrl_c_entered标志为True或者在worker函数执行过程中发生键盘中断，则返回值是KeyboardInterrupt的一个实例，主进程可以检查该实例以确定是否输入了Ctrl-C .

因此，所有提交的任务都将被允许启动，但将立即终止并返回 KeyBoardInterrupt 异常的值，并且一旦输入 Ctrl-C，装饰器函数将永远不会调用实际的工作函数。

import signal
from multiprocessing import Pool
from functools import wraps
import time

def handle_ctrl_c(func):
    """
    Decorator function.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        global ctrl_c_entered
        if not ctrl_c_entered:
            # re-enable keyboard interrups:
            signal.signal(signal.SIGINT, default_sigint_handler)
            try:
                return func(*args, **kwargs)
            except KeyboardInterrupt:
                ctrl_c_entered = True
                return KeyboardInterrupt()
            finally:
                signal.signal(signal.SIGINT, pool_ctrl_c_handler)
        else:
            return KeyboardInterrupt()
    return wrapper

def pool_ctrl_c_handler(*args, **kwargs):
    global ctrl_c_entered
    ctrl_c_entered = True

def init_pool():
    # set global variable for each process in the pool:
    global ctrl_c_entered
    global default_sigint_handler
    ctrl_c_entered = False
    default_sigint_handler = signal.signal(signal.SIGINT, pool_ctrl_c_handler)

@handle_ctrl_c
def perform_download(download):
    print('begin')
    time.sleep(2)
    print('end')
    return True

if __name__ == '__main__':
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    pool = Pool(initializer=init_pool)
    results = pool.map(perform_download, range(20))
    if any(map(lambda x: isinstance(x, KeyboardInterrupt), results)):
        print('Ctrl-C was entered.')
    print(results)