16

I'm currently launching a subprocess and parsing stdout on the go without waiting for it to finish to parse stdout.

for sample in all_samples:
    my_tool_subprocess = subprocess.Popen('mytool {}'.format(sample),shell=True, stdout=subprocess.PIPE)
    line = True
    while line:
        myline = my_tool_subprocess.stdout.readline()
        #here I parse stdout..

In my script I perform this action multiple times, indeed depending on the number of input samples.

Main problem here is that every subprocess is a program/tool that uses 1 CPU for 100% while it's running. And it takes sometime.. maybe 20-40 min per input.

What I would like to achieve, is to set a pool, queue (I'm not sure what's the exact terminology here) of N max subprocess job process running at same time. So I could maximize performance, and not proceed sequentially.

So an execution flow for example a max 4 jobs pool should be:

  • Launch 4 subprocess.
  • When one of jobs finishes, parse stdout and launch next.
  • Do this until all the jobs in queue are finished.

If I can achieve this I really don't know how I could identify which sample subprocess is the one that has finished. At this moment, I don't need to identify them since each subprocess runs sequentially and I parse stdout as subprocess is printing stdout.

This is really important, since I need to identify the output of each subprocess and assign it to it's corresponding input/sample.

3 Answers 3

28

ThreadPool could be a good fit for your problem, you set the number of worker threads and add jobs, and the threads will work their way through all the tasks.

from multiprocessing.pool import ThreadPool
import subprocess


def work(sample):
    my_tool_subprocess = subprocess.Popen('mytool {}'.format(sample),shell=True, stdout=subprocess.PIPE)
    line = True
    while line:
        myline = my_tool_subprocess.stdout.readline()
        #here I parse stdout..


num = None  # set to the number of workers you want (it defaults to the cpu count of your machine)
tp = ThreadPool(num)
for sample in all_samples:
    tp.apply_async(work, (sample,))

tp.close()
tp.join()
10
  • Wow! This seems to be exactly what I needed. If in a future the tool I use in work() can benefit from multiple core so I could use 2 CPUs(cores) per created thread is there any way to control this to not exceed computer number of cores? (ie: I spawn 4 threads each one using 4 cores when my machines has only 8 cores.)
    – gmarco
    Nov 7, 2014 at 8:31
  • @gmarco This can't tell anything about the type of work, or how many cores it uses/needs. If you don't specify the number of threads to ThreadPool (the first argument) it uses the function multiprocessing.cpu_count() to get the cpu count, and spins up that many threads. So if you know how many cores each job needs, you can work this out yourself (ThreadPool(multiprocessing.cpu_count() / 2) for example)
    – GP89
    Nov 7, 2014 at 11:54
  • Thanks! I had some serious issues with subprocess. Popen just not launching and generally not working for me for some reason and I just switched to ThreadPool + os.system() and it's even better than Popen was supposed to be!
    – Íhor Mé
    Mar 23, 2020 at 17:56
  • @ÍhorMé You're better off using subprocess. but you can deadlock it if there's too much output, and you're not handling stdout/err properly
    – GP89
    Mar 25, 2020 at 13:26
  • subprocess didn't spawn threads for some reason for me. I found no resolution to that problem. It's probably some bug. (Not on my side)
    – Íhor Mé
    Mar 27, 2020 at 22:34
0

well if this is the case you should write the same code above without proc.join() in this case the main thread (main) will start the other four threads, this the case that multithreading in a single process (in other words no benefits of multicore processor) to benefit from multicore processor you should use the multiprocessing module like this:

proc = multiprocessing.Process(target=func, args=(funarguments))      
proc.start()

this way each would be a separate process and separate processes can run completely independently of one another

1
  • 1
    There will be a benefit, as the subprocesses are in separate processes (the python threads are just waiting for output)
    – GP89
    Nov 6, 2014 at 15:56
0

well as i understood your question your problem is that the result of the first process after its finished is supplied to the second process, then to the third and so on, to achieve this you should import threading module and use the class Thread:

proc = threading.Thread(target=func, args=(func arguments) # Thread class
proc.start()                                   # starting the thread
proc.join()                                    # this ensures that the next thread does no 

start until the previous one has finished.....

1
  • Not really. I mean each process input produces an output. I don't previous process output to compute the next one. That's why I wish I could run them in parallel. The problem is that at this moment, I run them one by one in a sequential way.
    – gmarco
    Nov 6, 2014 at 14:40

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.