threading --- スレッドベースの並列処理

ソースコード: Lib/threading.py

このモジュールでは、高水準のスレッドインターフェースをより低水準 な _thread モジュールの上に構築しています。

利用可能な環境: not WASI.

このモジュールは WebAssembly では動作しないか、利用不可です。詳しくは、WebAssembly プラットフォーム を見てください。

はじめに

threading モジュールを使うと、1 つのプロセス内で複数の スレッド (プロセスより小さな実行単位) を並行して実行できます。 スレッドの作成と管理ができ、メモリ空間を共有しながらタスクを並列に実行できます。 スレッドは、ファイル操作やネットワークリクエストなど、外部リソースを待つ時間が大部分を占める I/O バウンドのタスクで特に役立ちます。

threading の典型的な用途には、複数のタスクを並行して処理するワーカースレッドのプールの管理があります。 以下は Thread を使ってスレッドを作成し、開始する基本的な例です。

import threading
import time

def crawl(link, delay=3):
    print(f"crawl started for {link}")
    time.sleep(delay)  # ブロックする I/O (ネットワークリクエストを模擬)
    print(f"crawl ended for {link}")

links = [
    "https://python.org",
    "https://docs.python.org",
    "https://peps.python.org",
]

# 各リンクのスレッドを開始する
threads = []
for link in links:
    # 位置引数は args、キーワード引数は kwargs で渡す
    t = threading.Thread(target=crawl, args=(link,), kwargs={"delay": 2})
    threads.append(t)

# 各スレッドを開始する
for t in threads:
    t.start()

# すべてのスレッドの終了を待つ
for t in threads:
    t.join()

バージョン 3.7 で変更: このモジュールは以前はオプションでしたが、常に利用可能なモジュールとなりました。

参考

concurrent.futures.ThreadPoolExecutor は、呼び出し側のスレッドをブロックせずにバックグラウンドのスレッドへタスクを渡し、必要なときに結果を取得できる、より高水準のインターフェースを提供します。

queue は、実行中のスレッド間でデータを交換するためのスレッドセーフなインターフェースを提供します。

asyncio は、オペレーティングシステムのスレッドを複数使わずに、タスクレベルの並行処理を実現する別の方法を提供します。

注釈

Python 2.x 系では、このモジュールの一部のメソッドや関数に camelCase 形式の名前がありました。 これらは Python 3.10 以降では非推奨ですが、Python 2.5 以前との互換性のため、引き続きサポートされています。

CPython は Global Interpreter Lock のため、ある時点で Python コードを実行できるスレッドは1つに限られます (ただし、いくつかのパフォーマンスが強く求められるライブラリはこの制限を克服しています)。アプリケーションにマルチコアマシンの計算能力をより良く利用させたい場合は、 multiprocessing モジュールや concurrent.futures.ProcessPoolExecutor の利用をお勧めします。 ただし、I/Oバウンドなタスクを並行して複数走らせたい場合においては、 マルチスレッドは正しい選択肢です。

GIL とパフォーマンスに関する考慮事項

別々のプロセスを使って グローバルインタープリタロック (GIL) を回避する multiprocessing モジュールとは異なり、threading モジュールは 1 つのプロセス内で動作するため、すべてのスレッドが同じメモリ空間を共有します。 ただし、一度に Python バイトコードを実行できるのは 1 つのスレッドだけなので、CPU バウンドのタスクでは GIL によってスレッドによる性能向上が制限されます。 それでも、多くの場面でスレッドは並行処理を実現する有用な手段です。

Python 3.13 以降では、フリースレッド ビルドで GIL を無効にし、スレッドを真に並列に実行できます。 ただし、この機能はデフォルトでは利用できません (PEP 703 を参照)。

リファレンス

このモジュールは以下の関数を定義しています:

threading.active_count()

生存中の Thread オブジェクトの数を返します。この数は enumerate() の返すリストの長さと同じです。

関数 activeCount はこの関数の非推奨エイリアスです。

threading.current_thread()

呼び出し元の実行スレッドに対応する、現在の Thread オブジェクトを返します。 呼び出し元のスレッドが threading モジュールを使って作成されたものでなければ、機能が制限されたダミーのスレッドオブジェクトを返します。

関数 currentThread はこの関数の非推奨エイリアスです。

threading.excepthook(args, /)

Thread.run() で発生したキャッチされない例外を処理する。

(実) 引数*args*は以下の属性をもちます:

  • exc_type: 例外の型

  • exc_value: 例外の値、None の可能性がある。

  • exc_traceback: 例外のトレースバック。 None の場合もあります。

  • thread: 例外を送出したスレッド。 None の場合もあります。

exc_type が SystemExit の場合は、何も表示せずに例外を無視します。 それ以外の場合は、sys.stderr に例外を出力します。

この関数が例外を送出した場合は、その例外を処理するために sys.excepthook() が呼び出されます。

threading.excepthook() を置き換えることで、Thread.run() から送出されて捕捉されなかった例外の処理方法を制御できます。

独自のフックで exc_value を保存すると、循環参照ができることがあります。 例外が不要になったときに明示的に消去して、循環参照を断ち切ってください。

独自のフックで thread を保存した際、その値がファイナライズ中のオブジェクトである場合は、そのオブジェクトを復活させる可能性があります。 オブジェクトの復活を避けるため、独自のフックの完了後まで thread を保存しないでください。

参考

sys.excepthook() は捕捉されなかった例外を処理します。

Added in version 3.8.

threading.__excepthook__

threading.excepthook() の元の値を保持します。 壊れたオブジェクトや別のオブジェクトに置き換えられた場合に元の値を復元できるよう、保存されています。

Added in version 3.10.

threading.get_ident()

現在のスレッドの 'スレッドID' を返します。非ゼロの整数です。この値は直接の意味を持っていません; 例えばスレッド特有のデータの辞書に索引をつけるためのような、マジッククッキーとして意図されています。スレッドが終了し、他のスレッドが作られたとき、スレッド ID は再利用されるかもしれません。

Added in version 3.3.

threading.get_native_id()

カーネルが割り当てた、現在のスレッドのネイティブなスレッド ID を返します。 これは非負の整数です。 この値で、システム全体においてこのスレッドを一意に識別できます (スレッドが終了するまでで、その後は OS が値を再利用することがあります)。

利用可能な環境: Windows, FreeBSD, Linux, macOS, OpenBSD, NetBSD, AIX, DragonFlyBSD, GNU/kFreeBSD, Solaris.

Added in version 3.8.

バージョン 3.13 で変更: GNU/kFreeBSD のサポートが追加されました。

バージョン 3.15 で変更: Solaris のサポートが追加されました。

threading.enumerate()

