CODEWITHSUNDEEP
pythonpython-3.xmultiprocessing
voltas
voltas

Reputation: 563

Python: store results of ProcessPoolExecutor

I'm very new to parallel processing with "concurrent.futures". Code seems to work, but I am not sure how to store the result of each process, therey by marking the build as failed at last, if any of processes's return value is not zero. Tried to create a list (exit_status) and append the results to that, but that shows IndexError. Wondering what can I do right?

#!/usr/bin/env python3

import concurrent.futures
import sys
import shutil
import os
import glob
import multiprocessing as mp
import json
from os import path


def slave(path1, path2, target):
    os.makedirs(target)
    shutil.copy(path1, target)
    shutil.copy(path2, target)
    os.system(<Login command>)
    os.system(<Image creation command>)
    os.system(<Copy to Other slaves or NFS>)
    
    #If any one of the above operation or command fails for any of the process, the script should return 1 at the end of the execution or fail the build at last.


def main():
    processed = {}
    exit_status = []
    with open('example.json', 'r') as f:
        data = json.load(f)
        for value in data.items():
            for line in value[1]:
                if line.endswith('.zip'):
                    targz = line
                elif line.endswith('.yaml'):
                    yaml = line
            processed[targz] = yaml


    with concurrent.futures.ProcessPoolExecutor() as executor:
        for id, (path2, path1) in enumerate(processed.items(), 1):
            target = path.join("/tmp", "dir" + str(id))
            ret = executor.submit(slave, path1, path2, target)
            exit_status.append(ret.result())

    for i in exit_status:
    print("##########Result status: ", i)
    
if __name__ == "__main__":
    mp.set_start_method('spawn')
    main()

exit_status list's output:

 ##########Result status:  None
##########Result status:  None

Upvotes: 0

Views: 2102

Answers (1)

Aaron
Aaron

Reputation: 11075

re; comments

If you want to get the result of a system call in order to act on the results of it, using subprocess.run is much more flexible and powerful than os.system. Additionally, if you actually want to perform the operations in parallel, you can't wait on result() after each task. Otherwise you're only ever doing one thing at a time. Better to submit all the tasks, and collect the Future objects. Then you can iterate over those and wait on each result() now that you've submitted all the work you want the executor to do.

def target_func(path1, path2, target):
    #...
    #instead of os.system, use subprocess.run
    
    #you can inspect the stdout from the process
    complete_process = subprocess.run(<Login command>, text=True, capture_output=True)
    if "success" not in complete_process.stdout:
        return "uh-oh"
    #you can also just check the return value (0 typically means clean exit)
    if subprocess.run(<Image creation command>).returncode == 0:
        return "uh-oh"
    #or you can tell `run` to generate an error if the returncode is non-zero
    try:
        subprocess.run(<Copy to Other slaves or NFS>, check=True)
    except subprocess.CalledProcessError:
        return "uh-oh"
    return "we did it!"

def main():
    #...
    #...
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for id, (path2, path1) in enumerate(processed.items(), 1):
            target = path.join("/tmp", "dir" + str(id))
            ret = executor.submit(slave, path1, path2, target)
            exit_status.append(ret)

    for i in exit_status:
        print("##########Result status: ", i.result())

Upvotes: 1

Related Questions