python task queue example

Developing an Asynchronous Task Queue in Python

Share this tutorial.

• Hacker News

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

Queue Data Structures

Following along, multiprocessing pool, multiprocessing queue.

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.

Let's start by creating a basic task:

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.

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

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

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.

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

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:

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:

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.

Let's look at a simple example:

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:

Here, we enqueued 40 tasks (ten for each text file) to the queue, created separate processes via the Process class, used start to start running the processes, and, finally, used join to complete the processes.

It should still take less than a second to run.

Challenge : Check your understanding by adding another queue to hold completed tasks. You can enqueue them within the process_tasks function.

The multiprocessing library provides support for logging as well:

To test, change task_queue.put("dracula.txt") to task_queue.put("drakula.txt") . You should see the following error outputted ten times in the terminal:

Want to log to disc?

Again, cause an error by altering one of the file names, and then run it. Take a look at process.log . It's not quite as organized as it should be since the Python logging library does not use shared locks between processes. To get around this, let's have each process write to its own file. To keep things organized, add a logs directory to your project folder:

Moving right along, instead of using an in-memory queue, let's add Redis into the mix.

Following along? Download and install Redis if you do not already have it installed. Then, install the Python interface : (env)$ pip install redis == 4 .5.5

We'll break the logic up into four files:

• redis_queue.py creates new queues and tasks via the SimpleQueue and SimpleTask classes, respectively.
• redis_queue_client enqueues new tasks.
• redis_queue_worker dequeues and processes tasks.
• redis_queue_server spawns worker processes.

Here, we defined two classes, SimpleQueue and SimpleTask :

• SimpleQueue creates a new queue and enqueues, dequeues, and gets the length of the queue.
• SimpleTask creates new tasks, which are used by the instance of the SimpleQueue class to enqueue new tasks, and processes new tasks.

Curious about lpush() , brpop() , and llen() ? Refer to the Command reference page. ( The brpop() function is particularly cool because it blocks the connection until a value exists to be popped!)

This module will create a new instance of Redis and the SimpleQueue class. It will then enqueue 40 tasks.

If a task is available, the dequeue method is called, which then de-serializes the task and calls the process_task method (in redis_queue.py ).

The run method spawns four new worker processes.

You probably don’t want four processes running at once all the time, but there may be times that you will need four or more processes. Think about how you could programmatically spin up and down additional workers based on demand.

To test, run redis_queue_server.py and redis_queue_client.py in separate terminal windows:

Check your understanding again by adding logging to the above application.

In this tutorial, we looked at a number of asynchronous task queue implementations in Python. If the requirements are simple enough, it may be easier to develop a queue in this manner. That said, if you're looking for more advanced features -- like task scheduling, batch processing, job prioritization, and retrying of failed tasks -- you should look into a full-blown solution. Check out Celery , RQ , or Huey .

Grab the final code from the simple-task-queue repo.

Full-text Search in Django with Postgres and Elasticsearch

Learn how to add full-text search to Django with both Postgres and Elasticsearch.

Recommended Tutorials

Stay sharp with course updates.

Join our mailing list to be notified about updates and new releases.

Send Us Feedback

Flask by Example – Implementing a Redis Task Queue

Table of Contents

Install Requirements

Set up the worker, update app.py, get results, what’s next.

This part of the tutorial details how to implement a Redis task queue to handle text processing.

• 02/12/2020: Upgraded to Python version 3.8.1 as well as the latest versions of Redis, Python Redis, and RQ. See below for details. Mention a bug in the latest RQ version and provide a solution. Solved the http before https bug.
• 03/22/2016: Upgraded to Python version 3.5.1 as well as the latest versions of Redis, Python Redis, and RQ. See below for details.
• 02/22/2015: Added Python 3 support.

Free Bonus: Click here to get access to a free Flask + Python video tutorial that shows you how to build Flask web app, step-by-step.

Remember: Here’s what we’re building—A Flask app that calculates word-frequency pairs based on the text from a given URL.

• Part One : Set up a local development environment and then deploy both a staging and a production environment on Heroku.
• Part Two : Set up a PostgreSQL database along with SQLAlchemy and Alembic to handle migrations.
• Part Three : Add in the back-end logic to scrape and then process the word counts from a webpage using the requests, BeautifulSoup, and Natural Language Toolkit (NLTK) libraries.
• Part Four: Implement a Redis task queue to handle the text processing. ( current )
• Part Five : Set up Angular on the front-end to continuously poll the back-end to see if the request is done processing.
• Part Six : Push to the staging server on Heroku - setting up Redis and detailing how to run two processes (web and worker) on a single Dyno.
• Part Seven : Update the front-end to make it more user-friendly.
• Part Eight : Create a custom Angular Directive to display a frequency distribution chart using JavaScript and D3.

Need the code? Grab it from the repo .

Tools used:

• Redis ( 5.0.7 )
• Python Redis ( 3.4.1 )
• RQ ( 1.2.2 ) - a simple library for creating a task queue

Start by downloading and installing Redis from either the official site or via Homebrew ( brew install redis ). Once installed, start the Redis server:

Next install Python Redis and RQ in a new terminal window:

Let’s start by creating a worker process to listen for queued tasks. Create a new file worker.py , and add this code:

Here, we listened for a queue called default and established a connection to the Redis server on localhost:6379 .

Fire this up in another terminal window:

Now we need to update our app.py to send jobs to the queue…

Add the following imports to app.py :

Then update the configuration section:

q = Queue(connection=conn) set up a Redis connection and initialized a queue based on that connection.

Move the text processing functionality out of our index route and into a new function called count_and_save_words() . This function accepts one argument, a URL, which we will pass to it when we call it from our index route.

Take note of the following code:

Note: We need to import the count_and_save_words function in our function index as the RQ package currently has a bug, where it won’t find functions in the same module.

Here we used the queue that we initialized earlier and called the enqueue_call() function. This added a new job to the queue and that job ran the count_and_save_words() function with the URL as the argument. The result_ttl=5000 line argument tells RQ how long to hold on to the result of the job for - 5,000 seconds, in this case. Then we outputted the job id to the terminal. This id is needed to see if the job is done processing.

Let’s setup a new route for that…

Let’s test this out.

Fire up the server, navigate to http://localhost:5000/ , use the URL https://realpython.com , and grab the job id from the terminal. Then use that id in the ‘/results/’ endpoint - i.e., http://localhost:5000/results/ef600206-3503-4b87-a436-ddd9438f2197 .

As long as less than 5,000 seconds have elapsed before you check the status, then you should see an id number, which is generated when we add the results to the database:

Now, let’s refactor the route slightly to return the actual results from the database in JSON:

Make sure to add the import:

Test this out again. If all went well, you should see something similar to in your browser:

In Part 5 we’ll bring the client and server together by adding Angular into the mix to create a poller , which will send a request every five seconds to the /results/<job_key> endpoint asking for updates. Once the data is available, we’ll add it to the DOM.

This is a collaboration piece between Cam Linke, co-founder of Startup Edmonton , and the folks at Real Python

🐍 Python Tricks 💌

Get a short & sweet Python Trick delivered to your inbox every couple of days. No spam ever. Unsubscribe any time. Curated by the Real Python team.

About The Team

Each tutorial at Real Python is created by a team of developers so that it meets our high quality standards. The team members who worked on this tutorial are:

Master Real-World Python Skills With Unlimited Access to Real Python

Join us and get access to thousands of tutorials, hands-on video courses, and a community of expert Pythonistas:

Join us and get access to thousands of tutorials, hands-on video courses, and a community of expert Pythonistas:

What Do You Think?

What’s your #1 takeaway or favorite thing you learned? How are you going to put your newfound skills to use? Leave a comment below and let us know.

Commenting Tips: The most useful comments are those written with the goal of learning from or helping out other students. Get tips for asking good questions and get answers to common questions in our support portal . Looking for a real-time conversation? Visit the Real Python Community Chat or join the next “Office Hours” Live Q&A Session . Happy Pythoning!

Keep Learning

Related Topics: databases flask web-dev

Keep reading Real Python by creating a free account or signing in:

Already have an account? Sign-In

Almost there! Complete this form and click the button below to gain instant access:

Free Flask Video Tutorial: Build a Python + Flask Web App, From Scratch

🔒 No spam. We take your privacy seriously.

Use Python to build your side business with the Python for Entrepreneurs video course!

Task queues

Task queues manage background work that must be executed outside the usual HTTP request-response cycle.

Why are task queues necessary?

Tasks are handled asynchronously either because they are not initiated by an HTTP request or because they are long-running jobs that would dramatically reduce the performance of an HTTP response.

For example, a web application could poll the GitHub API every 10 minutes to collect the names of the top 100 starred repositories. A task queue would handle invoking code to call the GitHub API, process the results and store them in a persistent database for later use.

Another example is when a database query would take too long during the HTTP request-response cycle. The query could be performed in the background on a fixed interval with the results stored in the database. When an HTTP request comes in that needs those results a query would simply fetch the precalculated result instead of re-executing the longer query. This precalculation scenario is a form of caching enabled by task queues.

Other types of jobs for task queues include

spreading out large numbers of independent database inserts over time instead of inserting everything at once

aggregating collected data values on a fixed interval, such as every 15 minutes

scheduling periodic jobs such as batch processes

Task queue projects

The defacto standard Python task queue is Celery. The other task queue projects that arise tend to come from the perspective that Celery is overly complicated for simple use cases. My recommendation is to put the effort into Celery's reasonable learning curve as it is worth the time it takes to understand how to use the project.

The Celery distributed task queue is the most commonly used Python library for handling asynchronous tasks and scheduling.

The RQ (Redis Queue) is a simple Python library for queueing jobs and processing them in the background with workers. RQ is backed by Redis and is designed to have a low barrier to entry. The intro post contains information on design decisions and how to use RQ.

Taskmaster is a lightweight simple distributed queue for handling large volumes of one-off tasks.

Huey is a simple task queue that uses Redis on the backend but otherwise does not depend on other libraries. The project was previously known as Invoker and the author changed the name.

Huey is a Redis-based task queue that aims to provide a simple, yet flexible framework for executing tasks. Huey supports task scheduling, crontab-like repeating tasks, result storage and automatic retry in the event of failure.

Hosted message and task queue services

Task queue third party services aim to solve the complexity issues that arise when scaling out a large deployment of distributed task queues.

Iron.io is a distributed messaging service platform that works with many types of task queues such as Celery. It also is built to work with other IaaS and PaaS environments such as Amazon Web Services and Heroku.

Amazon Simple Queue Service (SQS) is a set of five APIs for creating, sending, receiving, modifying and deleting messages.

CloudAMQP is at its core managed servers with RabbitMQ installed and configured. This service is an option if you are using RabbitMQ and do not want to maintain RabbitMQ installations on your own servers.

Open source examples that use task queues

Take a look at the code in this open source Flask application and this Django application for examples of how to use and deploy Celery with a Redis broker to send text messages with these frameworks.

flask-celery-example is a simple Flask application with Celery as a task queue and Redis as the broker.

Task queue resources

Getting Started Scheduling Tasks with Celery is a detailed walkthrough for setting up Celery with Django (although Celery can also be used without a problem with other frameworks).

International Space Station notifications with Python and Redis Queue (RQ) shows how to combine the RQ task queue library with Flask to send text message notifications every time a condition is met - in this blog post's case that the ISS is currently flying over your location on Earth.

Evaluating persistent, replicated message queues is a detailed comparison of Amazon SQS, MongoDB, RabbitMQ, HornetQ and Kafka's designs and performance.

Queues.io is a collection of task queue systems with short summaries for each one. The task queues are not all compatible with Python but ones that work with it are tagged with the "Python" keyword.

Why Task Queues is a presentation for what task queues are and why they are needed.

Flask by Example Implementing a Redis Task Queue provides a detailed walkthrough of setting up workers to use RQ with Redis.

How to use Celery with RabbitMQ is a detailed walkthrough for using these tools on an Ubuntu VPS.

