并行文件匹配，Python

我正在尝试改进扫描文件中是否存在恶意代码的脚本。我们在文件中有一个正则表达式模式列表，每行一个模式。这些正则表达式适用于 grep，因为我们当前的实现基本上是 bash 脚本 find\grep 组合。 bash 脚本在我的基准目录上花费了 358 秒。我能够编写一个 Python 脚本，在 72 秒内完成此操作，但希望进一步改进。首先，我将发布基本代码，然后调整我尝试过的：

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()

这显然是调试\丑陋的代码，不要介意百万个queue.put(-1)，我稍后会清理它。某些缩进未正确显示，特别是在 scanFile 中。

无论如何，我注意到了一些事情。使用 1 个、4 个甚至 8 个线程（对于 xrange(0,???) 中的扫描仪:) 没有什么区别。不管怎样，我仍然有大约 72 秒的时间。我认为这是由于 python 的 GIL 造成的。

与制作巨大的正则表达式相反，我尝试将每一行（模式）作为编译 RE 放在列表中，并在我的 scanfile 函数中迭代此列表。这导致执行时间更长。

为了避免 python 的 GIL，我尝试让每个线程 fork 到 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 - 只是为了让人们知道我知道我有一百万个queue.put(-1)

@Blender - 标记队列的底部。我的扫描器线程不断出队，直到到达底部的 -1（而 nextFile 不是 -1:）。处理器核心为 8 个，但由于 GIL 使用 1 个线程、4 个线程或 8 个线程，因此没有区别。生成 8 个子进程会导致代码显着变慢（142 秒 vs 72 秒）

@ed - 是的，它和 find\grep 组合一样慢，实际上更慢，因为它不加区别地 grep 不需要的文件

@Ron - 无法升级，这必须是通用的。你认为这会加速 > 72 秒吗？ bash grepper 执行 358 秒。我的 python 巨型 RE 方法需要 1-8 个线程执行 72 秒。 popen 方法有 8 个线程（8 个子进程），运行时间为 142 秒。到目前为止，巨型 RE python 方法显然是赢家

@intuted

这是我们当前的 find\grep 组合的核心内容（不是我的脚本）。这很简单。其中还有一些其他内容，例如 ls，但不会导致速度减慢 5 倍。即使 grep -r 的效率稍微高一点，5 倍也会造成巨大的减慢。

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

python代码效率更高，不知道为什么，但是我实验测试了一下。我更喜欢用 python 来做这件事。我已经用 python 实现了 5 倍的加速，我想让它加速更多。

::::::::::::::获胜者获胜者获胜者::::::::::::::::::

看来我们有赢家了。

intued 的 shell 脚本以 34 秒位居第二，但 @steveha 的 shell 脚本以 24 秒位居第一。由于我们的很多盒子没有 python2.6，我不得不 cx_freeze 它。我可以编写一个 shell 脚本包装器来获取 tar 并将其解压。不过，为了简单起见，我确实喜欢直觉。

感谢你们所有的帮助，我现在有了一个有效的系统管理工具

我认为，而不是使用threading模块，你应该使用multiprocessingPython 解决方案的模块。 Python 线程可能会与 GIL 发生冲突；如果您只是运行多个 Python 进程，那么 GIL 不是问题。

我认为对于你正在做的事情来说，工作进程池正是你想要的。默认情况下，池将默认为系统处理器中的每个核心分配一个进程。只需致电.map()带有要检查的文件名列表的方法以及执行检查的函数。

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

如果这不比你的快threading实施，那么我不认为 GIL 是你的问题。

编辑：好的，我正在添加一个可用的 Python 程序。这使用工作进程池来打开每个文件并搜索每个文件中的模式。当工作程序找到匹配的文件名时，它只是将其打印（到标准输出），以便您可以将此脚本的输出重定向到文件中，并且您将获得文件列表。

编辑：我认为这是一个更容易阅读的版本，更容易理解。

我计时了这一点，搜索了我计算机上 /usr/include 中的文件。它在大约半秒内完成搜索。使用find管道通过xargs运行尽可能少grep尽可能地处理，大约需要 0.05 秒，大约有 10 倍的加速。但我讨厌你必须使用巴洛克式的奇怪语言才能得到find可以正常工作，而且我喜欢 Python 版本。也许在真正大的目录上，差异会更小，因为 Python 的半秒的一部分一定是启动时间。对于大多数用途来说，也许半秒就足够快了！

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))