Подтвердить что ты не робот

Использование модуля Pipon Multiprocessing для одновременного и отдельного запуска моделей SEAWAT/MODFLOW

Я пытаюсь выполнить 100 моделей на моей 8-процессорной 64-разрядной машине Windows 7. Я бы хотел запустить 7 экземпляров модели одновременно, чтобы уменьшить общее время работы (около 9,5 минут на модельный пробег). Я просмотрел несколько потоков, относящихся к модулю многопроцессорности Python, но я все еще что-то пропустил.

Использование модуля многопроцессорности

Как создать параллельные дочерние процессы в многопроцессорной системе?

Многопроцессорная очередь Python

Мой процесс:

У меня есть 100 различных наборов параметров, которые я бы хотел запустить через SEAWAT/MODFLOW, чтобы сравнить результаты. Я предварительно создал входные файлы модели для каждого запуска модели и сохранил их в своих собственных каталогах. То, что я хотел бы иметь, это иметь 7 моделей, работающих одновременно, до тех пор, пока все реализации не будут завершены. Не должно быть связи между процессами или отображением результатов. До сих пор я смог последовательно генерировать модели:

import os,subprocess
import multiprocessing as mp

ws = r'D:\Data\Users\jbellino\Project\stJohnsDeepening\model\xsec_a'
files = []
for f in os.listdir(ws + r'\fieldgen\reals'):
    if f.endswith('.npy'):
        files.append(f)

## def work(cmd):
##     return subprocess.call(cmd, shell=False)

def run(f,def_param=ws):
    real = f.split('_')[2].split('.')[0]
    print 'Realization %s' % real

    mf2k = r'c:\modflow\mf2k.1_19\bin\mf2k.exe '
    mf2k5 = r'c:\modflow\MF2005_1_8\bin\mf2005.exe '
    seawatV4 = r'c:\modflow\swt_v4_00_04\exe\swt_v4.exe '
    seawatV4x64 = r'c:\modflow\swt_v4_00_04\exe\swt_v4x64.exe '

    exe = seawatV4x64
    swt_nam = ws + r'\reals\real%s\ss\ss.nam_swt' % real

    os.system( exe + swt_nam )


if __name__ == '__main__':
    p = mp.Pool(processes=mp.cpu_count()-1) #-leave 1 processor available for system and other processes
    tasks = range(len(files))
    results = []
    for f in files:
        r = p.map_async(run(f), tasks, callback=results.append)

Я изменил if __name__ == 'main': на следующее, надеясь, что он исправит недостаток parallelism, который, как мне кажется, передается на script выше for loop. Тем не менее, модель не может работать даже без ошибки Python:

if __name__ == '__main__':
    p = mp.Pool(processes=mp.cpu_count()-1) #-leave 1 processor available for system and other processes
    p.map_async(run,((files[f],) for f in range(len(files))))

Любая помощь приветствуется!

РЕДАКТИРОВАТЬ 3/26/2012 13:31 EST

Использование метода "Ручной пул" в @J.F. Ответ Себастьяна ниже. Я получаю параллельное выполнение моего внешнего .exe. Реализация модели вызывается партиями по 8 за раз, но она не дожидается завершения этих 8 прогонов до вызова следующей партии и т.д.:

from __future__ import print_function
import os,subprocess,sys
import multiprocessing as mp
from Queue import Queue
from threading import Thread

def run(f,ws):
    real = f.split('_')[-1].split('.')[0]
    print('Realization %s' % real)
    seawatV4x64 = r'c:\modflow\swt_v4_00_04\exe\swt_v4x64.exe '
    swt_nam = ws + r'\reals\real%s\ss\ss.nam_swt' % real
    subprocess.check_call([seawatV4x64, swt_nam])

def worker(queue):
    """Process files from the queue."""
    for args in iter(queue.get, None):
        try:
            run(*args)
        except Exception as e: # catch exceptions to avoid exiting the
                               # thread prematurely
            print('%r failed: %s' % (args, e,), file=sys.stderr)

