python - 並列ファイル マッチング、Python

Question

ファイルの悪意のあるコードをスキャンするスクリプトを改善しようとしています。ファイルには正規表現パターンのリストがあり、各行に 1 つのパターンがあります。現在の実装は基本的に bash スクリプトの find\grep コンボであるため、これらの正規表現は grep 用です。私のベンチマーク ディレクトリでは、bash スクリプトに 358 秒かかります。これを 72 秒で実行する Python スクリプトを作成できましたが、さらに改善したいと考えています。最初にベースコードを投稿し、次に試した微調整を投稿します。

import os, sys, Queue, threading, re

fileList = []
rootDir = sys.argv[1]

class Recurser(threading.Thread):

    def __init__(self, queue, dir):
    self.queue = queue
    self.dir = dir
    threading.Thread.__init__(self)

    def run(self):
    self.addToQueue(self.dir)

    ## HELPER FUNCTION FOR INTERNAL USE ONLY
    def addToQueue(self,  rootDir):
      for root, subFolders, files in os.walk(rootDir):
    for file in files:
       self.queue.put(os.path.join(root,file))
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)
      self.queue.put(-1)

class Scanner(threading.Thread):

    def __init__(self, queue, patterns):
    self.queue = queue
    self.patterns = patterns
    threading.Thread.__init__(self)

    def run(self):
    nextFile = self.queue.get()
    while nextFile is not -1:
       #print "Trying " + nextFile
       self.scanFile(nextFile)
       nextFile = self.queue.get()


    #HELPER FUNCTION FOR INTERNAL UES ONLY
    def scanFile(self, file):
       fp = open(file)
       contents = fp.read()
       i=0
       #for patt in self.patterns:
       if self.patterns.search(contents):
      print "Match " + str(i) + " found in " + file

############MAIN MAIN MAIN MAIN##################
############MAIN MAIN MAIN MAIN##################
############MAIN MAIN MAIN MAIN##################
############MAIN MAIN MAIN MAIN##################
############MAIN MAIN MAIN MAIN##################
############MAIN MAIN MAIN MAIN##################
############MAIN MAIN MAIN MAIN##################
############MAIN MAIN MAIN MAIN##################
############MAIN MAIN MAIN MAIN##################


fileQueue = Queue.Queue()

#Get the shell scanner patterns
patterns = []
fPatt = open('/root/patterns')
giantRE = '('
for line in fPatt:
   #patterns.append(re.compile(line.rstrip(), re.IGNORECASE))
   giantRE = giantRE + line.rstrip() + '|'

giantRE = giantRE[:-1] + ')'
giantRE = re.compile(giantRE, re.IGNORECASE)

#start recursing the directories
recurser = Recurser(fileQueue,rootDir)
recurser.start()

print "starting scanner"
#start checking the files
for scanner in xrange(0,8):
   scanner = Scanner(fileQueue, giantRE)
   scanner.start()

これは明らかにデバッグ\醜いコードです。100 万の queue.put(-1) は気にしないでください。後でクリーンアップします。特にscanFileで、いくつかのインデントが正しく表示されません。

とにかく気づいたことをいくつか。1、4、さらには 8 スレッド (xrange(0,???) のスキャナー用:) を使用しても違いはありません。それでも72秒以内です。これはpythonのGILによるものだと思います。

巨大な正規表現を作成するのではなく、各行 (パターン) を compilex RE としてリストに配置し、scanfile 関数でこのリストを反復処理してみました。これにより、実行時間が長くなりました。

PythonのGILを回避するために、次のように各スレッドをgrepにフォークさせてみました：

#HELPER FUNCTION FOR INTERNAL UES ONLY
def scanFile(self, file):
      s = subprocess.Popen(("grep", "-El", "--file=/root/patterns", file), stdout = subprocess.PIPE)
      output = s.communicate()[0]
      if output != '':
         print 'Matchfound in ' + file

これにより、実行時間が長くなりました。

パフォーマンスを改善するための提案。

：：：：：：：：：：：：：編集：：：：：：：：

私自身の質問への回答はまだ投稿できませんが、提起されたいくつかの点に対する回答は次のとおりです。

@David Nehme - 人々に知らせるために、私は100万個のqueue.put(-1)を持っているという事実を認識しています

@Blender - キューの一番下をマークします。私のスキャナースレッドは、一番下にある-1に達するまでデキューを続けます(nextFileは-1ではありません:)。プロセッサ コアは 8 ですが、GIL が 1 スレッド、4 スレッド、または 8 スレッドを使用しているため、違いはありません。8 つのサブプロセスを生成すると、コードが大幅に遅くなりました (142 秒対 72 秒)

