python multiprocessing庫使用記錄

需求是想並行調用形式化分析工具proverif，同時發起對多個query的分析（378個）。實驗室有40核心80線程的伺服器（雙cpu，至強gold 5218R*2)。

觀察到單個命令在分析時記憶體占用不大，且只使用單核心執行，因此考慮同時調用多個命令同時執行分析，加快結果輸出。

最底層的邏輯是調用多個命令行語句，和在命令行直接執行proverif語句類似。在python中也就是使用 os.system()函數實現命令調用。然而由於存在如下問題，需要考慮使用多進程multiprocessing庫。

• 如果使用多線程threading庫，由於GIL的存在，是否會因為一個進程未執行結束而無法發起新的進程？
• query數量很大的原因來自於多場景分析，同時對於同一場景下的query也希望可以並行推進，同時分析。
• query數量大+場景多，得到很多結果，每條分析語句都有各自不同的位置，需要生成大量的命令。
• 每條query執行完成後會給出分析結果。雖然分析結果會以html文件的形式輸出到指定結果文件夾，但是不能對分析結果做統一的分析，仍舊需要逐個閱讀。希望能在輸出後即時統計，原有輸出不變，還能給出分析結果表。
• 儘管proverif在分析上速度已經很好了，但是仍然有62條query在30000秒（8.3h）後未給出結果。希望能夠統計每一條query的運行時間並記錄，並能夠提供當前仍在執行的query數量。
  • 進一步的，設置最高分析時長上限（如48h），若超出上限則終止分析。
  • 對於一些可達性查詢（reachability，實現方法是：在實體執行最後，在公開通道上發送執行完成標記，檢查攻擊者是否檢驗實體代碼是否正確，以及攻擊者是否能夠阻止合法實體正常執行程式（如何做？）），會出現構建攻擊路徑很慢的情況。但是實際上已經給出了goal reachable的結果。對於這種其實無需浪費更多時間，可以把reachability的query添加 set reconstructTrace = false .以提前結束。
  • 對於數量監控，需要多進程讀寫共用變數；對於運行時間記錄，需要多進程讀寫同一個文件。

mutliprocessing庫使用

主要使用multiprocessing.Pool()來創建進程池，當前python進程會創建新的python進程用於執行函數。（win下是子進程，linux下是fork）

由於存在操作系統上的差異，請使用if __name__ == '__main__':來編寫主函數，否則可能出現問題。主函數內容如下。

query_num = multiprocessing.Value('i', 0)

def long_time_task(c, ):
    start = time.time()
    os.system(c)
    end = time.time()
    # task_name=...
    with query_num.get_lock():
        query_num.value -= 1
        print('Task %s runs %0.2f seconds. ' % (task_name, (end - start)) + str(query_num.value) + ' left.')
    return 'Task %s runs %0.2f seconds.' % (task_name, (end - start))


def call_back(s):
    with open('/home/dell/proverif/DDS/time.txt', "a+") as file:
        file.writelines(s + '\n')
        

if __name__ == '__main__':
    query_list = extract(path_query, 'query', '.')
    query_file_path_list = query_file(query_list)
    whole_cS = compromise_Scenarios(path_compromise, path_process_whole, work_path)
    MAC_cS = compromise_Scenarios(path_compromise, path_process_MAC, work_path)
    cmd = []
    cmd += (pv_cmd(query_file_path_list, whole_cS, path_result))
    cmd += (pv_cmd(query_file_path_list, MAC_cS, path_result))
    p = Pool(len(cmd))
    query_num.value = len(cmd)
    # for i in cmd:
    #     p.apply_async(long_time_task, args=(i,), callback=call_back)

    results = [p.apply_async(long_time_task, (i,), callback=call_back) for i in cmd]

    print('Waiting for all subprocesses done...')
    output = [result.get(timeout=24*60*60) for result in results]
    # p.close()
    # p.join()
    print('All subprocesses done.')
    

