Python3のyieldでのスレッド作成について

Question

前提・実現したいこと

Python3でThreadPoolExecutorを使用してスレッドを実行しようと、
下記の様なソースコードを試しました。

実現したいことは関数「create_tran(tg_list)」で出力で引数のリストの文字列を順に出力するスレッドを作成し、ThreadPoolExecutorに読み込ませて平行にメッセージを出力したいです。
(例：出力したいメッセージ)

下記のソースコードを実行したところ、リストの最終文字(ccccc)が出力されるスレッドが三回実行されてしまいます。
(例：下記のソースコードを実行した場合の出力)

ジェネレータ理解不足で申し訳ないのですが、原因とジェネレータでスレッドを作成する時の注意点などありましたらアドバイス頂きたいと考えています。

出力したい出力

aaaaa
bbbbb
ccccc

下記のソースコードを実行した場合の出力

ccccc
ccccc
ccccc

該当のソースコード

import time
import concurrent.futures


def create_tran(tg_list):
    for tg_name in tg_list:
        def _tran():
            time.sleep(3)
            print(tg_name)
        yield _tran


def main():
    tg_list = ['aaaaa', 'bbbbb', 'ccccc']
    tran_list = create_tran(tg_list)
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
    for tran in tran_list:
        executor.submit(tran)


if __name__ == "__main__":
    main()

試したこと

time.sleep(3)を削除しましたら、目的の出力が行われましたが関数「_tran()」内に重い処理を記載しましたらやはり同一の現象がおこりました。

Answer 1

tg_nameはcreate_tranのローカル名前空間に紐付いているのでfor文がまわるたびに更新されます。で、print(tg_name)の実行時にはfor文はすでに3回まわった状態なので意図と反する'ccccc'が出力されます。

def create_tran(tg_list):
    for tg_name in tg_list:
        def _tran():
            time.sleep(3)
            print(tg_name)  # for文1回目の間は'aaaaa'、2回目の間は'bbbbb'、3回目の間は'ccccc'
        yield _tran

意図する動作をさせたいなら以下のようにtg_nameは_tranのローカル名前空間に紐付けないとダメです。

def create_tran(tg_list):
    def _tran(tg_name):
        def inner():
            time.sleep(3)
            print(tg_name)
        return inner

    for t in tg_list:
        yield _tran(t)

Comments

[sdice]

誤動作する理由と修正方法を簡潔に記載して頂きありがとうございました。
ローカルの名前空間へ紐づける必要があったのですね。

Answer 2

これでいけますよ。

import time
import concurrent.futures


def create_tran(tg_list):
    for tg_name in tg_list:
        def _tran(tg_name=tg_name):
            time.sleep(3)
            print(tg_name)
        yield _tran


def main():
    tg_list = ['aaaaa', 'bbbbb', 'ccccc']
    tran_list = create_tran(tg_list)
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
    for tran in tran_list:
        executor.submit(tran)


if __name__ == "__main__":
    main()

Comments

[sdice]

短い修正で実現する方法を教えて頂きありがとうございます。

Answer 3

for文の制御変数tg_nameが関数_tranからどう参照されるのかの意味をよく注意すると原因と対処がわかってくると思います。本件の本質はどちらかといえばジェネレーターやスレッドとは独立した話題であり、単にPythonのfor文の制御変数と関数スコープに関する仕様にあると思います。

単純化した例を挙げますと・・・

リスト1

functions = []
for i in range(2):  # ...(A)
    def f():
        return i
    functions.append(f)
print([f() for f in functions])

この結果は[0 1]とはならず[1 1]となります。
なぜそうなるかを上のコードとほぼ同様の意味の別の表現で表してみると次のようになります。

リスト2

i = 0
def f0():
    return i
i = 1
def f1():
    return i
functions = [f0, f1]
print([f() for f in functions])

functionsの要素である2つの関数がprint文の引数を計算する際に順番に実行されますがそのときのiの値は1になっています。リスト1の(A)のfor文の制御変数iもリスト2の変数iも関数fから見れば実質的には同じであり、「iは関数fのスコープの外にある変数」なので関数fが実行された時点でのiの値が参照されることになります。

(for文の繰り返し処理の内側で関数を定義したとき、感覚的にはその関数を実行した際に関数を定義した時点でのiの値が関数内部から見えるかのように誤解しがちだと思いますが実はそうではなくあくまで変数iは関数を実行した時点での値が見えるだけであることに注意してください。)

---

対処：
あとから実行する関数の本体から参照している変数の値を「関数を定義した時点の値」としておきたいのはよくあること（というかより自然な設計）と思います。そうするにはその変数が「関数を定義した時点の値となる」ようにクロージャーを用いるとよいと思います。

リスト1を次のように書き換えるとそうできます。

リスト3

functions = []
def closure_of_f(i):
    def f():
        return i  # (B)このiはclosure_of_fの引数である
    return f

for j in range(2):
    functions.append(closure_of_f(j))  # (C)引数(=i)の値はjのこの時点での値により確定
print([f() for f in functions])
# => [0, 1]

closure_of_f関数は引数に指定した値が変数iの値となるような環境での関数fを返してくれます。
少々わかりにくいですがclosure_of_fという関数を定義せずともlambdaを用いると次のようにも書けます。

リスト4

def g(i):
    return i

functions = []
for i in range(2):
    functions.append((lambda i: lambda: g(i))(i))
print([f() for f in functions])
# => [0, 1]

リスト3, 4どちらがよいか(あるいは根本的にもっと別の方法がよいか)は意見が分かれるかもしれません。

Comments

[sdice]

丁寧に問題になる理由や、修正方法、今後理解を進めた方が良いところも解説して頂き、本当にありがとうございました。理由や着眼点を理解することができました。スコープ、クロージャを再度確認致します。