現在、アクティブな Thread オブジェクト全てのリストを返します。リストには、デーモンスレッド (daemonic thread)、 current_thread() の生成するダミースレッドオブジェクトが入ります。終了したスレッドとまだ開始していないスレッドは入りません。しかし、主スレッドは、たとえ終了しても、常に結果に含まれます。

threading.main_thread()

main Thread オブジェクトを返します。通常の条件では、メインスレッドはPythonインタプリタが起動したスレッドを指します。

Added in version 3.4.

threading.settrace(func)

threading モジュールから開始するすべてのスレッドにトレース関数を設定します。 各スレッドの run() メソッドが呼ばれる前に、func が sys.settrace() に渡されます。

threading.settrace_all_threads(func)

threading モジュールから開始するすべてのスレッドと、現在実行中のすべての Python スレッドにトレース関数を設定します。

The func will be passed to sys.settrace() for each thread, before its run() method is called.

Added in version 3.12.

threading.gettrace()

settrace() 関数で設定したトレース関数を取得します。

Added in version 3.10.

threading.setprofile(func)

threading モジュールから開始するすべてのスレッドにプロファイル関数を設定します。 各スレッドの run() メソッドが呼ばれる前に、func が sys.setprofile() に渡されます。

threading.setprofile_all_threads(func)

threading モジュールから開始するすべてのスレッドと、現在実行中のすべての Python スレッドにプロファイル関数を設定します。

The func will be passed to sys.setprofile() for each thread, before its run() method is called.

Added in version 3.12.

threading.getprofile()

setprofile() 関数で設定したプロファイラ関数を取得します。

Added in version 3.10.

threading.stack_size([size])

新しいスレッドを作るときのスレッドスタックサイズを返します。オプションの size 引数にはこれ以降に作成するスレッドのスタックサイズを指定し、0 (プラットフォームのデフォルト値または設定されたデフォルト値) か、 32,768 (32 KiB) 以上の正の整数でなければなりません。size が指定されない場合 0 が使われます。スレッドのスタックサイズの変更がサポートされていない場合、 RuntimeError を送出します。不正なスタックサイズが指定された場合、 ValueError を送出して、スタックサイズは変更されません。32 KiB は現在のインタープリタ自身のために十分であると保証された最小のスタックサイズです。いくつかのプラットフォームではスタックサイズに対して制限があることに注意してください。例えば最小のスタックサイズが 32 KiB より大きかったり、システムのメモリページサイズ の整数倍の必要があるなどです。この制限についてはプラットフォームのドキュメントを参照してください (一般的なページサイズは 4 KiB なので、プラットフォームに関する情報がない場合は 4096 の整数倍のスタックサイズを選ぶといいかもしれません)。

利用可能な環境: Windows, pthreads.

POSIX スレッドをサポートする Unix プラットフォーム。

このモジュールでは以下の定数も定義しています:

threading.TIMEOUT_MAX

ブロックする関数 (Lock.acquire()、RLock.acquire()、Condition.wait() など) の timeout 引数に指定できる最大値です。 この値を超えるタイムアウトを指定すると OverflowError が送出されます。

Added in version 3.2.

このモジュールは多くのクラスを定義しています。それらは下記のセクションで詳しく説明されます。

このモジュールのおおまかな設計は Java のスレッドモデルに基づいています。とはいえ、 Java がロックと条件変数を全てのオブジェクトの基本的な挙動にしているのに対し、 Python ではこれらを別個のオブジェクトに分けています。 Python の Thread クラスがサポートしているのは Java の Thread クラスの挙動のサブセットにすぎません; 現状では、優先度 (priority)やスレッドグループがなく、スレッドの破壊 (destroy)、中断 (stop)、一時停止 (suspend)、復帰 (resume)、割り込み (interrupt) は行えません。 Java の Thread クラスにおける静的メソッドに対応する機能が実装されている場合にはモジュールレベルの関数になっています。

以下に説明するメソッドは全て原子的 (atomic) に実行されます。

スレッドローカルデータ

スレッドローカルデータは、スレッドごとに固有の値を持つデータです。 スレッド内だけで使いたいデータがある場合は、local オブジェクトを作成し、その属性を使ってください。

>>> mydata = local()
>>> mydata.number = 42
>>> mydata.number
42

local オブジェクトの辞書にもアクセスできます。

>>> mydata.__dict__
{'number': 42}
>>> mydata.__dict__.setdefault('widgets', [])
[]
>>> mydata.widgets
[]

別のスレッドからデータにアクセスすると、次のようになります。

>>> log = []
>>> def f():
...     items = sorted(mydata.__dict__.items())
...     log.append(items)
...     mydata.number = 11
...     log.append(mydata.number)

>>> import threading
>>> thread = threading.Thread(target=f)
>>> thread.start()
>>> thread.join()
>>> log
[[], 11]

異なるデータが得られます。 また、別のスレッドで変更しても、このスレッドから見えるデータには影響しません。

>>> mydata.number
42

__dict__ 属性を含め、local オブジェクトから取得した値は、その属性を読み出したときのスレッドのものです。 値は取得元のスレッドだけに属するため、一般にはスレッドをまたいで保存しない方がよいでしょう。

local クラスを継承して、独自の local オブジェクトを作成できます。

>>> class MyLocal(local):
...     number = 2
...     def __init__(self, /, **kw):
...         self.__dict__.update(kw)
...     def squared(self):
...         return self.number ** 2

これはデフォルト値、メソッド、初期化をサポートするのに便利です。 __init__() メソッドを定義すると、各スレッドで local オブジェクトを初めて使うときに呼び出されることに注意してください。 各スレッドの辞書を初期化するために必要な動作です。

この local オブジェクトを作成すると、次のようになります。

>>> mydata = MyLocal(color='red')

number にはデフォルト値があります。

>>> mydata.number
2

color には初期値があります。

>>> mydata.color
'red'
>>> del mydata.color

データを操作するメソッドもあります。

>>> mydata.squared()
4

先ほどと同じように、別のスレッドからデータにアクセスできます。

>>> log = []
>>> thread = threading.Thread(target=f)
>>> thread.start()
>>> thread.join()
>>> log
[[('color', 'red')], 11]

このスレッドのデータには影響しません。

>>> mydata.number
2
>>> mydata.color
Traceback (most recent call last):
...
AttributeError: 'MyLocal' object has no attribute 'color'

サブクラスでは __slots__ を定義できますが、スロットはスレッドローカルではありません。 スレッド間で共有されます。

>>> class MyLocal(local):
...     __slots__ = 'number'

>>> mydata = MyLocal()
>>> mydata.number = 42
>>> mydata.color = 'red'

そのため、別のスレッドを実行すると、次のようになります。

