并行化蛮力生成

Question

我为此搜索了很多，但找不到我能掌握的东西。我正在寻找的是如何使这个算法并行。它的并行方式无关紧要，例如多线程或多处理甚至分布式计算，但我将如何在节点或核心之间分配工作？

def bruteforce(charset, maxlength):
    return (''.join(candidate)
        for candidate in itertools.chain.from_iterable(itertools.product(charset, repeat=i)
        for i in range(1, maxlength + 1)))

Answer 1

如果我正确理解了您的生成器表达式，那么您正在尝试生成所有长度为 1 .. maxlength 给定字母表的单词。

我不知道你想如何分解你的问题（你有N 工人吗？），但一个明显的方法是只按第一个字母分割单词列表，然后把它们交给各种并行工作人员，然后都必须将所有可能的单词组合附加 0 .. maxlength - 1 字母表中的字母。

Answer 2

您要做的第一件事就是对 1-liner disease 消除幻想 ;-) 也就是说，并行处理本身会增加许多复杂性，因此您希望使代码尽可能简单和透明。这是一种概括@BasSwinckels 建议的方法。不短！但它非常有效：无论你有多少个内核，它都会将你的 CPU 计量器钉在墙上。

CHARSET = "abcdefghijklmnopqrstuvwxyx"
MAX_LENGTH = 6  # generate all strings from CHARSET with length 1 thru MAX_LENGTH
NUM_PROCESSES = None # defaults to all available cores

from itertools import product

# string_gen is what the workers run.  Everything else
# runs in the main program.
def string_gen(prefix, suffix_len, length):
    # Generate all strings of length `length` starting with `prefix`.
    # If length > suffix_len, only the last suffix_len characters
    # need to be generated.
    num_done = 0
    if length <= suffix_len:
        assert prefix == ""
        for t in product(CHARSET, repeat=length):
            result = "".join(t)
            # do something with result
            num_done += 1
    else:
        assert len(prefix) + suffix_len == length
        for t in product(CHARSET, repeat=suffix_len):
            result = prefix + "".join(t)
            # do something with result
            num_done += 1
    return num_done

def record_done(num):
    global num_done
    num_done += num
    print num_done, "done so far"

def do_work(pool, strings_per_chunk=1000000):
    # What's the most chars we can cycle through without
    # exceeding strings_per_chunk?  Could do with this
    # logs, but I'm over-reacting to 1-liner disease ;-)
    N = len(CHARSET)
    suffix_len = 1
    while N**suffix_len <= strings_per_chunk:
        suffix_len += 1
    suffix_len -= 1
    print "workers will cycle through the last", suffix_len, "chars"

    # There's no point to splitting up very short strings.
    max_short_len = min(suffix_len, MAX_LENGTH)
    for length in range(1, max_short_len + 1):
        pool.apply_async(string_gen, args=("", suffix_len, length),
                         callback=record_done)
    # And now the longer strings.
    for length in range(max_short_len + 1, MAX_LENGTH + 1):
        for t in product(CHARSET, repeat=length-suffix_len):
            prefix = "".join(t)
            pool.apply_async(string_gen, args=(prefix, suffix_len, length),
                             callback=record_done)

if __name__ == "__main__":
    import multiprocessing
    pool = multiprocessing.Pool(NUM_PROCESSES)
    num_done = 0
    do_work(pool)
    pool.close()
    pool.join()
    expected = sum(len(CHARSET)**i
                   for i in range(1, MAX_LENGTH + 1))
    assert num_done == expected, (num_done, expected)

这有多个部分，因为您想要的是“块状”：各种大小的字符串。问题的结构越完全统一，并行的噱头通常就越容易。但是可以处理块状 - 它只需要更多代码。

请注意assert 语句和num_done 的“无用”计算。并行代码增加了全新维度的复杂性，所以请帮自己一个忙，从一开始就防御性地编写代码。您将尝试许多根本行不通的事情 - 每个人都会遇到这种情况。

还要注意，即使没有多核，摆脱 1-liner 疾病也可以提供更有效的方法：计算并加入 prefix 一次以获得更长的字符串将在更长的过程中节省数十亿的冗余连接运行。

玩得开心:-)