Heroku has a clear walkthrough for using RQ for background tasks .

Introducing Celery for Python+Django provides an introduction to the Celery task queue.

Celery - Best Practices explains things you should not do with Celery and shows some underused features for making task queues easier to work with.

The "Django in Production" series by Rob Golding contains a post specifically on Background Tasks .

Asynchronous Processing in Web Applications Part One and Part Two are great reads for understanding the difference between a task queue and why you shouldn't use your database as one.

Celery in Production on the Caktus Group blog contains good practices from their experience using Celery with RabbitMQ, monitoring tools and other aspects not often discussed in existing documentation.

A 4 Minute Intro to Celery is a short introductory task queue screencast.

Heroku wrote about how to secure Celery when tasks are otherwise sent over unencrypted networks.

Miguel Grinberg wrote a nice post on using the task queue Celery with Flask . He gives an overview of Celery followed by specific code to set up the task queue and integrate it with Flask.

3 Gotchas for Working with Celery are things to keep in mind when you're new to the Celery task queue implementation.

Deferred Tasks and Scheduled Jobs with Celery 3.1, Django 1.7 and Redis is a video along with code that shows how to set up Celery with Redis as the broker in a Django application.

Setting up an asynchronous task queue for Django using Celery and Redis is a straightforward tutorial for setting up the Celery task queue for Django web applications using the Redis broker on the back end.

Background jobs with Django and Celery shows the code and a simple explanation of how to use Celery with Django .

Asynchronous Tasks With Django and Celery shows how to integrate Celery with Django and create Periodic Tasks.

Three quick tips from two years with Celery provides some solid advice on retry delays, the -Ofair flag and global task timeouts for Celery.

Task queue learning checklist

Pick a slow function in your project that is called during an HTTP request.

Determine if you can precompute the results on a fixed interval instead of during the HTTP request. If so, create a separate function you can call from elsewhere then store the precomputed value in the database.

Read the Celery documentation and the links in the resources section below to understand how the project works.

Install a message broker such as RabbitMQ or Redis and then add Celery to your project. Configure Celery to work with the installed message broker.

Use Celery to invoke the function from step one on a regular basis.

Have the HTTP request function use the precomputed value instead of the slow running code it originally relied upon.

What's next to learn after task queues?

How do I log errors that occur in my application?

I want to learn more about app users via web analytics.

What tools exist for monitoring a deployed web app?

Sign up here to receive a monthly email with major updates to this site, tutorials and discount codes for Python books.

Searching for a complete, step-by-step deployment walkthrough? Learn more about The Full Stack Python Guide to Deployments book .

Email Updates

Sign up to get a monthly email with python tutorials and major updates to this site., table of contents, task queues.

• Contributing

RQ ( Redis Queue ) is a simple Python library for queueing jobs and processing them in the background with workers. It is backed by Redis and it is designed to have a low barrier to entry. It can be integrated in your web stack easily.

RQ requires Redis >= 3.0.0.

Getting started

First, run a Redis server. You can use an existing one. To put jobs on queues, you don’t have to do anything special, just define your typically lengthy or blocking function:

Then, create a RQ queue:

And enqueue the function call:

Scheduling jobs are similarly easy:

You can also ask RQ to retry failed jobs:

To start executing enqueued function calls in the background, start a worker from your project’s directory:

That’s about it.

Installation

Simply use the following command to install the latest released version:

If you want the cutting edge version (that may well be broken), use this:

Project history

This project has been inspired by the good parts of Celery , Resque and this snippet , and has been created as a lightweight alternative to existing queueing frameworks, with a low barrier to entry.

Using Python RQ for Task Queues in Python

This is a getting started on python-rq tutorial and I will demonstrate how to work with asynchronous tasks using python redis queue (python-rq).

What will we be doing

We want a client to submit 1000's of jobs in a non-blocking asynchronous fashion, and then we will have workers which will consume these jobs from our redis queue, and process those tasks at the rate of what our consumer can handle.

The nice thing about this is that, if our consumer is unavailable for processing the tasks will remain in the queue and once the consumer is ready to consume, the tasks will be executed. It's also nice that its asynchronous, so the client don't have to wait until the task has finished.

We will run a redis server using docker, which will be used to queue all our jobs, then we will go through the basics in python and python-rq such as:

• Writing a Task
• Enqueueing a Job
• Getting information from our queue, listing jobs, job statuses
• Running our workers to consume from the queue and action our tasks
• Basic application which queues jobs to the queue, consumes and action them and monitors the queue

Redis Server

You will require docker for this next step, to start the redis server:

Install python-rq:

Create the task which will be actioned by our workers, in our case it will just be a simple function that adds all the numbers from a given string to a list, then adds them up and return the total value.

This is however a very basic task, but its just for demonstration.

Our tasks.py :

To test this locally:

Now, lets import redis and redis-queue, with our tasks and instantiate a queue object:

Submit a Task to the Queue

Let's submit a task to the queue:

We have a couple of properties from result which we can inspect, first let's have a look at the id that we got back when we submitted our task to the queue:

We can also get the status from our task:

We can also view our results in json format:

If we dont have context of the job id, we can use get_jobs to get all the jobs which is queued:

Then we can loop through the results and get the id like below:

Or to get the job id's in a list:

Since we received the job id, we can use fetch_job to get more info about the job:

And as before we can view it in json format:

We can also view the key in redis by passing the job_id:

To view how many jobs are in our queue, we can either do:

Consuming from the Queue

Now that our task is queued, let's fire of our worker to consume the job from the queue and action the task:

Now, when we get the status of our job, you will see that it finished:

And to get the result from our worker:

And like before, if you dont have context of your job id, you can get the job id, then return the result:

Naming Queues

We can namespace our tasks into specific queues, for example if we want to create queue1 :

To verify the queue name:

As we can see our queue is empty:

Let's submit 10 jobs to our queue:

To verify the number of jobs in our queue:

And to count them:

Cleaning the Queue

Cleaning the queue can either be done with:

Then to verify that our queue is clean:

Naming Workers

The same way that we defined a name for our queue, we can define a name for our workers:

Which means you can have different workers consuming jobs from specific queues.

Documentation:

• https://python-rq.org/docs/
• https://python-rq.org/docs/workers/
• https://python-rq.org/docs/monitoring/

Thanks for reading, feel free to check out my website , and subscrube to my newsletter or follow me at @ruanbekker on Twitter.

• Linktree: https://go.ruan.dev/links
• Patreon: https://go.ruan.dev/patreon

Navigation Menu

Search code, repositories, users, issues, pull requests..., provide feedback.

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly.

To see all available qualifiers, see our documentation .

• Notifications You must be signed in to change notification settings

Python task queue using Redis

closeio/tasktiger

Folders and files.

Repository files navigation

TaskTiger is a Python task queue using Redis.

(Interested in working on projects like this? Close is looking for great engineers to join our team)

Quick start

Configuration, task decorator, task options, custom retrying, inspect, requeue and delete tasks, pause queue processing, rollbar error handling, cleaning up error'd tasks, running the test suite, releasing a new version.

Per-task fork or synchronous worker

By default, TaskTiger forks a subprocess for each task, This comes with several benefits: Memory leaks caused by tasks are avoided since the subprocess is terminated when the task is finished. A hard time limit can be set for each task, after which the task is killed if it hasn't completed. To ensure performance, any necessary Python modules can be preloaded in the parent process.

TaskTiger also supports synchronous workers, which allows for better performance due to no forking overhead, and tasks have the ability to reuse network connections. To prevent memory leaks from accumulating, workers can be set to shutdown after a certain amount of time, at which point a supervisor can restart them. Workers also automatically exit on on hard timeouts to prevent an inconsistent process state.

Unique queues

TaskTiger has the option to avoid duplicate tasks in the task queue. In some cases it is desirable to combine multiple similar tasks. For example, imagine a task that indexes objects (e.g. to make them searchable). If an object is already present in the task queue and hasn't been processed yet, a unique queue will ensure that the indexing task doesn't have to do duplicate work. However, if the task is already running while it's queued, the task will be executed another time to ensure that the indexing task always picks up the latest state.

TaskTiger can ensure to never execute more than one instance of tasks with similar arguments by acquiring a lock. If a task hits a lock, it is requeued and scheduled for later executions after a configurable interval.

Task retrying

TaskTiger lets you retry exceptions (all exceptions or a list of specific ones) and comes with configurable retry intervals (fixed, linear, exponential, custom).

Flexible queues

Tasks can be easily queued in separate queues. Workers pick tasks from a randomly chosen queue and can be configured to only process specific queues, ensuring that all queues are processed equally. TaskTiger also supports subqueues which are separated by a period. For example, you can have per-customer queues in the form process_emails.CUSTOMER_ID and start a worker to process process_emails and any of its subqueues. Since tasks are picked from a random queue, all customers get equal treatment: If one customer is queueing many tasks it can't block other customers' tasks from being processed. A maximum queue size can also be enforced.

Batch queues

Batch queues can be used to combine multiple queued tasks into one. That way, your task function can process multiple sets of arguments at the same time, which can improve performance. The batch size is configurable.

Scheduled and periodic tasks

Tasks can be scheduled for execution at a specific time. Tasks can also be executed periodically (e.g. every five seconds).

Structured logging

TaskTiger supports JSON-style logging via structlog, allowing more flexibility for tools to analyze the log. For example, you can use TaskTiger together with Logstash, Elasticsearch, and Kibana.

The structlog processor tasktiger.logging.tasktiger_processor can be used to inject the current task id into all log messages.

Reliability

TaskTiger atomically moves tasks between queue states, and will re-execute tasks after a timeout if a worker crashes.

Error handling

If an exception occurs during task execution and the task is not set up to be retried, TaskTiger stores the execution tracebacks in an error queue. The task can then be retried or deleted manually. TaskTiger can be easily integrated with error reporting services like Rollbar.

Admin interface

A simple admin interface using flask-admin exists as a separate project ( tasktiger-admin ).

It is easy to get started with TaskTiger.

Create a file that contains the task(s).

Queue the task using the delay method.

Run a worker (make sure the task code can be found, e.g. using PYTHONPATH ).

A TaskTiger object keeps track of TaskTiger's settings and is used to decorate and queue tasks. The constructor takes the following arguments:

Redis connection object. The connection should be initialized with decode_responses=True to avoid encoding problems on Python 3.

Dict with config options. Most configuration options don't need to be changed, and a full list can be seen within TaskTiger 's __init__ method.

Here are a few commonly used options:

ALWAYS_EAGER

If set to True , all tasks except future tasks ( when is a future time) will be executed locally by blocking until the task returns. This is useful for testing purposes.

BATCH_QUEUES

Set up queues that will be processed in batch, i.e. multiple jobs are taken out of the queue at the same time and passed as a list to the worker method. Takes a dict where the key represents the queue name and the value represents the batch size. Note that the task needs to be declared as batch=True . Also note that any subqueues will be automatically treated as batch queues, and the batch value of the most specific subqueue name takes precedence.

ONLY_QUEUES

If set to a non-empty list of queue names, a worker only processes the given queues (and their subqueues), unless explicit queues are passed to the command line.

setup_structlog

If set to True, sets up structured logging using structlog when initializing TaskTiger. This makes writing custom worker scripts easier since it doesn't require the user to set up structlog in advance.

TaskTiger provides a task decorator to specify task options. Note that simple tasks don't need to be decorated. However, decorating the task allows you to use an alternative syntax to queue the task, which is compatible with Celery:

Tasks support a variety of options that can be specified either in the task decorator, or when queueing a task. For the latter, the delay method must be called on the TaskTiger object, and any options in the task decorator are overridden.

When queueing a task, the task needs to be defined in a module other than the Python file which is being executed. In other words, the task can't be in the __main__ module. TaskTiger will give you back an error otherwise.

The following options are supported by both delay and the task decorator:

Name of the queue where the task will be queued.

hard_timeout

If the task runs longer than the given number of seconds, it will be killed and marked as failed.

