Threading Pool ähnlich dem Multiprocessing Pool?

Question

Gibt es eine Pool-Klasse für Arbeiter Gewinde ähnlich wie die des Multiprozessor-Moduls Pool-Klasse ?

Mir gefällt zum Beispiel die einfache Möglichkeit, eine Map-Funktion zu parallelisieren

def long_running_func(p):
    c_func_no_gil(p)

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

Ich möchte dies jedoch ohne den Aufwand für die Erstellung neuer Prozesse tun.

Ich weiß von der GIL. In meinem Anwendungsfall wird die Funktion jedoch eine IO-gebundene C-Funktion sein, für die der Python-Wrapper die GIL vor dem eigentlichen Funktionsaufruf freigeben wird.

Muss ich meinen eigenen Threading-Pool schreiben?

Answer 1

Ich habe gerade herausgefunden, dass es tatsächlich es eine Thread-basierte Pool-Schnittstelle in der multiprocessing Modul, allerdings ist es etwas versteckt und nicht richtig dokumentiert.

Es kann importiert werden über

from multiprocessing.pool import ThreadPool

Sie wird mit einer Dummy-Prozessklasse implementiert, die einen Python-Thread umhüllt. Diese Thread-basierte Process-Klasse finden Sie in multiprocessing.dummy die kurz in der docs . Dieses Dummy-Modul bietet angeblich die gesamte Multiprozessor-Schnittstelle auf der Grundlage von Threads.

Answer 2

In Python 3 können Sie concurrent.futures.ThreadPoolExecutor d.h.:

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

Siehe die docs für weitere Informationen und Beispiele.

Answer 3

Ja, und es scheint (mehr oder weniger) die gleiche API zu haben.

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

Answer 4

Für etwas sehr Einfaches und Leichtes (leicht abgewandelt von aquí ):

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

Um Rückrufe bei Aufgabenabschluss zu unterstützen, können Sie den Rückruf einfach zum Aufgabentupel hinzufügen.

Answer 5

Hallo, um den Thread-Pool in Python zu verwenden, können Sie diese Bibliothek verwenden:

from multiprocessing.dummy import Pool as ThreadPool

und dann für die Verwendung, diese Bibliothek tun, wie das :

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

Die Threads sind die Anzahl der gewünschten Threads und die Aufgaben sind eine Liste der Aufgaben, die dem Dienst am ehesten zugeordnet sind.