os.sched_getaffinity (0) vs os.cpu_count ()
Tak więc znam różnicę między dwiema metodami w tytule, ale nie znam ich praktycznych implikacji.
Z tego, co rozumiem: jeśli używasz więcej NUM_WORKERS niż faktycznie dostępnych rdzeni, napotykasz duże spadki wydajności, ponieważ Twój system operacyjny nieustannie przełącza się w tę iz powrotem, próbując zachować równoległość. Nie wiem, czy to prawda, ale przeczytałem to tutaj na SO gdzieś od kogoś mądrzejszego ode mnie.
A w dokumentach os.cpu_count()mówi:
Zwróć liczbę procesorów w systemie. Zwraca wartość None jeśli nieokreślona. Ta liczba nie jest równa liczbie procesorów, z których może korzystać bieżący proces. Liczbę użytecznych procesorów można uzyskać za pomocą len (os.sched_getaffinity (0))
Próbuję więc ustalić, do czego odnosi się „system”, jeśli proces może wykorzystywać więcej procesorów niż w „systemie”.
Chcę tylko bezpiecznie i efektywnie wdrożyć multiprocessing.poolfunkcjonalność. Oto podsumowanie mojego pytania:
Jakie są praktyczne konsekwencje:
NUM_WORKERS = os.cpu_count() - 1
# vs.
NUM_WORKERS = len(os.sched_getaffinity(0)) - 1
Dzieje się -1tak, ponieważ stwierdziłem, że mój system jest znacznie mniej opóźniony, jeśli próbuję pracować podczas przetwarzania danych.
Odpowiedzi
Gdybyś miał zadania, które były w 100% związane z procesorem, tj. Wykonywały tylko obliczenia, to oczywiście nic nie dałoby / nie dałoby się zyskać mając rozmiar puli procesów większy niż liczba procesorów dostępnych na twoim komputerze. Ale co by było, gdyby wrzucono mieszankę I / O, przez co proces zwolniłby procesor, czekając na zakończenie operacji we / wy (lub, na przykład, adres URL, który ma zostać zwrócony ze strony internetowej, co zajmuje stosunkowo dużo czasu) ? Nie jest dla mnie jasne, czy w tym scenariuszu nie można osiągnąć lepszej przepustowości przy rozmiarze puli procesów przekraczającym os.cpu_count().
Aktualizacja
Oto kod demonstrujący ten punkt. Ten kod, który prawdopodobnie najlepiej byłby obsługiwany przy użyciu wątków, używa procesów. Na swoim pulpicie mam 8 rdzeni. Program po prostu pobiera 54 adresy URL jednocześnie (lub równolegle w tym przypadku). Do programu przekazywany jest argument określający rozmiar puli do wykorzystania. Niestety, początkowe obciążenie wiąże się tylko z utworzeniem dodatkowych procesów, więc oszczędności zaczynają spadać, jeśli utworzysz zbyt wiele procesów. Ale jeśli zadanie trwało długo i miało dużo operacji we / wy, to koszt tworzenia procesów byłby tego wart:
from concurrent.futures import ProcessPoolExecutor, as_completed
import requests
from timing import time_it
def get_url(url):
resp = requests.get(url, headers={'user-agent': 'my-app/0.0.1'})
return resp.text
@time_it
def main(poolsize):
urls = [
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
'https://ibm.com',
'https://microsoft.com',
'https://google.com',
]
with ProcessPoolExecutor(poolsize) as executor:
futures = {executor.submit(get_url, url): url for url in urls}
for future in as_completed(futures):
text = future.result()
url = futures[future]
print(url, text[0:80])
print('-' * 100)
if __name__ == '__main__':
import sys
main(int(sys.argv[1]))
8 procesów: (liczba posiadanych rdzeni):
func: main args: [(8,), {}] took: 2.316840410232544 sec.
16 procesów:
func: main args: [(16,), {}] took: 1.7964842319488525 sec.
24 procesy:
func: main args: [(24,), {}] took: 2.2560818195343018 sec.
Te dwie funkcje są bardzo różne i NUM_WORKERS = os.sched_getaffinity(0) - 1po prostu natychmiast zawiodą, TypeErrorponieważ próbujesz odjąć liczbę całkowitą od zbioru. Podczas gdy os.cpu_count()mówi ci, ile rdzeni ma system, os.sched_getaffinity(pid)mówi ci, na których rdzeniach może działać określony wątek / proces .
os.cpu_count()
os.cpu_count()pokazuje liczbę dostępnych rdzeni znaną z systemu operacyjnego ( rdzenie wirtualne ). Najprawdopodobniej masz o połowę mniejszą liczbę rdzeni fizycznych . To, czy ma sens używanie większej liczby procesów niż rdzeni fizycznych, a nawet więcej niż rdzeni wirtualnych, zależy w dużej mierze od tego, co robisz. Im ściślejsza pętla obliczeniowa (mała różnorodność instrukcji, kilka chybień w pamięci podręcznej, ...), tym większe prawdopodobieństwo, że nie skorzystasz z większej liczby używanych rdzeni (przez użycie większej liczby procesów roboczych), a nawet nie doświadczysz spadku wydajności.
Oczywiście zależy to również od tego, co jeszcze działa w twoim systemie, ponieważ twój system próbuje nadać każdemu wątkowi (jako rzeczywistej jednostce wykonawczej procesu) w systemie sprawiedliwy udział czasu wykonywania na dostępnych rdzeniach. Dlatego nie ma możliwości uogólnienia liczby pracowników, których należy użyć. Ale jeśli na przykład masz ścisłą pętlę, a system pracuje na biegu jałowym, dobrym punktem wyjścia do optymalizacji jest
os.cpu_count() // 2 # same as mp.cpu_count() // 2
... i rośnie od tego momentu.
Jak już wspomniał @Frank Yellin, domyślnie multiprocessing.Poolużywa os.cpu_count()liczby pracowników.
os.sched_getaffinity(pid)
os.sched_getaffinity(pid)
Zwraca zestaw procesorów, do których ograniczony jest proces z PID pid (lub bieżący proces, jeśli zero).
Teraz core / cpu / procesor / -affinity dotyczy tego, na których konkretnych (wirtualnych) rdzeniach może działać twój wątek (w ramach procesu roboczego) . Twój system operacyjny nadaje każdemu rdzeniu identyfikator, od 0 do (liczba rdzeni - 1), a zmiana powinowactwa pozwala ograniczyć („przypinanie”), na którym rdzeniu (rdzeniach) może w ogóle działać określony wątek.
Przynajmniej w Linuksie stwierdziłem, że oznacza to, że jeśli żaden z dozwolonych rdzeni nie jest obecnie dostępny, wątek procesu potomnego nie zostanie uruchomiony, nawet jeśli inne, niedozwolone rdzenie byłyby bezczynne. Więc „powinowactwo” jest tutaj trochę mylące.
Celem majstrowania przy koligacji jest zminimalizowanie unieważnień pamięci podręcznej z przełączników kontekstu i migracji rdzenia. Twój system operacyjny ma zazwyczaj lepszy wgląd i już stara się utrzymywać pamięć podręczną „na gorąco” dzięki polityce planowania, więc jeśli nie wiesz, co robisz, nie możesz oczekiwać łatwych korzyści z ingerencji.
Domyślnie koligacja jest ustawiona na wszystkie rdzenie, a zmiana multiprocessing.Pooltego nie ma większego sensu, przynajmniej jeśli twój system jest bezczynny w inny sposób.
Zauważ, że pomimo faktu, że w tutejszych dokumentach mówi się o „procesie”, ustawienie powinowactwa jest naprawdę kwestią związaną z wątkiem. Na przykład ustawienie powinowactwa w wątku „podrzędnym” dla „bieżącego procesu, jeśli zero”, nie zmienia powinowactwa wątku głównego ani innych wątków w procesie. Ale wątki potomne dziedziczą swoje powinowactwo z wątku głównego, a procesy potomne (poprzez wątek główny) dziedziczą powinowactwo z głównego wątku procesu nadrzędnego. Ma to wpływ na wszystkie możliwe metody startowe ("spawn", "fork", "forkserver"). Poniższy przykład ilustruje to i jak zmodyfikować koligację przy użyciu multiprocessing.Pool.
import multiprocessing as mp
import threading
import os
def _location():
return f"{mp.current_process().name} {threading.current_thread().name}"
def thread_foo():
print(f"{_location()}, affinity before change: {os.sched_getaffinity(0)}")
os.sched_setaffinity(0, {4})
print(f"{_location()}, affinity after change: {os.sched_getaffinity(0)}")
def foo(_, iterations=200e6):
print(f"{_location()}, affinity before thread_foo:"
f" {os.sched_getaffinity(0)}")
for _ in range(int(iterations)): # some dummy computation
pass
t = threading.Thread(target=thread_foo)
t.start()
t.join()
print(f"{_location()}, affinity before exit is unchanged: "
f"{os.sched_getaffinity(0)}")
return _
if __name__ == '__main__':
mp.set_start_method("spawn") # alternatives on Unix: "fork", "forkserver"
# for current process, exclude cores 0,1 from affinity-mask
print(f"parent affinity before change: {os.sched_getaffinity(0)}")
excluded_cores = {0, 1}
os.sched_setaffinity(0, os.sched_getaffinity(0).difference(excluded_cores))
print(f"parent affinity after change: {os.sched_getaffinity(0)}")
with mp.Pool(2) as pool:
pool.map(foo, range(5))
Wynik:
parent affinity before change: {0, 1, 2, 3, 4, 5, 6, 7}
parent affinity after change: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-1 MainThread, affinity before thread_foo: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 MainThread, affinity before thread_foo: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-1 Thread-1, affinity before change: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-1 Thread-1, affinity after change: {4}
SpawnPoolWorker-1 MainThread, affinity before exit is unchanged: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-1 MainThread, affinity before thread_foo: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 Thread-1, affinity before change: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 Thread-1, affinity after change: {4}
SpawnPoolWorker-2 MainThread, affinity before exit is unchanged: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 MainThread, affinity before thread_foo: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 Thread-2, affinity before change: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 Thread-2, affinity after change: {4}
SpawnPoolWorker-2 MainThread, affinity before exit is unchanged: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 MainThread, affinity before thread_foo: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-1 Thread-2, affinity before change: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-1 Thread-2, affinity after change: {4}
SpawnPoolWorker-1 MainThread, affinity before exit is unchanged: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 Thread-3, affinity before change: {2, 3, 4, 5, 6, 7}
SpawnPoolWorker-2 Thread-3, affinity after change: {4}
SpawnPoolWorker-2 MainThread, affinity before exit is unchanged: {2, 3, 4, 5, 6, 7}
Wdrożenie wykorzystania multiprocessing.pool
if processes is None:
processes = os.cpu_count() or 1
Nie jestem pewien, czy to odpowiada na twoje pytanie, ale przynajmniej jest to punkt danych.