day29

Pythonでのマルチスレッドプログラミング、スレッドセキュリティとロック(一)
1.マルチスレッドプログラミングとスレッドセキュリティに関する重要な概念
GIL、スレッド、プロセス、スレッドセキュリティ、原子操作.

GIL:Global Interpreter Lock、グローバルインタプリタロック.マルチスレッド間のデータ整合性とステータス同期の問題を解決するために、任意の時点で1つのスレッドだけが解釈器で実行されるように設計されています.

スレッド:プログラム実行の最小単位.

プロセス:システムリソース割り当ての最小単位.

スレッドセキュリティ:マルチスレッド環境では、共有データは同じ時間に1つのスレッドしか操作できません.

原子操作:原子操作は、プロセスの同時またはスレッドの同時によって中断される操作ではありません.

もう1つの重要な結論は、グローバルリソースに書き込み操作がある場合、書き込みプロセスの原子性が保証されないと、スレッドが安全ではないという汚れた読み書きが発生するということです.PythonのGILは原子操作のスレッドセキュリティしか保証できないので,マルチスレッドプログラミングではロックをかけてスレッドセキュリティを保証する必要がある.
最も簡単なロックは反発ロック(同ステップロック)であり、反発ロックはio密集型シーンで発生した計算エラーを解決するために使用される.すなわち、共有されたデータを保護するために使用され、同じ時間に共有されたデータを変更するスレッドは1つしかない.
2. Threading.ロックが相互反発ロックを実現する簡単な例
私たちはThreadingを通じてロック()はロックを実現します.
スレッドが安全でない例を次に示します.

>>> import threading
>>> import time
>>> def sub1():
    global count
    tmp = count
    time.sleep(0.001)
    count = tmp + 1
    time.sleep(2)

    
>>> count = 0
>>> def verify(sub):
    global count
    thread_list = []
    for i in range(100):
        t = threading.Thread(target=sub,args=())
        t.start()
        thread_list.append(t)
    for j in thread_list:
        j.join()
    print(count)

    
>>> verify(sub1)
14

この例では

count+=1

代わりに

tmp = count
time.sleep(0.001)
count = tmp + 1

なぜなら、count+=1は非原子操作であるにもかかわらず、CPUの実行が速すぎるため、マルチプロセスの非原子操作によるプロセスの安全性を再現することが困難であるからである.代替後、sleepは0.001秒しか経っていないが、CPUの時間にとって非常に長く、このコードブロックが半分に実行され、GILロックが解放される.すなわち、tmpはcountの値を取得したが、tmp+1はcountに付与されていない.このとき他のスレッドがcount=tmp+1を実行した場合、元のスレッド実行に戻るとcountの値は更新されたにもかかわらず、count=tmp+1は付与操作であり、付与の結果はcountの更新の値と同じである.最終的には、私たちが累積した値が多く失われました.
スレッドのセキュリティの例を次に示しますthreadingを使用します.ロック()取得ロック
ロックを取得してロックを解放する文もPythonのwithで実現でき、より簡潔になります.

>>> count = 0
>>> def sub3():
    global count
    with lock:
        tmp = count
        time.sleep(0.001)
        count = tmp + 1
        time.sleep(2)

>>> def verify(sub):
    global count
    thread_list = []
    for i in range(100):
        t = threading.Thread(target=sub,args=())
        t.start()
        thread_list.append(t)
    for j in thread_list:
        j.join()
    print(count)
        
>>> verify(sub3)
100

3.二種類のデッドロック状況及び処理
デッドロックの原因
2つのデッドロック:
3.1反復デッドロックと再帰ロック(RLock)
この場合、スレッドが同じリソースを「反復」して要求すると、デッドロックが直接発生します.このデッドロックの原因は我々の標準的な反発ロックthreadingである.ロックの欠点が原因です.標準的なロックオブジェクト(threading.Lock)は、現在どのスレッドがロックを占有しているかに関心を持っていない.ロックが占有されている場合、ロックを取得しようとする他のスレッドはブロックされ、ロックを占有しているスレッドもブロックされます.
次に例を示します.

#/usr/bin/python3
# -*- coding: utf-8 -*-

import threading
import time

count_list = [0,0]
lock = threading.Lock()

def change_0():
    global count_list
    with lock:
        tmp = count_list[0]
        time.sleep(0.001)
        count_list[0] = tmp + 1
        time.sleep(2)
        print("Done. count_list[0]:%s" % count_list[0])
        
