Writing to multiple files using multiprocessing. Error: "TypeError: cannot serialize '_io.TextIOWrapper' object"

user_3458386 0
Viewed 6756

I'm trying to write to file the result from multiprocessing (4 cores/processes). Since the CPU cores work simultaneously, I thought of making 4 files, 0.txt, 1.txt, 2.txt and 3.txt and keep it in a multiprocessing.Manager().list(). But I'm getting error, TypeError: cannot serialize '_io.TextIOWrapper' object.

def run_solver(total, proc_id, result, fouts):
    for i in range(10)):
        fouts[proc_id].write('hi\n')

if __name__ == '__main__':
    processes = []
    fouts = Manager().list((open('0.txt', 'w'), open('1.txt', 'w'), open('2.txt', 'w'), open('3.txt', 'w')))
    for proc_id in range(os.cpu_count()):
        processes.append(Process(target=run_solver, args=(int(total/os.cpu_count()), proc_id, result, fouts)))

    for process in processes:
        process.start()

    for process in processes:
        process.join()

    for i in range(len(fouts)):
        fouts[i].close()

I've tried to populate the list with file handle inside the function too, like below.

def run_solver(total, proc_id, result, fouts):
    fout[proc_id] = open(str(proc_id)+'.txt', 'w')
    for i in range(10)):
        fouts[proc_id].write('hi\n')
    fout[proc_id].close()

if __name__ == '__main__':
    processes = []
    fouts = Manager().list([0]*os.cpu_count())

Both doesn't work and I understood there's something related to not able to serialize or unpickleable. But I don't know how to resolve this. Can someone suggest a solution?

python-3.x multiprocessing python
1 Answers

Open the files in each process. Do not open them in the manager, you can't send open files from the manager process to the executor processes.

def run_solver(total, proc_id, result, fouts):
    with open(fouts[proc_id], 'w') as openfile:
        for i in range(10)):
            openfile.write('hi\n')

if __name__ == '__main__':
    processes = []
    with Manager() as manager:
        fouts = manager.list(['0.txt', '1.txt', '2.txt', '3.txt'])
        for proc_id in range(os.cpu_count()):
            processes.append(Process(
                target=run_solver, args=(
                    int(total/os.cpu_count()), proc_id, result, fouts)
            ))

If you are sharing filenames between processes, you want to prevent race conditions when writing to those files, you really want to use a lock per file too:

def run_solver(total, proc_id, result, fouts, locks):
    with open(fouts[proc_id], 'a') as openfile:
        for i in range(10)):
            with locks[proc_id]:
                openfile.write('hi\n')
                openfile.flush()


if __name__ == '__main__':
    processes = []
    with Manager() as manager:
        fouts = manager.list(['0.txt', '1.txt', '2.txt', '3.txt'])
        locks = manager.list([Lock() for fout in fouts])

        for proc_id in range(os.cpu_count()):
            processes.append(Process(
                target=run_solver, args=(
                    int(total/os.cpu_count()), proc_id, result, fouts, locks
                )
            ))

Because the files are opened with with they are closed automatically each time, and they are opened in append mode so different processes don't clobber one another. You do need to remember to flush the write buffer before unlocking again.

As an aside, you probably want to look at the process pools rather than do your own manual pooling.

user_1002970
Related