Jakie są różnice między modułami obsługującymi wątki i wieloprocesorowe?

Uczę się, jak używać threadingi te multiprocessingmoduły w Pythonie, aby uruchomić pewne operacje równolegle i przyspieszyć mój kod.

Trudno mi (być może dlatego, że nie mam żadnych podstaw teoretycznych), aby zrozumieć, jaka jest różnica między threading.Thread()obiektem a multiprocessing.Process()jedynką.

Ponadto nie jest dla mnie całkowicie jasne, jak utworzyć wystąpienie kolejki zadań i mieć tylko 4 (na przykład) z nich uruchomionych równolegle, podczas gdy inne czekają na zwolnienie zasobów przed wykonaniem.

Uważam, że przykłady w dokumentacji są jasne, ale niezbyt wyczerpujące; gdy tylko próbuję trochę komplikować, otrzymuję wiele dziwnych błędów (takich jak metoda, której nie można wytrawić itd.).

Kiedy więc powinienem używać modułów threadingi multiprocessing?

Czy możesz połączyć mnie z niektórymi zasobami, które wyjaśniają koncepcje tych dwóch modułów i jak ich właściwie używać do złożonych zadań?

To, co mówi Giulio Franco, odnosi się do wielowątkowości w porównaniu z wielowątkowością w ogóle .

Jednak Python ^* ma dodatkowy problem: istnieje globalna blokada interpretera, która uniemożliwia dwóm wątkom w tym samym procesie uruchamianie kodu Pythona w tym samym czasie. Oznacza to, że jeśli masz 8 rdzeni i zmienisz kod tak, aby korzystał z 8 wątków, nie będzie on w stanie wykorzystać 800% procesora i działać 8 razy szybciej; będzie używać tego samego 100% procesora i działać z tą samą prędkością. (W rzeczywistości będzie działać trochę wolniej, ponieważ wątkowanie wiąże się z dodatkowymi kosztami, nawet jeśli nie masz żadnych udostępnionych danych, ale na razie zignoruj ​​to).

Są od tego wyjątki. Jeśli intensywne obliczenia twojego kodu w rzeczywistości nie są wykonywane w Pythonie, ale w jakiejś bibliotece z niestandardowym kodem C, która wykonuje właściwą obsługę GIL, jak aplikacja numpy, otrzymasz oczekiwaną korzyść wydajności z wątków. To samo dotyczy sytuacji, gdy ciężkie obliczenia są wykonywane przez jakiś podproces, który uruchamiasz i na który czekasz.

Co ważniejsze, są przypadki, w których to nie ma znaczenia. Na przykład serwer sieciowy spędza większość czasu na odczytywaniu pakietów z sieci, a aplikacja GUI spędza większość czasu na czekaniu na zdarzenia użytkownika. Jednym z powodów używania wątków w serwerze sieciowym lub aplikacji z graficznym interfejsem użytkownika jest umożliwienie wykonywania długotrwałych „zadań w tle” bez zatrzymywania kontynuowania obsługi pakietów sieciowych lub zdarzeń GUI przez główny wątek. I to działa dobrze z wątkami Pythona. (Z technicznego punktu widzenia oznacza to, że wątki Pythona zapewniają współbieżność, mimo że nie zapewniają równoległości rdzenia).

Ale jeśli piszesz program związany z procesorem w czystym Pythonie, używanie większej liczby wątków na ogół nie jest pomocne.

Stosowanie oddzielnych procesów nie stwarza takich problemów w GIL, ponieważ każdy proces ma swój własny, oddzielny GIL. Oczywiście nadal masz takie same kompromisy między wątkami i procesami, jak w każdym innym języku - udostępnianie danych między procesami jest trudniejsze i droższe niż między wątkami, uruchamianie ogromnej liczby procesów lub tworzenie i niszczenie może być kosztowne je często itd. Ale GIL w dużym stopniu waży na szali w procesach, w sposób, który nie jest prawdziwy, powiedzmy, C czy Java. Dlatego w Pythonie znacznie częściej będziesz korzystać z przetwarzania wieloprocesowego niż w C czy Javie.

W międzyczasie filozofia Pythona „w zestawie baterie” przynosi dobre wieści: bardzo łatwo jest napisać kod, który można przełączać między wątkami i procesami za pomocą jednej linii.

Jeśli projektujesz swój kod w kategoriach samodzielnych "zadań", które nie współużytkują niczego z innymi zadaniami (lub programem głównym) poza danymi wejściowymi i wyjściowymi, możesz użyć concurrent.futuresbiblioteki do napisania swojego kodu wokół puli wątków w następujący sposób:

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    executor.submit(job, argument)
    executor.map(some_function, collection_of_independent_things)
    # ...