Boolean to indicate whether the task will only be queued if there is no similar task with the same function, arguments, and keyword arguments in the queue. Note that multiple similar tasks may still be executed at the same time since the task will still be inserted into the queue if another one is being processed. Requeueing an already scheduled unique task will not change the time it was originally scheduled to execute at.

If set, this implies unique=True and specifies the list of kwargs to use to construct the unique key. By default, all args and kwargs are serialized and hashed.

Boolean to indicate whether to hold a lock while the task is being executed (for the given args and kwargs). If a task with similar args/kwargs is queued and tries to acquire the lock, it will be retried later.

If set, this implies lock=True and specifies the list of kwargs to use to construct the lock key. By default, all args and kwargs are serialized and hashed.

max_queue_size

A maximum queue size can be enforced by setting this to an integer value. The QueueFullException exception will be raised when queuing a task if this limit is reached. Tasks in the active , scheduled , and queued states are counted against this limit.

Takes either a datetime (for an absolute date) or a timedelta (relative to now). If given, the task will be scheduled for the given time.

Boolean to indicate whether to retry the task when it fails (either because of an exception or because of a timeout). To restrict the list of failures, use retry_on . Unless retry_method is given, the configured DEFAULT_RETRY_METHOD is used.

If a list is given, it implies retry=True . The task will be only retried on the given exceptions (or its subclasses). To retry the task when a hard timeout occurs, use JobTimeoutException .

retry_method

If given, implies retry=True . Pass either:

• a function that takes the retry number as an argument, or,
• a tuple (f, args) , where f takes the retry number as the first argument, followed by the additional args.

The function needs to return the desired retry interval in seconds, or raise StopRetry to stop retrying. The following built-in functions can be passed for common scenarios and return the appropriate tuple:

fixed(delay, max_retries)

Returns a method that returns the given delay (in seconds) or raises StopRetry if the number of retries exceeds max_retries .

linear(delay, increment, max_retries)

Like fixed , but starts off with the given delay and increments it by the given increment after every retry.

exponential(delay, factor, max_retries)

Like fixed , but starts off with the given delay and multiplies it by the given factor after every retry.

For example, to retry a task 3 times (for a total of 4 executions), and wait 60 seconds between executions, pass retry_method=fixed(60, 3) .

runner_class

If given, a Python class can be specified to influence task running behavior. The runner class should inherit tasktiger.runner.BaseRunner and implement the task execution behavior. The default implementation is available in tasktiger.runner.DefaultRunner . The following behavior can be achieved:

Execute specific code before or after the task is executed (in the forked child process), or customize the way task functions are called in either single or batch processing.

Note that if you want to execute specific code for all tasks, you should use the CHILD_CONTEXT_MANAGERS configuration option.

Control the hard timeout behavior of a task.

Execute specific code in the main worker process after a task failed permanently.

This is an advanced feature and the interface and requirements of the runner class can change in future TaskTiger versions.

The following options can be only specified in the task decorator:

If set to True , the task will receive a list of dicts with args and kwargs and can process multiple tasks of the same type at once. Example: [{"args": [1], "kwargs": {}}, {"args": [2], "kwargs": {}}] Note that the list will only contain multiple items if the worker has set up BATCH_QUEUES for the specific queue (see the Configuration section).

If given, makes a task execute periodically. Pass either:

• a function that takes the current datetime as an argument.
• a tuple (f, args) , where f takes the current datetime as the first argument, followed by the additional args.

The schedule function must return the next task execution datetime, or None to prevent periodic execution. The function is executed to determine the initial task execution date when a worker is initialized, and to determine the next execution date when the task is about to get executed.

For most common scenarios, the below mentioned built-in functions can be passed:

periodic(seconds=0, minutes=0, hours=0, days=0, weeks=0, start_date=None, end_date=None)

Use equal, periodic intervals, starting from start_date (defaults to 2000-01-01T00:00Z , a Saturday, if not given), ending at end_date (or never, if not given). For example, to run a task every five minutes indefinitely, use schedule=periodic(minutes=5) . To run a task every every Sunday at 4am UTC, you could use schedule=periodic(weeks=1, start_date=datetime.datetime(2000, 1, 2, 4)) .

cron_expr(expr, start_date=None, end_date=None)

start_date , to specify the periodic task start date. It defaults to 2000-01-01T00:00Z , a Saturday, if not given. end_date , to specify the periodic task end date. The task repeats forever if end_date is not given. For example, to run a task every hour indefinitely, use schedule=cron_expr("0 * * * *") . To run a task every Sunday at 4am UTC, you could use schedule=cron_expr("0 4 * * 0") .

In some cases the task retry options may not be flexible enough. For example, you might want to use a different retry method depending on the exception type, or you might want to like to suppress logging an error if a task fails after retries. In these cases, RetryException can be raised within the task function. The following options are supported:

Specify a custom retry method for this retry. If not given, the task's default retry method is used, or, if unspecified, the configured DEFAULT_RETRY_METHOD . Note that the number of retries passed to the retry method is always the total number of times this method has been executed, regardless of which retry method was used.

original_traceback

If RetryException is raised from within an except block and original_traceback is True, the original traceback will be logged (i.e. the stacktrace at the place where the caught exception was raised). False by default.

If set to False and the task fails permanently, a warning will be logged instead of an error, and the task will be removed from Redis when it completes. True by default.

Example usage:

The tasktiger command is used on the command line to invoke a worker. To invoke multiple workers, multiple instances need to be started. This can be easily done e.g. via Supervisor. The following Supervisor configuration file can be placed in /etc/supervisor/tasktiger.ini and runs 4 TaskTiger workers as the ubuntu user. For more information, read Supervisor's documentation.

Workers support the following options:

-q , --queues

If specified, only the given queue(s) are processed. Multiple queues can be separated by comma. Any subqueues of the given queues will be also processed. For example, -q first,second will process items from first , second , and subqueues such as first.CUSTOMER1 , first.CUSTOMER2 .

-e , --exclude-queues

If specified, exclude the given queue(s) from processing. Multiple queues can be separated by comma. Any subqueues of the given queues will also be excluded unless a more specific queue is specified with the -q option. For example, -q email,email.incoming.CUSTOMER1 -e email.incoming will process items from the email queue and subqueues like email.outgoing.CUSTOMER1 or email.incoming.CUSTOMER1 , but not email.incoming or email.incoming.CUSTOMER2 .

-m , --module

Module(s) to import when launching the worker. This improves task performance since the module doesn't have to be reimported every time a task is forked. Multiple modules can be separated by comma.

Another way to preload modules is to set up a custom TaskTiger launch script, which is described below.

-h , --host

Redis server hostname (if different from localhost ).

-p , --port

Redis server port (if different from 6379 ).

-a , --password

Redis server password (if required).

Redis server database number (if different from 0 ).

-M , --max-workers-per-queue

Maximum number of workers that are allowed to process a queue.

--store-tracebacks/--no-store-tracebacks

Store tracebacks with execution history (config defaults to True ).

Can be fork (default) or sync . Whether to execute tasks in a separate process via fork, or execute them synchronously in the same proces. See "Features" section for the benefits of either approach.

--exit-after

Exit the worker after the time in minutes has elapsed. This is mainly useful with the synchronous executor to prevent memory leaks from accumulating.

In some cases it is convenient to have a custom TaskTiger launch script. For example, your application may have a manage.py command that sets up the environment and you may want to launch TaskTiger workers using that script. To do that, you can use the run_worker_with_args method, which launches a TaskTiger worker and parses any command line arguments. Here is an example:

TaskTiger provides access to the Task class which lets you inspect queues and perform various actions on tasks.

Each queue can have tasks in the following states:

• queued : Tasks that are queued and waiting to be picked up by the workers.
• active : Tasks that are currently being processed by the workers.
• scheduled : Tasks that are scheduled for later execution.
• error : Tasks that failed with an error.

To get a list of all tasks for a given queue and state, use Task.tasks_from_queue . The method gives you back a tuple containing the total number of tasks in the queue (useful if the tasks are truncated) and a list of tasks in the queue, latest first. Using the skip and limit keyword arguments, you can fetch arbitrary slices of the queue. If you know the task ID, you can fetch a given task using Task.from_id . Both methods let you load tracebacks from failed task executions using the load_executions keyword argument, which accepts an integer indicating how many executions should be loaded.

Tasks can also be constructed and queued using the regular constructor, which takes the TaskTiger instance, the function name and the options described in the Task options section. The task can then be queued using its delay method. Note that the when argument needs to be passed to the delay method, if applicable. Unique tasks can be reconstructed using the same arguments.

The Task object has the following properties:

• id : The task ID.
• data : The raw data as a dict from Redis.
• executions : A list of failed task executions (as dicts). An execution dict contains the processing time in time_started and time_failed , the worker host in host , the exception name in exception_name and the full traceback in traceback .
• serialized_func , args , kwargs : The serialized function name with all of its arguments.
• func : The imported (executable) function

The Task object has the following methods:

• cancel : Cancel a scheduled task.
• delay : Queue the task for execution.
• delete : Remove the task from the error queue.
• execute : Run the task without queueing it.
• n_executions : Queries and returns the number of past task executions.
• retry : Requeue the task from the error queue for execution.
• update_scheduled_time : Updates a scheduled task's date to the given date.

The current task can be accessed within the task function while it's being executed: In case of a non-batch task, the current_task property of the TaskTiger instance returns the current Task instance. In case of a batch task the current_tasks property must be used which returns a list of tasks that are currently being processed (in the same order as they were passed to the task).

Example 1: Queueing a unique task and canceling it without a reference to the original task.

Example 2: Inspecting queues and retrying a task by ID.

Example 3: Accessing the task instances within a batch task function to determine how many times the currently processing tasks were previously executed.

The --max-workers-per-queue option uses queue locks to control the number of workers that can simultaneously process the same queue. When using this option a system lock can be placed on a queue which will keep workers from processing tasks from that queue until it expires. Use the set_queue_system_lock() method of the TaskTiger object to set this lock.

TaskTiger comes with Rollbar integration for error handling. When a task errors out, it can be logged to Rollbar, grouped by queue, task function name and exception type. To enable logging, initialize rollbar with the StructlogRollbarHandler provided in the tasktiger.rollbar module. The handler takes a string as an argument which is used to prefix all the messages reported to Rollbar. Here is a custom worker launch script:

Error'd tasks occasionally need to be purged from Redis, so TaskTiger exposes a purge_errored_tasks method to help. It might be useful to set this up as a periodic task as follows:

Tests can be run locally using the provided docker compose file. After installing docker, tests should be runnable with:

Tests can be more granularly run using normal pytest flags. For example:

• Make sure the code has been thoroughly reviewed and tested in a realistic production environment.
• Update setup.py and CHANGELOG.md . Make sure you include any breaking changes.
• Run python setup.py sdist and twine upload dist/<PACKAGE_TO_UPLOAD> .
• Push a new tag pointing to the released commit, format: v0.13 for example.
• Mark the tag as a release in GitHub's UI and include in the description the changelog entry for the version. An example would be: https://github.com/closeio/tasktiger/releases/tag/v0.13 .

Releases 17

Contributors 30.

• Python 98.2%
• Dockerfile 0.1%

How to Run Your First Task with RQ, Redis, and Python

Time to read: 7 minutes

• Facebook logo
• Twitter Logo Follow us on Twitter
• LinkedIn logo

As a developer, it can be very useful to learn how to run functions in the background while being able to monitor the queue in another tab or different system. This is incredibly helpful when managing heavy workloads that might not work efficiently when called all at once, or when making large numbers of calls to a database that returns data slowly over time rather than all at once.

In this tutorial we will implement a RQ queue in Python with the help of Redis to schedule and execute tasks in a timely manner.

Tutorial Requirements

• Python 3.6  or newer. If your operating system does not provide a Python interpreter, you can go to python.org  to download an installer.

Let’s talk about task queues

