Multi-Threading Do's and Don'ts

I've written a multi-threaded application using QRunnable with serial communication, timers, and inter-thread communication using queue.Queue(). I discovered I was unable to start a QTimer from a QRunnable — I had to use a signal back to the main event loop to start the timer there. Are there any rules to keep in mind when writing code that involves threads, timers, and signals?

Threading in GUI applications can feel tricky at first. There are a handful of rules that, once you know them, make everything much more predictable. This article walks through the most common pitfalls and best practices for working with threads in PyQt6 — covering timers, data sharing, signal handling, and keeping your interface responsive.

QTimer only works on threads with an event loop

One of the first surprises people encounter is that QTimer can only be started on a thread that has a running event loop. When a timer fires, the resulting timeout event needs somewhere to go — it gets placed onto the thread's event queue. The main thread always has an event loop (that's what app.exec() starts), but a QRunnable does not.

If you need a timer triggered from a worker thread, the usual approach is to emit a signal from the worker back to the main thread, and have a slot on the main thread start the timer. This is a perfectly normal pattern, not a workaround.

There's also a small quirk with QTimer worth knowing: when you create a repeating (interval) timer, you need to keep a reference to the timer object in Python, or it will be garbage collected. But when you use QTimer.singleShot(), you don't need to keep a reference — the static method just pushes an event onto the queue and returns None.

from PyQt6.QtCore import QTimer

# Interval timer — you must keep a reference
self.timer = QTimer()
self.timer.setInterval(1000)
self.timer.timeout.connect(self.on_timeout)
self.timer.start()

# Single shot — no reference needed
QTimer.singleShot(5000, self.do_something_once)

Watch out for deadlocks

When two threads communicate with each other — for example, a serial write thread and a read thread — there's always a risk of deadlock. A deadlock happens when both threads end up waiting on each other simultaneously: Thread A is waiting for data from Thread B, and Thread B is waiting for data from Thread A. Neither can make progress.

The way to avoid this is to think carefully about the direction of communication. Try to design your threads so that data flows in one direction wherever possible, and use non-blocking checks when you can. Python's queue.Queue is a good tool here because its get() method supports timeouts, so you can avoid blocking forever.

import queue

q = queue.Queue()

# In the consumer thread, use a timeout to avoid blocking forever
try:
    data = q.get(timeout=1.0)
except queue.Empty:
    # No data available, carry on
    pass

Be careful passing data between threads

When you send data between threads — whether through signals or shared references — you need to be aware that Python objects are passed by reference. This means both threads can end up pointing at the same object. If one thread modifies the object while the other is reading it, you'll get unpredictable behavior, and potentially segmentation faults.

Signals in PyQt6 send a reference to the same object, not a copy. So if you emit a signal with a list, and the worker thread then modifies that list, the main thread's slot might see partially modified data. For a deeper understanding of how signals and slots work in PyQt6, see the Signals, Slots & Events tutorial.

The complete guide to packaging Python GUI applications with PyInstaller.

Get the book

There are a few safe approaches:

Use a queue.Queue — this is thread-safe by design and a great choice for producer-consumer patterns.

Create a deep copy before sending — if you want to send data via a signal, copy it first. A shallow copy (copy.copy) duplicates the outer container, but the elements inside still point to the same objects. You need copy.deepcopy to get a fully independent copy.

from copy import copy, deepcopy

my_dict = {"a": [1, 2, 3], "b": [4, 5, 6]}

# Shallow copy — the inner lists are still shared
my_dict2 = copy(my_dict)
print(my_dict["a"] is my_dict2["a"])  # True — same object!

# Deep copy — everything is independent
my_dict3 = deepcopy(my_dict)
print(my_dict["a"] is my_dict3["a"])  # False — safe to modify separately

Send only immutable data — strings, numbers, tuples of numbers, and similar types are inherently safe because they can't be modified in place.

Keep the main thread responsive

The whole reason for putting work into threads is usually to keep the GUI responsive. But the way your threads communicate back to the main thread matters a lot.

Every signal emitted from a worker thread to the main thread is an interruption — the main thread has to pause what it's doing, switch into your Python slot function, execute it, and then return to processing other events (like repainting the window or handling mouse clicks).

There are a few things to keep in mind here:

Prefer many short interruptions over few long ones. If your slot does a lot of work (parsing large results, updating many widgets), the GUI will feel frozen while that slot runs. If you can split results into smaller chunks, each one processed quickly, the interface stays smooth because Qt can handle paint events and user input between each slot call.

Don't flood the event queue. On the other hand, emitting signals hundreds of times per second with tiny payloads can overwhelm the main thread. Each signal requires a context switch into Python from the Qt event loop, and that adds up. In extreme cases, the event queue fills up faster than it can be drained, and the application becomes unresponsive — or crashes.