>>> thread = threading.Thread(target=f)
>>> thread.start()
>>> thread.join()

このスレッドから見える値にも影響します。

>>> mydata.number
11
class threading.local

スレッドローカルデータを表現するクラス。

Thread オブジェクト

Thread クラスは、別の実行スレッドで行う処理を表します。 処理を指定する方法は 2 つあります。 呼び出し可能オブジェクトをコンストラクタに渡すか、サブクラスで run() メソッドをオーバーライドします。 サブクラスでは、コンストラクタを除く他のメソッドをオーバーライドしてはいけません。 つまり、このクラスの __init__() と run() メソッド だけ をオーバーライドしてください。

ひとたびスレッドオブジェクトを生成すると、スレッドの start() メソッドを呼び出して活動を開始しなければなりません。 これにより、別のスレッドでそのオブジェクトの run() メソッドが呼び出されます。

スレッドの活動が始まると、スレッドは '生存中 (alive)' とみなされます。 スレッドは、通常 run() メソッドが終了するまで、もしくは捕捉されない例外が送出されるまで生存中となります。 is_alive() メソッドは、スレッドが生存中であるかどうか調べます。

スレッドは他のスレッドの join() メソッドを呼び出すことができます。このメソッドは、 join() メソッドを呼ばれたスレッドが終了するまでメソッドの呼び出し元のスレッドをブロックします。

スレッドは名前を持っています。名前はコンストラクタに渡すことができ、 name 属性を通して読み出したり変更したりできます。

run() メソッドが例外を発生させた場合、threading.excepthook() が呼び出され、例外を処理します。デフォルトでは、threading.excepthook() は SystemExit を黙殺します。

スレッドには "デーモンスレッド (daemon thread)" であるというフラグを立てられます。 このフラグには、残っているスレッドがデーモンスレッドだけになった時に Python プログラム全体を終了させるという意味があります。フラグの初期値はスレッドを生成したスレッドから継承します。フラグの値は daemon プロパティまたは daemon コンストラクタ引数を通して設定できます。

注釈

デーモンスレッドは終了時にいきなり停止されます。デーモンスレッドで使われたリソース (開いているファイル、データベースのトランザクションなど) は適切に解放されないかもしれません。きちんと (gracefully) スレッドを停止したい場合は、スレッドを非デーモンスレッドにして、Event のような適切なシグナル送信機構を使用してください。

スレッドには "主スレッド (main thread)" オブジェクトがあります。主スレッドは Python プログラムを最初に制御していたスレッドです。主スレッドはデーモンスレッドではありません。

"ダミースレッドオブジェクト (dummy thread objects)" が作成される場合があります。ダミースレッドは、 "外来スレッド (alien thread)" に相当するスレッドオブジェクトです。ダミースレッドは、C コードから直接生成されたスレッドのような、 threading モジュールの外で開始された処理スレッドです。ダミースレッドオブジェクトには限られた機能しかなく、常に生存中、かつデーモンスレッドであるとみなされ、 join できません。また、外来スレッドの終了を検出するのは不可能なので、ダミースレッドは削除できません。

class threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None, context=None)

コンストラクタは常にキーワード引数を使って呼び出さなければなりません。各引数は以下の通りです:

group は ThreadGroup クラスが実装されたときの将来の拡張用に予約されているため、None でなければなりません。

target は run() メソッドによって起動される呼び出し可能オブジェクトです。デフォルトでは何も呼び出さないことを示す None になっています。

name*はスレッド名です。デフォルトでは、"Thread-*N"(N*は小さな10進数)、または*target*引数が指定された場合"Thread-*N (target)"("target"は target.__name__)という形式でユニークな名前が構成される。

args は target を呼び出すときの引数のリストかタプルです。デフォルトは () です。

kwargs は target を呼び出すときのキーワード引数の辞書です。デフォルトは {} です。

None でない場合、daemon はスレッドがデーモンかどうかを明示的に設定します。None の場合 (デフォルト)、デーモン属性は現在のスレッドから継承されます。

context はスレッドの開始時に使う Context の値です。 デフォルト値は None で、sys.flags.thread_inherit_context フラグが動作を制御します。 フラグが真の場合、スレッドは start() の呼び出し元のコンテキストのコピーで開始します。 偽の場合は空のコンテキストで開始します。 空のコンテキストで開始することを明示するには、新しい Context() インスタンスを渡してください。 現在のコンテキストのコピーで開始することを明示するには、copy_context() の戻り値を渡してください。 このフラグのデフォルトは、フリースレッドビルドでは真、それ以外では偽です。

サブクラスでコンストラクタをオーバーライドする場合は、そのスレッドオブジェクトに対して他の処理を行う前に、必ず基底クラスのコンストラクタ (Thread.__init__()) を呼び出してください。

バージョン 3.3 で変更: daemon 引数が追加されました。

バージョン 3.10 で変更: name 引数が省略された場合、target 名を使用します。

バージョン 3.14 で変更: context パラメータを追加しました。

start()

スレッドの活動を開始します。

このメソッドは、スレッドオブジェクトあたり一度しか呼び出してはなりません。 start() は、オブジェクトの run() メソッドが個別の処理スレッド中で呼び出されるように調整します。

同じスレッドオブジェクトに対し、このメソッドを2回以上呼び出した場合、 RuntimeError を送出します。

サポートされている場合、オペレーティングシステム上のスレッド名を threading.Thread.name に設定します。 名前は、オペレーティングシステムのスレッド名の長さ制限に応じて切り詰められることがあります。

バージョン 3.14 で変更: オペレーティングシステム上のスレッド名を設定するようになりました。

run()

スレッドの活動をもたらすメソッドです。

このメソッドはサブクラスでオーバライドできます。標準の run() メソッドでは、オブジェクトのコンストラクタの target 引数に呼び出し可能オブジェクトを指定した場合、 args および kwargs の位置引数およびキーワード引数とともに呼び出します。

Thread に渡される args の引数にリストやタプルを使っても、同じ効果が得られます。

以下はプログラム例です:

>>> from threading import Thread
>>> t = Thread(target=print, args=[1])
>>> t.run()
1
>>> t = Thread(target=print, args=(1,))
>>> t.run()
1
join(timeout=None)

スレッドが終了するまで待機します。 このメソッドは、 join() を呼ばれたスレッドが正常終了あるいは処理されない例外によって終了するか、オプションのタイムアウトが発生するまで、メソッドの呼び出し元のスレッドをブロックします。

timeout 引数を指定し、その値が None でない場合は、操作のタイムアウトを秒単位で表す実数 (小数も可) にしてください。 join() は常に None を返すため、タイムアウトしたかどうかは join() の後に is_alive() を呼び出して判断する必要があります。 スレッドがまだ生存していれば、join() はタイムアウトしています。

