Developing an Asynchronous Task Queue in Python

This tutorial looks at how to implement several asynchronous task queues using Python's multiprocessing library and Redis.

Queue Data Structures
Task
- Following along?
Multiprocessing Pool
Multiprocessing Queue
Logging
Redis
Conclusion

Queue Data Structures

A queue is a First-In-First-Out (FIFO) data structure.

an item is added at the tail (enqueue)
an item is removed at the head (dequeue)

You'll see this in practice as you code out the examples in this tutorial.

Task

Let's start by creating a basic task:

# tasks.py

import collections
import json
import os
import sys
import uuid
from pathlib import Path

from nltk.corpus import stopwords

COMMON_WORDS = set(stopwords.words("english"))
BASE_DIR = Path(__file__).resolve(strict=True).parent
DATA_DIR = Path(BASE_DIR).joinpath("data")
OUTPUT_DIR = Path(BASE_DIR).joinpath("output")


def save_file(filename, data):
    random_str = uuid.uuid4().hex
    outfile = f"{filename}_{random_str}.txt"
    with open(Path(OUTPUT_DIR).joinpath(outfile), "w") as outfile:
        outfile.write(data)


def get_word_counts(filename):
    wordcount = collections.Counter()
    # get counts
    with open(Path(DATA_DIR).joinpath(filename), "r") as f:
        for line in f:
            wordcount.update(line.split())
    for word in set(COMMON_WORDS):
        del wordcount[word]

    # save file
    save_file(filename, json.dumps(dict(wordcount.most_common(20))))

    proc = os.getpid()

    print(f"Processed {filename} with process id: {proc}")


if __name__ == "__main__":
    get_word_counts(sys.argv[1])

So, get_word_counts finds the twenty most frequent words from a given text file and saves them to an output file. It also prints the current process identifier (or pid) using Python's os library.

Following along?

Create a project directory along with a virtual environment. Then, use pip to install NLTK:

(env)$ pip install nltk==3.8.1

Once installed, invoke the Python shell and download the stopwords corpus:

>>> import nltk
>>> nltk.download("stopwords")

[nltk_data] Downloading package stopwords to
[nltk_data]     /Users/michael/nltk_data...
[nltk_data]   Unzipping corpora/stopwords.zip.
True

If you experience an SSL error refer to this article.

Example fix:

>>> import nltk
>>> nltk.download('stopwords')
[nltk_data] Error loading stopwords: <urlopen error [SSL:
[nltk_data]     CERTIFICATE_VERIFY_FAILED] certificate verify failed:
[nltk_data]     unable to get local issuer certificate (_ssl.c:1056)>
False
>>> import ssl
>>> try:
...     _create_unverified_https_context = ssl._create_unverified_context
... except AttributeError:
...     pass
... else:
...     ssl._create_default_https_context = _create_unverified_https_context
...
>>> nltk.download('stopwords')
[nltk_data] Downloading package stopwords to
[nltk_data]     /Users/michael.herman/nltk_data...
[nltk_data]   Unzipping corpora/stopwords.zip.
True

Add the above tasks.py file to your project directory but don't run it quite yet.

Multiprocessing Pool

We can run this task in parallel using the multiprocessing library:

# simple_pool.py

import multiprocessing
import time

from tasks import get_word_counts

PROCESSES = multiprocessing.cpu_count() - 1


def run():
    print(f"Running with {PROCESSES} processes!")

    start = time.time()
    with multiprocessing.Pool(PROCESSES) as p:
        p.map_async(
            get_word_counts,
            [
                "pride-and-prejudice.txt",
                "heart-of-darkness.txt",
                "frankenstein.txt",
                "dracula.txt",
            ],
        )
        # clean up
        p.close()
        p.join()

    print(f"Time taken = {time.time() - start:.10f}")


if __name__ == "__main__":
    run()

Here, using the Pool class, we processed four tasks with two processes.

Did you notice the map_async method? There are essentially four different methods available for mapping tasks to processes. When choosing one, you have to take multi-args, concurrency, blocking, and ordering into account:

Method	Multi-args	Concurrency	Blocking	Ordered-results
`map`	No	Yes	Yes	Yes
`map_async`	No	No	No	Yes
`apply`	Yes	No	Yes	No
`apply_async`	Yes	Yes	No	No

Without both close and join, garbage collection may not occur, which could lead to a memory leak.

close tells the pool not to accept any new tasks
join tells the pool to exit after all tasks have completed

Following along? Grab the Project Gutenberg sample text files from the "data" directory in the simple-task-queue repo, and then add an "output" directory.

Your project directory should look like this:
├── data
│   ├── dracula.txt
│   ├── frankenstein.txt
│   ├── heart-of-darkness.txt
│   └── pride-and-prejudice.txt
├── output
├── simple_pool.py
└── tasks.py

It should take less than a second to run:

(env)$ python simple_pool.py

Running with 15 processes!
Processed heart-of-darkness.txt with process id: 50510
Processed frankenstein.txt with process id: 50515
Processed pride-and-prejudice.txt with process id: 50511
Processed dracula.txt with process id: 50512

Time taken = 0.6383581161

This script ran on a i9 Macbook Pro with 16 cores.

So, the multiprocessing Pool class handles the queuing logic for us. It's perfect for running CPU-bound tasks or really any job that can be broken up and distributed independently. If you need more control over the queue or need to share data between multiple processes, you may want to look at the Queue class.

For more on this along with the difference between parallelism (multiprocessing) and concurrency (multithreading), review the Speeding Up Python with Concurrency, Parallelism, and asyncio article.

Multiprocessing Queue

Let's look at a simple example:

# simple_queue.py

import multiprocessing


def run():
    books = [
        "pride-and-prejudice.txt",
        "heart-of-darkness.txt",
        "frankenstein.txt",
        "dracula.txt",
    ]
    queue = multiprocessing.Queue()

    print("Enqueuing...")
    for book in books:
        print(book)
        queue.put(book)

    print("\nDequeuing...")
    while not queue.empty():
        print(queue.get())


if __name__ == "__main__":
    run()

The Queue class, also from the multiprocessing library, is a basic FIFO (first in, first out) data structure. It's similar to the queue.Queue class, but designed for interprocess communication. We used put to enqueue an item to the queue and get to dequeue an item.

Check out the Queue source code for a better understanding of the mechanics of this class.

Now, let's look at more advanced example:

# simple_task_queue.py

import multiprocessing
import time

from tasks import get_word_counts

PROCESSES = multiprocessing.cpu_count() - 1
NUMBER_OF_TASKS = 10


def process_tasks(task_queue):
    while not task_queue.empty():
        book = task_queue.get()
        get_word_counts(book)
    return True


def add_tasks(task_queue, number_of_tasks):
    for num in range(number_of_tasks):
        task_queue.put("pride-and-prejudice.txt")
        task_queue.put("heart-of-darkness.txt")
        task_queue.put("frankenstein.txt")
        task_queue.put("dracula.txt")
    return task_queue


def run():
    empty_task_queue = multiprocessing.Queue()
    full_task_queue = add_tasks(empty_task_queue, NUMBER_OF_TASKS)
    processes = []
    print(f"Running with {PROCESSES} processes!")
    start = time.time()
    for n in range(PROCESSES):
        p =

Share this tutorial

Contents

Queue Data Structures

Task

Following along?

Multiprocessing Pool

Multiprocessing Queue