Clean up on exit — Stopping threads when closing a PyQt6 application

I'm using QThreadPool and worker threads in my PyQt application. When I click the X button to close the window, the threads keep running in the background. What's the best way to clean everything up on application exit?

This is a very common situation when working with multithreaded PyQt6 applications. You've set up background workers using QThreadPool or QThread, everything runs great — but when you close the window, the application doesn't fully exit. The threads keep going, and you might even see errors in your console.

The solution is to hook into your window's close event and explicitly stop your background work before the window finishes closing. Let's walk through how to do this.

Understanding the problem

When you close a PyQt6 window by clicking the X button (or calling close()), Qt destroys the window and its widgets. But any threads you've started — whether via QThreadPool, QThread, or QRunnable — are managed separately. They don't automatically stop just because the window is gone.

This means your Python process can hang, or you might see tracebacks as threads try to interact with widgets that no longer exist.

Overriding closeEvent

Every QWidget (including QMainWindow) has a method called closeEvent that Qt calls whenever the widget is about to close. By overriding this method, you can run your own cleanup code at exactly the right moment.

Here's a minimal example:

from PyQt6.QtWidgets import QMainWindow


class MainWindow(QMainWindow):
    def closeEvent(self, event):
        # Put your cleanup code here
        print("Window is closing — cleaning up!")
        event.accept()

The event parameter is a QCloseEvent. Calling event.accept() tells Qt to go ahead and close the window. If you wanted to cancel the close (for example, to show a "Save changes?" dialog), you would call event.ignore() instead.

Stopping workers on close

If you're managing background workers — for example, through a QThreadPool — you'll want to signal them to stop and then wait for them to finish before allowing the window to close.

Here's how that looks in practice. First, let's set up a simple worker using QRunnable:

import time

from PyQt6.QtCore import QRunnable, pyqtSlot, QObject, pyqtSignal


class WorkerSignals(QObject):
    finished = pyqtSignal()


