Нишки

Проблем

Три отчета, един по един

Програмата сваля три месечни отчета от сървъра. Тук свалянето е пауза: time.sleep(секунди) спира програмата за толкова секунди.

import time


def download(file, seconds):
    time.sleep(seconds)
    print(f"Свалих {file}")


download("март.pdf", 0.6)
download("април.pdf", 0.2)
download("май.pdf", 0.4)
print("Готово")

Свалих март.pdf
Свалих април.pdf
Свалих май.pdf
Готово

Паузите са 0,6, 0,2 и 0,4 секунди, общо 1,2 секунди. През цялото време програмата само чака мрежата. Ако имаше бутон „Отказ“, той нямаше да реагира. А трите отчета не зависят един от друг и спокойно могат да се свалят едновременно.

Идеята

Процес и нишка

Когато пуснеш програмата, операционната система ѝ дава процес: своя памет, свои отворени файлове. Вътре в процеса работят една или повече нишки (threads). Нишката е поредица от команди, които се изпълняват една след друга. Досега всяка твоя програма имаше една нишка: тази, която изпълнява файла отгоре надолу.

Нишките в един процес делят паметта си: виждат едни и същи обекти. Това ги прави удобни (лесно си предават данни) и опасни (могат да се объркат взаимно, ще го видим след малко).

Паралелно значи наистина едновременно, на различни ядра на процесора.
Конкурентно значи на едно ядро, като се редуват толкова бързо, че изглежда едновременно.

За програмиста разликата рядко има значение: и в двата случая не знаеш коя нишка кога ще продължи.

В Python (CPython, този от python.org) има уловка: GIL (global interpreter lock). В даден момент само една нишка изпълнява Python код, дори процесорът да има много ядра. Затова нишките помагат, когато програмата чака (мрежа, файлове, sleep): докато една чака, друга работи. За тежки сметки нишките не ускоряват. Там се ползват процеси, ще ги видим в упражнението.

По-точно, за любопитните

Браузърът Chrome пуска отделен процес за всеки раздел. Затова един увиснал сайт не спира останалите, но и затова браузърът ползва много памет.
Нишки ползва всяка програма, която прави няколко неща наведнъж: играта пуска звука в една нишка, а логиката в друга; уеб сървърът обслужва всеки клиент в своя нишка.
Цената: всяка нишка има свой стек и паметта за нея се заделя отделно. Хиляди нишки струват. Затова в истински програми се ползва басейн от нишки (последната идея в урока). - Python 3.13 има и отделна версия без GIL (free-threaded). Тя още е експериментална, затова тук пишем за обикновения Python.

Виж го

Първата нишка

Класът Thread от модула threading е нишка. С target му казваш коя функция да изпълни, а с args с какви аргументи (кортеж):

worker = threading.Thread(target=download, args=("април.pdf", 0.2))
worker.start()
print("Програмата не чака")
worker.join()
print("Отчетът е свален")

Програмата не чака
Свалих април.pdf
Отчетът е свален

start() пуска новата нишка и се връща веднага. Програмата продължава, без да чака, затова „Програмата не чака“ се отпечатва първо.
join() кара програмата да изчака нишката да свърши.

download е функцията от първата карта. Python няма проверени изключения, затова тук няма какво да се обявява.

По-точно, за любопитните

target е най-краткият начин, но не единственият. Ще срещнеш и наследник на Thread, който пише свой run(), и ламбда:

class ReportThread(threading.Thread):
    def run(self):
        print("1. наследник на Thread")


def report_job():
    print("2. функция като target")


a = ReportThread()
b = threading.Thread(target=report_job)
c = threading.Thread(target=lambda: print("3. ламбда"))
for t in [a, b, c]:
    t.start()
    t.join()

1. наследник на Thread
2. функция като target
3. ламбда

Анонимни класове в Python няма. По-гъвкаво е да отделиш задачата (функцията) от нишката: така същата функция може да се даде на басейн от нишки.

Виж го

Три наведнъж

Сега трите отчета, всеки в своя нишка:

start = time.time()
march = threading.Thread(target=download, args=("март.pdf", 0.6))
april = threading.Thread(target=download, args=("април.pdf", 0.2))
may = threading.Thread(target=download, args=("май.pdf", 0.4))
march.start()
april.start()
may.start()
march.join()
april.join()
may.join()
seconds = time.time() - start
print(f"Готово за под секунда: {seconds < 1}")

Свалих април.pdf
Свалих май.pdf
Свалих март.pdf
Готово за под секунда: True

Нишките тръгват в реда март, април, май, но свършват по реда на паузите: април (0,2 s), май (0,4 s), март (0,6 s). Цялото чакане е колкото най-дългото, 0,6 секунди вместо 1,2.