timeout が指定されないかまたは None であるときは、この操作はスレッドが終了するまでブロックします。

1つのスレッドは何回も join されることができます。

現在のスレッドに対して join() を呼び出そうとすると、デッドロックを引き起こすため RuntimeError が送出されます。 スレッドが開始される前に join() を呼び出すことも同様のエラーのため、同じ例外が送出されます。

Python のファイナライズ の最終段階で、実行中のデーモンスレッドを join しようとすると、join() は PythonFinalizationError を送出します。

バージョン 3.14 で変更: PythonFinalizationError を送出する場合があります。

バージョン 3.15 で変更: timeout に、整数と浮動小数点数だけでなく任意の実数を受け付けるようになりました。

name

識別のためにのみ用いられる文字列です。名前には機能上の意味づけ (semantics) はありません。複数のスレッドに同じ名前をつけてもかまいません。名前の初期値はコンストラクタで設定されます。

一部のプラットフォームでは、スレッドの開始時にオペレーティングシステム上のスレッド名も設定され、タスクマネージャに表示されます。 この名前は、システム固有の制限 (例えば Linux では 15 バイト、macOS では 63 バイト) に収まるように切り詰められることがあります。

name の変更が OS に反映されるのは、現在実行しているスレッド自身の名前を変更した場合だけです。 別のスレッドの name 属性を設定しても、Python の Thread オブジェクトだけが更新されます。

getName()
setName()

name に対する非推奨 getter/setter API; 代わりにプロパティを直接使用してください。

バージョン 3.10 で非推奨.

ident

'スレッド識別子' 、または、スレッドが開始されていなければ None です。非ゼロの整数です。 get_ident() 関数を参照下さい。スレッド識別子は、スレッドが終了した後、新たなスレッドが生成された場合、再利用され得ます。スレッド識別子は、スレッドが終了した後でも利用できます。

native_id

OS (カーネル) が割り当てた、このスレッドのスレッド ID (TID) です。 非負の整数で、スレッドがまだ開始されていない場合は None です。 get_native_id() を参照してください。 この値で、システム全体においてこのスレッドを一意に識別できます (スレッドが終了するまでで、その後は OS が値を再利用することがあります)。

注釈

プロセス ID と同様に、スレッド ID が有効である (システム全体で一意であると保証される) のは、スレッドが作成されてから終了するまでの間だけです。

利用可能な環境: Windows, FreeBSD, Linux, macOS, OpenBSD, NetBSD, AIX, DragonFlyBSD.

Added in version 3.8.

is_alive()

スレッドが生存中かどうかを返します。

このメソッドは、 run() メソッドが起動する直前から run() メソッドが終了する直後までの間 True を返します。モジュール関数 enumerate() は、全ての生存中のスレッドのリストを返します。

daemon

このスレッドがデーモンスレッドか (True) か否か (False) を示すブール値。この値は start() の呼び出し前に設定されなければなりません。さもなければ RuntimeError が送出されます。初期値は生成側のスレッドから継承されます; メインスレッドはデーモンスレッドではないので、メインスレッドで作成されたすべてのスレッドは、デフォルトで daemon = False になります。

デーモンでない生存中のスレッドが全てなくなると、 Python プログラム全体が終了します。

isDaemon()
setDaemon()

daemon に対する非推奨 getter/setter API; 代わりにプロパティを直接使用してください。

バージョン 3.10 で非推奨.

Lock オブジェクト

プリミティブロックとは、ロックが生じた際に特定のスレッドによって所有されない同期プリミティブです。 Python では現在のところ拡張モジュール _thread で直接実装されている最も低水準の同期プリミティブを使えます。

プリミティブロックは2つの状態、 "ロック" または "アンロック" があります。ロックはアンロック状態で作成されます。ロックには基本となる二つのメソッド、 acquire() と release() があります。ロックの状態がアンロックである場合、 acquire() は状態をロックに変更して即座に処理を戻します。 状態がロックの場合、 acquire() は他のスレッドが release() を呼び出してロックの状態をアンロックに変更するまでブロックします。その後、 acquire() 呼び出しは状態を再度ロックに設定してから処理を戻します。 release() メソッドを呼び出すのはロック状態のときでなければなりません; このメソッドはロックの状態をアンロックに変更して、即座に処理を戻します。 アンロックの状態のロックを解放しようとすると RuntimeError が送出されます。

ロックは コンテキストマネージメントプロトコル もサポートします。

複数のスレッドにおいて acquire() がアンロック状態への遷移を待っているためにブロックが起きている時に release() を呼び出してロックの状態をアンロックにすると、一つのスレッドだけが処理を進行できます。 どのスレッドが処理を進行できるのかは定義されておらず、実装によって異なるかもしれません。

全てのメソッドはアトミックに実行されます。

class threading.Lock

プリミティブロック (primitive lock) オブジェクトを実装しているクラスです。スレッドが一度ロックを獲得すると、それ以後のロック獲得の試みはロックが解放されるまでブロックします。どのスレッドでもロックを解放できます。

バージョン 3.13 で変更: Lock はクラスになりました。 以前の Python では、Lock は内部の非公開ロック型のインスタンスを返すファクトリ関数でした。

acquire(blocking=True, timeout=-1)

ブロックあり、またはブロックなしでロックを獲得します。

引数 blocking を True (デフォルト) に設定して呼び出した場合、ロックがアンロック状態になるまでブロックします。そしてそれをロック状態にしてから True を返します。

引数 blocking の値を False にして呼び出すとブロックしません。blocking を True にして呼び出した場合にブロックするような状況では、直ちに False を返します。それ以外の場合には、ロックをロック状態にして True を返します。

timeout 引数に正の値を指定して呼び出すと、ロックを獲得できない間、最大で timeout 秒間ブロックします。 timeout に -1 を指定すると、時間の制限なく待機します。 blocking が False の場合に timeout を指定することは禁止されています。

ロックを獲得すると True を、ロックを獲得できなかったとき (例えば timeout が過ぎた場合) には False を返します。

バージョン 3.2 で変更: 新しい timeout 引数。

バージョン 3.2 で変更: POSIX で、内部のスレッド実装がサポートしていれば、ロックの獲得をシグナルで中断できるようになりました。

バージョン 3.14 で変更: Windows でロックの獲得をシグナルで中断できるようになりました。

バージョン 3.15 で変更: timeout に、整数と浮動小数点数だけでなく任意の実数を受け付けるようになりました。

release()

ロックを解放します。これはロックを獲得したスレッドだけでなく、任意のスレッドから呼ぶことができます。

