diff --git a/lib/Pipes/Cliff.hs b/lib/Pipes/Cliff.hs --- a/lib/Pipes/Cliff.hs +++ b/lib/Pipes/Cliff.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE RankNTypes #-} -- | Spawn subprocesses and interact with them using "Pipes" -- -- The interface in this module deliberately resembles the interface @@ -12,7 +11,7 @@ -- 'pipeInput' or 'pipeInputOutput' to specify what streams you want -- to use a 'Proxy' for and what streams you wish to be 'Inherit'ed -- or if you want to 'UseHandle'. You then send or receive --- information using the returned 'Proxy'. +-- information using one or more 'Proxy'. -- -- __Use the @-threaded@ GHC option__ when compiling your programs or -- when using GHCi. Internally, this module uses @@ -48,12 +47,16 @@ -- There you will find references to other libraries that you might -- find more useful than this one. -- --- For some simple examples, consult "Pipes.Cliff.Examples". +-- You will want to consult "Pipes.Cliff.Examples" for some examples +-- before getting started. There are some important notes in there +-- about how to run pipelines. module Pipes.Cliff ( -- * Specifying a subprocess's properties - NonPipe(..) + CmdSpec(..) + , NonPipe(..) , CreateProcess(..) , procSpec + , squelch -- * Creating processes -- $process @@ -78,418 +81,33 @@ , waitForProcess , waitForThread + -- * Errors and warnings + + -- | You will only need what's in this section if you want to + -- examine errors more closely. + , Activity(..) + , HandleDesc(..) + , HandleOopsie(..) + , Oopsie(..) + -- * Re-exports -- $reexports + , module Control.Concurrent.MVar , module Pipes , module Pipes.Safe , module System.Exit , module System.Process + + -- * Some design notes + -- $designNotes ) where -import System.IO +import Pipes.Cliff.Core import Pipes -import Pipes.Concurrent -import Control.Concurrent.Async (Async, async, cancel, wait) -import Data.ByteString (ByteString) -import qualified Data.ByteString as BS -import qualified System.Process as Process -import System.Process (ProcessHandle) -import qualified Control.Exception -import Pipes.Safe (runSafeT) -import qualified Pipes.Safe +import Pipes.Safe import System.Exit -import System.Environment (getProgName) -import qualified System.IO as IO -import Control.Monad (when, liftM) - --- # Exception handling - --- | Runs a particular action. Any IO errors are caught; a warning --- message is printed if we're not being quiet, and 'Nothing' is --- returned. Otherwise, the result is returned. - -catchAndWarn - :: (Pipes.Safe.MonadCatch m, MonadIO m) - => Bool - -- ^ Be quiet? - -> String - -- ^ What are we doing? Used for warning message. - -> m a - -- ^ Do this - -> m (Maybe a) -catchAndWarn beQuiet desc act - = Pipes.Safe.catch (liftM Just act) handle - where - handle e = do - let _types = e :: Control.Exception.IOException - when (not beQuiet) . liftIO $ do - pn <- getProgName - let msg = pn ++ ": warning: caught exception when " - ++ desc ++ ": " ++ show e - IO.hPutStrLn IO.stderr msg - return Nothing - --- | Runs a particular action, ignoring all IO errors. Sometimes --- using hClose will result in a broken pipe error. Since the process --- may have already been shut down, this is to be expected. Since --- there is nothing that can really be done to respond to any IO error --- that results from closing a handle, just ignore these errors. -ignoreIOExceptions :: Bool -> String -> IO () -> IO () -ignoreIOExceptions beQuiet desc a = Control.Exception.catch a f - where - f :: Control.Exception.IOException -> IO () - f exc = when (not beQuiet) $ do - pn <- getProgName - let msg = pn ++ ": warning: ignoring exception caught when " ++ desc - ++ ": " ++ show exc - IO.hPutStrLn IO.stderr msg - return () - --- | Close a handle, ignoring all IO exceptions. -closeHandle - :: Bool - -- ^ Be quiet? - -> Handle - -> IO () -closeHandle q h = ignoreIOExceptions q "closing handle" (hClose h) - --- | Terminate a process, ignoring all IO exceptions. -terminateProcess - :: Bool - -- ^ Be quiet? - -> Process.ProcessHandle - -> IO () -terminateProcess q h = ignoreIOExceptions q "terminating process" $ do - _ <- Process.terminateProcess h - _ <- Process.waitForProcess h - return () - - --- # Configuration and result types - --- | How will the subprocess get its information for this stream? -data NonPipe - = Inherit - -- ^ Use whatever stream that the parent process has. - | UseHandle Handle - -- ^ Use the given handle for input or output - --- | Like 'System.Process.CreateProcess' in "System.Process", --- this gives the necessary information to create a subprocess. All --- but one of these fields is also present in --- 'System.Process.CreateProcess', and they all have the same meaning; --- the only field that is different is the 'quiet' field. -data CreateProcess = CreateProcess - { cmdspec :: Process.CmdSpec - -- ^ Executable and arguments, or shell command - , cwd :: Maybe FilePath - -- ^ A new current working directory for the subprocess; if - -- 'Nothing', use the calling process's working directory. - , env :: Maybe [(String, String)] - -- ^ The environment for the subprocess; if 'Nothing', use the - -- calling process's working directory. - - , close_fds :: Bool - -- ^ If 'True', close all file descriptors other than the standard - -- descriptors. See the documentation for - -- 'System.Process.close_fds' for details on how this works in - -- Windows. - , create_group :: Bool - -- ^ If 'True', create a new process group. - , delegate_ctlc :: Bool - -- ^ See 'System.Process.delegate_ctlc' in the "System.Process" - -- module for details. - - , quiet :: Bool - -- ^ If True, does not print messages to standard error when IO - -- exceptions arise when closing handles or terminating processes. - -- Sometimes these errors arise due to broken pipes; this can be - -- normal, depending on the circumstances. For example, if you - -- are streaming a large set of values to a pager such as @less@ - -- and you expect that the user will often quit the pager without - -- viewing the whole result, a broken pipe will result, which will - -- print a warning message. That can be a nuisance. If you don't - -- want to see these errors, set 'quiet' to 'True'. - } - -convertCreateProcess - :: Maybe NonPipe - -> Maybe NonPipe - -> Maybe NonPipe - -> CreateProcess - -> Process.CreateProcess -convertCreateProcess inp out err a = Process.CreateProcess - { Process.cmdspec = cmdspec a - , Process.cwd = cwd a - , Process.env = env a - , Process.std_in = conv inp - , Process.std_out = conv out - , Process.std_err = conv err - , Process.close_fds = close_fds a - , Process.create_group = create_group a - , Process.delegate_ctlc = delegate_ctlc a - } - where - conv = convertNonPipe - -convertNonPipe :: Maybe NonPipe -> Process.StdStream -convertNonPipe a = case a of - Nothing -> Process.CreatePipe - Just Inherit -> Process.Inherit - Just (UseHandle h) -> Process.UseHandle h - --- | Create a 'CreateProcess' record with default settings. The --- default settings are: --- --- * a raw command (as opposed to a shell command) is created --- --- * the current working directory is not changed from the parent process --- --- * the environment is not changed from the parent process --- --- * the parent's file descriptors (other than standard input, --- standard output, and standard error) are inherited --- --- * no new process group is created --- --- * 'delegate_ctlc' is 'False' --- --- * 'quiet' is 'False' - -procSpec - :: String - -- ^ The name of the program to run, such as @less@. - -> [String] - -- ^ Command-line arguments - -> CreateProcess -procSpec prog args = CreateProcess - { cmdspec = Process.RawCommand prog args - , cwd = Nothing - , env = Nothing - , close_fds = False - , create_group = False - , delegate_ctlc = False - , quiet = False - } - - --- # Pipes - --- | I have no idea what this should be. I'll start with a simple --- small value and see how it works. -bufSize :: Int -bufSize = 1024 - --- | Create a 'Producer' from a 'Handle'. The 'Producer' will get --- 'ByteString' from the 'Handle' and produce them. Does nothing to --- close the given 'Handle' at any time. If there is an IO error --- while receiving data, the error is caught and production ceases. --- The exception is not re-thrown. -produceFromHandle - :: (Pipes.Safe.MonadCatch m, MonadIO m) - => Bool - -- ^ Be quiet? - -> Handle - -> Producer ByteString m () -produceFromHandle beQuiet h = catchAndWarn beQuiet desc act >>= go - where - desc = "receiving data from process" - act = liftIO (BS.hGetSome h bufSize) - go Nothing = return () - go (Just bs) - | BS.null bs = return () - | otherwise = yield bs >> produceFromHandle beQuiet h - - --- | Create a 'Consumer' from a 'Handle'. The 'Consumer' will put --- each 'ByteString' it receives into the 'Handle'. Does nothing to --- close the handle at any time. If there is an IO error while --- sending data, the error is caught and consumption ceases. The --- exception is not re-thrown. -consumeToHandle - :: (Pipes.Safe.MonadCatch m, MonadIO m) - => Bool - -- ^ Be quiet? - -> Handle - -> Consumer ByteString m () -consumeToHandle beQuiet h = do - bs <- await - mayRes <- catchAndWarn beQuiet "sending data to process" - (liftIO $ BS.hPut h bs) - case mayRes of - Nothing -> return () - Just _ -> consumeToHandle beQuiet h - - --- | A buffer that holds 10 messages. I have no idea if this is the --- ideal size. Don't use an unbounded buffer, though, because with --- unbounded producers an unbounded buffer will fill up your RAM. -messageBuffer :: Buffer a -messageBuffer = bounded 10 - --- | Acquires a resource and registers a finalizer. -initialize - :: Pipes.Safe.MonadSafe m - => IO a - -> (a -> IO ()) - -> m a -initialize make destroy = Pipes.Safe.mask $ \_ -> do - thing <- liftIO make - _ <- Pipes.Safe.register (liftIO (destroy thing)) - return thing - --- | Creates a mailbox; seals it when done. -newMailbox - :: Pipes.Safe.MonadSafe m - => m (Output a, Input a, STM ()) -newMailbox = - initialize (spawn' messageBuffer) - (\(_, _, seal) -> atomically seal) - --- | Runs a thread in the background. Initializes a finalizer that --- will cancel the thread if it is still running when the --- 'Pipes.Safe.MonadSafe' computation completes. -background :: Pipes.Safe.MonadSafe m => IO a -> m (Async a) -background action = initialize (async action) cancel - --- | Creates a thread that will run in the background and pump --- messages from the given mailbox to the process via its handle. --- Closes the Handle when done. -processPump - :: Pipes.Safe.MonadSafe m - => Bool - -- ^ Quiet? - -> Handle - -> (Input ByteString, STM ()) - -> m () -processPump beQuiet hndle (input, seal) = do - let pumper = flip Control.Exception.finally cleanup . - runEffect $ - fromInput input >-> consumeToHandle beQuiet hndle - _ <- background pumper - return () - where - cleanup = liftIO $ do - closeHandle beQuiet hndle - atomically seal - --- | Creates a thread that will run in the background and pull --- messages from the process and place them into the given mailbox. --- Closes the handle when done. -processPull - :: Pipes.Safe.MonadSafe m - => Bool - -- ^ Quiet? - -> Handle - -> (Output ByteString, STM ()) - -- ^ Output box, paired with action to close the box. - -> m () -processPull beQuiet hndle (output, seal) = do - let puller = flip Control.Exception.finally cleanup . - runEffect $ - produceFromHandle beQuiet hndle >-> toOutput output - _ <- background puller - return () - where - cleanup = liftIO $ do - closeHandle beQuiet hndle - atomically seal - - --- | Creates a mailbox that sends messages to the given process, and --- sets up and runs threads to pump messages to the process. -makeToProcess - :: (Pipes.Safe.MonadSafe mp, Pipes.Safe.MonadSafe m) - => Bool - -- ^ Quiet? - -> Handle - -> m (Consumer ByteString mp ()) -makeToProcess beQuiet hndle = do - (out, inp, seal) <- newMailbox - processPump beQuiet hndle (inp, seal) - return $ toOutput out `Pipes.Safe.finally` (liftIO (atomically seal)) - --- | Creates a mailbox that receives messages from the given process, --- and sets up and runs threads to receive the messages and deliver --- them to the mailbox. -makeFromProcess - :: (Pipes.Safe.MonadSafe m, Pipes.Safe.MonadSafe mp) - => Bool - -- ^ Quiet? - -> Handle - -> m (Producer ByteString mp ()) -makeFromProcess beQuiet hndle = do - (out, inp, seal) <- newMailbox - processPull beQuiet hndle (out, seal) - return $ fromInput inp `Pipes.Safe.finally` (liftIO (atomically seal)) - - --- # Subprocesses - --- Waiting on a process handle more than once will merely return the --- same code more than once. See the source code for System.Process. - --- | Launch and use a subprocess. --- --- /Warning/ - do not attempt to use any of the resources created by --- the process after leaving the 'ContT' computation. They are all --- destroyed. So any Pipes you create using the 'Output' or 'Input' --- that connect to the process must finish their IO before you leave --- the 'ContT' computation. It's okay to return the --- 'System.Exit.ExitCode' that you get from running the process, or --- any data you get from the process--you just can't return something --- that must perform IO to interact with the process. --- --- Also, exiting the 'ContT' computation immediately destroys all --- subprocesses. If you want to make sure the process terminates --- first, use 'Process.waitForProcess' on the handle which you can get --- from 'mbxHandle' before leaving the 'ContT' computation. --- --- The upside of this warning is that because all subprocess resources --- are destroyed after leaving the 'ContT' computation, this function --- is exception safe. --- --- To increase the safety when using values with the 'ContT' type, --- you can use 'safeCliff'. - - --- | Creates a subprocess. Registers destroyers for each handle --- created, as well as for the ProcessHandle. -createProcess - :: Pipes.Safe.MonadSafe m - => Bool - -- ^ Quiet? - -> Process.CreateProcess - -> m (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -createProcess beQuiet cp = initialize (Process.createProcess cp) destroy - where - destroy (mayIn, mayOut, mayErr, han) = do - let close = maybe (return ()) (closeHandle beQuiet) - close mayIn - close mayOut - close mayErr - terminateProcess beQuiet han - - --- | Runs in the background an effect, typically one that is moving --- data from one process to another. For examples of its usage, see --- "Pipes.Cliff.Examples". The associated thread is killed when the --- 'SafeT' computation completes. -conveyor :: Effect (Pipes.Safe.SafeT IO) () -> Pipes.Safe.SafeT IO () -conveyor efct - = (background . liftIO . runSafeT . runEffect $ efct) >> return () - --- | A version of 'System.Process.waitForProcess' with an overloaded --- 'MonadIO' return type. -waitForProcess :: MonadIO m => ProcessHandle -> m ExitCode -waitForProcess h = liftIO $ Process.waitForProcess h - --- | A version of 'Control.Concurrent.Async.wait' with an overloaded --- 'MonadIO' return type. Allows you to wait for the return value of --- threads launched with 'background'. If the thread throws an --- exception, 'waitForThread' will throw that same exception. -waitForThread :: MonadIO m => Async a -> m a -waitForThread = liftIO . wait +import System.Process (ProcessHandle) +import Control.Concurrent.MVar {- $process @@ -502,198 +120,76 @@ subprocess that does not create a Pipe for any of the standard streams, use 'pipeNone'. You must describe what you want done with standard input, standard output, and standard error. To create a -subprocess that creates a Pipe for standard input and standard +subprocess that creates a 'Proxy' for standard input and standard output, use 'pipeInputOutput'. You must describe what you want done with standard error. A 'Producer' is returned for standard output and a 'Consumer' for standard input. -Do NOT attempt to use any of the resources created by this function -outside of the 'Pipes.Safe.MonadSafe' computation. The -'Pipes.Safe.MonadSafe' computation will destroy all resources when -the 'Pipes.Safe.MonadSafe' computation is complete. This means that -all these functions are exception safe: all resources--threads and -processes included--are destroyed when the 'Pipes.Safe.MonadSafe' -computation is complete, even if an exception is thrown. However, -this does mean that you need to stay in the 'Pipes.Safe.MonadSafe' -computation if you need to keep a resource around. 'waitForProcess' -can be handy for this. All functions in this section return a -'ProcessHandle' for use with 'waitForProcess'. +If you are creating a 'Proxy' for only one stream (for instance, +you're using 'pipeOutput') then a single 'Proxy' is returned to you. +That 'Proxy' manages all the resources it creates; so, for example, +when you ultimately run your 'Effect', the process is created and then +destroyed when the 'MonadSafe' computation completes. -Each 'Proxy' automatically destroys associated file handles and other -behind-the-scenes resources after its computation finishes running; -that's why the monad stack of each 'Proxy' must contain a -'Pipes.Safe.MonadSafe'. For an example of how to deal with the -'Pipes.Safe.MonadSafe' class, consult "Pipes.Cliff.Examples". +If you are creating a 'Proxy' for more than one stream (for +instance, you're using 'pipeInputOutput') then the multiple 'Proxy' +are returned to you in a tuple in the 'MonadSafe' computation. The +'MonadSafe' computation will make sure that the resulting process +and handles are destroyed when you exit the 'MonadSafe' computation. +In such a case, you must make sure that you don't try to use the +streams outside of the 'MonadSafe' computation, because the +subprocess will already be destroyed. To make sure you are done +using the streams before leaving the 'MonadSafe' computation, you +will want to use 'waitForProcess' for one or more processes that you +are most interested in. +Every function in this section (except for the 'pipeNone' function) +returns a value of type @(a, h)@, where @a@ is the set of 'Proxy', +and @h@ is the 'ProcessHandle'; that means the functions that return +multiple 'Proxy' have a nested return type. That allows you to use +'fst' and 'snd' to pull out the part you are interested in. It +would have been more consistent for 'pipeNone' to return +@((), ProcessHandle)@ but that just seemed silly. + -} --- | Do not create any 'Proxy' to or from the process. -pipeNone - :: Pipes.Safe.MonadSafe m - => NonPipe - -- ^ Standard input - -> NonPipe - -- ^ Standard output - -> NonPipe - -- ^ Standard error - -> CreateProcess - -> m Process.ProcessHandle -pipeNone sIn sOut sErr cp = do - (_, _, _, han) <- createProcess (quiet cp) cp' - return han - where - cp' = convertCreateProcess (Just sIn) (Just sOut) (Just sErr) - cp +{- $reexports --- | Create a 'Consumer' for standard input. -pipeInput - :: (Pipes.Safe.MonadSafe mi, Pipes.Safe.MonadSafe m) - => NonPipe - -- ^ Standard output - -> NonPipe - -- ^ Standard error - -> CreateProcess - -> m (Consumer ByteString mi (), Process.ProcessHandle) - -- ^ A 'Consumer' for standard input, and the 'ProcessHandle' -pipeInput sOut sErr cp = do - (Just inp, _, _, han) <- createProcess (quiet cp) cp' - inp' <- makeToProcess (quiet cp) inp - return (inp', han) - where - cp' = convertCreateProcess Nothing (Just sOut) (Just sErr) - cp + * "Control.Concurrent.MVar" reexports all bindings --- | Create a 'Producer' for standard output. -pipeOutput - :: (Pipes.Safe.MonadSafe mi, Pipes.Safe.MonadSafe m) - => NonPipe - -- ^ Standard input - -> NonPipe - -- ^ Standard error - -> CreateProcess - -> m (Producer ByteString mi (), Process.ProcessHandle) - -- ^ A 'Producer' for standard output, and the 'ProcessHandle' -pipeOutput sIn sErr cp = do - (_, Just out, _, han) <- createProcess (quiet cp) cp' - out' <- makeFromProcess (quiet cp) out - return (out', han) - where - cp' = convertCreateProcess (Just sIn) Nothing (Just sErr) cp + * "Pipes" reexports all bindings --- | Create a 'Producer' for standard error. -pipeError - :: (Pipes.Safe.MonadSafe mi, Pipes.Safe.MonadSafe m) - => NonPipe - -- ^ Standard input - -> NonPipe - -- ^ Standard output - -> CreateProcess - -> m (Producer ByteString mi (), Process.ProcessHandle) - -- ^ A 'Producer' for standard error, and the 'ProcessHandle' -pipeError sIn sOut cp = do - (_, _, Just err, han) <- createProcess (quiet cp) cp' - err' <- makeFromProcess (quiet cp) err - return (err', han) - where - cp' = convertCreateProcess (Just sIn) (Just sOut) Nothing cp + * "Pipes.Safe" reexports all bindings --- | Create a 'Consumer' for standard input and a 'Producer' for --- standard output. -pipeInputOutput - :: ( Pipes.Safe.MonadSafe mi, Pipes.Safe.MonadSafe mo - , Pipes.Safe.MonadSafe m) - => NonPipe - -- ^ Standard error - -> CreateProcess - -> m ( Consumer ByteString mi () - , Producer ByteString mo () - , Process.ProcessHandle - ) - -- ^ A 'Consumer' for standard input, a 'Producer' for standard - -- output, and a 'ProcessHandle' -pipeInputOutput sErr cp = do - (Just inp, Just out, _, han) <- createProcess (quiet cp) cp' - inp' <- makeToProcess (quiet cp) inp - out' <- makeFromProcess (quiet cp) out - return (inp', out', han) - where - cp' = convertCreateProcess Nothing Nothing (Just sErr) cp + * "System.Exit" reexports all bindings --- | Create a 'Consumer' for standard input and a 'Producer' for --- standard error. -pipeInputError - :: ( Pipes.Safe.MonadSafe mi, Pipes.Safe.MonadSafe mo - , Pipes.Safe.MonadSafe m) - => NonPipe - -- ^ Standard output - -> CreateProcess - -> m ( Consumer ByteString mi () - , Producer ByteString mo () - , Process.ProcessHandle - ) - -- ^ A 'Consumer' for standard input, a 'Producer' for standard - -- error, and a 'ProcessHandle' -pipeInputError sOut cp = do - (Just inp, _, Just err, han) <- createProcess (quiet cp) cp' - inp' <- makeToProcess (quiet cp) inp - err' <- makeFromProcess (quiet cp) err - return (inp', err', han) - where - cp' = convertCreateProcess Nothing (Just sOut) Nothing cp + * "System.Process" reexports 'ProcessHandle' --- | Create a 'Producer' for standard output and a 'Producer' for --- standard error. -pipeOutputError - :: ( Pipes.Safe.MonadSafe mi - , Pipes.Safe.MonadSafe mo - , Pipes.Safe.MonadSafe m) - => NonPipe - -- ^ Standard input - -> CreateProcess - -> m ( Producer ByteString mi () - , Producer ByteString mo () - , Process.ProcessHandle - ) - -- ^ A 'Producer' for standard output, a 'Producer' for standard - -- error, and a 'ProcessHandle' -pipeOutputError sIn cp = do - (_, Just out, Just err, han) <- createProcess (quiet cp) cp' - out' <- makeFromProcess (quiet cp) out - err' <- makeFromProcess (quiet cp) err - return (out', err', han) - where - cp' = convertCreateProcess (Just sIn) Nothing Nothing cp +-} --- | Create a 'Consumer' for standard input, a 'Producer' for standard --- output, and a 'Producer' for standard error. -pipeInputOutputError - :: ( Pipes.Safe.MonadSafe mi, Pipes.Safe.MonadSafe mo, - Pipes.Safe.MonadSafe me, Pipes.Safe.MonadSafe m) - => CreateProcess - -> m ( Consumer ByteString mi () - , Producer ByteString mo () - , Producer ByteString me () - , Process.ProcessHandle - ) - -- ^ A 'Consumer' for standard input, a 'Producer' for standard - -- output, a 'Producer' for standard error, and a 'ProcessHandle' -pipeInputOutputError cp = do - (Just inp, Just out, Just err, han) <- createProcess (quiet cp) cp' - inp' <- makeToProcess (quiet cp) inp - out' <- makeFromProcess (quiet cp) out - err' <- makeFromProcess (quiet cp) err - return (inp', out', err', han) - where - cp' = convertCreateProcess Nothing Nothing Nothing cp +{- $designNotes -{- $reexports +Two overarching principles guided the design of this +library. First, I wanted the interface to use simple +ByteStrings. That most closely represents what a UNIX process +sees. If the user wants to use Text or String, it's easy +enough to convert between those types and a ByteString. Then +the user has to pay explicit attention to encoding issues--as +she should, because not all UNIX processes deal with encoded +textual data. - * "Pipes" reexports all bindings +Second, I paid meticulous attention to resource management. +Resources are deterministically destroyed immediately after +use. This eliminates many bugs. - * "Pipes.Safe" reexports 'runSafeT' +You might wonder why, if you are using an external process as +a pipeline, why can't you create, well, a 'Pipe'? Wouldn't +that be an obvious choice? Well, if such an interface is +possible using Pipes in its current incarnation, nobody has +figured it out yet. I don't think it's possible. See also - * "System.Exit" reexports all bindings + - * "System.Process" reexports 'ProcessHandle' +for a discussion about this. -} diff --git a/lib/Pipes/Cliff/Core.hs b/lib/Pipes/Cliff/Core.hs new file mode 100644 --- /dev/null +++ b/lib/Pipes/Cliff/Core.hs @@ -0,0 +1,709 @@ +{-# LANGUAGE FlexibleContexts #-} + +-- | This contains the innards of Cliff. You probably won't need +-- anything that's in here; "Pipes.Cliff" re-exports the most useful +-- bindings. But nothing will break if you use what's in here, so +-- it's here if you need it. +module Pipes.Cliff.Core where + +import System.Environment +import Data.List (intersperse) +import Control.Exception (IOException) +import System.IO +import qualified System.Process as Process +import System.Process (ProcessHandle) +import Pipes +import Pipes.Safe +import qualified Data.ByteString as BS +import qualified Pipes.Concurrent as PC +import Data.ByteString (ByteString) +import Control.Concurrent.Async +import System.Exit + +-- * Data types + +-- | Like 'Process.CmdSpec' in "System.Process", but also has an +-- instance for 'Show'. +data CmdSpec + = ShellCommand String + | RawCommand FilePath [String] + deriving (Eq, Ord, Show) + +convertCmdSpec :: CmdSpec -> Process.CmdSpec +convertCmdSpec (ShellCommand s) = Process.ShellCommand s +convertCmdSpec (RawCommand p ss) = Process.RawCommand p ss + +-- ** Errors + +-- | When dealing with a 'Handle', errors can occur when reading from, +-- writing to, or closing the handle. +data Activity + = Reading + | Writing + | Closing + deriving (Eq, Ord, Show) + +-- | Describes a handle. From the perspective of the subprocess; for +-- example, 'Input' means that this handle is connected to the +-- process's standard input. +data HandleDesc + = Input + | Output + | Error + deriving (Eq, Ord, Show) + +-- | Describes IO errors tha occur when dealing with a 'Handle'. +data HandleOopsie = HandleOopsie Activity HandleDesc + deriving (Eq,Show) + +-- | Describes all IO exceptions. The 'Oopsie' contains the +-- 'IOException' itself, along with the 'CmdSpec' that was running +-- when the exception occurred. If the exception occurred while +-- dealing with a 'Handle', there is also a 'HandleOopsie'. If there +-- is no 'HandleOopsie', this means that the exception arose when +-- running 'terminateProcess'. +-- +-- The exceptions that are caught and placed into an 'Oopsie' may +-- arise from reading data from or writing data to a 'Handle'. In +-- these errors, the associated 'Producer' or 'Consumer' will +-- terminate (which may trigger various cleanup actions in the +-- 'MonadSafe' computation) but the exception itself is not re-thrown; +-- rather, it is passed to the 'handler'. Similarly, an exception may +-- occur while closing a handle; these exceptions are caught, not +-- rethrown, and are passed to the 'handler'. If an exception arises +-- when terminating a process (I'm not sure this is possible) then it +-- is also caught, not rethrown, and passed to the 'handler'. +-- +-- If an exception arises when creating a process--such as a command +-- not being found--the exception is /not/ caught, handled, or passed +-- to the 'handler'. Also, an 'Oopsie' is created only for an +-- 'IOException'; no other exceptions of any kind are caught or +-- handled. However, exceptions of any kind will still trigger +-- appropriate cleanup actions in the 'MonadSafe' computation. +data Oopsie = Oopsie (Maybe HandleOopsie) CmdSpec IOException + deriving (Eq, Show) + +-- | Formats an 'Oopsie' for display. +renderOopsie + :: String + -- ^ The name of the currently runnning program + -> Oopsie + -> String +renderOopsie pn (Oopsie mayHan cmd ioe) = + pn ++ ": warning: when running command " + ++ renderCommand cmd ++ ": " ++ renderMayHan mayHan + ++ ": " ++ show ioe + where + renderCommand (ShellCommand str) = show str + renderCommand (RawCommand fp ss) + = concat . intersperse " " . map show + $ fp : ss + + renderMayHan Nothing = "when terminating process" + renderMayHan (Just (HandleOopsie act desc)) = + "when " ++ actStr ++ " " ++ descStr + where + actStr = case act of + Reading -> "reading from" + Writing -> "writing to" + Closing -> "closing the handle associated with" + descStr = "standard " ++ case desc of + Input -> "input" + Output -> "output" + Error -> "error" + +-- | The default handler when receiving an 'Oopsie'; simply uses +-- 'renderOopsie' to format it nicely and put it on standard error. +defaultHandler :: Oopsie -> IO () +defaultHandler oops = do + pn <- getProgName + hPutStrLn stderr $ renderOopsie pn oops + +-- ** Configuration types + +-- | How will the subprocess get its information for this stream? A +-- 'NonPipe' is used for streams that will not be assigned to a +-- 'Proxy' but, instead, will be inherited from the parent or directed +-- from an existing 'Handle'. +data NonPipe + = Inherit + -- ^ Use whatever stream that the parent process has. + | UseHandle Handle + -- ^ Use the given handle for input or output + +convertNonPipe :: Maybe NonPipe -> Process.StdStream +convertNonPipe a = case a of + Nothing -> Process.CreatePipe + Just Inherit -> Process.Inherit + Just (UseHandle h) -> Process.UseHandle h + +-- | Like 'System.Process.CreateProcess' in "System.Process", +-- this gives the necessary information to create a subprocess. All +-- but one of these fields is also present in +-- 'System.Process.CreateProcess', and they all have the same meaning; +-- the only field that is different is the 'handler' field. +data CreateProcess = CreateProcess + { cmdspec :: CmdSpec + -- ^ Executable and arguments, or shell command + + , cwd :: Maybe FilePath + -- ^ A new current working directory for the subprocess; if + -- 'Nothing', use the calling process's working directory. + + , env :: Maybe [(String, String)] + -- ^ The environment for the subprocess; if 'Nothing', use the + -- calling process's working directory. + + , close_fds :: Bool + -- ^ If 'True', close all file descriptors other than the standard + -- descriptors. See the documentation for + -- 'System.Process.close_fds' for details on how this works in + -- Windows. + + , create_group :: Bool + -- ^ If 'True', create a new process group. + + , delegate_ctlc :: Bool + -- ^ See 'System.Process.delegate_ctlc' in the "System.Process" + -- module for details. + + , handler :: Oopsie -> IO () + -- ^ Whenever an IO exception arises during the course of various + -- IO actios, the exception is caught and placed into an 'Oopsie' + -- that indicates why and where the exception happened. The + -- 'handler' determines what happens when an 'Oopsie' comes in. + -- See 'Oopsie' for details. + -- + -- The default 'handler' created by 'procSpec' is + -- 'defaultHandler', which will simply print the exceptions to + -- standard error. You may not want to see the exceptions at all. + -- For example, many exceptions come from broken pipes. A broken + -- pipe might be entirely normal in your circumstance. For + -- example, if you are streaming a large set of values to a pager + -- such as @less@ and you expect that the user will often quit the + -- pager without viewing the whole result, a broken pipe will + -- result, which will print a warning message. That can be a + -- nuisance. + -- + -- If you don't want to see the exceptions at all, just set + -- 'handler' to 'squelch', which simply discards the exceptions. + -- + -- Conceivably you could rig up an elaborate mechanism that puts + -- the 'Oopsie's into a "Pipes.Concurrent" mailbox or something. + -- Indeed, when using 'defaultHandler' each thread will print its + -- warnings to standard error at any time. If you are using + -- multiple processes and each prints warnings at the same time, + -- total gibberish can result as the text gets mixed in. You + -- could solve this by putting the errors into a + -- "Pipes.Concurrent" mailbox and having a single thread print the + -- errors; building this sort of functionality directly in to the + -- library would clutter up the API somewhat so I have been + -- reluctant to do it. + } + +-- | Do not show or do anything with exceptions; useful to use as a +-- 'handler'. +squelch :: Monad m => a -> m () +squelch = const (return ()) + +-- | Create a 'CreateProcess' record with default settings. The +-- default settings are: +-- +-- * a raw command (as opposed to a shell command) is created +-- +-- * the current working directory is not changed from the parent process +-- +-- * the environment is not changed from the parent process +-- +-- * the parent's file descriptors (other than standard input, +-- standard output, and standard error) are inherited +-- +-- * no new process group is created +-- +-- * 'delegate_ctlc' is 'False' +-- +-- * 'storeProcessHandle' is 'Nothing' +-- +-- * 'handler' is 'defaultHandler' + +procSpec + :: String + -- ^ The name of the program to run, such as @less@. + -> [String] + -- ^ Command-line arguments + -> CreateProcess +procSpec prog args = CreateProcess + { cmdspec = RawCommand prog args + , cwd = Nothing + , env = Nothing + , close_fds = False + , create_group = False + , delegate_ctlc = False + , handler = defaultHandler + } + +convertCreateProcess + :: Maybe NonPipe + -> Maybe NonPipe + -> Maybe NonPipe + -> CreateProcess + -> Process.CreateProcess +convertCreateProcess inp out err a = Process.CreateProcess + { Process.cmdspec = convertCmdSpec $ cmdspec a + , Process.cwd = cwd a + , Process.env = env a + , Process.std_in = conv inp + , Process.std_out = conv out + , Process.std_err = conv err + , Process.close_fds = close_fds a + , Process.create_group = create_group a + , Process.delegate_ctlc = delegate_ctlc a + } + where + conv = convertNonPipe + +-- * ErrSpec + +-- | Contains data necessary to deal with exceptions. +data ErrSpec = ErrSpec + { esErrorHandler :: Oopsie -> IO () + , esCmdSpec :: CmdSpec + } + +makeErrSpec + :: CreateProcess + -> ErrSpec +makeErrSpec cp = ErrSpec + { esErrorHandler = handler cp + , esCmdSpec = cmdspec cp + } + + +-- * Exception handling + +-- | Sends an exception using the exception handler specified in the +-- 'ErrSpec'. +handleException + :: MonadIO m + => Maybe HandleOopsie + -> IOException + -> ErrSpec + -> m () +handleException mayOops exc ev = liftIO $ sender oops + where + spec = esCmdSpec ev + sender = esErrorHandler ev + oops = Oopsie mayOops spec exc + + +-- | Run an action, taking all IO errors and sending them to the handler. +handleErrors + :: (MonadCatch m, MonadIO m) + => Maybe HandleOopsie + -> ErrSpec + -> m () + -> m () +handleErrors mayHandleOops ev act = catch act catcher + where + catcher e = liftIO $ hndlr oops + where + spec = esCmdSpec ev + hndlr = esErrorHandler ev + oops = Oopsie mayHandleOops spec e + + +-- | Close a handle. Catches any exceptions and passes them to the handler. +closeHandleNoThrow + :: (MonadCatch m, MonadIO m) + => Handle + -> HandleDesc + -> ErrSpec + -> m () +closeHandleNoThrow hand desc ev = handleErrors (Just (HandleOopsie Closing desc)) + ev (liftIO $ hClose hand) + +-- | Terminates a process; sends any IO errors to the handler. +terminateProcess + :: (MonadCatch m, MonadIO m) + => Process.ProcessHandle + -> ErrSpec + -> m () +terminateProcess han ev = do + _ <- handleErrors Nothing ev (liftIO (Process.terminateProcess han)) + handleErrors Nothing ev . liftIO $ do + _ <- Process.waitForProcess han + return () + + +-- | Acquires a resource and ensures it will be destroyed when the +-- 'MonadSafe' computation completes. +acquire + :: MonadSafe m + => Base m a + -- ^ Acquirer. + -> (a -> Base m ()) + -- ^ Destroyer. + -> m a +acquire acq rel = mask $ \restore -> do + a <- liftBase acq + _ <- register (rel a) + restore $ return a + + +-- * Threads + +-- | Runs a thread in the background. The thread is terminated when +-- the 'MonadSafe' computation completes. +background + :: MonadSafe m + => IO a + -> m (Async a) +background act = acquire (liftIO $ async act) (liftIO . cancel) + +-- | Runs in the background an effect, typically one that is moving +-- data from one process to another. For examples of its usage, see +-- "Pipes.Cliff.Examples". The associated thread is killed when the +-- 'MonadSafe' computation completes. +conveyor :: MonadSafe m => Effect (SafeT IO) () -> m () +conveyor efct + = (background . liftIO . runSafeT . runEffect $ efct) >> return () + + +-- | A version of 'Control.Concurrent.Async.wait' with an overloaded +-- 'MonadIO' return type. Allows you to wait for the return value of +-- threads launched with 'background'. If the thread throws an +-- exception, 'waitForThread' will throw that same exception. +waitForThread :: MonadIO m => Async a -> m a +waitForThread = liftIO . wait + +-- | An overloaded version of the 'Process.waitForProcess' from +-- "System.Process". +waitForProcess :: MonadIO m => ProcessHandle -> m ExitCode +waitForProcess h = liftIO $ Process.waitForProcess h + + +-- * Mailboxes + +-- | A buffer that holds 1 message. I have no idea if this is the +-- ideal size. Don't use an unbounded buffer, though, because with +-- unbounded producers an unbounded buffer will fill up your RAM. +-- +-- Since the buffer just holds one size, you might think \"why not +-- just use an MVar\"? At least, I have been silly enough to think +-- that. Using @Pipes.Concurrent@ also give the mailbox the ability +-- to be sealed; sealing the mailbox signals to the other side that it +-- won't be getting any more input or be allowed to send any more +-- output, which tells the whole pipeline to start shutting down. +messageBuffer :: PC.Buffer a +messageBuffer = PC.bounded 1 + +-- | Creates a new mailbox and returns 'Proxy' that stream values +-- into and out of the mailbox. Each 'Proxy' is equipped with a +-- finalizer that will seal the mailbox immediately after production +-- or consumption has completed, even if such completion is not due +-- to an exhausted mailbox. This will signal to the other side of +-- the mailbox that the mailbox is sealed. +newMailbox + :: (MonadIO m, MonadSafe mi, MonadSafe mo) + => m (Consumer a mi (), Producer a mo ()) +newMailbox = do + (toBox, fromBox, seal) <- liftIO $ PC.spawn' messageBuffer + let csmr = register (liftIO $ PC.atomically seal) + >> PC.toOutput toBox + pdcr = register (liftIO $ PC.atomically seal) + >> PC.fromInput fromBox + return (csmr, pdcr) + + +-- * Production from and consumption to 'Handle's + +-- | I have no idea what this should be. I'll start with a simple +-- small value and see how it works. +bufSize :: Int +bufSize = 1024 + + +-- | Create a 'Producer' that produces from a 'Handle'. Takes +-- ownership of the 'Handle'; closes it when the 'Producer' +-- terminates. If any IO errors arise either during production or +-- when the 'Handle' is closed, they are caught and passed to the +-- handler. +produceFromHandle + :: (MonadSafe m, MonadCatch (Base m)) + => HandleDesc + -> Handle + -> ErrSpec + -> Producer ByteString m () +produceFromHandle hDesc h ev = do + _ <- register (closeHandleNoThrow h hDesc ev) + let hndlr e = lift $ handleException (Just oops) e ev + oops = HandleOopsie Reading hDesc + produce = liftIO (BS.hGetSome h bufSize) >>= go + go bs + | BS.null bs = return () + | otherwise = yield bs >> produce + produce `catch` hndlr + + +-- | Runs a 'Consumer'; registers the handle so that it is closed +-- when consumption finishes. If any IO errors arise either during +-- consumption or when the 'Handle' is closed, they are caught and +-- passed to the handler. +consumeToHandle + :: (MonadSafe m, MonadCatch (Base m)) + => Handle + -> ErrSpec + -> Consumer ByteString m () +consumeToHandle h ev = do + _ <- register $ closeHandleNoThrow h Input ev + let hndlr e = lift $ handleException (Just oops) e ev + oops = HandleOopsie Writing Input + go = do + bs <- await + liftIO $ BS.hPut h bs + go + go `catch` hndlr + + +-- | Creates a background thread that will consume to the given Handle +-- from the given Producer. Takes ownership of the 'Handle' and +-- closes it when done. +backgroundSendToProcess + :: MonadSafe m + => Handle + -> Producer ByteString (SafeT IO) () + -> ErrSpec + -> m () +backgroundSendToProcess han prod ev = background act >> return () + where + csmr = consumeToHandle han ev + act = runSafeT . runEffect $ prod >-> csmr + +-- | Creates a background thread that will produce from the given +-- Handle into the given Consumer. Takes possession of the Handle and +-- closes it when done. +backgroundReceiveFromProcess + :: MonadSafe m + => HandleDesc + -> Handle + -> Consumer ByteString (SafeT IO) () + -> ErrSpec + -> m () +backgroundReceiveFromProcess desc han csmr ev = background act >> return () + where + prod = produceFromHandle desc han ev + act = runSafeT . runEffect $ prod >-> csmr + +-- | Does everything necessary to run a 'Handle' that is created to a +-- process standard input. Creates mailbox, runs background thread +-- that pumps data out of the mailbox and into the process standard +-- input, and returns a Consumer that consumes and places what it +-- consumes into the mailbox for delivery to the background process. +runInputHandle + :: (MonadSafe m, MonadSafe mi) + => Handle + -> ErrSpec + -> m (Consumer ByteString mi ()) +runInputHandle inp ev = do + (toMbox, fromMbox) <- liftIO newMailbox + backgroundSendToProcess inp fromMbox ev + return toMbox + + +-- | Does everything necessary to run a 'Handle' that is created to a +-- process standard output or standard error. Creates mailbox, runs +-- background thread that pumps data from the process output 'Handle' +-- into the mailbox, and returns a Producer that produces what comes +-- into the mailbox. +runOutputHandle + :: (MonadSafe m, MonadSafe mo) + => HandleDesc + -> Handle + -> ErrSpec + -> m (Producer ByteString mo ()) +runOutputHandle desc out ev = do + (toMbox, fromMbox) <- liftIO $ newMailbox + backgroundReceiveFromProcess desc out toMbox ev + return fromMbox + + +-- * Creating subprocesses + + +-- | Creates a subprocess. Registers destroyers for each handle +-- created, as well as for the ProcessHandle. +createProcess + :: (MonadSafe m, MonadCatch (Base m)) + => Process.CreateProcess + -> ErrSpec + -> m (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) +createProcess cp ev = mask $ \restore -> do + (mayIn, mayOut, mayErr, han) <- liftIO $ Process.createProcess cp + let close mayHan desc = maybe (return ()) + (\h -> closeHandleNoThrow h desc ev) mayHan + _ <- register (close mayIn Input) + _ <- register (close mayOut Output) + _ <- register (close mayErr Error) + _ <- register (terminateProcess han ev) + restore $ return (mayIn, mayOut, mayErr, han) + + +-- | Convenience wrapper for 'createProcess'. The subprocess is +-- terminated and all its handles destroyed when the 'MonadSafe' +-- computation completes. +runCreateProcess + :: (MonadSafe m, MonadCatch (Base m)) + => Maybe NonPipe + -- ^ Standard input + -> Maybe NonPipe + -- ^ Standard output + -> Maybe NonPipe + -- ^ Standard error + -> CreateProcess + -> m (Maybe Handle, Maybe Handle, Maybe Handle, ErrSpec, ProcessHandle) +runCreateProcess inp out err cp = do + let ev = makeErrSpec cp + (inp', out', err', phan) <- + createProcess (convertCreateProcess inp out err cp) ev + return (inp', out', err', ev, phan) + +-- * Creating Proxy + + +-- | Do not create any 'Proxy' to or from the process. +pipeNone + :: (MonadSafe m, MonadCatch (Base m)) + => NonPipe + -- ^ Standard input + -> NonPipe + -- ^ Standard output + -> NonPipe + -- ^ Standard error + -> CreateProcess + -> m ProcessHandle +pipeNone inp out err cp = do + (_, _, _, _, phan) <- + runCreateProcess (Just inp) (Just out) (Just err) cp + return phan + + +-- | Create a 'Consumer' for standard input. +pipeInput + :: (MonadSafe mi, MonadSafe m, MonadCatch (Base m)) + => NonPipe + -- ^ Standard output + -> NonPipe + -- ^ Standard error + -> CreateProcess + -> m (Consumer ByteString mi (), ProcessHandle) + -- ^ A 'Consumer' for standard input +pipeInput out err cp = do + (Just inp, _, _, ev, phan) <- + runCreateProcess Nothing (Just out) (Just err) cp + ih <- runInputHandle inp ev + return (ih, phan) + + + +-- | Create a 'Producer' for standard output. +pipeOutput + :: (MonadSafe mo, MonadSafe m, MonadCatch (Base m)) + => NonPipe + -- ^ Standard input + -> NonPipe + -- ^ Standard error + -> CreateProcess + -> m (Producer ByteString mo (), ProcessHandle) + -- ^ A 'Producer' for standard output +pipeOutput inp err cp = do + (_, Just out, _, ev, phan) <- runCreateProcess (Just inp) + Nothing (Just err) cp + oh <- runOutputHandle Output out ev + return (oh, phan) + +-- | Create a 'Producer' for standard error. +pipeError + :: (MonadSafe m, MonadCatch (Base m)) + => NonPipe + -- ^ Standard input + -> NonPipe + -- ^ Standard output + -> CreateProcess + -> m (Producer ByteString m (), ProcessHandle) + -- ^ A 'Producer' for standard error +pipeError inp out cp = do + (_, _, Just err, ev, phan) <- runCreateProcess (Just inp) (Just out) Nothing cp + eh <- runOutputHandle Error err ev + return (eh, phan) + +-- | Create a 'Consumer' for standard input and a 'Producer' for +-- standard output. +pipeInputOutput + :: (MonadSafe mi, MonadSafe mo, MonadSafe m, MonadCatch (Base m)) + => NonPipe + -- ^ Standard error + -> CreateProcess + -> m ((Consumer ByteString mi (), Producer ByteString mo ()), ProcessHandle) + -- ^ A 'Consumer' for standard input, a 'Producer' for standard + -- output +pipeInputOutput err cp = do + (Just inp, Just out, _, ev, phan) <- + runCreateProcess Nothing Nothing (Just err) cp + ih <- runInputHandle inp ev + oh <- runOutputHandle Output out ev + return ((ih, oh), phan) + +-- | Create a 'Consumer' for standard input and a 'Producer' for +-- standard error. +pipeInputError + :: (MonadSafe mi, MonadSafe me, MonadSafe m, MonadCatch (Base m)) + => NonPipe + -- ^ Standard output + -> CreateProcess + -> m ( (Consumer ByteString mi (), Producer ByteString me ()) + , ProcessHandle) + -- ^ A 'Consumer' for standard input, a 'Producer' for standard + -- error +pipeInputError out cp = do + (Just inp, _, Just err, ev, phan) <- + runCreateProcess Nothing (Just out) Nothing cp + ih <- runInputHandle inp ev + eh <- runOutputHandle Error err ev + return $ ((ih, eh), phan) + +-- | Create a 'Producer' for standard output and a 'Producer' for +-- standard error. +pipeOutputError + :: (MonadSafe mo, MonadSafe me, MonadSafe m, MonadCatch (Base m)) + => NonPipe + -- ^ Standard input + -> CreateProcess + -> m ((Producer ByteString mo (), Producer ByteString me ()), ProcessHandle) + -- ^ A 'Producer' for standard output, a 'Producer' for standard + -- error +pipeOutputError inp cp = do + (_, Just out, Just err, ev, phan) <- + runCreateProcess (Just inp) Nothing Nothing cp + oh <- runOutputHandle Output out ev + eh <- runOutputHandle Error err ev + return ((oh, eh), phan) + + +-- | Create a 'Consumer' for standard input, a 'Producer' for standard +-- output, and a 'Producer' for standard error. +pipeInputOutputError + :: ( MonadSafe mi, MonadSafe mo, MonadSafe me, + MonadSafe m, MonadCatch (Base m)) + => CreateProcess + -> m (( Consumer ByteString mi () + , Producer ByteString mo () + , Producer ByteString me ()), ProcessHandle) + -- ^ A 'Consumer' for standard input, a 'Producer' for standard + -- output, a 'Producer' for standard error +pipeInputOutputError cp = do + (Just inp, Just out, Just err, ev, phan) <- + runCreateProcess Nothing Nothing Nothing cp + ih <- runInputHandle inp ev + oh <- runOutputHandle Output out ev + eh <- runOutputHandle Error err ev + return $ ((ih, oh, eh), phan) diff --git a/lib/Pipes/Cliff/Examples.hs b/lib/Pipes/Cliff/Examples.hs --- a/lib/Pipes/Cliff/Examples.hs +++ b/lib/Pipes/Cliff/Examples.hs @@ -7,9 +7,15 @@ -- -- -- --- __Be sure to use the @-threaded@ option__ when compiling code that --- uses "Pipes.Cliff", including this code; see the warning in +-- __Be sure to use the @-threaded@ option__ when compiling code +-- that uses "Pipes.Cliff", including this code; see the warning in -- "Pipes.Cliff" for more details. +-- +-- Notice throughout how pipelines that move data from one process +-- to another typically are run in the background using 'conveyor', +-- and that threads that produce information you need to use are run +-- in the 'background' so you can use 'waitForThread' to retrieve +-- their results. module Pipes.Cliff.Examples where @@ -24,7 +30,8 @@ -- can use "Pipes.Cliff" even for non-finite 'Producer's. Don't try -- to go to the end of the input in @less@, though. When you quit -- @less@, you will get broken pipe warnings printed to standard --- error. This is normal. To suppress them, see the 'quiet' option. +-- error. This is normal. To suppress them, see the 'handler' +-- option. numsToLess :: IO ExitCode numsToLess = runSafeT $ do @@ -32,32 +39,22 @@ conveyor $ produceNumbers >-> toLess waitForProcess han + -- | Streams an infinite list of numbers to @tr@ and then to @less@. --- Perfectly useless, but shows how to build pipelines. Notice how --- the components of the pipeline are run in the background using --- 'conveyor'; if you run one of them in the foreground, you might get --- a deadlock. +-- Perfectly useless, but shows how to build pipelines. Also +-- squlches warning messages using the 'handler' option. alphaNumbers :: IO ExitCode alphaNumbers = runSafeT $ do - (toTr, fromTr, _) <- pipeInputOutput Inherit - (procSpec "tr" ["[0-9]", "[a-z]"]) + ((toTr, fromTr), _) <- pipeInputOutput Inherit + (procSpec "tr" ["[0-9]", "[a-z]"]) { handler = squelch } (toLess, lessHan) <- pipeInput Inherit Inherit - (procSpec "less" []) + (procSpec "less" []) { handler = squelch } conveyor $ produceNumbers >-> toTr conveyor $ fromTr >-> toLess waitForProcess lessHan --- | Produces a stream of 'BS8.ByteString', where each --- 'BS8.ByteString' is a shown integer. - -produceNumbers :: Monad m => Producer BS8.ByteString m () -produceNumbers = each . fmap mkNumStr $ [(0 :: Int) ..] - where - mkNumStr = flip BS8.snoc '\n' . BS8.pack . show - - -- | Produces an infinite stream of numbers, sends it to @tr@ for some -- mangling, and then to @sh@, which will copy each line both to -- standard output and to standard error. From @sh@, standard output @@ -74,29 +71,30 @@ -- output the user actually viewed in @less@. standardOutputAndError :: IO BS8.ByteString standardOutputAndError = runSafeT $ do - (toTr, fromTr, _) <- pipeInputOutput Inherit + ((toTr, fromTr), _) <- pipeInputOutput Inherit (procSpec "tr" ["[0-9]", "[a-z]"]) - (toSh, fromShOut, fromShErr, _) <- pipeInputOutputError + ((toSh, fromShOut, fromShErr), _) <- pipeInputOutputError (procSpec "sh" ["-c", script]) - (toLess, _) <- pipeInput Inherit Inherit - (procSpec "less" []) + (toLess, lessHan) <- pipeInput Inherit Inherit (procSpec "less" []) conveyor $ produceNumbers >-> toTr conveyor $ fromTr >-> toSh conveyor $ fromShOut >-> toLess - thread <- background + foldHan <- + background . runSafeT - . P.fold BS8.append BS8.empty id - $ fromShErr - waitForThread thread + $ P.fold BS8.append BS8.empty id fromShErr + _ <- waitForProcess lessHan + waitForThread foldHan where script = "while read line; do echo $line; echo $line 1>&2; done" -- | Like 'alphaNumbers' but just sends a limited number -- of numbers to @cat@. A useful test to make sure that pipelines --- shut down automatically. +-- shut down automatically. Runs both pipelines in the background and +-- uses 'waitForProcess' to wait until @cat@ is done. limitedAlphaNumbers :: IO ExitCode limitedAlphaNumbers = runSafeT $ do - (toTr, fromTr, _) <- pipeInputOutput Inherit + ((toTr, fromTr), _) <- pipeInputOutput Inherit (procSpec "tr" ["[0-9]", "[a-z]"]) (toCat, catHan) <- pipeInput Inherit Inherit (procSpec "cat" []) @@ -106,25 +104,27 @@ -- | Produces a finite list of numbers, sends it to @tr@ for some -- mangling, and then puts the results into a 'BS8.ByteString' for --- further processing. Unlike previous examples, there is no use of --- 'waitForProcess' or 'conveyor'. This is OK because the 'Effect' --- that retrieves the results from @tr@ will pull all the data; the --- @tr@ process will then shut down because its standard input will be --- closed when the source 'Producer' is exhausted. This example shows --- how you can use this library to place the results of a pipeline --- into a simple strict data type. --- --- When the 'SafeT' computation completes, a --- 'System.Process.terminateProcess' is automatically sent to the @tr@ --- process--which does nothing, as @tr@ has already died. Only after --- the process is waited for is it fully removed from the system --- process table. A 'System.Process.waitForProcess' from --- "System.Process" is automatically done as well. Therefore, you --- will not get zombie processes if you use this library. +-- further processing. This example shows how you can use this +-- library to place the results of a pipeline into a simple strict +-- data type. alphaNumbersByteString :: IO BS8.ByteString alphaNumbersByteString = runSafeT $ do - (toTr, fromTr, _) <- pipeInputOutput Inherit + ((toTr, fromTr), _) <- pipeInputOutput Inherit (procSpec "tr" ["[0-9]", "[a-z]"]) conveyor $ produceNumbers >-> P.take 300 >-> toTr - P.fold BS8.append BS8.empty id fromTr + threadHan <- + background + . runSafeT + $ P.fold BS8.append BS8.empty id fromTr + waitForThread threadHan + + +-- | Produces a stream of 'BS8.ByteString', where each +-- 'BS8.ByteString' is a shown integer. + +produceNumbers :: Monad m => Producer BS8.ByteString m () +produceNumbers = each . fmap mkNumStr $ [(0 :: Int) ..] + where + mkNumStr = flip BS8.snoc '\n' . BS8.pack . show + diff --git a/pipes-cliff.cabal b/pipes-cliff.cabal --- a/pipes-cliff.cabal +++ b/pipes-cliff.cabal @@ -3,11 +3,11 @@ -- http://www.github.com/massysett/cartel -- -- Script name used to generate: genCabal.hs --- Generated on: 2015-03-22 13:30:11.54764 EDT +-- Generated on: 2015-03-24 13:45:58.663565 EDT -- Cartel library version: 0.14.2.0 name: pipes-cliff -version: 0.4.0.0 +version: 0.6.0.0 cabal-version: >= 1.16 license: BSD3 license-file: LICENSE @@ -45,6 +45,7 @@ Library exposed-modules: Pipes.Cliff + Pipes.Cliff.Core Pipes.Cliff.Examples hs-source-dirs: lib @@ -55,7 +56,7 @@ , bytestring >= 0.10.4 && < 0.11 , process >= 1.2.0.0 && < 1.3 , async >= 2.0 && < 2.1 - , pipes-concurrency >= 2.0 && < 2.1 + , pipes-concurrency >= 2.0.3 && < 2.1 default-language: Haskell2010 ghc-options: -Wall @@ -76,6 +77,7 @@ tests other-modules: Pipes.Cliff + Pipes.Cliff.Core Pipes.Cliff.Examples build-depends: base >= 4.6.0.0 && < 4.8 @@ -84,7 +86,7 @@ , bytestring >= 0.10.4 && < 0.11 , process >= 1.2.0.0 && < 1.3 , async >= 2.0 && < 2.1 - , pipes-concurrency >= 2.0 && < 2.1 + , pipes-concurrency >= 2.0.3 && < 2.1 ghc-options: -threaded else @@ -102,6 +104,7 @@ tests other-modules: Pipes.Cliff + Pipes.Cliff.Core Pipes.Cliff.Examples build-depends: base >= 4.6.0.0 && < 4.8 @@ -110,7 +113,7 @@ , bytestring >= 0.10.4 && < 0.11 , process >= 1.2.0.0 && < 1.3 , async >= 2.0 && < 2.1 - , pipes-concurrency >= 2.0 && < 2.1 + , pipes-concurrency >= 2.0.3 && < 2.1 ghc-options: -threaded else @@ -128,6 +131,7 @@ tests other-modules: Pipes.Cliff + Pipes.Cliff.Core Pipes.Cliff.Examples build-depends: base >= 4.6.0.0 && < 4.8 @@ -136,7 +140,7 @@ , bytestring >= 0.10.4 && < 0.11 , process >= 1.2.0.0 && < 1.3 , async >= 2.0 && < 2.1 - , pipes-concurrency >= 2.0 && < 2.1 + , pipes-concurrency >= 2.0.3 && < 2.1 ghc-options: -threaded else @@ -154,6 +158,7 @@ tests other-modules: Pipes.Cliff + Pipes.Cliff.Core Pipes.Cliff.Examples build-depends: base >= 4.6.0.0 && < 4.8 @@ -162,7 +167,7 @@ , bytestring >= 0.10.4 && < 0.11 , process >= 1.2.0.0 && < 1.3 , async >= 2.0 && < 2.1 - , pipes-concurrency >= 2.0 && < 2.1 + , pipes-concurrency >= 2.0.3 && < 2.1 ghc-options: -threaded else @@ -180,6 +185,7 @@ tests other-modules: Pipes.Cliff + Pipes.Cliff.Core Pipes.Cliff.Examples build-depends: base >= 4.6.0.0 && < 4.8 @@ -188,7 +194,7 @@ , bytestring >= 0.10.4 && < 0.11 , process >= 1.2.0.0 && < 1.3 , async >= 2.0 && < 2.1 - , pipes-concurrency >= 2.0 && < 2.1 + , pipes-concurrency >= 2.0.3 && < 2.1 ghc-options: -threaded else