def main():
    # populate files
    ws = r'D:\Data\Users\jbellino\Project\stJohnsDeepening\model\xsec_a'
    wdir = os.path.join(ws, r'fieldgen\reals')
    q = Queue()
    for f in os.listdir(wdir):
        if f.endswith('.npy'):
            q.put_nowait((os.path.join(wdir, f), ws))

    # start threads
    threads = [Thread(target=worker, args=(q,)) for _ in range(8)]
    for t in threads:
        t.daemon = True # threads die if the program dies
        t.start()

    for _ in threads: q.put_nowait(None) # signal no more files
    for t in threads: t.join() # wait for completion

if __name__ == '__main__':

    mp.freeze_support() # optional if the program is not frozen
    main()

Отсутствует трассировка ошибок. Функция run() выполняет свою обязанность при вызове одного файла реализации модели, как с несколькими файлами. Единственное различие заключается в том, что с несколькими файлами он называется len(files) раз, хотя каждый из экземпляров немедленно закрывается, и только один запуск модели разрешен, и в это время script выходит изящно (код выхода 0).

Добавление некоторых операторов печати в main() показывает некоторую информацию об активных подсчетах потоков, а также о состоянии потока (обратите внимание, что это тест только для 8 файлов реализации, чтобы сделать снимок экрана более управляемым, теоретически все 8 файлов должны запускаться одновременно, однако поведение продолжается там, где они появляются и сразу умирают, кроме одного):

def main():
    # populate files
    ws = r'D:\Data\Users\jbellino\Project\stJohnsDeepening\model\xsec_a'
    wdir = os.path.join(ws, r'fieldgen\test')
    q = Queue()
    for f in os.listdir(wdir):
        if f.endswith('.npy'):
            q.put_nowait((os.path.join(wdir, f), ws))

    # start threads
    threads = [Thread(target=worker, args=(q,)) for _ in range(mp.cpu_count())]
    for t in threads:
        t.daemon = True # threads die if the program dies
        t.start()
    print('Active Count a',threading.activeCount())
    for _ in threads:
        print(_)
        q.put_nowait(None) # signal no more files
    for t in threads: 
        print(t)
        t.join() # wait for completion
    print('Active Count b',threading.activeCount())

screenshot

** Строка, которая читает "D:\\Data\\Users...", - это информация об ошибке, которая была сброшена, когда я вручную остановил работу модели до ее завершения. Как только я остановлю работу модели, появятся остальные строки состояния потока, и script завершает работу.

РЕДАКТИРОВАТЬ 3/26/2012 16:24 EST

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

Я принимаю @J.F. Себастьян отвечает, так как вполне вероятно, что, как только я разрешу эту проблему с исполнением модели, предоставленный им код потоковой передачи получит меня там, где мне нужно.

ЗАКЛЮЧИТЕЛЬНЫЙ КОД

Добавлен аргумент cwd в subprocess.check_call для запуска каждого экземпляра SEAWAT в его собственном каталоге. Очень важно.

from __future__ import print_function
import os,subprocess,sys
import multiprocessing as mp
from Queue import Queue
from threading import Thread
import threading

def run(f,ws):
    real = f.split('_')[-1].split('.')[0]
    print('Realization %s' % real)
    seawatV4x64 = r'c:\modflow\swt_v4_00_04\exe\swt_v4x64.exe '
    cwd = ws + r'\reals\real%s\ss' % real
    swt_nam = ws + r'\reals\real%s\ss\ss.nam_swt' % real
    subprocess.check_call([seawatV4x64, swt_nam],cwd=cwd)

def worker(queue):
    """Process files from the queue."""
    for args in iter(queue.get, None):
        try:
            run(*args)
        except Exception as e: # catch exceptions to avoid exiting the
                               # thread prematurely
            print('%r failed: %s' % (args, e,), file=sys.stderr)