ロックの状態がロックのとき、状態をアンロックにリセットして処理を戻します。他のスレッドがロックがアンロック状態になるのを待ってブロックしている場合、ただ一つのスレッドだけが処理を継続できるようにします。

アンロック状態のロックに対して呼び出された場合、RuntimeError が送出されます。

戻り値はありません。

locked()

ロック状態のとき True を返します。

RLock オブジェクト

再入可能ロック (reentrant lock) とは、同じスレッドが複数回獲得できるような同期プリミティブです。再入可能ロックの内部では、プリミティブロックの使うロック／アンロック状態に加え、 "所有スレッド (owning thread)" と "再帰レベル (recursion level)" という概念を用いています。ロック状態では何らかのスレッドがロックを所有しており、アンロック状態ではいかなるスレッドもロックを所有していません。

Threads call a lock's acquire() method to lock it, and its release() method to unlock it.

注釈

再入可能ロックは コンテキスト管理プロトコル をサポートしています。 そのため、コードブロックでのロックの獲得と解放には、acquire() と release() を手動で呼び出すよりも with を使うことを推奨します。

RLock's acquire()/release() call pairs may be nested, unlike Lock's acquire()/release(). Only the final release() (the release() of the outermost pair) resets the lock to an unlocked state and allows another thread blocked in acquire() to proceed.

acquire() と release() は対にして使わなければなりません。 ロックを獲得するたびに、獲得したスレッドで解放する必要があります。 獲得した回数と同じ回数だけ解放しないと、デッドロックにつながる可能性があります。

class threading.RLock

このクラスは再入可能ロックオブジェクトを実装します。再入可能ロックはそれを獲得したスレッドによって解放されなければなりません。いったんスレッドが再入可能ロックを獲得すると、同じスレッドはブロックされずにもう一度それを獲得できます ; そのスレッドは獲得した回数だけ解放しなければいけません。

RLock は実際にはファクトリ関数で、プラットフォームでサポートされる最も効率的なバージョンの具体的な RLock クラスのインスタンスを返すことに注意してください。

acquire(blocking=True, timeout=-1)

ブロックあり、またはブロックなしでロックを獲得します。

参考

RLock をコンテキストマネージャとして使う

実用上可能なら、acquire() と release() を手動で呼び出すよりも推奨されます。

blocking 引数を True (デフォルト) にして呼び出した場合:

  • どのスレッドもロックを所有していなければ、ロックを獲得して直ちに戻ります。

  • 別のスレッドがロックを所有している場合は、ロックを獲得できるようになるまでブロックします。 timeout に正の値が指定されていれば、その時間が経過した場合にも待機を終了します。

  • 同じスレッドがロックを所有している場合は、再びロックを獲得して直ちに戻ります。 この点が Lock と RLock の違いです。 Lock はこの場合も前項と同様に扱い、ロックを獲得できるまでブロックします。

blocking 引数を False にして呼び出した場合:

  • どのスレッドもロックを所有していなければ、ロックを獲得して直ちに戻ります。

  • 別のスレッドがロックを所有している場合は、直ちに戻ります。

  • 同じスレッドがロックを所有している場合は、再びロックを獲得して直ちに戻ります。

いずれの場合も、スレッドがロックを獲得できれば True を返します。 獲得できなければ (ブロックしない場合やタイムアウトに達した場合など)、False を返します。

複数回呼び出した場合、同じ回数だけ release() を呼び出さないと、デッドロックにつながる可能性があります。 acquire/release を直接呼び出す代わりに、RLock をコンテキストマネージャとして使うことを検討してください。

バージョン 3.2 で変更: 新しい timeout 引数。

バージョン 3.15 で変更: timeout に、整数と浮動小数点数だけでなく任意の実数を受け付けるようになりました。

release()

再帰レベルをデクリメントしてロックを解放します。デクリメント後に再帰レベルがゼロになった場合、ロックの状態をアンロック (いかなるスレッドにも所有されていない状態) にリセットし、ロックの状態がアンロックになるのを待ってブロックしているスレッドがある場合にはその中のただ一つだけが処理を進行できるようにします。デクリメント後も再帰レベルがゼロでない場合、ロックの状態はロックのままで、呼び出し側のスレッドに所有されたままになります。

このメソッドは、呼び出し元のスレッドがロックを所有している場合にだけ呼び出してください。 ロックを獲得していないときに呼び出すと、RuntimeError を送出します。

戻り値はありません。

locked()

このオブジェクトが現在ロックされているかどうかを示すブール値を返します。

Added in version 3.14.

Condition オブジェクト

条件変数 (condition variable) は、常にある種のロックに関連付けられています; このロックは明示的に渡すことも、デフォルトで生成させることもできます。複数の条件変数で同じロックを共有しなければならない場合には、引渡しによる関連付けが便利です。ロックは条件オブジェクトの一部です: それを別々に扱う必要はありません。

条件変数は コンテキスト管理プロトコル に従います: with 文を使って囲まれたブロックの間だけ関連付けられたロックを獲得することができます。 acquire() メソッドと release() メソッドは、さらに関連付けられたロックの対応するメソッドを呼び出します。

他のメソッドは、関連付けられたロックを保持した状態で呼び出さなければなりません。 wait() メソッドはロックを解放します。そして別のスレッドが notify() または notify_all() を呼ぶことによってスレッドを起こすまでブロックします。一旦起こされたなら、 wait() は再びロックを得て戻ります。タイムアウトを指定することも可能です。

notify() メソッドは条件変数待ちのスレッドを1つ起こします。 notify_all() メソッドは条件変数待ちの全てのスレッドを起こします。

注意: notify() と notify_all() はロックを解放しません; 従って、スレッドが起こされたとき、 wait() の呼び出しは即座に処理を戻すわけではなく、 notify() または notify_all() を呼び出したスレッドが最終的にロックの所有権を放棄したときに初めて処理を返すのです。

条件変数を使う典型的なプログラミングスタイルでは、何らかの共有された状態変数へのアクセスを同期させるためにロックを使います; 状態変数が特定の状態に変化したことを知りたいスレッドは、自分の望む状態になるまで繰り返し wait() を呼び出します。その一方で、状態変更を行うスレッドは、前者のスレッドが待ち望んでいる状態であるかもしれないような状態へ変更を行ったときに notify() や notify_all() を呼び出します。例えば、以下のコードは無制限のバッファ容量のときの一般的な生産者-消費者問題です:

# 要素を 1 つ消費する
with cv:
    while not an_item_is_available():
        cv.wait()
    get_an_available_item()

# 要素を 1 つ生成する
with cv:
    make_an_item_available()
    cv.notify()

