Надеюсь, на этот раз я не остановлен. Я некоторое время боролся с параллельной обработкой в Python (ровно 2 дня). Я проверить эти ресурсы (неполный список приведен здесь:Лучшие примеры параллельной обработки в Python
(а) http://eli.thegreenplace.net/2013/01/16/python-paralellizing-cpu-bound-tasks-with-concurrent-futures
(б) https://pythonadventures.wordpress.com/tag/processpoolexecutor/
Я отклеивался То, что я хочу сделать это:.
Master :
Break up the file into chunks(strings or numbers)
Broadcast a pattern to be searched to all the workers
Receive the offsets in the file where the pattern was found
Рабочие:
Receive pattern and chunk of text from the master
Compute()
Send back the offsets to the master.
Я попытался реализовать это с использованием MPI/concurrent.futures/multiprocessing и вышел из строя.
Моей наивная реализация с использованием MultiProcessing модуля
import multiprocessing
filename = "file1.txt"
pat = "afow"
N = 1000
""" This is the naive string search algorithm"""
def search(pat, txt):
patLen = len(pat)
txtLen = len(txt)
offsets = []
# A loop to slide pattern[] one by one
# Range generates numbers up to but not including that number
for i in range ((txtLen - patLen) + 1):
# Can not use a for loop here
# For loops in C with && statements must be
# converted to while statements in python
counter = 0
while(counter < patLen) and pat[counter] == txt[counter + i]:
counter += 1
if counter >= patLen:
offsets.append(i)
return str(offsets).strip('[]')
""""
This is what I want
if __name__ == "__main__":
tasks = []
pool_outputs = []
pool = multiprocessing.Pool(processes=5)
with open(filename, 'r') as infile:
lines = []
for line in infile:
lines.append(line.rstrip())
if len(lines) > N:
pool_output = pool.map(search, tasks)
pool_outputs.append(pool_output)
lines = []
if len(lines) > 0:
pool_output = pool.map(search, tasks)
pool_outputs.append(pool_output)
pool.close()
pool.join()
print('Pool:', pool_outputs)
"""""
with open(filename, 'r') as infile:
for line in infile:
print(search(pat, line))
Я был бы признателен за какие-либо указания, особенно с concurrent.futures. Спасибо за ваше время. Валерий помог мне с его дополнением, и я благодарю его за это.
Но если кто-то может просто баловаться меня на данный момент, это код, который я работал над для concurrent.futures (отработка примера я видел где-то)
from concurrent.futures import ProcessPoolExecutor, as_completed
import math
def search(pat, txt):
patLen = len(pat)
txtLen = len(txt)
offsets = []
# A loop to slide pattern[] one by one
# Range generates numbers up to but not including that number
for i in range ((txtLen - patLen) + 1):
# Can not use a for loop here
# For loops in C with && statements must be
# converted to while statements in python
counter = 0
while(counter < patLen) and pat[counter] == txt[counter + i]:
counter += 1
if counter >= patLen:
offsets.append(i)
return str(offsets).strip('[]')
#Check a list of strings
def chunked_worker(lines):
return {0: search("fmo", line) for line in lines}
def pool_bruteforce(filename, nprocs):
lines = []
with open(filename) as f:
lines = [line.rstrip('\n') for line in f]
chunksize = int(math.ceil(len(lines)/float(nprocs)))
futures = []
with ProcessPoolExecutor() as executor:
for i in range(nprocs):
chunk = lines[(chunksize * i): (chunksize * (i + 1))]
futures.append(executor.submit(chunked_worker, chunk))
resultdict = {}
for f in as_completed(futures):
resultdict.update(f.result())
return resultdict
filename = "file1.txt"
pool_bruteforce(filename, 5)
Еще раз спасибо, Валерий и любой который пытается помочь мне решить мою загадку.
Валерий: Спасибо. Что же такое частичное? Знаете ли вы о каких-либо ресурсах, которые тщательно решают параллельную обработку в python? Еще раз спасибо. – corax
https://docs.python.org/2/library/functools.html#functools.partial –
Валерий: Я прочитал это и не мог этого понять. Мне жаль, но я имел в виду пример в функции. Благодарю. – corax