運行子進程時檢查條件

我正在使用subprocess.Popen（）執行命令。我想在執行剩餘的代碼之前等待該過程完成，但同時我想在運行子進程2分鐘後檢查生成的文件的狀態。如果文件的大小爲零，那麼我想停止該過程。目前我的代碼如下。有沒有更聰明的方法來做到這一點？運行子進程時檢查條件

def execute(command,outputfilename): 
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) 
    start_time=time.time() 
    Is_Working=False 
    while True: 
    process.poll() 
    if not Is_Working: 
     #allow 2 minutes for the process to create results 
     if process.returncode == None and (time.time()-start_time)//60>1: 
      Is_Working=True 
      if (os.stat(outputfilename)[6]==0): 
       process.kill() 
       return 
    if process.returncode != None: 
     break 

    output, errors=process.communicate()

來源

2014-03-06 Reyhaneh

這個問題似乎是題外話題，因爲它是關於審查一個工作代碼，這是主題在http://codereview.stackexchange.com – zmo

不相關：使用'是None'來比較'None'，因爲它是一個單例，'=='可以被覆蓋，即使對於非'Non'對象也返回'True'。 – jfs

全球您的代碼對我來說很好。只有幾個細節：

在(time.time()-start_time)//60>1，我覺得有一個沒用使用//，因爲你不一定需要落地的結果，並轉換爲整數除法的LHS的結果。保持它全部浮動應該可以用於比較，這是所有基本的機器邏輯;
你能避免從一個無限循環與while process.returncode is not None:…
突破，通過改變循環條件，使之更簡單，其實我循環，直到文件大小爲!=0，然後調用process.wait()只是在循環之後。
其將被使用while/else結構

所以，一個改善將是一個很好的方案，所以你可以做的東西（如清理或重試...）中的處理結束後（無論成功或失敗）：

def execute(command,outputfilename): 
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) 
    start_time=time.time() 
    Is_Working=False 
    while process.returncode == None: 
     process.poll() 
     #allow 2 minutes for the process to create results 
     if (time.time()-start_time)/60 > 1: 
      if (os.stat(outputfilename)[6]==0): 
       process.kill() 
       break 
    else: 
     # the process has not been killed 
     # wait until it finishes 
     process.wait() 
     output, errors=process.communicate() 
     # do stuff with the output 
     […] 

    # here the process may have been killed or not 
    […]

或其他好的選擇是拋出異常：

def execute(command,outputfilename): 
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) 
    start_time=time.time() 
    Is_Working=False 
    while process.returncode == None: 
     process.poll() 
     #allow 2 minutes for the process to create results 
     if (time.time()-start_time)/60 > 1: 
      if (os.stat(outputfilename)[6]==0): 
       process.kill() 
       raise Exception("Process has failed to do its job") 
    # the process has not been killed 
    # wait until it finishes 
    process.wait() 
    output, errors=process.communicate() 
    # do stuff with the output 
    […]

HTH

來源

2014-03-06 15:02:57 zmo

Thanks.I使用第一個選項 – Reyhaneh

殺子過程在2分鐘內，如果outputfilename是空的（我假設outputfilename是由一些外部進程修改），你可以使用threading.Timer：

import os 
from subprocess import Popen, PIPE 
from threading import Timer 

def kill_if_empty(proc, filename): 
    if proc.poll() is None and os.path.getsize(filename) == 0: 
     proc.kill() 

def execute(command, outputfilename): 
    p = Popen(command, stdout=PIPE, stderr=PIPE) 
    Timer(2*60, kill_if_empty, [p, outputfilename]).start() 
    output, errors = p.communicate() 
    ...

此代碼收集標準輸出/標準錯誤分開，並避免您的代碼中存在可能存在的死鎖問題。

來源

2014-03-06 18:37:18 jfs

運行子進程時檢查條件

回答

相關問題