在Python 3中進行多重處理

Question

我正在搞一個zip文件破解程序，並決定使用多處理模塊來加速進程。這是一個完整的痛苦，因爲這是我第一次使用該模塊，我甚至還沒有完全理解它。但是，我得到它的工作。

問題是它沒有完成單詞列表;它只是在單詞列表中隨機停止放置，如果找到密碼，它將繼續通過單詞列表，而不僅僅是停止該過程。

有誰知道它爲什麼會出現這種行爲？

源代碼的ZipFile餅乾

#!/usr/bin/env python3 

import multiprocessing as mp 
import zipfile # Handeling the zipfile 
import sys # Command line arguments, and quiting application 
import time # To calculate runtime 

def usage(program_name): 
    print("Usage: {0} <path to zipfile> <dictionary>".format(program_name)) 
    sys.exit(1) 

def cracker(password): 
    try: 
     zFile.extractall(pwd=password) 
     print("[+] Password Found! : {0}".format(password.decode('utf-8'))) 
     pool.close() 
    except: 
     pass 

def main(): 
    global zFile 
    global pool 

    if len(sys.argv) < 3: 
     usage(sys.argv[0]) 

    zFile = zipfile.ZipFile(sys.argv[1]) 

    print("[*] Started Cracking") 

    startime = time.time() 
    pool = mp.Pool() 

    for i in open(sys.argv[2], 'r', errors='ignore'): 
     pswd = bytes(i.strip('\n'), 'utf-8') 
     pool.apply_async(cracker, (pswd,)) 

    print (pswd) 
    runtime = round(time.time() - startime, 5) 
    print ("[*] Runtime:", runtime, 'seconds') 
    sys.exit(0) 

if __name__ == "__main__": 
    main() 

Answer 1

下面是@phihag's意見的實施和@Equality 7-2521's answers：

#!/usr/bin/env python3 
"""Brute force zip password. 

Usage: brute-force-zip-password <zip archive> <passwords> 
""" 
import sys 
from multiprocessing import Pool 
from time import monotonic as timer 
from zipfile import ZipFile 

def init(archive): # run at the start of a worker process 
    global zfile 
    zfile = ZipFile(open(archive, 'rb')) # open file in each process once 

def check(password): 
    assert password 
    try: 
     with zfile.open(zfile.infolist()[0], pwd=password): 
      return password # assume success 
    except Exception as e: 
     if e.args[0] != 'Bad password for file': 
      # assume all other errors happen after the password was accepted 
      raise RuntimeError(password) from e 

def main(): 
    if len(sys.argv) != 3: 
     sys.exit(__doc__) # print usage 

    start = timer() 
    # decode passwords using the preferred locale encoding 
    with open(sys.argv[2], errors='ignore') as file, \ 
     Pool(initializer=init, initargs=[sys.argv[1]]) as pool: # use all CPUs 
     # check passwords encoded using utf-8 
     passwords = (line.rstrip('\n').encode('utf-8') for line in file) 
     passwords = filter(None, passwords) # filter empty passwords 
     for password in pool.imap_unordered(check, passwords, chunksize=100): 
      if password is not None: # found 
       print("Password: '{}'".format(password.decode('utf-8'))) 
       break 
     else: 
      sys.exit('Unable to find password') 
    print('Runtime: %.5f seconds' % (timer() - start,)) 

if __name__=="__main__": 
    main() 

注：

• 每個工作進程都有自己的ZipFile對象和壓縮文件在每個進程中打開一次：它應該使它更加便攜（Windows支持）並提高時間性能
• 內容未被提取：check(password)試圖在成功時打開並立即關閉歸檔成員：它更安全並且應該提高時間性能（不需要創建目錄等）
• 解密歸檔文件時除了'Bad password for file'之外的所有錯誤成員被假定爲發生後密碼被接受：理性是避免沉默意外的錯誤 - 每個異常應單獨考慮
• check(password)預計非空密碼
• chunksize參數可以大幅提高性能
• 難得for/else語法時，舉報箱時，沒有發現密碼
• 的語句來爲你

Answer 2

你太早期終止程序。要測試此方法，請在cracker方法中添加一個無害的time.sleep(10)，並觀察您的程序仍在一秒鐘內終止。

呼叫join等待池完成：

pool = mp.Pool() 
for i in open(sys.argv[2], 'r', errors='ignore'): 
    pswd = bytes(i.strip('\n'), 'utf-8') 
    pool.apply_async(cracker, (pswd,)) 

