# Pyqt Threading

> PyQt/PySide6 threading and concurrency - QThread, QThreadPool, QTimer, thread safety, concurrent patterns

- Skill: `codeatcode/pyqt-threading` (Agent Skill)
- Install (CLI): `npx skillmds@latest add codeatcode/pyqt-threading`
- Raw SKILL.md: https://api.skillmd.com/api/skills/codeatcode/pyqt-threading/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: codeatcode (https://skillmd.com/u/codeatcode)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/codeatcode/pyqt-threading

---


# PyQt Threading - Concurrency and Thread Safety

Comprehensive guide to threading in PyQt applications.

## Thread Safety Rules

**CRITICAL**: Qt/PyQt is NOT thread-safe for UI operations. You MUST follow these rules:

1. **Never access widgets from worker threads** - Only the main thread can modify UI
2. **Use signals for cross-thread communication** - Emit signals from worker, connect to slots in main thread
3. **Use Qt.QueuedConnection for thread-safe signal delivery** - AutoConnection handles this automatically
4. **Never block the main thread** - Long operations will freeze the UI

```python
# ❌ 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 with Worker Object (Recommended Pattern)

The most flexible pattern separates the worker logic from thread lifecycle:

```python
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
```

## QThread Subclass (Simpler Pattern)

For simpler cases, subclass QThread directly:

```python
from PySide6.QtCore import QThread, Signal

class DataProcessor(QThread):
    """Thread that processes data and emits progress."""
    progress = Signal(int)
    result_ready = Signal(list)
    error_occurred = Signal(str)
    finished = Signal()
    
    def __init__(self, input_data, parent=None):
        super().__init__(parent)
        self.input_data = input_data
        self._cancelled = False
    
    def run(self):
        """Thread entry point - called by start()."""
        try:
            results = []
            total = len(self.input_data)
            
            for i, item in enumerate(self.input_data):
                if self._cancelled:
                    self.error_occurred.emit("Cancelled")
                    return
                
                # Process item (heavy work here)
                processed = self.process_item(item)
                results.append(processed)
                
                # Emit progress
                progress_percent = int((i + 1) / total * 100)
                self.progress.emit(progress_percent)
            
            self.result_ready.emit(results)
        except Exception as e:
            self.error_occurred.emit(str(e))
        finally:
            self.finished.emit()
    
    def process_item(self, item):
        import time
        time.sleep(0.05)  # Simulate work
        return str(item).upper()
    
    def cancel(self):
        self._cancelled = True

# Usage
class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.processor = None
        
        self.progress = QProgressBar()
        self.start_btn = QPushButton("Start")
        self.cancel_btn = QPushButton("Cancel")
        
        self.start_btn.clicked.connect(self.start_processing)
        self.cancel_btn.clicked.connect(self.cancel_processing)
    
    def start_processing(self):
        data = ["item1", "item2", "item3", "item4", "item5"]
        
        self.processor = DataProcessor(data)
        self.processor.progress.connect(self.progress.setValue)
        self.processor.result_ready.connect(self.on_results)
        self.processor.error_occurred.connect(self.on_error)
        self.processor.finished.connect(self.on_finished)
        
        self.processor.start()
        self.start_btn.setEnabled(False)
    
    def cancel_processing(self):
        if self.processor:
            self.processor.cancel()
    
    def on_results(self, results):
        print(f"Got {len(results)} results")
    
    def on_error(self, error):
        QMessageBox.warning(self, "Error", error)
    
    def on_finished(self):
        self.start_btn.setEnabled(True)
        self.progress.setValue(0)
        self.processor = None
```

## QThreadPool with QRunnable

For parallel execution of independent tasks:

```python
from PySide6.QtCore import QThreadPool, QRunnable, Signal, QObject, QThread
import time

class TaskSignals(QObject):
    """Signals for QRunnable (QRunnable cannot have signals directly)."""
    finished = Signal(object)
    error = Signal(str)
    progress = Signal(int)

