python - 在进程中安全运行代码，在 multithreading.Process 中重定向 stdout

Question

我正在处理来自 MOOC 的数据集。我有很多 python3 代码片段需要运行并从中获取结果。为此，我编写了一个循环遍历每个片段的 python 脚本。对于每个片段我:

1. 创建新的 StringIO 对象
2. 将 sys.stdout 和 sys.stderr 设置为我的 stringIO 缓冲区
3. 在threading.thread 对象中执行代码片段
4. 加入话题
5. 将结果记录在 stringIO 缓冲区中
6. 恢复标准输出和标准错误

这对于“正确”的代码工作正常，但在其他情况下会出现问题:

• 当代码出现无限循环时，thread.join 不会终止线程。该线程是一个守护线程，因此它在后台安静地运行，直到我的循环结束。
• 当代码有一个带有 print() 的无限循环时，当我将它设置回默认值(远离 StringIO 缓冲区)时，线程开始覆盖我的实际标准输出。这污染了我的报道。

这是我当前的代码:

def execCode(code, testScript=None):
    # create file-like string to capture output
    codeOut = io.StringIO()
    codeErr = io.StringIO()

    # capture output and errors
    sys.stdout = codeOut
    sys.stderr = codeErr

    def worker():
        exec(code, globals())

        if testScript:
            # flush stdout/stderror
            sys.stdout.truncate(0)
            sys.stdout.seek(0)
            # sys.stderr.truncate(0)
            # sys.stderr.seek(0)
            exec(testScript)

    thread = threading.Thread(target=worker, daemon=True)
    # thread = Process(target=worker) #, stdout=codeOut, stderr=codeErr)
    thread.start()
    thread.join(0.5)  # 500ms

    execError = codeErr.getvalue().strip()
    execOutput = codeOut.getvalue().strip()

    if thread.is_alive():
        thread.terminate()
        execError = "TimeError: run time exceeded"

    codeOut.close()
    codeErr.close()

    # restore stdout and stderr
    sys.stdout = sys.__stdout__
    sys.stderr = sys.__stderr__

    # restore any overridden functions
    restoreBuiltinFunctions()

    if execError:
        return False, stripOuterException(execError)
    else:
        return True, execOutput

为了处理这种情况，我一直在尝试使用 multithreading.Process 和/或 contextlib.redirect_stdout 在进程中运行代码(然后我可以调用process.terminate())，但我没有成功捕获 stdout/stderr。

所以我的问题是:如何从进程中重定向或捕获标准输出/标准错误？或者，是否有其他方法可以尝试运行和捕获任意代码的输出？

(是的，我知道这通常是个坏主意；我在虚拟机中运行它以防万一那里有恶意代码)

Python版本为3.5.3

---

更新

我想到在这种情况下有更多的灵 active 。我有一个函数 preprocess(code) 接受代码提交作为字符串并更改它。大多数情况下，我一直在使用它来使用正则表达式换出一些变量的值。

这是一个示例实现:

def preprocess(code):
    import re
    rx = re.compile('earlier_date\s*=\s*.+')
    code = re.sub(rx, "earlier_date = date(2016, 5, 3)", code)
    rx = re.compile('later_date\s*=\s*.+')
    code = re.sub(rx, "later_date = date(2016, 5, 24)", code)
    return code

我可以使用预处理函数来帮助重定向 STDOUT

Answer 1

在 Python 中与正在运行的进程通信并不简单。由于某种原因，您只能在子流程生命周期中执行一次。根据我的经验，最好运行一个启动进程的线程，并在超时后获取其输出并终止子进程。

类似于:

def subprocess_with_timeout(cmd, timeout_sec, stdin_data=None):
    """Execute `cmd` in a subprocess and enforce timeout `timeout_sec` seconds.

    Send `stdin_data` to the subprocess.

    Return subprocess exit code and outputs on natural completion of the subprocess.
    Raise an exception if timeout expires before subprocess completes."""
    proc = os.subprocess.Popen(cmd,
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE)
    timer = threading.Timer(timeout_sec, proc.kill)
    # this will terminate subprocess after timeout
    timer.start()

    # you will be blocked here until process terminates (by itself or by timeout death switch)
    stdoutdata, stderrdata = proc.communicate(stdin_data) 

    if timer.is_alive():
        # Process completed naturally - cancel timer and return exit code
        timer.cancel()
        return proc.returncode, stdoutdata, stderrdata
    # Process killed by timer - raise exception
    raise TimeoutError('Process #%d killed after %f seconds' % (proc.pid, timeout_sec))

因此，运行一个调用 subprocess_with_timeout 的线程执行器。它应该处理输入并保存结果。

另一个想法是使用网络服务器来执行 IPC。参见 this link