13. Конкурентност

19 май 2014

Паралелизация vs конкурентност

Всъщност каква е разликата?

Moore's law

А какво става, когато имаме много ядра?

Подходи

Отделни процеси
Нишки (два вида)
Актьори
Software Transactional Memory
Мега умни компилатори?

В Python

Отделни процеси
Нишки

Видове нишки

Зелени нишки
Системни нишки

Global Interpreter Lock

Прави нещата по-неприятни

fork

fork създава ново копие на програмата, която изпълняваме
Всички ресурси и променливи запазват стойността си в процеса-син
След създаването на новия процес, всички промени са локални
Все едно клонираме хора, за да вършим повече работа едновременно

Пример с fork (на C)

#include <stdio.h>

int main()
{   printf("before\n");
    if (fork())
        printf("father\n");
    else
        printf("son\n");
    printf("both\n");
}


before
son
both
father
both

Пример с fork (на Python)

import os

print("before")
if os.fork():
    print("father")
else:
    print("son")
print("both")

По-сложен пример с fork

import os
import time

def log(msg): print("\n* " + msg)

orders = 0
while True:
    order = input('Enter order: ')
    if not order: continue
    if order in ('q', 'x', 'quit', 'exit'): break
    pid = os.fork()
    if pid == 0:
        time.sleep(3)
        log("Order '{0}' is ready!".format(order))
        break
    else:
        log("Roger that '{0}'({1}). Please, wait.".format(order, orders))
        orders += 1

Синхронизация при fork

import os

pid = os.fork()
if pid == 0:
    os.execlp('date', 'date')
else:
    status = os.wait()
    print("Father: son has finished {0}".format(status))

можем да ползваме и сигнали import signal

Предимства и недостатъци на fork

Против:

Само за UNIX
Създаването на нов процес е бавно и паметоемко
Комуникацията между процеси е трудна - нямат обща памет

За:

Стабилност
Синът е независим - ако омаже нещо, бащата няма да пострада

Нишки

Наричат ги lightweight процеси
Много нишки живеят в един и същи процес
Глобалните променливи са общи за нишките
Създават се бързо и лесно
В python модулът threading

Създаване на нова нишка

даваме функция с параметри към нея

или

Нишка — наследник на threading.Thread
Код — в метода run
Създаване и изпълнение на кода паралелно — метод start

Пример с нишки

import threading

def f(name):
    print("Hello from {0}".format(name))

thread = threading.Thread(target=f, args=('Bob',))
thread.start()
thread.join()

Пример с нишки(2)

import threading
import time

orders = 0

class Chef(threading.Thread):
    def __init__(self, order):
        self.order = order
        threading.Thread.__init__(self)
    def run(self):
        time.sleep(3)
        log("Order '{0}' is ready!".format(self.order))

while True:
    order = input('Enter order: ')
    if not order: continue
    if order in ('q', 'x', 'quit', 'exit'): break
    chef = Chef(order)
    chef.start()
    log("Roger that '{0}'. Please, wait in quiet desperation.".format(order))
    orders += 1

Вечерящи философи

import random, time, threading

taken = False
class Philosopher(threading.Thread):
    def __init__(self, name):
        super().__init__(); self.name = name

    def log(self, msg): print("{0}: {1}".format(self.name, msg))
    def eat(self): time.sleep(random.random())
    def ponder(self): time.sleep(random.random())

    def refresh(self):
        global taken
        self.log("Please excuse me...");
        while taken: pass;
        taken = True; self.log("--> (entered the bathroom)")
        time.sleep(random.random())
        taken = False; self.log("<-- (left the bathroom)")

    def run(self):
        while True:
            self.eat(); self.ponder(); self.refresh()

Критични секции

Части от кода, които могат да бъдат изпълнени само от една нишка/процес в даден момент се наричат критични секции
Те са критична част от многозадачното програмиране
Има много похвати за реализирането на критични секции.
Всичките са равномощни на нещо, наречено семафор.

threading.Lock

threading.Lock() ни връща Lock обект
Викането на метода му acquire() ни гарантира, че само ние притежаваме този Lock
Викането на release() освобождава Lock-а и разрешава някой друг да го заключи с acquire()
Ако викнем acquire() докато Lock-а е зает — методът чака, докато не се освободи

Философите, отново

import random, time, threading

bathroom = threading.Lock()

class Philosopher(threading.Thread):
    def __init__(self, name):
        super().__init__(); self.name = name

    def log(self, msg): print("{0}: {1}".format(self.name, msg))
    def eat(self): time.sleep(random.random())
    def ponder(self): time.sleep(random.random())

    def refresh(self):
        self.log("Please excuse me...")
        bathroom.acquire(); self.log("--> (entered the bathroom)")
        time.sleep(random.random())
        bathroom.release(); self.log("<-- (left the bathroom)")

    def run(self):
        while True:
            self.eat(); self.ponder(); self.refresh()