class ParallelTask(QRunnable):
    """Runnable task for thread pool."""
    
    def __init__(self, task_id, data):
        super().__init__()
        self.task_id = task_id
        self.data = data
        self.signals = TaskSignals()
        self._cancelled = False
    
    def run(self):
        """Executed by thread pool."""
        try:
            time.sleep(0.5)  # Simulate work
            
            if self._cancelled:
                return
            
            result = {
                "id": self.task_id,
                "processed": str(self.data).upper(),
                "thread": int(QThread.currentThreadId())
            }
            
            self.signals.finished.emit(result)
        except Exception as e:
            self.signals.error.emit(str(e))
    
    def cancel(self):
        self._cancelled = True

class ThreadPoolManager(QObject):
    """Manages parallel task execution."""
    all_finished = Signal(int)
    
    def __init__(self, max_threads=4):
        super().__init__()
        self.pool = QThreadPool()
        self.pool.setMaxThreadCount(max_threads)
        self.active_tasks = {}
        self.completed_count = 0
        self.total_tasks = 0
    
    def run_parallel(self, tasks):
        """Run multiple tasks in parallel."""
        self.completed_count = 0
        self.total_tasks = len(tasks)
        self.active_tasks.clear()
        
        for task_id, data in enumerate(tasks):
            task = ParallelTask(task_id, data)
            task.signals.finished.connect(
                lambda result, tid=task_id: self.on_task_finished(result)
            )
            task.signals.error.connect(self.on_task_error)
            self.active_tasks[task_id] = task
            self.pool.start(task)
    
    def on_task_finished(self, result):
        self.completed_count += 1
        task_id = result["id"]
        del self.active_tasks[task_id]
        
        if self.completed_count >= self.total_tasks:
            self.all_finished.emit(self.completed_count)
    
    def on_task_error(self, error):
        print(f"Task error: {error}")
    
    def cancel_all(self):
        for task in self.active_tasks.values():
            task.cancel()
        self.active_tasks.clear()
```

## QTimer for Periodic Updates

```python
from PySide6.QtCore import QTimer, Slot

class PollingWidget(QWidget):
    def __init__(self):
        super().__init__()
        
        # Create timer
        self.timer = QTimer(self)
        self.timer.timeout.connect(self.on_timeout)
        
        # UI
        self.status_label = QLabel("Last update: Never")
        self.poll_btn = QPushButton("Start Polling")
        self.poll_btn.setCheckable(True)
        
        layout = QVBoxLayout(self)
        layout.addWidget(self.status_label)
        layout.addWidget(self.poll_btn)
        
        self.poll_btn.toggled.connect(self.toggle_polling)
    
    @Slot()
    def toggle_polling(self, checked):
        if checked:
            self.timer.start(1000)  # Poll every second
            self.poll_btn.setText("Stop Polling")
        else:
            self.timer.stop()
            self.poll_btn.setText("Start Polling")
    
    @Slot()
    def on_timeout(self):
        from datetime import datetime
        self.status_label.setText(f"Last update: {datetime.now().strftime('%H:%M:%S')}")
```

## QThreadPool with QRunnable - Fire-and-Forget Pattern

For fire-and-forget tasks where you don't need to wait for results:

```python
from PySide6.QtCore import QThreadPool, QRunnable, Signal, QObject
import time

class BackgroundTask(QRunnable):
    """Runnable for fire-and-forget operations."""
    
    def __init__(self, task_id, data):
        super().__init__()
        self.task_id = task_id
        self.data = data
        self._cancelled = False
    
    def run(self):
        """Executed by thread pool - auto-managed lifecycle."""
        try:
            # Simulate background work
            for i in range(10):
                if self._cancelled:
                    return
                import time
                time.sleep(0.1)
            
            print(f"Task {self.task_id} completed")
            
        except Exception as e:
            # Log error silently for fire-and-forget
            print(f"Task {self.task_id} failed: {e}")