def main():
    # populate files
    ws = r'D:\Data\Users\jbellino\Project\stJohnsDeepening\model\xsec_a'
    wdir = os.path.join(ws, r'fieldgen\reals')
    q = Queue()
    for f in os.listdir(wdir):
        if f.endswith('.npy'):
            q.put_nowait((os.path.join(wdir, f), ws))

    # start threads
    threads = [Thread(target=worker, args=(q,)) for _ in range(mp.cpu_count()-1)]
    for t in threads:
        t.daemon = True # threads die if the program dies
        t.start()
    for _ in threads: q.put_nowait(None) # signal no more files
    for t in threads: t.join() # wait for completion

if __name__ == '__main__':
    mp.freeze_support() # optional if the program is not frozen
    main()
4b9b3361

ОТВЕТЫ

Ответ 1

Я не вижу никаких вычислений в коде Python. Если вам просто нужно выполнить несколько внешних программ параллельно, достаточно использовать subprocess для запуска программ и модуля threading для поддержания постоянного количества запущенных процессов, но простейший код использует multiprocessing.Pool:

#!/usr/bin/env python
import os
import multiprocessing as mp

def run(filename_def_param): 
    filename, def_param = filename_def_param # unpack arguments
    ... # call external program on `filename`

def safe_run(*args, **kwargs):
    """Call run(), catch exceptions."""
    try: run(*args, **kwargs)
    except Exception as e:
        print("error: %s run(*%r, **%r)" % (e, args, kwargs))

def main():
    # populate files
    ws = r'D:\Data\Users\jbellino\Project\stJohnsDeepening\model\xsec_a'
    workdir = os.path.join(ws, r'fieldgen\reals')
    files = ((os.path.join(workdir, f), ws)
             for f in os.listdir(workdir) if f.endswith('.npy'))

    # start processes
    pool = mp.Pool() # use all available CPUs
    pool.map(safe_run, files)

if __name__=="__main__":
    mp.freeze_support() # optional if the program is not frozen
    main()

Если есть много файлов, тогда pool.map() можно заменить на for _ in pool.imap_unordered(safe_run, files): pass.

Существует также mutiprocessing.dummy.Pool, который предоставляет тот же интерфейс, что и multiprocessing.Pool, но использует потоки вместо процессов, которые могут быть более подходящими в этом случае.

Вам не нужно освобождать некоторые процессоры. Просто используйте команду, которая запускает ваши исполняемые файлы с низким приоритетом (в Linux это программа nice).

ThreadPoolExecutor пример

concurrent.futures.ThreadPoolExecutor будет простым и достаточным, но для этого требуется зависимость от сторонних разработчиков от Python 2.x (это в stdlib с Python 3.2).

#!/usr/bin/env python
import os
import concurrent.futures

def run(filename, def_param):
    ... # call external program on `filename`

# populate files
ws = r'D:\Data\Users\jbellino\Project\stJohnsDeepening\model\xsec_a'
wdir = os.path.join(ws, r'fieldgen\reals')
files = (os.path.join(wdir, f) for f in os.listdir(wdir) if f.endswith('.npy'))

# start threads
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
    future_to_file = dict((executor.submit(run, f, ws), f) for f in files)

    for future in concurrent.futures.as_completed(future_to_file):
        f = future_to_file[future]
        if future.exception() is not None:
           print('%r generated an exception: %s' % (f, future.exception()))
        # run() doesn't return anything so `future.result()` is always `None`

Или если игнорировать исключения, вызванные run():

from itertools import repeat

... # the same

# start threads
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
     executor.map(run, files, repeat(ws))
     # run() doesn't return anything so `map()` results can be ignored

subprocess + threading (ручной пул) решение

#!/usr/bin/env python
from __future__ import print_function
import os
import subprocess
import sys
from Queue import Queue
from threading import Thread

def run(filename, def_param):
    ... # define exe, swt_nam
    subprocess.check_call([exe, swt_nam]) # run external program