アプリケーションの条件をチェックする while ループは必須です。なぜなら、 wait() が任意の長時間の後で返り、 notify() 呼び出しを促した条件がもはや真でないことがありえるからです。これはマルチスレッドプログラミングに固有です。条件チェックを自動化するために wait_for() メソッドを使うことができ、それはタイムアウトの計算を簡略化します:

# 要素を 1 つ消費する
with cv:
    cv.wait_for(an_item_is_available)
    get_an_available_item()

notify() と notify_all() のどちらを使うかは、その状態の変化に興味を持っている待ちスレッドが一つだけなのか、あるいは複数なのかで考えます。例えば、典型的な生産者-消費者問題では、バッファに1つの要素を加えた場合には消費者スレッドを 1 つしか起こさなくてかまいません。

class threading.Condition(lock=None)

このクラスは条件変数 (condition variable) オブジェクトを実装します。条件変数を使うと、1つ以上のスレッドを別のスレッドの通知があるまで待機させておけます。

lock に None でない値を指定した場合、その値は Lock または RLock オブジェクトでなければなりません。 この場合、 lock は根底にあるロックオブジェクトとして使われます。 それ以外の場合には、 RLock オブジェクトを新しく作成して使います。

バージョン 3.3 で変更: ファクトリ関数からクラスに変更されました。

acquire(*args)

根底にあるロックを獲得します。このメソッドは根底にあるロックの対応するメソッドを呼び出します。そのメソッドの戻り値を返します。

release()

根底にあるロックを解放します。このメソッドは根底にあるロックの対応するメソッドを呼び出します。戻り値はありません。

locked()

このオブジェクトが現在ロックされているかどうかを示すブール値を返します。

Added in version 3.14.

wait(timeout=None)

通知 (notify) を受けるか、タイムアウトするまで待機します。呼び出し側のスレッドがロックを獲得していないときにこのメソッドを呼び出すと RuntimeError が送出されます。

このメソッドは内部のロックを解放し、他のスレッドが同じ条件変数の notify() または notify_all() を呼び出して起こすか、指定されたタイムアウトが発生するまでブロックします。 起こされた場合もタイムアウトした場合も、ロックを再び獲得してから戻ります。

timeout 引数を指定し、その値が None でない場合は、操作のタイムアウトを秒単位で表す実数 (小数も可) にしてください。

根底にあるロックが RLock である場合、 release() メソッドではロックは解放されません。というのも、ロックが再帰的に複数回獲得されている場合には、 release() によって実際にアンロックが行われないかもしれないからです。その代わり、ロックが再帰的に複数回獲得されていても確実にアンロックを行える RLock クラスの内部インターフェースを使います。その後ロックを再獲得する時に、もう一つの内部インターフェースを使ってロックの再帰レベルを復帰します。

与えられた timeout が過ぎていなければ返り値は True です。タイムアウトした場合には False が返ります。

バージョン 3.2 で変更: 以前は、このメソッドは常に None を返していました。

wait_for(predicate, timeout=None)

条件が真と判定されるまで待ちます。 predicate は呼び出し可能オブジェクトでなければならず、その結果はブール値として解釈されます。 最大の待ち時間を指定する timeout を与えることができます。

このユーティリティメソッドは、述語が満たされるかタイムアウトが発生するまで wait() を繰り返し呼び出す場合があります。戻り値は述語の最後の戻り値で、もしメソッドがタイムアウトすれば、 False と評価されます。

タイムアウト機能を無視すれば、このメソッドの呼び出しは以下のように書くのとほぼ等価です:

while not predicate():
    cv.wait()

したがって、 wait() と同じルールが適用されます: 呼び出された時にロックを保持していなければならず、戻るときにロックが再度獲得されます。述語はロックを保持した状態で評価されます。

Added in version 3.2.

notify(n=1)

デフォルトで、この条件変数を待っている1つのスレッドを起こします。 呼び出し側のスレッドがロックを獲得していないときにこのメソッドを呼び出すと RuntimeError が送出されます。

この条件変数を待っているスレッドを最大 n 個起こします。 待機中のスレッドがなければ何もしません。

現在の実装では、少なくとも n スレッドが待機中であれば、ちょうど n スレッドを起こします。 とはいえ、この挙動に依存するのは安全ではありません。 将来、実装の最適化によって、n 個を超えるスレッドを起こすことがあるかもしれないからです。

注意: 起こされたスレッドは実際にロックを再獲得できるまで wait() 呼び出しから戻りません。 notify() はロックを解放しないので、 notify() 呼び出し側は明示的にロックを解放しなければなりません。

notify_all()

この条件を待っているすべてのスレッドを起こします。このメソッドは notify() のように動作しますが、 1 つではなくすべての待ちスレッドを起こします。呼び出し側のスレッドがロックを獲得していない場合、 RuntimeError が送出されます。

メソッド notifyAll は、このメソッドの非推奨エイリアスです。

Semaphore オブジェクト

セマフォ (semaphore) は、計算機科学史上最も古い同期プリミティブの一つ で、草創期のオランダ計算機科学者 Edsger W. Dijkstra によって発明されました (彼は acquire() と release() の代わりに P() と V() を使いました)。

セマフォは acquire() でデクリメントされ release() でインクリメントされるような内部カウンタを管理します。 カウンタは決してゼロより小さくはなりません; acquire() は、カウンタがゼロになっている場合、他のスレッドが release() を呼び出すまでブロックします。

セマフォは コンテキストマネージメントプロトコル もサポートします。

class threading.Semaphore(value=1)

このクラスはセマフォ (semaphore) オブジェクトを実装します。 セマフォは、release() を呼び出した回数から acquire() を呼び出した回数を引き、初期値を足した値を表すアトミックなカウンタを管理します。 acquire() メソッドは、カウンタの値を負にせずに処理を戻せるまで、必要に応じてブロックします。 value を指定しない場合、デフォルトの値は 1 になります。

オプションの引数には、内部カウンタの初期値を指定します。デフォルトは 1 です。与えられた value が 0 より小さい場合、 ValueError が送出されます。

バージョン 3.3 で変更: ファクトリ関数からクラスに変更されました。

acquire(blocking=True, timeout=None)

セマフォを獲得します。

引数なしで呼び出した場合:

  • 呼び出し時に内部カウンタが 0 より大きければ、1 減らして直ちに True を返します。

  • If the internal counter is zero on entry, block until awoken by a call to release(). Once awoken (and the counter is greater than 0), decrement the counter by 1 and return True. Exactly one thread will be awoken by each call to release(). The order in which threads are awoken should not be relied on.

blocking を False にして呼び出すとブロックしません。引数なしで呼び出した場合にブロックするような状況であった場合には直ちに False を返します。それ以外の場合には、引数なしで呼び出したときと同じ処理を行い True を返します。

