2

Надеюсь, на этот раз я не остановлен. Я некоторое время боролся с параллельной обработкой в ​​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) 

Еще раз спасибо, Валерий и любой который пытается помочь мне решить мою загадку.

ответ

0

Вы используете несколько аргументов, так:

import multiprocessing 
from functools import partial 
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('[]') 


if __name__ == "__main__": 
    tasks = [] 
    pool_outputs = [] 
    pool = multiprocessing.Pool(processes=5) 
    lines = [] 
    with open(filename, 'r') as infile: 
     for line in infile: 
      lines.append(line.rstrip())     
    tasks = lines 
    func = partial(search, pat) 
    if len(lines) > N: 
     pool_output = pool.map(func, lines) 
     pool_outputs.append(pool_output)  
    elif len(lines) > 0: 
     pool_output = pool.map(func, lines) 
     pool_outputs.append(pool_output) 
    pool.close() 
    pool.join() 
    print('Pool:', pool_outputs) 
+0

Валерий: Спасибо. Что же такое частичное? Знаете ли вы о каких-либо ресурсах, которые тщательно решают параллельную обработку в python? Еще раз спасибо. – corax

+0

https://docs.python.org/2/library/functools.html#functools.partial –

+0

Валерий: Я прочитал это и не мог этого понять. Мне жаль, но я имел в виду пример в функции. Благодарю. – corax