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

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 来添加它。然而，拥有一个从列表中生成值但在某些信号上停止的生成器还不够，因为池会急切地耗尽提供给它的任何生成器。所以我们需要一个非常精心设计的迭代器。

懒惰的池

我们需要的通用工具是一种仅当工作人员可用时才为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

当然，add 和 close 本身可以在循环中。