Python Thread Priority Tutorial with Examples

In most programming languages, thread priority allows you to give certain threads precedence over others, meaning that higher-priority threads get more CPU time compared to lower-priority ones.

However, Python does not provide explicit mechanisms for controlling thread priority due to the presence of the Global Interpreter Lock (GIL) and the way the Python interpreter manages threads.

That being said, you can still influence the behavior of threads using alternative approaches such as controlling thread sleep times, using priority queues, or manually managing tasks in a way that mimics thread priority.

In this tutorial, you will learn:

Why Python does not have native thread priority
Mimicking thread priority with time.sleep()
Using priority queues to control task execution order
Working with the concurrent.futures module
Practical examples to simulate thread priority

1. Why Python Does Not Have Native Thread Priority

Python's Global Interpreter Lock (GIL) ensures that only one thread executes Python bytecode at a time, even in a multi-core processor. This is done to protect the memory management of Python objects. Due to the GIL, Python threads cannot execute truly concurrently on multiple CPU cores for CPU-bound tasks.

Because of this, controlling thread priority directly would not provide significant performance benefits in Python for CPU-bound tasks.

However, for I/O-bound tasks (e.g., network requests, file I/O), Python threads are more effective since they spend time waiting on external operations, and you can simulate thread priority using certain techniques.

2. Mimicking Thread Priority with time.sleep()

One way to simulate thread priority in Python is by controlling the frequency at which threads yield control. By making lower-priority threads sleep longer, you can give higher-priority threads more CPU time.

Example 1: Simulating Thread Priority with Sleep Time

import threading
import time

def high_priority_task():
    for i in range(5):
        print(f"High-priority task running {i}")
        time.sleep(0.1)  # Shorter sleep for high priority

def low_priority_task():
    for i in range(5):
        print(f"Low-priority task running {i}")
        time.sleep(0.5)  # Longer sleep for low priority

# Creating threads
high_priority_thread = threading.Thread(target=high_priority_task)
low_priority_thread = threading.Thread(target=low_priority_task)

# Start threads
high_priority_thread.start()
low_priority_thread.start()

# Wait for both threads to complete
high_priority_thread.join()
low_priority_thread.join()

print("Both tasks are complete.")

Output (high-priority task runs more frequently):

High-priority task running 0
Low-priority task running 0
High-priority task running 1
High-priority task running 2
High-priority task running 3
Low-priority task running 1
High-priority task running 4
Low-priority task running 2
Low-priority task running 3
Low-priority task running 4
Both tasks are complete.

In this example:

The high-priority task sleeps for a shorter time (0.1 seconds), allowing it to execute more frequently.
The low-priority task sleeps for a longer time (0.5 seconds), giving the high-priority task more CPU time.

3. Using Priority Queues to Control Task Execution Order

A more structured way to control task execution based on priority is by using a priority queue. The queue.PriorityQueue class in Python allows you to assign priorities to tasks, and the task with the highest priority is always executed first.

Example 2: Using Priority Queue for Task Execution

import threading
import queue
import time

def task(name, duration):
    print(f"{name} is running")
    time.sleep(duration)
    print(f"{name} is complete")

# Create a priority queue
task_queue = queue.PriorityQueue()

# Add tasks to the queue with priority (lower number = higher priority)
task_queue.put((1, ("High-priority task", 1)))  # Higher priority (1)
task_queue.put((3, ("Low-priority task", 3)))   # Lower priority (3)
task_queue.put((2, ("Medium-priority task", 2)))  # Medium priority (2)

def worker():
    while not task_queue.empty():
        priority, task_info = task_queue.get()
        task_name, task_duration = task_info
        task(task_name, task_duration)
        task_queue.task_done()

# Create a thread to process tasks from the queue
thread = threading.Thread(target=worker)
thread.start()

# Wait for the thread to finish
thread.join()

print("All tasks completed.")

Output (tasks execute in priority order):

High-priority task is running
High-priority task is complete
Medium-priority task is running
Medium-priority task is complete
Low-priority task is running
Low-priority task is complete
All tasks completed.

In this example:

Tasks are added to a priority queue with assigned priorities (lower number = higher priority).
The worker thread processes tasks based on priority, ensuring that the high-priority task is executed before the medium-priority task and the low-priority task.

4. Working with concurrent.futures Module

The concurrent.futures module provides a higher-level interface for asynchronously executing tasks using thread or process pools. You can simulate task priority by submitting tasks to the pool based on their priority or completion time.

Example 3: Simulating Task Priority with concurrent.futures

import concurrent.futures
import time

def task(name, duration):
    print(f"Task {name} is starting")
    time.sleep(duration)
    print(f"Task {name} is complete")
    return name

# Create a thread pool
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # Submit tasks with different "priority" based on their sleep time
    future_high = executor.submit(task, "High-priority", 1)
    future_low = executor.submit(task, "Low-priority", 3)
    future_medium = executor.submit(task, "Medium-priority", 2)

    # Wait for results
    print(future_high.result())
    print(future_medium.result())
    print(future_low.result())

print("All tasks completed.")

Output:

Task High-priority is starting
Task Low-priority is starting
Task Medium-priority is starting
Task High-priority is complete
High-priority
Task Medium-priority is complete
Medium-priority
Task Low-priority is complete
Low-priority
All tasks completed.

In this example:

Tasks are submitted with different “priorities” based on their execution time (duration).
The ThreadPoolExecutor allows tasks to run concurrently, but their completion order is based on how quickly each task finishes.

5. Practical Examples of Simulating Thread Priority

Example 4: Simulating Download Tasks with Priority

Let’s simulate downloading files where high-priority files are downloaded before lower-priority files using a priority queue.

import threading
import queue
import time

def download_file(filename, duration):
    print(f"Downloading {filename}...")
    time.sleep(duration)
    print(f"{filename} download complete.")

# Create a priority queue
download_queue = queue.PriorityQueue()

# Add files to download with priority (lower number = higher priority)
download_queue.put((1, ("High-priority file", 2)))
download_queue.put((3, ("Low-priority file", 4)))
download_queue.put((2, ("Medium-priority file", 3)))

def worker():
    while not download_queue.empty():
        priority, file_info = download_queue.get()
        filename, download_duration = file_info
        download_file(filename, download_duration)
        download_queue.task_done()

# Create a thread to download files
download_thread = threading.Thread(target=worker)
download_thread.start()

# Wait for the download thread to finish
download_thread.join()

print("All files downloaded.")

Output:

Downloading High-priority file...
High-priority file download complete.
Downloading Medium-priority file...
Medium-priority file download complete.
Downloading Low-priority file...
Low-priority file download complete.
All files downloaded.

In this example, high-priority files are downloaded first, followed by medium and low-priority files.

Example 5: Simulating Task Priority with Multiple Threads

You can create multiple threads and control their task execution based on priority by controlling their sleep times or by using a priority queue.

import threading
import time

def high_priority_task():
    for i in range(3):
        print(f"High-priority task {i}")
        time.sleep(0.1)

def low_priority_task():
    for i in range(3):
        print(f"Low-priority task {i}")
        time.sleep(0.5)

# Creating threads
high_priority_thread = threading.Thread(target=high_priority_task)
low_priority_thread = threading.Thread(target=low_priority_task)

# Start threads
high_priority_thread.start()
low_priority_thread.start()

# Wait for both threads to complete
high_priority_thread.join()
low_priority_thread.join()

print("Both tasks are complete.")

Output:

High-priority task 0
Low-priority task 0
High-priority task 1
High-priority task 2
Low-priority task 1
Low-priority task 2
Both tasks are complete.

Here, we control the task priority by giving the high-priority task a shorter sleep time.

Conclusion

Python does not provide native support for thread priority due to the presence of the Global Interpreter Lock (GIL). However, you can simulate thread priority by:

Using time.sleep() to control how frequently threads yield control.
Using a priority queue to control task execution based on priority.
Utilizing the concurrent.futures module to manage tasks efficiently.
In most Python programs, especially for I/O-bound tasks, these techniques can help you manage thread-like behaviors efficiently. For CPU-bound tasks, you may want to consider other options such as multiprocessing for true parallelism.

With these techniques, you're now ready to simulate thread priority in Python for various tasks!