title: Using Python's Multiprocessing module to execute simultaneous and separate SEAWAT/MODFLOW model runs
body: <p>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.</p>
  
  <p><a href="https://stackoverflow.com/questions/7780478/using-the-multiprocessing-module">Using the multiprocessing module</a></p>
  
  <p><a href="https://stackoverflow.com/questions/884650/python-spawn-parallel-child-processes-on-a-multi-processor-system-use-multipr">How to spawn parallel child processes on a multi-processor system?</a></p>
  
  <p><a href="https://stackoverflow.com/questions/5506227/python-multiprocessing-queue">Python Multiprocessing queue</a></p>
  
  <p><strong>My Process:</strong></p>
  
  <p>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:</p>
  
  <pre><code>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)
  </code></pre>
  
  <p>I changed the <code>if __name__ == 'main':</code> to the following in hopes it would fix the lack of parallelism I feel is being imparted on the above script by the <code>for loop</code>. However, the model fails to even run (no Python error):</p>
  
  <pre><code>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))))
  </code></pre>
  
  <p>Any and all help is greatly appreciated!</p>
  
  <p><strong>EDIT 3/26/2012 13:31 EST</strong></p>
  
  <p>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:</p>
  
  <pre><code>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()
  </code></pre>
  
  <p>No  error traceback is available. The <code>run()</code> 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 <code>len(files)</code> 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).</p>
  
  <p>Adding some print statements to <code>main()</code> 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):</p>
  
  <pre><code>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())
  </code></pre>
  
  <p><img src="https://i.stack.imgur.com/6yffC.png" alt="screenshot"></p>
  
  <p>**The line which reads "<code>D:\\Data\\Users...</code>" 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.</p>
  
  <p><strong>EDIT 3/26/2012 16:24 EST</strong></p>
  
  <p>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.</p>
  
  <p>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.</p>
  
  <p><strong>FINAL CODE</strong></p>
  
  <p>Added cwd argument in subprocess.check_call to start each instance of SEAWAT in its own directory. Very key.</p>
  
  <pre><code>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()
  </code></pre>