Comment puis-je utiliser le filetage en Python?

Depuis que cette question a été posée en 2010, il y a eu une réelle simplification dans la façon de faire simple multithreading avec Python avec carte et piscine.

Le code ci-dessous provient d'un article/billet de blog que vous devriez certainement vérifier (pas d'affiliation) - Le parallélisme en une ligne: Un Meilleur Modèle pour le quotidien le Filetage des Tâches. Je vais résumer ci-dessous -, il finit par être juste quelques lignes de code:

from multiprocessing.dummy import Pool as ThreadPool
pool = ThreadPool(4)
results = pool.map(my_function, my_array)

Qui est la version multithread:

results = []
for item in my_array:
    results.append(my_function(item))

---

Description

La carte est un peu cool de la fonction, et la clé de facilement injecter du parallélisme dans votre code Python. Pour ceux qui ne connaissent pas, la carte est levé quelque chose de fonctionnel langages comme Lisp. C'est une fonction qui associe à une autre fonction au cours d'une séquence.

Carte gère l'itération sur la séquence pour nous, applique la fonction, et stocke tous les résultats dans une liste pratique à la fin.

---

Mise en œuvre

Versions parallèles de la carte de fonction sont fournis par les deux bibliothèques:multitraitement, et aussi sa peu connu, mais tout aussi fantastique étape de l'enfant:multitraitement.dummy.

multiprocessing.dummy est exactement le même que le multitraitement module, mais utilise des threads au lieu (une distinction importante - utiliser plusieurs processus pour les tâches consommatrices d'UC; les fils (et pendant) I/O):

multiprocessing.mannequin réplique de l'API de multitraitement, mais n'est pas plus qu'un wrapper autour de l'enfilage du module.

import urllib2
from multiprocessing.dummy import Pool as ThreadPool

urls = [
  'http://www.python.org',
  'http://www.python.org/about/',
  'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html',
  'http://www.python.org/doc/',
  'http://www.python.org/download/',
  'http://www.python.org/getit/',
  'http://www.python.org/community/',
  'https://wiki.python.org/moin/',
]

# Make the Pool of workers
pool = ThreadPool(4)

# Open the URLs in their own threads
# and return the results
results = pool.map(urllib2.urlopen, urls)

# Close the pool and wait for the work to finish
pool.close()
pool.join()

Et le calendrier résultats:

Single thread:   14.4 seconds
       4 Pool:   3.1 seconds
       8 Pool:   1.4 seconds
      13 Pool:   1.3 seconds

---

Passer plusieurs arguments (fonctionne comme ceci seulement en Python 3.3 et versions ultérieures):

Pour passer plusieurs tableaux:

results = pool.starmap(function, zip(list_a, list_b))

Ou de passer d'une constante et d'un tableau:

results = pool.starmap(function, zip(itertools.repeat(constant), list_a))

Si vous utilisez une version antérieure de Python, vous pouvez passer plusieurs arguments via cette solution de contournement).

(Grâce à user136036 pour le commentaire.)

NOTE: Pour réelle la parallélisation en Python, vous devez utiliser le multitraitement module de la fourche à la fourchette de multiples processus qui s'exécutent en parallèle (en raison du blocage de l'interpreteur global, Python fils de fournir de l'entrelacement, mais ils sont en fait exécutées en série, non pas en parallèle, et ne sont utiles que si l'entrelacement des opérations d'e/S).

Toutefois, si vous êtes simplement à la recherche pour l'entrelacement (ou faire des opérations d'e/S peut être parallélisée malgré le blocage de l'interpreteur global), puis le filetage module est l'endroit pour commencer. Comme vraiment un exemple simple, considérons le problème de l'addition d'une large gamme en additionnant les sous-plages en parallèle:

import threading

class SummingThread(threading.Thread):
     def __init__(self,low,high):
         super(SummingThread, self).__init__()
         self.low=low
         self.high=high
         self.total=0

     def run(self):
         for i in range(self.low,self.high):
             self.total+=i


thread1 = SummingThread(0,500000)
thread2 = SummingThread(500000,1000000)
thread1.start() # This actually causes the thread to run
thread2.start()
thread1.join()  # This waits until the thread has completed
thread2.join()
# At this point, both threads have completed
result = thread1.total + thread2.total
print result

Remarque que ci-dessus est un très stupide exemple, comme il ne fait absolument aucun I/O et seront exécutées en série mais entrelacés (avec la charge supplémentaire de la commutation de contexte) dans Disponible fait de la mondialisation de l'interprète de verrouillage.

Juste une remarque: Une file d'attente n'est pas nécessaire pour l'enfilage.

C'est l'exemple le plus simple que j'ai pu imaginer que la montre 10 processus qui s'exécutent simultanément.

import threading
from random import randint
from time import sleep


def print_number(number):

    # Sleeps a random 1 to 10 seconds
    rand_int_var = randint(1, 10)
    sleep(rand_int_var)
    print "Thread " + str(number) + " slept for " + str(rand_int_var) + " seconds"