# Usage - fire and forget
task = BackgroundTask(42, "some data")
QThreadPool.globalInstance().start(task)
# Task automatically deleted after run() completes
```

**Key Behaviors:**
- **Auto-delete**: QRunnable is deleted automatically after `run()` completes
- **No explicit lifecycle management needed**: Pool handles creation and destruction
- **Global thread pool**: `QThreadPool.globalInstance()` returns the default singleton
- **Default limit**: Typically 8 threads (can be configured via `setMaxThreadCount()`)
- **Thread recycling**: Completed tasks' threads are reused for new tasks

## moveToThread() Pattern - Worker Object Semantics

### Correct Pattern: Worker + Thread Controller

The **worker object pattern** is the recommended approach for explicit thread control:

```python
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()
```

### Ownership Semantics

| Object | Owner | Lifetime | Deletion |
|--------|-------|----------|----------|
| **Worker** | ThreadController | Until thread.quit() + wait() | worker.deleteLater() |
| **Thread** | ThreadController | Until deleted | thread.deleteLater() |
| **Signals** | Their parent | Until parent deleted | Automatic |

### Common Mistakes to Avoid

```python
# ❌ 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()
```

## Thread Lifecycle Management

### Thread Signals and State

```python
from PySide6.QtCore import QThread, Signal, QObject

class MyThread(QThread):
    started = Signal()  # Emitted when thread() is called
    finished = Signal()  # Emitted when finished() is called
    isRunningChanged = Signal(bool)  # Emitted when running state changes
    
    def __init__(self):
        super().__init__()
        self._running = False
    
    def run(self):
        self._running = True
        self.isRunningChanged.emit(True)
        self.started.emit()
        try:
            # Work here
            pass
        finally:
            self.finished.emit()
```

### Proper Cleanup Pattern

```python
class ThreadManager(QObject):
    def __init__(self):
        super().__init__()
        self.threads = {}
    
    def create_thread(self, name, worker):
        thread = QThread()
        thread.setObjectName(name)
        
        worker.moveToThread(thread)
        thread.started.connect(worker.start_work)
        
        # Cleanup when thread finishes
        thread.finished.connect(self._on_thread_finished, Qt.QueuedConnection)
        
        thread.start()
        self.threads[name] = thread
        
        return thread
    
    def _on_thread_finished(self, thread, name):
        """Clean up resources when thread exits."""
        # Signal for parent to handle cleanup
        self.on_thread_cleanup.emit(thread, name)
        
        # Auto-delete worker
        if name in self.thread_workers:
            worker = self.thread_workers.pop(name)
            worker.deleteLater()
        
        # Thread will be deleted by connect above
        print(f"Thread {name} cleaned up")
    
    def on_thread_cleanup(self, thread, name):
        """Override to handle custom cleanup."""
        del self.threads[name]

# Usage
manager = ThreadManager()
manager.thread_workers = {"worker1": worker}
thread = manager.create_thread("WorkerThread", worker)
# When thread finishes, worker is auto-deleted
```

### Graceful Shutdown Patterns

```python
class GracefulWorker(QObject):
    finished = Signal()
    
    def __init__(self):
        super().__init__()
        self._shutdown_requested = False
        self._current_task = None
    
    def shutdown(self):
        """Request graceful shutdown."""
        self._shutdown_requested = True
        if self._current_task:
            self._current_task.cancel()
    
    def run(self):
        while not self._shutdown_requested:
            # Check shutdown before starting new task
            if self._check_ready_for_work():
                self._do_work()
            else:
                import time
                time.sleep(0.01)  # Small delay to avoid busy-wait
        
        self.finished.emit()

# Shutdown sequence
def shutdown_worker(worker, thread):
    """Clean shutdown of worker and thread."""
    worker.shutdown()
    
    # Wait for current work to complete
    for _ in range(100):
        if worker.isFinished():  # Qt isSignal (weak reference)
            break
        import time
        time.sleep(0.1)
    
    # Quit thread
    thread.quit()
    thread.wait()
    
    # Final cleanup
    worker.deleteLater()
    thread.deleteLater()
```

## Thread Safety - Advanced Patterns

### Thread-Safe State Management

```python
from PySide6.QtCore import QObject, QObject, QMutex, QMutexLocker, QAtomicPointer
from PySide6.QtCore import QThread, Signal
import threading