Task queues are a great way to allow tasks to work asynchronously outside of the main application flow. There are many task queues in Python to assist you in your project, however, we’ll be discussing a solution today known as RQ.

RQ, also known as Redis Queue , is a Python library that allows developers to enqueue jobs to be processed in the background with workers . The RQ workers will be called when it's time to execute the queue in the background. Using a connection to Redis , it’s no surprise that this library is super lightweight and offers support for those getting started for the first time.

By using this particular task queue, it is possible to process jobs in the background with little to no hassle.

Set up the environment

Create a project directory in your terminal called “rq-test” to follow along.

Install a virtual environment and copy and paste the commands to install rq and related packages. If you are using a Unix or MacOS system, enter the following commands:

If you are on a Windows machine, enter the following commands in a prompt window:

RQ requires a Redis installation  on your machine which can be done using the following commands using wget . Redis is on version 6.0.6 at the time of this article publication.

If you are using a Unix or MacOS system, enter these commands to install Redis. This is my personal favorite way to install Redis, but there are alternatives below:

If you have Homebrew installed, you can type brew install redis in the terminal and refer to this GitHub gist to install Redis on the Mac . For developers using Ubuntu Linux, the command sudo apt-get install redis would get the job done as well.

Run the Redis server in a separate terminal window on the default port with the command src/redis-server from the directory where it's installed.

For Windows users, you would have to follow a separate tutorial to run Redis on Windows . Download the latest zip file on GitHub  and extract the contents. Run the redis-server.exe file that was extracted from the zip file to start the Redis server.

The output should look similar to the following after running Redis:

Build out the tasks

In this case, a task for Redis Queue is merely a Python function. For this article, we’ll tell the task to print a message to the terminal for a “x” amount of seconds to demonstrate the use of RQ.

Copy and paste the following code to a file named “tasks.py” in your directory.

These are simple tasks that print out numbers and text on the terminal so that we can see if the tasks are executed properly. Using the time.sleep(1) function from the Python time library  will allow your task to be suspended for the given number of seconds and overall extend the time of the task so that we can examine their progress.

Feel free to alter this code after the tutorial and create your own tasks. Some other popular tasks are sending a fax message or email by connecting to your email client.

Create your queue

Create another file in the root directory and name it “app.py”. Copy and paste the following code:

The queue object sets up a connection to Redis and initializes a queue based on that connection. This queue can hold all the jobs required to run in the background with workers.

As seen in the code, the tasks.print_task function is added using the enqueue function. This means that the task added to the queue will be executed immediately

The enqueue_in function is another nifty RQ function because it expects a timedelta in order to schedule the specified job. In this case, seconds is specified, but this variable can be changed according to the time schedule expected for your usage. Check out other ways to schedule a job  on this GitHub README.

Since we are testing out the RQ queue, I have enqueued both the tasks.print_task and tasks.print_numbers functions so that we can see their output on the terminal. The third argument passed in is a "5" which also stands for the argument passed into the respective functions. In this case, we are expecting to see print_task() print "Hello World!" five times and for print_numbers() to print 5 numbers in order.

If you have created any additional task, be sure to import your tasks at the top of the file so that all the tasks in your Python file can be accessed.

Run the queue

For the purposes of this article, the gif demo below will show a perfect execution of the tasks in queue so no exceptions will be raised.

The Redis server should still be running in a tab from earlier in the tutorial at this point. If it stopped, run the command src/redis-server inside the redis-6.0.6 folder on one tab, or for developers with a Windows machine, start redis-cli.exe . Open another tab solely to run the RQ scheduler with the command rq worker --with-scheduler .

This should be the output after running the command above.

The worker command activated a worker process in order to connect to Redis and look for any jobs assigned to the queue from the code in app.py .

Lastly, open a third tab in the terminal for the root project directory. Start up the virtual environment again with the command source venv/bin/activate . Then type python app.py to run the project.

Go back to the tab that is running rq worker --with-scheduler . Wait 5 more seconds after the first task is executed to see the next task. Although the live demo gif below wasn’t able to capture the best timing due to having to run the program and record, it is noticeable that there was a pause between tasks until execution and that both tasks were completed within 15 seconds.

Here’s the sample output inside of the rqworker tab:

As seen in the output above, if the tasks written in task.py  had a line to return anything, then the result of both tasks are kept for 500 seconds which is the default. A developer can  alter the return value's time to live by passing in a result_ttl parameter when adding tasks to the queue.

Handle exceptions and try again

If a job were to fail, you can always set up a log to keep track of the error messages, or you can use the RQ queue to enqueue and retry failed jobs. By using RQ's FailedJobRegistry package, you can keep track of the jobs that failed during runtime. The RQ documentation discusses how it handles the exceptions  and how data regarding the job can help the developer figure out how to resubmit the job.  

However, RQ also supports developers in handling exceptions in their own way by injecting your own logic to the rq workers . This may be a helpful option for you if you are executing many tasks in your project and those that failed are not worth retrying.

Force a failed task to retry

Since this is an introductory article to run your first task with RQ, let's try to purposely fail one of the tasks from earlier to test out RQ's retry object.

Go to the tasks.py  file and alter the print_task() function so that random numbers can be generated and determine if the function will be executed or not. We will be using the random Python library  to assist us in generating numbers. Don't forget to include the import random at the top of the file.

Copy and paste the following lines of code to change the print_task() function in the tasks.py  file.

Go back to the app.py  file to change the queue. Instead of using the enqueue_in function to execute the tasks.print_task function, delete the line and replace it with queue.enqueue(tasks.print_task, 5, retry=Retry(max=2)) .

The retry object is imported with rq so make sure you add from rq import Retry at the top of the file as well in order to use this functionality. This object accepts max and interval arguments to specify when the particular function will be retried. In the newly changed line, the tasks.print_task function will pass in the function we want to retry, the argument parameter "5" which stands for the seconds of execution, and lastly the maximum amount of times we want the queue to retry.

The tasks in queue should now look like this:

When running the print_task task, there is a 50/50 chance that tasks.print_task() will execute properly since we're only generating a 1 or 2, and the print statement will only happen if you generate a 1. A RuntimeError will be raised otherwise and the queue will retry the task immediately as many times as it takes to successfully print "Hello World!".

What’s next for task queues?

Congratulations! You have successfully learned and implemented the basics of scheduling tasks in the RQ queue. Perhaps now you can tell the worker command to add a task that prints out an infinite number of "Congratulations" messages in a timely manner!

Otherwise, check out these different tasks that you can build in to your Redis Queue:

• Schedule Twilio SMS to a list of contacts  quickly!
• Use Redis Queue to generate a fan fiction with OpenAI GPT-3
• Queue Emails with Twilio SendGrid using Redis Queue

Let me know what you have been building by reaching out to me over email!

Diane Phan is a developer on the Developer Voices team. She loves to help programmers tackle difficult challenges that might prevent them from bringing their projects to life. She can be reached at dphan [at] twilio.com or LinkedIn .

Related Posts

Related Resources

Twilio docs, from apis to sdks to sample apps.

API reference documentation, SDKs, helper libraries, quickstarts, and tutorials for your language and platform.

Resource Center

The latest ebooks, industry reports, and webinars.

Learn from customer engagement experts to improve your own communication.

Twilio's developer community hub

Best practices, code samples, and inspiration to build communications and digital engagement experiences.

Celery is a task queue implementation for Python web applications used to asynchronously execute work outside the HTTP request-response cycle.

Why is Celery useful?

Task queues and the Celery implementation in particular are one of the trickier parts of a Python web application stack to understand.

If you are a junior developer it can be unclear why moving work outside the HTTP request-response cycle is important. In short, you want your WSGI server to respond to incoming requests as quickly as possible because each request ties up a worker process until the response is finished. Moving work off those workers by spinning up asynchronous jobs as tasks in a queue is a straightforward way to improve WSGI server response times.

What's the difference between Celeryd and Celerybeat?

Celery can be used to run batch jobs in the background on a regular schedule. A key concept in Celery is the difference between the Celery daemon (celeryd), which executes tasks, Celerybeat, which is a scheduler. Think of Celeryd as a tunnel-vision set of one or more workers that handle whatever tasks you put in front of them. Each worker will perform a task and when the task is completed will pick up the next one. The cycle will repeat continously, only waiting idly when there are no more tasks to put in front of them.

Celerybeat on the other hand is like a boss who keeps track of when tasks should be executed. Your application can tell Celerybeat to execute a task at time intervals, such as every 5 seconds or once a week. Celerybeat can also be instructed to run tasks on a specific date or time, such as 5:03pm every Sunday. When the interval or specific time is hit, Celerybeat will hand the job over to Celeryd to execute on the next available worker.

Celery tutorials and advice

Celery is a powerful tool that can be difficult to wrap your mind around at first. Be sure to read up on task queue concepts then dive into these specific Celery tutorials.

A 4 Minute Intro to Celery is a short introductory task queue screencast.

This blog post series on Celery's architecture , Celery in the wild: tips and tricks to run async tasks in the real world and dealing with resource-consuming tasks on Celery provide great context for how Celery works and how to handle some of the trickier bits to working with the task queue.

How to use Celery with RabbitMQ is a detailed walkthrough for using these tools on an Ubuntu VPS.

Celery - Best Practices explains things you should not do with Celery and shows some underused features for making task queues easier to work with.

Celery Best Practices is a different author's follow up to the above best practices post that builds upon some of his own learnings from 3+ years using Celery.

Common Issues Using Celery (And Other Task Queues) contains good advice about mistakes to avoid in your task configurations, such as database transaction usage and retrying failed tasks.

Asynchronous Processing in Web Applications Part One and Part Two are great reads for understanding the difference between a task queue and why you shouldn't use your database as one.

My Experiences With A Long-Running Celery-Based Microprocess gives some good tips and advice based on experience with Celery workers that take a long time to complete their jobs.

Checklist to build great Celery async tasks is a site specifically designed to give you a list of good practices to follow as you design your task queue configuration and deploy to development, staging and production environments.

Heroku wrote about how to secure Celery when tasks are otherwise sent over unencrypted networks.

Unit testing Celery tasks explains three strategies for testing code within functions that Celery executes. The post concludes that calling Celery tasks synchronously to test them is the best strategy without any downsides. However, keep in mind that any testing method that is not the same as how the function will execute in a production environment can potentially lead to overlooked bugs. There is also an open source Git repository with all of the source code from the post.

Rollbar monitoring of Celery in a Django app explains how to use Rollbar to monitor tasks. Super useful when workers invariably die for no apparent reason.

3 Gotchas for Working with Celery are things to keep in mind when you're new to the Celery task queue implementation.

Dask and Celery compares Dask.distributed with Celery for Python projects. The post gives code examples to show how to execute tasks with either task queue.

Python+Celery: Chaining jobs? explains that Celery tasks should be dependent upon each other using Celery chains, not direct dependencies between tasks.

Celery with web frameworks

Celery is typically used with a web framework such as Django , Flask or Pyramid . These resources show you how to integrate the Celery task queue with the web framework of your choice.

How to Use Celery and RabbitMQ with Django is a great tutorial that shows how to both install and set up a basic task with Django.

Miguel Grinberg wrote a nice post on using the task queue Celery with Flask . He gives an overview of Celery followed by specific code to set up the task queue and integrate it with Flask.

Setting up an asynchronous task queue for Django using Celery and Redis is a straightforward tutorial for setting up the Celery task queue for Django web applications using the Redis broker on the back end.

A Guide to Sending Scheduled Reports Via Email Using Django And Celery shows you how to use django-celery in your application. Note however there are other ways of integrating Celery with Django that do not require the django-celery dependency.

Flask asynchronous background tasks with Celery and Redis combines Celery with Redis as the broker and Flask for the example application's framework.

Celery and Django and Docker: Oh My! shows how to create Celery tasks for Django within a Docker container. It also provides some

Asynchronous Tasks With Django and Celery shows how to integrate Celery with Django and create Periodic Tasks.

