Threading pool similaire au pool multiprocessing?

Existe-t-il une classe Pool pour les threads de travail , similaire à la classe Pool du module de multitraitement ?

J'aime par exemple la manière simple de paralléliser une fonction de carte

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 connais le GIL. Cependant, dans mon cas d'utilisation, la fonction sera une fonction C liée aux IO pour laquelle le wrapper python libérera le GIL avant l'appel de la fonction réelle.

Dois-je écrire mon propre pool de threads?

Je viens de découvrir qu'il existe en fait une interface de pool basée sur les threads dans le multiprocessingmodule, mais elle est quelque peu cachée et n'est pas correctement documentée.

Il peut être importé via

from multiprocessing.pool import ThreadPool

Il est implémenté à l'aide d'une classe Process factice enveloppant un thread python. Cette classe de processus basée sur les threads peut être trouvée dans multiprocessing.dummylaquelle est brièvement mentionnée dans la documentation . Ce module factice fournit censément l'ensemble de l'interface de multitraitement basée sur les threads.

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

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

Voir la documentation pour plus d'informations et d'exemples.

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

import multiprocessing

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

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

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

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.

Salut pour utiliser le pool de threads en Python, vous pouvez utiliser cette bibliothèque:

from multiprocessing.dummy import Pool as ThreadPool

puis pour utilisation, cette bibliothèque fait comme ça:

pool = ThreadPool(threads)
results = pool.map(service, tasks)
pool.close()
pool.join()
return results

Les threads sont le nombre de threads que vous souhaitez et les tâches sont une liste des tâches les plus mappées au service.

Voici le résultat que j'ai finalement utilisé. C'est une version modifiée des classes par dgorissen ci-dessus.

Fichier: threadpool.py

from queue import Queue, Empty
import threading
from threading import Thread


class Worker(Thread):
    _TIMEOUT = 2
    """ Thread executing tasks from a given tasks queue. Thread is signalable, 
        to exit
    """
    def __init__(self, tasks, th_num):
        Thread.__init__(self)
        self.tasks = tasks
        self.daemon, self.th_num = True, th_num
        self.done = threading.Event()
        self.start()

    def run(self):       
        while not self.done.is_set():
            try:
                func, args, kwargs = self.tasks.get(block=True,
                                                   timeout=self._TIMEOUT)
                try:
                    func(*args, **kwargs)
                except Exception as e:
                    print(e)
                finally:
                    self.tasks.task_done()
            except Empty as e:
                pass
        return

    def signal_exit(self):
        """ Signal to thread to exit """
        self.done.set()


class ThreadPool:
    """Pool of threads consuming tasks from a queue"""
    def __init__(self, num_threads, tasks=[]):
        self.tasks = Queue(num_threads)
        self.workers = []
        self.done = False
        self._init_workers(num_threads)
        for task in tasks:
            self.tasks.put(task)

    def _init_workers(self, num_threads):
        for i in range(num_threads):
            self.workers.append(Worker(self.tasks, i))

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

    def _close_all_threads(self):
        """ Signal all threads to exit and lose the references to them """
        for workr in self.workers:
            workr.signal_exit()
        self.workers = []

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

    def __del__(self):
        self._close_all_threads()


def create_task(func, *args, **kwargs):
    return (func, args, kwargs)

Pour utiliser la piscine

from random import randrange
from time import sleep

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

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()

Les frais généraux liés à la création de nouveaux processus sont minimes, surtout lorsqu'il ne s'agit que de 4 d'entre eux. Je doute que ce soit un point chaud de performance de votre application. Restez simple, optimisez où vous devez et où les résultats du profilage pointent.

Il n'y a pas de pool basé sur les threads. Cependant, il peut être très rapide d'implémenter une file d'attente producteur / consommateur avec la Queueclasse.

De: https://docs.python.org/2/library/queue.html

from threading import Thread
from Queue import Queue
def worker():
    while True:
        item = q.get()
        do_work(item)
        q.task_done()

q = Queue()
for i in range(num_worker_threads):
     t = Thread(target=worker)
     t.daemon = True
     t.start()

for item in source():
    q.put(item)

q.join()       # block until all tasks are done

une autre façon peut être d'ajouter le processus au pool de files d'attente

import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=cpus) as executor:
    for i in range(0, len(list_of_files) - 1):
        a = executor.submit(loop_files2, i, list_of_files2, mt_list, temp_path, mt_dicto)