损坏管道错误：从 Go 到 Python 应用程序写入 stdin 时发生问题

编程并不是一个机械性的工作，而是需要有思考，有创新的工作，语法是固定的，但解决问题的思路则是依靠人的思维，这就需要我们坚持学习和更新自己的知识。今天golang学习网就整理分享《损坏管道错误：从 Go 到 Python 应用程序写入 stdin 时发生问题》，文章讲解的知识点主要包括，如果你对Golang方面的知识点感兴趣，就不要错过golang学习网，在这可以对大家的知识积累有所帮助，助力开发能力的提升。

我已经实现了一个工作池来将我的作业提交到 python 脚本。

func newworker(index int, workerconfig config.appconfig, logger log.prefixedlogger) error {
    var worker entities.worker

    worker.id = index
    worker.config = workerconfig
    strcommand := workerconfig.ridemodelscriptpath
    command := exec.command(strcommand)

    stdin, err := command.stdinpipe()
    if err != nil {
        logger.error("worker_pool.initworkerpool", "error resolving stdin pipe from command", err.error())
        return err
    }
    worker.stdin = stdin

    stdout, err := command.stdoutpipe()
    if err != nil {
        logger.error("worker_pool.initworkerpool", "error resolving stdout pipe from command", err.error())
        return err
    }
    worker.stdoutreader = bufio.newreadersize(stdout, workerconfig.maxrequestsize)

    stderr, err := command.stderrpipe()
    if err != nil {
        logger.error("worker_pool.initworkerpool", "error resolving stderror pipe from command", err.error())
        return err
    }
    worker.stderror = stderr

    err = command.start()
    if err != nil {
        logger.error("worker_pool.initworkerpool", "error starting command", err.error())
        return err
    }

    go processworkerpool(&worker, reqchan, logger)
    return err
}

当共享通道接收到作业时，它会被消耗并发送到 python 脚本。

func processworkerpool(worker *entities.worker, reqchannel chan entities.reqmessage, logger log.prefixedlogger) {

    for request := range reqchannel {
        bufferlatency.observe(float64(time.since(request.senttime).nanoseconds()/1e6), map[string]string{"name": "buffer", "error": "false"})

        logger.info("worker.processworkerpool", request.request)

        starttime := time.now()

        //send request to worker
        _, err := io.writestring(worker.stdin, request.request)
        if err != nil {
            scriptlatency.observe(float64(time.since(starttime).nanoseconds()/1e6), map[string]string{"name": "script", "error": "true"})
            log.errorcontext(context.background(), log.withprefix("worker.processworkerpool", err))
            return
        }

        //get response from worker
        result := copyoutput(logger, worker.stdoutreader)

        scriptlatency.observe(float64(time.since(starttime).nanoseconds()/1e6), map[string]string{"name": "script", "error": "false"})
        request.responsechannel <- result
    }
}

要从 python 脚本的 stdout 读取结果，我使用以下辅助函数

func copyoutput(logger log.prefixedlogger, r io.reader) string {
    scanner := bufio.newscanner(r)
    result := ""

    for scanner.scan() {
        output := scanner.text()

        switch {
        case strings.contains(output, "error"):
            errormsg := strings.splitafter(output, "error: ")[1]
            err := errors.new(errormsg)
            logger.error("worker.copyoutput", "error log: ", err.error())
            return err.error()
        case strings.contains(output, "output"):
            result = strings.splitafter(output, "output: ")[1]
            logger.debug("worker.copyoutput", "output log: ", result)
            return result
        default:
            logger.debug("worker.copyoutput", "debug log: ", output)
        }
    }
    return result
}

在 python 端我的脚本看起来像这样

#!/usr/bin/python3
import sys
import json
from threading import thread

from vrpsolver.ride_model import rides_model
from preprocessor.config_loader import config


# load configs
configs = config('/opt/pool-processor/configs/configs.yaml')

while true:
    
    # input = json.loads(sys.argv[1])
    # model = sys.argv[2]
    # file = sys.argv[3]
    
    threads = []
    try:
        inputdatastream = sys.stdin.readline()
        inputdatastream = inputdatastream.strip()
        data = inputdatastream.split(' ')
        model = data[1]
    except (exception) as ex:
        sys.stdout.write('error: error occured while reading stdin: {}\n'.format(str(ex)))
        sys.stdout.flush()
        continue
    
    try:
        input = json.loads(data[0])
    except (exception, ioerror) as ex:
        sys.stdout.write('error: error occured while parsing data to json: {}\n'.format(str(ex)))
        continue

    try:
        result = rides_model(input, configs)
        sys.stdout.write('output: {}\n'.format(json.dumps(result)))
        sys.stdout.flush()
    except (exception, ioerror) as ex:
        sys.stdout.write('error: error occured while processing: {}\n'.format(str(ex)))
        sys.stdout.flush()
        continue

当我运行程序一段时间后，我得到了

write |1: broken pipe on /build/pool-engine/worker_pool/worker.go:76
write |1: broken pipe on /build/pool-engine/worker_pool/worker.go:83

从以下几行开始

_, err := io.WriteString(worker.Stdin, request.Request)
result := CopyOutput(logger, worker.StdOutReader)

我现在在这个问题上停留了一段时间，对此的任何意见都会受到赞赏。我的猜测是，一段时间后 python 脚本崩溃了，结果我收到了这个错误。我不确定为什么没有从异常中捕获崩溃错误。

正确答案

这个错误的基本答案是：由于某种原因，你的python进程已经关闭了它的stdin（它可能由于某种原因退出了），检查为什么它退出得太早。

关于为什么你很难看到你的 python 进程做了什么的一些元素：

• 其主要“活动日志”位于 sys.stdout，
• stdout 被你的 go 程序捕获并处理（副作用是：它不会打印到控制台），
• 在 go 中处理子进程输出的方式存在一些问题。

为了使调试更容易，我建议您让 python 脚本也将其输出写入日志文件中。

我看到的前三个问题是：

• 您在 python 进程上设置了 stderrpipe，但从未使用过它，因此 stderr 完全被静音

尝试不重定向stderr（您应该看到stderr打印在控制台上），或者至少添加一个额外的goroutine来排出并在某处打印其内容（在您的go进程stderr上，在一个日志文件...）

例如：

go func() {
   io.Copy(os.Stderr, worker.StdErrReader)
}()

• 要将子进程输出读取为文本行，请通过 stdout 管道重复创建一个新的 bufio.scanner

当您运行 bufio.newscanner(...) 时，将创建一个带有新缓冲区的新缓冲读取器。如果你丢弃它并创建新的 scanner，则之前的缓冲区将被丢弃，并且你不知道从底层 io.reader 读取了多少字节（有些可能已被缓冲...）。

至少，您应该仅实例化您的 bufio.newscanner() 一次（在 processworkerpool() 中），并在该单个 *bufio.scanner 实例上重复调用 scanner.scan() ，以便使用相同的缓冲区。

• 您应该以某种方式监视正在运行的进程的状态

保留一个访问commmand.processstate的方法，并检查您的外部命令是否已完成。

好了，本文到此结束，带大家了解了《损坏管道错误：从 Go 到 Python 应用程序写入 stdin 时发生问题》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！