thread_list = []

for i in range(1, 10):

    # Instantiates the thread
    # (i) does not make a sequence, so (i,)
    t = threading.Thread(target=print_number, args=(i,))
    # Sticks the thread in a list so that it remains accessible
    thread_list.append(t)

# Starts threads
for thread in thread_list:
    thread.start()

# This blocks the calling thread until the thread whose join() method is called is terminated.
# From http://docs.python.org/2/library/threading.html#thread-objects
for thread in thread_list:
    thread.join()

# Demonstrates that the main process waited for threads to complete
print "Done"

J'ai trouvé cela très utile: créer autant de threads que de cœurs et de les laisser exécuter un (grand) nombre de tâches (dans ce cas, l'appel d'un programme shell):

import Queue
import threading
import multiprocessing
import subprocess

q = Queue.Queue()
for i in range(30): # Put 30 tasks in the queue
    q.put(i)

def worker():
    while True:
        item = q.get()
        # Execute a task: call a shell program and wait until it completes
        subprocess.call("echo " + str(item), shell=True)
        q.task_done()

cpus = multiprocessing.cpu_count() # Detect number of cores
print("Creating %d threads" % cpus)
for i in range(cpus):
     t = threading.Thread(target=worker)
     t.daemon = True
     t.start()

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

Python 3 a de la facilité de le lancement de tâches en parallèle. Cela rend notre travail plus facile.

Il a mise en pool de threads et processus de mise en commun.

La suite donne un aperçu:

ThreadPoolExecutor Exemple (source)

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
        'http://www.cnn.com/',
        'http://europe.wsj.com/',
        'http://www.bbc.co.uk/',
        'http://some-made-up-domain.com/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

ProcessPoolExecutor (source)

import concurrent.futures
import math

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]

def is_prime(n):
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))

if __name__ == '__main__':
    main()

Pour moi, l'exemple parfait pour le filetage est de la surveillance des événements asynchrones. Regardez ce code.

# thread_test.py
import threading
import time

class Monitor(threading.Thread):
    def __init__(self, mon):
        threading.Thread.__init__(self)
        self.mon = mon

    def run(self):
        while True:
            if self.mon[0] == 2:
                print "Mon = 2"
                self.mon[0] = 3;

Vous pouvez jouer avec ce code par l'ouverture d'un IPython session et de faire quelque chose comme:

>>> from thread_test import Monitor
>>> a = [0]
>>> mon = Monitor(a)
>>> mon.start()
>>> a[0] = 2
Mon = 2
>>>a[0] = 2
Mon = 2

Attendre quelques minutes

>>> a[0] = 2
Mon = 2

J'ai vu beaucoup d'exemples ici où aucun véritable travail a été effectué, et ils étaient pour la plupart par le CPU. Voici un exemple d'un CPU tâche qui calcule tous les nombres premiers compris entre 10 millions et 10,05 millions de dollars. J'ai utilisé toutes les quatre méthodes ici:

import math
import timeit
import threading
import multiprocessing
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
def time_stuff(fn):
"""
Measure time of execution of a function
"""
def wrapper(*args, **kwargs):
t0 = timeit.default_timer()
fn(*args, **kwargs)
t1 = timeit.default_timer()
print("{} seconds".format(t1 - t0))
return wrapper
def find_primes_in(nmin, nmax):
"""
Compute a list of prime numbers between the given minimum and maximum arguments
"""
primes = []
# Loop from minimum to maximum
for current in range(nmin, nmax + 1):
# Take the square root of the current number
sqrt_n = int(math.sqrt(current))
found = False
# Check if the any number from 2 to the square root + 1 divides the current numnber under consideration
for number in range(2, sqrt_n + 1):
# If divisible we have found a factor, hence this is not a prime number, lets move to the next one
if current % number == 0:
found = True
break
# If not divisible, add this number to the list of primes that we have found so far
if not found:
primes.append(current)
# I am merely printing the length of the array containing all the primes, but feel free to do what you want
print(len(primes))
@time_stuff
def sequential_prime_finder(nmin, nmax):
"""
Use the main process and main thread to compute everything in this case
"""
find_primes_in(nmin, nmax)
@time_stuff
def threading_prime_finder(nmin, nmax):
"""
If the minimum is 1000 and the maximum is 2000 and we have four workers,
1000 - 1250 to worker 1
1250 - 1500 to worker 2
1500 - 1750 to worker 3
1750 - 2000 to worker 4
so let’s split the minimum and maximum values according to the number of workers
"""
nrange = nmax - nmin
threads = []
for i in range(8):
start = int(nmin + i * nrange/8)
end = int(nmin + (i + 1) * nrange/8)
# Start the thread with the minimum and maximum split up to compute
# Parallel computation will not work here due to the GIL since this is a CPU-bound task
t = threading.Thread(target = find_primes_in, args = (start, end))
threads.append(t)
t.start()
# Don’t forget to wait for the threads to finish
for t in threads:
t.join()
@time_stuff
def processing_prime_finder(nmin, nmax):
"""
Split the minimum, maximum interval similar to the threading method above, but use processes this time
"""
nrange = nmax - nmin
processes = []
for i in range(8):
start = int(nmin + i * nrange/8)
end = int(nmin + (i + 1) * nrange/8)
p = multiprocessing.Process(target = find_primes_in, args = (start, end))
processes.append(p)
p.start()
for p in processes:
p.join()
@time_stuff
def thread_executor_prime_finder(nmin, nmax):
"""
Split the min max interval similar to the threading method, but use a thread pool executor this time.
This method is slightly faster than using pure threading as the pools manage threads more efficiently.
This method is still slow due to the GIL limitations since we are doing a CPU-bound task.
"""
nrange = nmax - nmin
with ThreadPoolExecutor(max_workers = 8) as e:
for i in range(8):
start = int(nmin + i * nrange/8)
end = int(nmin + (i + 1) * nrange/8)
e.submit(find_primes_in, start, end)
@time_stuff
def process_executor_prime_finder(nmin, nmax):
"""
Split the min max interval similar to the threading method, but use the process pool executor.
This is the fastest method recorded so far as it manages process efficiently + overcomes GIL limitations.
RECOMMENDED METHOD FOR CPU-BOUND TASKS
"""
nrange = nmax - nmin
with ProcessPoolExecutor(max_workers = 8) as e:
for i in range(8):
start = int(nmin + i * nrange/8)
end = int(nmin + (i + 1) * nrange/8)
e.submit(find_primes_in, start, end)
def main():
nmin = int(1e7)
nmax = int(1.05e7)
print("Sequential Prime Finder Starting")
sequential_prime_finder(nmin, nmax)
print("Threading Prime Finder Starting")
threading_prime_finder(nmin, nmax)
print("Processing Prime Finder Starting")
processing_prime_finder(nmin, nmax)
print("Thread Executor Prime Finder Starting")
thread_executor_prime_finder(nmin, nmax)
print("Process Executor Finder Starting")
process_executor_prime_finder(nmin, nmax)
main()

Voici le résultat sur mon Mac OS X quatre-core de la machine

Sequential Prime Finder Starting
9.708213827005238 seconds
Threading Prime Finder Starting
9.81836523200036 seconds
Processing Prime Finder Starting
3.2467174359990167 seconds
Thread Executor Prime Finder Starting
10.228896902000997 seconds
Process Executor Finder Starting
2.656402041000547 seconds

Ici est multi threading avec un exemple simple qui peut être utile. Vous pouvez l'exécuter et de comprendre facilement comment multi threading est de travailler en Python. J'ai utilisé un verrou pour empêcher l'accès à d'autres threads jusqu'à ce que la précédente threads terminé leur travail. Par l'utilisation de cette ligne de code,

tLock = filetage.BoundedSemaphore(valeur=4)

vous pouvez permettre à un certain nombre de processus à la fois et garder le reste des fils qui se déroulera plus tard, ou après avoir terminé le processus précédents.

import threading
import time
#tLock = threading.Lock()
tLock = threading.BoundedSemaphore(value=4)
def timer(name, delay, repeat):
print  "\r\nTimer: ", name, " Started"
tLock.acquire()
print "\r\n", name, " has the acquired the lock"
while repeat > 0:
time.sleep(delay)
print "\r\n", name, ": ", str(time.ctime(time.time()))
repeat -= 1
print "\r\n", name, " is releaseing the lock"
tLock.release()
print "\r\nTimer: ", name, " Completed"
def Main():
t1 = threading.Thread(target=timer, args=("Timer1", 2, 5))
t2 = threading.Thread(target=timer, args=("Timer2", 3, 5))
t3 = threading.Thread(target=timer, args=("Timer3", 4, 5))
t4 = threading.Thread(target=timer, args=("Timer4", 5, 5))
t5 = threading.Thread(target=timer, args=("Timer5", 0.1, 5))
t1.start()
t2.start()
t3.start()
t4.start()
t5.start()
print "\r\nMain Complete"
if __name__ == "__main__":
Main()

Aucune des solutions précédentes effectivement utilisé plusieurs cœurs sur mon GNU/Linux server (où je n'ai pas les droits administrateur). Ils ont juste couru sur un seul core.

J'ai utilisé le niveau inférieur os.fork interface pour frayer les multiples processus. C'est le code qui a fonctionné pour moi:

from os import fork
values = ['different', 'values', 'for', 'threads']
for i in range(len(values)):
p = fork()
if p == 0:
my_function(values[i])
break

import threading
import requests
def send():
r = requests.get('https://www.stackoverlow.com')
thread = []
t = threading.Thread(target=send())
thread.append(t)
t.start()