使用timeout-decorator為python函數任務設置超時時間

需求背景

在python程式碼的實現中,假如我們有一個需要執行時間跨度非常大的for循環,如果在中間的某處我們需要定時停止這個函數,而不停止整個程式。那麼初步的就可以想到兩種方案:第一種方案是我們先預估for循環或者while中的每一步所需要的運行時間,然後設定在到達某一個迭代次數之後就自動退出循環;第二種方案是,在需要設置超時任務的前方引入超時的裝飾器,使得超過指定時間之後自動退出函數執行。這裡我們將針對第二種方案,進行展開介紹。

timeout-decorator的安裝

在pypi的標準庫中也包含有timeout-decorator模組,因此可以通過pip來直接安裝:

[dechin@dechin-manjaro timeout]$ python3 -m pip install timeout_decorator
Collecting timeout_decorator
  Downloading timeout-decorator-0.5.0.tar.gz (4.8 kB)
Building wheels for collected packages: timeout-decorator
  Building wheel for timeout-decorator (setup.py) ... done
  Created wheel for timeout-decorator: filename=timeout_decorator-0.5.0-py3-none-any.whl size=5029 sha256=279f8585a08d5e5c87de887492169d1a81e02060c8ea3b62fdd6f062b7f83601
  Stored in directory: /home/dechin/.cache/pip/wheels/38/05/4e/161d1463ca145ec1023bd4e5e1f31cbf9239aa8f39a2a2b643
Successfully built timeout-decorator
Installing collected packages: timeout-decorator
Successfully installed timeout-decorator-0.5.0

配置一個超時任務

這裡我們先展示示例程式碼,再展開介紹其中各個模組的含義:

# timeout_test1.py
from tqdm import trange
import sys
import time
import timeout_decorator

@timeout_decorator.timeout(int(sys.argv[2]))
def test():
    if sys.argv[1] == '--timeout':
        for i in trange(3):
            time.sleep(1)
            print ('>>> {} seconds passed.'.format(i+1))
    return 0

if __name__ == '__main__':
    try:
        test()
    except Exception as e:
        print ('Timeout Error Catched!')
        print (e)
    print ("Timeout Task Ended!")

timeout-decorator裝飾器的使用

該超時模組採用裝飾器的形式來進行調用,使用時先import該模組,然後在需要設置定時任務的函數前添加@timeout_decorator.timeout(3)即可,這裡括弧中的3表示超時時間設置為3s,也就是3s後該函數就會停止運行。前面寫過一篇部落格介紹如何自定義一個裝飾器,感興趣的讀者可以自行閱讀。在上述的用例中,為了使得超時時間的定義更加靈活,我們採取了從用戶輸入獲取參數的方案,具體內容參考下一章節的介紹。

通過sys獲取timeout參數

在上述用例的裝飾器中,我們看到了int(sys.argv[2])這樣的一個參數,這個參數的意思是用戶輸入命令行的第三個用空格隔開的參數。舉例子說,如果用戶執行了python3 test.py -t 1,那麼這裡就會產生三個輸入參數:argv[0]就是test.pyargv[1]就是-targv[2]就是1,是一個數組的格式。需要注意的是,argv數組的每一個元素都是字元串格式,如果需要使用數字需要先進行格式轉換。這裡針對於超時任務的處理,我們指定的執行策略為類似python3 task.py --timeout 5的格式,--timeout後面的數字表示任務執行超時的秒數。如果輸入變數格式不正確,或者不滿足3個以上的變數輸入要求,或者第二個參數不是--timeout,都有可能運行報錯。

異常捕獲

在定義好超時任務之後,如果達到了設定好的超時時間,系統會給出timeout_decorator.timeout_decorator.TimeoutError報錯並結束程式運行。但是我們這裡配置超時任務的目的其實是希望在超時任務的函數到達指定時間之後退出,但是不影響其他模組程式的運行,因此這裡我們需要對程式給出的報錯進行異常捕獲,並且通報與抑制該異常。比較簡單的方案就是採用except Exception as e的方式,一般Exception最好可以指向指定的報錯類型,而不是通用的Exception處理,這有可能帶來其他的一些風險。

