streamly-process (empty) → 0.1.0
raw patch · 12 files changed
+2245/−0 lines, 12 filesdep +QuickCheckdep +basedep +directorysetup-changed
Dependencies added: QuickCheck, base, directory, exceptions, fusion-plugin, gauge, hspec, process, streamly, streamly-process, tasty-bench, unix
Files
- Benchmark/System/Process.hs +240/−0
- CHANGELOG.md +5/−0
- LICENSE +177/−0
- NOTICE +5/−0
- README.md +12/−0
- Setup.hs +2/−0
- design/proposal.md +61/−0
- src/Streamly/Internal/System/Process.hs +618/−0
- src/Streamly/Internal/System/Process/Posix.hs +317/−0
- src/Streamly/System/Process.hs +139/−0
- streamly-process.cabal +136/−0
- test/Streamly/System/Process.hs +533/−0
+ Benchmark/System/Process.hs view
@@ -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+ )
+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Changelog++## 0.1.0 (Jul 2021)++* Initial version.
+ LICENSE view
@@ -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
+ NOTICE view
@@ -0,0 +1,5 @@+streamly-process+Copyright 2020 Composewell Technologies++This product includes software developed at+Composewell Technologies (http://www.composewell.com).
+ README.md view
@@ -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).
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ design/proposal.md view
@@ -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
+ src/Streamly/Internal/System/Process.hs view
@@ -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
+ src/Streamly/Internal/System/Process/Posix.hs view
@@ -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 = "<stdioFdToHandle file descriptor: " ++ show fd ++ ">"+ 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
+ src/Streamly/System/Process.hs view
@@ -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
+ streamly-process.cabal view
@@ -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
+ test/Streamly/System/Process.hs view
@@ -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