Ghi dữ liệu vào tập tin hdf bằng cách sử dụng đa xử lý

Question

Điều này có vẻ như là một vấn đề đơn giản nhưng tôi không thể có được đầu của tôi xung quanh nó.

Tôi có mô phỏng chạy trong vòng lặp đôi và ghi kết quả vào tệp HDF. Một phiên bản đơn giản của chương trình này được hiển thị dưới đây:

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() 

Mã này thực hiện chính xác những gì tôi muốn nhưng vì quá trình này có thể mất nhiều thời gian để chạy tôi cố gắng sử dụng các mô-đun đa và sử dụng đoạn mã sau:

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() 

Tuy nhiên, điều này chỉ in mô phỏng cuối cùng cho tệp HDF, bằng cách nào đó nó ghi đè tất cả các nhóm khác.

Answer 1

Mỗi khi bạn mở một tệp trong chế độ ghi (w), một tệp mới sẽ được tạo - vì vậy nội dung của tệp sẽ bị mất nếu nó đã tồn tại. Chỉ xử lý tệp cuối cùng mới có thể ghi thành công vào tệp. Ngay cả khi bạn thay đổi chế độ đó để thêm, bạn không nên cố gắng ghi vào cùng một tệp từ nhiều quy trình - đầu ra sẽ bị cắt xén nếu hai quá trình cố gắng viết cùng một lúc.

Thay vào đó, có tất cả các quá trình lao động đưa ra trong một hàng đợi, và có một quá trình chuyên dụng đơn (hoặc là một tiến trình con hoặc quá trình chính) xử lý đầu ra khỏi hàng đợi và ghi vào tập tin:

---

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() 

---

Để so sánh, đây là một phiên bản trong đó sử dụng mp.Pool:

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() 

Có vẻ đơn giản hơn phải không? Tuy nhiên có một sự khác biệt đáng kể. Mã ban đầu được sử dụng output.put để gửi args đến handle_output đang chạy trong tiến trình con của riêng nó. handle_output sẽ mất args từ hàng đợi output và xử lý chúng ngay lập tức. Với mã Hồ bơi ở trên, Simulation tích lũy toàn bộ một bó số args trong result và result không được gửi đến handle_output cho đến khi sau Simulation trả lại.

Nếu Simulation mất nhiều thời gian, sẽ có thời gian chờ đợi lâu trong khi không có gì được ghi vào simulation.h5.