class ThreadSafeCounter:
    """Atomic counter for thread-safe increment."""
    
    def __init__(self, initial=0):
        self._value = QAtomicPointer(initial)
    
    def increment(self):
        """Atomic increment operation."""
        return QAtomicPointer.fetchAndAddRelaxed(self._value, 1)
    
    def get(self):
        """Atomic read."""
        return QAtomicPointer.load(self._value)
    
    def set(self, value):
        """Atomic write."""
        QAtomicPointer.store(self._value, value)

class SharedResource:
    """Advanced thread-safe resource."""
    
    def __init__(self):
        self._data = {}
        self._mutex = threading.Lock()  # Platform lock for cross-thread use
        self._ref_count = 0
        self._lock = QMutex()
    
    def acquire(self):
        """Thread-safe acquisition with timeout."""
        acquired = QMutex.tryLock(self._lock, 1000)  # 1 second timeout
        if not acquired:
            raise RuntimeError("Failed to acquire lock within timeout")
        self._ref_count += 1
    
    def release(self):
        """Thread-safe release."""
        QMutex.unlock(self._lock)
        self._ref_count -= 1
    
    def update(self, key, value):
        """Thread-safe update with atomicity."""
        locker = QMutexLocker(self._lock)
        # Critical section - only one thread here at a time
        self._data[key] = value
```

### Avoiding UI State Races

```python
from PySide6.QtWidgets import QApplication
from PySide6.QtCore import QObject, Signal, Slot
import threading

class UIStateManager:
    """Prevent UI state races using mutex guards."""
    
    def __init__(self):
        self._state_lock = threading.Lock()
        self._update_in_progress = False
    
    def safe_update_ui(self, value):
        """Ensure only one UI update at a time."""
        if threading.current_thread() is threading.main_thread():
            # Main thread - direct update
            self._apply_ui_update(value)
        else:
            # Worker thread - schedule update
            self.update_ui_slot(value)
    
    @Slot()
    def update_ui_slot(self, value):
        """Thread-safe slot for UI updates."""
        with self._state_lock:
            if self._update_in_progress:
                # Already updating - cancel and request again
                self._cancel_pending_update()
            self._update_in_progress = True
            try:
                # Apply UI update from main thread
                QApplication.instance().postEvent(
                    QApplication.instance(),
                    self._update_event(value)
                )
            finally:
                self._update_in_progress = False
    
    def _apply_ui_update(self, value):
        """Actual UI update - must be called from main thread."""
        self.status_label.setText(str(value))
    
    def _cancel_pending_update(self):
        """Cancel any pending UI updates."""
        self.status_label.setText("Updating...")
```

### Resource Exhaustion Prevention

```python
class ThreadPoolManager:
    """Configurable thread pool with resource limits."""
    
    def __init__(self, max_threads=4, max_concurrent=2):
        self.pool = QThreadPool()
        self.pool.setMaxThreadCount(max_threads)
        self.max_concurrent = max_concurrent
        self.active_count = 0
        self._lock = QMutex()
    
    def submit_safe(self, task):
        """Submit task only if under concurrency limit."""
        if self._should_proceed():
            self.pool.start(task)
        else:
            # Queue for later or reject
            print("Too many concurrent tasks")
    
    def _should_proceed(self):
        """Check if we can proceed with new task."""
        with QMutexLocker(self._lock):
            self.active_count += 1
            result = self.active_count < self.max_concurrent
            if not result:
                self.active_count -= 1
            return result
    
    def _cleanup_finished(self, task):
        """Called when task completes."""
        with QMutexLocker(self._lock):
            self.active_count -= 1

class LeakyResourceGuard:
    """Prevent resource leaks in long-running threads."""
    
    def __init__(self, max_duration=300, max_memory=100 * 1024 * 1024):  # 5 min, 100MB
        self.max_duration = max_duration
        self.max_memory = max_memory
        self._start_time = None
        self._memory_usage = 0
    
    def start(self):
        self._start_time = time.time()
        self._memory_usage = self._get_memory_usage()
    
    def check(self):
        """Check for resource exhaustion."""
        elapsed = time.time() - self._start_time
        current_memory = self._get_memory_usage()
        
        # Time limit
        if elapsed > self.max_duration:
            raise TimeoutError(f"Thread ran for {elapsed}s, exceeded {self.max_duration}s")
        
        # Memory limit
        memory_delta = current_memory - self._memory_usage
        if memory_delta > self.max_memory:
            raise MemoryError(f"Thread leaked {memory_delta} bytes")
    
    def _get_memory_usage(self):
        """Get approximate memory usage."""
        import resource
        return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
