pyqt-threading

PyQt のスレッド処理 - 基本パターン

スレッドセーフティのルール

重要：Qt/PyQt は UI 操作に関してスレッドセーフではありません。以下のルールに必ず従ってください。

1. ワーカースレッドからウィジェットにアクセスしない - UI を変更できるのはメインスレッドだけです
2. スレッド間の通信にはシグナルを使用する - ワーカーからシグナルを発行し、メインスレッドのスロットに接続します
3. スレッドセーフなシグナル配信には Qt.QueuedConnection を使用する - AutoConnection がこれを自動的に処理します
4. メインスレッドをブロックしない - 時間のかかる処理によって UI がフリーズします

# ❌ WRONG: Direct UI access from thread
class BadWorker(QThread):
    def run(self):
        # This will crash or cause undefined behavior!
        self.label.setText("Done")

# ✅ CORRECT: Use signals
class GoodWorker(QThread):
    finished = Signal(str)
    
    def run(self):
        result = self.process_data()
        self.finished.emit(result)  # Signal emitted, UI updated in main thread

ワーカーオブジェクトを使用する QThread（推奨パターン）

最も柔軟なパターンでは、ワーカーのロジックとスレッドのライフサイクルを分離します。

from PySide6.QtCore import QThread, Signal, QObject, Slot

class Worker(QObject):
    """Worker object that does the actual work."""
    finished = Signal(object)
    progress = Signal(int)
    error = Signal(str)
    
    def __init__(self, data):
        super().__init__()
        self.data = data
        self._is_cancelled = False
    
    @Slot()
    def process(self):
        """Main processing method called from thread."""
        try:
            for i, item in enumerate(self.data):
                if self._is_cancelled:
                    return
                
                # Simulate heavy work
                result = self.process_item(item)
                self.progress.emit(int((i + 1) / len(self.data) * 100))
            
            self.finished.emit({"status": "success", "count": len(self.data)})
        except Exception as e:
            self.error.emit(str(e))
    
    def cancel(self):
        self._is_cancelled = True
    
    def process_item(self, item):
        import time
        time.sleep(0.1)  # Simulate work
        return item * 2

class ThreadController(QObject):
    """Manages worker thread lifecycle."""
    def __init__(self):
        super().__init__()
        self.thread = None
        self.worker = None
    
    def start_work(self, data):
        # Create thread and worker
        self.thread = QThread()
        self.worker = Worker(data)
        
        # Move worker to thread
        self.worker.moveToThread(self.thread)
        
        # Connect signals
        self.worker.finished.connect(self.on_finished)
        self.worker.progress.connect(self.on_progress)
        self.worker.error.connect(self.on_error)
        
        # Thread lifecycle
        self.thread.started.connect(self.worker.process)
        self.thread.finished.connect(self.thread.deleteLater)
        
        # Start thread
        self.thread.start()
    
    def cancel_work(self):
        if self.worker:
            self.worker.cancel()
        if self.thread:
            self.thread.quit()
            self.thread.wait()
    
    @Slot()
    def on_finished(self, result):
        print(f"Work completed: {result}")
        self.cleanup()
    
    @Slot()
    def on_progress(self, percent):
        print(f"Progress: {percent}%")
    
    @Slot()
    def on_error(self, error):
        print(f"Error: {error}")
        self.cleanup()
    
    def cleanup(self):
        self.thread = None
        self.worker = None

moveToThread() パターン - ワーカーオブジェクトのセマンティクス

正しいパターン：ワーカー + スレッドコントローラー

ワーカーオブジェクトパターンは、明示的なスレッド制御に推奨されるアプローチです。

from PySide6.QtCore import QThread, Signal, QObject, Slot

class Worker(QObject):
    """Worker owns the work, controller owns the thread."""
    progress = Signal(int)
    finished = Signal(object)
    
    def __init__(self, data):
        super().__init__()
        self.data = data
        self._is_running = False
    
    @Slot()
    def process(self):
        """Main work method - called from thread."""
        self._is_running = True
        try:
            # Heavy computation here
            for i in range(100):
                self.progress.emit(i)
            self.finished.emit(None)
        finally:
            self._is_running = False

class ThreadController(QObject):
    """Controller owns thread and manages worker."""
    def __init__(self):
        super().__init__()
        self.thread = None
        self.worker = None
    
    def start_work(self, data):
        # Create NEW thread and worker
        self.thread = QThread()
        self.worker = Worker(data)
        
        # CRITICAL: Move worker TO thread
        self.worker.moveToThread(self.thread)
        
        # Worker lifetime tied to thread lifetime
        self.thread.started.connect(self.worker.process)
        self.thread.finished.connect(self.thread.quit)
        self.thread.finished.connect(self.thread.deleteLater)
        
        self.thread.start()

