如何将 IO 操作的输出通过管道传输到 haskell 中的进程中

Question

我想创建一个进程并定期将一些文本从我的 haskell 程序写入进程的标准输入（来自 IO 操作）。

以下内容在 GHCi 中正常工作，但在构建和运行时无法正常工作。在 GHCi 中，一切都运行良好，并且定期输入 IO 操作的值。然而，当构建并运行时，它似乎在写入进程的标准输入时暂停了任意长的时间。

我使用CreateProcess（来自System.Process）创建句柄并尝试hPutStrLn（缓冲区设置为NoBuffering -- LineBuffering 也不起作用）。

所以我正在尝试process-streaming 包和pipes，但似乎根本无法工作。

真正的问题是：我如何从 haskell 创建一个进程并定期写入它？

展示此行为的最小示例：

import System.Process
import Data.IORef
import qualified Data.Text as T  -- from the text package
import qualified Data.Text.IO as TIO
import Control.Concurrent.Timer  -- from the timers package
import Control.Concurrent.Suspend -- from the suspend package

main = do
    (Just hin, _,_,_) <- createProcess_ "bgProcess" $
        (System.Process.proc "grep"  ["10"]) { std_in = CreatePipe }

    ref <- newIORef 0 :: IO (IORef Int)
    flip repeatedTimer (msDelay 1000) $ do
        x <- atomicModifyIORef' ref $ \x -> (x + 1, x)
        hSetBuffering hin NoBuffering
        TIO.hPutStrLn hin $ T.pack $ show x

任何帮助将不胜感激。

Answer 1

这是一个管道Producer，它发出一个带有第二个延迟的数字序列：

{-# language NumDecimals #-}
import Control.Concurrent
import Pipes
import qualified Data.ByteString.Char8 as Bytes

periodic :: Producer Bytes.ByteString IO ()
periodic = go 0
    where
        go n = do
            d <- liftIO (pure (Bytes.pack (show n ++ "\n"))) -- put your IO action here
            Pipes.yield d
            liftIO (threadDelay 1e6)
            go (succ n)

并且，使用process-streaming，我们可以将生产者提供给这样的外部进程：

import System.Process.Streaming

main :: IO ()
main = do
    executeInteractive (shell "grep 10"){ std_in = CreatePipe } (feedProducer periodic)

我使用了executeInteractive，它将std_in 自动设置为NoBuffering。

此外，如果您使用管道 std_out 并希望立即处理每个匹配项，请务必将 --line-buffered 选项传递给 grep（或使用 stdbuf 命令）以确保匹配项在输出中立即可用。

Answer 2

如何使用threadDelay，例如：

import Control.Monad (forever)
import Control.Concurrent (threadDelay)
...

forever $ do
    x <- atomicModifyIORef' ref $ \x -> (x + 1, x)
    hSetBuffering hin NoBuffering
    TIO.hPutStrLn hin $ T.pack $ show x
    threadDelay 1000000  -- 1 sec

如果您需要同时做其他工作，请在另一个线程中执行此操作。

您可以通过以下方式消除他对 IORef 的需求：

loop h x = do 
    hSetBuffering h NoBuffering
    TIO.hPutStrLn h $ T.pack $ show x
    threadDelay 1000000
    loop h (x+1)

当然，您只需要执行一次hSetBuffering - 例如在进入循环之前执行此操作。