Как обмениваться и обновлять объекты классов / переменных в многопроцессорных системах, каждый из которых имеет несколько потоков? - PullRequest
Новая
       23

Как обмениваться и обновлять объекты классов / переменных в многопроцессорных системах, каждый из которых имеет несколько потоков?

0 голосов
/ 28 января 2019

Я настраиваю пользовательский сканер глубокого файла для файла очень большого размера (в диапазоне 10 ГБ +).Я должен оптимизировать поиск этих файлов по «n» количеству ключевых слов, и такие значения, как номер строки и количество для расчета.Таким образом, регулярное применение алгоритмов должно быть изменено, чтобы соответствовать особым потребностям.

Я также использую многопроцессорность, чтобы разделить сканирование каждого файла на разные процессы.в рамках каждого такого процесса, поскольку размеры файлов варьируются от нескольких килобайт до нескольких 100 Мб, я решил разделить строки и использовать многопоточность в каждом порожденном процессе, чтобы сканировать набор строк для ключевых слов 'n'.Таким образом, каждый поток будет иметь одну строку одного файла и будет сканировать ключевые слова n над этой строкой и получать необходимые данные (многопоточность, поскольку содержимое файла и другие данные передаются совместно).

Сначала: это хороший дизайн?

Во-вторых: как эффективно реализовать глубокий сканер ключевых слов?

В-третьих: я хочу использовать функции Pool.map для обработки многопоточных и многопроцессорных процессов, как наилучшим образом использовать общие переменные, используя это.?

Я пытался использовать очереди, значение и, наконец, процесс сервера диспетчера для передачи значений между процессами и потоками.Я запутался, как получить мой результат. 

from multiprocessing import cpu_count, Pool, Manager, 
from contextlib import contextmanager
from multiprocessing.dummy import Pool as ThreadPool
from itertools import count, product

key_list = []

@contextmanager
def poolContext(*args, **kwargs):
    pool = Pool(*args, **kwargs)
    yield pool
    pool.terminate()

class FileDetails:
    listvar = [] 
    def __init__ (self, *args):
        file_name =filename
        file_path = filepath


class KeyWord:
    dictvar1 = defaultdict()
    dictvar2 = defaultdict ()

    def __init__(self,key_name,lang_name):
        self.name = key_name
        self.lang_name = lang_name

    def __call__(self,*args):
        * update the dictvars1,dictvar2 with the arguments passed *

def scanEngine(line_no, line, key, filename,  key_obj_list)
    * scan algorithm *
    *if condition is satisfied *
    for keyobj in key_obj_list:
        if keyobj.name == key:
            * update the object parameter *
        else: 
            continue

def unpacker2(args):
    scan_engine(*args)

def threadingFunct(file_obj,key_obj_list):
    args_list_thread = []
    tpool = ThreadPool(4)
    file_name = file_obj.file_name

    with open(file_obj.file_path,'r') as content : 
        file_content = content.readlines()

    for line_no,line in enumerate(file_content):
       for keyword in file_obj.key_list:
           args_list_thread.append(line_no+1,line, keyword,file_name,  key_obj_list)    #i want to update the details in the key_object_list for each keyword scanned.so its passed to the main scan engine function.

    results = tpool.map(unpacker2, args_list_thread)

def unpacker(args):
    return threadingFunct(*args)

def multiprocessingFunct(filtered_file_lst):
    args_list = [] 
    manager = Manager() 
    manager_key_list = manager.List(key_list) #so that i can access is the processes.
    for obj in filtered_file_lst):
        args_list.append((obj, key_list)) # for the pool arguments.. getting each object a copy of the object2_list
    with poolContext(processes = cpu_count) as p:
        results = p.map(unpacker ,args_list[])


def main(arg1, arg2):
    file_list = getObject1List(*args) #function defined to create file objects
    key_list = getObject2List(*args)  #function defined to create keyword objects
    for obj in file_list:
        if condition== True:
            filtered_file_list.append(obj)

    multiprocessingFunct(filtered_file_list)

ОЖИДАЕМЫЙ РЕЗУЛЬТАТ: После каждого выполнения одного положительного потока объект Keyword с соответствующим ключевым словом должен добавлять к нему значения, включая номер строки без повторений.

...