持久的子进程管道 - 没有读取标准输出

Posted 2021-04-12

我有一个处理链，沿着这些方向：

1. 通过几个步骤预处理数据，包括从单个Bash脚本调用perl，Bash和python脚本，通过管道连接这些脚本
2. 在python中转换数据（我遗憾地使用的程序不能在Python 3上运行，所以我想我被迫运行2.7）
3. 后处理数据就像在预处理步骤中一样

以前的一种方法是

cat input | preprocess.sh | transform.py | postprocess.sh

这适用于处理批量输入数据。

但是，我现在发现自己需要在Python中将其实现为服务器功能 - 我需要能够接受单个数据项，运行管道并快速将其吐出。

我只是在Python中调用的中心步骤，所以这是最简单的部分。后处理也相对容易。

问题在于：预处理代码由4个不同的脚本组成，每个脚本将数据输出到下一个脚本，其中两个脚本需要从磁盘加载模型文件才能工作。这种加载速度相对较慢，对我的执行时间来说也很糟糕。因此，我认为我需要以某种方式将它们保存在内存中，写入它们的stdins并读取输出。

但是，我发现对于我的链中的每个链接，我都不能写入stdin并在不关闭stdout的情况下读取stdin，这会使该方法无用，因为我必须重新打开该过程并再次加载模型。

请注意，对于链中的每个链接，这不是我的脚本的问题

cat input_data | preprocessing_script_i.sh

返回Bash中的内容。

这是我到目前为止尝试过的事情：

• 只需写入stdin和flush它 - 在readline上无限期等待
• process.communicate - 杀死这个过程，因此是不可能的。
• 使用主和从pty句柄 - 挂在readline上
• 在从主线程写入stdout时使用队列和线程读取stdin
• 在bufsize的电话中与subprocess搞乱

有没有办法从Python做到这一点？这是否有可能，因为我开始怀疑这一点？可以重新实现这个管道（不触及元素，因为对我的用例来说不太可行）在另一种语言中为我解决这个问题？

答案

最简单的可能是来自同一文件系统的mv文件（因为重命名是原子文件操作而cp不是原子的）到“输入目录”。 shell无限循环并等待新文件mv到“工作目录”处理它和mvit在“完成目录”或“错误目录”中。

另一答案

您可以避免使用FIFO的stdin / stdout相关问题

os.mkfifo("/tmp/fifo")

然后，您可以将Python中的这个用作文件，用于从不同进程读取和写入目的，您甚至可以查看读者中的fifo（Python: Check if named pipe has data）以检查是否有东西可以在那里阅读。

如果没有这样的帮助，另一种方法是用消息传递平台取代进程间通信。 ZeroMQ（zmq）很容易实现，不需要任何服务器组件，然后你就可以摆脱链接的输入和输出。您只需从一个进程发布消息并从另一个进程读取它们。数据传输但使用线程读取器，您不会遇到阻塞IO。

另一答案

对不起，提出的想法很棒，这可能不会帮助很多人，但这就是我解决问题的方法。

事实证明perl有一个-b标志用于在线缓冲模式下打印。一旦我将其插入处理管道的perl -b script.perl部分，事情开始顺利进行，简单的process.write()跟随.flush()足以获得输出。

我将尝试更改问题标签和标题以更好地适应实际问题。