def worker(queue):
    """Process files from the queue."""
    for args in iter(queue.get, None):
        try:
            run(*args)
        except Exception as e: # catch exceptions to avoid exiting the
                               # thread prematurely
            print('%r failed: %s' % (args, e,), file=sys.stderr)

# start threads
q = Queue()
threads = [Thread(target=worker, args=(q,)) for _ in range(8)]
for t in threads:
    t.daemon = True # threads die if the program dies
    t.start()

# populate files
ws = r'D:\Data\Users\jbellino\Project\stJohnsDeepening\model\xsec_a'
wdir = os.path.join(ws, r'fieldgen\reals')
for f in os.listdir(wdir):
    if f.endswith('.npy'):
        q.put_nowait((os.path.join(wdir, f), ws))

for _ in threads: q.put_nowait(None) # signal no more files
for t in threads: t.join() # wait for completion

Ответ 2

Вот мой способ поддерживать минимальное количество потоков в памяти. Это комбинация многопоточных и многопроцессорных модулей. Это может быть необычно для других методов, таких как уважаемые коллеги, объяснил выше, но может быть значительным. Для объяснения, я беру сценарий сканирования по меньшей мере 5 сайтов за раз.

так что вот: -

#importing dependencies.
from multiprocessing import Process
from threading import Thread
import threading

# Crawler function
def crawler(domain):
    # define crawler technique here.
    output.write(scrapeddata + "\n")
    pass

Далее функция threadController. Эта функция будет управлять потоком потоков в основной памяти. Он будет продолжать активировать потоки, чтобы поддерживать минимальный предел threadNum, т.е. 5. Также он не выйдет до тех пор, пока все активные потоки (acitveCount) не будут завершены.

Он будет поддерживать минимум потоков threadNum (5) startProcess (эти потоки в конечном итоге запустит процессы из списка процессов, присоединив их с временем в 60 секунд). После просмотра threadController было бы 2 потока, которые не включены в вышеуказанный предел 5, т.е. Основной поток и поток threadController. вот почему используется threading.activeCount()!= 2.

def threadController():
    print "Thread count before child thread starts is:-", threading.activeCount(), len(processList)
    # staring first thread. This will make the activeCount=3
    Thread(target = startProcess).start()
    # loop while thread List is not empty OR active threads have not finished up.
    while len(processList) != 0 or threading.activeCount() != 2:
        if (threading.activeCount() < (threadNum + 2) and # if count of active threads are less than the Minimum AND
            len(processList) != 0):                            # processList is not empty
                Thread(target = startProcess).start()         # This line would start startThreads function as a seperate thread **

Функция startProcess, как отдельный поток, запускает процессы из списка процессов. Назначение этой функции (** началось как другой поток) заключается в том, что она станет родительским потоком для процессов. Поэтому, когда он присоединится к ним с тайм-аутом в 60 секунд, это остановит поток startProcess, чтобы двигаться вперед, но это не остановит выполнение threadController. Таким образом, threadController будет работать по мере необходимости.

def startProcess():
    pr = processList.pop(0)
    pr.start()
    pr.join(60.00) # joining the thread with time out of 60 seconds as a float.

if __name__ == '__main__':
    # a file holding a list of domains
    domains = open("Domains.txt", "r").read().split("\n")
    output = open("test.txt", "a")
    processList = [] # thread list
    threadNum = 5 # number of thread initiated processes to be run at one time

    # making process List
    for r in range(0, len(domains), 1):
        domain = domains[r].strip()
        p = Process(target = crawler, args = (domain,))
        processList.append(p) # making a list of performer threads.

    # starting the threadController as a seperate thread.
    mt = Thread(target = threadController)
    mt.start()
    mt.join() # won't let go next until threadController thread finishes.

    output.close()
    print "Done"

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

Я надеюсь, что эта конструкция поможет любому в этом мире. С Уважением, Викас Гаутам