@ed-はい、それはfind\grepコンボと同じくらい遅く、実際には必要のないファイルを無差別にgrepするため遅くなります

@Ron - アップグレードできません。これはユニバーサルでなければなりません。これで 72 秒以上速くなると思いますか? bash grepper は 358 秒かかります。私の Python の巨大な RE メソッドは、1 ～ 8 スレッドで 72 秒かかります。8 スレッド (8 サブプロセス) の popen メソッドは 142 秒で実行されました。これまでのところ、巨大な RE python のみの方法が明らかに勝者です。

@intted

これが現在の find\grep コンボの要点です (私のスクリプトではありません)。とてもシンプルです。そこにはlsのような追加のものがありますが、5倍の速度低下を引き起こすものは何もありません. grep -r がわずかに効率的であっても、5x は非常に遅くなります。

 find "${TARGET}" -type f -size "${SZLIMIT}" -exec grep -Eaq --file="${HOME}/patterns" "{}" \; -and -ls | tee -a "${HOME}/found.txt"

Python コードの方が効率的です。理由はわかりませんが、実験的にテストしました。私はPythonでこれを行うことを好みます。私はすでにpythonで5倍のスピードアップを達成しました.もっとスピードアップしたいと思います.

:::::::::::::勝者 勝者 勝者:::::::::::::::::

勝者がいるようです。

intued のシェル スクリプトは 34 秒で 2 位になりましたが、@steveha は 24 秒で 1 位になりました。多くのボックスに python2.6 がないため、cx_freeze を使用する必要がありました。tar を wget して解凍するためのシェル スクリプト ラッパーを作成できます。ただし、単純にするために intued が好きです。

助けてくれてありがとう、システム管理のための効率的なツールを手に入れました

Answer 1

あなたの Python スクリプトが検索と grep の組み合わせよりもどのように高速になったかについて、私は少し混乱しています。grepロン・スミスの回答で提案されているものと多少似た方法で使用したい場合は、次のようなことができます

find -type f | xargs -d \\n -P 8 -n 100 grep --file=/root/patterns

grep終了する前に 100 個のファイルを処理するプロセスを起動し、一度に最大 8 個のプロセスをアクティブに保ちます。それらに 100 個のファイルを処理させると、それぞれのプロセスの起動オーバーヘッド時間が無視できるようになります。

注:-d \\nオプション toxargsは GNU 拡張であり、すべての POSIX っぽいシステムでは機能しません。ファイル名間の * d * 区切り文字が改行であることを指定します。技術的にはファイル名に改行を含めることができますが、実際には誰もこれを行って仕事を続けていません。非 GNU との互換性のために、オプションをに追加してwithの代わりに使用xargsする必要があります。これにより、 との両方でヌル バイト(16 進数) が区切り文字として使用されるようになります。-print0find-0-d \\nxargs\00x00findxargs

grep されるファイルの数を最初にカウントするアプローチを取ることもできます

NUMFILES="$(find -type f | wc -l)";

そして、その数を使用して8つのプロセス間で均等に分割します（bashシェルと仮定）

find -type f | xargs -d \\n -P 8 -n $(($NUMFILES / 8 + 1)) grep --file=/root/patterns

findのディスク I/O がさまざまな のディスク I/O に干渉しないため、これはうまくいくと思いますgrep。ファイルの大きさと、それらが連続して保存されているかどうかにも部分的に依存すると思います.小さなファイルでは、ディスクはとにかく多くのシークを行うので、それほど重要ではありません. また、特に適切な量の RAM がある場合は、ファイルの一部がメモリ キャッシュに保存されるため、このようなコマンドの後続の実行が高速になることにも注意してください。

もちろん、 をパラメーター化し8て、さまざまな数の同時プロセスを簡単に試すことができます。

編として。コメントで言及されているように、このアプローチのパフォーマンスが単一プロセスのパフォーマンスよりも印象的ではない可能性は十分にありますgrep -r。ディスク[アレイ]の相対速度、システム内のプロセッサの数などに依存すると思います.

Answer 2

モジュールを使用するのではなく、Python ソリューションにモジュールをthreading使用する必要があると思います。multiprocessingPython スレッドは GIL に違反する可能性があります。複数の Python プロセスが実行されているだけであれば、GIL は問題になりません。

あなたがやっていることは、ワーカープロセスのプールがまさにあなたが望むものだと思います。デフォルトでは、プールはシステム プロセッサのコアごとに 1 つのプロセスにデフォルト設定されます。.map()チェックするファイル名のリストとチェックを行う関数を指定してメソッドを呼び出すだけです。

