When Can the Speedup Ratio of Multi-processing Be Greater Than 1? · English (unofficial) translations of posts at kexue.fm

Parallel acceleration using multi-processing or multi-threading is no longer a difficult task nowadays, and I believe many readers have experienced it. Generally speaking, we tend to conclude that the speedup ratio of multi-processing is difficult to reach 1. In other words, when you use 10 processes to run a task in parallel, you usually only get less than a 10-fold speedup, and the more processes you use, the lower this speedup ratio often becomes.

Note that when we say it is “difficult to reach 1,” it implies our subconscious belief that the speedup ratio can at most be 1. Theoretically, this is true; how could using 10 processes result in a 20-fold speedup? Wouldn’t that be a “free lunch”? However, I did encounter an example a few days ago where the speedup ratio was significantly greater than 1, so I would like to share it here.

Word Frequency Counting

My original task was to count word frequencies: I have many articles, and we need to perform word segmentation (tokenization) on these articles and finally aggregate a word frequency table. A typical implementation looks like this:

tokens = {}

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

This implementation took about 20 minutes when I was counting the word frequencies of all articles in the THUCNews dataset.

Multi-processing Version

Next, let’s compare it with a multi-processing version. I have previously introduced multi-processing techniques in Python in the article “Multi-processing Programming Tips in Python”. For ease of reuse, I have encapsulated it into a function:

def parallel_apply(func,
                   iterable,
                   workers,
                   max_queue_size,
                   callback=None,
                   dummy=False):
    """
    Apply func to each element of iterable using multi-processing or multi-threading.
    Note that this apply is asynchronous and unordered, meaning if you input a, b, c,
    the output might be func(c), func(a), func(b).
    Parameters:
        dummy: False for multi-processing, True for multi-threading;
        callback: Callback function to handle a single output;
    """
    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):
        # Wrap single-step function into a loop
        while True:
            d = in_queue.get()
            r = func(d)
            out_queue.put(r)

    # Start multi-processing/threading
    pool = Pool(workers, worker_step, (in_queue, out_queue))

    if callback is None:
        results = []

    # Post-processing function
    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

    # Input data, retrieve results
    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 to call this function for multi-processing word frequency counting is roughly as follows:

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

# Use 10 processes to complete word frequency counting
parallel_apply(
    func=_tokenize_and_count,
    iterable=_batch_texts(),
    workers=10,
    max_queue_size=200,
    callback=_total_count,
)

The entire process is: _batch_texts divides the text into batches, with each batch containing 1000 texts; _tokenize_and_count is used to count each batch; _total_count aggregates the results of each batch; and finally, parallel_apply implements this process using 10 processes.

How long did this take? The result was 55 seconds! This means a 20-fold speedup, and the speedup ratio (efficiency) is 2!

Principle Analysis

Why can we achieve a speedup ratio greater than 1? Actually, the reason lies in the fact that in the initial single-process implementation, the line tokens[token] = tokens.get(token, 0) + 1 becomes slower and slower. As the counting progresses, the number of elements in tokens increases, making the CRUD (Create, Read, Update, Delete) operations on tokens increasingly slow.

In the multi-processing version, the line tokens[token] = tokens.get(token, 0) + 1 is only executed for batches of no more than 1000 samples, which obviously maintains a fast speed throughout. Although the final aggregation also involves frequent reads and writes to tokens, the frequency is far lower than in the original implementation, so it is also very fast. Therefore, the multi-processing version can achieve a 20-fold speedup, rather than just the theoretical limit of 10-fold.

Of course, readers may have already sensed that this isn’t truly making the speedup ratio exceed 1, but rather an illusion caused by the poor writing of the original single-process version. It could be improved with the following code:

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 still the approach of batch counting followed by aggregation, but it is single-processed. While this way of writing looks roundabout and unintuitive, it actually only took 8 minutes, which is about one-third of the original version! From this, we can see that the actual speedup ratio is approximately 0.8.

Summary

This article briefly discussed the issue of multi-processing in Python, provided an example where the speedup ratio seemingly exceeds 1, and analyzed the reasons behind it. From another perspective, this also serves as a reminder when writing similar code: even in a single-process scenario, the efficiency of batch calculation followed by aggregation is usually higher than calculating the entire batch all at once.