問題描述

我正在嘗試在我的 8 處理器 64 位 Windows 7 機(jī)器上運(yùn)行 100 個(gè)模型.我想同時(shí)運(yùn)行 7 個(gè)模型實(shí)例以減少我的總運(yùn)行時(shí)間(每個(gè)模型運(yùn)行大約 9.5 分鐘).我已經(jīng)查看了與 Python 的多處理模塊有關(guān)的幾個(gè)線程，但仍然缺少一些東西.

concurrent.futures.ThreadPoolExecutor 既簡(jiǎn)單又足夠了，但它需要 第三方對(duì) Python 2.x 的依賴(自 Python 3.2 起就在 stdlib 中).

#!/usr/bin/env python導(dǎo)入操作系統(tǒng)導(dǎo)入并發(fā)期貨def 運(yùn)行(文件名，def_param):... # 在 `filename` 上調(diào)用外部程序# 填充文件ws = r'D:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'wdir = os.path.join(ws, r'fieldgen
eals')files = (os.path.join(wdir, f) for f in os.listdir(wdir) if f.endswith('.npy'))# 啟動(dòng)線程以 concurrent.futures.ThreadPoolExecutor(max_workers=8) 作為執(zhí)行者:future_to_file = dict((executor.submit(run, f, ws), f) for f in files)concurrent.futures.as_completed(future_to_file) 中的未來:f = future_to_file[未來]如果 future.exception() 不是無:print('%r 產(chǎn)生異常: %s' % (f, future.exception()))# run() 不返回任何東西，所以 `future.result()` 總是 `None`

或者如果我們忽略 run() 引發(fā)的異常:

從 itertools 導(dǎo)入重復(fù)... # 相同# 啟動(dòng)線程以 concurrent.futures.ThreadPoolExecutor(max_workers=8) 作為執(zhí)行者:executor.map(運(yùn)行，文件，重復(fù)(ws))# run() 不返回任何內(nèi)容，因此可以忽略 `map()` 結(jié)果

subprocess + threading(手動(dòng)池)解決方案

#!/usr/bin/env python從 __future__ 導(dǎo)入 print_function導(dǎo)入操作系統(tǒng)導(dǎo)入子流程導(dǎo)入系統(tǒng)從隊(duì)列導(dǎo)入隊(duì)列從線程導(dǎo)入線程def 運(yùn)行(文件名，def_param):... # 定義 exe, swt_namsubprocess.check_call([exe, swt_nam]) # 運(yùn)行外部程序def 工人(隊(duì)列):"""處理隊(duì)列中的文件."""對(duì)于 iter(queue.get, None) 中的 args:嘗試:運(yùn)行(*參數(shù))except Exception as e: # 捕獲異常以避免退出# 過早線程print('%r failed: %s' % (args, e,), file=sys.stderr)# 啟動(dòng)線程q = 隊(duì)列()線程= [線程(目標(biāo)=工作者，參數(shù)=(q，))為_在范圍內(nèi)(8)]對(duì)于線程中的 t:t.daemon = True # 如果程序死了，線程就死了t.start()# 填充文件ws = r'D:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'wdir = os.path.join(ws, r'fieldgen
eals')對(duì)于 os.listdir(wdir) 中的 f:如果 f.endswith('.npy'):q.put_nowait((os.path.join(wdir, f), ws))for _ in threads: q.put_nowait(None) # 表示不再有文件for t in threads: t.join() # 等待完成

I'm trying to complete 100 model runs on my 8-processor 64-bit Windows 7 machine. I'd like to run 7 instances of the model concurrently to decrease my total run time (approx. 9.5 min per model run). I've looked at several threads pertaining to the Multiprocessing module of Python, but am still missing something.

Using the multiprocessing module

How to spawn parallel child processes on a multi-processor system?

Python Multiprocessing queue

My Process:

I have 100 different parameter sets I'd like to run through SEAWAT/MODFLOW to compare the results. I have pre-built the model input files for each model run and stored them in their own directories. What I'd like to be able to do is have 7 models running at a time until all realizations have been completed. There needn't be communication between processes or display of results. So far I have only been able to spawn the models sequentially:

import os,subprocess
import multiprocessing as mp

ws = r'D:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'
files = []
for f in os.listdir(ws + r'fieldgen
eals'):
    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:modflowmf2k.1_19inmf2k.exe '
    mf2k5 = r'c:modflowMF2005_1_8inmf2005.exe '
    seawatV4 = r'c:modflowswt_v4_00_04exeswt_v4.exe '
    seawatV4x64 = r'c:modflowswt_v4_00_04exeswt_v4x64.exe '

    exe = seawatV4x64
    swt_nam = ws + r'
eals
eal%sssss.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)