None 以外の timeout で起動された場合、最大で timeout 秒ブロックします。 acquire が その間隔の間で完了しなかった場合は False が返ります。そうでなければ True が返ります。

バージョン 3.2 で変更: 新しい timeout 引数。

バージョン 3.15 で変更: timeout に、整数と浮動小数点数だけでなく任意の実数を受け付けるようになりました。

release(n=1)

内部カウンタを n インクリメントして、セマフォを解放します。 release() 処理に入ったときにカウンタがゼロであり、カウンタの値がゼロより大きくなるのを待っている別のスレッドがあった場合、それらのスレッドから n 個を起こします。

バージョン 3.9 で変更: 複数の待機中のスレッドを一度に解放する n パラメータを追加しました。

class threading.BoundedSemaphore(value=1)

有限セマフォ (bounded semaphore) オブジェクトを実装しているクラスです。 有限セマフォは、現在の値が初期値を超過しないようチェックを行います。 超過を起こした場合、 ValueError を送出します。 たいていの場合、セマフォは限られた容量のリソースを保護するために使われるものです。 従って、セマフォを必要以上の回数解放することはバグが生じているしるしです。 value を指定しない場合、デフォルトの値は 1 になります。

バージョン 3.3 で変更: ファクトリ関数からクラスに変更されました。

Semaphore の使用例

セマフォはしばしば、容量に限りのある資源、例えばデータベースサーバなどを保護するために使われます。リソースが固定の状況では、常に有限セマフォを使わなければなりません。主スレッドは、作業スレッドを立ち上げる前にセマフォを初期化します:

maxconnections = 5
# ...
pool_sema = BoundedSemaphore(value=maxconnections)

作業スレッドは、ひとたび立ち上がると、サーバへ接続する必要が生じたときにセマフォの acquire() および release() メソッドを呼び出します:

with pool_sema:
    conn = connectdb()
    try:
        # ... 接続を使う ...
    finally:
        conn.close()

有限セマフォを使うと、セマフォを獲得回数以上に解放してしまうというプログラム上の間違いを見逃しにくくします。

Event オブジェクト

イベントは、あるスレッドがイベントを発信し、他のスレッドはそれを待つという、スレッド間で通信を行うための最も単純なメカニズムの一つです。

イベントオブジェクトは内部フラグを管理します。このフラグは set() メソッドで値を true に、 clear() メソッドで値を false にリセットします。 wait() メソッドはフラグが true になるまでブロックします。

class threading.Event

イベントオブジェクトを実装しているクラスです。 イベントは set() メソッドを使うと True に、 clear() メソッドを使うと False にセットされるようなフラグを管理します。 wait() メソッドは、そのフラグが true になるまでブロックするようになっています。 フラグの初期値は false です。

バージョン 3.3 で変更: ファクトリ関数からクラスに変更されました。

is_set()

内部フラグが真のとき True を返します。

メソッド isSet はこのメソッドの非推奨エイリアスです。

set()

内部フラグの値を true にセットします。フラグの値が true になるのを待っている全てのスレッドを起こします。一旦フラグが true になると、スレッドが wait() を呼び出しても全くブロックしなくなります。

clear()

内部フラグの値を false にリセットします。以降は、 set() を呼び出して再び内部フラグの値を true にセットするまで、 wait() を呼び出したスレッドはブロックするようになります。

wait(timeout=None)

内部フラグが偽で、タイムアウトが指定されている場合はその時間が経過していない間、ブロックします。 戻り値は、このブロックするメソッドが処理を戻した理由を示します。 内部フラグが真になって戻った場合は True、タイムアウトが指定されており、その待ち時間内に内部フラグが真にならなかった場合は False です。

timeout 引数を指定し、その値が None でない場合は、操作のタイムアウトを秒単位で表す実数 (小数も可) にしてください。

バージョン 3.1 で変更: 以前は、このメソッドは常に None を返していました。

Timer オブジェクト

このクラスは、一定時間経過後に実行される活動、すなわちタイマ活動を表現します。 Timer は Thread のサブクラスであり、自作のスレッドを構築した一例でもあります。

タイマは、スレッドと同様に Timer.start メソッドを呼び出して開始します。 cancel() メソッドを呼び出すと、処理を開始する前であればタイマを停止できます。 処理の実行前にタイマが待つ時間は、指定された時間と厳密には一致しない場合があります。

例えば:

def hello():
    print("hello, world")

t = Timer(30.0, hello)
t.start()  # 30 秒後に "hello, world" が表示される
class threading.Timer(interval, function, args=None, kwargs=None)

interval 秒後に引数 args キーワード引数 kwargs で function を実行するようなタイマを生成します。args*が ``None`` (デフォルト) なら空のリストが使用されます。*kwargs が None (デフォルト) なら空の辞書が使用されます。

バージョン 3.3 で変更: ファクトリ関数からクラスに変更されました。

cancel()

タイマをストップして、その動作の実行をキャンセルします。このメソッドはタイマがまだ活動待ち状態にある場合にのみ動作します。

Barrier オブジェクト

Added in version 3.2.

このクラスは、互いを待つ必要のある固定の数のスレッドで使用するための単純な同期プリミティブを提供します。それぞれのスレッドは wait() メソッドを呼ぶことによりバリアを通ろうとしてブロックします。すべてのスレッドがそれぞれの wait() メソッドを呼び出した時点で、すべてのスレッドが同時に解放されます。

バリアは同じ数のスレッドに対して何度でも再利用することができます。

例として、クライアントとサーバの間でスレッドを同期させる単純な方法を紹介します:

b = Barrier(2, timeout=5)

def server():
    start_server()
    b.wait()
    while True:
        connection = accept_connection()
        process_server_connection(connection)

def client():
    b.wait()
    while True:
        connection = make_connection()
        process_client_connection(connection)
class threading.Barrier(parties, action=None, timeout=None)

parties 個のスレッドのためのバリアオブジェクトを作成します。 action を指定する場合は、スレッドの解放時に、そのうちの 1 つから呼び出される呼び出し可能オブジェクトにしてください。 timeout は、wait() メソッドでタイムアウトを指定しなかった場合に使うデフォルト値です。

wait(timeout=None)

バリアを通ります。バリアに対するすべてのスレッドがこの関数を呼んだ時に、それらは同時にすべて解放されます。timeout が提供される場合、それはクラスコンストラクタに渡された値に優先して使用されます。

返り値は 0 から parties -- 1 の範囲の整数で、それぞれのスレッドに対して異なります。これは、特別な後始末 (housekeeping) を行うスレッドを選択するために使用することができます。例えば:

i = barrier.wait()
if i == 0:
    # 1 つのスレッドだけが表示すればよい
    print("passed the barrier")

