Pool de threads Python similaire au pool multi-traitement?

Question

Il y a une Piscine de classe worker threads, semblable à la multitraitement du module de classe du Pool?

J'aime par exemple le moyen le plus facile pour paralléliser une carte de fonction

def long_running_func(p):
    c_func_no_gil(p)

p = multiprocessing.Pool(4)
xs = p.map(long_running_func, range(100))

cependant, je voudrais le faire sans la surcharge de créer de nouveaux processus.

Je sais au sujet de la GIL. Cependant, dans mon cas d'utilisation, la fonction sera un IO-borne C de la fonction pour laquelle le wrapper python publiera le GIL avant l'appel de fonction.

Dois-je écrire mon propre filetage piscine?

Answer 1

Je viens de découvrir qu'il y en est un basé sur le thread du Pool d'interface dans l' multiprocessing module, cependant, il est caché quelque peu, et ne sont pas correctement documentés.

Il peut être importé via

from multiprocessing.pool import ThreadPool

Il est mis en œuvre à l'aide d'un mannequin de classe de Processus d'emballage d'un python fil. Ce fil basée sur la classe de Processus peuvent être trouvés en multiprocessing.dummy qui est brièvement mentionnée dans les docs. Ce module factice soi-disant fournit l'ensemble de multitraitement interface basée sur les threads.

Answer 2

En Python 3, vous pouvez utiliser concurrent.futures.ThreadPoolExecutor , c'est-à-dire:

 executor = ThreadPoolExecutor(max_workers=10)
a = executor.submit(my_function)
 

Voir les documents pour plus d'informations et des exemples.

Answer 3

Oui, et il semble avoir (plus ou moins) la même API.

 import multiprocessing as mult

def worker(lnk):
    ....    
def start_process():
    .....
....

if(PROCESS):
    pool = mult.Pool(processes=POOL_SIZE, initializer=start_process)
else:
    from multiprocessing.pool import ThreadPool
    pool = ThreadPool(processes=POOL_SIZE, initializer=start_process)

pool.map(worker, inputs)
....
 

Answer 4

Pour quelque chose de très simple et léger (légèrement modifié d' ici ):

 from Queue import Queue
from threading import Thread

    class Worker(Thread):
        """Thread executing tasks from a given tasks queue"""
        def __init__(self, tasks):
            Thread.__init__(self)
            self.tasks = tasks
            self.daemon = True
            self.start()

        def run(self):
            while True:
                func, args, kargs = self.tasks.get()
                try:
                    func(*args, **kargs)
                except Exception, e:
                    print e
                finally:
                    self.tasks.task_done()

    class ThreadPool:
        """Pool of threads consuming tasks from a queue"""
        def __init__(self, num_threads):
            self.tasks = Queue(num_threads)
            for _ in range(num_threads): Worker(self.tasks)

        def add_task(self, func, *args, **kargs):
            """Add a task to the queue"""
            self.tasks.put((func, args, kargs))

        def wait_completion(self):
            """Wait for completion of all the tasks in the queue"""
            self.tasks.join()

if __name__ == '__main__':
    from random import randrange
    from time import sleep

    delays = [randrange(1, 10) for i in range(100)]

    def wait_delay(d):
        print 'sleeping for (%d)sec' % d
        sleep(d)

    pool = ThreadPool(20)

    for i, d in enumerate(delays):
        pool.add_task(wait_delay, d)

    pool.wait_completion()
 

Pour prendre en charge les rappels à la fin de la tâche, vous pouvez simplement ajouter le rappel au tuple de la tâche.

Answer 5

Voici quelque chose qui semble prometteur dans le livre de recettes Python:

Recette 576519: Pool de threads avec la même API que (multi) processing.Pool (Python)