Popen.communicate触发ValueError: 已关闭文件I/O操作异常求助

报错回溯 #

运行代码时抛出未捕获异常，完整栈信息如下：

Uncaught exception <class 'ValueError'>: I/O operation on closed file.Traceback (most recent call last):
    File "/home/gidumah/miniconda/envs/ytune/lib/python3.7/runpy.py", line 193, in _run_module_as_main
        "__main__", mod_spec)
      File "/home/gidumah/miniconda/envs/ytune/lib/python3.7/runpy.py", line 85, in _run_code
        exec(code, run_globals)
      File "/home/gidumah/ytopt/ytopt/ytopt/search/ambs.py", line 128, in <module>
        search.main()
      File "/home/gidumah/ytopt/ytopt/ytopt/search/ambs.py", line 105, in main
        results = list(self.evaluator.get_finished_evals())
      File "/home/gidumah/ytopt/ytopt/ytopt/evaluator/evaluate.py", line 200, in get_finished_evals
        y = future.result()
      File "/home/gidumah/ytopt/ytopt/ytopt/evaluator/subprocess_evaluator.py", line 41, in result
        stdout, stderr_data = self.proc.communicate()
      File "/home/gidumah/miniconda/envs/ytune/lib/python3.7/subprocess.py", line 951, in communicate
        stdout = self.stdout.read()
    ValueError: I/O operation on closed file.

问题代码 #

出问题的subprocess_evaluator.py中PopenFuture类实现如下：

class PopenFuture:
    FAIL_RETURN_VALUE = Evaluator.FAIL_RETURN_VALUE

    def __init__(self, args, parse_fxn):
        self.proc = subprocess.Popen(args, shell=True, stdout=subprocess.PIPE,
                                     stderr=subprocess.STDOUT, encoding='utf-8')

        self._state = 'active'
        self._result = None
        self._parse = parse_fxn

    def _poll(self):
        if not self._state == 'active':
            return
        retcode = self.proc.poll()
        if retcode is None:
            self._state = 'active'
            stdout, stderr_data = self.proc.communicate()
            tmp_res = self._parse(stdout)
            if tmp_res != sys.float_info.max:
                self._result = tmp_res
        elif retcode == 0:
            self._state = 'done'
        else:
            self._state = 'failed'

    def result(self):
        if self._result is not None:
            return self._result
        self.proc.wait()
        stdout, stderr_data = self.proc.communicate()
        if self.done:
            self._result = self._parse(stdout)
        else:
            self._result = self.FAIL_RETURN_VALUE
            logger.error(f"Eval failed: {stdout}")
        ####
        # if stdout:
        #     print (stdout)#.split('\n')[:-2])#)[:-1])
        ####   
        return self._result

根因分析 #

异常确实是communicate()方法调用方式错误导致的，核心问题有三个：

• communicate()只能对同一个Popen实例调用一次：该方法执行时会阻塞等待子进程结束，读取完所有管道输出后会自动关闭stdout/stderr对应的文件描述符，后续再次调用就会尝试读取已关闭的文件句柄，直接抛出你遇到的ValueError。
• 代码中存在重复调用communicate()的逻辑：
  • _poll()方法中，当检测到进程还在运行（retcode is None）时就调用了一次communicate()，调用完成后管道已经被关闭，但此时你没有更新实例状态，_state仍然保持为active
  • 后续业务逻辑调用result()方法时，会再次执行communicate()，此时管道早已关闭，直接触发异常
• 状态流转逻辑完全混乱：_poll()的设计初衷是做非阻塞的状态检查，但你在进程未结束时就调用阻塞的communicate()，违背了poll的设计；同时_poll()中读取完输出、拿到结果后没有把状态标记为完成，result()方法中调用的self.done也没有在类中实现，属于隐藏bug。

修复方案 #

按照以下逻辑调整代码即可解决问题：

• 非阻塞的_poll()方法只做返回码检查，不要调用communicate()读取输出
• 把communicate()调用收口到result()方法中，保证整个生命周期只调用一次
• 补全状态流转逻辑，实现done状态判断属性，进程结束后正确标记状态

修复后的参考实现：

class PopenFuture:
    FAIL_RETURN_VALUE = Evaluator.FAIL_RETURN_VALUE

    def __init__(self, args, parse_fxn):
        self.proc = subprocess.Popen(args, shell=True, stdout=subprocess.PIPE,
                                     stderr=subprocess.STDOUT, encoding='utf-8')
        self._state = 'active'
        self._result = None
        self._parse = parse_fxn

    def _poll(self):
        if self._state != 'active':
            return
        retcode = self.proc.poll()
        if retcode is not None:
            self._state = 'done' if retcode == 0 else 'failed'

    @property
    def done(self):
        self._poll()
        return self._state in ('done', 'failed')

    def result(self):
        if self._result is not None:
            return self._result
        # 仅在此处调用一次communicate，读取输出、等待进程结束
        stdout, _ = self.proc.communicate()
        self._poll()
        if self._state == 'done':
            self._result = self._parse(stdout)
        else:
            self._result = self.FAIL_RETURN_VALUE
            logger.error(f"Eval failed: {stdout}")
        return self._result

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