多处理 - 取消池中剩余的作业而不破坏池

Question

我正在使用 map_async 创建一个由 4 个工作人员组成的池。并为其提供要处理的图像文件列表 [Set 1]。
有时，我需要取消中间的处理，以便我可以处理一组不同的文件 [Set 2]。

所以一个例子是，我给了 map_async 1000 个文件来处理。然后想在处理完大约 200 个文件后取消剩余作业的处理。
此外，我想在不破坏/终止池的情况下进行此取消。这可能吗？

我不想终止池，因为在 Windows 上重新创建池是一个缓慢的过程（因为它使用“spawn”而不是“fork”）。而且我需要使用同一个池来处理一组不同的图像文件 [Set 2]..

# Putting job_set1 through processing. It may consist of 1000 images
cpu = multiprocessing.cpu_count()
pool = Pool(processes=cpu)
result = pool.map_async(job_set1, thumb_ts_list, chunksize=chunksize)

现在，我需要取消对这组 1 的处理。然后移动到另一个组（等待所有 1000 张图像完成处理不是一个选项，但我可以等待正在处理的当前图像完成)

<Somehow cancel processing of job_set1>
result = pool.map_async(job_set2, thumb_ts_list, chunksize=chunksize)

Answer 1

现在是fundamental theorem of software engineering 的时候了：虽然multiprocessing.Pool 不提供取消功能，但我们可以通过从精心设计的迭代中读取Pool 来添加它。然而，拥有一个从列表中获取yields 值但在某些信号上停止的生成器是不够的，因为Pool 急切地耗尽了给它的任何生成器。所以我们需要一个非常精心设计的可迭代对象。

一个懒惰的Pool

我们需要的通用工具是一种仅在工作人员可用时为Pool 构建任务的方法（或者最多提前一个任务，以防构建它们需要大量时间）。基本思想是减慢Pool 的线程收集工作，仅在任务完成时才增加信号量。 （我们从imap_unordered 的可观察行为中知道存在这样的线程。）

import multiprocessing
from threading import Semaphore

size=multiprocessing.cpu_count()  # or whatever Pool size to use

# How many workers are waiting for work?  Add one to buffer one task.
work=Semaphore(size)

def feed0(it):
  it=iter(it)
  try:
    while True:
      # Don't ask the iterable until we have a customer, in case better
      # instructions become available:
      work.acquire()
      yield next(it)
  except StopIteration: pass
  work.release()
def feed(p,f,it):
  import sys,traceback
  iu=p.imap_unordered(f,feed0(it))
  while True:
    try: x=next(iu)
    except StopIteration: return
    except Exception: traceback.print_exception(*sys.exc_info())
    work.release()
    yield x

feed 中的 try 可防止子级中的故障破坏信号量的计数，但请注意，它不能防止父级中的故障。

可取消的迭代器

现在我们可以实时控制Pool 输入，使任何调度策略都变得简单明了。例如，这里有类似 itertools.chain 的内容，但能够异步丢弃输入序列之一中的任何剩余元素：

import collections,queue

class Cancel:
  closed=False
  cur=()
  def __init__(self): self.data=queue.Queue() # of deques
  def add(self,d):
    d=collections.deque(d)
    self.data.put(d)
    return d
  def __iter__(self):
    while True:
      try: yield self.cur.popleft()
      except IndexError:
        self.cur=self.data.get()
        if self.cur is None: break
  @staticmethod
  def cancel(d): d.clear()
  def close(self): self.data.put(None)

尽管没有锁定，但这是线程安全的（至少在 CPython 中），因为像 deque.clear 这样的操作在 Python 检查方面是原子的（而且我们不会单独检查 self.cur 是否为空）。

用法

让其中一个看起来像

pool=mp.Pool(size)
can=Cancel()
many=can.add(range(1000))
few=can.add(["some","words"])
can.close()
for x in feed(pool,assess_happiness,can):
  if happy_with(x): can.cancel(many)  # straight onto few, then out

adds 和 close 当然可以自己在循环中。

Answer 2

multiprocessing 模块似乎没有取消的概念。您可以使用concurrent.futures.ProcessPoolExecutor 包装器并在您有足够的结果时取消待处理的期货。

这是一个示例，它从一组路径中挑选出 10 个 JPEG，并取消待处理的期货，同时让进程池在之后保持可用：

import concurrent.futures


def interesting_path(path):
    """Gives path if is a JPEG else ``None``."""
    with open(path, 'rb') as f:
        if f.read(3) == b'\xff\xd8\xff':
            return path
        return None


def find_interesting(paths, count=10):
     """Yields count from paths which are 'interesting' by multiprocess task."""
    with concurrent.futures.ProcessPoolExecutor() as pool:
        futures = {pool.submit(interesting_path, p) for p in paths}
        print ('Started {}'.format(len(futures)))
        for future in concurrent.futures.as_completed(futures):
            res = future.result()
            futures.remove(future)
            if res is not None:
                yield res
                count -= 1
                if count == 0:
                    break
        cancelled = 0
        for future in futures:
            cancelled += future.cancel()
        print ('Cancelled {}'.format(cancelled))
        concurrent.futures.wait(futures)
        # Can still use pool here for more processing as needed

请注意，选择如何将工作分解为期货仍然很棘手，更大的集合会带来更多开销，但也意味着浪费的工作更少。这也可以很容易地适应 Python 3.6 异步语法。