所有権のセマンティクス

オブジェクト	所有者	ライフタイム	削除
ワーカー	ThreadController	thread.quit() + wait() まで	worker.deleteLater()
スレッド	ThreadController	削除されるまで	thread.deleteLater()
シグナル	その親	親が削除されるまで	自動

避けるべきよくある間違い

# ❌ WRONG: Worker outlives thread
controller = ThreadController()
controller.thread = QThread()
controller.worker = Worker()
controller.thread.moveToThread(controller.worker)  # Wrong direction!
controller.thread.start()
# Problem: Worker destroyed before thread finishes

# ❌ WRONG: Forgetting thread lifecycle
def start_work():
    thread = QThread()
    worker = Worker()
    worker.moveToThread(thread)
    thread.start()  # Never quit() or wait()!
# Problem: Zombie thread keeps running

# ✅ CORRECT: Proper ownership chain
controller = ThreadController()
controller.start_work(data)
# Later when done:
controller.thread.quit()
controller.thread.wait()
# Now delete: controller.worker.deleteLater()

よくある問題

問題	原因	解決策
UI がフリーズする	メインスレッドでブロッキング処理を実行している	ワーカースレッドに移動する
ウィジェットへのアクセス時にクラッシュする	ワーカースレッドから UI にアクセスしている	代わりにシグナルを使用する
メモリーリーク	スレッドがクリーンアップされていない	deleteLater() と適切なライフサイクルを使用する
デッドロック	複数のミューテックスを異なる順序で取得している	常に同じ順序で取得し、タイムアウトを使用する
競合状態	ロックなしで共有データを使用している	QMutex またはアトミック操作を使用する

ベストプラクティス

1. スレッド間の通信には必ずシグナルを使用する - スレッドからウィジェットに直接アクセスするとクラッシュします
2. ワーカーオブジェクトのスレッド親和性を意識する - QObject がメインスレッドにあると決めつけないでください
3. スレッドを適切にクリーンアップする - deleteLater() と quit() + wait() を使用します
4. キャンセルを処理する - 時間のかかる処理では、定期的にフラグを確認します
5. 並列実行可能な独立タスクには QThreadPool を使用する - デフォルトのプールがリソース制限を管理します
6. 明示的なスレッド制御には moveToThread() を使用する - ワーカーオブジェクトパターンが推奨されます
7. メインスレッドで time.sleep() を使用しない - 代わりに QTimer またはワーカーを使用します
8. ロックの保持時間を短くする - ミューテックスはクリティカルセクションの間だけ保持します
9. 読み取り中心のデータには QReadWriteLock を使用する - 複数の読み取りと単一の書き込みが可能です
10. スレッドプールの上限を監視する - リソースの枯渇を防ぐために maxThreadCount を設定します

よくある落とし穴

問題	原因	解決策
ウィジェットへのアクセス時に UI がクラッシュする	ワーカースレッドからウィジェットにアクセスしている	メインスレッドから UI を更新するには、必ずシグナルを使用する
デッドロック	複数のミューテックスを異なる順序で取得している	常に一貫した順序で取得し、タイムアウトを使用する
競合状態	ロックなしで共有データを使用している	QMutex、QAtomicPointer、またはアトミック操作を使用する
メモリーリーク	スレッドがクリーンアップされていない	スレッドとワーカーで deleteLater() を使用する
スレッドが停止しない	quit() + wait() の順序で呼び出していない	必ず thread.quit() の後に thread.wait() を呼び出す
シグナルが誤ったスレッドから発行される	AutoConnection がキュー経由の配信を使用している	AutoConnection が正しいため、変更しないでください
再入性の問題	シグナルハンドラーがスロットを再帰的に呼び出している	フラグを使って状態の変更を追跡する
リソースの枯渇	スレッドプールのスレッド数に上限がない	QThreadPool の maxThreadCount を設定する
ビジーウェイトループ	スレッドがスリープせずにポーリングしている	ポーリングの代わりに QTimer を使用する
ロックが解放されない	ロック解除の前に例外が発生している	RAII スタイルのクリーンアップには QMutexLocker を使用する

避けるべきよくあるミス 5 つ

