Iterator with multithreading

Question

I need to implement an Iterator that returns two values (nothing unusual so far), but these values need to be continuously computed/generated in parallel, even when the iterator is not solicited.

Here is an example to explain what I need.

def GenerateValues()
    #I do the math for value1 in the first thread
    #I do the math for value2 in the second thread
    return value1 value2

def myIterator()
    while 1:
        yield GenerateValues()

In this situation, value1 and value2 are computed/generated in parallel only when the function myIterator is called. But in my problem, it takes a long time to compute/generate value1 and value2, but it also takes a very long time to process value1 and value2. So when my software is processing value1 and value2, I would like it to compute in parallel the new value1 and new value2.

So it would be something like:

def GenerateValues()
    #If value1 and value 2 are not computed, then wait.

    #I do the math for the new value1 in the first thread without blocking
    #I do the math for the new value2 in the second thread without blocking
    return value1 value2

def myIterator()
    while 1:
        yield GenerateValues()

With such configuration, the new value1 and new value2 are computed/generated while value1 and value2 are returned to be processed.

• Is that clear enough?
• If yes, how could I do such synchronization?

Thanks in advance for your help!

PS: I need the while 1, no need to comment on that point.

Answer 1

Not sure if I fully understand what you are trying to do, if you are trying to calculate value1 and value2 in parallel, you could use multiprocessing or threading. I do recommend multiprocessing if the task is CPU-bound, to fully take advantage of your CPU to "side-step" the Global Interpreter Lock (GIL) by using subprocesses instead of threads.

This is a rather straight forward example of using multiprocessing:

from multiprocessing import Queue, Process

def cal_value1(queue):
    # do the job
    queue.put({'value1': value1})

def cal_value2(queue):
    # do the job
    queue.put({'value2': value2})

def GenerateValues()
    #If value1 and value 2 are not computed, then wait.
    queue = Queue() # 
    process_1 = Process(target=cal_value1, args=(queue, ))
    process_2 = Process(target=cal_value2, args=(queue, ))
    process_1.start()
    process_2.start()  # start both processes
    process_1.join()
    process_2.join()  # wait for both to finish

    result = queue.get()
    result.update(queue.get()) # get results

    return result['value1'], result['value2']

P.S. You can easily use threading.Thread and Queue.Queue to replace multiprocessing.Process and multiprocessing.Queue if you want.

Edit: Now let's make cal_value1 and cal_value2 long running processes, and you might want to start these 2 processes in the beginning of your script.

from multiprocessing import Queue, Process

def cal_value1(tasks, results):
    while True:
        task = tasks.get() # this will block until a new task coming in
        # calculate value1
        results.put({'value1': value1})

def cal_value2(tasks, results):
    while True:
        task = tasks.get() # this will block until a new task coming in
        # calculate value2
        results.put({'value2': value2})

def main(）:
    cal_value1_tasks, cal_value2_tasks, results = Queue(), Queue(), Queue()
    process_1 = Process(target=cal_value1, args=(cal_value1_tasks, results, ))
    process_2 = Process(target=cal_value2, args=(cal_value2_tasks, results, ))
    process_1.start()
    process_2.start()
    cal_value1_tasks.put('cal_value1')
    cal_value2_tasks.put('cal_value2') # Start to calculate the first pair
    values = GenerateValues(cal_value1_tasks, cal_value2_tasks, results)

def GenerateValues(cal_value1_tasks, cal_value2_tasks, results):
    values = results.get() # get results
    values.update(queue.get()) # notice that it'll block until both value1 and value 2 calculated
    cal_value1_tasks.put('cal_value1')
    cal_value2_tasks.put('cal_value2') # before returning, start to calculate the next round of value1 and value2
    return values['value1'], values['value2]

Answer 2

In CPython (the most-used implementation that you get from python.org) threads do not really help with parallelizing computations done in python.

Because to make memory management easier (among other things) only one thread at a time can be executing Python bytecode. This is enforced by the global interpreter lock ("GIL").

(If you are doing all your calculations in an extension like numpy, which releases the GIL when it's working, this restriction mostly does not apply)

You could use multiprocessing or (or the ProcesPoolExecutor from concurrent.futures from python 3.2 onwards) to spread the calculation out over multiple processes. There are examples for both in the implementation.

Below is an example where I use a ProcessPoolExecutor to convert DICOM images to JPEG. It uses the "wand" python bindings to ImageMagick. What it does it to make a list of jobs (futures), and start those. The as_completed function returns the result for each future in the order that they finish.

def convert(filename):
    """Convert a DICOM file to a JPEG file, removing the blank areas from the
    Philips x-ray detector.

    Arguments:
        filename: name of the file to convert.

    Returns:
        Tuple of (input filename, output filename)
    """
    outname = filename.strip() + '.jpg'
    with Image(filename=filename) as img:
        with img.convert('jpg') as converted:
            converted.units = 'pixelsperinch'
            converted.resolution = (300, 300)
            converted.crop(left=232, top=0, width=1574, height=2048)
            converted.save(filename=outname)
    return filename, outname


def main(argv):
    """Main entry point for dicom2jpg.py.

    Arguments:
        argv: command line arguments
    """
    if len(argv) == 1:
        binary = os.path.basename(argv[0])
        print("{} ver. {}".format(binary, __version__), file=sys.stderr)
        print("Usage: {} [file ...]\n".format(binary), file=sys.stderr)
        print(__doc__)
        sys.exit(0)
    del argv[0]  # Remove the name of the script from the arguments.
    es = 'Finished conversion of {} to {}'
    with cf.ProcessPoolExecutor(max_workers=os.cpu_count()) as tp:
        fl = [tp.submit(convert, fn) for fn in argv]
        for fut in cf.as_completed(fl):
            infn, outfn = fut.result()
            print(es.format(infn, outfn))

You can find this and other examples in my scripts repository on github.