```

## Common Issues

| Issue | Cause | Solution |
|-------|-------|----------|
| UI freezes | Blocking operation in main thread | Move to worker thread |
| Crashes on widget access | Accessing UI from worker thread | Use signals instead |
| Memory leaks | Thread not cleaned up | Use deleteLater() and proper lifecycle |
| Deadlocks | Multiple mutexes acquired in different order | Always acquire in same order, use timeout |
| Race conditions | Shared data without locks | Use QMutex or atomic operations |

## Best Practices

1. **Always use signals for cross-thread communication** - Direct widget access from threads causes crashes
2. **Keep worker objects thread-affinity aware** - Never assume QObject is in main thread
3. **Clean up threads properly** - Use deleteLater() and quit() + wait()
4. **Handle cancellation** - Check flags periodically in long operations
5. **Use QThreadPool for parallel independent tasks** - Default pool manages resource limits
6. **Use moveToThread() for explicit thread control** - Worker object pattern is recommended
7. **Never use time.sleep() in main thread** - Use QTimer or workers instead
8. **Keep locks short-lived** - Only hold mutexes for critical section duration
9. **Use QReadWriteLock for read-heavy data** - Multiple readers possible, single writer
10. **Monitor thread pool limits** - Set maxThreadCount to prevent resource exhaustion

## Common Pitfalls

| Issue | Cause | Solution |
|-------|-------|----------|
| UI crashes on widget access | Widget accessed from worker thread | Always use signals to update UI from main thread |
| Deadlock | Multiple mutexes acquired in different order | Always acquire in consistent order, use timeouts |
| Race conditions | Shared data without locks | Use QMutex, QAtomicPointer, or atomic operations |
| Memory leaks | Threads not cleaned up | Use deleteLater() on threads and workers |
| Thread not stopping | No quit() + wait() sequence | Always call thread.quit() then thread.wait() |
| Signals firing from wrong thread | AutoConnection uses queued delivery | AutoConnection is correct - don't change |
| Re-entrancy issues | Signal handler calls slot recursively | Use flags to track state changes |
| Resource exhaustion | Unlimited thread pool threads | Set maxThreadCount on QThreadPool |
| Busy-wait loops | Thread polling without sleep | Use QTimer instead of polling |
| Lock not released | Exception before unlock | Use QMutexLocker for RAII-style cleanup |

### Top 5 Mistakes to Avoid

1. **Never access UI from worker threads** - The most common crash cause
   ```python
   # ❌ 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. **Forgetting thread lifecycle management** - Threads become zombies
   ```python
   # ❌ 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. **Acquiring locks in wrong order** - Deadlock
   ```python
   # ❌ 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. **Using locks too long** - Performance issues
   ```python
   # ❌ 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. **Forgetting to check cancellation** - Threads don't stop
   ```python
   # ❌ 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

### Official Documentation
- **Qt Threads and Objects**: 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
- **Signal/Slot**: 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

### Community Resources
- **Threading Best Practices**: https://realpython.com/python-threading/ (adapted for Qt)
- **Qt Multithreading Guide**: https://www.qt.io/blog/solving-multithreading-problems-in-qt-6
- **PySide6 Examples**: https://github.com/pyside/pyside-examples
- **PyQt6 Threading**: https://www.riverbankcomputing.com/static/Docs/PyQt6/qthread.html#queuing-mechanism

### Advanced Topics
- **Atomic Operations**: https://doc.qt.io/qt-6/qatomicint.html (atomic types)
- **Inter-Thread Communication**: https://doc.qt.io/qt-6/inter-thread-communication.html
- **Thread Safety Analysis**: https://doc.qt.io/qt-6/thread-safety-analysis.html

### Testing
- **pytest-qt for Threading Tests**: https://pytest-qt.readthedocs.io/
- **Async Testing Patterns**: https://doc.qt.io/qt-6/async-testing.html