Getting Started Scheduling Tasks with Celery is a detailed walkthrough for setting up Celery with Django (although Celery can also be used without a problem with other frameworks).

Asynchronous Tasks with Falcon and Celery configures Celery with the Falcon framework, which is less commonly-used in web tutorials.

Custom Celery task states is an advanced post on creating custom states, which is especially useful for transient states in your application that are not covered by the default Celery configuration.

Asynchronous Tasks with Django and Celery looks at how to configure Celery to handle long-running tasks in a Django app.

Celery deployment resources

Celery and its broker run separately from your web and WSGI servers so it adds some additional complexity to your deployments . The following resources walk you through how to handle deployments and get the right configuration settings in place.

• How to run celery as a daemon? is a short post with the minimal code for running the Celery daemon and Celerybeat as system services on Linux.

Do you want to learn more about task queues, or another topic?

How do I execute code outside the HTTP request-response cycle?

I've built a Python web app, now how do I deploy it?

What tools exist for monitoring a deployed web app?

Table of Contents

Full stack python.

• Skip to main content
• Skip to primary sidebar
• Skip to footer

Additional menu

Super Fast Python

making you awesome at concurrency

Queue task_done() and join() in Python

April 14, 2022 by Jason Brownlee in Python Threading

Last Updated on September 12, 2022

You can mark queue tasks done via task_done() and be notified when all tasks are done via join() .

In this tutorial you will discover how to use queue task done and join in Python .

Let’s get started.

Table of Contents

Need To Know When All Tasks are Done

A thread is a thread of execution in a computer program.

Every Python program has at least one thread of execution called the main thread. Both processes and threads are created and managed by the underlying operating system.

Sometimes we may need to create additional threads in our program in order to execute code concurrently.

Python provides the ability to create and manage new threads via the threading module and the threading.Thread class .

You can learn more about Python threads in the guide:

• Threading in Python: The Complete Guide

Threads can share data with each other using thread-safe queues, such as the queue.Queue class .

A problem when using queues is knowing when all items in the queue have been processed by consumer threads.

How can we know when all items have been processed in a queue?

Run loops using all CPUs, download your FREE book to learn how.

Why Care When All Tasks Are Done

There are many reasons why a thread may want to know when all tasks in a queue have been processed.

For example:

• A producer thread may want to wait until all work is done before adding new work.
• A producer thread may want to wait for all tasks to be done before sending a shutdown signal.
• A main thread may want to wait for all tasks to be done before terminating the program.

There are two aspects to this, they are:

• A thread blocking until tasks are done.
• A task being done is more than being retrieved from the queue.

Specifically, waiting means that the thread is blocked until the condition is met.

The condition of all tasks being processed is not only the case that all items put on the queue have been retrieved, but have been processed by the thread that retrieved them.

Next, let’s look at how we might do this in Python.

Download Now: Free Threading PDF Cheat Sheet

How to Know When All Tasks Are Done

Python provides thread-safe queue data structures in the queue module, such as the queue.Queue , queue.LifoQueue and queue.PriorityQueue classes.

Objects can be added to the queue via calls to Queue.put() and removed from the queue via calls to Queue.get() .

Thread-safe means that multiple threads may put and get items from the queue concurrently without fear of a race condition or corruption of the internal data structure.

A thread can block and be notified when all current items or tasks on the queue are done by calling the Queue.join() function .

Block means that the calling thread will wait until the condition is met, specifically that all tasks are done. The Queue.join() function will not return until then.

Notified means that the blocked thread will be woken up and allowed to proceed when all tasks are done. This means that the join() function will return at this time allowing the thread to continue on with the next instructions.

Multiple different threads may join the queue and await the state that all tasks are marked done.

If there are no tasks in the queue, e.g. the queue is empty, then the join() function will return immediately.

The count of unfinished tasks goes up whenever an item is added to the queue. The count goes down whenever a consumer thread calls task_done() to indicate that the item was retrieved and all work on it is complete. When the count of unfinished tasks drops to zero, join() unblocks. — queue — A synchronized queue class

The Queue.join() function only works if threads retrieving items or tasks from the queue via Queue.get() also call the Queue.task_done() function .

The Queue.task_done() function is called by the consumer thread (e.g. the thread that calls Queue.get() ) only after the thread has finished processing the task retrieved from the queue.

This will be application specific, but is more than simply retrieving the task from the queue.

Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete. — queue — A synchronized queue class

If there are multiple consumer threads, then for join() to function correctly, each consumer thread must mark tasks as done.

If processing the task may fail with an unexpected Error or Exception , it is a good practice to wrap the processing of the task in a try-finally pattern.

This ensures that the task is marked done in the queue, even if processing the task fails.

In turn, it allows any threads blocked by calling Queue.join() to be appropriately notified when all items are retrieved from the queue, avoiding a possible deadlock concurrency failure mode.

Now that we are familiar with the Queue.task_done() and Queue.join() functions, let’s look at some worked examples.

Free Python Threading Course

Download your FREE threading PDF cheat sheet and get BONUS access to my free 7-day crash course on the threading API.

Discover how to use the Python threading module including how to create and start new threads and how to use a mutex locks and semaphores

Learn more  

Example of Queue Join and Task Done

We can explore an example of how to use join() and task_done() .

In this example we will have a producer thread that will add ten tasks to the queue and then signal that no further tasks are to be expected. The consumer thread will get the tasks, process them and mark them as done. When the signal to exit is received, the consumer thread will terminate. The main thread will wait for the producer thread to add all items to the queue, and will wait for all items on the queue to be processed before moving on.

Let’s dive in.

Producer Thread

First, we can define the function to be executed by the producer thread.

The task will iterate ten times in a loop.

Each iteration, it will generate a new random value between 0 and 1 via the random.random() function. It will then pair the generated value with an integer timestamp from 0 to 9 into a tuple and put the value on the queue.

Finally, the producer will put the value None on the queue to signal to the consumer that there are no further tasks.

This is called a Sentinel Value and is a common way for threads to communicate via queues to signal an important event, like a shutdown.

The producer() function below implements this by taking the queue instance as an argument.

Consumer Thread

Next, we can define the function to be executed by the consumer thread.

The consumer thread will loop forever.

Each iteration, it will get an item from the queue and block if there is no item yet available.

If the item retrieved from the queue is the value None , then the task will break the loop and terminate the thread.

Otherwise, the fractional value is used to block with a call to time.sleep() and is then reported. The item is then marked as processed via a call to task_done() .

Finally, just prior to the thread exiting, it will mark the signal to terminate as processed.

The consumer() function below implements this and takes the queue instance as an argument.

Create Queue and Threads

In the main thread we can create the shared queue instance.

Then we can configure and start the producer thread, which will generate tasks and add them to the queue for the consumer to retrieve.

We can then configure and start the consumer thread, which will patiently wait for work to arrive on the queue.

The main thread will then block until the producer thread has added all work to the queue and the thread has terminated.

The main thread will then block on the queue with a call to join() until the consumer has retrieved all values from the queue and processed them appropriately. This includes the final signal that there are no further task items to process.

It is important that the main thread blocks on the producer thread first, before blocking on the queue. This is to avoid a possible race condition.

For example, if the main thread blocked on the queue directly, it is possible that at that time for the queue to be empty, in which case the call would return immediately. Alternatively, it may join at a time when there are only a few tasks on the queue, they are consumed by the consumer thread and the join call returns.

The problem is that in both of these cases, we don’t know if the call to join returned because all tasks were marked done or just a subset of tasks that had been added to the queue at the time join was called.

By waiting for the producer thread to terminate first, we know that all tasks that could be added to the queue have been added to the queue. By blocking on the queue after the producer has finished, we know that when the call to join returns, that all tasks were added to the queue, all tasks were retrieved from the queue and all retrieved tasks were marked as done.

Complete Example

Tying this together, the complete example is listed below.

Running the example first creates the queue to be shared between the producer and consumer threads.

Then the producer thread is created and configured to execute our producer() function and passed in the shared queue instance. The producer thread is then started.

Next, the consumer thread is configured to execute our consumer() function and is then started.

The producer thread runs as fast as it is able, each iteration generating a random number and adding it along with the task number to the queue. Once all ten tasks have been added to the queue, a None message is sent to the queue to indicate no further messages are to be expected and the producer thread terminates.

The main thread notices that the producer thread has terminated, then blocks on the queue itself waiting for all tasks to be retrieved and marked as done.

The consumer thread retrieves tasks one at a time from the queue, blocks and reports their value, then marks the task as done. This is repeated until the sentinel message is received and the loop is broken. Just before the consumer thread is terminated, the final sentinel message is marked as done.

The main thread notices that all tasks were marked done and is then free to continue on with other tasks. In this case, the main thread reports a message and terminates.

A sample of the output from the program is listed below. Note, your specific results will differ given the use of random numbers.

In this case, we can see that the producer ran first, adding all tasks to the queue. Then the consumer thread ran, reading off each value. After all tasks are marked done, finally the main thread continues on.

This highlights how to use task_done() and join() on a thread queue.

Next, let’s look at what happens if all tasks are already marked as done on the queue.

Overwhelmed by the python concurrency APIs? Find relief, download my FREE Python Concurrency Mind Maps

Example of Join When All Tasks Are Already Done

If all tasks on the thread queue have been marked as done, then any call to join() will return immediately.

This can happen if the queue is empty, for example no tasks were added to the queue.

This can also happen if tasks were added to the queue, then retrieved and marked done and no further tasks were added.

We can update the example from the previous section to demonstrate this case.

Specifically, after the main thread is notified that all tasks were marked as done by the consumer thread, it can then attempt to join the queue again.

Running the example creates the shared queue, then creates and starts the producer and consumer threads.

The producer thread puts ten tasks to the queue, then the sentinel message. The consumer thread gets the ten tasks and marks each as done, then gets the sentinel message and marks it as done.

The main thread is then notified that all tasks on the queue are done.

The main thread then attempts to block on the queue again until all tasks are done. There are no tasks on the queue, therefore the call to join() returns immediately.

This example highlights that a call to join() will only block if there is at least one task that has not been marked as done.

Next, let’s look at what happens if we call the task_done() function too many times.

Loving The Tutorials?

Why not take the next step? Get the book.

Example of Too Many Calls to Task Done

The task_done() function may be called more times than there are tasks on the queue.

This will result in a ValueError being raised.

This can happen if task_done() is called when there are no items in the queue.

It can also happen if all tasks in the queue have been retrieved and marked as done.

We can update the above example to demonstrate this case.

Consider if the consumer thread marked the sentinel message as done just prior to breaking the loop, and again after the loop.

This would be one too many calls to task_done() compared to tasks on the queue and will result in a ValueError .

The updated version of the consumer() function with this change is listed below.

The producer thread puts ten tasks to the queue, then the sentinel message.

The consumer thread gets the ten tasks and marks each as done, then gets the sentinel message and marks it as done.

The main thread is then notified that all tasks on the queue are done as per normal.

The consumer thread breaks the loop, then marks the sentinel message as done again. This causes a ValueError to be raised, which unravels the stack and ultimately terminates consumer thread.

The program achieves the desired outcome, but the consumer thread does not terminate normally.

This example highlights what happens if too many calls are made to the task_done() function.

Next, let’s look at an example of a deadlock caused by too few calls to the task_done() function.

Example of Join Deadlock Due to Missed Task Done

A deadlock may result if there is at least one thread waiting on the queue via a call to join() and too few calls are made to the task_done() function.

This can happen if an Error or Exception is raised by a consumer thread while processing a task. For example, between a call to get() and the call to task_done() on the queue.

Consider the loop in consumer() function. We can contrive a case where if a task has the identifier of five, to raise an exception.

An exception raised at this point will cause this item on the queue to not have the task_done() function called.

Additionally, any items not retrieved from the queue will remain on the queue and will also not be marked as task_done() .

The exception will cause the consumer thread to unravel the stack and to terminate. The result will be that the main thread will block and wait forever on the queue as the remaining tasks will never be marked done.