def change_1():
    global count_list
    with lock:
        tmp = count_list[1]
        time.sleep(0.001)
        count_list[1] = tmp + 1
        time.sleep(2)
        print("Done. count_list[1]:%s" % count_list[1])
        
def change():
    with lock:
        change_0()        time.sleep(0.001)
        change_1()
    
def verify(sub):
    global count_list
    thread_list = []
    for i in range(100):
        t = threading.Thread(target=sub, args=())
        t.start()
        thread_list.append(t)
    for j in thread_list:
        j.join()
    print(count_list)
    
if __name__ == "__main__":
    verify(change)

例では、共有リソースcount_があります.Listは、この共有リソースの第1および第2の部分をそれぞれ取得する2つの数字(count_list[0]およびcount_list[1])を有する.両方のアクセス関数は、データの取得時に他のスレッドが対応する共有データを変更していないことを確認するためにロックを使用します.次に、3番目の関数を追加して2つの部分のデータを取得する方法を考えます.簡単な方法は、この2つの関数を順番に呼び出し、結合した結果を返すことです.
ここでの問題は、あるスレッドが2つの関数呼び出し間で共有リソースを変更した場合、最終的に一致しないデータが得られることです.
最も明らかな解決法は,この関数においてもlockを用いることである.しかし、これは不可能です.外側の文がロックを占有しているため、内側の2つのアクセス関数がブロックされます.
結果は出力がなく、デッドロックです.
この問題を解決するためにthreadingを使うことができます.RLockはthreadingの代わりにLock

#/usr/bin/python3
# -*- coding: utf-8 -*-

import threading
import time

count_list = [0,0]
lock = threading.RLock()

def change_0():
    global count_list
    with lock:
        tmp = count_list[0]
        time.sleep(0.001)
        count_list[0] = tmp + 1
        time.sleep(2)
        print("Done. count_list[0]:%s" % count_list[0])
        
def change_1():
    global count_list
    with lock:
        tmp = count_list[1]
        time.sleep(0.001)
        count_list[1] = tmp + 1
        time.sleep(2)
        print("Done. count_list[1]:%s" % count_list[1])
        
def change():
    with lock:
        change_0()        time.sleep(0.001)
        change_1()
    
def verify(sub):
    global count_list
    thread_list = []
    for i in range(100):
        t = threading.Thread(target=sub, args=())
        t.start()
        thread_list.append(t)
    for j in thread_list:
        j.join()
    print(count_list)
    
if __name__ == "__main__":
    verify(change)

3.2デッドロックとロックの昇順使用を互いに待つ
デッドロックのもう1つの原因は、2つのプロセスが取得したいロックが相手のプロセスによって取得され、互いに待つしかなく、取得したロックを解放することができず、デッドロックを招くことです.銀行システムにおいて、ユーザaがユーザbに100元を振替しようとするとともに、ユーザbがユーザaに500元を振替しようとすると、デッドロックが発生する可能性がある.2つのスレッドは互いに相手のロックを待ち合い,互いにリソースを占有して解放しない.
次に、相互呼び出しによるデッドロックの例を示します.

#/usr/bin/python3
# -*- coding: utf-8 -*-

import threading
import time

class Account(object):
    def __init__(self, name, balance, lock):
        self.name = name
        self.balance = balance
        self.lock = lock
        
    def withdraw(self, amount):
        self.balance -= amount
        
    def deposit(self, amount):
        self.balance += amount
        
def transfer(from_account, to_account, amount):
    with from_account.lock:
        from_account.withdraw(amount)
        time.sleep(1)
        print("trying to get %s's lock..." % to_account.name)
        with to_account.lock:
            to_account_deposit(amount)
    print("transfer finish")
    
if __name__ == "__main__":
    a = Account('a',1000, threading.Lock())
    b = Account('b',1000, threading.Lock())
    thread_list = []
    thread_list.append(threading.Thread(target = transfer, args=(a,b,100)))
    thread_list.append(threading.Thread(target = transfer, args=(b,a,500)))
    for i in thread_list:
        i.start()
    for j in thread_list:
        j.join()
    

最終的な結果はデッドロックです.

trying to get account a's lock...
trying to get account b's lock...

