gistfile1.txt

what ho?

procfunc.py

#!/usr/bin/env python

# from python3 at
# http://groups.google.com/group/comp.lang.python/browse_thread/thread/e6c67a09fb20cdec

import time 
import random 

from multiprocessing import Pool, Queue, Manager, current_process

def procfunc(queue): 
    time.sleep(random.random() * 2) 
    print current_process()
    return queue.get()*2 


def printrange(n): 
    for i in range(n): 
        print("generated " + str(i)) 
        yield i 


def printfunc(s):
    print s


if __name__ == "__main__": 
    sm = Manager() 
    pool = Pool() 
    queuelen = len(pool._pool) * 2 
    queue = sm.Queue(queuelen) 
    for i in printrange(10): 
        queue.put(i) 
        pool.apply_async(procfunc, (queue,), callback=printfunc) 
    print 'closing'
    pool.close() 
    print 'joining'
    pool.join() 
    print 'done'

thing.py

import time
from multiprocessing import Process, Pipe

class Processor(object):

    def __init__(self, sleeptime=0.1):
        self.p = Process(target=self.do)
        self.sleeptime = sleeptime

    def start(self):
        self.p.start()

    def do(self):
        self.pid = self.p.pid
        while True:
            self.loop_once()
            time.sleep(self.sleeptime)

    def loop_once(self):
        pass

class Receiver(Processor):

    def __init__(self, conn):
        self.connection = conn
        Processor.__init__(self)
        self.receiver = self.receiver_gen()

    def receiver_gen(self):
        while True:
            yield self.connection.recv()

    def _input(self):
        return self.receiver.next()

    def loop_once(self):
        input = self._input()
        print '%s: got %s from %s' % (self.pid, input[0], input[1])

class Sender(Processor):

    def __init__(self, conn):
        self.connection = conn
        Processor.__init__(self)
        self.sender = self.sender_gen()
        self.sender.send(None)

    def sender_gen(self):
        while True:
            self.connection.send((yield))

    def _output(self, output):
        self.sender.send(output)


    def loop_once(self):
        self._output((time.time(), self.pid))

class FileSender(Sender):

    def __init__(self, conn):
        Sender.__init__(self, conn)
        self.file = open('./sample')

    def loop_once(self):
        where = self.file.tell()
        line = self.file.readline()
        if not line:
            self.file.seek(where)
            return
        else:
            self._output((line, self.pid))

class FileReceiver(Receiver):

    def __init__(self, conn):
        Receiver.__init__(self, conn)
        self.line_count = 0

    def loop_once(self):
        input = self._input()
        self.line_count += 1
        print 'lines: %s' % self.line_count



def startup():
    (conna, connb) = Pipe()
    nancy = FileSender(conna)
    paul = FileReceiver(connb)
    nancy.start()
    paul.start()


if __name__ == '__main__':
    startup()