trjhtr

This code is a multi-threaded version of filter that uses pipes to communicate between the workers and the main thread. I am looking for any ways to make this code cleaner. I'm also interested in if there is anything I can do to make this play nicely with infinite iterators.

from multiprocessing import Process, Pipe
from itertools import cycle, count

def distributer(f, data):
 def worker(pipe):
 while True:
 piece = pipe.recv()
 if f(piece):
 pipe.send(piece)
 pipe.send(False)

 pipes = 
 for _ in range(8):
 parent_conn, child_conn = Pipe()
 p = Process(target=worker, args=(child_conn,))
 pipes.append(parent_conn)
 p.start()
 parent_conn.send(next(data))

 for pipe in cycle(pipes):
 if pipe.poll(.0001):
 ans = pipe.recv()
 if ans:
 yield ans
 pipe.send(next(data))

if __name__ == '__main__':
 def g(x):
 return x%1000==0
 data = count(1)
 for ans in distributer(g, data):
 print(ans)

I recommend a signature of:

def distributer(f, data, num_workers=8):
...
 for _ in range(num_workers):

Perhaps f is the appropriate identifier. But it leaves me scratching my head a bit. I wouldn't mind a docstring explaining the function's responsibility.

 p = Process(target=worker, args=(child_conn,))

Ordinarily this would be perfect. But given that you're also manipulating pipes, it wouldn't hurt to assign to proc instead.

Your prose suggests that you're filtering, in this case picking out one value in a thousand. Perhaps the appropriate function name would have been filter_worker().

Phrasing it as "reduce" may have been clearer than "filter". For example, every thousandth row return sum of previous rows, that sort of thing, if the code is strictly for illustrative purposes.

In summary, as a reader I'm looking for a comment / docstring that explains what is interesting about the top level code doing busy work forever (I see no benchmark timings), and I'm interested in a docstring on worker() explaining the protocol it implements.