【Python】自作関数に『タイムアウト』を設定するコード例【Windows】

自作関数に『タイムアウト』を設定して実行する Python コード例を書きました。

Windows 10』で Python を使用しているときに、タイムアウトを設定するコード例です。

タイムアウト処理は、『wrapt-timeout-decorator』というライブラリを使用したら、簡単に設定することができました。ありがとうございます。

実際にタイムアウトを設定してみたところ、指定した秒数を超えたところで、TimeoutError が発生しました。

期待した通りの動作です。

自作関数の実行を、時間経過で打ち切ることができました。

ところで、タイムアウトを設定したら、関数の実行速度が少し遅くなりました。

タイムアウト処理を実現するために、新しいプロセスを開始したり終了したりする関係で、遅くなっていたようでした。

wrapt-timeout-decorator

wrapt-timeout-decorator は、Windows でも簡単に『タイムアウト』が設定できるようになる Python ライブラリでした。

公式サイトの場所です。

(PyPI) wrapt-timeout-decorator

(GitHub) wrapt_timeout_decorator: Python Powerful Timeout Decorator that can be used safely on classes, methods, class methods

Python マニュアル

コード例で使用した Python 機能のマニュアルの場所です。

(Python) 引数リストのアンパック * 演算子

(Python) classmethod datetime.now(tz=None) 現在の時刻

(Python) timedelta.total_seconds() トータルの秒数

(Python) os.getpid() 現在のプロセス id

(Python) os.getppid() 親プロセスのプロセス id

(Python) exception TimeoutError 組み込み例外

(Python) traceback.format_exception_only(etype, value) 例外メッセージの最後の行を取得する

(Python) class type(object) object の型を取得

(Python) instance.__class__ 特殊属性

(Python) definition.__name__ 特殊属性

(Python) str.rstrip([chars]) 末尾から指定した文字を除去

(Python) class multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

(Python) map(func, iterable[, chunksize]) マルチプロセス処理を実行する

(Python) time.sleep(secs) スリープ

コード例

自作関数に『タイムアウト (timeout)』を設定して実行する Python コード例です。

タイムアウトの設定手順です。

  1. 『タイムアウトを設定したい関数』を、別の Python ファイル (jisaku.py) に分けて書きます。
  2. デコレータで、タイムアウトの秒数を設定します。

以上です。

@wrapt_timeout_decorator.timeout(dec_timeout=秒数) で設定したタイムアウトは、『シングルプロセス処理』で実行したときも機能しましたし、『マルチプロセス処理』で実行したときにも機能しました。

メイン関数

メイン関数のコードです (main.py)。

自作関数は、別のファイルに書いて、import jisaku でインポートしました。

"""
自作関数に『タイムアウト』を設定して実行するコード例です。
(main.py)
"""

import datetime # デバッグ用
import os # デバッグ用
import traceback # デバッグ用
import multiprocessing
import jisaku


def main():
    """メイン関数"""
    print('start\n')

    # 使いたい引数のリストを作ります。
    args_list = [
        [1, 'あ'],
        [2, 'い'],
        [3, 'う'],
        [4, 'え'],
        [5, 'お'],
        ]

    print('実行中...')

    processes = 1 # シングルプロセス処理
    # processes = 3 # マルチプロセス処理

    if processes == 1:
        print(f'=== シングルプロセス処理 (processes={processes}) ===')
        results = []
        for args in args_list:
            result = proc(*args)
            results.append(result)
    elif processes >= 2:
        print(f'=== マルチプロセス処理 (processes={processes}) ===')
        with multiprocessing.Pool(processes=processes) as pool:
            results = pool.map(mp_proc_wrapper, args_list)

    print('\n結果を表示します。')
    for result in results:
        print(result)

    print('\nend')
    return


def mp_proc_wrapper(args):
    """
    (シングルプロセス処理の時は必要無いです)
    マルチプロセス処理の時に引数をアンパックして渡す関数。
    """
    return proc(*args)


