2

処理するファイルが多数あります。必要なデータを取得、並べ替え、プロットするスクリプトを作成しました。ここまでは順調ですね。私はそれをテストしましたが、望ましい結果が得られます。

次に、マルチスレッドを使用してこれを実行したいと考えました。インターネットでドキュメントと例を調べましたが、プログラムで 1 つのスレッドを使用すると問題なく動作します。しかし、もっと使用すると、ある時点でランダムなmatplotlibエラーが発生し、プロットの名前を持つ関数を使用しているにもかかわらず、競合が疑われ、問題がどこにあるのかわかりません。

さらにコメントが必要な場合は、スクリプト全体を次に示します。追加します。ありがとうございました。

#!/usr/bin/python
import matplotlib
matplotlib.use('GTKAgg')
import numpy as np
from scipy.interpolate import griddata

import matplotlib.pyplot as plt
import matplotlib.colors as mcl
from matplotlib import rc #for latex

import time as tm
import sys
import threading
import Queue #queue in 3.2 and Queue in 2.7 !

import pdb #the debugger

rc('text', usetex=True)#for latex

map=0 #initialize the map index. It will be use to index the array like     this: array[map,[x,y]]
time=np.zeros(1) #an array to store the time
middle_h=np.zeros((0,3)) #x phi c

#for the middle of the box
current_file=open("single_void_cyl_periodic_phi_c_middle_h_out",'r')
for line in current_file:
    if line.startswith('# ===  time'):
        map+=1
        np.append(time,[float(line.strip('# ===  time  '))])
    elif line.startswith('#'):
        pass
    else:
        v=np.fromstring(line,dtype=float,sep=' ')
        middle_h=np.vstack( (middle_h,v[[1,3,4]]) ) 
current_file.close()
middle_h=middle_h.reshape((map,-1,3)) #3d array: map, x, phi,c 

#####
def load_and_plot(): #will load a map file, and plot it along with the     corresponding profile loaded before
    while not exit_flag:
        print("fecthing work ...")
        #try:
        if not tasks_queue.empty():
            map_index=tasks_queue.get()
            print("----> working on map: %s" %map_index)
            x,y,zp=np.loadtxt("single_void_cyl_growth_periodic_post_map_"+str(map_index),unpack=True, usecols=[1, 2,3])
            for i,el in enumerate(zp):
                if el<0.:
                    zp[i]=0.
            xv=np.unique(x)
            yv=np.unique(y)
            X,Y= np.meshgrid(xv,yv)
            Z = griddata((x, y), zp, (X, Y),method='nearest')

            figure=plt.figure(num=map_index,figsize=(14, 8))
            ax1=plt.subplot2grid((2,2),(0,0))
            ax1.plot(middle_h[map_index,:,0],middle_h[map_index,:,1],'*b')
            ax1.grid(True)
            ax1.axis([-15, 15, 0, 1])
            ax1.set_title('Profiles')
            ax1.set_ylabel(r'$\phi$')
            ax1.set_xlabel('x')

            ax2=plt.subplot2grid((2,2),(1,0))
            ax2.plot(middle_h[map_index,:,0],middle_h[map_index,:,2],'*r')
            ax2.grid(True)
            ax2.axis([-15, 15, 0, 1])
            ax2.set_ylabel('c')
            ax2.set_xlabel('x')

            ax3=plt.subplot2grid((2,2),(0,1),rowspan=2,aspect='equal')
            sub_contour=ax3.contourf(X,Y,Z,np.linspace(0,1,11),vmin=0.)
            figure.colorbar(sub_contour,ax=ax3)
            figure.savefig('single_void_cyl_'+str(map_index)+'.png')
            plt.close(map_index)
            tasks_queue.task_done()
        else:
            print("nothing left to do, other threads finishing,sleeping 2 seconds...")
            tm.sleep(2)
   # except:
   #     print("failed this time: %s" %map_index+". Sleeping 2 seconds")
   #     tm.sleep(2)
#####
exit_flag=0
nb_threads=2
tasks_queue=Queue.Queue()
threads_list=[]

jobs=list(range(map)) #each job is composed of a map
print("inserting jobs in the queue...")
for job in jobs:
    tasks_queue.put(job)
print("done")

#launch the threads
for i in range(nb_threads):
    working_bee=threading.Thread(target=load_and_plot)
    working_bee.daemon=True
    print("starting thread "+str(i)+' ...')
    threads_list.append(working_bee)
working_bee.start()