Remember the GIL. Python's Global Interpreter Lock means that only one thread can execute Python code at a time. If your worker thread is doing pure Python work (not calling into a C library that releases the GIL), then while your main thread is executing a slot, the worker thread is paused too. And while the worker is running Python code, the main thread's slots have to wait. This means CPU-bound Python work in a thread doesn't truly run in parallel with your GUI — it just interleaves. For CPU-bound tasks, consider using QProcess or Python's multiprocessing module instead.

The right balance depends on your application. If you're unsure, add some timing to your slots to measure how often they fire and how long they take.

Use QRunnable and QThreadPool for efficient thread management

If your application involves many short-lived tasks, QRunnable with QThreadPool is an excellent architecture. When a QRunnable finishes, the thread it was running on doesn't shut down immediately — it goes back into the pool, ready to pick up the next waiting task. This avoids the overhead of repeatedly creating and destroying threads. For a comprehensive introduction to this approach, see the Multithreading PyQt6 applications with QThreadPool tutorial.

This also means that if you have a long-running job and want to restructure it into many small jobs, you can do so without worrying about thread creation overhead. The thread pool handles it efficiently.

Here's a complete working example that demonstrates these principles — a QRunnable worker that performs a task, sends progress updates via signals, and communicates results back to the main thread safely:

import sys
import time
from copy import deepcopy

from PyQt6.QtCore import QObject, QRunnable, QThreadPool, QTimer, pyqtSignal, pyqtSlot
from PyQt6.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QLabel, QPushButton, QWidget


class WorkerSignals(QObject):
    """Signals available from a running worker thread."""
    progress = pyqtSignal(int)
    result = pyqtSignal(dict)
    finished = pyqtSignal()


class Worker(QRunnable):
    """A worker that performs a task and reports progress."""

    def __init__(self, data):
        super().__init__()
        # Deep copy the input data so the worker owns its own copy.
        self.data = deepcopy(data)
        self.signals = WorkerSignals()

    @pyqtSlot()
    def run(self):
        """Simulate a long-running task broken into small steps."""
        total = self.data["steps"]
        results = []

        for i in range(total):
            time.sleep(0.2)  # Simulate work
            results.append(i * i)
            # Send progress as a percentage
            self.signals.progress.emit(int((i + 1) / total * 100))

        # Send the result back — it's a new dict, so it's safe
        self.signals.result.emit({"output": results})
        self.signals.finished.emit()


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Threading Do's and Don'ts")

        layout = QVBoxLayout()

        self.status_label = QLabel("Ready")
        layout.addWidget(self.status_label)

        self.progress_label = QLabel("Progress: 0%")
        layout.addWidget(self.progress_label)

        self.result_label = QLabel("Result: —")
        layout.addWidget(self.result_label)

        self.start_button = QPushButton("Start Worker")
        self.start_button.clicked.connect(self.start_worker)
        layout.addWidget(self.start_button)

        self.timer_label = QLabel("Timer: not started")
        layout.addWidget(self.timer_label)

        container = QWidget()
        container.setLayout(layout)
        self.setCentralWidget(container)

        self.threadpool = QThreadPool()

        # Create an interval timer on the main thread.
        # We keep a reference to it so it isn't garbage collected.
        self.tick_count = 0
        self.timer = QTimer()
        self.timer.setInterval(500)
        self.timer.timeout.connect(self.on_timer_tick)

    def start_worker(self):
        self.status_label.setText("Running...")
        self.start_button.setEnabled(False)

        # Start the timer on the main thread (where the event loop lives)
        self.tick_count = 0
        self.timer.start()

        # Prepare input data — the worker will deep copy this
        task_data = {"steps": 10}

        worker = Worker(task_data)
        worker.signals.progress.connect(self.on_progress)
        worker.signals.result.connect(self.on_result)
        worker.signals.finished.connect(self.on_finished)

        # Submit to the thread pool
        self.threadpool.start(worker)

    def on_progress(self, percent):
        self.progress_label.setText(f"Progress: {percent}%")

    def on_result(self, data):
        # This data dict was created in the worker thread,
        # so it's safe to read here without copying.
        output = data["output"]
        self.result_label.setText(f"Result: {output[:5]}...")

    def on_finished(self):
        self.status_label.setText("Finished!")
        self.start_button.setEnabled(True)
        self.timer.stop()

    def on_timer_tick(self):
        self.tick_count += 1
        self.timer_label.setText(f"Timer: {self.tick_count} ticks")


app = QApplication(sys.argv)
window = MainWindow()
window.show()
app.exec()

This example puts several of the guidelines into practice:

• The QTimer is created and started on the main thread, where there's an event loop to handle its timeout events.
• Input data is deep-copied in the Worker.__init__, so the worker has its own independent copy.
• Progress signals are emitted at a reasonable rate (every 200ms), sending just an integer — lightweight enough that the main thread stays responsive.
• The result is a new dictionary created inside the worker, so there's no shared-state issue when the main thread reads it.
• QThreadPool manages the threads, so if you click the button multiple times, threads are reused efficiently.

If you're new to building PyQt6 applications, you may also want to start with creating your first window to get familiar with the basics before diving into threading.