class Worker(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = WorkerSignals()
        self.is_running = True

    @pyqtSlot()
    def run(self):
        while self.is_running:
            print("Worker is working...")
            time.sleep(1)
        print("Worker stopped.")
        self.signals.finished.emit()

    def stop(self):
        self.is_running = False

The worker runs in a loop, checking self.is_running on each iteration. When stop() is called, it sets the flag to False, and the loop exits on the next check.

To stop QRunnable objects you need to use this flag-watching approach.

If your runnable doesn't have a loop, and is instead doing a long series of processing steps, you instead will need check the flag state multiple times during that code and return or raise to exit the runner.

The exit can only happen at the points where the flag is checked.

To avoid multiple if checks in the code, an alternative is to have a check method that raises an exception for the stop state. For example:

    def maybe_stop(self):
        if not self.is_running:
            raise Exception("Worker stopped.")

You can then use this as follows:

import time

from PyQt6.QtCore import QRunnable, pyqtSlot, QObject, pyqtSignal


class WorkerSignals(QObject):
    finished = pyqtSignal()


class Worker(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = WorkerSignals()
        self.is_running = True

    @pyqtSlot()
    def run(self):
        print("Worker is working...")
        self.maybe_stop()
        time.sleep(5) # <- do some work.
        self.maybe_stop()
        time.sleep(5) # <- do some more work.
        self.maybe_stop()
        time.sleep(5) # <- do some more work.
        self.maybe_stop()
        time.sleep(5) # <- do some more work.
        self.maybe_stop()
        time.sleep(5) # <- do some more work.
        # <- no point stopping now.
        print("Worker stopped.")
        self.signals.finished.emit()

    def stop(self):
        self.is_running = False

    def maybe_stop(self):
        if not self.is_running:
            raise Exception("Worker stopped.")

We'll not use this approach in our example here, since for testing purposes it is better to have a worker that doesn't stop. But you may find it useful in your own code.

Now let's put together a QMainWindow that starts a worker and cleans it up on close:

import sys
import time

from PyQt6.QtCore import QRunnable, QThreadPool, pyqtSlot, QObject, pyqtSignal
from PyQt6.QtWidgets import QApplication, QMainWindow, QLabel


class WorkerSignals(QObject):
    finished = pyqtSignal()


class Worker(QRunnable):
    def __init__(self):
        super().__init__()
        self.signals = WorkerSignals()
        self.is_running = True

    @pyqtSlot()
    def run(self):
        while self.is_running:
            print("Worker is working...")
            time.sleep(1)
        print("Worker stopped.")
        self.signals.finished.emit()

    def stop(self):
        self.is_running = False


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Thread Cleanup Example")

        label = QLabel("Close this window to stop the worker.")
        self.setCentralWidget(label)

        self.threadpool = QThreadPool()
        self.workers = []

        # Start a background worker.
        worker = Worker()
        self.workers.append(worker)
        self.threadpool.start(worker)

    def closeEvent(self, event):
        # Signal all workers to stop.
        for worker in self.workers:
            worker.stop()

        # Wait for all threads in the pool to finish.
        self.threadpool.waitForDone()
        print("All workers stopped. Closing application.")
        event.accept()


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

When you run this and close the window, you'll see the worker print its "stopped" message, and the application exits cleanly.

Let's look at what's happening in closeEvent:

First, we loop through all our tracked workers and call stop() on each one, setting the flag that tells them to finish. Then we call self.threadpool.waitForDone(), which blocks until every runnable in the pool has completed. This ensures we don't pull the rug out from under a running thread. Finally, we call event.accept() to let the window close.

Managing multiple worker groups

If your application has different categories of workers — say, one group handling camera feeds and another running inference engines — you can keep separate lists and stop them all in closeEvent:

def closeEvent(self, event):
    for worker in self.feed_workers:
        worker.stop()
    for worker in self.engine_workers:
        worker.stop()

    self.threadpool.waitForDone()
    event.accept()

The same principle applies: signal every worker to stop, then wait for the thread pool to drain.

Ensuring a clean exit with sys.exit

You might notice that even after the window closes, the Python process occasionally doesn't exit cleanly. This usually happens when sys.exit() isn't receiving the application's return code properly.

The standard way to launch and exit a PyQt6 application is:

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

app.exec() starts the Qt event loop and returns an exit code (an integer) when the loop ends. Passing that code to sys.exit() ensures Python terminates with the correct status.

If you skip sys.exit() and just call app.exec(), Python may not tear down all its resources properly — particularly if threads or other objects are still being referenced. Wrapping it in sys.exit() triggers a proper SystemExit exception, which gives Python the chance to clean everything up.

Complete working example

Here's the full working example:

import sys
import time

from PyQt6.QtCore import QRunnable, QThreadPool, pyqtSlot, QObject, pyqtSignal
from PyQt6.QtWidgets import QApplication, QMainWindow, QLabel


class WorkerSignals(QObject):
    finished = pyqtSignal()


class Worker(QRunnable):
    def __init__(self, worker_id):
        super().__init__()
        self.worker_id = worker_id
        self.signals = WorkerSignals()
        self.is_running = True

    @pyqtSlot()
    def run(self):
        while self.is_running:
            print(f"Worker {self.worker_id} is working...")
            time.sleep(1)
        print(f"Worker {self.worker_id} stopped.")
        self.signals.finished.emit()

    def stop(self):
        self.is_running = False


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Thread Cleanup on Exit")
        self.resize(400, 200)

        label = QLabel("Close this window to stop all workers.")
        label.setMargin(20)
        self.setCentralWidget(label)

        self.threadpool = QThreadPool()
        self.workers = []

        # Start a few background workers.
        for i in range(3):
            worker = Worker(worker_id=i)
            self.workers.append(worker)
            self.threadpool.start(worker)

    def closeEvent(self, event):
        print("Close event received. Stopping workers...")

        for worker in self.workers:
            worker.stop()

        self.threadpool.waitForDone()
        print("All workers stopped. Goodbye!")
        event.accept()


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

When you run this, you'll see three workers printing messages to the console. Close the window, and you'll see each worker confirm it has stopped before the application exits.

Summary

• Override closeEvent on your main window to run cleanup code when the user closes the application.
• Set a flag on each worker to signal it should stop, and check that flag in the worker's loop.
• Use QThreadPool.waitForDone() to block until all workers have finished before allowing the window to close.
• Always wrap app.exec() with sys.exit() to ensure a clean process exit.

This pattern works well for any PyQt6 application with background threads, whether you're processing data, handling network requests, or running live camera feeds. Once you have it in place, your application will shut down gracefully every time. For a complete introduction to using QThreadPool and QRunnable for multithreading in PyQt6, see our Multithreading PyQt6 applications with QThreadPool tutorial. You may also find our guides on signals and slots and creating your first PyQt6 window helpful as you build out your application.