Concurrent Programming in Python

01オブジェクト指向とは

Java 版で述べたとおり、オブジェクト指向は Simula67 に始まる考え方で、世界をもの(object) の 集まりとして捉え、もの同士のメッセージの受渡しで現象が進むと考える。オブジェクトは 状態・メソッド・計算を実行する主体の 3 つから成る、という見方も同じである。

Python はこの考え方を徹底していて、整数も関数もクラスも、すべてがオブジェクトである。 一方で「計算を実行する主体」については独特の事情があり、それがこの資料の後半の主題になる。

02Python のオブジェクト

Python でオブジェクトの状態は、既定では辞書(__dict__) に入っている。 外から自由に足せてしまうのは、Java とも JavaScript の #private とも違う点である。

class Tree:
    def __init__(self):
        self.status = '立っている'

    def fall(self):
        self.status = '倒れた'

t = Tree()
t.fall()
print(t.status)      # 倒れた
print(t.__dict__)    # {'status': '倒れた'}

t.color = '緑'       # 定義していない属性も後から生やせる

「外から触ってほしくない」ことを示すには、名前の先頭にアンダースコアを付ける習慣がある。 アンダースコア 2 つにすると名前マングリングが働き、クラスの外からは _クラス名__名前 でしか触れなくなる。ただしこれも「隠す」のではなく 「うっかり衝突させない」ための仕組みで、Java の private のような強制力はない。

class Tree:
    def __init__(self):
        self._status = '立っている'    # 「触らないでね」という合図（強制力なし）
        self.__secret = 42             # _Tree__secret に名前が変わる

t = Tree()
t._status            # 触れてしまう
t.__secret           # AttributeError
t._Tree__secret      # 42  ← 実は触れる

03class 構文

Python にはメソッドの多重定義（オーバーロード）が無い。同じ名前で 2 回定義すると 後の定義が前を上書きする。これは JavaScript と同じ事情である。

class Calc:
    def add(self, a, b):    ...
    def add(self, a, b, c): ...    # ← 前の add を消してしまう

引数の数で使い分けたいときは、既定値・可変長引数で書く。型で分けたいときは functools.singledispatch がある。

from functools import singledispatch

@singledispatch
def draw(x):            return f'なにか: {x}'
@draw.register
def _(x: int):          return f'整数: {x}'
@draw.register
def _(x: str):          return f'文字列: {x}'

draw(3)      # 整数: 3
draw('あ')   # 文字列: あ

04Python の並行性

ここからが本題である。Python には本物のスレッド（OS のスレッド）がある。 Java と同じように threading.Thread で作れるし、いつ割り込まれるか分からない プリエンプティブな切り替えも起きる。

ところが GIL(Global Interpreter Lock、グローバルインタプリタロック) があるため、 同時に Python のバイトコードを実行できるスレッドは常に 1 つだけである。 スレッドは存在するのに、計算は並列にならない。

つまり Python のスレッドは「速くするため」ではなく「待ち time を重ねるため」にある。 通信やファイル入出力の待ちの間は GIL が解放されるので、そこは重ねられる。

05threading

スレッドの作り方は Java とよく似ている。

import threading, time

def worker(name):
    for i in range(3):
        print(name, i)
        time.sleep(0.1)      # ここで GIL を手放す

ts = [threading.Thread(target=worker, args=(f'T{i}',)) for i in range(3)]
for t in ts: t.start()
for t in ts: t.join()        # 終わるまで待つ

Java 版の WireFrame の例のように、各オブジェクトに自分のスレッドを持たせることもできる。

class Rotator(threading.Thread):
    def __init__(self, shape):
        super().__init__(daemon=True)
        self.shape = shape

    def run(self):               # start() から呼ばれる
        while True:
            self.shape.rotate()
            time.sleep(1 / 30)

06同期をとる

Python の threading には、Java 版で扱った道具がひととおり揃っている。

セマフォの P／V は、Python では acquire()／release() である。

sem = threading.Semaphore(2)     # 資源は 2 つ