with и обекти с acquire и release

Всички Lock-оподбни обекти от threading са и context manager-и
Ако ги ползваме с with ни се гарантира викането на acquire() преди и на release() след блока

with bathroom:
    self.log("--> (entered the bathroom)")
    time.sleep(random.random())
    self.log("<-- (left the bathroom)")

Модерно строителство, модерни ресторанти и семафори

ако има повече от една тоалетна в сградата?

Или още по-добре

Ресторант с 10 човека
Всеки е едновременно готвач и сервитьор
5 фурни

Семафори в орехова черупка

Числова променлива v
Операция P(v) — чакай докато v > 0, след което v -= 1
Операция V(v) — v += 1
Предложени от Едсгер Дейкстра (Едсгар Дийкстра?)

threading.Semaphore

threading.Semaphore(k) ни връща семафор с интерфейс като на Lock и стойност k
При всяко изпълнение на acquire() стойността се намалява с 1
При всяко изпълнение на release() стойността се увеличава с 1
Ако стойността е 0, acquire() спи, докато някой не я увеличи с release()
Lock() е еквивалентен на Semaphore(1)

Ресторанта на Грибоедов

import threading, random, time

ovens = threading.Semaphore(5)

class WaiterChef(threading.Thread):
    def __init__(self, name):
        super(WaiterChef, self).__init__()
        self.name = name

    def run(self):
        while True:
            print("...({0}) waiting for an oven".format(self.name))
            ovens.acquire()
            print("--> ({0}) Cooking...".format(self.name))
            time.sleep(random.random() * 10)
            ovens.release()
            print("<-- ({0}) Serving...".format(self.name))
            time.sleep(random.random() * 4)


for _ in range(0, 10):
  WaiterChef(_).start()

threading.Event

По-проста комуникация, в която няколко нишки изчакват събитие от друга
wait() блокира докато събитието не се случи
set() „случва“ събитието

Коледа на село

2 баби, майсторки баничарки, пекат баници за 3-мата си внука
Искаме след като бабите опекат нещо, някой от свободните внуци да започва да яде
Бабите се казват producer-и, внуците — consumer-и

threading.Condition

wait() — чака, докато някоя баба не произведе нова баница
notify() — това трябва да каже една баба, която е опекла баница. Ще събуди някой от чакащите. Ако няма чакащи няма да направи нищо.
notifyAll() — ще събуди всички чакащи
release() и acquire() работят върху вътрешен за Condition Lock, който може да се подаде при конструиране
wait и notify работят само ако владеем вътрешния Lock

threading.local

Инстанцира се чрез threading.local()
Позволява свободен достъп до атрибути
Всяка нишка вижда различни стойности зад атрибутите на този обект

multiprocessing

multiprocessing модулът

Дава ни подобни възможности като threading, но за процеси
Crossplatform - Unix/Windows
Благинки за синхронизация - Semaphore, Lock, RLock, Condition, Event
Благинки за обмен на данни - Queue, Pipe
Възможности за обща памен (Value, Array) от елементарни данни (int/float/byte/...) и ctypes структури
Възможност за общи обекти - Manager
Pool

multiprocessing - пример

from multiprocessing import Process
import os

def info(title):
    print(title)
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())

def f(name):
    info('function f')
    print('hello', name)

if __name__ == '__main__':
    info('main line')
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

multiprocessing - пример с обща памет

from multiprocessing import Process, Value

def f(n):
    # work
    v = n.value
    for x in range(0, 30000): x=x+2
    n.value = v + 1
    # work

if __name__ == '__main__':
    num = Value('i', 0)
    processes = [Process(target=f, args=(num,)) for i in range(0, 10)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(num.value)

multiprocessing - Lock

Lock - осигурява че само един процес може да го държи

from multiprocessing import Process, Value, Lock

def f(n, lock):
    # work
    lock.acquire()
    v = n.value
    for x in range(0, 30000): x=x+2
    n.value = v + 1
    lock.release()
    # work

if __name__ == '__main__':
    num = Value('i', 0)
    lock = Lock()
    processes = [Process(target=f, args=(num, lock)) for i in range(0, 10)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(num.value)

multiprocessing - Manager

Позволява създаването на споделени обекти

по-удобно от обща памет
но и по-бавно

from multiprocessing import Process, Manager

def f(d, l):
    d[1] = '1'
    d['2'] = 2
    d[0.25] = None
    l.reverse()

if __name__ == '__main__':
    manager = Manager()
    d, l = manager.dict(), manager.list(range(10))
    p = Process(target=f, args=(d, l))
    p.start()
    p.join()
    print(d, l)

{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

multiprocessing - неща, за които трябва да се внимава под Windows

аргументите на Process.__init__() трябва да са „picklable“
не е хубаво да достъпвате глобални променливи
задължително if __name__ == '__main__'
трябва да се внимава с действията при import