многопроцессорная работа с функцией python

Question

Как я могу реализовать многопроцессорную функцию для моей функции. Я так пробовал, но не работал.

def steric_clashes_parallel(system): 
    rna_st = system[MolWithResID("G")].molecule() 
    for i in system.molNums(): 
     peg_st = system[i].molecule() 
     if rna_st != peg_st: 
      print(peg_st) 
      for i in rna_st.atoms(AtomIdx()): 
       for j in peg_st.atoms(AtomIdx()): 
#     print(Vector.distance(i.evaluate().center(), j.evaluate().center())) 
        dist = Vector.distance(i.evaluate().center(), j.evaluate().center()) 
        if dist<2: 
         return print("there is a steric clash") 
    return print("there is no steric clashes") 

mix = PDB().read("clash_1.pdb") 
system = System() 
system.add(mix)  
from multiprocessing import Pool 
p = Pool(4) 
p.map(steric_clashes_parallel,system) 

У меня есть тысячи файлов pdb или системных файлов для проверки этой функции. Требуется 2 часа для одного файла на одном ядре без модуля многопроцессорности. Любое предложение было бы большой помощью.

Моя отслеживающий выглядит примерно так:

self.run() 
File "/home/sajid/sire.app/bundled/lib/python3.3/threading.py", line 858, 
    in run self._target(*self._args, **self._kwargs) 
    File "/home/sajid/sire.app/bundled/lib/python3.3/multiprocessing/pool.py", line 351, 
     in _handle_tasks put(task) 
     File "/home/sajid/sire.app/bundled/lib/python3.3/multiprocessing/connection.py", line 206, 
      in send ForkingPickler(buf, pickle.HIGHEST_PROTOCOL).dump(obj) 
RuntimeError: Pickling of "Sire.System._System.System" instances is not enabled 
(boost.org/libs/python/doc/v2/pickle.html) 

Answer 1

Проблема заключается в том, что Sire.System._System.System не может быть сериализовать, поэтому он не может быть отправлен в дочерний процесс. Для многопроцессорной обработки используется модуль pickle для сериализации, и вы можете часто выполнять проверку работоспособности в основной программе с помощью pickle.dumps(my_mp_object) для проверки.

У вас есть другая проблема, хотя (или я думаю, что вы делаете, основываясь на именах переменных). метод map принимает итерируемый и вентилятор своими итерированными объектами для объединения членов, но кажется, что вы хотите обработать system сам, а не что-то при этом итерации.

Один трюк для многопроцессорной обработки - это сохранить полезную нагрузку, которую вы отправляете от родителя к ребенку, и дать ребенку тяжелый подъем создания его объектов. Здесь вам может быть лучше просто отправить имена файлов и позволить детям выполнять большую часть работы.

def steric_clashes_from_file(filename): 
    mix = PDB().read(filename) 
    system = System() 
    system.add(mix)  
    steric_clashes_parallel(system) 

def steric_clashes_parallel(system): 
    rna_st = system[MolWithResID("G")].molecule() 
    for i in system.molNums(): 
     peg_st = system[i].molecule() 
     if rna_st != peg_st: 
      print(peg_st) 
      for i in rna_st.atoms(AtomIdx()): 
       for j in peg_st.atoms(AtomIdx()): 
#     print(Vector.distance(i.evaluate().center(), j.evaluate().center())) 
        dist = Vector.distance(i.evaluate().center(), j.evaluate().center()) 
        if dist<2: 
         return print("there is a steric clash") 
    return print("there is no steric clashes") 

filenames = ["clash_1.pdb",] 
from multiprocessing import Pool 
p = Pool(4, chunksize=1) 
p.map(steric_clashes_from_file,filenames) 

Answer 2

@ Мартино: Я проверил команду рассол, и он дал мне;

----> 1 pickle.dumps(clash_1.pdb) 
    RuntimeError: Pickling of "Sire.Mol._Mol.MoleculeGroup" instances is not enabled (http://www.boost.org/libs/python/doc/v2/pickle.html) 
    ----> 1 pickle.dumps(system) 
    RuntimeError: Pickling of "Sire.System._System.System" instances is not enabled (http://www.boost.org/libs/python/doc/v2/pickle.html) 

С вашим сценарием потребовалось то же самое время и использование только одного ядра. однако, строка dist iterable. Могу ли я запустить эту одиночную линию над многоядерными? Я изменяю строку как;

for i in rna_st.atoms(AtomIdx()): 
        icent = i.evaluate().center() 
        for j in peg_st.atoms(AtomIdx()): 
         dist = Vector.distance(icent, j.evaluate().center()) 

Answer 3

Существует один трюк вы можете сделать, чтобы получить более быстрое вычисление для каждого файла - обработки каждого файла последовательно, но обработки содержимого файла параллельно. Это зависит от ряда оговорок:

1. Вы работаете в системе, которая может обрабатывать процессы (например, Linux).
2. Вычисления, которые вы выполняете, не имеют побочных эффектов, которые влияют на результат будущих вычислений.

Похоже, что это имеет место в вашей ситуации, но я не могу быть уверенным на 100%.

Когда процесс разветвляется, вся память в дочернем процессе дублируется из родительского процесса (более того, он продублирован эффективным образом - биты памяти, которые только считываются, не дублируются). Это позволяет легко разделить большие, сложные начальные состояния между процессами. Однако, как только дочерние процессы запущены, они не будут видеть никаких изменений в объектах, выполненных в родительском процессе (и наоборот).

Пример код:

import multiprocessing 

system = None 
rna_st = None 

class StericClash(Exception): 
    """Exception used to halt processing of a file. Could be modified to 
    include information about what caused the clash if this is useful.""" 
    pass 


def steric_clashes_parallel(system_index): 
    peg_st = system[system_index].molecule() 
    if rna_st != peg_st: 
     for i in rna_st.atoms(AtomIdx()): 
      for j in peg_st.atoms(AtomIdx()): 
       dist = Vector.distance(i.evaluate().center(), 
        j.evaluate().center()) 
       if dist < 2: 
        raise StericClash() 


def process_file(filename): 
    global system, rna_st 

    # initialise global values before creating pool  
    mix = PDB().read(filename) 
    system = System() 
    system.add(mix) 
    rna_st = system[MolWithResID("G")].molecule() 

    with multiprocessing.Pool() as pool: 
     # contents of file processed in parallel 
     try: 
      pool.map(steric_clashes_parallel, range(system.molNums())) 
     except StericClash: 
      # terminate called to halt current jobs and further processing 
      # of file 
      pool.terminate() 
      # wait for pool processes to terminate before returning 
      pool.join() 
      return False 
     else: 
      pool.close() 
      pool.join() 
      return True 
     finally: 
      # reset globals 
      system = rna_st = None 

if __name__ == "__main__": 
    for filename in get_files_to_be_processed(): 
     # files are being processed in serial 
     result = process_file(filename) 
     save_result_to_disk(filename, result)