Как заставить все процессы pool.apply_async остановиться, когда какой-либо один процесс нашел совпадение в python
У меня есть следующий код, который использует многопроцессорность, чтобы перебрать большой список и найти совпадение. Как я могу заставить все процессы остановиться, когда совпадение найдено в каком-либо одном процессе? Я видел примеры, но я ни один из них, кажется, не вписывается в то, что я делаю здесь.
#!/usr/bin/env python3.5
import sys, itertools, multiprocessing, functools
alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ12234567890!@#$%^&*?,()-=+[]/;"
num_parts = 4
part_size = len(alphabet) // num_parts
def do_job(first_bits):
for x in itertools.product(first_bits, *itertools.repeat(alphabet, num_parts-1)):
# CHECK FOR MATCH HERE
print(''.join(x))
# EXIT ALL PROCESSES IF MATCH FOUND
if __name__ == '__main__':
pool = multiprocessing.Pool(processes=4)
results = []
for i in range(num_parts):
if i == num_parts - 1:
first_bit = alphabet[part_size * i :]
else:
first_bit = alphabet[part_size * i : part_size * (i+1)]
pool.apply_async(do_job, (first_bit,))
pool.close()
pool.join()
Спасибо за ваше время.
ОБНОВЛЕНИЕ 1:
Я реализовал изменения, предложенные в великолепном подходе @ShadowRanger, и он почти работает так, как я хочу. Поэтому я добавил некоторые записи в журнал, чтобы дать представление о прогрессе, и добавил «тестовый» ключ для соответствия. Я хочу иметь возможность увеличивать / уменьшать iNumberOfProcessors независимо от num_parts. На этом этапе, когда у меня они оба на 4, все работает как положено, 4 процесса раскручиваются (один дополнительный для консоли). Когда я изменяю iNumberOfProcessors = 6, 6 процессов ускоряются, но только для них загрузка ЦП. Так что, похоже, 2 простаивают. Где, как и в моем предыдущем решении, я смог увеличить количество ядер без увеличения num_parts, и все процессы будут использованы.
Я не уверен, как изменить этот новый подход, чтобы дать мне ту же функциональность. Можете ли вы взглянуть и дать мне какое-то руководство по рефакторингу, необходимому для того, чтобы иметь возможность устанавливать iNumberOfProcessors и num_parts независимо друг от друга и при этом все еще использовать все процессы?
Вот обновленный код:
#!/usr/bin/env python3.5
import sys, itertools, multiprocessing, functools
alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ12234567890!@#$%^&*?,()-=+[]/;"
num_parts = 4
part_size = len(alphabet) // num_parts
iProgressInterval = 10000
iNumberOfProcessors = 6
def do_job(first_bits):
iAttemptNumber = 0
iLastProgressUpdate = 0
for x in itertools.product(first_bits, *itertools.repeat(alphabet, num_parts-1)):
sKey = ''.join(x)
iAttemptNumber = iAttemptNumber + 1
if iLastProgressUpdate + iProgressInterval <= iAttemptNumber:
iLastProgressUpdate = iLastProgressUpdate + iProgressInterval
print("Attempt#:", iAttemptNumber, "Key:", sKey)
if sKey == 'test':
print("KEY FOUND!! Attempt#:", iAttemptNumber, "Key:", sKey)
return True
def get_part(i):
if i == num_parts - 1:
first_bit = alphabet[part_size * i :]
else:
first_bit = alphabet[part_size * i : part_size * (i+1)]
return first_bit
if __name__ == '__main__':
# with statement with Py3 multiprocessing.Pool terminates when block exits
with multiprocessing.Pool(processes = iNumberOfProcessors) as pool:
# Don't need special case for final block; slices can
for gotmatch in pool.imap_unordered(do_job, map(get_part, range(num_parts))):
if gotmatch:
break
else:
print("No matches found")
ОБНОВЛЕНИЕ 2:
Хорошо, вот моя попытка попробовать предложение @noxdafox. Я собрал следующее на основе ссылки, которую он дал со своим предложением. К сожалению, когда я запускаю его, я получаю сообщение об ошибке:
... строка 322, в apply_async поднять ValueError («Пул не запущен») ValueError: Пул не запущен
Может ли кто-нибудь дать мне некоторое руководство о том, как заставить это работать.
По сути, проблема в том, что моя первая попытка выполняла многопроцессорную обработку, но не поддерживала отмену всех процессов после того, как было найдено совпадение.
Моя вторая попытка (основанная на предложении @ShadowRanger) решила эту проблему, но сломала функциональность возможности независимо масштабировать количество процессов и размер num_parts, что я мог сделать в своей первой попытке.
Моя третья попытка (на основе предложения @noxdafox) выдает ошибку, описанную выше.
Если кто-нибудь может дать мне какое-то указание о том, как сохранить функциональность моей первой попытки (возможность независимо масштабировать количество процессов и размер num_parts) и добавить функцию отмены всех процессов после того, как совпадение было найдено, это было бы очень полезно ,
Спасибо за ваше время.
Вот код из моей третьей попытки, основанный на предложении @noxdafox:
#!/usr/bin/env python3.5
import sys, itertools, multiprocessing, functools
alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ12234567890!@#$%^&*?,()-=+[]/;"
num_parts = 4
part_size = len(alphabet) // num_parts
iProgressInterval = 10000
iNumberOfProcessors = 4
def find_match(first_bits):
iAttemptNumber = 0
iLastProgressUpdate = 0
for x in itertools.product(first_bits, *itertools.repeat(alphabet, num_parts-1)):
sKey = ''.join(x)
iAttemptNumber = iAttemptNumber + 1
if iLastProgressUpdate + iProgressInterval <= iAttemptNumber:
iLastProgressUpdate = iLastProgressUpdate + iProgressInterval
print("Attempt#:", iAttemptNumber, "Key:", sKey)
if sKey == 'test':
print("KEY FOUND!! Attempt#:", iAttemptNumber, "Key:", sKey)
return True
def get_part(i):
if i == num_parts - 1:
first_bit = alphabet[part_size * i :]
else:
first_bit = alphabet[part_size * i : part_size * (i+1)]
return first_bit
def grouper(iterable, n, fillvalue=None):
args = [iter(iterable)] * n
return itertools.zip_longest(*args, fillvalue=fillvalue)
class Worker():
def __init__(self, workers):
self.workers = workers
def callback(self, result):
if result:
self.pool.terminate()
def do_job(self):
print(self.workers)
pool = multiprocessing.Pool(processes=self.workers)
for part in grouper(alphabet, part_size):
pool.apply_async(do_job, (part,), callback=self.callback)
pool.close()
pool.join()
print("All Jobs Queued")
if __name__ == '__main__':
w = Worker(4)
w.do_job()