pool.close() # Indicate that no more data is coming 
pool.join() # Wait for pool to finish processing 

runtime = round(time.time() - startime, 5) 
print ("[*] Runtime:", runtime, 'seconds') 
sys.exit(0) 

此外，一旦你找到了正確的密碼，稱close只是表明沒有更多的未來的任務來了 - 已經提交了所有的任務仍然會做。相反，請致電terminate來終止池而不處理任何更多任務。

此外，根據multiprocessing.Pool的實現細節，全局變量pool在您需要時可能無法使用（並且其值無法反序列化）。爲了解決這個問題，你可以在

def cracker(password): 
    try: 
     zFile.extractall(pwd=password) 
    except RuntimeError: 
     return 
    return password 

def callback(found): 
    if found: 
     pool.terminate() 
... 
pool.apply_async(cracker, (pswd,), callback=cb) 

當然可以用一個回調，因爲，因爲你現在看結果的時候，apply是不正確的方式去。相反，你可以使用imap_unordered編寫代碼：

with open(sys.argv[2], 'r', errors='ignore') as passf, \ 
     multiprocessing.Pool() as pool: 
    passwords = (line.strip('\n').encode('utf-8') for line in passf) 
    for found in pool.imap_unordered(cracker, passwords): 
     if found: 
      break 

而不是使用全局變量，你可能還需要打開壓縮文件（並創建一個ZipFile對象）中的每一個過程，通過使用initializer爲池。更好（更快），放棄所有的I/O，只讀一次你需要的字節，然後傳給孩子們。

Answer 3

phihag的答案是正確的解決方案。

我只是想提供有關在找到正確密碼時撥打terminate()的更多詳細信息。我運行代碼時未定義pool變量cracker()。所以，試圖從那裏調用它只是拋出一個異常：

NameError: name 'pool' is not defined 

（我fork()經驗較弱，所以我不完全理解爲什麼全球zFile被複制到子進程成功而pool甚至不如果它被複制，它在父進程中不會是相同的pool，對嗎？因此，調用它的任何方法都不會影響父進程中的真實池。無論如何，我更喜歡multiprocessing模塊的編程指南中列出this建議部分：明確地傳遞資源，子進程）

我的建議是讓cracker()回報，如果它是正確的密碼，否則返回None。然後將回撥傳遞給apply_async()，記錄正確的密碼以及終止池。這是我採取在修改代碼來做到這一點：

#!/usr/bin/env python3 

import multiprocessing as mp 
import zipfile # Handeling the zipfile 
import sys # Command line arguments, and quiting application 
import time # To calculate runtime 
import os 

def usage(program_name): 
    print("Usage: {0} <path to zipfile> <dictionary>".format(program_name)) 
    sys.exit(1) 

def cracker(zip_file_path, password): 
    print('[*] Starting new cracker (pid={0}, password="{1}")'.format(os.getpid(), password)) 

    try: 
     time.sleep(1) # XXX: to simulate the task taking a bit of time 
     with zipfile.ZipFile(zip_file_path) as zFile: 
      zFile.extractall(pwd=bytes(password, 'utf-8')) 
     return password 
    except: 
     return None 

def main(): 
    if len(sys.argv) < 3: 
     usage(sys.argv[0]) 

    print('[*] Starting main (pid={0})'.format(os.getpid())) 

    zip_file_path = sys.argv[1] 
    password_file_path = sys.argv[2] 
    startime = time.time() 
    actual_password = None 

    with mp.Pool() as pool: 
     def set_actual_password(password): 
      nonlocal actual_password 
      if password: 
       print('[*] Found password; stopping future tasks') 
       pool.terminate() 
       actual_password = password 

     with open(password_file_path, 'r', errors='ignore') as password_file: 
      for pswd in password_file: 
       pswd = pswd.strip('\n') 
       pool.apply_async(cracker, (zip_file_path, pswd,), callback=set_actual_password) 

     pool.close() 
     pool.join() 

    if actual_password: 
     print('[*] Cracked password: "{0}"'.format(actual_password)) 
    else: 
     print('[*] Unable to crack password') 
    runtime = round(time.time() - startime, 5) 
    print("[*] Runtime:", runtime, 'seconds') 
    sys.exit(0) 

if __name__ == "__main__": 
    main()