Możesz nawet pobrać wyniki tych zadań i przekazać je do dalszych zadań, czekać na rzeczy w kolejności wykonania lub kolejności ukończenia itp .; przeczytaj sekcję o Futureobiektach po szczegóły.

Teraz, jeśli okaże się, że twój program stale używa 100% procesora, a dodanie większej liczby wątków tylko go spowolni, oznacza to, że masz problem z GIL, więc musisz przełączyć się na procesy. Wszystko, co musisz zrobić, to zmienić pierwszą linię:

with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:

Jedynym prawdziwym zastrzeżeniem jest to, że argumenty i wartości zwracane zadań muszą być trawione (i nie zajmować zbyt dużo czasu ani pamięci), aby można je było wykorzystać w procesach krzyżowych. Zwykle nie stanowi to problemu, ale czasami jest.

Ale co, jeśli twoja praca nie może być samowystarczalna? Jeśli potrafisz zaprojektować swój kod pod kątem zadań, które przekazują wiadomości od jednego do drugiego, nadal jest to całkiem proste. Być może będziesz musiał użyć threading.Threadlub multiprocessing.Processzamiast polegać na pulach. I trzeba będzie utworzyć queue.Queuelub multiprocessing.Queueobiekty wyraźnie. (Jest wiele innych opcji - potoki, gniazda, pliki ze stokami ... ale chodzi o to, że musisz zrobić coś ręcznie, jeśli automatyczna magia Executora jest niewystarczająca.)

Ale co, jeśli nie możesz nawet polegać na przekazywaniu wiadomości? A jeśli potrzebujesz dwóch miejsc pracy, aby zmutować tę samą strukturę i zobaczyć zmiany innych? W takim przypadku będziesz musiał wykonać ręczną synchronizację (blokady, semafory, warunki itp.) I, jeśli chcesz używać procesów, jawne obiekty pamięci współdzielonej do rozruchu. Dzieje się tak, gdy wielowątkowość (lub wieloprocesowość) staje się trudna. Jeśli możesz tego uniknąć, świetnie; jeśli nie możesz, będziesz musiał przeczytać więcej, niż ktoś może udzielić odpowiedzi TAK.

Z komentarza chciałeś wiedzieć, czym różnią się wątki i procesy w Pythonie. Naprawdę, jeśli przeczytasz odpowiedź Giulio Franco, moją i wszystkie nasze linki, to powinno obejmować wszystko… ale podsumowanie z pewnością byłoby przydatne, więc oto:

1. Wątki domyślnie udostępniają dane; procesy nie.
2. W konsekwencji (1) przesyłanie danych między procesami wymaga zazwyczaj wytrawiania i wytrawiania. ^**
3. Inną konsekwencją (1) jest bezpośrednie udostępnianie danych między procesami na ogół wymaga umieszczenia ich w formatach niskiego poziomu, takich jak wartość, tablica i ctypestypy.
4. Procesy nie podlegają GIL.
5. Na niektórych platformach (głównie Windows) tworzenie i niszczenie procesów jest znacznie droższe.
6. Istnieją dodatkowe ograniczenia dotyczące procesów, z których niektóre są różne na różnych platformach. Szczegółowe informacje można znaleźć w wytycznych dotyczących programowania .
7. threadingModuł nie posiada niektórych cech multiprocessingmodułu. (Możesz użyć, multiprocessing.dummyaby uzyskać większość brakującego interfejsu API na wierzchu wątków, lub możesz użyć modułów wyższego poziomu, takich jak concurrent.futuresi nie martw się o to.)

_{* Właściwie to nie Python, język, ma ten problem, ale CPython, „standardowa” implementacja tego języka. Niektóre inne implementacje nie mają GIL, jak Jython.}

_{** Jeśli używasz metody fork start do przetwarzania wieloprocesowego - co jest możliwe na większości platform innych niż Windows - każdy proces potomny otrzymuje zasoby, które posiadał rodzic, gdy dziecko zostało uruchomione, co może być innym sposobem przekazywania danych dzieciom.}

W jednym procesie może istnieć wiele wątków. Wątki należące do tego samego procesu współużytkują ten sam obszar pamięci (mogą czytać i zapisywać do tych samych zmiennych oraz mogą wzajemnie się zakłócać). Wręcz przeciwnie, w różnych obszarach pamięci istnieją różne procesy, a każdy z nich ma swoje własne zmienne. Aby się komunikować, procesy muszą używać innych kanałów (plików, potoków lub gniazd).

Jeśli chcesz zrównoleglać obliczenia, prawdopodobnie będziesz potrzebować wielowątkowości, ponieważ prawdopodobnie chcesz, aby wątki współpracowały w tej samej pamięci.

