# Parallel processing

import Queue

import warnings

class SyncPool:

    '''A dummy synchronous Pool to use if multiprocessing is not available'''

    def __init__(self, processes=None): pass

    class Result:

        def __init__(self, value): self.value = value

        def get(timeout=None): return self.value

        def wait(timeout=None): pass

        def ready(): return True

        def successful(): return True # TODO: False if raised exception

    def apply_async(self, func, args=None, kw_args=None, callback=None):

        if args == None: args = ()

        if kw_args == None: kw_args = {}

        if callback == None: callback = lambda v: None

        value = func(*args, **kw_args)

        callback(value)

        return self.Result(value)

class MultiProducerPool:

    '''A multi-producer pool. You must call pool.main_loop() in the thread that

    created this to process new tasks.'''

    def __init__(self, processes=None):

        '''

        @param processes If 0, uses SyncPool

        @post The # processes actually used is made available in self.process_ct

        '''

        try:

            if processes == 0: raise ImportError('turned off')

            import multiprocessing

            import multiprocessing.pool

        except ImportError, e:

            warnings.warn(UserWarning('Not using parallel processing: '+str(e)))

            processes = 1

            Pool_ = SyncPool

            Queue_ = Queue.Queue

        else:

            if processes == None: processes = multiprocessing.cpu_count()

            Pool_ = multiprocessing.pool.Pool

            Queue_ = multiprocessing.Queue

        self.process_ct = processes

        self.pool = Pool_(processes)

        self.queue = Queue_()

    def main_loop(self):

        '''@param pool Must be a pool returned by mk_pool()'''

        try:

            # block=False raises Empty immediately if the queue is empty,

            # which indicates that the program is done

            while True: self.queue.get(block=False)() # each elem is a function

        except Queue.Empty: pass

    class Result:

        def get(timeout=None): raise NotImplementedError()

        def wait(timeout=None): raise NotImplementedError()

        def ready(): raise NotImplementedError()

        def successful(): raise NotImplementedError()

    def apply_async(self, *args, **kw_args):

        def call(): self.pool.apply_async(*args, **kw_args)

        self.queue.put_nowait(call)

        return self.Result()