使用Concurrent.Futures.ProcessPoolExecutor运行同时与独立ABAQUS模型

我希望运行总共nAnalysis = 25个的Abaqus模型，核中的每使用X数目，我可以同时运行nParallelLoops = 5这些模型的。如果目前的5所分析的一个结束，然后又分析应该开始，直到所有nAnalysis完成。使用Concurrent.Futures.ProcessPoolExecutor运行同时与独立ABAQUS模型

我使用和中发布的解决方案实现了以下代码。然而，我错过了一些东西，因为所有n分析尝试从“一次”开始，代码发生死锁并且没有分析完成，因为很多时候可能想要使用与已经开始的分析相同的内核。

def runABQfile(*args):  
    import subprocess 
    import os 

    inpFile,path,jobVars = args 

    prcStr1 = (path+'/runJob.sh') 

    process = subprocess.check_call(prcStr1, stdin=None, stdout=None, stderr=None, shell=True, cwd=path) 

def safeABQrun(*args): 
    import os 

    try: 
     runABQfile(*args) 
    except Exception as e: 
     print("Tread Error: %s runABQfile(*%r)" % (e, args)) 

def errFunction(ppos, *args): 
    import os 
    from concurrent.futures import ProcessPoolExecutor 
    from concurrent.futures import as_completed 
    from concurrent.futures import wait 

    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     future_to_file = dict((executor.submit(safeABQrun, inpFiles[k], aPath[k], jobVars), k) for k in range(0,nAnalysis)) # 5Nodes 
     wait(future_to_file,timeout=None,return_when='ALL_COMPLETED')

最多只有这样，现在我能够运行是，如果我修改errFunction在下面的时候用了整整5分析。然而，这种方法有时会导致在分析的每一个组中采取比其他4长得多（每ProcessPoolExecutor调用），因此5下一组不会不顾资源的可用性（核）开始。最终这会导致更多时间完成所有25个型号。

def errFunction(ppos, *args): 
    import os 
    from concurrent.futures import ProcessPoolExecutor 
    from concurrent.futures import as_completed 
    from concurrent.futures import wait  

    # Group 1 
    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     future_to_file = dict((executor.submit(safeABQrun, inpFiles[k], aPath[k], jobVars), k) for k in range(0,5)) # 5Nodes   
     wait(future_to_file,timeout=None,return_when='ALL_COMPLETED') 

    # Group 2 
    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     future_to_file = dict((executor.submit(safeABQrun, inpFiles[k], aPath[k], jobVars), k) for k in range(5,10)) # 5Nodes   
     wait(future_to_file,timeout=None,return_when='ALL_COMPLETED') 

    # Group 3 
    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     future_to_file = dict((executor.submit(safeABQrun, inpFiles[k], aPath[k], jobVars), k) for k in range(10,15)) # 5Nodes   
     wait(future_to_file,timeout=None,return_when='ALL_COMPLETED') 

    # Group 4 
    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     future_to_file = dict((executor.submit(safeABQrun, inpFiles[k], aPath[k], jobVars), k) for k in range(15,20)) # 5Nodes   
     wait(future_to_file,timeout=None,return_when='ALL_COMPLETED') 

    # Group 5 
    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     future_to_file = dict((executor.submit(safeABQrun, inpFiles[k], aPath[k], jobVars), k) for k in range(20,25)) # 5Nodes   
     wait(future_to_file,timeout=None,return_when='ALL_COMPLETED')

我试过使用as_completed函数，但它似乎不工作。

请你能帮助找出正确的并行化，所以我可以运行nAnalysis，总是具有nParallelLoops同时运行？您的帮助表示感谢。我使用Python 2.7

最好成绩，大卫P.

UPDATE JULY二千零十六分之三十〇：

我在safeABQrun推出了循环和管理的5个不同的“队列”。循环是避免分析尝试在节点中运行而另一个仍在运行的情况下所必需的。分析预先配置为在开始任何实际分析之前在其中一个请求节点中运行。

def safeABQrun(*list_args): 
    import os 

    inpFiles,paths,jobVars = list_args 

    nA = len(inpFiles) 
    for k in range(0,nA): 
     args = (inpFiles[k],paths[k],jobVars[k]) 
     try: 
      runABQfile(*args) # Actual Run Function 
     except Exception as e: 
      print("Tread Error: %s runABQfile(*%r)" % (e, args)) 

def errFunction(ppos, *args): 
    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     futures = dict((executor.submit(safeABQrun, inpF, aPth, jVrs), k) for inpF, aPth, jVrs, k in list_args) # 5Nodes 

     for f in as_completed(futures): 
      print("|=== Finish Process Train %d ===|" % futures[f]) 
      if f.exception() is not None: 
       print('%r generated an exception: %s' % (futures[f], f.exception()))

来源

2016-07-29 David P.

它看起来OK我，但是，我是不能运行代码。如何极大地尝试一些简单的，然后添加的事情，直到“问题”出现？例如，下面显示了你想要的行为类型吗？它在我的机器上，但我正在运行Python 3.5.2。你说你正在运行2.7，但concurrent.futures没有在Python 2存在 - 如果你使用的是2。7，你必须运行某人的图书馆后台，也许问题在于此。尝试以下应有助于回答是否是这样的话：

from concurrent.futures import ProcessPoolExecutor, wait, as_completed 

def worker(i): 
    from time import sleep 
    from random import randrange 
    s = randrange(1, 10) 
    print("%d started and sleeping for %d" % (i, s)) 
    sleep(s) 

if __name__ == "__main__": 
    nAnalysis = 25 
    nParallelLoops = 5 
    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     futures = dict((executor.submit(worker, k), k) for k in range(nAnalysis)) 
     for f in as_completed(futures): 
      print("got %d" % futures[f])

典型输出：

0 started and sleeping for 4 
1 started and sleeping for 1 
2 started and sleeping for 1 
3 started and sleeping for 6 
4 started and sleeping for 5 
5 started and sleeping for 9 
got 1 
6 started and sleeping for 5 
got 2 
7 started and sleeping for 6 
got 0 
8 started and sleeping for 6 
got 4 
9 started and sleeping for 8 
got 6 
10 started and sleeping for 9 
got 3 
11 started and sleeping for 6 
got 7 
12 started and sleeping for 9 
got 5 
...

来源

2016-07-29 21:59:02

def safeABQrun(*list_args): 
    import os 

    inpFiles,paths,jobVars = list_args 

    nA = len(inpFiles) 
    for k in range(0,nA): 
     args = (inpFiles[k],paths[k],jobVars[k]) 
     try: 
      runABQfile(*args) # Actual Run Function 
     except Exception as e: 
      print("Tread Error: %s runABQfile(*%r)" % (e, args)) 

def errFunction(ppos, *args): 
    with ProcessPoolExecutor(max_workers=nParallelLoops) as executor: 
     futures = dict((executor.submit(safeABQrun, inpF, aPth, jVrs), k) for inpF, aPth, jVrs, k in list_args) # 5Nodes 

     for f in as_completed(futures): 
      print("|=== Finish Process Train %d ===|" % futures[f]) 
      if f.exception() is not None: 
       print('%r generated an exception: %s' % (futures[f], f.exception()))

来源

2016-07-30 19:19:16

使用Concurrent.Futures.ProcessPoolExecutor运行同时与独立ABAQUS模型

回答

相关问题