主函數前7行為文本處理，其內容不細表。

第8行p = Pool(len(cmd))創建了進程池，其長度為cmd的個數，也就是我們要同時發起這麼多個進程。接下來註釋掉的迴圈是常規的多進程發起辦法，即使用apply_async函數執行我們要的函數。args是long_time_task的參數，由於需要為Iterable且只有一個參數，因此以元組形式傳入。

call_back參數為回調函數，這裡很像go語言下的defer，會在函數執行後再執行。回調函數接受long_time_task的返回值作為參數，我們使用這個機制實現多進程寫文件。long_time_task在返回後會受到進程池p的調度，依次執行寫文件操作，因此避免了同時寫引起衝突。

對於剩餘的query數量，使用全局變數query_num = multiprocessing.Value('i', 0)，這樣的變數具有鎖，可以供多進程讀寫。每個query在完成後會將數量減一，輸出時間和剩餘數量。使用with query_num.get_lock():獲得鎖，避免讀寫衝突，併在使用完成後自動釋放。

這已經滿足了基本需求。還有一個定時終止的功能有待實現。接下來再介紹我不斷修改的思路。

多進程定時終止

單進程定時終止

process = multiprocessing.Process(target=long_time_task)

# 啟動進程
process.start()

# 設置運行時長上限（48小時）
timeout = 48 * 60 * 60  # 以秒為單位

# 創建定時器，在指定時間後終止進程
timer = multiprocessing.Timer(timeout, process.terminate)
timer.start()

# 等待進程結束
process.join()

使用定時器的辦法，在一定時間後調用我們創建進程的process.terminate()方法結束進程。但我們需要多進程並行。

多進程定時終止

pool = multiprocessing.Pool()

# 準備要執行函數的參數列表
inputs = [1, 2, 3, 4, 5]

# 執行函數，並設置最大運行時長為30秒
result = pool.map_async(long_time_task, inputs)

# 獲取結果，最多等待30秒
output = result.get(timeout=48 * 60 * 60)

map_async方法可以將函數應用於可迭代的參數列表，並返回一個AsyncResult對象，可以使用該對象的get方法獲取結果。map_async方法將任務提交給進程池後會立即返回，並不會等待所有任務執行完成。如果在get方法獲取結果時，其中某些任務仍在執行，將會等待直到超時。get方法擁有timeout參數，超時後會raise TimeoutError，報錯終止python程式的運行。因此如果想輸出已完成的結果，有兩個思路：

1. try-except捕獲TimeoutError，並針對處理。
2. 對每個結果都使用get方法並設置超時時間。

列表推導式

results = [p.apply_async(long_time_task, (i,), callback=call_back) for i in cmd]

print('Waiting for all subprocesses done...')
output = [result.get(timeout=24*60*60) for result in results]
# p.close()
# p.join()
print('All subprocesses done.')

使用apply_async方法來執行函數，該方法會也會返回一個AsyncResult對象。我們將這些對象放入results數組，接著使用數組中每個元素的結果組成output數組並定義超時時間。這樣就可以執行call_back函數了。output內容其實不是很重要，主要是為了使用AsyncResult對象的get方法來設置定時器。

不過這樣還是需要try-except捕獲TimeoutError，以處理超時未完成的query。這樣做比map_async好在哪裡？我在使用的時候map_async似乎不能成功調用回調函數，還有待試驗。此外，該方法並不能在設定時間時準時停下，例如我設置時間5s，則會在約12秒時才停止。

還有一個問題是，在pycharm里運行腳本時，會有部分進程無法結束。暫不清楚其原因，也不確定命令行下執行腳本是否存在同樣的問題。

與Go相比

顯著的感覺到python在處理多進程、多線程、併發等問題上有一定的弱點。雖然能夠通過一系列操作實現，但是做起來比較吃力，也不算太優雅。現在的腳本已經可以並行分析了。然而在任務管理器中，除了看到了378個proverif進程，還看到了378個sh和378個python