When Can the Speedup of Multiprocessing Exceed 1?

什么时候多进程的加速比可以大于1？ 苏剑林 · 2019-10-27 · Machine Learning programming Python

Parallel speedups from multiprocessing or multithreading are no longer anything exotic, and I'm sure many readers have experienced them firsthand. Generally speaking, we have this conclusion: it's very hard for multiprocessing to reach a speedup of 1. In other words, when you use 10 processes to run a task in parallel, you typically get less than a 10x speedup, and the more processes you use, the lower this speedup ratio tends to be.

Notice that when we say "very hard to reach 1," this implies our subconscious assumption is that the speedup ratio is capped at 1. In theory, that's indeed the case — how could using 10 processes possibly yield a 20x speedup? That would be too good to be true. However, I actually ran into an example a few days ago where the speedup ratio was far greater than 1, so I'd like to share it here.

Word Frequency Counting

My original task was to count word frequencies: I had a large collection of articles, and I needed to tokenize these articles and finally aggregate a word frequency table. The typical way to write this is as follows:

tokens = {}

for text in read_texts():
    for token in tokenize(text):
        tokens[token] = tokens.get(token, 0) + 1

more

Using this approach to count word frequencies over the entire THUCNews] corpus took about 20 minutes.

The Multiprocess Version

Now let's compare this to the multiprocess version. I already introduced the technique for writing multiprocess code in the article Python Multiprocessing Tips], and to make it easy to reuse, I wrapped it into a function:

def parallel_apply(func,
                   iterable,
                   workers,
                   max_queue_size,
                   callback=None,
                   dummy=False):
    """多进程或多线程地将func应用到iterable的每个元素中。
    注意这个apply是异步且无序的，也就是说依次输入a,b,c，但是
    输出可能是func(c), func(a), func(b)。
    参数：
        dummy: False是多进程/线性，True则是多线程/线性；
        callback: 处理单个输出的回调函数；
    """
    if dummy:
        from multiprocessing.dummy import Pool, Queue
    else:
        from multiprocessing import Pool, Queue
    from six.moves import queue

    in_queue, out_queue = Queue(max_queue_size), Queue()

    def worker_step(in_queue, out_queue):
        # 单步函数包装成循环执行
        while True:
            d = in_queue.get()
            r = func(d)
            out_queue.put(r)

    # 启动多进程/线程
    pool = Pool(workers, worker_step, (in_queue, out_queue))

    if callback is None:
        results = []

    # 后处理函数
    def process_out_queue():
        out_count = 0
        for _ in range(out_queue.qsize()):
            d = out_queue.get()
            out_count += 1
            if callback is None:
                results.append(d)
            else:
                callback(d)
        return out_count

    # 存入数据，取出结果
    in_count, out_count = 0, 0
    for d in iterable:
        in_count += 1
        while True:
            try:
                in_queue.put(d, block=False)
                break
            except queue.Full:
                out_count += process_out_queue()
        if in_count % max_queue_size == 0:
            out_count += process_out_queue()

    while out_count != in_count:
        out_count += process_out_queue()

    pool.terminate()

    if callback is None:
        return results

The code for calling this function to count word frequencies with multiprocessing looks roughly like this:

def _batch_texts():
    texts = []
    for text in read_texts():
        texts.append(text)
        if len(texts) == 1000:
            yield texts
            texts = []
    if texts:
        yield texts

def _tokenize_and_count(texts):
    tokens = {}
    for text in texts:
        for token in tokenize(text):
            tokens[token] = tokens.get(token, 0) + 1
    return tokens

tokens = {}
def _total_count(result):
    for k, v in result.items()
        tokens[k] = tokens.get(k, 0) + v

# 10进程来完成词频统计
parallel_apply(
    func=_tokenize_and_count,
    iterable=_batch_texts(),
    workers=10,
    max_queue_size=200,
    callback=_total_count,
)

The overall flow is: _batch_texts splits the text into batches, with each batch containing 1000 documents; _tokenize_and_count is used to perform the count on each batch of samples; _total_count aggregates the results across batches; and finally parallel_apply implements this whole process using 10 processes.

How long did this take? The answer is 55 seconds! That's a 20x speedup — meaning a speedup ratio of 2!

Analysis of the Underlying Cause

Why does this achieve a speedup ratio greater than 1? The reason, it turns out, lies in the fact that in the original single-process implementation, the line tokens[token] = tokens.get(token, 0) + 1 becomes progressively slower, because as the counting proceeds, tokens accumulates more and more elements, making insertions, deletions, lookups, and updates to tokens increasingly slow.

In the multiprocess version, however, the line tokens[token] = tokens.get(token, 0) + 1 is only ever executed on batches of no more than 1000 samples, so it consistently stays fast. Although the final merging step also involves frequent reads and writes to tokens, this is far less frequent than in the original implementation, so it too remains fast. This is why the multiprocess version achieves a 20x speedup, rather than being capped at the theoretical limit of 10x.

Of course, readers may already suspect that this isn't truly a case of the speedup ratio exceeding 1 — rather, it's a symptom of the original single-process version being poorly written. If we rewrite it as follows, the problem goes away:

count = 0
tokens = {}
_tokens = {}

for text in read_texts():
    for token in tokenize(text):
        _tokens[token] = _tokens.get(token, 0) + 1
    count += 1
    if count == 1000:
        for k, v in _tokens.items():
            tokens[k] = tokens.get(k, 0) + v
        count = 0
        _tokens = {}

for k, v in _tokens.items():
    tokens[k] = tokens.get(k, 0) + v

This is essentially the same batch-then-aggregate approach, just done within a single process. It might look convoluted and less straightforward, but in fact it only took 8 minutes — roughly a third of the time taken by the original version! From this we can see that the actual speedup ratio is about 0.8.

Summary

This post gave a brief discussion of a multiprocessing issue in Python, presenting an example that appears to have a speedup ratio greater than 1, and then analyzed the reason behind it. Looked at from another angle, this also serves as a useful reminder for writing similar code: even in the single-process case, computing in batches and then aggregating is usually more efficient than processing everything in one giant batch at once.