Can I tell Python's multiprocessing pool not to get too far ahead?

I think you can create a Queue to store your temp files. As Queue can set a max length, if it is full when you use queue.put, it will block until there is still some space. In this case, you can easily make your process to pause.

I took the suggestion to use multiprocessing.Queue, and after a lot of experimentation, I came up with an imap-like function that runs things in parallel using multiple CPUs, like Pool.imap_unordered, but avoids getting too far ahead by using a limited-size queue.

import os
from time import sleep
from multiprocessing import Queue, Process
import atexit
from contextlib import contextmanager

def feed_queue(q, items, sentinel=None, num_sentinels=0):
    for x in items:
        if x == sentinel:
            break
        # print("Feeding {:.1f} into queue".format(x))
        q.put(x)
    for i in range(num_sentinels):
        q.put(sentinel)

class Sentinel:
    def __eq__(self, other):
        return isinstance(other, Sentinel)

class ParallelMapWorkerProcess(Process):
    def __init__(self, target, q_in, q_out, sentinel=None, *args, **kwargs):
        self.q_in = q_in
        self.q_out = q_out
        self.target = target
        self.sentinel_value = sentinel
        super().__init__(*args, **kwargs)

    def run(self):
        try:
            while True:
                x = self.q_in.get()
                if x == self.sentinel_value:
                    break
                result = self.target(x)
                self.q_out.put(result)
            while True:
                self.q_out.put(self.sentinel_value)
        except KeyboardInterrupt:
            pass

@contextmanager
def parallel_imap_buffered(func, iterable, num_cpus=1, buffer_size=1):
    input_queue = Queue(1)
    input_queue.cancel_join_thread()
    output_queue = Queue(buffer_size)
    output_queue.cancel_join_thread()
    sentinel = Sentinel()
    feeder_proc = Process(target=feed_queue, args=(input_queue, iterable, sentinel, num_cpus))
    worker_procs = [ ParallelMapWorkerProcess(func, input_queue, output_queue, sentinel)
                     for i in range(num_cpus) ]
    try:
        feeder_proc.start()
        for p in worker_procs:
            p.start()
        yield iter(output_queue.get, sentinel)
    finally:
        feeder_proc.terminate()
        for p in worker_procs:
            p.terminate()

This abstracts away the details of managing queues and processes. It can be used like this:

def fast_step(x):
    print("Running fast step for {:.1f}".format(x))
    return x + 0.1

def slow_step(x):
    print("Starting slow step for {:.1f}".format(x))
    # Make it slow
    sleep(0.1)
    print("Finishing slow step for {:.1f}".format(x))
    return x + 0.1

input_values = range(50)

with parallel_imap_buffered(fast_step, input_values, num_cpus=4, buffer_size=2) as fast_step_results, \
     parallel_imap_buffered(slow_step, fast_step_results, num_cpus=1, buffer_size=2) as slow_step_results:
    for x in slow_step_results:
        if x >= 10:
            raise Exception("Mid-pipeline exception")
        print('Got value: {:.1f}'.format(x))
    print("Finished")

Using a context manager allows the iterator to kill off the child processes once the iterator is no longer needed, regardless of whether the iterator has been exhausted or not. This seems to work when the main process throws an exception, as demonstrated. If anyone can give an example where this code fails, deadlocks, or does something else bad, please comment. (Edit: After some testing, this code is not ideal. It can run into issues when exceptions are raised from certain locations.)

I'm a little surprised that something like this isn't built into the Python standard library.

You can split it up into batches. That way the main process can sync up with the pool processes in between batches while limiting the number of resources (temp files in your case) used by the pool.

import math
import os
from time import sleep
from multiprocessing.pool import Pool

input_values = list(range(10))

def fast_step(x):
    print("Running fast step for {x}".format(x=x))
    return x

def slow_step(x):
    print("Starting slow step for {x}".format(x=x))
    sleep(1)
    print("Finishing slow step for {x}".format(x=x))
    return x

mypool = Pool(2)

batch_size = 2
batch_count = math.ceil(len(input_values) / batch_size)

for batch_start in range(0, batch_count, batch_size):
    batch_stop = min(batch_start + batch_size, len(input_values))
    input_batch = input_values[batch_start:batch_stop]

    for i in mypool.imap(fast_step, input batch):
        slow_step(i)

I used this solution when I wanted to create images in a pool and stream them into a video in the main process. This way I could limit the max memory usage.

The code example is untested.