Mówiąc o wydajności, wątki są szybsze w tworzeniu i zarządzaniu niż procesy (ponieważ system operacyjny nie musi przydzielać zupełnie nowego obszaru pamięci wirtualnej), a komunikacja między wątkami jest zwykle szybsza niż komunikacja między procesami. Ale wątki są trudniejsze do zaprogramowania. Wątki mogą ze sobą kolidować i mogą zapisywać się nawzajem w pamięci, ale sposób, w jaki to się dzieje, nie zawsze jest oczywisty (ze względu na kilka czynników, głównie zmianę kolejności instrukcji i buforowanie pamięci), dlatego do kontroli dostępu będą potrzebne prymitywy synchronizacji do swoich zmiennych.

Uważam, że ten link w elegancki sposób odpowiada na Twoje pytanie.

Krótko mówiąc, jeśli jeden z podproblemów musi czekać, aż inny się zakończy, wielowątkowość jest dobra (na przykład w ciężkich operacjach we / wy); Z drugiej strony, jeśli twoje podproblemy naprawdę mogą się zdarzyć w tym samym czasie, sugerowane jest przetwarzanie wieloprocesowe. Jednak nie utworzysz więcej procesów niż liczba rdzeni.

Cytaty z dokumentacji Pythona

Podkreśliłem kluczowe cytaty z dokumentacji Pythona dotyczące Process vs Threads i GIL na: Co to jest globalna blokada interpretera (GIL) w CPythonie?

Eksperymenty z procesami a wątkami

Przeprowadziłem trochę benchmarkingu, aby bardziej konkretnie pokazać różnicę.

W teście porównawczym ustaliłem czas pracy związanej z procesorem i we / wy dla różnych liczb wątków na 8-wątkowym procesorze. Praca dostarczana na wątek jest zawsze taka sama, więc więcej wątków oznacza większą całkowitą dostarczoną pracę.

Wyniki były następujące:

Dane wykresu .

Wnioski:

• w przypadku pracy związanej z procesorem wieloprocesorowość jest zawsze szybsza, prawdopodobnie dzięki GIL

• do pracy związanej z IO. obie mają dokładnie taką samą prędkość

• wątki skalują się tylko do około 4x zamiast oczekiwanego 8x, ponieważ jestem na maszynie 8-wątkowej.

  Porównaj to z pracą związaną z procesorem C-POSIX, która osiąga oczekiwane 8-krotne przyspieszenie: co oznaczają „rzeczywisty”, „użytkownik” i „sys” w danych wyjściowych czasu (1)?

  DO ZROBIENIA: Nie wiem, dlaczego tak się dzieje, muszą pojawić się inne problemy z wydajnością Pythona.

Kod testowy:

#!/usr/bin/env python3

import multiprocessing
import threading
import time
import sys

def cpu_func(result, niters):
    '''
    A useless CPU bound function.
    '''
    for i in range(niters):
        result = (result * result * i + 2 * result * i * i + 3) % 10000000
    return result

class CpuThread(threading.Thread):
    def __init__(self, niters):
        super().__init__()
        self.niters = niters
        self.result = 1
    def run(self):
        self.result = cpu_func(self.result, self.niters)

class CpuProcess(multiprocessing.Process):
    def __init__(self, niters):
        super().__init__()
        self.niters = niters
        self.result = 1
    def run(self):
        self.result = cpu_func(self.result, self.niters)

class IoThread(threading.Thread):
    def __init__(self, sleep):
        super().__init__()
        self.sleep = sleep
        self.result = self.sleep
    def run(self):
        time.sleep(self.sleep)

class IoProcess(multiprocessing.Process):
    def __init__(self, sleep):
        super().__init__()
        self.sleep = sleep
        self.result = self.sleep
    def run(self):
        time.sleep(self.sleep)

if __name__ == '__main__':
    cpu_n_iters = int(sys.argv[1])
    sleep = 1
    cpu_count = multiprocessing.cpu_count()
    input_params = [
        (CpuThread, cpu_n_iters),
        (CpuProcess, cpu_n_iters),
        (IoThread, sleep),
        (IoProcess, sleep),
    ]
    header = ['nthreads']
    for thread_class, _ in input_params:
        header.append(thread_class.__name__)
    print(' '.join(header))
    for nthreads in range(1, 2 * cpu_count):
        results = [nthreads]
        for thread_class, work_size in input_params:
            start_time = time.time()
            threads = []
            for i in range(nthreads):
                thread = thread_class(work_size)
                threads.append(thread)
                thread.start()
            for i, thread in enumerate(threads):
                thread.join()
            results.append(time.time() - start_time)
        print(' '.join('{:.6e}'.format(result) for result in results))

GitHub upstream + kreślenie kodu w tym samym katalogu .