This is a deadlock and the program must be terminated forcefully, such as via Control-C.

The consumer thread gets the ten tasks and marks each as done. When a task is retrieved that has an integer value of 5, the Exception is triggered and the thread is terminated.

The main thread continues to block, waiting for the consumer thread to mark all tasks as done.

This will never happen and the main thread is stuck, blocking forever for a condition that will never be met. A deadlock.

The program must then be forcefully terminated, such as via Control-C or via the kill command on POSIX operating systems like Linux or MacOS.

This highlights that a deadlock is possible in those cases where processing a task could result in an Error or Exception .

A possible fix is to wrap the processing of the task in a try-catch-finally pattern.

Specifically:

• The call to Queue.get() occurs before the try block.
• The processing of the task occurs within the try block.
• The expected Error or Exception is handled in the except block.
• The call to task_done() occurs within the finally block.

This approach ensures that even if the processing of a task fails, that the task is still marked as done and that any threads waiting on the queue will be appropriately notified.

Next, let’s look at some common questions about queue join() and task_done() .

Common Questions

This section lists some common questions about thread queue task_done() and join() .

Do you have any questions? Let me know in the comments below and I may add your question to this section.

Why Not Just Use Queue.qsize() or Queue.empty()?

The Queue.qsize() function will report the number of items in the queue and will return a value of zero when there are no tasks in the queue at the time of calling.

The Queue.empty() function will return True when there are no tasks in the queue at the time of calling.

Both functions can be used to report if there are no items on the queue, but neither have any way of determining if the tasks retrieved from the queue have been processed and marked done and neither have any way of notifung waiting threads.

Nevertheless, these functions can be used instead of join() in some limited cases.

If join() is called at a time when the thread knows that no further tasks will be added to the queue, such as in the above situations, then qsize() or empty() can be used instead within a busy wait loop.

A busy wait loop or busy waiting is a way for a thread to block by repeatedly checking a condition. Once the condition is met, the thread may then proceed.

For example, below is a busy wait loop using the qsize() function:

We can also use the empty() function in a busy wait loop.

The problem with the busy wait loop is that the thread will remain active and consume computational resources while waiting for the condition to be met.

The join() function on the other hand will block until the condition is met (e.g. until it is notified), meaning that the thread will not consume any computational resources while waiting.

You can learn more about busy wait loops in this tutorial:

• How to Use Busy Waiting in Python

Is Queue.join() related to Thread.join()?

Calling join() on a queue will block and return once the number of calls to task_done() matches the number of calls to put() at the time that join() is called.

Calling join() a thread will block until the target thread terminates.

You can learn more about Thread.join() in the tutorial:

• How to Join a Thread in Python

Do All Consumer Threads Need To Call task_done()?

If a thread is blocked by calling join() on the queue, then all threads that are retrieving items from the queue via get() must call task_done() in order for the blocked thread to be notified correctly.

What if the Queue is Empty?

A thread may call join() on an empty thread.

If a thread calls join() on a queue that is empty, then it will not block and the call will return immediately.

What if Tasks Are Put After a Call to Join?

Tasks may be put on the queue after a thread has called join() .

Each call to put() will increment an internal counter in the queue. Each call to task_done() decrements the counter. A call to join() will return once the counter has reached zero.

This means that any calls to put() while a thread is blocked on a call to join() will require a subsequent call to task_done() by a consumer thread in order for the blocked thread to be notified.

Can task_done() Be Called Without Calls to get()?

A thread blocked with a call to join() will be notified after a task_done() call has been made for each call to put() .

Any calls to get() are not considered by the blocked thread.

This means that a thread blocked by a call to join() may be notified by the correct number of calls to task_done() , even though all of the tasks may still remain on the queue given the absence of calls to get() .

When Should I Call join()?

The join() function should be called by a thread on a queue when the calling thread wants to be notified when all tasks on the queue have been marked as done.

Ideally, the join() function would be called after all tasks that could be added to the queue have been added to the queue. This is to avoid a race condition where the blocked thread is notified because all tasks have been retrieved from the queue and marked as done, yet the producer thread has more tasks to add to the queue.

When Should I call task_done()?

The task_done() should be called on the queue after a thread has called get() to retrieve the item and after that retrieved item has been processed to the satisfaction of the application.

What if Processing a Task May Raise an Error or Exception?

Processing tasks retrieved from a queue may fail with an Error or Exception .

If this is possible, a try-except-finally or try-finally pattern may be used.

The call to get() on the queue would be made prior to the try block and the call to task_done() would be made in the finally block.

This ensures that regardless of the success or failure of processing the task, the task is marked as done and any thread blocked until all tasks are done will be correctly notified.

Otherwise, if a task_done() call is missed because of an error while processing, any threads waiting on the queue with a call to join() will block forever, resulting in a deadlock.

Further Reading

This section provides additional resources that you may find helpful.

Python Threading Books

• Python Threading Jump-Start , Jason Brownlee ( my book! )
• Threading API Interview Questions
• Threading Module API Cheat Sheet

I also recommend specific chapters in the following books:

• See: Chapter 12: Concurrency
• See: Chapter 7: Concurrency and Parallelism
• See: Chapter: 14: Threads and Processes
• Python Threading: The Complete Guide
• Python ThreadPoolExecutor: The Complete Guide
• Python ThreadPool: The Complete Guide
• threading - Thread-based parallelism
• queue — A synchronized queue class
• Thread (computing), Wikipedia.
• Process (computing), Wikipedia.

You now know how to wait for all tasks to be done on a thread queue.

Do you have any questions? Ask your questions in the comments below and I will do my best to answer.

Photo by Alvis Taurēns on Unsplash

Share this:

Related tutorials:.

About Jason Brownlee

Hi, my name is Jason Brownlee, Ph.D. and I’m the guy behind this website. I am obsessed with Python Concurrency.

I help python developers learn concurrency, super fast. Learn more .

Parallel Loops in Python

Discover how to run your loops in parallel, download your free book now:

Your free book " Parallel Loops in Python " includes complete and working code templates that you can modify and use right now in your own projects.

Reader Interactions

Do you have any questions cancel reply, learn threading systematically.

What if you could develop Python programs that were concurrent from the start?

The threading module provides easy-to-use thread-based concurrency.

Introducing: " Python Threading Jump-Start ".

A new book designed to teach you the threading module step-by-step, super fast!

Insert/edit link

Enter the destination URL

Or link to existing content

task-queue 2.14.0

pip install task-queue Copy PIP instructions

Released: May 7, 2024

Multithreaded cloud queue client.

Verified details

Maintainers.

Unverified details

Project links, github statistics.

• Open issues:

License: BSD License (BSD)

Author: Ignacio Tartavull, William Silversmith, and others

Classifiers

• 5 - Production/Stable
• OSI Approved :: BSD License
• Python :: 3
• Python :: 3.5
• Python :: 3.6
• Python :: 3.7
• Python :: 3.8

Project description

python-task-queue

A client and system for generating, uploading, leasing, and executing dependency free tasks both locally and in the cloud using AWS SQS or on a single machine or cluster with a common file system using file based queues. Of note, file queue requires no setup or queue service and can be used in a distributed fashion on a network filesystem.

The task queue uses JSON as a messaging medium, so be aware that e.g. integer dictionary keys can be turned into strings when bound as a parameter.

Installation

The task queue uses your CloudVolume secrets located in $HOME/.cloudvolume/secrets/ . When using AWS SQS as your queue backend, you must provide $HOME/.cloudvolume/secrets/aws-secret.json . See the CloudVolume repo for additional instructions.

As of version 2.7.0, there are two ways to create a queueable task. The new way is simpler and probably preferred.

MacOS Only: Note that proxy servers are disabled for parallel operation due to libdispatch being not fork-safe.

New School: Queueable Functions

Designate a function as queueable using the @queueable decorator. Currently variable positional arguments ( *args ) and variable keyword arguments ( **kwargs ) are not yet supported. If a function is not marked with the decorator, it cannot be executed via the queue.

You then create queueable instantiations of these functions by using the standard library partial function to create a concrete binding.

Old School: RegisteredTask Subclasses

Define a class that inherits from taskqueue.RegisteredTask and implements the execute method. RegisteredTasks contain logic that will render their attributes into a JSON payload and can be reconstituted into a live class on the other side of a task queue.

Tasks can be loaded into queues locally or in the cloud and executed later. Here's an example implementation of a trivial PrintTask . The attributes of your container class should be simple values that can be easily encoded into JSON such as ints, floats, strings, and numpy arrays. Let the execute method download and manipulate heavier data. If you're feeling curious, you can see what JSON a task will turn into by calling task.payload() .

Local Usage

For small jobs, you might want to use one or more processes to execute the tasks.

This will load the queue with 1000 print tasks then execute them across five processes.

Cloud and Cluster Usage

Set up an SQS queue and acquire an aws-secret.json that is compatible with CloudVolume. Generate the tasks and insert them into the cloud queue.

You can alternatively set up a file based queue that has the same time-based leasing property of an SQS queue.

IMPORTANT: You must import the tasks that will be executed, otherwise the code to execute them has not been loaded.

This inserts 1000 PrintTask JSON descriptions into your SQS queue.

Somewhere else, you'll do the following (probably across multiple workers):

Poll will check the queue for a new task periodically. If a task is found, it will execute it immediately, delete the task from the queue, and request another. If no task is found, a random exponential backoff of up to 120sec is built in to prevent workers from attempting to DDOS the queue. If the task fails to complete, the task will eventually recirculate within the queue, ensuring that all tasks will eventually complete provided they are not fundementally flawed in some way.

Local Container testing

If there is a AWS compatible queue running on a local cluster, e.g. alpine-sqs , the underlying connection client needs additional parameters. These can be passed into the TaskQueue constructor.

The following code on a worker will work in local and production contexts:

Example docker-compose.yml for local testing:

Example docker-compose.yml for production:

Notes on Google PubSub

TaskQueue will try to connect to pubsub using the credentials it finds at ~/.cloudvolume/secrets/google-secret.json

You must first make both a topic and a subscription that is subscribed to that topic.

Then you can specify a taskqueue using this url format (which we invented to include both the project_id, topic and subscription)

Note that Google PubSub doesn't have all the same features as Amazon SQS, including statistics reporting, and so some features do not function properly with this backend.

Also, google pubsub libraries are not installed by default and so if you want to use this backend install with the pubsub option

Notes on File Queue

