在线程上下文中捕获 stdout/stderr 和函数返回值答案

【问题标题】：Capturing stdout/stderr and function return value in a threaded context在线程上下文中捕获 stdout/stderr 和函数返回值
【发布时间】：2018-03-13 03:48:41
【问题描述】：

我需要能够在线程上下文中捕获函数的返回值及其stdout/stderr。目前，我正在通过在每个子类中调用is_true 方法来评估基于此基类（my_base_class）的许多子类（my_class_foo、my_class_bar）。

我希望能够同时捕获每个 is_true 以及 stdout/stderr 的返回值。下面的当前解决方案适用于非线程或非多进程上下文。但是，它依赖于重定向 stdout/stderr，如果我同时评估多个子类，这显然是行不通的。

我查看了 concurrent.futures 和 multiprocess 和 subprocess 包，但找不到解决方案。

我试图避免使用记录器，以便用户可以只依赖打印到stdout 而不是使用显式方法。

我想并行执行my_class_foo 和my_class_bar 中的is_true 方法，并能够使用每个类的返回值捕获stdout 消息。

class my_class_foo(my_base_class):
    def is_true(self):
        print('foo')
        return True


class my_class_bar(my_base_class):
    def is_true(self):
        print('bar')
        return False


class my_base_class(object):

    def is_true(self):
        Raise NotImplementedError

    def evaluate_node_is_true(self):
        with Capturing() as is_true_stdout:
            node_is_true = self.is_true()
            self.output = ''.join(is_true_stdout)


class Capturing(list):
    """
    Context manager for capturing the stdout of the is_true() function call
    """
    def __enter__(self):
        self._stdout = sys.stdout
        self._stderr = sys.stderr
        sys.stdout = self._stringio_out = io.StringIO()
        sys.stderr = self._stringio_err = io.StringIO()
        return self
    def __exit__(self, *args):
        self.extend(self._stringio_out.getvalue().splitlines())
        self.extend(self._stringio_err.getvalue().splitlines())
        del self._stringio_out
        del self._stringio_err
        sys.stdout = self._stdout
        sys.stderr = self._stderr

【问题讨论】：

这必须与线程一起使用，还是只能与进程一起使用？因为后者要容易得多。
另外：“我查看了 concurrent.futures 以及多进程和子进程包……”您尝试了什么？使用捕获的 stdio 在 sys.executable 上触发 subprocess.run 非常简单，我不明白为什么它对你不起作用。
顺便说一句，一旦stdout 和朋友与子解释器正确绑定在一起，就可以将它们包装在 PEP 567 上下文变量中。不要屏住呼吸——它可能发生的最早可能是 3.9，即使他们决定这样做——但这显然会使这个问题变得容易得多。
如果我可以让它与流程一起工作，那么我会很满意。进程的问题是我不知道如何在下面的答案中获取 sys.executable 的返回值。有没有办法做到这一点？提前致谢。
嗯，一个子进程除了一个 8 位整数之外不能返回任何东西，所以你需要找到一些其他的方法来传递它——在标准输出中的一个特殊的最后一行，或者你需要的管道作为额外的 fd 等传递下去。

标签： python multithreading function return stdout

【解决方案1】：

最简单的解决方案可能是subprocess。当然，如果您想与孩子共享任何数据，这将不起作用。它需要为子进程编写一个简单的单独驱动程序脚本，而不是依赖与主程序相同的脚本。但如果有效，就这么简单：

res = subprocess.run([sys.executable, driver_script_name],
                     stdout=subprocess.PIPE, stderr=subprocess.PIPE)

然后你就可以读取res.stdout 和res.stderr。

要同时处理多个子进程，最简单的解决方案是为每个subprocess.run 触发一个线程。如果您一次只需要一个池，例如 8 个，请改用 ThreadPoolExecutor。

对于像您这样的问题，subprocess 的最大问题是您需要 stdout、stderr 和返回值。进程为您提供标准输出、标准错误和返回代码，但该代码只是一个 8 位数字。

在您的情况下（基于您的 cmets），返回值实际上是一个布尔值，所以这可能没问题。 Unix 命令行工具有一个准标准，它执行布尔操作以返回 0 为真，1 为假——与 true 和 false 工具一样。

这与您可能期望的相反，但它符合更通用的准标准，即 0 表示成功，1 表示一般错误，2 表示参数错误，3-127 是特定于工具的错误，以及不使用 128-255 是因为如果被信号杀死，您有时会丢失最后一位。如果您（或您的用户）可能想要从 shell 测试您的子程序，请选择 0 表示 true。

您也可以使用multiprocessing 执行此操作，但它有点棘手。您可以手动创建管道并通过 stdio 复制它们，但这很难做到正确。更糟糕的是，我认为这样做的方法没有记录在案，所以你必须深入研究multiprocessing 的源代码才能做到这一点。

事实上，至少如果您使用 *nix，在这种情况下手动 fork 可能会更容易。当然，它是低级的，而且有很大的出错空间——但是（假设你对 Unix fork 足够了解）你就知道该做什么以及在哪里做。

multiprocessing 的一个更简单的选择是执行 this answer 演示的操作，让每个子级将其标准输出重定向到一个文件，然后父级可以稍后读取该文件。

一个甚至适用于线程的选项是将sys.stdout 替换为一个自定义的类似文件的对象，该对象附加到特定于线程的缓冲区。您可以使用线程本地数据，然后在最后将其复制回来，但仅使用由线程 ID 键入的字典可能更简单。像这样的：

class ThreadedStdWriter(io.RawIOBase):
    ThreadedStdWriter.buffers = {}
    def __init__(self):
        ThreadedStdWriter.buffers[threading.get_ident()] = []
    def write(self, b):
        ThreadedStdWriter.buffers.append(b)

...然后将其包装在BufferedWriter 和TextIOWrapper 中并将结果存储为sys.stdout，然后对sys.stderr 执行相同操作。那么，任何线程的标准输出就是b''.join(ThreadedStdWriter.buffers[thread_id])。

【讨论】：