Testowano na Ubuntu 18.10, Python 3.6.7, w laptopie Lenovo ThinkPad P51 z procesorem: procesor Intel Core i7-7820HQ (4 rdzenie / 8 wątków), pamięć RAM: 2x Samsung M471A2K43BB1-CRC (2x 16GiB), dysk SSD: Samsung MZVLB512HAJQ- 000L7 (3000 MB / s).

Wizualizuj, które wątki działają w danym momencie

Ten post https://rohanvarma.me/GIL/ nauczył mnie, że możesz uruchomić wywołanie zwrotne za każdym razem, gdy zaplanowany jest wątek z target=argumentemthreading.Thread i to samo for multiprocessing.Process.

To pozwala nam zobaczyć dokładnie, który wątek jest uruchamiany za każdym razem. Po wykonaniu tej czynności zobaczylibyśmy coś takiego (zrobiłem ten konkretny wykres):

            +--------------------------------------+
            + Active threads / processes           +
+-----------+--------------------------------------+
|Thread   1 |********     ************             |
|         2 |        *****            *************|
+-----------+--------------------------------------+
|Process  1 |***  ************** ******  ****      |
|         2 |** **** ****** ** ********* **********|
+-----------+--------------------------------------+
            + Time -->                             +
            +--------------------------------------+

co by pokazało, że:

• wątki są w pełni serializowane przez GIL
• procesy mogą przebiegać równolegle

Oto kilka danych dotyczących wydajności dla języka Python 2.6.x, które podważają pogląd, że wątkowanie jest bardziej wydajne niż przetwarzanie wieloprocesowe w scenariuszach związanych z IO. Te wyniki pochodzą z 40-procesorowego IBM System x3650 M4 BD.

Przetwarzanie powiązane we / wy: pula procesów działa lepiej niż pula wątków

>>> do_work(50, 300, 'thread','fileio')
do_work function took 455.752 ms

>>> do_work(50, 300, 'process','fileio')
do_work function took 319.279 ms

Przetwarzanie związane z procesorem: Pula procesów działa lepiej niż pula wątków

>>> do_work(50, 2000, 'thread','square')
do_work function took 338.309 ms

>>> do_work(50, 2000, 'process','square')
do_work function took 287.488 ms

Nie są to rygorystyczne testy, ale mówią mi, że przetwarzanie wieloprocesowe nie jest całkowicie nieefektywne w porównaniu z wątkami.

Kod używany w interaktywnej konsoli Pythona do powyższych testów

from multiprocessing import Pool
from multiprocessing.pool import ThreadPool
import time
import sys
import os
from glob import glob

text_for_test = str(range(1,100000))

def fileio(i):
 try :
  os.remove(glob('./test/test-*'))
 except : 
  pass
 f=open('./test/test-'+str(i),'a')
 f.write(text_for_test)
 f.close()
 f=open('./test/test-'+str(i),'r')
 text = f.read()
 f.close()


def square(i):
 return i*i

def timing(f):
 def wrap(*args):
  time1 = time.time()
  ret = f(*args)
  time2 = time.time()
  print '%s function took %0.3f ms' % (f.func_name, (time2-time1)*1000.0)
  return ret
 return wrap

result = None

@timing
def do_work(process_count, items, process_type, method) :
 pool = None
 if process_type == 'process' :
  pool = Pool(processes=process_count)
 else :
  pool = ThreadPool(processes=process_count)
 if method == 'square' : 
  multiple_results = [pool.apply_async(square,(a,)) for a in range(1,items)]
  result = [res.get()  for res in multiple_results]
 else :
  multiple_results = [pool.apply_async(fileio,(a,)) for a in range(1,items)]
  result = [res.get()  for res in multiple_results]


do_work(50, 300, 'thread','fileio')
do_work(50, 300, 'process','fileio')

do_work(50, 2000, 'thread','square')
do_work(50, 2000, 'process','square')

Cóż, na większość pytań odpowiada Giulio Franco. W dalszej części opiszę problem konsument-producent, który, jak sądzę, poprowadzi Cię na właściwą ścieżkę do rozwiązania problemu korzystania z aplikacji wielowątkowej.

fill_count = Semaphore(0) # items produced
empty_count = Semaphore(BUFFER_SIZE) # remaining space
buffer = Buffer()

def producer(fill_count, empty_count, buffer):
    while True:
        item = produceItem()
        empty_count.down();
        buffer.push(item)
        fill_count.up()

def consumer(fill_count, empty_count, buffer):
    while True:
        fill_count.down()
        item = buffer.pop()
        empty_count.up()
        consume_item(item)

Możesz przeczytać więcej na temat prymitywów synchronizacji z:

 http://linux.die.net/man/7/sem_overview
 http://docs.python.org/2/library/threading.html

Pseudokod jest powyżej. Przypuszczam, że powinieneś przeszukać problem producent-konsument, aby uzyskać więcej referencji.