def proc(number, text):
    """自作の処理"""

    result = None

    # (デバッグ) 現在の時刻を取得します。
    t1 = datetime.datetime.now()

    # (デバッグ) 親プロセスのプロセス ID を取得します。
    ppid = f'PPID={os.getppid()}' # the parent's process id.

    # (デバッグ) 現在のプロセスのプロセス ID を取得します。
    pid = f'PID={os.getpid()}' # the current process id.

    try:
        #『長い時間がかかるかもしれない関数』を実行します。
        result = jisaku.func(number, text)
    except TimeoutError as e:
        # タイムアウトしました。
        result = e.__class__.__name__
        print(traceback.format_exception_only(type(e), e)[0].rstrip('\n'))
        print(f'jisaku.func(number={number}, text="{text}") Error')
    else:
        # 何のエラーも起きませんでした。
        print(f'jisaku.func(number={number}, text="{text}") OK')

    # (デバッグ) 経過時間を計算します。
    elapsed_time = (datetime.datetime.now() - t1).total_seconds()
    total_seconds = '%.3f 秒' % elapsed_time

    return (number, text, total_seconds, ppid, pid, result)


if __name__ == "__main__":
    main()

タイムアウトを設定した関数

タイムアウトを設定した関数のコードです (jisaku.py)。

自作関数は、メイン関数とは別の Python ファイルに書きます。

"""
タイムアウト付きの『自作関数』を書いたモジュールです。
ちょっと面倒ですが、別のファイルに書きます。
(jisaku.py)
"""

import time # デバッグ用
import wrapt_timeout_decorator

# デコレータでタイムアウトの秒数を設定します
@wrapt_timeout_decorator.timeout(dec_timeout=3.5)
def func(number, text):
    """『長い時間がかかるかもしれない関数』"""

    # (デバッグ) 長い時間待機してみる。
    time.sleep(number)
    z = f'{number} 秒のスリープ OK。text="{text}"'

    return z

実行結果

シングルプロセス処理の場合

『シングルプロセス処理 (processes=1)』で、自作関数のタイムアウトを発生させた場合です。

期待した通り、指定した秒数(3.5 秒)を超えたところで、TimeoutError が発生しました。

start

実行中...
=== シングルプロセス処理 (processes=1) ===
jisaku.func(number=1, text="あ") OK
jisaku.func(number=2, text="い") OK
jisaku.func(number=3, text="う") OK
TimeoutError: Function func timed out after 3.5 seconds
jisaku.func(number=4, text="え") Error
TimeoutError: Function func timed out after 3.5 seconds
jisaku.func(number=5, text="お") Error

結果を表示します。
(1, 'あ', '1.127 秒', 'PPID=232', 'PID=4748', '1 秒のスリープ OK。text="あ"')
(2, 'い', '2.121 秒', 'PPID=232', 'PID=4748', '2 秒のスリープ OK。text="い"')
(3, 'う', '3.113 秒', 'PPID=232', 'PID=4748', '3 秒のスリープ OK。text="う"')
(4, 'え', '3.608 秒', 'PPID=232', 'PID=4748', 'TimeoutError')
(5, 'お', '3.623 秒', 'PPID=232', 'PID=4748', 'TimeoutError')

end

マルチプロセス処理の場合

『マルチプロセス処理 (processes=3)』で、自作関数のタイムアウトを発生させた場合です。

期待した通り、指定した秒数(3.5 秒)を超えたところで、TimeoutError が発生しました。

start

実行中...
=== マルチプロセス処理 (processes=3) ===
jisaku.func(number=1, text="あ") OK
jisaku.func(number=2, text="い") OK
jisaku.func(number=3, text="う") OK
TimeoutError: Function func timed out after 3.5 seconds
jisaku.func(number=4, text="え") Error
TimeoutError: Function func timed out after 3.5 seconds
jisaku.func(number=5, text="お") Error

結果を表示します。
(1, 'あ', '1.179 秒', 'PPID=9852', 'PID=1516', '1 秒のスリープ OK。text="あ"')
(2, 'い', '2.139 秒', 'PPID=9852', 'PID=9108', '2 秒のスリープ OK。text="い"')
(3, 'う', '3.148 秒', 'PPID=9852', 'PID=9572', '3 秒のスリープ OK。text="う"')
(4, 'え', '3.617 秒', 'PPID=9852', 'PID=1516', 'TimeoutError')
(5, 'お', '3.604 秒', 'PPID=9852', 'PID=9108', 'TimeoutError')

end

以上です。

タイトルとURLをコピーしました