Реда, в който нишките стигат до print, решава операционната система. Тук паузите се различават достатъчно, затова редът е един и същ. Ако бяха еднакви, при всяко пускане можеше да е различен.

Пробвай

start или run

Какво ще отпечата програмата? (name е името на нишката)

class Report(threading.Thread):
    def run(self):
        print(f"Работи в: {threading.current_thread().name}")


t = Report(name="сваляне")
t.run()
t.start()
t.join()

Отговор

Работи в: сваляне
Работи в: сваляне

Работи в: сваляне
Работи в: MainThread

Работи в: MainThread
Работи в: сваляне

t.run() е обикновено извикване на метод: кодът се изпълнява веднага, в главната нишка, която Python нарича MainThread. Нова нишка няма. Чак t.start() пуска нишката „сваляне“, която изпълнява същия run(). Да напишеш run() вместо start() е честа грешка: програмата работи, само че нищо не е едновременно.

Идеята

Животът на нишката

is_alive() казва дали нишката работи в момента: от start() до края на функцията ѝ:

t = threading.Thread(target=download, args=("май.pdf", 0.5))
print(t.is_alive())
t.start()
time.sleep(0.1)
print(t.is_alive())
t.join()
print(t.is_alive())
try:
    t.start()
except RuntimeError as e:
    print(f"RuntimeError: {e}")

False
True
Свалих май.pdf
False
RuntimeError: threads can only be started once

Python не казва повече от това: дали чака или смята, не се вижда. Нишката се пуска само веднъж. Втори start() на същата нишка хвърля RuntimeError.

По-точно, за любопитните

Нишка в Python не може да се спре отвън. Ако трябва да спре, тя сама проверява сигнал, например threading.Event, който друга нишка вдига с set().
name и native_id са име и номер на нишката. threading.current_thread() дава нишката, в която се изпълнява кодът.
time.sleep(0) подсказва, че нишката може да отстъпи на друга. Ще го ползваме след малко.

Проблем

Загубените посещения

Сайтът брои посещенията. Две нишки обслужват посетители и всяка вика increment() 10 000 пъти. Между прочитането и записването на брояча нишката прави нещо друго. Тук това е time.sleep(0): „пусни другите нишки за момент“:

class Counter:
    def __init__(self):
        self.count = 0

    def increment(self):
        value = self.count
        time.sleep(0)  # тук нишката отстъпва на другата
        self.count = value + 1


visits = Counter()


def work():
    for i in range(10_000):
        visits.increment()


first = threading.Thread(target=work)
second = threading.Thread(target=work)
first.start()
second.start()
first.join()
second.join()
print(f"Посещения: {visits.count}")

Очакваш 20 000. Пет пускания:

Посещения: 10000
Посещения: 10001
Посещения: 10001
Посещения: 10000
Посещения: 10000

Всеки път по-малко, почти наполовина. Причината е, че увеличаването не е една стъпка, а три:

value = self.count       # 1. прочети
value = value + 1        # 2. добави 1
self.count = value       # 3. запиши

class Counter:
    def __init__(self):
        self.count = 5
        self.lock = threading.Lock()

    def increment(self):
        self.count += 1  # прочети, добави 1, запиши

    def safe_increment(self):
        with self.lock:
            self.count += 1

Стъпка по стъпка, с думи

count е 5. Нишките A и B викат increment() едновременно.
A прочита count: 5.
B прочита count, преди A да е записала: също 5.
A добавя 1 в своето копие и записва 6.
B също записва 6. Две извиквания, а count е 6, не 7: едното добавяне е загубено.
Отначало, с with self.lock: A взима ключалката, прочита 5 и записва 6.
B стига до with self.lock и чака, докато A пусне ключалката.
B взима ключалката, прочита 6 и записва 7: нищо не е загубено.

Когато двете нишки прочетат една и съща стойност, преди някоя да е записала, едното добавяне се губи. Това се казва състезание (race condition). Дали ще стане, зависи от това кога се сменят нишките, затова резултатът е различен при всяко пускане.

По-точно, за любопитните

Без time.sleep(0), със self.count += 1 и 100 000 увеличения на нишка, пет пускания в Python 3.12 дадоха пет пъти 200 000:

class Counter:
    def __init__(self):
        self.count = 0

    def increment(self):
        self.count += 1

Посещения: 200000
Посещения: 200000
Посещения: 200000
Посещения: 200000
Посещения: 200000

Това не е защото е безопасно. GIL сменя нишките само на определени места в изпълнението, и тук случайно не е между прочитането и записването. Python не обещава това: друга версия, друг код в метода или Python без GIL губят посещения. Затова общите данни се пазят с ключалка винаги.

Идеята

Lock: по една

