mp.set_start_method('spawn')
total_count = Counter(0)
pool = mp.Pool(initializer=init, initargs=(total_count,), processes=num_proc)
pool.map(part_crack_helper, product(seed_str, repeat=4))
pool.close()
pool.join()
所以我有一个工作进程池可以做一些工作。它只需要找到一个解决方案。因此,当其中一个工作进程找到解决方案时,我想停止一切。
我想到的一种方法是调用sys.exit((。但是,这似乎无法正常工作,因为其他进程正在运行。
另一种方法是检查每个进程调用的返回值(part_crack_helper函数的返回值(并在该进程上调用终止。但是,使用该地图功能时我不知道该怎么做。
我应该如何实现这一点?
您可以使用来自 Pool.apply_async
的回调。
像这样的东西应该为你做这项工作。
from multiprocessing import Pool
def part_crack_helper(args):
solution = do_job(args)
if solution:
return True
else:
return False
class Worker():
def __init__(self, workers, initializer, initargs):
self.pool = Pool(processes=workers,
initializer=initializer,
initargs=initargs)
def callback(self, result):
if result:
print("Solution found! Yay!")
self.pool.terminate()
def do_job(self):
for args in product(seed_str, repeat=4):
self.pool.apply_async(part_crack_helper,
args=args,
callback=self.callback)
self.pool.close()
self.pool.join()
print("good bye")
w = Worker(num_proc, init, [total_count])
w.do_job()
如果您同意使用其他库,则可以使用 Pebble 通过以下方式解决它。此解决方案的优点是可以另外指定超时。这意味着如果有一个成功的工人,或者如果它用完了时间,程序就会结束:
from pebble import ProcessPool, ProcessExpired
from concurrent.futures import TimeoutError
import time
pool = ProcessPool()
def my_function(args):
print("running " + str(args))
time.sleep((args + 1) * 30)
print("process won:" + str(args))
return True
start_time = time.time()
future = pool.map(my_function, range(4), timeout=65)
iterator = future.result()
while True:
try:
result = next(iterator)
if result:
pool.stop()
pool.join(timeout=0)
break
except StopIteration:
break
except TimeoutError as error:
print("function took longer than %d seconds" % error.args[1])
except ProcessExpired as error:
print("%s. Exit code: %d" % (error, error.exitcode))
except Exception as error:
print("function raised %s" % error)
print(error.traceback) # Python's traceback of remote process
print("whole time: " + str(time.time() - start_time))