पाइथन मल्टीप्रोसेसिंग सुरक्षित रूप से एक फ़ाइल

Question

पर लिखने के लिए मैं एक बड़ी संख्यात्मक समस्या को हल करने की कोशिश कर रहा हूं जिसमें बहुत से उपप्रवाह शामिल हैं, और मैं विभिन्न कोर पर विभिन्न स्वतंत्र उपप्रोबम्स को विभाजित करने के लिए पाइथन के मल्टीप्रोसेसिंग मॉड्यूल (विशेष रूप से Pool.map) का उपयोग कर रहा हूं। प्रत्येक सबप्रोबलेम में बहुत से सब-सबप्रोबलेम्स की गणना करना शामिल है, और मैं इन परिणामों को प्रभावी रूप से किसी फ़ाइल में संग्रहीत करके याद कर रहा हूं, अगर उन्हें किसी भी प्रक्रिया द्वारा गणना नहीं की गई है, अन्यथा गणना को छोड़ दें और केवल फाइल से परिणाम पढ़ें।

मुझे फाइलों के साथ समवर्ती समस्याएं हैं: विभिन्न प्रक्रियाएं कभी-कभी यह देखने के लिए जांच करती हैं कि उप-सबप्रोबैम की गणना क्यों की गई है (फ़ाइल को देखकर जहां परिणाम संग्रहीत किए जाएंगे), देखें कि यह नहीं है, गणना चलाएं, फिर परिणामों को एक ही फ़ाइल में एक ही समय में लिखने का प्रयास करें। मैं इस तरह टकराव लिखने से कैसे बचूं?

Answer 1

@ GP89 ने एक अच्छा समाधान का उल्लेख किया। लेखन कार्यों को एक समर्पित प्रक्रिया में भेजने के लिए एक कतार का उपयोग करें जिसमें फ़ाइल में एकमात्र लेखन पहुंच है। अन्य सभी श्रमिकों ने केवल पहुंच पढ़ी है। यह टकराव को खत्म कर देगा। यहाँ एक उदाहरण apply_async का उपयोग करता है, लेकिन यह नक्शे के साथ भी काम करेगा:

import multiprocessing as mp 
import time 

fn = 'c:/temp/temp.txt' 

def worker(arg, q): 
    '''stupidly simulates long running process''' 
    start = time.clock() 
    s = 'this is a test' 
    txt = s 
    for i in xrange(200000): 
     txt += s 
    done = time.clock() - start 
    with open(fn, 'rb') as f: 
     size = len(f.read()) 
    res = 'Process' + str(arg), str(size), done 
    q.put(res) 
    return res 

def listener(q): 
    '''listens for messages on the q, writes to file. ''' 

    f = open(fn, 'wb') 
    while 1: 
     m = q.get() 
     if m == 'kill': 
      f.write('killed') 
      break 
     f.write(str(m) + '\n') 
     f.flush() 
    f.close() 

def main(): 
    #must use Manager queue here, or will not work 
    manager = mp.Manager() 
    q = manager.Queue()  
    pool = mp.Pool(mp.cpu_count() + 2) 

    #put listener to work first 
    watcher = pool.apply_async(listener, (q,)) 

    #fire off workers 
    jobs = [] 
    for i in range(80): 
     job = pool.apply_async(worker, (i, q)) 
     jobs.append(job) 

    # collect results from the workers through the pool result queue 
    for job in jobs: 
     job.get() 

    #now we are done, kill the listener 
    q.put('kill') 
    pool.close() 

if __name__ == "__main__": 
    main() 

अच्छी किस्मत,

माइक

Answer 2

यह मेरे लिए लग रहा है कि आप अस्थायी रूप से अपने परिणाम को बचाने के लिए प्रबंधक का उपयोग करने की आवश्यकता है एक सूची में और फिर सूची से परिणामों को एक फ़ाइल में लिखें। साथ ही, उस ऑब्जेक्ट को पास करने के लिए स्टैम्पैप का उपयोग करें जिसे आप संसाधित करना चाहते हैं और प्रबंधित सूची। पहला कदम स्टैरैप को पारित करने के लिए पैरामीटर बनाना है, जिसमें प्रबंधित सूची शामिल है।

from multiprocessing import Manager 
from multiprocessing import Pool 
import pandas as pd``` 

def worker(row, param): 
    # do something here and then append it to row 
    x = param**2 
    row.append(x) 

if __name__ == '__main__': 
    pool_parameter = [] # list of objects to process 
    with Manager() as mgr: 
     row = mgr.list([]) 

     # build list of parameters to send to starmap 
     for param in pool_parameter: 
      params.append([row,param]) 

     with Pool() as p: 
      p.starmap(worker, params) 

इस बिंदु से आपको यह तय करने की आवश्यकता है कि आप सूची को कैसे संभालने जा रहे हैं। यदि आपके पास बहुत सी रैम है और एक विशाल डेटा सेट पांडा का उपयोग करके संयोजित करने के लिए स्वतंत्र महसूस करता है। फिर आप फ़ाइल को सीएसवी या अचार के रूप में आसानी से सहेज सकते हैं। [ `Multiprocessing.Lock`] (http://docs.python.org/2/library/multiprocessing.html#synchronization-between-processes) एकाधिक सिंक्रनाइज़ करने के लिए

 df = pd.concat(row, ignore_index=True) 

     df.to_pickle('data.pickle') 
     df.to_csv('data.csv')