Ачааллж байна...
Өмнөх хичээлд бид threading.Thread-ийг гараар үүсгэж, start ба join-оор удирдаж, таван хүлээлтийг нэг секундэд багтаасан. Тэр урьдчилсан харц ажилласан, гэвч бодит ажилд гурван асуудал бий. Нэгдүгээрт, thread-ийн функц юу буцаасныг авах арга байхгүй: Thread объект үр дүнг хадгалдаггүй. Хоёрдугаарт, thread дотор алдаа гарвал тэр алдаа үндсэн програмд хүрэхгүй, зөвхөн терминалд хэвлэгдээд алга болно. Гуравдугаарт, мянган хуудас татах бол мянган thread үүсгэх нь хэтэрхий: thread бүр санах ой эзэлдэг, үйлдлийн систем хэт олон thread-ийг удирдахад хүндэрнэ.
Стандарт сангийн concurrent.futures модуль энэ гурвыг бүгдийг шийддэг: ThreadPoolExecutor. “Pool” бол "цөөрөм", тогтмол тооны thread-ийн бүлэг. “Executor” бол "гүйцэтгэгч". Та ажлуудыг executor-т өгнө, тэр тэдгээрийг бэлэн thread-үүдэд хуваарилж, үр дүн ба алдааг танд буцаана. Энэ хичээлийн эхний хагаст бид executor-ийн хоёр хэлбэр, map ба submit-ийг эзэмшиж, хэдэн thread хамгийн ашигтайг хэмжиж, алдааг Бүлэг 11-ийн загвараар зөв цуглуулна.
Хичээлийн хоёр дахь хагас өмнөх хичээлийн сануулсан аюулд зориулагдсан: race condition. Бид түүнийг санаатайгаар үүсгэж, үр дүн ажиллуулах бүрт өөр байхыг харж, Бүлэг 6-ын хэрэгсэл болох context manager хэлбэрийн Lock-оор засна. Эцэст нь түгжээнээс илүү сайн шийдлийг харна: thread-үүд төлөв хуваалцахгүй байх.
Хэвлэх үйлчилгээний газрыг төсөөлөөд үзээрэй. Үйлчлүүлэгчид ирж захиалга өгнө. Газар тогтмол таван ажилтантай. Захиалга бүрийг хүлээн авах цэг дээр бүртгэж, баримт өгнө: "таны захиалгын дугаар 17". Чөлөөтэй ажилтан дараалсан захиалгыг авч гүйцэтгэнэ. Үйлчлүүлэгч дараа нь баримтаа үзүүлж бэлэн бүтээгдэхүүнээ авна. Хэрэв хэвлэх явцад принтер гацсан бол баримт дээр "амжилтгүй, шалтгаан: цаас гацсан" гэж тэмдэглэнэ.
Ажилтан бүрийг захиалга бүрт шинээр ажилд авахгүй: тав нь байнга бэлэн, захиалгуудыг ээлжлэн авна. Энэ бол pool. Захиалгын баримт бол Future буюу "ирээдүй": ажил хараахан дуусаагүй байж болох, гэхдээ дараа нь үр дүн эсвэл алдааг авах эрхийн бичиг. Python-ий ThreadPoolExecutor яг ийм ажилладаг.
Одоо хоёр дахь дүр зураг. Газрын нийт орлогыг ганц дэвтэрт бичдэг. Хоёр ажилтан нэг зэрэг дэвтэр рүү очиж, хоёулаа "одоогийн дүн: 100,000" гэж уншаад, тус бүр өөрийн 5,000-ыг нэмээд “105,000” гэж бичнэ. Хоёр захиалга орсон, гэвч дэвтэрт ганцынх л тусгагдсан. Нэг орлого алга болсон. Хэн ч буруу зүйл хийгээгүй: хоёулаа зөв уншиж, зөв нэмж, зөв бичсэн. Асуудал нь уншиж бичих хооронд нөгөө нь орж ирсэнд. Шийдэл нь дэвтрийн хажууд түлхүүр тавих: дэвтэрт бичихийн тулд түлхүүрийг авах ёстой, нэг удаад нэг хүн л авна. Энэ бол Lock.
Өмнөх хичээлийн fake_download-ийг ашиглана:
python
import time
from concurrent.futures import ThreadPoolExecutor
def fake_download(name, seconds=1.0):
time.sleep(seconds)
return f"{name}: татагдлаа"
names = [f"хуудас-{i}" for i in range(5)]
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as executor:
results = list(executor.map(fake_download, names))
print(f"{time.perf_counter() - start:.2f} с")
for line in results:
print(line)1.00 с
хуудас-0: татагдлаа
хуудас-1: татагдлаа
хуудас-2: татагдлаа
хуудас-3: татагдлаа
хуудас-4: татагдлааЭнэ хэдэн мөр өмнөх хичээлийн in_threads-ийн бүх ажлыг хийж, түүнээс илүүг өгнө. ThreadPoolExecutor(max_workers=5) нь таван thread-тэй pool үүсгэнэ. executor.map нь Хичээл 3.1-ийн map-тай ижил хэлбэртэй: функцийг жагсаалтын элемент бүрд хэрэглэнэ, гэхдээ дуудлагууд thread-үүд дээр зэрэг явна. Хамгийн чухал нь: үр дүн оролтын дарааллаар буцна, аль thread эхэлж дууссанаас үл хамааран. results жагсаалт names-тэй яг ижил дараалалтай.
with блок бол Бүлэг 6-ын context manager: блокийн төгсгөлд executor бүх ажил дуусахыг хүлээж, thread-үүдийг цэвэрлэнэ. Өмнөх хичээлийн join-ийг гараар бичих шаардлагагүй, мартах боломжгүй. Мөн executor.map нь Бүлэг 2-ын адил iterator буцаадаг: list() дуудсанаар бүх үр дүнг цуглуулсан.
max_workers-ийг өгөхгүй бол Python өөрөө сонгоно: процессорын цөмийн тоо дээр дөрвийг нэмж, 32-оос хэтрүүлэхгүй. Энэ дүрэм I/O-bound ажилд ерөнхий тохиромжтой эхлэл. Гэвч Бүлэг 12-ын хууль энд ч хүчинтэй: хэмжье. Хорин хуудас, тус бүр нэг секунд:
python
def download_all(names, workers):
with ThreadPoolExecutor(max_workers=workers) as executor:
return list(executor.map(fake_download, names))
names = [f"хуудас-{i}" for i in range(20)]
for workers in (1, 5, 10, 20, 50):
start = time.perf_counter()
download_all(names, workers)
print(f"{workers:>2} thread: {time.perf_counter() - start:5.2f} с") 1 thread: 20.02 с
5 thread: 4.01 с
10 thread: 2.01 с
20 thread: 1.00 с
50 thread: 1.00 сХэв маяг тодорхой. Нэг thread бол дараалсан програм: 20 секунд. Thread-ийн тоог нэмэх бүрт хугацаа пропорциональ буурна: 5 thread дөрвөн "давалгаа", 10 thread хоёр. 20 thread бүх хүлээлтийг нэг дор давхцуулна. 50 thread юу ч нэмээгүй: 20 ажилд 20-оос илүү ажилтан хэрэггүй, илүү 30 нь сул сууна.
Бодит ажилд хоёр хязгаар нэмэгдэнэ. Нэгдүгээрт, гадны сервер: нэг вэб сайт руу зэрэг 50 хүсэлт илгээх нь эелдэг биш бөгөөд олон сервер ийм үед хүсэлтээ татгалзаж, хариугаа “429 Too Many Requests”, "хэт олон хүсэлт" гэсэн кодоор буцаадаг. Хоёрдугаарт, таны сүлжээний хурд: thread нэмэх нь хүлээлтийг давхцуулна, харин сүлжээний сувгийг өргөсгөхгүй. Тиймээс бодит тохиолдолд эхлээд 5-10 thread-ээр эхэлж, хэмжээд, шаардлагатай бол нэмэгдүүлэх нь зөв арга.
map бүх ажил ижил функц, үр дүн оролтын дарааллаар хэрэгтэй үед тохиромжтой. Хэрэв ажлууд өөр өөр хугацаатай бөгөөд үр дүнг дуусах дарааллаар нь боловсруулахыг хүсвэл submit хэрэглэнэ:
python
from concurrent.futures import ThreadPoolExecutor, as_completed
jobs = {"том-файл": 3.0, "жижиг-файл": 1.0, "дунд-файл": 2.0}
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=3) as executor:
futures = {executor.submit(fake_download, name, seconds): name
for name, seconds in jobs.items()}
for future in as_completed(futures):
elapsed = time.perf_counter() - start
print(f"{elapsed:.1f} с: {future.result()}")1.0 с: жижиг-файл: татагдлаа
2.0 с: дунд-файл: татагдлаа
3.0 с: том-файл: татагдлааexecutor.submit(func, *args) нэг ажлыг илгээж, шууд Future буцаана: хэвлэх газрын баримт. Ажил thread дээр явж байх хооронд үндсэн програм цааш үргэлжилнэ. futures dict нь Future бүрийг ажлын нэртэй холбоно: үр дүн ирэхэд аль ажил байсныг мэдэх хэрэгтэй үед. as_completed(futures) нь Future-уудыг дуусах дарааллаар нь өгдөг iterator: жижиг файл эхэлж, том файл сүүлд. Илгээсэн дараалал нь "том, жижиг, дунд" байсныг анзаараарай. future.result() үр дүнг авна.
as_completed-ийн давуу тал: эхний үр дүн нэг секундэд л бэлэн, түүнийг шууд боловсруулж, хэрэглэгчид харуулж болно. map-ын хувьд эхний элемент, том файл, гурван секунд хүлээлгэх байсан.
Thread дотор гарсан алдаа юу болох вэ? future.result() дуудахад тэр алдаа дахин шидэгдэнэ, үндсэн thread-д. Энэ бол Future-ийн хамгийн чухал онцлог: алдаа алга болохгүй, баримт дээр тэмдэглэгдэж хадгалагдана. Нэг хуудас бүтэлгүйтдэг хувилбар:
python
def flaky_download(name):
time.sleep(1.0)
if name == "хуудас-2":
raise ConnectionError(f"{name}: холболт тасарлаа")
return f"{name}: татагдлаа"
names = [f"хуудас-{i}" for i in range(5)]
done, failed = [], []
with ThreadPoolExecutor(max_workers=5) as executor:
futures = {executor.submit(flaky_download, name): name for name in names}
for future in as_completed(futures):
try:
done.append(future.result())
except ConnectionError as error:
failed.append((futures[future], error))
print(f"Амжилттай: {len(done)}, амжилтгүй: {len(failed)}")
for name, error in failed:
print(f" {name}: {error}")Амжилттай: 4, амжилтгүй: 1
хуудас-2: хуудас-2: холболт тасарлааЭнэ бол Хичээл 12.4-ийн импортлогчийн загвар зэрэгцээ хэлбэрээр: нэг ажлын алдаа бусдыг зогсоохгүй, алдаанууд нэр ба шалтгаантайгаа цуглуулагдана. Хичээл 11.1-ийн дүрмийн дагуу бид зөвхөн юу хийхээ мэддэг ConnectionError-ийг барьсан: хэрэв flaky_download-д кодын алдаа, жишээлбэл NameError, байсан бол тэр үндсэн програмд шууд хүрч, нуугдахгүй.
executor.map-д алдаа өөрөөр ажиллана: үр дүнг давтаж байхад алдаатай элементэд хүрэхэд алдаа шидэгдэж, давталт зогсоно. Тэр элементээс хойших үр дүнгүүдийг авах боломжгүй болно. Тиймээс алдаа гарч болзошгүй ажилд submit ба as_completed илүү тохиромжтой.
Одоо аюул руу. Хичээл 4.2-ын түрийвчийг thread-тэй хувилбараар бичье:
python
import time
from concurrent.futures import ThreadPoolExecutor
class Account:
def __init__(self):
self.balance = 0
def deposit(self, amount):
current = self.balance
time.sleep(0)
self.balance = current + amount
account = Account()
def make_deposits(count):
for _ in range(count):
account.deposit(1)
with ThreadPoolExecutor(max_workers=10) as executor:
for _ in range(10):
executor.submit(make_deposits, 1_000)
print("Хүлээсэн: 10,000")
print(f"Бодит: {account.balance:,}")Хоёр удаагийн ажиллуулалт:
Хүлээсэн: 10,000
Бодит: 1,062Хүлээсэн: 10,000
Бодит: 1,019Арван мянган орлого орох ёстой байсан, мянга орчим л тусгагдсан. Хоёр ажиллуулалт өөр үр дүн өгсөн: ижил код, ижил өгөгдөл.
deposit-ийг зүйрлэлийн дэвтэртэй харьцуул. current = self.balance бол дэвтрээс унших. self.balance = current + amount бол дэвтэрт бичих. Хооронд нь time.sleep(0) байна: тэг секунд унтах. Тэр юу ч хүлээхгүй, гэвч өмнөх хичээлээс санаарай: thread унтах үед GIL-ийг суллана. Тиймээс энэ мөр "бусад thread-д ээлж өгөх" дохио. Trace хийе:
Алхам 1. Thread A balance-ийг уншина: current = 500. sleep(0)-д GIL-ийг суллана.
Алхам 2. Thread B ээлж авч, balance-ийг уншина: current = 500, учир нь A хараахан бичээгүй. B мөн sleep(0)-д ээлж өгнө.
Алхам 3. A буцаж ирж бичнэ: balance = 501.
Алхам 4. B буцаж ирж бичнэ: balance = 501. A-ийн орлого дарагдаж алга болсон.
Арван thread бүр ээлжлэн ингэж байгаа тул ихэнх орлого алдагдсан.
Шударга тэмдэглэл: sleep(0)-ийг бид алдааг найдвартай харуулахын тулд санаатайгаар оруулсан. Хэрэв түүнийг арилгаад self.balance += amount болговол стандарт Python-ий орчин үеийн хувилбаруудад үр дүн ихэвчлэн зөв 10,000 гарна: тайлбарлагч thread-үүдийг ийм богино кодын дунд ховор сольдог. Гэвч "ихэвчлэн" бол баталгаа биш. Бодит кодонд уншиж бичих хооронд log бичих, мэдээллийн сангаас хайх, функц дуудах бүх зүйл ээлж өгөх боломж. GIL-гүй free-threaded Python дээр энгийн += ч алдаа гаргана. Энэ алдааны хамгийн аюултай шинж нь яг энэ: тест дээр зөв ажиллаж, production-д сард нэг удаа буруу тоо гаргадаг.
threading.Lock бол дэвтрийн түлхүүр. Бүлэг 6-ын context manager: with блок руу орохдоо түлхүүрийг авч, гарахдаа, алдаатай ч, буцааж өгнө:
python
import threading
class Account:
def __init__(self):
self.balance = 0
self._lock = threading.Lock()
def deposit(self, amount):
with self._lock:
current = self.balance
time.sleep(0)
self.balance = current + amountХүлээсэн: 10,000
Бодит: 10,000Хэдэн удаа ажиллуулсан ч 10,000. Thread A түлхүүрийг авсан бол Thread B with мөрөнд хүлээнэ: A уншиж, бичиж, түлхүүрийг өгөх хүртэл. Уншиж бичих хос хуваагдашгүй болсон, Хичээл 11.3-ын atomic write-ийн адил санаа. Lock-ийг __init__-д нэг удаа үүсгэж, объектын бүх дуудлага ижил түлхүүрийг хуваалцаж байгааг анзаараарай. Түгээмэл алдаанууд хэсэгт эсрэг тохиолдлыг харна.
Lock үнэгүй биш. Түгжээн доторх код нэг агшинд нэг thread-ээр л ажиллана: тэр хэсэгт зэрэгцээ ажиллагаа байхгүй. Хэрэв түгжээн дотор удаан ажил хийвэл бүх thread-ийн давуу тал алга болно. Хэмжье:
python
lock = threading.Lock()
def download_locked(name):
with lock:
return fake_download(name)
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as executor:
list(executor.map(download_locked, [f"хуудас-{i}" for i in range(5)]))
print(f"{time.perf_counter() - start:.2f} с")5.00 сТаван thread, таван секунд: дараалсан програмтай ижил. Түгжээ хүлээлтүүдийг дараалалд оруулсан. Дүрэм: түгжээн доторх кодыг аль болох богино байлга, зөвхөн хуваалцсан төлөвийг унших ба бичих. Хүлээлт, тооцоолол, I/O түгжээнээс гадна.
Race condition зөвхөн thread-үүд нэг өөрчлөгддөг объектыг хуваалцах үед гарна. Хэрэв thread бүр өөрийн үр дүнг буцааж, нэгтгэх ажлыг үндсэн thread хийвэл түгжээ огт хэрэггүй:
python
def count_deposits(count):
total = 0
for _ in range(count):
total += 1
return total
with ThreadPoolExecutor(max_workers=10) as executor:
balance = sum(executor.map(count_deposits, [1_000] * 10))
print(f"Бодит: {balance:,}")Бодит: 10,000total бол функцийн дотоод хувьсагч: thread бүр өөрийн хувийг эзэмшинэ, хэн ч хуваалцахгүй. sum үр дүнгүүдийг үндсэн thread дээр нэгтгэнэ. Энэ загвар Бүлэг 3-ын functional хэв маягтай ижил: төлөв өөрчлөхийн оронд утга буцаах. Боломжтой бүх газарт энэ загварыг сонго, Lock-ийг хуваалцах зайлшгүй үед л хэрэглэ.
Хүсвэл Хичээл 5.3-ын retry decorator-ийг flaky_download дээр хэрэглэж, бүтэлгүйтсэн хуудсыг дахин оролдох executor бүтээж үзээрэй. Сонирхвол deposit-ийн Lock-тэй хувилбарыг дахин хэмжиж, Lock-гүй хувилбартай хурдыг харьцуулж болно: Lock хэр үнэтэй вэ? Мөн race condition-ийг өөр хэлбэрээр үүсгэх нь сургамжтай: хоёр thread нэг list-д "байхгүй бол нэм" гэсэн шалгалт хийж, давхардсан элементүүд гарч байгааг харах. Эдгээр бүгд санал бөгөөд дараагийн хичээлүүд үүнийг хийсэн гэж үзэхгүй.
RuntimeError: cannot schedule new futures after shutdown. "Унтарсны дараа шинэ ажил төлөвлөх боломжгүй" гэсэн утгатай. with блок дууссаны дараа executor.submit дуудахад гарна: блокийн төгсгөлд executor хаагдсан. Бүх submit-ийг with блок дотор хий.
Алдаа чимээгүй алга болох. Мессежгүй. submit хийсэн ажил алдаа шидсэн ч future.result() хэзээ ч дуудагдаагүй бол алдаа Future дотор хадгалагдсан хэвээр, хэн ч мэдэхгүй. Програм амжилттай дууссан мэт харагдана. Future бүрийн result()-ийг дуудах, эсвэл map-ын үр дүнг бүрэн давтах нь алдааг гаргаж ирэх цорын ганц арга.
Програм хариугүй зогсох: давхар түгжээ. Мессежгүй, програм мөнхөд хүлээнэ. threading.Lock дахин орох боломжгүй: түлхүүрийг аль хэдийн барьж буй thread түүнийг дахин авах гэвэл өөрийгөө хүлээнэ. Жишээ нь deposit Lock дотроосоо өөр нэг Lock-той method дуудвал. Энэ төрлийн мөнхийн хүлээлтийг deadlock гэдэг. Терминалд Ctrl+C дарж зогсооно. Шийдэл: түгжээтэй method-ууд бие биеэ дуудахгүй байхаар зохион байгуулах. Хэрэв нэг thread нэг түгжээг давтан авах зайлшгүй бол threading.RLock буюу “reentrant”, дахин орох боломжтой түгжээ.
Түгжээг дуудлага бүрт шинээр үүсгэх. Мессежгүй. def deposit(self, amount): with threading.Lock(): ... гэж бичвэл дуудлага бүр өөрийн шинэ түгжээ авна, хэн ч түүнийг хүлээхгүй, хамгаалалт огт байхгүй. Зүйрлэлээр: ажилтан бүр өөрийн түлхүүртэй. Race condition хэвээр, гэвч код хамгаалагдсан мэт харагдана. Түгжээг хамгаалах өгөгдөлтэй нь хамт, __init__-д нэг удаа үүсгэ.
time.sleep(0)-гүй тестэд итгэх. Мессежгүй. Race condition-ий тест “10,000 гарлаа, асуудалгүй” гэж дүгнэх нь Бүлэг 12-ын "жижиг өгөгдөл дээр хэмжих" алдааны ах дүү. Алдаа гарахгүй байх нь алдаа байхгүйн нотолгоо биш. Хуваалцсан өөрчлөгддөг төлөв байгаа бол түүнийг Lock-оор хамгаал, эсвэл хуваалцахаа боль.
ThreadPoolExecutor тогтмол тооны thread-ийг удирдаж, үр дүн ба алдааг буцаадаг. with блок бүх ажил дуусахыг хүлээж, thread-үүдийг цэвэрлэнэ. executor.map функцийг элемент бүрд зэрэг хэрэглэж, үр дүнг оролтын дарааллаар өгнө. executor.submit Future буцаана: ирээдүйн үр дүн эсвэл алдааны баримт. as_completed Future-уудыг дуусах дарааллаар өгнө. future.result() thread дотор гарсан алдааг үндсэн thread-д дахин шиднэ, тиймээс алдааг Бүлэг 11-ийн загвараар цуглуулж болно. Thread-ийн тоог хэмжилтээр сонгоно: ажлын тооноос илүү thread ашиггүй, бодит серверүүд хэт олон хүсэлтийг татгалздаг. Race condition бол хуваалцсан төлөвийг унших ба бичих хооронд өөр thread орж ирэхэд өгөгдөл алдагдах. Үр дүн ажиллуулах бүрт өөр бөгөөд тест дээр нуугдаж болно. threading.Lock бол context manager: уншиж бичих хосыг хуваагдашгүй болгоно. Түгжээн доторх кодыг богино байлга, тэнд хүлээлт байвал зэрэгцээ ажиллагаа алга болно. Хамгийн сайн шийдэл бол төлөв хуваалцахгүй байх: thread бүр утга буцааж, үндсэн thread нэгтгэнэ.
Энэ хоёр хичээлд thread I/O-bound ажлыг гайхалтай хурдасгасан, CPU-bound ажлыг огт хурдасгаагүй. Шалтгаан нь GIL гэдгийг бид нэрлэсэн боловч тайлбарлаагүй. Дараагийн хичээл энэ бүлгийг дуусгана: GIL яагаад бий, Python-ий шинэ хувилбаруудад юу өөрчлөгдөж байна, мөн CPU-bound ажлыг жинхэнэ parallelism-оор хурдасгах стандарт хэрэгсэл ProcessPoolExecutor. Гайхалтай нь энэ хичээлийн бүх хэв маяг, map, submit, as_completed, Future, түүнд яг ижлээр ажиллана. Executor-ийн нэг үгийг солиход л хангалттай.
Бүртгэлтэй болсноор энэ сургалтын бүх хичээлд хандах эрх авна.