diff --git a/Benchmark/System/Process.hs b/Benchmark/System/Process.hs new file mode 100644 --- /dev/null +++ b/Benchmark/System/Process.hs @@ -0,0 +1,240 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +module Main (main) where + +import Control.Exception (finally) +import Data.Either (isRight, fromRight, isLeft, fromLeft) +import Data.Word (Word8) +import Gauge (defaultMain, bench, nfIO) +import System.Directory (removeFile, findExecutable) +import System.IO + ( Handle + , IOMode(..) + , openFile + , hClose + ) +import System.Process (proc, createProcess, waitForProcess, callCommand) + +import qualified Streamly.Data.Fold as FL +import qualified Streamly.FileSystem.Handle as FH +import qualified Streamly.Prelude as S +import qualified Streamly.System.Process as Proc + +-- Internal imports +import qualified Streamly.Internal.FileSystem.Handle + as FH (toBytes, toChunks, putBytes, putChunks) +import qualified Streamly.Internal.System.Process as Proc + +-- XXX replace with streamly versions once they are fixed +{-# INLINE rights #-} +rights :: (S.IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b +rights = fmap (fromRight undefined) . S.filter isRight + +{-# INLINE lefts #-} +lefts :: (S.IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a +lefts = fmap (fromLeft undefined) . S.filter isLeft + +------------------------------------------------------------------------------- +-- Constants and utils +------------------------------------------------------------------------------- + +_a :: Word8 +_a = 97 + +-- XXX portability on macOS +devRandom :: String +devRandom = "/dev/urandom" + +devNull :: String +devNull = "/dev/null" + +which :: String -> IO FilePath +which cmd = do + r <- findExecutable cmd + case r of + Just path -> return path + _ -> error $ "Required command " ++ cmd ++ " not found" + +------------------------------------------------------------------------------- +-- Create a data file filled with random data +------------------------------------------------------------------------------- + +ddBlockSize :: Int +ddBlockSize = 1024 * 1024 + +ddBlockCount :: Int +ddBlockCount = 10 + +largeByteFile :: String +largeByteFile = "./largeByteFile" + +generateByteFile :: IO () +generateByteFile = do + ddPath <- which "dd" + let procObj = proc ddPath [ + "if=" ++ devRandom, + "of=" ++ largeByteFile, + "count=" ++ show ddBlockCount, + "bs=" ++ show ddBlockSize + ] + + (_, _, _, procHandle) <- createProcess procObj + _ <- waitForProcess procHandle + return () + +------------------------------------------------------------------------------- +-- Create a file filled with ascii chars +------------------------------------------------------------------------------- + +largeCharFile :: String +largeCharFile = "./largeCharFile" + +numCharInCharFile :: Int +numCharInCharFile = 10 * 1024 * 1024 + +generateCharFile :: IO () +generateCharFile = do + handle <- openFile largeCharFile WriteMode + FH.putBytes handle (S.replicate numCharInCharFile _a) + hClose handle + +------------------------------------------------------------------------------- +-- Create a utility that writes to stderr +------------------------------------------------------------------------------- + +trToStderr :: String +trToStderr = "./writeTrToError.sh" + +trToStderrContent :: String +trToStderrContent = + "tr [a-z] [A-Z] <&0 >&2" + +createExecutable :: IO () +createExecutable = do + writeFile trToStderr trToStderrContent + callCommand ("chmod +x " ++ trToStderr) + +------------------------------------------------------------------------------- +-- Create and delete the temp data/exec files +------------------------------------------------------------------------------- + +generateFiles :: IO () +generateFiles = do + createExecutable + generateByteFile + generateCharFile + +deleteFiles :: IO () +deleteFiles = do + removeFile trToStderr + removeFile largeByteFile + removeFile largeCharFile + +------------------------------------------------------------------------------- +-- Benchmark functions +------------------------------------------------------------------------------- + +toBytes' :: String-> Handle -> IO () +toBytes' catPath outH = + FH.putBytes outH + $ rights + $ Proc.toBytes' catPath [largeByteFile] + +toChunks' :: String -> Handle -> IO () +toChunks' catPath hdl = + FH.putChunks hdl + $ rights + $ Proc.toChunks' catPath [largeByteFile] + +processBytes' :: String-> Handle -> IO () +processBytes' trPath outputHdl = do + inputHdl <- openFile largeCharFile ReadMode + _ <- S.fold (FL.partition (FH.write outputHdl) (FH.write outputHdl)) + $ Proc.processBytes' + trPath + ["[a-z]", "[A-Z]"] + $ FH.toBytes inputHdl + hClose inputHdl + +processBytes :: String-> Handle -> IO () +processBytes trPath outputHdl = do + inputHdl <- openFile largeCharFile ReadMode + FH.putBytes outputHdl + $ Proc.processBytes + trPath + ["[a-z]", "[A-Z]"] + $ FH.toBytes inputHdl + hClose inputHdl + +processBytesToStderr :: Handle -> IO () +processBytesToStderr outputHdl = do + inputHdl <- openFile largeCharFile ReadMode + FH.putBytes outputHdl + $ lefts + $ Proc.processBytes' + trToStderr + ["[a-z]", "[A-Z]"] + $ FH.toBytes inputHdl + hClose inputHdl + +processChunks :: String -> Handle -> IO () +processChunks trPath outputHdl = do + inputHdl <- openFile largeCharFile ReadMode + FH.putChunks outputHdl $ + Proc.processChunks + trPath + ["[a-z]", "[A-Z]"] + $ FH.toChunks inputHdl + hClose inputHdl + +processChunks' :: String -> Handle -> IO () +processChunks' trPath outputHdl = do + inputHdl <- openFile largeCharFile ReadMode + _ <- S.fold + (FL.partition + (FH.writeChunks outputHdl) (FH.writeChunks outputHdl) + ) + $ Proc.processChunks' + trPath + ["[a-z]", "[A-Z]"] + (FH.toChunks inputHdl) + hClose inputHdl + +processChunksToStderr :: Handle -> IO () +processChunksToStderr outputHdl = do + inputHdl <- openFile largeCharFile ReadMode + FH.putChunks outputHdl + $ lefts + $ Proc.processChunks' + trToStderr + ["[a-z]", "[A-Z]"] + (FH.toChunks inputHdl) + hClose inputHdl + +------------------------------------------------------------------------------- +-- Main +------------------------------------------------------------------------------- + +main :: IO () +main = do + putStrLn "Generating files..." + generateFiles + trPath <- which "tr" + catPath <- which "cat" + nullH <- openFile devNull WriteMode + putStrLn "Running benchmarks..." + + defaultMain + [ bench "toBytes'" $ nfIO $ toBytes' catPath nullH + , bench "toChunks'" $ nfIO $ toChunks' catPath nullH + , bench "processBytes tr" $ nfIO $ processBytes trPath nullH + , bench "processBytes' tr" $ nfIO $ processBytes' trPath nullH + , bench "processBytesToStderr tr" $ nfIO $ processBytesToStderr nullH + , bench "processChunks tr" $ nfIO (processChunks trPath nullH) + , bench "processChunks' tr" $ nfIO (processChunks' trPath nullH) + , bench "processChunksToStderr" $ nfIO $ processChunksToStderr nullH + ] `finally` (do + putStrLn "cleanup ..." + hClose nullH + deleteFiles + ) diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Changelog + +## 0.1.0 (Jul 2021) + +* Initial version. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,177 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS diff --git a/NOTICE b/NOTICE new file mode 100644 --- /dev/null +++ b/NOTICE @@ -0,0 +1,5 @@ +streamly-process +Copyright 2020 Composewell Technologies + +This product includes software developed at +Composewell Technologies (http://www.composewell.com). diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,12 @@ +# OS processes as streams + +Run operating system processes as stream source, sink or transformation +functions. Use them seamlessly in a streaming data pipeline in the same way +as any other Haskell functions. + +Please visit [Streamly homepage](https://streamly.composewell.com) for more +details about streamly. + +# Licensing + +Available under [Apache-2.0 license](LICENSE). diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/design/proposal.md b/design/proposal.md new file mode 100644 --- /dev/null +++ b/design/proposal.md @@ -0,0 +1,61 @@ +# Streaming System Processes + +Use streamly to execute system commands and connect the input, output and +error streams to create processing pipelines. + +## Scope of the Project + +A system process created by executing a command or binary executable can be +modeled as a stream transformation whose standard input is a stream and +standard output is another stream. The overall idea is to provide +functionality close to how shell executes processes and connects them via +pipes. The interface has to be as simple as possible. + +_Exception Handling:_ The standard error is yet another stream that needs to be +handled by an exception handling mechanism. We can also implements equivalents +of `pipefail` and `!` to apply a default exception handling policy on a +pipeline. Something like `set -e` would be default in a streamly streams +composition, however, we can have an option to achieve the opposite to ignore +exceptions on a portion of a pipeline. + +_Signal handling:_ Implement the equivalent of the `trap` shell builtin. + +_Redirections:_ Implement input and output redirections like the shell. We can +use operators for example, `|>`, `<|` and `|>>` etc. to represent redirections. +Standard error can also be merged with standard output, something like the +`2>&1`, `|&` idioms of shell. Redirections using here documents (`|<<`) and +here string (`|<<<`) can also be supported. Duplicating and moving file +descriptors can also be supported. + +_Background Jobs:_ Functionality equivalent to the `&` operator and `wait` +command in shell. This should be trivially equivalent to a `Sink` in streamly. +Investigate if there is anything extra required to implement the bash +coprocesses like functionality. + +_Network Pipes:_ In addition to stdin and stdout, processes can also +communicate via sockets. We can implement ways to connect input/output streams +to sockets owned by the processes as well. `socat` and `netcat` utilities are +examples of such functionality. + +## References + +* See the [`bash` man page](https://linux.die.net/man/1/bash) +* https://hackage.haskell.org/package/process +* https://hackage.haskell.org/package/typed-process +* http://hackage.haskell.org/package/turtle + +## Prerequisites + +Working knowledge of the Haskell programming language is needed, advanced +knowledge is not required. If you have a good knowledge of basics and can work +with Haskell lists that should be enough. Familiarity with process management +related POSIX systems calls, signal handling and Haskell C FFI may be useful. + +## Difficulty Level + +Intermediate. + +## Mentors + +* Harendra Kumar +* Pranay Sashank diff --git a/src/Streamly/Internal/System/Process.hs b/src/Streamly/Internal/System/Process.hs new file mode 100644 --- /dev/null +++ b/src/Streamly/Internal/System/Process.hs @@ -0,0 +1,618 @@ +-- | +-- Module : Streamly.Internal.System.Process +-- Copyright : (c) 2020 Composewell Technologies +-- License : Apache-2.0 +-- Maintainer : streamly@composewell.com +-- Stability : experimental +-- Portability : GHC +-- + +{-# LANGUAGE CPP #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- TODO: +-- +-- Remove dependency on the "process" package. We do not need a lot of it or +-- need to reimplement significantly. +-- +-- Interactive processes: +-- +-- To understand process groups and sessions, see the "General Terminal +-- Interface" section in the [POSIX +-- standard](https://pubs.opengroup.org/onlinepubs/9699919799/). +-- +-- For running processes interactively we need to make a new process group for +-- the new process and make it the foreground process group. When the process +-- is done we can make the parent process group as the foreground process group +-- again. We need to ensure that this works properly under exceptions. We can +-- provide an "interact" function to do so. +-- +-- - Need a way to specify additional parameters for process creation. +-- Possibly use something like @processBytesWith spec@ etc. +-- +-- - Need a way to access the pid and manage the processes and process groups. +-- We can treat the processes in the same way as we treat threads. We can +-- compose processes in parallel, and cleanup can happen in the same way as +-- tids are cleaned up. But do we need this when we have threads anyway? +-- +-- - Use unfolds for generation? +-- +-- - Folds for composing process sinks? Input may be taken as input of the +-- fold and the output of the process can be consumed by another fold. +-- +-- - Replace FilePath with a typed path. +-- +{-# LANGUAGE FlexibleContexts #-} + +module Streamly.Internal.System.Process + ( + -- * Process Configuration + Config + + -- * Exceptions + , ProcessFailure (..) + + -- * Generation + , toBytes + , toBytes' + , toChunks + , toChunks' + + -- * Transformation + , processBytes + , processBytes' + , processChunksWith + , processChunks + , processChunks'With + , processChunks' + ) +where + +-- #define USE_NATIVE + +import Control.Monad.Catch (MonadCatch, throwM) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Word (Word8) +import Streamly.Data.Array.Foreign (Array) +import Streamly.Prelude (MonadAsync, parallel, IsStream, adapt) +import System.Exit (ExitCode(..)) +import System.IO (hClose, Handle) + +#ifdef USE_NATIVE +import Control.Exception (Exception(..), catch, throwIO, SomeException) +import System.Posix.Process (ProcessStatus(..)) +import Streamly.Internal.System.Process.Posix +#else +import Control.Concurrent (forkIO) +import Control.Exception (Exception(..), catch, throwIO) +import Control.Monad (void, unless) +import Foreign.C.Error (Errno(..), ePIPE) +import GHC.IO.Exception (IOException(..), IOErrorType(..)) +import System.Process + ( ProcessHandle + , CreateProcess(..) + , StdStream (..) + , createProcess + , waitForProcess + , CmdSpec(..) + , terminateProcess + ) +#endif + +import qualified Streamly.Data.Array.Foreign as Array +import qualified Streamly.Prelude as Stream + +-- Internal imports +-- XXX chunked IO chunk size should be moved to Streamly.System.IO +import Streamly.Internal.Data.Array.Foreign.Type (defaultChunkSize) +import Streamly.Internal.Data.Stream.StreamD.Step (Step (..)) +import Streamly.Internal.Data.Stream.StreamD.Type + (Stream (..), fromStreamD, toStreamD) +import Streamly.Internal.Data.SVar (adaptState) +import Streamly.Internal.Data.Unfold.Type (Unfold(..)) + +import qualified Streamly.Internal.Data.Array.Stream.Foreign + as ArrayStream (arraysOf) +import qualified Streamly.Internal.FileSystem.Handle + as Handle (toChunks, putChunks) +-- XXX To be exposed by streamly +-- import qualified Streamly.Internal.Data.Stream.IsStream as Stream (bracket') + +-- $setup +-- >>> :set -XFlexibleContexts +-- >>> import Data.Char (toUpper) +-- >>> import Data.Function ((&)) +-- >>> import qualified Streamly.Console.Stdio as Stdio +-- >>> import qualified Streamly.Data.Fold as Fold +-- >>> import qualified Streamly.Prelude as Stream +-- >>> import qualified Streamly.System.Process as Process +-- >>> import qualified Streamly.Unicode.Stream as Unicode +-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream + +-- XXX To be imported from streamly in future releases. + +data UnfoldManyEither o i f = + UnfoldManyEitherOuter o + | UnfoldManyEitherInner o i f + +{-# INLINE [1] unfoldManyEitherD #-} +unfoldManyEitherD :: Monad m => + Unfold m a b -> Stream m (Either a a) -> Stream m (Either b b) +unfoldManyEitherD (Unfold istep inject) (Stream ostep ost) = + Stream step (UnfoldManyEitherOuter ost) + where + {-# INLINE [0] step #-} + step gst (UnfoldManyEitherOuter o) = do + r <- ostep (adaptState gst) o + case r of + Yield (Left a) o' -> do + i <- inject a + i `seq` return (Skip (UnfoldManyEitherInner o' i Left)) + Yield (Right a) o' -> do + i <- inject a + i `seq` return (Skip (UnfoldManyEitherInner o' i Right)) + Skip o' -> return $ Skip (UnfoldManyEitherOuter o') + Stop -> return Stop + + step _ (UnfoldManyEitherInner o i f) = do + r <- istep i + return $ case r of + Yield x i' -> Yield (f x) (UnfoldManyEitherInner o i' f) + Skip i' -> Skip (UnfoldManyEitherInner o i' f) + Stop -> Skip (UnfoldManyEitherOuter o) + +{-# INLINE unfoldManyEither #-} +unfoldManyEither ::(IsStream t, Monad m) => + Unfold m a b -> t m (Either a a) -> t m (Either b b) +unfoldManyEither u m = fromStreamD $ unfoldManyEitherD u (toStreamD m) + +------------------------------------------------------------------------------- +-- Config +------------------------------------------------------------------------------- + +-- | Process configuration used for creating a new process. +-- +-- By default the process config is setup to inherit the following attributes +-- from the parent process: +-- +-- * Current working directory +-- * Environment +-- * Open file descriptors +-- * Process group +-- * Process uid and gid +-- * Signal handlers +-- * Terminal (Session) +-- +#ifdef USE_NATIVE + +type ProcessHandle = Process + +-- XXX After the fork specify what code to run in parent and in child before +-- exec. Also, use config to control whether to search the binary in the PATH +-- or not. +newtype Config = Config Bool + +mkConfig :: FilePath -> [String] -> Config +mkConfig _ _ = Config False + +pipeStdErr :: Config -> Config +pipeStdErr (Config _) = Config True +#else +newtype Config = Config CreateProcess + +-- | Create a default process configuration from an executable file path and +-- an argument list. +-- +mkConfig :: FilePath -> [String] -> Config +mkConfig path args = Config $ CreateProcess + { cmdspec = RawCommand path args + , cwd = Nothing -- inherit + , env = Nothing -- inherit + + -- File descriptors + , std_in = CreatePipe + , std_out = CreatePipe + , std_err = Inherit + , close_fds = False + + -- Session/group/setuid/setgid + , create_group = False + , child_user = Nothing -- Posix only + , child_group = Nothing -- Posix only + + -- Signals (Posix only) behavior + -- Reset SIGINT (Ctrl-C) and SIGQUIT (Ctrl-\) to default handlers. + , delegate_ctlc = False + + -- Terminal behavior + , new_session = False -- Posix only + , detach_console = False -- Windows only + , create_new_console = False -- Windows Only + + -- Added by commit 6b8ffe2ec3d115df9ccc047599545ca55c393005 + , use_process_jobs = True -- Windows only + } + +pipeStdErr :: Config -> Config +pipeStdErr (Config cfg) = Config $ cfg { std_err = CreatePipe } +#endif + +------------------------------------------------------------------------------- +-- Exceptions +------------------------------------------------------------------------------- +-- +-- TODO Add the path of the executable and the PID of the process to the +-- exception info to aid debugging. + +-- | An exception that is raised when a process fails. +-- +-- @since 0.1.0 +newtype ProcessFailure = ProcessFailure Int -- ^ The exit code of the process. + deriving Show + +-- Exception instance of ProcessFailure +instance Exception ProcessFailure where + + displayException (ProcessFailure exitCode) = + "Process failed with exit code: " ++ show exitCode + +------------------------------------------------------------------------------- +-- Transformation +------------------------------------------------------------------------------- +-- +-- | On normal cleanup we do not need to close the pipe handles as they are +-- already guaranteed to be closed (we can assert that) by the time we reach +-- here. We should not kill the process, rather wait for it to terminate +-- normally. +cleanupNormal :: MonadIO m => + (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> m () +cleanupNormal (_, _, _, procHandle) = liftIO $ do +#ifdef USE_NATIVE + -- liftIO $ putStrLn "cleanupNormal waiting" + status <- wait procHandle + -- liftIO $ putStrLn "cleanupNormal done" + case status of + Exited ExitSuccess -> return () + Exited (ExitFailure code) -> throwM $ ProcessFailure code + Terminated signal _ -> + throwM $ ProcessFailure (negate $ fromIntegral signal) + Stopped signal -> + throwM $ ProcessFailure (negate $ fromIntegral signal) +#else + exitCode <- waitForProcess procHandle + case exitCode of + ExitSuccess -> return () + ExitFailure code -> throwM $ ProcessFailure code + +-- | On an exception or if the process is getting garbage collected we need to +-- close the pipe handles, and send a SIGTERM to the process to clean it up. +-- Since we are using SIGTERM to kill the process, it may block forever. We can +-- possibly use a timer and send a SIGKILL after the timeout if the process is +-- still hanging around. +_cleanupException :: MonadIO m => + (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> m () +_cleanupException (Just stdinH, Just stdoutH, stderrMaybe, ph) = liftIO $ do + -- Send a SIGTERM to the process + terminateProcess ph + + -- Ideally we should be closing the handle without flushing the buffers so + -- that we cannot get a SIGPIPE. But there seems to be no way to do that as + -- of now so we just ignore the SIGPIPE. + hClose stdinH `catch` eatSIGPIPE + hClose stdoutH + whenJust hClose stderrMaybe + + -- Non-blocking wait for the process to go away + void $ forkIO (void $ waitForProcess ph) + + where + + whenJust action mb = maybe (pure ()) action mb + + isSIGPIPE e = + case e of + IOError + { ioe_type = ResourceVanished + , ioe_errno = Just ioe + } -> Errno ioe == ePIPE + _ -> False + + eatSIGPIPE e = unless (isSIGPIPE e) $ throwIO e +_cleanupException _ = error "cleanupProcess: Not reachable" +#endif + +-- | Creates a system process from an executable path and arguments. For the +-- default attributes used to create the process see 'mkConfig'. +-- +createProc' :: + (Config -> Config) -- ^ Process attribute modifier + -> FilePath -- ^ Executable path + -> [String] -- ^ Arguments + -> IO (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) + -- ^ (Input Handle, Output Handle, Error Handle, Process Handle) +createProc' modCfg path args = do +#ifdef USE_NATIVE + ((inp, out, err, _excParent, _excChild), parent, child, failure) <- + mkStdioPipes cfg + -- XXX Pass the exChild handle to the child process action + proc <- newProcess child path args Nothing + `catch` (\(e :: SomeException) -> failure >> throwIO e) + -- XXX Read the exception channel and reap the process if it failed before + -- exec. + parent + return (Just inp, Just out, err, proc) +#else + createProcess cfg +#endif + + where + + Config cfg = modCfg $ mkConfig path args + +{-# INLINE putChunksClose #-} +putChunksClose :: (MonadIO m, IsStream t) => + Handle -> t m (Array Word8) -> t m a +putChunksClose h input = + Stream.before + (Handle.putChunks h (adapt input) >> liftIO (hClose h)) + Stream.nil + +{-# INLINE toChunksClose #-} +toChunksClose :: (IsStream t, MonadAsync m) => Handle -> t m (Array Word8) +toChunksClose h = Stream.after (liftIO $ hClose h) (Handle.toChunks h) + +{-# INLINE processChunksWithAction #-} +processChunksWithAction :: + (IsStream t, MonadCatch m, MonadAsync m) + => ((Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> t m a) + -> (Config -> Config) + -> FilePath -- ^ Path to Executable + -> [String] -- ^ Arguments + -> t m a -- ^ Output stream +processChunksWithAction run modCfg path args = + -- Stream.bracket' + -- alloc cleanupNormal _cleanupException _cleanupException run + Stream.bracket alloc cleanupNormal run + + where + + alloc = liftIO $ createProc' modCfg path args + +{-# INLINE processChunks'With #-} +processChunks'With :: + (IsStream t, MonadCatch m, MonadAsync m) + => (Config -> Config) -- ^ Config modifier + -> FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m (Array Word8) -- ^ Input stream + -> t m (Either (Array Word8) (Array Word8)) -- ^ Output stream +processChunks'With modifier path args input = + processChunksWithAction run (modifier . pipeStdErr) path args + + where + + run (Just stdinH, Just stdoutH, Just stderrH, _) = + putChunksClose stdinH input + `parallel` Stream.map Left (toChunksClose stderrH) + `parallel` Stream.map Right (toChunksClose stdoutH) + run _ = error "processChunks': Not reachable" + +{-# INLINE processChunks' #-} +processChunks' :: + (IsStream t, MonadCatch m, MonadAsync m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m (Array Word8) -- ^ Input stream + -> t m (Either (Array Word8) (Array Word8)) -- ^ Output stream +processChunks' = processChunks'With id + +-- | @processBytes' path args input@ runs the executable at @path@ using @args@ +-- as arguments and @input@ stream as its standard input. The error stream of +-- the executable is presented as 'Left' values in the resulting stream and +-- output stream as 'Right' values. +-- +-- Raises 'ProcessFailure' exception in case of failure. +-- +-- For example, the following is equivalent to the shell command @echo "hello +-- world" | tr [:lower:] [:upper:]@: +-- +-- >>> :{ +-- processBytes' "echo" ["hello world"] Stream.nil +-- & Stream.rights +-- & processBytes' "tr" ["[:lower:]", "[:upper:]"] +-- & Stream.rights +-- & Stream.fold Stdio.write +-- :} +--HELLO WORLD +-- +-- @since 0.1.0 +{-# INLINE processBytes' #-} +processBytes' :: + (IsStream t, MonadCatch m, MonadAsync m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m Word8 -- ^ Input Stream + -> t m (Either Word8 Word8) -- ^ Output Stream +processBytes' path args input = + let input1 = ArrayStream.arraysOf defaultChunkSize input + output = processChunks' path args input1 + in unfoldManyEither Array.read output + +{-# INLINE processChunksWith #-} +processChunksWith :: + (IsStream t, MonadCatch m, MonadAsync m) + => (Config -> Config) -- ^ Config modifier + -> FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m (Array Word8) -- ^ Input stream + -> t m (Array Word8) -- ^ Output stream +processChunksWith modifier path args input = + processChunksWithAction run modifier path args + + where + + run (Just stdinH, Just stdoutH, _, _) = + putChunksClose stdinH input `parallel` toChunksClose stdoutH + run _ = error "processChunks: Not reachable" + +-- | @processChunks file args input@ runs the executable @file@ specified by +-- its name or path using @args@ as arguments and @input@ stream as its +-- standard input. Returns the standard output of the executable as a stream. +-- +-- If only the name of an executable file is specified instead of its path then +-- the file name is searched in the directories specified by the PATH +-- environment variable. +-- +-- If the input stream throws an exception or if the output stream is garbage +-- collected before it could finish then the process is sent a SIGTERM and we +-- wait for it to terminate gracefully. +-- +-- If the process terminates with a non-zero exit code then a 'ProcessFailure' +-- exception is raised. +-- +-- The following code is equivalent to the shell command @echo "hello world" | +-- tr [a-z] [A-Z]@: +-- +-- >>> :{ +-- Process.toChunks "echo" ["hello world"] +-- & Process.processChunks "tr" ["[a-z]", "[A-Z]"] +-- & Stream.fold Stdio.writeChunks +-- :} +--HELLO WORLD +-- +-- @since 0.1.0 +{-# INLINE processChunks #-} +processChunks :: + (IsStream t, MonadCatch m, MonadAsync m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m (Array Word8) -- ^ Input stream + -> t m (Array Word8) -- ^ Output stream +processChunks = processChunksWith id + +-- | Like 'processChunks' except that it works on a stream of bytes instead of +-- a stream of chunks. +-- +-- We can write the example in 'processChunks' as follows. Notice how +-- seamlessly we can replace the @tr@ process with the Haskell @toUpper@ +-- function: +-- +-- >>> :{ +-- Process.toBytes "echo" ["hello world"] +-- & Unicode.decodeLatin1 & Stream.map toUpper & Unicode.encodeLatin1 +-- & Stream.fold Stdio.write +-- :} +--HELLO WORLD +-- +-- @since 0.1.0 +{-# INLINE processBytes #-} +processBytes :: + (IsStream t, MonadCatch m, MonadAsync m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m Word8 -- ^ Input Stream + -> t m Word8 -- ^ Output Stream +processBytes path args input = -- rights . processBytes' path args + let input1 = ArrayStream.arraysOf defaultChunkSize input + output = processChunks path args input1 + in Stream.unfoldMany Array.read output + +------------------------------------------------------------------------------- +-- Generation +------------------------------------------------------------------------------- +-- +-- | @toBytes' path args@ runs the executable at @path@ using @args@ as +-- arguments and returns a stream of 'Either' bytes. The 'Left' values are from +-- @stderr@ and the 'Right' values are from @stdout@ of the executable. +-- +-- Raises 'ProcessFailure' exception in case of failure. +-- +-- The following example uses @echo@ to write @hello@ to @stdout@ and @world@ +-- to @stderr@, then uses folds from "Streamly.Console.Stdio" to write them +-- back to @stdout@ and @stderr@ respectively: +-- +-- +-- >>> :{ +-- Process.toBytes' "/bin/bash" ["-c", "echo 'hello'; echo 'world' 1>&2"] +-- & Stream.fold (Fold.partition Stdio.writeErr Stdio.write) +-- :} +-- world +-- hello +-- ((),()) +-- +-- >>> toBytes' path args = Process.processBytes' path args Stream.nil +-- +-- @since 0.1.0 +{-# INLINE toBytes' #-} +toBytes' :: + (IsStream t, MonadAsync m, MonadCatch m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m (Either Word8 Word8) -- ^ Output Stream +toBytes' path args = processBytes' path args Stream.nil + +-- | See 'processBytes'. 'toBytes' is defined as: +-- +-- >>> toBytes path args = processBytes path args Stream.nil +-- +-- The following code is equivalent to the shell command @echo "hello world"@: +-- +-- >>> :{ +-- Process.toBytes "echo" ["hello world"] +-- & Stream.fold Stdio.write +-- :} +--hello world +-- +-- @since 0.1.0 +{-# INLINE toBytes #-} +toBytes :: + (IsStream t, MonadAsync m, MonadCatch m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m Word8 -- ^ Output Stream +toBytes path args = processBytes path args Stream.nil + +-- | Like 'toBytes' but generates a stream of @Array Word8@ instead of a stream +-- of @Word8@. +-- +-- >>> :{ +-- toChunks' "bash" ["-c", "echo 'hello'; echo 'world' 1>&2"] +-- & Stream.fold (Fold.partition Stdio.writeErrChunks Stdio.writeChunks) +-- :} +-- world +-- hello +-- ((),()) +-- +-- >>> toChunks' path args = processChunks' path args Stream.nil +-- +-- Prefer 'toChunks' over 'toBytes' when performance matters. +-- +-- @since 0.1.0 +{-# INLINE toChunks' #-} +toChunks' :: + (IsStream t, MonadAsync m, MonadCatch m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m (Either (Array Word8) (Array Word8)) -- ^ Output Stream +toChunks' path args = processChunks' path args Stream.nil + +-- | See 'processChunks'. 'toChunks' is defined as: +-- +-- >>> toChunks path args = processChunks path args Stream.nil +-- +-- The following code is equivalent to the shell command @echo "hello world"@: +-- +-- >>> :{ +-- Process.toChunks "echo" ["hello world"] +-- & Stream.fold Stdio.writeChunks +-- :} +--hello world +-- +-- @since 0.1.0 +{-# INLINE toChunks #-} +toChunks :: + (IsStream t, MonadAsync m, MonadCatch m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> t m (Array Word8) -- ^ Output Stream +toChunks path args = processChunks path args Stream.nil diff --git a/src/Streamly/Internal/System/Process/Posix.hs b/src/Streamly/Internal/System/Process/Posix.hs new file mode 100644 --- /dev/null +++ b/src/Streamly/Internal/System/Process/Posix.hs @@ -0,0 +1,317 @@ +-- | +-- Module : Streamly.Internal.System.Process.Posix +-- Copyright : (c) 2020 Composewell Technologies +-- License : Apache-2.0 +-- Maintainer : streamly@composewell.com +-- Stability : experimental +-- Portability : GHC +-- + +{-# LANGUAGE CPP #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Streamly.Internal.System.Process.Posix + ( + -- * Processes + Process + , newProcess + , wait + , getStatus + + -- * IPC + , mkPipe + , mkStdioPipes + ) +where + +import Control.Concurrent + (MVar, newMVar, readMVar, withMVar, modifyMVar, modifyMVar_) +import Control.Exception (catch, throwIO, Exception(..), onException) +import Control.Monad (void) +import Data.Bifunctor (first) +import Data.Tuple (swap) +import GHC.IO.Device (IODeviceType(..)) +import GHC.IO.Encoding (getLocaleEncoding) +import GHC.IO.Handle.FD (mkHandleFromFD) +import System.IO (IOMode(..), Handle) +import System.IO.Error (isDoesNotExistError) +import System.Posix.IO (createPipe, dupTo, closeFd) +import System.Posix.Process (forkProcess, executeFile, ProcessStatus) +import System.Posix.Types (ProcessID, Fd(..), CDev, CIno) +import System.Posix.Internals (fdGetMode) + +import qualified GHC.IO.FD as FD +import qualified System.Posix.Process as Posix + +------------------------------------------------------------------------------- +-- Utilities to create stdio handles +------------------------------------------------------------------------------- + +-- See GHC.IO.Handle.FD +-- We have to put the FDs into binary mode on Windows to avoid the newline +-- translation that the CRT IO library does. +setBinaryMode :: FD.FD -> IO () +#if defined(mingw32_HOST_OS) +setBinaryMode fd = do + _ <- setmode (FD.fdFD fd) True + return () +#else +setBinaryMode _ = return () +#endif + +-- See Posix.fdToHandle and GHC.IO.Handle.FD.fdToHandle +-- See stdin, stdout, stderr in module GHC.IO.Handle.FD +-- +-- This routines avoids a few system calls and does a few more things compared +-- to fdToHandle. +stdioFdToHandle :: + Bool -> Maybe IOMode -> Maybe (IODeviceType, CDev, CIno) -> Fd -> IO Handle +stdioFdToHandle binary mbIOMode mbStat (Fd fdint) = do + iomode <- + case mbIOMode of + Just mode -> return mode + Nothing -> fdGetMode fdint + (fd, fd_type) <- FD.mkFD fdint iomode mbStat + False{-is_socket-} + -- NB. the is_socket flag is False, meaning that: + -- on Windows we're guessing this is not a socket (XXX) + False{-is_nonblock-} + -- file descriptors that we get from external sources are + -- not put into non-blocking mode, because that would affect + -- other users of the file descriptor + setBinaryMode fd + enc <- if binary then return Nothing else fmap Just getLocaleEncoding + -- Should we set the FD non-blocking? + -- See https://gitlab.haskell.org/ghc/ghc/-/issues/3316 + let fd_str = "" + mkHandleFromFD fd fd_type fd_str iomode True{-non-block-} enc + +------------------------------------------------------------------------------- +-- Setup pipes between parent and child processes +------------------------------------------------------------------------------- + +-- | ParentToChild: parent writes, child reads +data Direction = ParentToChild | ChildToParent deriving (Show, Eq) + +-- | return (parent, child, (parentAction, childAction, failureAction)) +mkPipe :: Direction -> IO (Fd, Fd, (IO (), IO (), IO ())) +mkPipe direction = do + let setDirection = if direction == ParentToChild then id else swap + (child, parent) <- fmap setDirection createPipe + let parentAction = closeFd child + childAction = closeFd parent + failureAction = closeFd child >> closeFd parent + return (parent, child, (parentAction, childAction, failureAction)) + +-- | The child end of the pipe is duped to the supplied fd. The parent side fd +-- of the pipe is returned. +mkPipeDupChild :: Direction -> Fd -> IO (Fd, (IO (), IO (), IO ())) +mkPipeDupChild direction childFd = do + let setDirection = if direction == ParentToChild then id else swap + (child, parent) <- fmap setDirection createPipe + let parentAction = closeFd child + childAction = + closeFd parent >> void (dupTo child childFd) >> closeFd child + failureAction = closeFd child >> closeFd parent + return (parent, (parentAction, childAction, failureAction)) + +-- XXX We could possibly combine the triples from individual pipes in a more +-- idiomatic manner. +mkStdioPipes :: Bool -> IO ((Handle, Handle, Maybe Handle, Handle, Handle), IO (), IO (), IO ()) +mkStdioPipes pipeStdErr = do + -- stdin + (inp, (inpParent, inpChild, inpFail)) <- mkPipeDupChild ParentToChild 0 + + -- stdout + (out, (outParent, outChild, outFail)) <- mkPipeDupChild ChildToParent 1 + `onException` inpFail + + -- stderr + (err, (errParent, errChild, errFail)) <- + if pipeStdErr + then first Just <$> mkPipeDupChild ChildToParent 2 + `onException` (inpFail >> outFail) + else return (Nothing, (return (), return (), return ())) + + {- + -- exception channel + (efdParent, efdChild, (excParent, excChild, excFail)) <- + mkPipe ChildToParent + `onException` (inpFail >> outFail >> errFail) + -} + + let parentAction = inpParent >> outParent >> errParent -- >> excParent + childAction = inpChild >> outChild >> errChild -- >> excChild + failureAction = inpFail >> outFail >> errFail -- >> excFail + + inpH <- toHandle WriteMode inp + outH <- toHandle ReadMode out + errH <- + case err of + Just x -> Just <$> toHandle ReadMode x + Nothing -> return Nothing + -- ehParent <- stdioFdToHandle + -- True (Just ReadMode) (Just (Stream, 0, 0)) efdParent + -- ehChild Paren<- stdioFdToHandle + -- True (Just ReadMode) (Just (Stream, 0, 0)) efdChild + let ehParent = undefined + let ehChild = undefined + return ( (inpH, outH, errH, ehParent, ehChild) + , parentAction + , childAction + , failureAction + ) + + where + + toHandle mode = stdioFdToHandle False (Just mode) (Just (Stream, 0, 0)) + +------------------------------------------------------------------------------- +-- Process handle +------------------------------------------------------------------------------- + +-- Note: We have two types of users reading and modifying the process handle, +-- (1) blocking waiters - wait, (2) non-blocking users - +-- getStatus. We need to ensure that blocking waiters always get the +-- status, and non-blocking waiters always get the status if the process is +-- done otherwise return Nothing. +-- +-- One locking strategy could be that blocking waiters take a lock, and +-- non-blocking waiters try the same lock and if they cannot acquire it then +-- return Nothing. But the problem with this is that even after the process is +-- done non-blocking waiters locking may fail due to contention and it may +-- return Nothing, which is wrong. +-- +-- Instead we use the following strategy. Everyone first looks up the status +-- under the ProcessStatus lock. If the process is done we return the status. +-- If not done then: +-- +-- - blocking users take the "waitlock" and perform a blocking waitpid, then +-- update the status by taking the ProcessStatus lock. This ensures that all +-- the blocking waiters are synchronized. +-- +-- - non-blocking users perform non-blocking waitpid under the ProcessStatus +-- lock, this ensures that blocking users will not miss a reaping done by +-- non-blocking users. But non-blocking users may still miss a reaping done by +-- blocking users, if a blocking user reaped but is waiting for ProcessStatus +-- lock to update it. To take care of that case non-blocking users use the +-- result of waitpid to detect if the process has already been reaped and if so +-- they try again using the blocking users' waitlock. We know that this cannot +-- block for a long time any more since the process has been reaped. This time +-- if we still cannot get the status then it is some real error or bug so we +-- raise an exception. +-- +-- | Thread safe, mutable process handle. Process status is stored in the +-- handle and is modified by the process inspection operations. +data Process = + Process + ProcessID -- Read only + (MVar ()) -- waitlock + (MVar (Maybe ProcessStatus)) -- ProcessStatus lock + +------------------------------------------------------------------------------- +-- Creating a Process +------------------------------------------------------------------------------- + +-- If this API is to be exported then we should wrap it in another function +-- that checks if the pid really exists by doing a non-blocking waitpid on it. +-- +-- | Turn an existing process pid into a 'Process' handle. +pidToProcess :: ProcessID -> Maybe ProcessStatus -> IO Process +pidToProcess pid status = do + waitLock <- newMVar () + st <- newMVar status + return $ Process pid waitLock st + +-- | Creates a new process, executes the specified action in the cloned process +-- and then performs an @exec@ system call using the provided path, arguments +-- and environment. The PATH is searched for the specified binary when the +-- specified path is not absolute? +newProcess :: + IO () -- ^ Execute after fork, before exec + -> FilePath -- ^ Executable path + -> [String] -- ^ Arguments + -> Maybe [(String, String)] -- ^ Environment + -> IO Process +newProcess action path args env = do + pid <- forkProcess exec + pidToProcess pid Nothing + + where + + -- XXX What if exec failed or the "action" failed? Need to send the error + -- to the parent and clean up the parent fds. We can send the exceptions + -- via a pipe like we do for threads. + -- + exec = action >> executeFile path True args env + +newtype ProcessDoesNotExist = ProcessDoesNotExist ProcessID deriving Show + +instance Exception ProcessDoesNotExist where + + displayException (ProcessDoesNotExist pid) = + "Bug: Process with pid " ++ show pid ++ " does not exist." + +-- | Wait until the process exits by itself or gets terminated due to a signal. +-- Returns the 'ProcessStatus' which includes the termination reason or exit +-- code. +-- +-- Thread safe. +wait :: Process -> IO ProcessStatus +wait (Process pid waitLock procStatus) = do + status <- readMVar procStatus + case status of + Just st -> return st + Nothing -> withMVar waitLock $ \() -> waitStatus `catch` eChild + + where + + waitStatus = do + st <- Posix.getProcessStatus True{-block-} False{-stopped-} pid + case st of + Nothing -> error $ "wait: Bug: Posix.getProcessStatus " + ++ "returned Nothing in blocking mode" + Just s -> do + modifyMVar_ procStatus $ \_ -> return st + return s + + -- Got eCHILD, some non-blocking user already reaped the process. + eChild e = do + if isDoesNotExistError e + then do + st <- readMVar procStatus + case st of + Nothing -> throwIO $ ProcessDoesNotExist pid + Just s -> return s + else throwIO e + +-- | Get the current status of a process. A 'Nothing' value means the process +-- is still running, a 'Just' value means the process is terminated and +-- provides the status of the process. +-- +-- Thread safe. +-- +getStatus :: Process -> IO (Maybe ProcessStatus) +getStatus proc@(Process pid _ procStatus) = do + r <- modifyMVar + procStatus + $ \old -> + case old of + Just _ -> return (old, Just old) + Nothing -> fetchStatus `catch` eChild + case r of + Just st -> return st + Nothing -> Just <$> wait proc + + where + + fetchStatus = do + st <- Posix.getProcessStatus False{-block-} False{-stopped-} pid + return (st, Just st) + + -- Got eCHILD, some blocking user already reaped the process. + -- We need to go through the blocking wait API to synchronize. + eChild e = do + if isDoesNotExistError e + then return (Nothing, Nothing) + else throwIO e diff --git a/src/Streamly/System/Process.hs b/src/Streamly/System/Process.hs new file mode 100644 --- /dev/null +++ b/src/Streamly/System/Process.hs @@ -0,0 +1,139 @@ +-- | +-- Module : Streamly.System.Process +-- Copyright : (c) 2020 Composewell Technologies +-- License : Apache-2.0 +-- Maintainer : streamly@composewell.com +-- Stability : experimental +-- Portability : GHC +-- +-- Use OS processes as stream transformation functions. +-- +-- This module provides functions to run operating system processes as stream +-- source, sink or transformation functions. Thus OS processes can be used in +-- the same way as Haskell functions and all the streaming combinators in +-- streamly can be used to combine them. This allows you to seamlessly +-- integrate external programs into your Haskell program. +-- +-- We recommend that you use Streamly threads instead of system processes where +-- possible as they have a simpler programming model and processes have a +-- larger performance overhead. +-- +-- = Imports for examples +-- +-- Use the following imports for examples below. +-- +-- >>> :set -XFlexibleContexts +-- >>> :set -XScopedTypeVariables +-- >>> import Data.Char (toUpper) +-- >>> import Data.Function ((&)) +-- >>> import qualified Streamly.Console.Stdio as Stdio +-- >>> import qualified Streamly.Data.Array.Foreign as Array +-- >>> import qualified Streamly.Data.Fold as Fold +-- >>> import qualified Streamly.Prelude as Stream +-- >>> import qualified Streamly.System.Process as Process +-- >>> import qualified Streamly.Unicode.Stream as Unicode +-- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir +-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream +-- +-- = Executables as functions +-- +-- Streamly provides powerful ways to combine streams. Processes can be +-- composed in a streaming pipeline just like a Posix shell command pipeline +-- except that we use @&@ instead of @|@. Moreover, we can mix processes and +-- Haskell functions seamlessly in a processing pipeline. For example: +-- +-- >>> :{ +-- Process.toBytes "echo" ["hello world"] +-- & Process.processBytes "tr" ["[a-z]", "[A-Z]"] +-- & Stream.fold Stdio.write +-- :} +-- HELLO WORLD +-- +-- Of course, you can use a Haskell function instead of "tr": +-- +-- >>> :{ +-- Process.toBytes "echo" ["hello world"] +-- & Unicode.decodeLatin1 & Stream.map toUpper & Unicode.encodeLatin1 +-- & Stream.fold Stdio.write +-- :} +-- HELLO WORLD +-- +-- = Shell commands as functions +-- +-- Using a shell as the command interpreter we can use shell commands in a data +-- processing pipeline: +-- +-- >>> :{ +-- Process.toBytes "sh" ["-c", "echo hello | tr [a-z] [A-Z]"] +-- & Stream.fold Stdio.write +-- :} +-- HELLO +-- +-- = Concurrent Processing +-- +-- We can run executables or commands concurrently as we would run any other +-- functions in Streamly. For example, the following program greps the word +-- "to" in all the files in the current directory concurrently: +-- +-- >>> :{ +-- grep file = +-- Process.toBytes "grep" ["-H", "to", file] +-- & Stream.handle (\(_ :: Process.ProcessFailure) -> Stream.nil) +-- & Stream.splitWithSuffix (== 10) Array.write +-- :} +-- +-- >>> :{ +-- pgrep = +-- Dir.toFiles "." +-- & Stream.concatMapWith Stream.async grep +-- & Stream.fold Stdio.writeChunks +-- :} +-- +-- = Exception handling +-- +-- Since we are composing using Streamly streaming pipeline there is nothing +-- special about exception handling, it works the same as in Streamly. Like +-- the @pipefail@ option in shells, exceptions are propagated if any of the +-- stages fail. +-- +-- = Process Attributes +-- +-- When a new process is spawned, the following attributes are inherited from +-- the parent process: +-- +-- * Current working directory +-- * Environment +-- * Open file descriptors +-- * Process group +-- * Process uid and gid +-- * Signal handlers +-- * Terminal (Session) + +module Streamly.System.Process + ( ProcessFailure (..) + + -- * Generation + , toChunks + , toBytes + + -- * Transformation + , processChunks + , processBytes + ) +where + +import Streamly.Internal.System.Process + +-- $setup +-- >>> :set -XFlexibleContexts +-- >>> :set -XScopedTypeVariables +-- >>> import Data.Char (toUpper) +-- >>> import Data.Function ((&)) +-- >>> import qualified Streamly.Console.Stdio as Stdio +-- >>> import qualified Streamly.Data.Array.Foreign as Array +-- >>> import qualified Streamly.Data.Fold as Fold +-- >>> import qualified Streamly.Prelude as Stream +-- >>> import qualified Streamly.System.Process as Process +-- >>> import qualified Streamly.Unicode.Stream as Unicode +-- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir +-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream diff --git a/streamly-process.cabal b/streamly-process.cabal new file mode 100644 --- /dev/null +++ b/streamly-process.cabal @@ -0,0 +1,136 @@ +cabal-version: 2.2 +name: streamly-process +version: 0.1.0 +synopsis: Use OS processes as stream transformation functions +description: + Run operating system processes as stream source, sink or transformation + functions. Use them seamlessly in a streaming data pipeline in the same way + as any other Haskell functions. +homepage: https://streamly.composewell.com +bug-reports: https://github.com/composewell/streamly-process/issues +license: Apache-2.0 +license-file: LICENSE +author: Composewell Technologies +maintainer: streamly@composewell.com +copyright: Composewell Technologies +category: Streamly, Streaming, System +stability: Experimental +tested-with: GHC==9.0.1, GHC==8.10.5, GHC==8.8.4, GHC==8.6.5, GHC==8.4.4 +build-type: Simple +extra-source-files: + CHANGELOG.md + NOTICE + README.md + design/proposal.md +source-repository head + type: git + location: https://github.com/composewell/streamly-process + +flag fusion-plugin + description: Use fusion plugin for benchmarks and executables + +flag use-gauge + description: Use gauge instead of tasty-bench for benchmarking + manual: True + default: False + +common compile-options + default-language: Haskell2010 + ghc-options: + -Wall + -Wcompat + -Wunrecognised-warning-flags + -Widentities + -Wincomplete-record-updates + -Wincomplete-uni-patterns + -Wredundant-constraints + -Wnoncanonical-monad-instances + +common optimization-options + ghc-options: + -O2 + -fdicts-strict + -fspec-constr-recursive=16 + -fmax-worker-args=16 + +library + import: compile-options, optimization-options + hs-source-dirs: src + exposed-modules: + Streamly.System.Process + Streamly.Internal.System.Process + if !os(windows) + exposed-modules: + Streamly.Internal.System.Process.Posix + ghc-options: + -Wall + -Wcompat + -Wunrecognised-warning-flags + -Widentities + -Wincomplete-record-updates + -Wincomplete-uni-patterns + -Wredundant-constraints + -Wnoncanonical-monad-instances + build-depends: + base >= 4.8 && < 5 + , process >= 1.0 && < 1.7 + -- Uses internal APIs + , streamly >= 0.8 && < 0.8.1 + , exceptions >= 0.8 && < 0.11 + if !os(windows) + build-depends: + unix >= 2.5 && < 2.8 + +------------------------------------------------------------------------------- +-- Benchmarks +------------------------------------------------------------------------------- + +common threading-options + ghc-options: + -threaded + -with-rtsopts=-N + +benchmark Benchmark.System.Process + import: compile-options, optimization-options + if flag(fusion-plugin) && !impl(ghc < 8.6) + ghc-options: -fplugin Fusion.Plugin + type: exitcode-stdio-1.0 + main-is: Benchmark/System/Process.hs + build-depends: + streamly-process + , base >= 4.8 && < 5 + , directory >= 1.2.2 && < 1.4 + , process >= 1.0 && < 1.7 + -- Uses internal APIs + , streamly >= 0.8 && < 0.8.1 + + if flag(fusion-plugin) && !impl(ghc < 8.6) + build-depends: + fusion-plugin >= 0.2 && < 0.3 + if flag(use-gauge) + build-depends: gauge >= 0.2.4 && < 0.3 + else + build-depends: tasty-bench >= 0.2.5 && < 0.4 + mixins: tasty-bench + (Test.Tasty.Bench as Gauge + , Test.Tasty.Bench as Gauge.Main + ) + +------------------------------------------------------------------------------- +-- Test Suites +------------------------------------------------------------------------------- + +test-suite Test.System.Process + import: compile-options, optimization-options, threading-options + type: exitcode-stdio-1.0 + main-is: test/Streamly/System/Process.hs + build-depends: + streamly-process + , base >= 4.8 && < 5 + , directory >= 1.2.2 && < 1.4 + , exceptions >= 0.8 && < 0.11 + , hspec >= 2.0 && < 3 + , process >= 1.0 && < 1.7 + , QuickCheck >= 2.10 && < 2.15 + -- Uses internal APIs + , streamly >= 0.8 && < 0.8.1 diff --git a/test/Streamly/System/Process.hs b/test/Streamly/System/Process.hs new file mode 100644 --- /dev/null +++ b/test/Streamly/System/Process.hs @@ -0,0 +1,533 @@ +{-# LANGUAGE CPP #-} + +module Main where + +import Data.Function ((&)) +import Data.List ((\\)) +import Data.Maybe (fromMaybe) +import Data.Word (Word8) +import Control.Exception (Exception) +import Control.Monad (unless) +import Control.Monad.Catch (throwM, catch) +import Control.Monad.IO.Class (MonadIO(..)) +import Streamly.System.Process (ProcessFailure (..)) +import System.Directory (removeFile, findExecutable) +import System.IO (IOMode(..), openFile, hClose) +import System.Process (callCommand) +import Test.Hspec (hspec, describe) +import Test.Hspec.QuickCheck +import Test.QuickCheck + ( forAll + , choose + , Property + , listOf + , counterexample + ) +import Test.QuickCheck.Monadic (monadicIO, PropertyM, assert, monitor, run) + +import qualified Streamly.Data.Fold as Fold +import qualified Streamly.Prelude as S +import qualified Streamly.System.Process as Proc + +-- Internal imports +import qualified Streamly.Internal.Data.Array.Stream.Foreign + as AS (arraysOf, concat) +import qualified Streamly.Internal.Data.Stream.IsStream as S (nilM, lefts, rights) +import qualified Streamly.Internal.FileSystem.Handle as FH (putBytes, toBytes) +import qualified Streamly.Internal.System.Process as Proc + +newtype SimpleError = SimpleError String + deriving Show + +instance Exception SimpleError + +_a :: Word8 +_a = 97 + +_z :: Word8 +_z = 122 + +_A :: Word8 +_A = 65 + +_Z :: Word8 +_Z = 90 + +which :: String -> IO FilePath +which cmd = do + r <- findExecutable cmd + case r of + Just path -> return path + _ -> error $ "Required command " ++ cmd ++ " not found" + +failErrorMessage :: String +failErrorMessage = "fail" + +minBlockCount :: Int +minBlockCount = 1 + +maxBlockCount :: Int +maxBlockCount = 100 + +minNumChar :: Int +minNumChar = 1 + +maxNumChar :: Int +maxNumChar = 100 * 1024 + +arrayChunkSize :: Int +arrayChunkSize = 100 + +-- XXX Commit these files to repo +executableFile :: FilePath +#if mingw32_HOST_OS == 1 +executableFile = "./writeTrToError.bat" +#else +executableFile = "./writeTrToError.sh" +#endif + +executableFileContent :: String +#if mingw32_HOST_OS == 1 +executableFileContent = "@echo off\r\ntr [a-z] [A-Z] >&2\r\n" +#else +executableFileContent = + "tr [a-z] [A-Z] >&2" +#endif + +executableFileFail :: FilePath +#if mingw32_HOST_OS == 1 +executableFileFail = "./failExec.bat" +#else +executableFileFail = "./failExec.sh" +#endif + +executableFileFailContent :: String +executableFileFailContent = + "exit 1" + +executableFilePass :: FilePath +#if mingw32_HOST_OS == 1 +executableFilePass = "./passExec.bat" +#else +executableFilePass = "./passExec.sh" +#endif + +executableFilePassContent :: String +executableFilePassContent = + "exit 0" + +createExecutable :: FilePath -> String -> IO () +createExecutable file content = do + writeFile file content + callCommand ("chmod +x " ++ file) + +createExecutables :: IO () +createExecutables = do + createExecutable executableFile executableFileContent + createExecutable executableFileFail executableFileFailContent + createExecutable executableFilePass executableFilePassContent + +removeExecutables :: IO () +removeExecutables = do + removeFile executableFile + removeFile executableFileFail + removeFile executableFilePass + +toUpper :: Word8 -> Word8 +toUpper char = + if _a <= char && char <= _z + then (char - _a) + _A + else char + +listEquals :: (Show a, Eq a, MonadIO m) + => ([a] -> [a] -> Bool) -> [a] -> [a] -> PropertyM m () +listEquals eq process_output list = do + unless (process_output `eq` list) $ liftIO $ putStrLn $ + "process output " ++ show process_output + ++ "\nlist " ++ show list + ++ "\nprocess output \\\\ list " ++ show (process_output \\ list) + ++ "\nlist \\\\ process output " ++ show (list \\ process_output) + unless (process_output `eq` list) $ + monitor + (counterexample $ + "process output " ++ show process_output + ++ "\nlist " ++ show list + ++ "\nprocess output \\\\ list " ++ show (process_output \\ list) + ++ "\nlist \\\\ process output " ++ show (list \\ process_output) + ) + assert (process_output `eq` list) + +charFile :: String +charFile = "./largeCharFile" + +generateCharFile :: Int -> IO () +generateCharFile numCharInCharFile = do + handle <- openFile charFile WriteMode + FH.putBytes handle (S.replicate numCharInCharFile _a) + hClose handle + +toBytes1 :: Property +toBytes1 = + forAll (choose (minBlockCount, maxBlockCount)) $ \numBlock -> + monadicIO $ do + catBinary <- run $ which "cat" + let + genStrm = S.rights $ Proc.toBytes' catBinary [charFile] + + ioByteStrm = do + handle <- openFile charFile ReadMode + let strm = FH.toBytes handle + return $ S.finally (hClose handle) strm + + run $ generateCharFile numBlock + byteStrm <- run ioByteStrm + genList <- run $ S.toList genStrm + byteList <- run $ S.toList byteStrm + run $ removeFile charFile + listEquals (==) genList byteList + +toBytes2 :: Property +toBytes2 = monadicIO $ run checkFailAction + where + + action = do + S.drain $ Proc.toBytes' executableFileFail [] + return False + + failAction (ProcessFailure exitCode) = + return (exitCode == 1) + + checkFailAction = catch action failAction + +toChunks1 :: Property +toChunks1 = + forAll (choose (minBlockCount, maxBlockCount)) $ \numBlock -> + monadicIO $ do + catBinary <- run $ which "cat" + let + genStrm = S.rights $ Proc.toChunks' catBinary [charFile] + + ioByteStrm = do + handle <- openFile charFile ReadMode + let strm = FH.toBytes handle + return $ S.finally (hClose handle) strm + + run $ generateCharFile numBlock + byteStrm <- run ioByteStrm + genList <- run $ S.toList (AS.concat genStrm) + byteList <- run $ S.toList byteStrm + run $ removeFile charFile + listEquals (==) genList byteList + +toChunks2 :: Property +toChunks2 = monadicIO $ run checkFailAction + where + + action = do + S.drain $ Proc.toChunks' executableFileFail [] + return False + + failAction (ProcessFailure exitCode) = + return (exitCode == 1) + + checkFailAction = catch action failAction + +processBytes1 :: Property +processBytes1 = + forAll (listOf (choose(_a, _z))) $ \ls -> + monadicIO $ do + trBinary <- run $ which "tr" + let + inputStream = S.fromList ls + genStrm = Proc.processBytes + trBinary + ["[a-z]", "[A-Z]"] + inputStream + charUpperStrm = S.map toUpper inputStream + + genList <- run $ S.toList genStrm + charList <- run $ S.toList charUpperStrm + listEquals (==) genList charList + +processBytes2 :: Property +processBytes2 = monadicIO $ run checkFailAction + where + + action = do + S.drain $ Proc.processBytes executableFileFail [] S.nil + return False + + failAction (ProcessFailure exitCode) = + return (exitCode == 1) + + checkFailAction = catch action failAction + +processBytes3 :: Property +processBytes3 = monadicIO $ run checkFailAction + where + + action = do + S.drain $ + Proc.processBytes + executableFilePass + [] + (S.nilM $ throwM (SimpleError failErrorMessage)) + return False + + failAction (SimpleError err) = + return (err == failErrorMessage) + + checkFailAction = catch action failAction + +-- Termination on input termination +processChunksConsumeAllInput :: Property +processChunksConsumeAllInput = + forAll (listOf (choose(_a, _z))) $ \ls -> + monadicIO $ do + trBinary <- run $ which "tr" + let + inputStream = S.fromList ls + + genStrm = AS.concat $ + Proc.processChunks + trBinary + ["[a-z]", "[A-Z]"] + (AS.arraysOf arrayChunkSize inputStream) + + charUpperStrm = S.map toUpper inputStream + + genList <- run $ S.toList genStrm + charList <- run $ S.toList charUpperStrm + listEquals (==) genList charList + +processChunksConsumePartialInput :: Property +processChunksConsumePartialInput = + forAll (listOf (choose(9, _z))) $ \ls -> + monadicIO $ do + path <- run $ which "head" + let + inputStream = S.fromList ls + + procOutput = AS.concat $ + Proc.processChunks + path + ["-n", "1"] + (AS.arraysOf arrayChunkSize inputStream) + + headLine = + S.splitWithSuffix (== 10) Fold.toList inputStream + & S.head + + procList <- run $ S.toList procOutput + expectedList <- run $ fmap (fromMaybe []) headLine + listEquals (==) procList expectedList + +processChunksProcessFailure :: Property +processChunksProcessFailure = monadicIO $ run $ catch runProcess checkExitCode + + where + + runProcess = do + S.drain $ Proc.processChunks executableFileFail [] S.nil + return False + + checkExitCode (ProcessFailure exitCode) = return (exitCode == 1) + +processChunksInputFailure :: Property +processChunksInputFailure = monadicIO $ run $ catch runProcess checkException + + where + + runProcess = do + S.drain $ + Proc.processChunks + executableFilePass + [] + (S.nilM $ throwM (SimpleError failErrorMessage)) + return False + + checkException (SimpleError err) = return (err == failErrorMessage) + +processBytes'1 :: Property +processBytes'1 = + forAll (listOf (choose(_a, _z))) $ \ls -> + monadicIO $ do + trBinary <- run $ which "tr" + let + inputStream = S.fromList ls + genStrm = S.rights $ Proc.processBytes' + trBinary + ["[a-z]", "[A-Z]"] + inputStream + charUpperStrm = S.map toUpper inputStream + + genList <- run $ S.toList genStrm + charList <- run $ S.toList charUpperStrm + listEquals (==) genList charList + +processBytes'2 :: Property +processBytes'2 = + forAll (listOf (choose(_a, _z))) $ \ls -> + monadicIO $ do + let + inputStream = S.fromList ls + outStream = S.lefts $ + Proc.processBytes' + executableFile + ["[a-z]", "[A-Z]"] + inputStream + + charUpperStrm = S.map toUpper inputStream + + genList <- run $ S.toList outStream + charList <- run $ S.toList charUpperStrm + listEquals (==) genList charList + +processBytes'3 :: Property +processBytes'3 = monadicIO $ run checkFailAction + where + + action = do + S.drain $ Proc.processBytes' executableFileFail [] S.nil + return False + + failAction (ProcessFailure exitCode) = + return (exitCode == 1) + + checkFailAction = catch action failAction + +processBytes'4 :: Property +processBytes'4 = monadicIO $ run checkFailAction + where + + action = do + S.drain $ + Proc.processBytes' + executableFilePass + [] + (S.nilM $ throwM (SimpleError failErrorMessage)) + return False + + failAction (SimpleError err) = + return (err == failErrorMessage) + + checkFailAction = catch action failAction + +processChunks'1 :: Property +processChunks'1 = + forAll (listOf (choose(_a, _z))) $ \ls -> + monadicIO $ do + trBinary <- run $ which "tr" + let + inputStream = S.fromList ls + + genStrm = AS.concat $ S.rights $ + Proc.processChunks' + trBinary + ["[a-z]", "[A-Z]"] + (AS.arraysOf arrayChunkSize inputStream) + + charUpperStrm = S.map toUpper inputStream + + genList <- run $ S.toList genStrm + charList <- run $ S.toList charUpperStrm + listEquals (==) genList charList + +processChunks'2 :: Property +processChunks'2 = + forAll (listOf (choose(_a, _z))) $ \ls -> + monadicIO $ do + let + inputStream = S.fromList ls + outStream = AS.concat $ S.lefts $ + Proc.processChunks' + executableFile + ["[a-z]", "[A-Z]"] + (AS.arraysOf arrayChunkSize inputStream) + + charUpperStrm = S.map toUpper inputStream + + genList <- run $ S.toList outStream + charList <- run $ S.toList charUpperStrm + listEquals (==) genList charList + +processChunks'3 :: Property +processChunks'3 = monadicIO $ run checkFailAction + where + + action = do + S.drain $ Proc.processChunks' executableFileFail [] S.nil + return False + + failAction (ProcessFailure exitCode) = + return (exitCode == 1) + + checkFailAction = catch action failAction + +processChunks'4 :: Property +processChunks'4 = monadicIO $ run checkFailAction + where + + action = do + S.drain $ + Proc.processChunks' + executableFilePass + [] + (S.nilM $ throwM (SimpleError failErrorMessage)) + return False + + failAction (SimpleError err) = + return (err == failErrorMessage) + + checkFailAction = catch action failAction + +main :: IO () +main = do + createExecutables + hspec $ do + describe "Streamly.System.Process" $ do + -- XXX Add a test for garbage collection case. Also check whether + -- the process is really gone after exception or gc. + -- + -- Keep the tests in dependency order so that we test the basic + -- things first. + describe "processChunks'" $ do + prop + "AS.concat $ processChunks' tr = map toUpper" + processChunks'1 + prop + "error stream of processChunks' tr = map toUpper" + processChunks'2 + prop "processChunks' on failing executable" processChunks'3 + prop "processChunks' using error stream" processChunks'4 + + describe "processChunks" $ do + prop "consumeAllInput" processChunksConsumeAllInput + prop "consumePartialInput" processChunksConsumePartialInput + prop "ProcessFailure" processChunksProcessFailure + prop "inputFailure" processChunksInputFailure + + -- based on processChunks + describe "processBytes'" $ do + prop "processBytes' tr = map toUpper" processBytes'1 + prop + "error stream of processBytes' tr = map toUpper" + processBytes'2 + prop "processBytes' on failing executable" processBytes'3 + prop "processBytes' using error stream" processBytes'4 + + describe "processBytes" $ do + prop "processBytes tr = map toUpper" processBytes1 + prop "processBytes on failing executable" processBytes2 + prop "processBytes using error stream" processBytes3 + + -- Based on processBytes/Chunks + describe "toChunks'" $ do + prop "toChunks' cat = FH.toChunks" toChunks1 + prop "toChunks' on failing executable" toChunks2 + + describe "toBytes'" $ do + prop "toBytes' cat = FH.toBytes" toBytes1 + prop "toBytes' on failing executable" toBytes2 + + removeExecutables