[Haskell-cafe] Blocking IO & FIFOs

[Jason Dusek]

2012/10/20 Wilfried van Asten <sniperrifle2004 at gmail.com>:
> Perhaps an interleaving solution as in process-conduit is
> still viable:
>
>  - Check if one or both of the fifo's are still ready (Based
>    on your statement about the reading end receiving EOF
>    hIsEOF should work here). If both fifos are done the query
>    is finished so break the loop.

Alas, checking for EOF does not work. I mentioned this in
passing in my prior email; the code was somewhat involved and I
have deleted it. Here is a simple example of something that does
not work as expected:

  In the first terminal:

   :; mkfifo fifo
   :; ghci
    --  :m + GHC.IO.Handle.FD System.IO
    --  do { h <- openFileBlocking "fifo" ReadMode ; hGetContents h }

  In the second terminal, *after* doing everything in the first
  terminal:

   :; cat > fifo
    < type some characters here >
    ^D

Notice that the characters appear in the first terminal, as the
output of hGetContents. Sending ^D to end cat does not register
any effect in GHCi; hGetContents dutifully waits and you can in
fact run cat on the FIFO again to send more characters to the
same instances of hGetContents. This would seem to be due to
non-blocking IO, deep in the IO manager.

>  - Check if some output is available on oh. If so read some of
>    it. Repeat.
>
>  - Otherwise check if some output is available on eh. If so
>    read some of it. Repeat loop
>
> I also see you don't do anything with the std_out and std_err
> pipes of bash as given by runInteractiveProcess. These could
> also cause a problem even when the FIFO's are working
> correctly. Replace these by handles to the null file or let
> the output be dumped on the parent's std_in and std_out
> (StdStream Inherit).

I would prefer to leave them be, since they're passed in from
the caller, who nominally owns them. If you mean that I should
close them in `start', well, that would make it hard to debug
this stuff; and if I simply tie them to the parent's file
descriptors, it will make it hard to deal with more than a few
CoBashes at one time while testing.

Using cat to read the FIFOs and allowing Haskell to read from
cat does work, actually.

  https://gist.github.com/3923673

Shell really is such a nice language for tying together
processes.

--
Jason Dusek
pgp // solidsnack // C1EBC57DC55144F35460C8DF1FD4C6C1FED18A2B




{-# LANGUAGE OverloadedStrings
           , ScopedTypeVariables
           , ParallelListComp
           , TupleSections #-}

module CoBash where

import           Control.Applicative
import           Control.Concurrent
import           Control.Concurrent.MVar
import           Control.Exception
import           Control.Monad
import           Data.Bits
import           Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as Bytes
import           Data.Maybe
import           Data.Monoid
import qualified GHC.IO.Handle.FD
import           System.IO
import           System.IO.Error
import           System.Process
import           System.Posix.ByteString

import           System.IO.Temp

import qualified Text.ShellEscape as Esc


start :: IO (Handle, Handle, Handle, ProcessHandle)
start = runInteractiveProcess "bash" [] Nothing (Just [])

query :: (Handle, Handle, Handle, ProcessHandle) -> ByteString
      -> IO (ByteString, ByteString)
query (i, _, _, _) query = withFIFOs query'
 where query' ofo efo = do
         Bytes.hPut i cmd
         hFlush i
         [ob, eb] <- backgroundReadFIFOs [ofo, efo]
         return (ob, eb)
        where cmd = Bytes.unlines ["{", query, "} 1>" <> ofo <> " 2>" <> efo]

shutdown :: (Handle, Handle, Handle, ProcessHandle) -> IO ()
shutdown (i, _, _, p) = () <$ hClose i <* waitForProcess p


openFIFO path = GHC.IO.Handle.FD.openFileBlocking (Bytes.unpack path) ReadMode

-- | Run an IO action with two FIFOs in scope, which will removed after it
--   completes.
withFIFOs :: (RawFilePath -> RawFilePath -> IO a) -> IO a
withFIFOs m = withSystemTempDirectory "cobash." m'
 where m'   = (uncurry m =<<) . mk . Bytes.pack
       mk d = (o, e) <$ (createNamedPipe o mode >> createNamedPipe e mode)
        where (o, e) = (d <> "/o", d <> "/e")
              mode   = ownerReadMode .|. ownerWriteMode .|. namedPipeMode

drainFIFO :: ByteString -> IO ByteString
drainFIFO path = do
  (i, o, e, p) <- bash ["-c", "exec cat <"<>(Bytes.unpack path)]
  hClose i
  hClose e
  Bytes.hGetContents o <* waitForProcess p

backgroundReadFIFOs theFIFOs = do
  cells <- sequence (newEmptyMVar <$ theFIFOs)
  sequence_ [ forkIO (drainFIFO p >>= putMVar c) | p <- theFIFOs | c <- cells ]
  sequence (takeMVar <$> cells)

bash args = runInteractiveProcess "bash" args Nothing (Just [])