Решението е тези три стъпки да ги прави само една нишка наведнъж. Това прави ключалката threading.Lock: блокът with self.lock: е само за една нишка в даден момент:

class Counter:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            value = self.count
            time.sleep(0)
            self.count = value + 1

Посещения: 20000

Нишката взима ключалката, когато влезе в with, и я пуска, когато излезе, дори при изключение. Другата нишка чака пред with. Сега резултатът е 20 000 при всяко пускане. Готов атомарен брояч, като в Java, в Python няма: ключалката е начинът.

По-точно, за любопитните

Condition: wait() и notify_all(): нишка в with блок на Condition може да пусне ключалката и да заспи, докато друга не я събуди. Така кухнята чака поръчка, а касата чака кухнята да вземе предишната (задачата „производител и потребител“):

class OrderBox:
    def __init__(self):
        self.order = None
        self.changed = threading.Condition()

    def put(self, new_order):
        with self.changed:
            while self.order is not None:
                self.changed.wait()
            self.order = new_order
            self.changed.notify_all()

    def take(self):
        with self.changed:
            while self.order is None:
                self.changed.wait()
            result = self.order
            self.order = None
            self.changed.notify_all()
            return result


box = OrderBox()


def kitchen():
    for i in range(3):
        print(f"Кухнята прие {box.take()}")


cook = threading.Thread(target=kitchen)
cook.start()
box.put("пица")
box.put("салата")
box.put("супа")
cook.join()

Кухнята прие пица
Кухнята прие салата
Кухнята прие супа

wait() е в while, а не в if: нишката може да се събуди и без причина, затова проверява условието наново. Същото готово прави queue.Queue(maxsize=1): put чака, ако опашката е пълна, а get чака, ако е празна.

Мъртва хватка (deadlock): две нишки чакат една друга завинаги. Офисът държи принтера и чака скенера, а рецепцията държи скенера и чака принтера:

printer = threading.Lock()
scanner = threading.Lock()


def office():
    with printer:
        time.sleep(0.1)
        with scanner:
            print("Офисът копира")


def reception():
    with scanner:
        time.sleep(0.1)
        with printer:
            print("Рецепцията копира")

Програмата увисва и никога не печата нищо. Начинът да го избегнеш е всички нишки да взимат ключалките в един и същ ред.

Приоритети нишките в Python нямат. Има фонови нишки (daemon=True): програмата свършва, без да ги чака:

report = threading.Thread(target=download, args=("годишен.pdf", 0.1), daemon=True)
print(report.name, report.daemon)

Thread-1 (download) True

Пробвай

Два банкомата

Двама души теглят по 80 € от една сметка със 100 € в един и същ момент. Понякога и двамата получават парите, а балансът става -60 €. На кой ред е грешката?

Избери реда с грешката.

Проверката self.balance >= amount и тегленето не са едно цяло. Ако нишката се смени между тях, и двете виждат 100 и двете теглят. С GIL това става рядко, затова грешката е коварна: минава хиляда пъти и пада при клиента. Поправката е ключалка около двата реда:

class Account:
    def __init__(self):
        self.balance = 100
        self.lock = threading.Lock()

    def withdraw(self, amount):
        with self.lock:
            if self.balance >= amount:
                self.balance -= amount

Тогава вторият банкомат чака първия, вижда 20 € и отказва. Балансът остава 20 €.

Идеята

Басейн от нишки и Future

Да правиш нишка за всяка задача е скъпо, а при хиляда задачи и опасно. Басейнът от нишки (thread pool) е няколко нишки, които си взимат задачи от опашка. В Python той е ThreadPoolExecutor (от concurrent.futures):

def size_of(file, seconds):
    time.sleep(seconds)
    return len(file) * 100

files = ["март.pdf", "април.pdf", "май.pdf", "юни.pdf"]
with ThreadPoolExecutor(max_workers=2) as pool:
    sizes = [pool.submit(size_of, file, 0.3) for file in files]
    for file, size in zip(files, sizes):
        print(f"{file}: {size.result()} KB")

март.pdf: 800 KB
април.pdf: 900 KB
май.pdf: 700 KB
юни.pdf: 700 KB

ThreadPoolExecutor(max_workers=2) прави басейн от 2 нишки. Четирите задачи се редуват в тях.
submit(функция, аргументи) приема задача и веднага дава Future: обещание, че резултатът ще дойде.
result() чака задачата да свърши и връща резултата. Ако задачата е хвърлила изключение, result() го хвърля наново.
with спира басейна накрая: чака задачите да свършат и спира нишките му.

По-точно, за любопитните

Ако задачата не връща нищо, просто не пазиш Future. Краят на with чака всички:

with ThreadPoolExecutor(max_workers=2) as pool:
    for i in range(1, 6):
        pool.submit(download, f"отчет-{i}.pdf", 0.1)