with sem:            # acquire()（＝P）、抜けるとき release()（＝V）
    ...              # ここを通れるのは同時に 2 スレッドまで

07multiprocessing

CPU を使う処理を本当に速くしたいなら、プロセスを分ける。 プロセスごとに Python インタプリタが 1 つずつ動くので、GIL も 1 つずつになり、 本当に並列に実行される。

from concurrent.futures import ProcessPoolExecutor

def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)

with ProcessPoolExecutor() as ex:
    print(list(ex.map(fib, [32, 32, 32, 32])))   # 4 コアならほぼ 1 個分の時間

代わりにメモリを共有しない。やりとりする値は pickle で直列化されて プロセス間を渡るので、渡せない値（ラムダ式、開いたファイルなど）は使えず、 大きなデータのやりとりには時間がかかる。これは Web Worker の postMessage とまったく同じ事情である。

# 共有したいときは専用の入れ物を使う
from multiprocessing import Value, Lock

count = Value('i', 0)          # 共有メモリ上の整数
lock  = Lock()

with lock:
    count.value += 1           # ここでもロックは要る

08asyncio

もう一つの道具が asyncio である。こちらはスレッドを使わず、 1 本の流れの中で待ち合わせ点を作って譲り合う。JavaScript のイベントループと 同じ考え方で、実際 async／await という書き方まで同じである。

import asyncio

async def worker(name):
    print(name, '開始')
    await asyncio.sleep(1)      # ここで他に譲る（1秒間ずっと譲っている）
    print(name, '終了')

async def main():
    await asyncio.gather(worker('A'), worker('B'), worker('C'))

asyncio.run(main())
# 3 つ同時に開始し、約 1 秒後に 3 つ同時に終わる

ここでも競合は起きる。await をまたいで「読む→書く」をすると、 JavaScript 版で見たのと同じように更新が消える。

count = 0

async def inc():
    global count
    v = count               # 読む
    await asyncio.sleep(0)  # ← ここで譲る
    count = v + 1           # 古い v をもとに書く

asyncio.run(asyncio.gather(inc(), inc()))
# count は 2 ではなく 1

直すには asyncio.Lock を使う。threading.Lock とは別物なので 取り違えないこと（threading.Lock をイベントループの中で使うと、 待っている間ループ全体が止まってしまう）。

lock = asyncio.Lock()

async def inc():
    global count
    async with lock:
        v = count
        await asyncio.sleep(0)
        count = v + 1

09どれを使うか

3 つの道具の使い分けは、処理が何を待っているかで決まる。

asyncio と threading はどちらも入出力待ちに使えるが、性格が違う。 asyncio は譲る場所が await に限られるので、どこで割り込まれるかが コードから読める。threading はどこでも割り込まれる代わりに、 既存の同期的なコードをそのまま動かせる。次のデモで、条件を変えて所要時間を比べられる。

10課題

課題 1　次のクラス定義について、D.__mro__ を手で求めよ。 C3 線形化の 3 つの規則のどれがどこで効いているかを説明すること。求まったら MRO のデモで確かめよ。

class A: pass
class B(A): pass
class C(A): pass
class D(C, B): pass      # ← B, C ではなく C, B

課題 2　count += 1 を 10 万回ずつ 2 スレッドで実行したとき、 なぜ 20 万にならないことがあるのか。dis の出力のどの命令とどの命令の間で 切り替わると更新が消えるのかを示して説明せよ。

課題 3　fib(32) を 4 個計算する処理を、①素直に順番に ②ThreadPoolExecutor ③ProcessPoolExecutor の 3 通りで書き、所要時間を測って比べよ。 ②が速くならない理由を GIL の言葉で説明すること。

課題 4　上の BoundedBuffer の while を if に書き換えると、 どんなときに壊れるか。生産者 2 人・消費者 1 人の場合で、誰がいつ起こされるかを追って示せ。

課題 5　asyncio で、同時に 3 本までしか走らせない async def limit(n, coros) を asyncio.Semaphore を使って書け。 JavaScript 版の課題 3 と同じ問題を、Python で解くとどうなるか比べよ。