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函数的返回值),然后在该进程上终止调用。但是,使用该map函数时,我不知道该怎么做。
我应该如何实现?
您可以使用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))