Linux下PyQt5中QProcess引发段错误与UI崩溃的解决求助

问题背景 #

开发的Python多线程PyQt5窗口应用，通过按钮触发QProcess执行命令，Windows环境运行正常，但Linux下频繁出现无响应或段错误崩溃。任务平均耗时8分钟，使用waitForFinished(-1)确保任务串行执行，已通过-X faulthandler定位错误位置。

环境版本 #

OS: Linux, Python: 3.6.9, PyQt5: 5.15.6

段错误信息 #

Thread 0x00007fde5fffe700 (most recent call first):
File .../test.py line 27 in run
File .../test.py line 107 in process_single
File .../test.py line 51 in run

Thread 0x00007fde63ffe700 (most recent call first):
File .../test.py line 27 in run
File .../test.py line 107 in process_single
File .../test.py line 51 in run

Current thread 0x00007fdfd245f740(most recent call first):
File .../test.py line 40 in process_output
File .../test.py line 132 in <module> 

原始代码 #

import sys
import threading
from functools import partial
from queue import Queue
from PyQt5.QtCore import *
from PyQt5.QtWidgets import *
from PyQt5.QtGui import *

class Job(QObject):
    task_msg = pyqtSignal(str)
    task_started = pyqtSignal()
    task_finished = pyqtSignal()

    def __init__(self, job_build_cmd):
        super().__init__()
        self.job_build_cmd = job_build_cmd  
        self.task_process = None

    def run(self):
        self.task_process = QProcess()
        env = QProcessEnvironment.systemEnvironment()
        env.insert('PATH', '/usr/bin:'+env.value('PATH'))
        self.task_process.setProcessEnvironment(env)
        self.task_process.readyReadStandardOutput.connect(self.process_output)
        self.task_process.finished.connect(lambda exitcode: self.process_finished(exitcode))
        self.task_process.start('bash', ['-c', self.job_build_cmd])
        self.task_process.waitForFinished(-1)

    def process_finished(self, exitCode):
        self.process_output()
        self.task_finished.emit()
        if self.task_process is not None:
            if self.task_process.isOpen():
                self.task_process.close()
            self.task_process.deleteLater()
            self.task_process = None
        
    def process_output(self):
        if self.task_process is not None and self.task_process.isOpen():
            line_data = bytes(self.task_process.readAllStandardOutput()).decode('utf-8', errors = 'replace').strip()
            for line in line_data.splitlines():
                self.task_msg.emit(f'{str(line)}')

class WorkThread(QThread):
    def __init__(self, process_single, t_id):
        super().__init__()
        self.process_single = process_single
        self.t_id = t_id
        
    def run(self):
        self.process_single(self.t_id)

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.job_queue = Queue()
        self.work_threads = []
        self.stop_job = False
        self.lock_get_job = threading.Lock()
        self.process = []
        self.initUI()

    def initUI(self):
        self.setGeometry(100, 100, 100, 100)
        layout = QVBoxLayout()

        self.startbtn = QPushButton('Start')
        self.startbtn.clicked.connect(self.run_build)
        layout.addWidget(self.startbtn)

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

    def run_build(self):
        self.startbtn.setEnabled(False)
        tasks = {
            'Job 1': 'for i in $(seq 1$((RANDOM%101+50))); do len=$(($RANDOM % 100 + 30)); echo "Random text $i: $(tr -dc A-Za-z0-9 </dev/urandom | head -c $len)"; sleep 0.$((RANDOM % 10 + 1)); done; exit $((RANDOM % 2))'
        }
        for name, cmd in tasks.items():
            job = Job(cmd)
            job.task_msg.connect(print)
            self.job_queue.put(job)
            self.process.append(job)
            job.task_finished.connect(partial(self.sync_task_finish, job))
        self.create_worker()
            
    def create_worker(self):
        self.num_of_worker = 2
        self.work_threads =[]
        for id in range(self.num_of_worker):
            worker = WorkThread(self.process_single, id)
            self.work_threads.append(worker)
            worker.finished.connect(worker.deleteLater)
            worker.start()
            
    def process_single(self, t_id):
        while not self.stop_job:
            try:
                with self.lock_get_job :
                    if not self.job_queue.empty():
                        job = self.job_queue.get()
                    else:
                        break
                if job is not None:
                    job.task_finished.connect(self.release_process)
                    job.run()
            except Exception as e:
                if job is not None:
                    with self.lock_get_job:
                        self.job_queue.task_done()

    def release_process(self):
        with self.lock_get_job :
            self.job_queue.task_done()

    def sync_task_finish(self, job):
        with self.lock_get_job :
            if job in self.process:
                self.process.remove(job)
            if len(self.process)==0:
                self.job_finish()
    
    def job_finish(self):
        if self.job_queue.empty():
            self.startbtn.setEnabled(True)

if __name__ == '__main__':
    app = QApplication(sys.argv)
    mainWindow = MainWindow()
    mainWindow.show()
    sys.exit(app.exec_())

问题分析 #

1. QProcess跨线程风险：PyQt的QObject子类（含QProcess）默认归属创建它的线程，原代码中Job.run()在自定义WorkThread中执行，导致QProcess在非主线程创建，跨线程信号槽连接在Linux下易触发线程安全问题，引发段错误。
2. 阻塞式等待隐患：waitForFinished(-1)长时间阻塞子线程，QProcess的信号（如readyReadStandardOutput）在主线程处理，时序错乱会导致资源访问异常。
3. 信号重复连接：job.task_finished在run_build和process_single中各连接一次，重复触发任务完成逻辑，引发队列状态异常。

修复方案 #

核心修改点 #