用例測試

以下按照輸入參數的不同,我們先劃分為幾個模組來分析輸出結果以及原因。

超時任務為2s

[dechin@dechin-manjaro timeout]$ python3 timeout_test.py --timeout 2
  0%|                                                 | 0/3 [00:00<?, ?it/s]>>> 1 seconds passed.
 33%|█████████████▋                           | 1/3 [00:01<00:03,  1.99s/it]
Timeout Error Catched!
'Timed Out'
Timeout Task Ended!

結果分析:由於我們在程式中給定了一個一共會執行3s的任務,而這裡在命令行中我們將超時時間設置為了2s,因此還沒執行完程式就拋出並捕獲了異常,成功列印了Timeout Task Ended!這一超時任務之外的任務。

超時任務為3s

[dechin@dechin-manjaro timeout]$ python3 timeout_test.py --timeout 3
  0%|                                                 | 0/3 [00:00<?, ?it/s]>>> 1 seconds passed.
 33%|█████████████▋                           | 1/3 [00:01<00:02,  1.00s/it]>>> 2 seconds passed.
 67%|███████████████████████████▎             | 2/3 [00:02<00:01,  1.50s/it]
Timeout Error Catched!
'Timed Out'
Timeout Task Ended!

結果分析:由於我們在程式中給定了一個一共會執行3s的任務,雖然在命令行的輸入參數中我們給定了3s的執行時間,但是最終程式還是沒有執行結束並拋出了異常。這是因為sleep(1)並不是精準的1s,也許是1.0000001但是這超出來的時間也會對最終執行的總時間產生影響,況且還有其他模組程式所導致的overlap,因此最後也沒有執行完成。而且從進度條來看,上面一個章節中時間設置為3s的時候,其實也只是完成了33%的任務而不是67%的任務,這也是符合我們的預期的。

超時任務為4s

[dechin@dechin-manjaro timeout]$ python3 timeout_test.py --timeout 4
  0%|                                                 | 0/3 [00:00<?, ?it/s]>>> 1 seconds passed.
 33%|█████████████▋                           | 1/3 [00:01<00:02,  1.00s/it]>>> 2 seconds passed.
 67%|███████████████████████████▎             | 2/3 [00:02<00:01,  1.00s/it]>>> 3 seconds passed.
100%|█████████████████████████████████████████| 3/3 [00:03<00:00,  1.00s/it]
Timeout Task Ended!

結果分析:由於我們在程式中給定了一個一共會執行3s的任務,而在參數輸入時配置了4s的超時時間,因此最終任務可以順利執行完成。這裡為了驗證上面一個小章節中提到的overlap,我們可以嘗試使用系統自帶的時間測試模組來測試,如果該程式執行完成之後,一共需要多少的時間:

[dechin@dechin-manjaro timeout]$ time python3 timeout_test.py --timeout 4
  0%|                                                 | 0/3 [00:00<?, ?it/s]>>> 1 seconds passed.
 33%|█████████████▋                           | 1/3 [00:01<00:02,  1.00s/it]>>> 2 seconds passed.
 67%|███████████████████████████▎             | 2/3 [00:02<00:01,  1.00s/it]>>> 3 seconds passed.
100%|█████████████████████████████████████████| 3/3 [00:03<00:00,  1.00s/it]
Timeout Task Ended!

real    0m3.167s
user    0m0.147s
sys     0m0.017s

這裡我們就可以看到,其實額定為3s的任務,執行完成需要約3.2s的實際時間,多出來的時間就是所謂的overlap

總結概要

函數的超時設置是一個比較小眾使用的功能,可以用於任務的暫停(並非截斷)等場景,並且配合上面章節提到的異常捕獲和參數輸入來使用,會使得任務更加優雅且合理。

版權聲明

本文首發鏈接為://www.cnblogs.com/dechinphy/p/timeout.html
作者ID:DechinPhy
更多原著文章請參考://www.cnblogs.com/dechinphy/