http://docs.python.org/library/multiprocessing.html

これがthreading実装よりも高速でない場合、GIL が問題ではないと思います。

編集:さて、動作する Python プログラムを追加しています。これは、ワーカー プロセスのプールを使用して各ファイルを開き、それぞれのパターンを検索します。ワーカーが一致するファイル名を見つけると、それを (標準出力に) 表示するだけなので、このスクリプトの出力をファイルにリダイレクトして、ファイルのリストを取得できます。

編集：これは少し読みやすく、理解しやすいバージョンだと思います。

コンピューターの /usr/include 内のファイルを検索して、これを計りました。約0.5秒で検索が完了します。findパイプスルーを使用しxargsて実行するgrepプロセスをできるだけ少なくすると、約 0.05 秒かかり、約 10 倍の速度になります。しかし、適切に動作させるために使用しなければならないバロック様式の奇妙な言語は嫌いでfind、Python バージョンが好きです。そして、おそらく非常に大きなディレクトリでは、Python の 0.5 秒の一部が起動時間であったに違いないため、格差は小さくなります。そしておそらく、ほとんどの目的には 0.5 秒で十分です!

import multiprocessing as mp
import os
import re
import sys

from stat import S_ISREG


# uncomment these if you really want a hard-coded $HOME/patterns file
#home = os.environ.get('HOME')
#patterns_file = os.path.join(home, 'patterns')

target = sys.argv[1]
size_limit = int(sys.argv[2])
assert size_limit >= 0
patterns_file = sys.argv[3]


# build s_pat as string like:  (?:foo|bar|baz)
# This will match any of the sub-patterns foo, bar, or baz
# but the '?:' means Python won't bother to build a "match group".
with open(patterns_file) as f:
    s_pat = r'(?:{})'.format('|'.join(line.strip() for line in f))

# pre-compile pattern for speed
pat = re.compile(s_pat)


def walk_files(topdir):
    """yield up full pathname for each file in tree under topdir"""
    for dirpath, dirnames, filenames in os.walk(topdir):
        for fname in filenames:
            pathname = os.path.join(dirpath, fname)
            yield pathname

def files_to_search(topdir):
    """yield up full pathname for only files we want to search"""
    for fname in walk_files(topdir):
        try:
            # if it is a regular file and big enough, we want to search it
            sr = os.stat(fname)
            if S_ISREG(sr.st_mode) and sr.st_size >= size_limit:
                yield fname
        except OSError:
            pass

def worker_search_fn(fname):
    with open(fname, 'rt') as f:
        # read one line at a time from file
        for line in f:
            if re.search(pat, line):
                # found a match! print filename to stdout
                print(fname)
                # stop reading file; just return
                return

mp.Pool().map(worker_search_fn, files_to_search(target))

Answer 3

バージョン 3.2 以降にアップグレードする場合は、concurrent.futures.ProcessPoolExecutor を利用できます。popen メソッドが毎回新しいプロセスを作成するプロセスのプールを事前に作成するため、試みた popen メソッドよりもパフォーマンスが向上すると思います。何らかの理由で 3.2 に移行できない場合は、以前のバージョンで同じことを行う独自のコードを作成できます。

Answer 4

また、並列 Python アプリケーションを作成するためのオープンソース フレームワークであるRayでこれを行う方法も紹介します。このアプローチの利点は、高速で、記述と拡張が容易であり (タスク間で大量のデータを渡したい場合や、ステートフルな蓄積を行いたい場合など)、変更なしでクラスターまたはクラウドで実行できることです。また、単一のマシン (100 コアのような非常に大きなマシンの場合でも) ですべてのコアを利用し、タスク間のデータ転送を非常に効率的に行うことができます。

import os
import ray
import re

ray.init()

patterns_file = os.path.expanduser("~/patterns")
topdir = os.path.expanduser("~/folder")

with open(patterns_file) as f:
    s_pat = r'(?:{})'.format('|'.join(line.strip() for line in f))

regex = re.compile(s_pat)

@ray.remote
def match(pattern, fname):
    results = []
    with open(fname, 'rt') as f:
        for line in f:
            if re.search(pattern, line):
                results.append(fname)
    return results

results = []
for dirpath, dirnames, filenames in os.walk(topdir):
    for fname in filenames:
        pathname = os.path.join(dirpath, fname)
        results.append(match.remote(regex, pathname))

print("matched files", ray.get(results))

これをクラスターまたはクラウドで実行する方法などの詳細については、ドキュメントを参照してください。