print("Всички свършиха.")

Свалих отчет-2.pdf
Свалих отчет-1.pdf
Свалих отчет-3.pdf
Свалих отчет-4.pdf
Свалих отчет-5.pdf
Всички свършиха.

Първите два отчета вървят едновременно, затова редът им се сменя от пускане на пускане. Без with пишеш pool.shutdown(wait=True) накрая.

Пробвай

Кой резултат пръв?

„архив.zip“ се сваля 0,6 s, а „бележка.txt“ 0,1 s. Какво ще отпечата програмата?

with ThreadPoolExecutor(max_workers=2) as pool:
    big = pool.submit(size_of, "архив.zip", 0.6)
    small = pool.submit(size_of, "бележка.txt", 0.1)
    print(f"Голям: {big.result()}")
    print(f"Малък: {small.result()}")

Отговор

Зависи от това коя задача свърши първа

Голям: 900
Малък: 1100

Малък: 1100
Голям: 900

Упражнение

Програми с нишки

Реши задачите в редактора си.

1. Паралелна сума. Списъкът sales има 1 000 000 продажби, sales[i] = i % 100. Раздели го на 4 равни части и сумирай всяка в своя нишка. Всяка нишка пише в свое място в списък partial, затова не трябва синхронизация. Изчакай ги с join и събери частите:

Части: [12375000, 12375000, 12375000, 12375000]
Общо: 49500000

По-точно, за любопитните

Заради GIL четирите нишки не смятат по-бързо от една. Истински паралелно е с процеси: ProcessPoolExecutor пуска отделни копия на Python, всяко със свой GIL. Частите им се подават, а резултатът се връща:

from concurrent.futures import ProcessPoolExecutor


if __name__ == "__main__":
    sales = [i % 100 for i in range(1_000_000)]
    size = len(sales) // 4
    chunks = [sales[p * size:(p + 1) * size] for p in range(4)]
    with ProcessPoolExecutor(max_workers=4) as pool:
        partial = list(pool.map(sum, chunks))
    print(f"Части: {partial}")
    print(f"Общо: {sum(partial)}")

Части: [12375000, 12375000, 12375000, 12375000]
Общо: 49500000

if __name__ == "__main__": пази главния код да не се изпълни наново във всеки процес: на Windows и macOS процесът тръгва, като зарежда файла отначало.

2. Сваляне с басейн. Свали шест отчета (март.pdf, април.pdf, май.pdf, юни.pdf, юли.pdf, август.pdf) с басейн от 3 нишки. Ползвай size_of(file, 0.3) от картата за басейна. Отпечатай размерите по реда на файловете, сбора и дали всичко е станало за под 1,5 секунди (едно след друго би било 1,8 s):

март.pdf: 800 KB
април.pdf: 900 KB
май.pdf: 700 KB
юни.pdf: 700 KB
юли.pdf: 700 KB
август.pdf: 1000 KB
Общо: 4800 KB
Под 1,5 секунди: True

3. Брояч без загуби. Четири нишки, всяка с по 50 000 посещения, увеличават общ брояч. Напиши брояча така, че резултатът да е винаги 200 000. Опитай първо без Lock и с time.sleep(0) между прочитането и записването и виж колко се губи, после поправи с Lock.

Посещения: 200000

Още задачи

Тест за брояча. Напиши unittest тест, който пуска 8 нишки по 10 000 увеличения и проверява 80 000 с assertEqual. Пусни го срещу брояча с time.sleep(0) и без Lock няколко пъти. Тест, който пада понякога, е типичният знак за състезание.
Мъртвата хватка. Пусни примера с принтера и скенера от „По-точно“ (с две нишки и join) и виж как увисва. Поправи го, като и двете нишки взимат първо принтера.

Тест

Провери се

1. Каква е разликата между t.start() и t.run()?Няма разликаstart() пуска run() в нова нишка, а run() го изпълнява веднага в текущатаrun() пуска нова нишка, а start() не

2. Защо програмата вика join() преди да отпечата резултата?За да спре нишката веднагаЗа да изчака нишката да свършиЗа да стартира нишката

3. Две нишки увеличават общ брояч без Lock. Какво обещава Python за резултата?Нищо: ако нишката смени между прочитането и записването, едно увеличение се губиЧе винаги е верен заради GILЧе винаги е наполовина

4. Какво гарантира with self.lock: около няколко реда?Че нишките се изпълняват по реда на стартаЧе редовете са по-бързиЧе в даден момент само една нишка е вътре в блока с тази ключалка

5. Пускаш три задачи с pool.submit и пазиш Future за всяка. Какво прави future.result()?Връща резултата веднага, дори да не е готовПуска задачата нановоЧака задачата да свърши и връща резултата ѝ

Нишки

Бисквитки