FileQueue ( fq:// ) is designed to simulate the timed task leasing feature from SQS and exploits a common filesystem to avoid requiring an additional queue server. You can read in detail about its design on the wiki .

There are a few things FileQueue can do that SQS can't and also some quirks you should be aware of. For one, FileQueue can track the number of task completions ( tq.completions , tq.poll(..., tally=True) ), but it does so by appending a byte to a file called completions for each completion. The size of the file in bytes is the number of completions. This design is an attempt to avoid problems with locking and race conditions. FileQueue also tracks insertions ( tq.insertions ) in a more typical way in an insertions file. Also unlike SQS, FileQueue allows listing all tasks at once.

FileQueue also allows releasing all current tasks from their leases, something impossible in SQS. Sometimes a few tasks will die immediately after leasing, but with a long lease, and you'll figure out how to fix them. Instead of starting over or waiting possibly hours, you can set the queue to be made available again ( tq.release_all() ).

As FileQueue is based on the filesystem, it can be managed somewhat via the command line. To delete a queue, just rm -r $QUEUE_PATH . To reset a counter: rm $QUEUE_PATH/completions (e.g.). If you are brave, you could even use the mv command to reassign a task's availability.

We also discovered that FileQueues are also amenable to fixing problems on the fly. In one case, we generated a set of tasks that took 4.5 hours of computation time and decided to run those tasks on a different cluster. The 500k tasks each contained a path to the old storage cluster. Using find , xargs , and sed we were able to fix them efficiently.

Bundled ptq CLI Tool

As of 2.5.0, we now bundle a command line tool ptq to make managing running FileQueues easier.

Distributed dependency free task execution engines (such as Igneous ) often make use of cloud based queues like Amazon Simple Queue Service (SQS). In the connectomics field we process petascale images which requires generating hundreds of thousands or millions of cloud tasks per a run. In one case, we were processing serial blocks of a large image where each block depended on the previous block's completion. Each block's run required the generation and upload of millions of tasks and the use of thousands of workers. The workers would rapidly drain the task queue and it was important to ensure that it could be fed fast enough to prevent starvation of this enormous cluster.

There are a few strategies for accomplishing this. One way might be to use a fully featured DAG supporting engine which could generate the next task on demand. However, we were experienced with SQS and had designed our architecture around it. Furthermore, it was, in our experience, robust to thousands of machines knocking on it. This does not discount that there could be better methods out there, but this was convenient for us.

The two major ways to populate the SQS queue at scale would be a task generating task so a single processor could could enlist hundreds or thousands of others or we could just make our task generating client fast and memory efficient and use a handful of cores for multiprocessing. Keeping things simple and local allows for greater operational flexibility and the addition of a drop-in mutiprocessing execution engine allows for the omission of cloud services for small jobs. Importantly, improved small scale performance doesn't preclude the later development of metageneration facilities.

By default, the Python task queue libraries are single threaded and blocking, resulting in upload rates of at most tens of tasks per second. It is possible to do much better by using threads, multiple processes, and by batching requests. TaskQueue has achivied upload rates of over 3000 tasks per second single core, and around 10,000 per second multicore on a single machine. This is sufficient to keep our cluster fed and allows for programmer flexibility as they can populate queues from their local machine using simple scripts.

How to Achieve High Performance

Attaining the quoted upload rates is simple but takes a few tricks to tune the queue. By default, TaskQueue will upload hundreds of tasks per second using its threading model. We'll show via progressive examples how to tune your upload script to get many thousands of tasks per second with near zero latency and memory usage. Note that the examples below use sqs:// , but apply to fq:// as well. These examples also use the old school style of task instantiation, but you can substitute the new style without consequence.

This first example shows how you might use the queue in the most naive fashion. The tasks list takes a long time to compute, uses a lot of memory, and then inserts a single task at a time, failing to exploit the threading model in TaskQueue. Note that this behavior has changed from previous versions where we endorsed the "with" statement where this form was faster, though still problematic.

The listing above allows you to use ordinary iterative programming techniques to achieve an upload rate of hundreds per a second without much configuration, a marked improvement over simply using boto nakedly. However, the initial generation of a list of tasks uses a lot of memory and introduces a delay while the list is generated.

This form also takes advantage of SQS batch upload which allows for submitting 10 tasks at once. As the overhead for submitting a task lies mainly in HTTP/1.1 TCP/IP connection overhead, batching 10 requests results in nearly a 10x improvement in performance. However, in this case we've created all the tasks up front again in order to batch them correctly which results in the same memory and latency issues as in Listing 1.

In Listing 3, we've started using generators instead of lists. Generators are essentially lazy-lists that compute the next list element on demand. Defining a generator is fast and takes constant time, so we are able to begin production of new elements nearly instantly. The elements are produced on demand and consumed instantly, resulting in a small constant memory overhead that can be typically measured in kilobytes to megabytes.

As generators do not support the len operator, we manually pass in the number of items to display a progress bar.

In Listing 4, we use the green=True argument to use cooperative threads. Under the hood, TaskQueue relies on Python kernel threads to achieve concurrent IO. However, on systems with mutliple cores, especially those in a virutalized or NUMA context, the OS will tend to distribute the threads fairly evenly between cores leading to high context-switching overhead. Ironically, a more powerful multicore system can lead to lower performance. To remedy this issue, we introduce a user-space cooperative threading model (green threads) using gevent (which depending on your system is uses either libev or libuv for an event loop).

This can result in a substantial performance increase on some systems. Typically a single core will be fully utilized with extremely low overhead. However, using cooperative threading with networked IO in Python requires monkey patching the standard library (!!). Refusing to patch the standard library will result in single threaded performance. Thus, using GreenTaskQueue can introduce problems into many larger applications (we've seen problems with multiprocessing and ipython). However, often the task upload script can be isolated from the rest of the system and this allows monkey patching to be safely performed. To give users more control over when they wish to accept the risk of monkey patching, it is not performed automatically and a warning will appear with instructions for amending your program.

In Listing 5, we finally move to multiprocessing to attain the highest speeds. There are three critical pieces of this construction to note.

First, we do not use the usual multiprocessing package and instead use concurrent.futures.ProcessPoolExecutor . If a child process dies in multiprocessing , the parent process will simply hang (this is by design unfortunately...). Using this alternative package, at least an exception will be thrown.

Second, we pass parameters for task generation to the child proceses, not tasks. It is not possible to pass generators from parent to child processes in CPython [1]. It is also inefficient to pass tasks directly as it requires first generating them (as in Listing 1) and then invisibly pickling and unpickling them as they are passed to the child processes. Therefore, we pass only a small number of small picklable objects that are used for constructing a task generator on the other side.

Third, as described in the narrative for Listing 5, the GreenTaskQueue has less context-switching overhead than ordinary multithreaded TaskQueue. Using GreenTaskQueue will cause each core to efficiently run independently of the others. At this point, your main bottlenecks will probably be OS/network card related (let us know if they aren't!). Multiprocessing does scale task production, but it's sub-linear in the number of processes. The task upload rate per a process will fall with each additional core added, but each core still adds additional throughput up to some inflection point.

If you insist on wanting to pass generators to your subprocesses, you can use iterators instead. The construction above allows us to write the generator call up front, pass only a few primatives through the pickling process, and transparently call the generator on the other side. We can even support the len() function which is not available for generators.

If you design your iterators such that the slice operator works, TaskQueue can automatically resection the iterator such that it can be fed to multiple processes. Notably, we don't return PrintTaskIterator(self.start+slc.start, self.start+slc.stop) because it triggers an endless recursion during pickling. However, the runtime copy implementation above sidesteps this issue. Internally, PrintTaskIterator(0,200) will be turned into [ PrintTaskIterator(0,100), PrintTaskIterator(100,200) ] . We also perform tracking of exceptions raised by child processes in a queue. gevent.monkey.patch_all(thread=False) was necessary to avoid multiprocess hanging.

[1] You can't pass generators in CPython but you can pass iterators . You can pass generators if you use Pypy or Stackless Python.

-- Made with <3.

Project details

Release history release notifications | rss feed.

May 7, 2024

Sep 1, 2023

Sep 26, 2022

Jan 6, 2022

Jul 27, 2021

Jul 14, 2021

May 14, 2021

May 10, 2021

Apr 6, 2021

Mar 25, 2021

Feb 10, 2021

Jan 15, 2021

Dec 28, 2020

Dec 6, 2020

Dec 2, 2020

Nov 15, 2020

Nov 13, 2020

Nov 12, 2020

Oct 29, 2020

Sep 25, 2020

Sep 1, 2020

Jul 31, 2020

Jan 27, 2020

Nov 15, 2019

Sep 3, 2019

Apr 30, 2019

Feb 28, 2019

Feb 24, 2019

Feb 22, 2019

Feb 19, 2019

Jan 30, 2019

Jan 27, 2019

Jan 25, 2019

Jan 24, 2019

Oct 3, 2018

Jun 28, 2018

May 10, 2018

May 4, 2018

Mar 9, 2018

Mar 2, 2018

Jan 4, 2018

Jan 3, 2018

Dec 21, 2017

Nov 4, 2017

Nov 3, 2017

Oct 27, 2017

Oct 21, 2017

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages .

Source Distribution

Uploaded May 7, 2024 Source

Built Distribution

Uploaded May 7, 2024 Python 3

Hashes for task-queue-2.14.0.tar.gz

Hashes for task_queue-2.14.0-py3-none-any.whl

• português (Brasil)

Supported by

Implementing a Priority Queue in Python

A priority queue is a special type of queue where each element is associated with a priority. In a priority queue, elements are dequeued in order of their priority, rather than their insertion order. This makes priority queues a powerful data structure for scenarios where elements with higher priorities need to be processed first.

Importance of Priority Queues in Programming

Priority queues are widely used in various programming applications, including:

Task scheduling in operating systems

Managing the order of events in simulations

Implementing Dijkstra's algorithm for finding the shortest path in graphs

Managing a to-do list where tasks have different levels of urgency

Implementing a Priority Queue Using heapq

Python provides a built-in library called heapq that can be used to implement priority queues. The heapq module offers an efficient way to maintain a heap, which is a binary tree where the parent node is always smaller than or equal to its child nodes (min-heap).

Basic Operations: Push, Pop, Peek

Let's explore the basic operations of a priority queue using the  heapq module.

Pushing an Element

To add an element to the priority queue, use the  heappush  function:

Popping an Element

To remove and return the smallest element from the priority queue, use the  heappop function:

Peeking at the Smallest Element

To look at the smallest element without removing it, simply access the first element of the list:

Example Use Cases for Priority Queues

Let's look at different use cases for priority queues.

Task Scheduling

In task scheduling, tasks with higher priorities should be executed before those with lower priorities. A priority queue can manage the order of task execution efficiently.

Dijkstra's Algorithm

Dijkstra's algorithm finds the shortest path from a source node to all other nodes in a graph. A priority queue is used to select the next node to process based on the shortest known distance.

Advanced Priority Queue Operations

In this section, we will look closer at advanced usage of priority queues.

Updating an Element's Priority

Updating an element's priority involves removing the element and adding it again with the new priority. This can be inefficient, but it's a necessary step since  heapq  does not support direct priority updates.

Merging Two Priority Queues

To merge two priority queues, use the heapq.merge  function, which returns an iterator over the sorted values.

Practical Code Examples

Priority queue with custom objects.

You can use custom objects in a priority queue by implementing comparison methods.

Using Priority Queue in a Multi-threaded Environment

Python's queue.PriorityQueue  is a thread-safe priority queue that can be used in multi-threaded applications.

Priority queues are essential for efficiently managing tasks and resources based on priority. Python's heapq and queue.PriorityQueue modules provide powerful tools to implement and manipulate priority queues. Whether for simple tasks or complex algorithms like Dijkstra's, understanding how to use priority queues effectively can significantly enhance your programming toolkit.

For further reading and additional resources, consider the following:

• Python Documentation: heapq
• Python Documentation: queue.PriorityQueue

• Python »
• 3.12.4 Documentation »
• The Python Standard Library »
• Networking and Interprocess Communication »
• asyncio — Asynchronous I/O »
• Theme Auto Light Dark |

Source code: Lib/asyncio/queues.py

asyncio queues are designed to be similar to classes of the queue module. Although asyncio queues are not thread-safe, they are designed to be used specifically in async/await code.

Note that methods of asyncio queues don’t have a timeout parameter; use asyncio.wait_for() function to do queue operations with a timeout.

See also the Examples section below.

A first in, first out (FIFO) queue.

If maxsize is less than or equal to zero, the queue size is infinite. If it is an integer greater than 0 , then await put() blocks when the queue reaches maxsize until an item is removed by get() .

Unlike the standard library threading queue , the size of the queue is always known and can be returned by calling the qsize() method.

Changed in version 3.10: Removed the loop parameter.

This class is not thread safe .

Number of items allowed in the queue.

Return True if the queue is empty, False otherwise.

Return True if there are maxsize items in the queue.

If the queue was initialized with maxsize=0 (the default), then full() never returns True .

Remove and return an item from the queue. If queue is empty, wait until an item is available.

Return an item if one is immediately available, else raise QueueEmpty .

Block until all items in the queue have been received and processed.

The count of unfinished tasks goes up whenever an item is added to the queue. The count goes down whenever a consumer coroutine calls task_done() to indicate that the item was retrieved and all work on it is complete. When the count of unfinished tasks drops to zero, join() unblocks.

Put an item into the queue. If the queue is full, wait until a free slot is available before adding the item.

Put an item into the queue without blocking.

If no free slot is immediately available, raise QueueFull .

Return the number of items in the queue.

Indicate that a formerly enqueued task is complete.

Used by queue consumers. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.

If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue).

Raises ValueError if called more times than there were items placed in the queue.

Priority Queue ¶

A variant of Queue ; retrieves entries in priority order (lowest first).

Entries are typically tuples of the form (priority_number, data) .

LIFO Queue ¶

A variant of Queue that retrieves most recently added entries first (last in, first out).

Exceptions ¶

This exception is raised when the get_nowait() method is called on an empty queue.

Exception raised when the put_nowait() method is called on a queue that has reached its maxsize .

Queues can be used to distribute workload between several concurrent tasks:

Table of Contents

• Priority Queue

Previous topic

Subprocesses

• Report a Bug
• Show Source

• Python Basics
• Interview Questions
• Python Quiz
• Popular Packages
• Python Projects
• Practice Python
• AI With Python
• Learn Python3
• Python Automation
• Python Web Dev
• DSA with Python
• Python OOPs
• Dictionaries

Extracting text from HTML file using Python

Extracting text from an HTML file is a common task in web scraping and data extraction. Python provides powerful libraries such as BeautifulSoup that make this task straightforward. In this article we will explore the process of extracting text from an HTML file using Python .

Use the below command to install the BeautifulSoup library:

Using BeautifulSoup for Text Extraction

BeautifulSoup helps us to parse HTML documents and extract data from them.

Example: In the below example we are extracting text from html file.

Handling Complex HTML Structures

Sometimes HTML files can have complex structures and we might need to extract text from nested tags or specific parts of the HTML. For such scenarios BeautifulSoup provides various methods to handle it.

Extracting Text from Nested Tags

To extract text from nested tags we will navigate through the parse tree using tag names, attributes, or CSS selectors.

Example: In below example we will extract text from nested tags.

Extracting Text Using CSS Selectors

We can also use CSS selectors to target specific elements.

Conclusion:

By following the steps explained in this article, we can efficiently parse and extract text from HTML documents. For complex HTML structures BeautifulSoup’s has powerful methods and CSS selectors. With these techniques, we can perform web scraping and data extraction tasks effectively.

Please Login to comment...

Similar reads.

• Python BeautifulSoup

Improve your Coding Skills with Practice

What kind of Experience do you want to share?

• Stack Overflow for Teams Where developers & technologists share private knowledge with coworkers
• Advertising & Talent Reach devs & technologists worldwide about your product, service or employer brand
• OverflowAI GenAI features for Teams
• OverflowAPI Train & fine-tune LLMs
• Labs The future of collective knowledge sharing
• About the company Visit the blog

Collectives™ on Stack Overflow

Find centralized, trusted content and collaborate around the technologies you use most.

Q&A for work

Connect and share knowledge within a single location that is structured and easy to search.

Get early access and see previews of new features.

Python - What is queue.task_done() used for?

I wrote a script that has multiple threads (created with threading.Thread ) fetching URLs from a Queue using queue.get_nowait() , and then processing the HTML. I am new to multi-threaded programming, and am having trouble understanding the purpose of the queue.task_done() function.

When the Queue is empty, it automatically returns the queue.Empty exception. So I don't understand the need for each thread to call the task_done() function. We know that we're done with the queue when its empty, so why do we need to notify it that the worker threads have finished their work (which has nothing to do with the queue, after they've gotten the URL from it)?

Could someone provide me with a code example (ideally using urllib , file I/O, or something other than fibonacci numbers and printing "Hello") that shows me how this function would be used in practical applications?

4 Answers 4

Queue.task_done is not there for the workers' benefit. It is there to support Queue.join .

If I give you a box of work assignments, do I care about when you've taken everything out of the box?

No. I care about when the work is done . Looking at an empty box doesn't tell me that. You and 5 other guys might still be working on stuff you took out of the box.

Queue.task_done lets workers say when a task is done . Someone waiting for all the work to be done with Queue.join will wait until enough task_done calls have been made, not when the queue is empty.

eigenfield points out in the comments that it seems really weird for a queue to have task_done / join methods. That's true, but it's really a naming problem. The queue module has bad name choices that make it sound like a general-purpose queue library, when it's really a thread communication library.

It'd be weird for a general-purpose queue to have task_done / join methods, but it's entirely reasonable for an inter-thread message channel to have a way to indicate that messages have been processed. If the class was called thread_communication.MessageChannel instead of queue.Queue and task_done was called message_processed , the intent would be a lot clearer.

(If you need a general-purpose queue rather than an inter-thread message channel, use collections.deque .)

• 2 Yes, but what I'm asking is why the queue needs to know when the task (i.e. processing the HTML, inserting the extracted data into the DB, etc) is done at all? When it's empty I'm done with it. Why can't I just know that I'm done when the queue.Empty exception is thrown? Why does a queue of URLs need to know that I successfully entered something into the database, or processing some HTML? All the queue should care about is whether all of the URLs have been dispatched to some thread. –  J. Taylor Commented Apr 3, 2018 at 19:05
• 2 @J.Taylor: Whoever's calling Queue.join needs to know. –  user2357112 Commented Apr 3, 2018 at 19:06
• 2 @J.Taylor: Depends on who needs to know the work is done, and whether work is added to the queue while work is being done. The workers don't need to call task_done to know whether they're done. task_done is a way to communicate to any threads waiting for work to finish. –  user2357112 Commented Apr 3, 2018 at 19:55
• 13 To me, the concept of a Queue that implies task_is_done / join semantics is unusual. A queue should only be concerned with put and get and wither it is empty or full. Not task_done() stuff. Am i in twilight zone here? –  daparic Commented Jul 16, 2018 at 3:25
• 3 @ifelsemonkey It's just convenient and useful in a very common usage scenario, distributing tasks among workers. And it only applies if you need it. If you never plan to call join() on the queue, you don't need to know or care about task_done() either. –  user4815162342 Commented Dec 13, 2018 at 21:53

.task_done() is used to mark .join() that the processing is done.

💡 If you use .join() and don't call .task_done() for every processed item, your script will hang forever.

Ain't nothin' like a short example;

• 2 Why do you need the additional test of "if item is None:" when you will never reach it (unless you actually store None values in the items_queue which you don't do it right now) ? As far as I see the Empty exception is thrown if queue is empty so the execution will "jump" into the catch clause. –  Alex Commented Jan 21, 2021 at 7:57
• Is your "💡 If you use .join() and don't call .task_done() for every processed item, your script will hang forever." an actual quote from somewhere else? If so, please link to the source. Otherwise it would be nice to have some info that this is "just" the gist of your (great) answer. –  bugmenot123 Commented Oct 5, 2022 at 9:29

"Read the source, Luke!" -- Obi-one Codobi

The source for ayncio.queue is pretty short.

• the number of unfinished tasks goes up by one when you put to the queue.
• it goes down by one with you call task_done
• join() awaits there being no unfinished tasks.

This makes join useful if and only if you are calling task_done(). Using the classic bank analogy:

• people come in the doors and get in line; door is a producer doing a q.put()
• when a teller is idle and a person is in line, they go to the teller window. teller does a q.get() .
• When the teller has finished helping the person, they are ready for the next one. teller does a q.task_done()
• at 5 p.m., the doors are locked door task finishes
• you wait until both the line is empty and each teller has finished helping the person in front of them. await q.join(tellers)
• then you send the tellers home, who are now all idling with an empty queue. for teller in tellers: teller.cancel()

Without the task_done(), you cannot know every teller is done with people. You cannot send a teller home while they have a person at his or her window.

• Excellent example. Thanks! –  Jabba Commented Aug 27, 2021 at 18:23

Could someone provide me with a code example (ideally using urllib, file I/O, or something other than fibonacci numbers and printing "Hello") that shows me how this function would be used in practical applications?

@user2357112's answer nicely explains the purpose of task_done , but lacks the requested example. Here is a function that calculates checksums of an arbitrary number of files and returns a dict mapping each file name to the corresponding checksum. Internal to the function, the work is divided among a several threads.

The function uses of Queue.join to wait until the workers have finished their assigned tasks, so it is safe to return the dictionary to the caller. It is a convenient way to wait for all files being processed , as opposed to them being merely dequeued.

A note on the GIL: since the code in hashlib internally releases the GIL while calculating the checksum, using multiple threads yields a measurable (1.75x-2x depending on Python version) speedup compared to the single-threaded variant.

Your Answer

Reminder: Answers generated by artificial intelligence tools are not allowed on Stack Overflow. Learn more

Sign up or log in

Post as a guest.

Required, but never shown

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy .

Not the answer you're looking for? Browse other questions tagged python queue task or ask your own question .

• Featured on Meta
• We spent a sprint addressing your requests — here’s how it went
• Upcoming initiatives on Stack Overflow and across the Stack Exchange network...
• What makes a homepage useful for logged-in users

Hot Network Questions

• How to choose between 3/4 and 6/8 time?
• What was the first science fiction or fantasy element to appear in any Batman comic?
• On the Rambam's view on funding Torah scholars
• How will a planet on asteroid belt see Jupiter?
• Are the North Star and the moon ever visible in the night sky at the same time?
• Accommodating whiteboard glare for low-vision student
• Does 誰と mean 誰とも?
• Is an employment Conflict of Interest necessary when redundant with my Affiliation?
• Does the damage from Thunderwave occur before or after the target is moved
• As a DM, what should I do if a person decides to play a rogue?
• Coping with consequences of a dog bite before buying a puppy
• Custom panel is created but does not display properly in custom plug-in
• Reference about cancellation property for semigroups
• Does there exist a nontrivial "good" set?
• Where do we go if we gain knowledge of the absolute truth?
• Of "ils" and "elles", which pronoun is, grammatically speaking, used to refer to a group with an overwhelming female majority?
• Reviewer "rejected" my submission, and then submitted one by their own. What did they do wrong?
• Dual of slope semistable vector bundle on higher dimensional variety
• Are there any reasons I shouldn't remove this odd nook from a basement room?
• Error concerning projectile motion in respected textbook?
• USB Data communication issue on a panelized pcba hardware test platform
• In the travel industry, why is the "business" term coined in for luxury or premium services?
• Using register after multiplier in the MACC
• Equivalence of first/second choice with naive probability - I don't buy it

An official website of the United States government

Here’s how you know

Official websites use .gov A .gov website belongs to an official government organization in the United States.

Secure .gov websites use HTTPS A lock ( Lock A locked padlock ) or https:// means you’ve safely connected to the .gov website. Share sensitive information only on official, secure websites.

Free Cyber Services #protect2024 Secure Our World Shields Up Report A Cyber Issue

Vulnerability Summary for the Week of June 24, 2024

The CISA Vulnerability Bulletin provides a summary of new vulnerabilities that have been recorded by the  National Institute of Standards and Technology  (NIST)  National Vulnerability Database  (NVD) in the past week. In some cases, the vulnerabilities in the bulletin may not yet have assigned CVSS scores. Please visit NVD for updated vulnerability entries, which include CVSS scores once they are available.

Vulnerabilities are based on the  Common Vulnerabilities and Exposures  (CVE) vulnerability naming standard and are organized according to severity, determined by the  Common Vulnerability Scoring System  (CVSS) standard. The division of high, medium, and low severities correspond to the following scores:

• High : vulnerabilities with a CVSS base score of 7.0–10.0
• Medium : vulnerabilities with a CVSS base score of 4.0–6.9
• Low : vulnerabilities with a CVSS base score of 0.0–3.9

Entries may include additional information provided by organizations and efforts sponsored by CISA. This information may include identifying information, values, definitions, and related links. Patch information is provided when available. Please note that some of the information in the bulletin is compiled from external, open-source reports and is not a direct result of CISA analysis.  

High Vulnerabilities

Back to top

Medium Vulnerabilities

Low Vulnerabilities

Severity Not Yet Assigned