Python multiprocessing safely writing to a file

by

@GP89 mentioned a good solution. Use a queue to send the writing tasks to a dedicated process that has sole write access to the file. All the other workers have read only access. This will eliminate collisions. Here is an example that uses apply_async, but it will work with map too:

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 range(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. '''

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

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()
    pool.join()

if __name__ == "__main__":
   main()

More Related Contents:

  1. Why can’t I call read() twice on an open file?
  2. Editing specific line in text file in Python
  3. Why does multiprocessing use only a single core after I import numpy?
  4. Sharing a result queue among several processes
  5. Keyboard Interrupts with python’s multiprocessing Pool
  6. What exactly is file.flush() doing?
  7. multiprocessing: Understanding logic behind `chunksize`
  8. Multiprocessing: How to use Pool.map on a function defined in a class?
  9. multiprocessing in python – sharing large object (e.g. pandas dataframe) between multiple processes
  10. pandas multiprocessing apply
  11. Multiprocessing.Pool makes Numpy matrix multiplication slower
  12. Can I use a multiprocessing Queue in a function called by Pool.imap?
  13. Using “readlines()” twice in a row [duplicate]
  14. Accessing an attribute of a multiprocessing Proxy of a class
  15. What’s the difference between io.open() and os.open() on Python?
  16. call multiprocessing in class method Python
  17. Leveraging “Copy-on-Write” to Copy Data to Multiprocessing.Pool() Worker Processes
  18. Reading a file with a specified delimiter for newline
  19. Tkinter GUI, I/O & Threading: When to use queues, when events?
  20. How to get the return value of a function passed to multiprocessing.Process?
  21. How to share numpy random state of a parent process with child processes?
  22. multiprocessing vs multithreading vs asyncio
  23. Python multiprocessing – tracking the process of pool.map operation
  24. Creating and updating nested dictionaries and lists inside multiprocessing.Manager object
  25. Can I get a return value from multiprocessing.Process?
  26. How to utilize all cores with python multiprocessing
  27. Tensorflow and Multiprocessing: Passing Sessions
  28. How can I share a class between processes?
  29. Multiprocessing useless with urllib2?
  30. Python multiprocessing linux windows difference

Leave a Comment