मल्टीप्रोसेसिंग

Question

का उपयोग करके एचडीएफ फ़ाइल में डेटा लिखें यह एक साधारण समस्या की तरह लगता है लेकिन मैं इसके चारों ओर अपना सिर नहीं प्राप्त कर सकता।

मेरे पास एक सिमुलेशन है जो लूप के लिए डबल में चलता है और परिणाम एचडीएफ फ़ाइल में लिखता है। इस कार्यक्रम का एक सरल संस्करण नीचे दिखाया गया है:

import tables as pt 

a = range(10) 
b = range(5) 

def Simulation(): 
    hdf = pt.openFile('simulation.h5',mode='w') 
    for ii in a: 
     print(ii) 
     hdf.createGroup('/','A%s'%ii) 
     for i in b: 
      hdf.createArray('/A%s'%ii,'B%s'%i,[ii,i]) 
     hdf.close() 
    return 
Simulation() 

इस कोड है कि वास्तव में क्या मैं चाहता हूँ लेकिन जब से प्रक्रिया को चलाने के लिए काफी समय लग सकता है मैं बहु मॉड्यूल का उपयोग करता है और निम्नलिखित कोड का उपयोग करने की कोशिश की:

import multiprocessing 
import tables as pt 

a = range(10) 
b = range(5) 

def Simulation(ii): 
    hdf = pt.openFile('simulation.h5',mode='w') 
    print(ii) 
     hdf.createGroup('/','A%s'%ii) 
     for i in b: 
      hdf.createArray('/A%s'%ii,'B%s'%i,[ii,i]) 
     hdf.close() 
    return 

if __name__ == '__main__': 
    jobs = [] 
    for ii in a: 
     p = multiprocessing.Process(target=Simulation, args=(ii,)) 
     jobs.append(p)  
     p.start() 

हालांकि यह केवल एचडीएफ फ़ाइल के अंतिम अनुकरण को प्रिंट करता है, किसी भी तरह यह अन्य सभी समूहों को ओवरराइट करता है।

Answer 1

प्रत्येक बार जब आप लिखने में फ़ाइल खोलते हैं (w) मोड, एक नई फ़ाइल बनाई जाती है - इसलिए फ़ाइल की सामग्री खो जाती है यदि यह पहले से मौजूद है। केवल अंतिम फ़ाइल हैंडल फ़ाइल को सफलतापूर्वक लिख सकता है। यहां तक ​​कि यदि आपने मोड को जोड़ने के लिए इसे बदल दिया है, तो आपको एक ही फाइल को एकाधिक प्रक्रियाओं से लिखने की कोशिश नहीं करनी चाहिए - यदि दो प्रक्रियाएं एक ही समय में लिखने की कोशिश करती हैं तो आउटपुट खराब हो जाएगा।

इसके बजाय, सभी कार्यकर्ता एक कतार में उत्पादन डाल प्रक्रियाओं है, और (या तो एक उपप्रक्रिया या मुख्य प्रक्रिया) एक एकल समर्पित प्रक्रिया है कतार से उत्पादन को संभालने और फ़ाइल पर लिखने:

---

import multiprocessing as mp 
import tables as pt 


num_arrays = 100 
num_processes = mp.cpu_count() 
num_simulations = 1000 


def Simulation(ii): 
    result = [] 
    result.append(('createGroup', ('/', 'A%s' % ii))) 
    for i in range(num_arrays): 
     result.append(('createArray', ('/A%s' % ii, 'B%s' % i, [ii, i]))) 
    return result 


def handle_output(result): 
    hdf = pt.openFile('simulation.h5', mode='a') 
    for args in result: 
     method, args = args 
     getattr(hdf, method)(*args) 
    hdf.close() 


if __name__ == '__main__': 
    # clear the file 
    hdf = pt.openFile('simulation.h5', mode='w') 
    hdf.close() 
    pool = mp.Pool(num_processes) 
    for i in range(num_simulations): 
     pool.apply_async(Simulation, (i,), callback=handle_output) 
    pool.close() 
    pool.join() 

:

import multiprocessing as mp 
import tables as pt 


num_arrays = 100 
num_processes = mp.cpu_count() 
num_simulations = 1000 
sentinel = None 


def Simulation(inqueue, output): 
    for ii in iter(inqueue.get, sentinel): 
     output.put(('createGroup', ('/', 'A%s' % ii))) 
     for i in range(num_arrays): 
      output.put(('createArray', ('/A%s' % ii, 'B%s' % i, [ii, i]))) 


def handle_output(output): 
    hdf = pt.openFile('simulation.h5', mode='w') 
    while True: 
     args = output.get() 
     if args: 
      method, args = args 
      getattr(hdf, method)(*args) 
     else: 
      break 
    hdf.close() 

if __name__ == '__main__': 
    output = mp.Queue() 
    inqueue = mp.Queue() 
    jobs = [] 
    proc = mp.Process(target=handle_output, args=(output,)) 
    proc.start() 
    for i in range(num_processes): 
     p = mp.Process(target=Simulation, args=(inqueue, output)) 
     jobs.append(p) 
     p.start() 
    for i in range(num_simulations): 
     inqueue.put(i) 
    for i in range(num_processes): 
     # Send the sentinal to tell Simulation to end 
     inqueue.put(sentinel) 
    for p in jobs: 
     p.join() 
    output.put(None) 
    proc.join() 

---

तुलना के लिए, यहाँ एक संस्करण है जो mp.Pool का उपयोग करता है

यह आसान लगता है ना? हालांकि एक संकेतक अंतर है। handle_output पर तर्क भेजने के लिए मूल कोड output.put का उपयोग किया गया था जो अपने स्वयं के उपप्रजाति में चल रहा था। handle_outputargsoutput कतार से ले जाएगा और उन्हें तुरंत संभालेंगे। ऊपर पूल कोड के साथ, Simulationargsresult और result में handle_output पर Simulation रिटर्न के बाद तक args का पूरा समूह जमा करता है।

यदि Simulation में काफी समय लगता है, तो एक लंबी प्रतीक्षा अवधि होगी जबकि simulation.h5 पर कुछ भी नहीं लिखा जा रहा है।