• 强制QProcess在主线程创建和管理，子线程仅负责任务调度，不直接操作QProcess。
• 用finished信号触发下一个任务，替代waitForFinished的阻塞方式，避免线程长时间挂起。
• 统一信号连接逻辑，严格管理QProcess生命周期，确保资源正确释放。

修改后的代码 #

import sys
import threading
from queue import Queue
from PyQt5.QtCore import *
from PyQt5.QtWidgets import *
from PyQt5.QtGui import *

class Job(QObject):
    task_msg = pyqtSignal(str)
    task_finished = pyqtSignal(object)  # 传递自身供主线程处理
    start_job = pyqtSignal()  # 触发任务启动的信号

    def __init__(self, job_build_cmd):
        super().__init__()
        self.job_build_cmd = job_build_cmd  
        self.task_process = None
        self.start_job.connect(self._start_process)  # 信号绑定主线程处理函数

    def _start_process(self):
        # 主线程执行该函数，确保QProcess归属主线程
        self.task_process = QProcess(self)
        env = QProcessEnvironment.systemEnvironment()
        env.insert('PATH', '/usr/bin:' + env.value('PATH'))
        self.task_process.setProcessEnvironment(env)
        self.task_process.readyReadStandardOutput.connect(self.process_output)
        self.task_process.finished.connect(self._on_process_finished)
        self.task_process.start('bash', ['-c', self.job_build_cmd])

    def _on_process_finished(self, exitCode):
        self.process_output()
        if self.task_process:
            self.task_process.close()
            self.task_process.deleteLater()
            self.task_process = None
        self.task_finished.emit(self)  # 通知任务完成

    def process_output(self):
        if self.task_process and self.task_process.isOpen():
            line_data = bytes(self.task_process.readAllStandardOutput()).decode('utf-8', errors='replace').strip()
            for line in line_data.splitlines():
                self.task_msg.emit(f'{str(line)}')

class Worker(QThread):
    job_ready = pyqtSignal(object)  # 通知主线程启动下一个任务

    def __init__(self, job_queue, lock):
        super().__init__()
        self.job_queue = job_queue
        self.lock = lock
        self.stop_flag = False

    def run(self):
        while not self.stop_flag:
            with self.lock:
                if self.job_queue.empty():
                    break
                job = self.job_queue.get()
            # 通过信号通知主线程启动任务，避免子线程操作QProcess
            self.job_ready.emit(job)
            # 用事件循环等待任务完成，非阻塞
            loop = QEventLoop()
            job.task_finished.connect(lambda: loop.quit())
            loop.exec_()
            with self.lock:
                self.job_queue.task_done()

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.job_queue = Queue()
        self.workers = []
        self.active_jobs = []
        self.lock = threading.Lock()
        self.initUI()

    def initUI(self):
        self.setGeometry(100, 100, 300, 100)
        layout = QVBoxLayout()

        self.startbtn = QPushButton('Start')
        self.startbtn.clicked.connect(self.run_build)
        layout.addWidget(self.startbtn)

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

    def run_build(self):
        self.startbtn.setEnabled(False)
        # 模拟多任务场景
        tasks = {
            'Job 1': 'for i in $(seq 1 $((RANDOM%101+50))); do len=$(($RANDOM % 100 + 30)); echo "Random text $i: $(tr -dc A-Za-z0-9 </dev/urandom | head -c $len)"; sleep 0.$((RANDOM % 10 + 1)); done; exit $((RANDOM % 2))',
            'Job 2': 'for i in $(seq 1 $((RANDOM%101+50))); do len=$(($RANDOM % 100 + 30)); echo "Job2 text $i: $(tr -dc A-Za-z0-9 </dev/urandom | head -c $len)"; sleep 0.$((RANDOM % 10 + 1)); done; exit $((RANDOM % 2))'
        }
        with self.lock:
            self.active_jobs.clear()
            for name, cmd in tasks.items():
                job = Job(cmd)
                job.task_msg.connect(print)
                job.task_finished.connect(self._on_job_finished)
                self.job_queue.put(job)
                self.active_jobs.append(job)
        self._create_workers()

    def _create_workers(self):
        num_workers = 2
        self.workers = []
        for _ in range(num_workers):
            worker = Worker(self.job_queue, self.lock)
            worker.job_ready.connect(self._start_job)
            worker.finished.connect(worker.deleteLater)
            self.workers.append(worker)
            worker.start()

    def _start_job(self, job):
        # 主线程触发任务启动
        job.start_job.emit()

    def _on_job_finished(self, job):
        with self.lock:
            if job in self.active_jobs:
                self.active_jobs.remove(job)
            if not self.active_jobs and self.job_queue.empty():
                self._all_jobs_finished()

    def _all_jobs_finished(self):
        # 等待所有工作线程结束
        for worker in self.workers:
            worker.stop_flag = True
            worker.wait()
        self.startbtn.setEnabled(True)

if __name__ == '__main__':
    app = QApplication(sys.argv)
    mainWindow = MainWindow()
    mainWindow.show()
    sys.exit(app.exec_())

修复说明 #

1. 主线程管理QProcess：通过start_job信号触发主线程的_start_process函数创建QProcess，确保QProcess归属主线程，消除跨线程操作的安全隐患。
2. 非阻塞任务等待：工作线程使用QEventLoop等待任务完成信号，替代阻塞式的waitForFinished，线程不会长时间挂起，信号处理时序更稳定。
3. 统一资源管理：每个Job的信号仅连接一次，任务完成后由主线程统一处理资源释放和队列状态更新，避免重复操作引发的异常。
4. 线程安全调度：通过Lock保证队列操作的原子性，工作线程仅负责任务分发，不直接操作QProcess，减少线程间资源竞争。

内容的提问来源于stack exchange，提问作者Guanz T