action がコンストラクタに渡されていれば、スレッドのうちの1つが解放される前にそれを呼び出します。万一この呼び出しでエラーが発生した場合、バリアは broken な状態に陥ります。

この呼び出しがタイムアウトする場合、バリアは broken な状態に陥ります。

スレッドが待っている間にバリアが broken になるかリセットされた場合、このメソッドは BrokenBarrierError 例外を送出するかもしれません。

reset()

バリアをデフォルトの空の状態に戻します。そのバリアの上で待っているすべてのスレッドは BrokenBarrierError 例外を受け取ります。

状態が未知の他のスレッドがある場合、この関数を使用するのに何らかの外部同期を必要とするかもしれないことに注意してください。バリアが broken な場合、単にそれをそのままにして新しいものを作成する方がよいでしょう。

abort()

バリアを broken な状態にします。これによって、現在または将来の wait() 呼び出しが BrokenBarrierError とともに失敗するようになります。これを使うと、例えばスレッドが異常終了する必要がある場合にアプリケーションがデッドロックするのを避けることができます。

スレッドのうちの1つが返ってこないことに対して自動的に保護するように、単純に常識的な timeout 値でバリアを作成することは望ましいかもしれません。

parties

バリアを通るために要求されるスレッドの数。

n_waiting

現在バリアの中で待っているスレッドの数。

broken

バリアが broken な状態である場合に True となるブール値。

exception threading.BrokenBarrierError

Barrier オブジェクトがリセットされるか broken な場合に、この例外 (RuntimeError のサブクラス) が送出されます。

with 文でのロック・条件変数・セマフォの使い方

このモジュールが提供するオブジェクトのうち、acquire メソッドと release メソッドを持つものはすべて、with 文のコンテキストマネージャとして使えます。 ブロックに入るときに acquire、ブロックを出るときに release が呼ばれます。 したがって、次のコードは

with some_lock:
    # 何らかの処理を行う...

は、以下と同じです

some_lock.acquire()
try:
    # 何らかの処理を行う...
finally:
    some_lock.release()

現在のところ、 Lock 、 RLock 、 Condition 、 Semaphore 、 BoundedSemaphore を with 文のコンテキストマネージャとして使うことができます。

イテレータの同期

デフォルトでは、Python のイテレータは並行アクセスをサポートしません。 ほとんどのイテレータは、複数のスレッドから同時にアクセスされた場合の動作を保証しません。 例えばジェネレータイテレータは、ジェネレータがすでに実行中であるときにイテレータのメソッドが呼び出されると、ValueError を送出します。 この節のツールを使うと、通常のイテレータやイテレータを生成する呼び出し可能オブジェクトに、確実に動作する並行アクセスのサポートを追加できます。

serialize_iterator ラッパーを使うと、複数のスレッドが 1 つのイテレータを共有し、順番に値を取り出せます。 あるスレッドが __next__() を実行している間、他のスレッドはイテレータが利用可能になるまでブロックします。 元のイテレータが生成する各値は、ちょうど 1 つの呼び出し元に渡されます。

concurrent_tee() 関数を使うと、複数のスレッドがそれぞれ、1 つの元のイテレータの値をすべて受け取れます。 同じ取得元から値を取り出す、独立したイテレータを作成します。 値は、作成されたすべてのイテレータが消費するまでバッファに保持されます。

class threading.serialize_iterator(iterable)

__next__() への並行した呼び出しを、ロックを使って 1 つずつ実行するイテレータラッパーを返します。

ラップしたイテレータに send()、throw()、close() も定義されている場合は、それらの呼び出しも 1 つずつ実行します。

これにより、ジェネレータイテレータを含め、1 つのイテレータを複数のスレッドで共有できます。 ロックによって、呼び出しは一度に 1 つずつ処理されます。 ラッパー自体が値を重複させたり飛ばしたりすることはありません。 元のイテレータの各要素は、ちょうど 1 つの呼び出し元に渡されます。

このラッパーは値をコピーせず、バッファにも保持しません。 別のスレッドがイテレータを進めているときに next() を呼び出すスレッドは、実行中の呼び出しが完了するまでブロックします。

例:

import threading

def squares(n):
    for x in range(n):
        yield x * x

def consume(name, iterable):
    for item in iterable:
        print(name, item)

source = threading.serialize_iterator(squares(5))

t1 = threading.Thread(target=consume, args=("left", source))
t2 = threading.Thread(target=consume, args=("right", source))
t1.start()
t2.start()
t1.join()
t2.join()

この例では、各数値はちょうど一度ずつ表示されますが、処理は 2 つのスレッドに分担されます。

Added in version 3.15.

threading.synchronized_iterator(func)

イテレータを生成する呼び出し可能オブジェクトをラップし、それが返す各イテレータを自動的に serialize_iterator に渡すようにします。

ジェネレータ関数の デコレータ として特に便利で、ジェネレータイテレータの値を複数のスレッドから取り出せるようにします。

例:

import threading

@threading.synchronized_iterator
def squares(n):
    for x in range(n):
        yield x * x

def consume(name, iterable):
    for item in iterable:
        print(name, item)

source = squares(5)

t1 = threading.Thread(target=consume, args=("left", source))
t2 = threading.Thread(target=consume, args=("right", source))
t1.start()
t2.start()
t1.join()
t2.join()

返されるラッパーは、名前やラップした関数への参照など、func のメタデータを保持します。

Added in version 3.15.

threading.concurrent_tee(iterable, n=2)

1 つの入力 iterable から n 個の独立したイテレータを返します。 作成されたイテレータから並行して値を取り出す場合の動作が保証されます。

この関数は itertools.tee() に似ていますが、別々のスレッドで動作する消費側に、元のイテレータから値を供給する場合を想定しています。 返される各イテレータは、元のイテラブルのすべての値を同じ順序で生成します。

内部では、作成されたすべてのイテレータが各値を消費するまで、その値をバッファに保持します。

返されるイテレータは、同じ同期用のロックを共有します。 作成された各イテレータは、一度に 1 つのスレッドから値を取り出すことを想定しています。 作成されたイテレータの 1 つを複数のスレッドで共有する必要がある場合は、serialize_iterator でラップしてください。

n が 0 の場合は空のタプルを返します。 n が負の場合は ValueError を送出します。

例:

import threading

def squares(n):
    for x in range(n):
        yield x * x

def consume(name, iterable):
    for item in iterable:
        print(name, item)

source = squares(5)
left, right = threading.concurrent_tee(source)

t1 = threading.Thread(target=consume, args=("left", left))
t2 = threading.Thread(target=consume, args=("right", right))
t1.start()
t2.start()
t1.join()
t2.join()

In this example, both consumer threads see the full sequence of squares from a single generator expression.

Added in version 3.15.