私たちの問題は
マルチスレッドプログラムを書いています.スレッドは一度に複数のロックを取得する必要があります.この場合、デッドロックの問題を回避するにはどうすればいいですか.解決策:マルチスレッドプログラムでは、デッドロックの問題の大部分は、スレッドが複数のロックを同時に取得することによるものです.例を挙げると、1つのスレッドが1つ目のロックを取得し、2つ目のロックを取得するときにブロックが発生すると、このスレッドは他のスレッドの実行をブロックし、プログラム全体が偽死する可能性があります.実はこの問題を解決して、核心の思想も特に簡単です:現在私達が出会った問題は2つのスレッドが取得したいロックで、すべて相手のスレッドに手に入れられて、それでは私達はこの2つのスレッドの中で保証するだけで、ロックを取得する順序は一致していればいいです.例としてスレッドthread_がありますa, thread_b,ロックロックロック_1, lock_2.ロックの使用順序を決めさえすれば、例えばロックを先に使う.1、またlock_2スレッドthread_a lock_を取得1の場合、thread_のような他のスレッドb lock_を取得できません1このロックは、次の操作(lock_2というロックを取得)を行うこともできず、互いに待ち合うことによるデッドロックを招くこともありません.簡単に言えば、デッドロックの問題を解決する1つの方法は、プログラム内の各ロックに一意のidを割り当て、昇順ルールに従って複数のロックのみを使用することを許可することです.このルールは、コンテキストマネージャを使用すると非常に容易に実現できます.例は、次のとおりです.

#/usr/bin/python3
# -*- coding: utf-8 -*-

import threading
import time
from contextlib import contextmanager

thread_local = threading.local()

@contextmanager
def acquire(*locks):
    #sort locks by object identifier
    locks = sorted(locks, key=lambda x: id(x))
    
    #make sure lock order of previously acquired locks is not violated
    acquired = getattr(thread_local,'acquired',[])
    if acquired and (max(id(lock) for lock in acquired) >= id(locks[0])):
        raise RuntimeError('Lock Order Violation')
    
    # Acquire all the locks
    acquired.extend(locks)
    thread_local.acquired = acquired
    
    try:
        for lock in locks:
            lock.acquire()
        yield
    finally:
        for lock in reversed(locks):
            lock.release()
        del acquired[-len(locks):]

class Account(object):
    def __init__(self, name, balance, lock):
        self.name = name
        self.balance = balance
        self.lock = lock
        
    def withdraw(self, amount):
        self.balance -= amount
        
    def deposit(self, amount):
        self.balance += amount
        
def transfer(from_account, to_account, amount):
    print("%s transfer..." % amount)
    with acquire(from_account.lock, to_account.lock):
        from_account.withdraw(amount)
        time.sleep(1)
        to_account.deposit(amount)
    print("%s transfer... %s:%s ,%s: %s" % (amount,from_account.name,from_account.balance,to_account.name, to_account.balance))
    print("transfer finish")
    
if __name__ == "__main__":
    a = Account('a',1000, threading.Lock())
    b = Account('b',1000, threading.Lock())
    thread_list = []
    thread_list.append(threading.Thread(target = transfer, args=(a,b,100)))
    thread_list.append(threading.Thread(target = transfer, args=(b,a,500)))
    for i in thread_list:
        i.start()
    for j in thread_list:
        j.join()
    

私たちが得た結果は

100 transfer...
500 transfer...
100 transfer... a:900 ,b:1100
transfer finish
500 transfer... b:600, a:1400
transfer finish

互いに待つことによるデッドロックの問題を回避することに成功した.
上記のコードでは、いくつかの構文を説明する必要があります.

\1. アクセラレータ@contextmanagerは、with文でロックを呼び出すことができ、ロックの取得と解放プロセスを簡素化するために使用されます.with文については、Pythonのwith文を参照してください.簡単に言えば、with文は呼び出し時にenter()メソッドを実行し、with構造内の文を実行し、最後に__を実行する.exit__()文.デコレーション@contextmanager.ジェネレータ関数のyieldより前の文はenter()メソッドで実行され、yieldより後の文はexit()で実行され、yieldが生成した値はasサブ文のvalue変数に割り当てられます.

\2. try文とfinally文では、ロックの取得と解放が実現されます.

\3. try以前の文は,ロックのソート,およびロックのソートが破壊されたか否かの判断を実現する.