Python - 独自関数にタイムアウトを設定

最近Pythonで実装しているあるバッチ処理が、正常終了にもエラーにもならずプロセスとして残ってしまっていることが分かりました。

無限ループしているわけではなく、サーバの状態異常のため特定のLinuxコマンドの実行結果が戻ってこないことが原因でした。

なんとかしたいと思い調査したところ、関数にタイムアウトを設定できるライブラリがありましたので試してみたいと思います。

ライブラリのインストール

タイムアウト処理を行うためのライブラリをインストールします。

[コンソール]

1
pip install wrapt-timeout-decorator

Python3.6以降でないとインストールできないようですので、ご注意ください。

サンプルソース

無限ループする関数を定義し、その関数にタイムアウトの設定を行います。

タイムアウトの設定は4行目のデコレータで行っています。

デコレータとはある関数を拡張する機能で、関数の開始前や終了後などに処理を付け加えることができる機能のことです。

[コンソール]

1
2
3
4
5
6
7
8
9
10
import wrapt_timeout_decorator

# デコレータでタイムアウトの秒数を設定
@wrapt_timeout_decorator.timeout(dec_timeout=30)
def func():
    while True:
        pass

if __name__ == '__main__':
    func()

実行結果は以下の通りです。

[実行結果]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Traceback (most recent call last):
  File "test.py", line 11, in <module>
    func()
  File "C:\Util\anaconda3\lib\site-packages\wrapt_timeout_decorator\wrapt_timeout_decorator.py", line 123, in wrapper
    return wrapped_with_timeout(wrap_helper)
  File "C:\Util\anaconda3\lib\site-packages\wrapt_timeout_decorator\wrapt_timeout_decorator.py", line 131, in wrapped_with_timeout
    return wrapped_with_timeout_process(wrap_helper)
  File "C:\Util\anaconda3\lib\site-packages\wrapt_timeout_decorator\wrapt_timeout_decorator.py", line 145, in wrapped_with_timeout_process
    return timeout_wrapper()
  File "C:\Util\anaconda3\lib\site-packages\wrapt_timeout_decorator\wrap_function_multiprocess.py", line 43, in __call__
    self.cancel()
  File "C:\Util\anaconda3\lib\site-packages\wrapt_timeout_decorator\wrap_function_multiprocess.py", line 51, in cancel
    raise_exception(self.wrap_helper.timeout_exception, self.wrap_helper.exception_message)
  File "C:\Util\anaconda3\lib\site-packages\wrapt_timeout_decorator\wrap_helper.py", line 178, in raise_exception
    raise exception(exception_message)
TimeoutError: Function func timed out after 30.0 seconds

最終行に30秒タイムアウトが発生したことが表示されています。

デコレータを追加するだけで簡単にタイムアウトの設定ができるのはとても便利ですね。