#wait for all tasks to be treated
tasks_queue.join()

#flip the flag, so the threads know it's time to stop
exit_flag=1

for t in threads_list:
    print("waiting for threads %s to stop..."%t)
    t.join()

print("all threads stopped")
4

1 に答える 1

3

デビッドの提案に従って、私はマルチプロセッシングでそれを行いました。8個のプロセッサで5倍の速度が得られます。残りは、スクリプトの冒頭での単一プロセスの作業であると思います。編集：ただし、スクリプトが正しいマップを生成したとしても、最後のマップで「ハング」することがあり、次のエラーが発生します。

tasks_queue.join（）のファイル "single_void_cyl_plot_mprocess.py"、行90

ファイル"/usr/local/epd-7.0-2-rh5-x86_64/lib/python2.7/multiprocessing/queues.py"、316行目、join self._cond.wait（）

ファイル"/usr/local/epd-7.0-2-rh5-x86_64/lib/python2.7/multiprocessing/synchronize.py"、220行目、待機中self._wait_semaphore.acquire（True、timeout）

import numpy as np
from scipy.interpolate import griddata

import matplotlib.pyplot as plt
from matplotlib import rc #for latex

from multiprocessing import Process, JoinableQueue

import pdb #the debugger

rc('text', usetex=True)#for latex

map=0 #initialize the map index. It will be use to index the array     like this: array[map,x,y,...]
time=np.zeros(1) #an array to store the time
middle_h=np.zeros((0,3)) #x phi c

#for the middle of the box
current_file=open("single_void_cyl_periodic_phi_c_middle_h_out",'r')
for line in current_file.readlines():
    if line.startswith('# ===  time'):
        map+=1
        np.append(time,[float(line.strip('# ===  time  '))])
    elif line.startswith('#'):
        pass
    else:
        v=np.fromstring(line,dtype=float,sep=' ')
        middle_h=np.vstack( (middle_h,v[[1,3,4]]) ) 
current_file.close()
middle_h=middle_h.reshape((map,-1,3)) #3d array: map, x, phi,c 

#######
def load_and_plot(): #will load a map file, and plot it along with     the corresponding profile loaded before
    while tasks_queue.empty()==False:
        print("fecthing work ...")
        try:
            map_index=tasks_queue.get() #get some work to do from     the queue
            print("----> working on map: %s" %map_index)
                 x,y,zp=np.loadtxt("single_void_cyl_growth_periodic_post_map_"+str(map_index),\
                unpack=True, usecols=[1, 2,3])
            for i,el in enumerate(zp):
                if el<0.:
                    zp[i]=0.
            xv=np.unique(x)
            yv=np.unique(y)
            X,Y= np.meshgrid(xv,yv)
            Z = griddata((x, y), zp, (X, Y),method='nearest')

            figure=plt.figure(num=map_index,figsize=(14, 8))
            ax1=plt.subplot2grid((2,2),(0,0))
                ax1.plot(middle_h[map_index,:,0],middle_h[map_index,:,1],'*b')
            ax1.grid(True)
            ax1.axis([-15, 15, 0, 1])
            ax1.set_title('Profiles')
            ax1.set_ylabel(r'$\phi$')
            ax1.set_xlabel('x')

            ax2=plt.subplot2grid((2,2),(1,0))
                ax2.plot(middle_h[map_index,:,0],middle_h[map_index,:,2],'*r')
            ax2.grid(True)
            ax2.axis([-15, 15, 0, 1])
            ax2.set_ylabel('c')
            ax2.set_xlabel('x')

            ax3=plt.subplot2grid((2,2),    (0,1),rowspan=2,aspect='equal')
                sub_contour=ax3.contourf(X,Y,Z,np.linspace(0,1,11),vmin=0.)
            figure.colorbar(sub_contour,ax=ax3)
                figure.savefig('single_void_cyl_'+str(map_index)+'.png')
            plt.close(map_index)
            tasks_queue.task_done() #work for this item finished
        except:
            print("failed this time: %s" %map_index)
#######

nb_proc=8 #number of processes
tasks_queue=JoinableQueue() #a queue to pile up the work to do

jobs=list(range(map)) #each job is composed of a map
print("inserting jobs in the queue...")
for job in jobs:
    tasks_queue.put(job)
print("done")

#launch the processes
for i in range(nb_proc):
    current_process=Process(target=load_and_plot)
    current_process.start()

#wait for all tasks to be treated
tasks_queue.join()
于 2012-12-13T16:23:44.220 に答える