1. ワーカースレッドから UI にアクセスしないでください - クラッシュの最も一般的な原因です

   # ❌ CRASH: Direct UI access
   def run(self):
       self.label.setText("Done")  # Crashes!
   
   # ✅ SAFE: Use signals
   def run(self):
       self.finished.emit("Done")  # UI updated in main thread
   

2. スレッドのライフサイクル管理を忘れないでください - スレッドがゾンビ化します

   # ❌ BAD: No cleanup
   thread = QThread()
   thread.start()  # Never quit() or wait()
   
   # ✅ GOOD: Proper lifecycle
   thread.start()
   thread.quit()
   thread.wait()  # Wait for thread to finish
   

3. 間違った順序でロックを取得しないでください - デッドロックが発生します

   # ❌ DEADLOCK
   def do_work(self):
       with lock_a:
           with lock_b:
               pass
   
   def do_other_work(self):
       with lock_b:  # Different order!
           with lock_a:
               pass
   
   # ✅ SAFE: Consistent ordering
   # Always acquire locks in same order (e.g., by ID)
   

4. ロックを長時間保持しないでください - パフォーマンスの問題が発生します

   # ❌ BAD: Lock held for long operation
   with lock:
       result = heavy_computation()
   
   # ✅ GOOD: Hold lock only for data access
   with lock:
       data = self.shared_data
   result = heavy_computation(data)  # Outside lock
   

5. キャンセルの確認を忘れないでください - スレッドが停止しません

   # ❌ BAD: Infinite loop
   def run(self):
       while True:
           do_work()
   
   # ✅ GOOD: Check cancellation flag
   def run(self):
       while not self._cancelled:
           if self._work_done():
               break
           do_work()
   

詳細解説

高度なパターンについて詳しくは、次のリファレンスファイルを参照してください。

• プールのパターン (references/pool-patterns.md) - QThread のサブクラス、QThreadPool/QRunnable、QTimer、ファイア・アンド・フォーゲット
• Qt Concurrent (references/qt-concurrent.md) - マップ／フィルター／リデュース操作、進捗の追跡、シンプルなバックグラウンドタスク
• ライフサイクルとクリーンアップ (references/lifecycle-cleanup.md) - スレッドシグナル、適切なクリーンアップ、正常なシャットダウン
• 高度な安全対策 (references/advanced-safety.md) - スレッドセーフな状態、UI 状態の競合、リソース枯渇
• テスト (references/testing.md) - pytest-qt のパターン、シグナルのテスト、スレッドのライフサイクルテスト、競合状態のテスト

参考資料

公式ドキュメント

• Qt スレッドとオブジェクト: https://doc.qt.io/qt-6/threads-and-qobjects.html
• QThread: https://doc.qt.io/qt-6/qthread.html
• QThreadPool: https://doc.qt.io/qt-6/qpthreadpool.html
• QMutex: https://doc.qt.io/qt-6/qmutex.html
• QMutexLocker: https://doc.qt.io/qt-6/qmutexlocker.html
• QReadWriteLock: https://doc.qt.io/qt-6/qreadwritelock.html
• QAtomicPointer: https://doc.qt.io/qt-6/qatomicpointer.html
• シグナル／スロット: https://doc.qt.io/qt-6/signalsandslots.html

Qt for Python（PySide6/PyQt6）

• PySide6 QThread: https://doc.qt.io/qtforpython-6/PySide6/QtCore/QThread.html
• PySide6 QThreadPool: https://doc.qt.io/qtforpython-6/PySide6/QtCore/QThreadPool.html
• PyQt6 QThread: https://www.riverbankcomputing.com/static/Docs/PyQt6/qthread.html

コミュニティリソース

• スレッド処理のベストプラクティス: https://realpython.com/python-threading/（Qt 向けに調整）
• Qt マルチスレッドガイド: https://www.qt.io/blog/solving-multithreading-problems-in-qt-6
• PySide6 のサンプル: https://github.com/pyside/pyside-examples
• PyQt6 のスレッド処理: https://www.riverbankcomputing.com/static/Docs/PyQt6/qthread.html#queuing-mechanism

高度なトピック

• アトミック操作: https://doc.qt.io/qt-6/qatomicint.html（アトミック型）
• スレッド間通信: https://doc.qt.io/qt-6/inter-thread-communication.html
• スレッドセーフティ分析: https://doc.qt.io/qt-6/thread-safety-analysis.html

テスト

• スレッド処理テスト向け pytest-qt: https://pytest-qt.readthedocs.io/
• 非同期テストのパターン: https://doc.qt.io/qt-6/async-testing.html