I changed the if __name__ == 'main': to the following in hopes it would fix the lack of parallelism I feel is being imparted on the above script by the for loop. However, the model fails to even run (no Python error):

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))))

Any and all help is greatly appreciated!

EDIT 3/26/2012 13:31 EST

Using the "Manual Pool" method in @J.F. Sebastian's answer below I get parallel execution of my external .exe. Model realizations are called up in batches of 8 at a time, but it doesn't wait for those 8 runs to complete before calling up the next batch and so on:

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:modflowswt_v4_00_04exeswt_v4x64.exe '
    swt_nam = ws + r'
eals
eal%sssss.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:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'
    wdir = os.path.join(ws, r'fieldgen
eals')
    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()

No error traceback is available. The run() function performs its duty when called upon a single model realization file as with mutiple files. The only difference is that with multiple files, it is called len(files) times though each of the instances immediately closes and only one model run is allowed to finish at which time the script exits gracefully (exit code 0).

Adding some print statements to main() reveals some information about active thread-counts as well as thread status (note that this is a test on only 8 of the realization files to make the screenshot more manageable, theoretically all 8 files should be run concurrently, however the behavior continues where they are spawn and immediately die except one):

def main():
    # populate files
    ws = r'D:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'
    wdir = os.path.join(ws, r'fieldgen	est')
    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())

**The line which reads "D:\Data\Users..." is the error information thrown when I manually stop the model from running to completion. Once I stop the model running, the remaining thread status lines get reported and the script exits.

EDIT 3/26/2012 16:24 EST

SEAWAT does allow concurrent execution as I've done this in the past, spawning instances manually using iPython and launching from each model file folder. This time around, I'm launching all model runs from a single location, namely the directory where my script resides. It looks like the culprit may be in the way SEAWAT is saving some of the output. When SEAWAT is run, it immediately creates files pertaining to the model run. One of these files is not being saved to the directory in which the model realization is located, but in the top directory where the script is located. This is preventing any subsequent threads from saving the same file name in the same location (which they all want to do since these filenames are generic and non-specific to each realization). The SEAWAT windows were not staying open long enough for me to read or even see that there was an error message, I only realized this when I went back and tried to run the code using iPython which directly displays the printout from SEAWAT instead of opening a new window to run the program.

I am accepting @J.F. Sebastian's answer as it is likely that once I resolve this model-executable issue, the threading code he has provided will get me where I need to be.

FINAL CODE

Added cwd argument in subprocess.check_call to start each instance of SEAWAT in its own directory. Very key.

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:modflowswt_v4_00_04exeswt_v4x64.exe '
    cwd = ws + r'
eals
eal%sss' % real
    swt_nam = ws + r'
eals
eal%sssss.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:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'
    wdir = os.path.join(ws, r'fieldgen
eals')
    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()

I don't see any computations in the Python code. If you just need to execute several external programs in parallel it is sufficient to use subprocess to run the programs and threading module to maintain constant number of processes running, but the simplest code is using 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:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'
    workdir = os.path.join(ws, r'fieldgen
eals')
    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()

If there are many files then pool.map() could be replaced by for _ in pool.imap_unordered(safe_run, files): pass.

There is also mutiprocessing.dummy.Pool that provides the same interface as multiprocessing.Pool but uses threads instead of processes that might be more appropriate in this case.

You don't need to keep some CPUs free. Just use a command that starts your executables with a low priority (on Linux it is a nice program).

ThreadPoolExecutor example

concurrent.futures.ThreadPoolExecutor would be both simple and sufficient but it requires 3rd-party dependency on Python 2.x (it is in the stdlib since 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:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'
wdir = os.path.join(ws, r'fieldgen
eals')
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`

Or if we ignore exceptions raised by 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 (manual pool) solution

#!/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:DataUsersjbellinoProjectstJohnsDeepeningmodelxsec_a'
wdir = os.path.join(ws, r'fieldgen
eals')
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

這篇關(guān)于使用 Python 的 Multiprocessing 模塊執(zhí)行同時(shí)和單獨(dú)的 SEAWAT/MODFLOW 模型運(yùn)行的文章就介紹到這了，希望我們推薦的答案對(duì)大家有所幫助，也希望大家多多支持html5模板網(wǎng)！