diff --git a/Benchmark/System/Process.hs b/Benchmark/System/Process.hs --- a/Benchmark/System/Process.hs +++ b/Benchmark/System/Process.hs @@ -1,11 +1,12 @@ {-# LANGUAGE ScopedTypeVariables #-} +{-# OPTIONS_GHC -fspec-constr-recursive=4 #-} 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 Test.Tasty.Bench (defaultMain, bench, nfIO) import System.Directory (removeFile, findExecutable) import System.IO ( Handle @@ -13,25 +14,24 @@ , openFile , hClose ) -import System.Process (proc, createProcess, waitForProcess, callCommand) +import Streamly.Data.Stream (Stream) import qualified Streamly.Data.Fold as FL -import qualified Streamly.FileSystem.Handle as FH -import qualified Streamly.Prelude as S +import qualified Streamly.Data.Stream as S import qualified Streamly.System.Process as Proc +import qualified Streamly.Internal.System.Command as Cmd -- Internal imports -import qualified Streamly.Internal.FileSystem.Handle - as FH (toBytes, toChunks, putBytes, putChunks) +import qualified Streamly.Internal.FileSystem.Handle as FH 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 :: Monad m => Stream m (Either a b) -> Stream 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 :: Monad m => Stream m (Either a b) -> Stream m a lefts = fmap (fromLeft undefined) . S.filter isLeft ------------------------------------------------------------------------------- @@ -71,16 +71,12 @@ 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 () + Cmd.toStdout + $ ddPath + ++ " if=" ++ devRandom + ++ " of=" ++ largeByteFile + ++ " count=" ++ show ddBlockCount + ++ " bs=" ++ show ddBlockSize ------------------------------------------------------------------------------- -- Create a file filled with ascii chars @@ -107,12 +103,12 @@ trToStderrContent :: String trToStderrContent = - "tr [a-z] [A-Z] <&0 >&2" + "#!/bin/sh\ntr [a-z] [A-Z] <&0 >&2" createExecutable :: IO () createExecutable = do writeFile trToStderr trToStderrContent - callCommand ("chmod +x " ++ trToStderr) + Cmd.toStdout ("chmod +x " ++ trToStderr) ------------------------------------------------------------------------------- -- Create and delete the temp data/exec files @@ -134,36 +130,36 @@ -- Benchmark functions ------------------------------------------------------------------------------- -toBytes' :: String-> Handle -> IO () -toBytes' catPath outH = +toBytesEither :: String-> Handle -> IO () +toBytesEither catPath outH = FH.putBytes outH $ rights - $ Proc.toBytes' catPath [largeByteFile] + $ Proc.toBytesEither catPath [largeByteFile] -toChunks' :: String -> Handle -> IO () -toChunks' catPath hdl = +toChunksEither :: String -> Handle -> IO () +toChunksEither catPath hdl = FH.putChunks hdl $ rights - $ Proc.toChunks' catPath [largeByteFile] + $ Proc.toChunksEither catPath [largeByteFile] -processBytes' :: String-> Handle -> IO () -processBytes' trPath outputHdl = do +pipeBytesEither :: String-> Handle -> IO () +pipeBytesEither trPath outputHdl = do inputHdl <- openFile largeCharFile ReadMode _ <- S.fold (FL.partition (FH.write outputHdl) (FH.write outputHdl)) - $ Proc.processBytes' + $ Proc.pipeBytesEither trPath ["[a-z]", "[A-Z]"] - $ FH.toBytes inputHdl + $ FH.read inputHdl hClose inputHdl -processBytes :: String-> Handle -> IO () -processBytes trPath outputHdl = do +pipeBytes :: String-> Handle -> IO () +pipeBytes trPath outputHdl = do inputHdl <- openFile largeCharFile ReadMode FH.putBytes outputHdl - $ Proc.processBytes + $ Proc.pipeBytes trPath ["[a-z]", "[A-Z]"] - $ FH.toBytes inputHdl + $ FH.read inputHdl hClose inputHdl processBytesToStderr :: Handle -> IO () @@ -171,33 +167,33 @@ inputHdl <- openFile largeCharFile ReadMode FH.putBytes outputHdl $ lefts - $ Proc.processBytes' + $ Proc.pipeBytesEither trToStderr ["[a-z]", "[A-Z]"] - $ FH.toBytes inputHdl + $ FH.read inputHdl hClose inputHdl -processChunks :: String -> Handle -> IO () -processChunks trPath outputHdl = do +pipeChunks :: String -> Handle -> IO () +pipeChunks trPath outputHdl = do inputHdl <- openFile largeCharFile ReadMode FH.putChunks outputHdl $ - Proc.processChunks + Proc.pipeChunks trPath ["[a-z]", "[A-Z]"] - $ FH.toChunks inputHdl + $ FH.readChunks inputHdl hClose inputHdl -processChunks' :: String -> Handle -> IO () -processChunks' trPath outputHdl = do +pipeChunksEither :: String -> Handle -> IO () +pipeChunksEither trPath outputHdl = do inputHdl <- openFile largeCharFile ReadMode _ <- S.fold (FL.partition (FH.writeChunks outputHdl) (FH.writeChunks outputHdl) ) - $ Proc.processChunks' + $ Proc.pipeChunksEither trPath ["[a-z]", "[A-Z]"] - (FH.toChunks inputHdl) + (FH.readChunks inputHdl) hClose inputHdl processChunksToStderr :: Handle -> IO () @@ -205,10 +201,10 @@ inputHdl <- openFile largeCharFile ReadMode FH.putChunks outputHdl $ lefts - $ Proc.processChunks' + $ Proc.pipeChunksEither trToStderr ["[a-z]", "[A-Z]"] - (FH.toChunks inputHdl) + (FH.readChunks inputHdl) hClose inputHdl ------------------------------------------------------------------------------- @@ -225,13 +221,13 @@ 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 "toBytesEither" $ nfIO $ toBytesEither catPath nullH + , bench "toChunksEither" $ nfIO $ toChunksEither catPath nullH + , bench "pipeBytes tr" $ nfIO $ pipeBytes trPath nullH + , bench "pipeBytesEither tr" $ nfIO $ pipeBytesEither trPath nullH , bench "processBytesToStderr tr" $ nfIO $ processBytesToStderr nullH - , bench "processChunks tr" $ nfIO (processChunks trPath nullH) - , bench "processChunks' tr" $ nfIO (processChunks' trPath nullH) + , bench "pipeChunks tr" $ nfIO (pipeChunks trPath nullH) + , bench "pipeChunksEither tr" $ nfIO (pipeChunksEither trPath nullH) , bench "processChunksToStderr" $ nfIO $ processChunksToStderr nullH ] `finally` (do putStrLn "cleanup ..." diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## 0.3.0 (Apr 2023) + +* Added a `Streamly.System.Command` module +* `toChunks` and `toBytes` now make the stdin available to the process being + executed. +* Allow streamly 0.9.0 +* Signature changes - removed the IsStream constraint, now all APIs use the + monomorphic `Stream` type. + ## 0.2.0.1 (Mar 2022) * Fix the test suite. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,8 +1,77 @@ # 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. +Use operating system (OS) commands in Haskell programs as if they were +native Haskell functions, by treating their inputs and outputs as +Haskell streams. This allows you to write high-level Haskell scripts +that can perform tasks similar to shell scripts, but with C-like +performance, and with strong safety guarantees, refactorability, and +modularity. + +Use the following imports in the examples below: + +```haskell +>>> :set -XFlexibleContexts +>>> :set -XScopedTypeVariables +>>> :set -XQuasiQuotes +>>> import Data.Function ((&)) +>>> import Streamly.Unicode.String (str) +>>> import qualified Streamly.Data.Array as Array +>>> import qualified Streamly.Console.Stdio as Stdio +>>> import qualified Streamly.Data.Fold as Fold +>>> import qualified Streamly.Data.Stream.Prelude as Stream +>>> import qualified Streamly.System.Command as Command +>>> import qualified Streamly.Internal.FileSystem.Dir as Dir +``` + +## Commands as streaming functions + +The shell command `echo "hello world" | tr [a-z] [A-Z]` can be written as +follows using this package: + +```haskell +>>> :{ + Command.toBytes [str|echo "hello world"|] -- Stream IO Word8 + & Command.pipeBytes [str|tr [a-z] [A-Z]|] -- Stream IO Word8 + & Stream.fold Stdio.write -- IO () + :} + HELLO WORLD +``` + +## Shell commands as streaming functions + +If you want to execute the same command using the shell pipe syntax: + +```haskell +>>> :{ + Command.toBytes [str|sh "-c" "echo 'hello world' | tr [a-z] [A-Z]"|] -- Stream IO Word8 + & Stream.fold Stdio.write -- IO () + :} + HELLO WORLD +``` + +## Running Commands Concurrently + +You can run commands concurrently by using streamly's concurrent combinators. + +Running @grep@ concurrently on many files: + +```haskell +>>> :{ +grep file = + Command.toBytes [str|grep -H "pattern" #{file}|] -- Stream IO Word8 + & Stream.handle (\(_ :: Command.ProcessFailure) -> Stream.nil) -- Stream IO Word8 + & Stream.foldMany (Fold.takeEndBy (== 10) Array.write) -- Stream IO (Array Word8) + :} + +>>> :{ +pgrep = + Dir.readFiles "." -- Stream IO FilePath + & Stream.parConcatMap id grep -- Stream IO (Array Word8) + & Stream.fold Stdio.writeChunks -- IO () +:} +``` + +## Streamly Please visit [Streamly homepage](https://streamly.composewell.com) for more details about streamly. diff --git a/src/Streamly/Internal/System/Command.hs b/src/Streamly/Internal/System/Command.hs new file mode 100644 --- /dev/null +++ b/src/Streamly/Internal/System/Command.hs @@ -0,0 +1,318 @@ +-- | +-- Module : Streamly.Internal.System.Command +-- Copyright : (c) 2022 Composewell Technologies +-- License : Apache-2.0 +-- Maintainer : streamly@composewell.com +-- Stability : experimental +-- Portability : GHC +-- +{-# LANGUAGE CPP #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE FlexibleContexts #-} + +module Streamly.Internal.System.Command + ( + -- * Generation + toBytes + , toChunks + , toChars + , toLines + + -- * Effects + , toString + , toStdout + , toNull + + -- * Transformation + , pipeBytes + , pipeChars + , pipeChunks + + -- * Helpers + , runWith + , streamWith + , pipeWith + ) +where + +import Control.Monad.Catch (MonadCatch) +import Data.Char (isSpace) +import Data.Word (Word8) +import Streamly.Data.Array (Array) +import Streamly.Data.Fold (Fold) +import Streamly.Data.Parser (Parser) +import Streamly.Data.Stream.Prelude (MonadAsync, Stream) + +import qualified Streamly.Data.Fold as Fold +import qualified Streamly.Data.Parser as Parser +import qualified Streamly.Data.Stream.Prelude as Stream +import qualified Streamly.Internal.System.Process as Process + +-- Keep it synced with the released module + +-- $setup +-- >>> :set -XFlexibleContexts +-- >>> :set -XQuasiQuotes +-- >>> import Data.Char (toUpper) +-- >>> import Data.Function ((&)) +-- >>> import Streamly.Unicode.String (str) +-- >>> import qualified Streamly.Data.Array as Array +-- >>> import qualified Streamly.Console.Stdio as Stdio +-- >>> import qualified Streamly.Data.Fold as Fold +-- >>> import qualified Streamly.Data.Stream.Prelude as Stream +-- >>> import qualified Streamly.System.Command as Command +-- >>> import qualified Streamly.Unicode.Stream as Unicode +-- +-- >>> import qualified Streamly.Internal.System.Process as Process +-- >>> import qualified Streamly.Internal.Console.Stdio as Stdio +-- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir + +{-# INLINE quotedWord #-} +quotedWord :: MonadCatch m => Parser Char m String +quotedWord = + let toRQuote x = + case x of + '"' -> Just x + '\'' -> Just x + _ -> Nothing + trEsc q x = if q == x then Just x else Nothing + in Parser.wordWithQuotes False trEsc '\\' toRQuote isSpace Fold.toList + +-- | A modifier for stream generation APIs in "Streamly.System.Process" to +-- generate streams from command strings. +-- +-- For example: +-- +-- >>> streamWith Process.toBytes "echo hello" & Stdio.putBytes +-- hello +-- >>> streamWith Process.toChunks "echo hello" & Stdio.putChunks +-- hello +-- +-- /Internal/ +{-# INLINE streamWith #-} +streamWith :: MonadCatch m => + (FilePath -> [String] -> Stream m a) -> String -> Stream m a +streamWith f cmd = + Stream.concatEffect $ do + xs <- Stream.fold Fold.toList + $ Stream.catRights + $ Stream.parseMany quotedWord + $ Stream.fromList cmd + case xs of + y:ys -> return $ f y ys + _ -> error "streamWith: empty command" + +-- | A modifier for process running APIs in "Streamly.System.Process" to run +-- command strings. +-- +-- For example: +-- +-- >>> runWith Process.toString "echo hello" +-- "hello\n" +-- >>> runWith Process.toStdout "echo hello" +-- hello +-- +-- /Internal/ +{-# INLINE runWith #-} +runWith :: MonadCatch m => + (FilePath -> [String] -> m a) -> String -> m a +runWith f cmd = do + xs <- Stream.fold Fold.toList + $ Stream.catRights + $ Stream.parseMany quotedWord + $ Stream.fromList cmd + case xs of + y:ys -> f y ys + _ -> error "streamWith: empty command" + +-- | A modifier for process piping APIs in "Streamly.System.Process" to pipe +-- data through processes specified by command strings. +-- +-- For example: +-- +-- >>> :{ +-- toChunks "echo hello" +-- & pipeWith Process.pipeChunks "tr [a-z] [A-Z]" +-- & Stdio.putChunks +-- :} +--HELLO +-- +-- /Internal/ +pipeWith :: MonadCatch m => + (FilePath -> [String] -> Stream m a -> Stream m b) + -> String + -> Stream m a + -> Stream m b +pipeWith f cmd input = + Stream.concatEffect $ do + xs <- Stream.fold Fold.toList + $ Stream.catRights + $ Stream.parseMany quotedWord + $ Stream.fromList cmd + case xs of + y:ys -> return $ f y ys input + _ -> error "streamWith: empty command" + + +-- | @pipeChunks command input@ runs the executable with arguments specified by +-- @command@ and supplying @input@ stream as its standard input. Returns the +-- standard output of the executable as a stream of byte arrays. +-- +-- 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 terminated with SIGTERM. +-- +-- 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]@: +-- +-- >>> :{ +-- toChunks "echo hello world" +-- & pipeChunks "tr [a-z] [A-Z]" +-- & Stdio.putChunks +-- :} +--HELLO WORLD +-- +-- /Pre-release/ +{-# INLINE pipeChunks #-} +pipeChunks :: (MonadAsync m, MonadCatch m) => + String -> Stream m (Array Word8) -> Stream m (Array Word8) +pipeChunks = pipeWith Process.pipeChunks + +-- | Like 'pipeChunks' except that it works on a stream of bytes instead of +-- a stream of chunks. +-- +-- >>> :{ +-- toBytes "echo hello world" +-- & pipeBytes "tr [a-z] [A-Z]" +-- & Stdio.putBytes +-- :} +--HELLO WORLD +-- +-- /Pre-release/ +{-# INLINE pipeBytes #-} +pipeBytes :: (MonadAsync m, MonadCatch m) => + String -> Stream m Word8 -> Stream m Word8 +pipeBytes = pipeWith Process.pipeBytes + +-- | Like 'pipeChunks' except that it works on a stream of chars instead of +-- a stream of chunks. +-- +-- >>> :{ +-- toChars "echo hello world" +-- & pipeChars "tr [a-z] [A-Z]" +-- & Stdio.putChars +-- :} +--HELLO WORLD +-- +-- /Pre-release/ +{-# INLINE pipeChars #-} +pipeChars :: (MonadAsync m, MonadCatch m) => + String -> Stream m Char -> Stream m Char +pipeChars = pipeWith Process.pipeChars + +------------------------------------------------------------------------------- +-- Generation +------------------------------------------------------------------------------- + +-- | +-- +-- >>> toBytes = streamWith Process.toBytes +-- +-- >>> toBytes "echo hello world" & Stdio.putBytes +--hello world +-- >>> toBytes "echo hello\\ world" & Stdio.putBytes +--hello world +-- >>> toBytes "echo 'hello world'" & Stdio.putBytes +--hello world +-- >>> toBytes "echo \"hello world\"" & Stdio.putBytes +--hello world +-- +-- /Pre-release/ +{-# INLINE toBytes #-} +toBytes :: (MonadAsync m, MonadCatch m) => String -> Stream m Word8 +toBytes = streamWith Process.toBytes + +-- | +-- +-- >>> toChunks = streamWith Process.toChunks +-- +-- >>> toChunks "echo hello world" & Stdio.putChunks +--hello world +-- +-- /Pre-release/ +{-# INLINE toChunks #-} +toChunks :: (MonadAsync m, MonadCatch m) => String -> Stream m (Array Word8) +toChunks = streamWith Process.toChunks + +-- | +-- >>> toChars = streamWith Process.toChars +-- +-- >>> toChars "echo hello world" & Stdio.putChars +--hello world +-- +-- /Pre-release/ +{-# INLINE toChars #-} +toChars :: (MonadAsync m, MonadCatch m) => String -> Stream m Char +toChars = streamWith Process.toChars + +-- | +-- >>> toLines f = streamWith (Process.toLines f) +-- +-- >>> toLines Fold.toList "echo -e hello\\\\nworld" & Stream.fold Fold.toList +-- ["hello","world"] +-- +-- /Pre-release/ +{-# INLINE toLines #-} +toLines :: + (MonadAsync m, MonadCatch m) + => Fold m Char a + -> String -- ^ Command + -> Stream m a -- ^ Output Stream +toLines f = streamWith (Process.toLines f) + +-- | +-- >>> toString = runWith Process.toString +-- +-- >>> toString "echo hello world" +--"hello world\n" +-- +-- /Pre-release/ +{-# INLINE toString #-} +toString :: + (MonadAsync m, MonadCatch m) + => String -- ^ Command + -> m String +toString = runWith Process.toString + +-- | +-- >>> toStdout = runWith Process.toStdout +-- +-- >>> toStdout "echo hello world" +-- hello world +-- +-- /Pre-release/ +{-# INLINE toStdout #-} +toStdout :: + (MonadAsync m, MonadCatch m) + => String -- ^ Command + -> m () +toStdout = runWith Process.toStdout + +-- | +-- >>> toNull = runWith Process.toNull +-- +-- >>> toNull "echo hello world" +-- +-- /Pre-release/ +{-# INLINE toNull #-} +toNull :: + (MonadAsync m, MonadCatch m) + => String -- ^ Command + -> m () +toNull = runWith Process.toNull diff --git a/src/Streamly/Internal/System/Process.hs b/src/Streamly/Internal/System/Process.hs --- a/src/Streamly/Internal/System/Process.hs +++ b/src/Streamly/Internal/System/Process.hs @@ -49,45 +49,101 @@ -- * Process Configuration Config + -- ** Common Config Options + -- | These options apply to both POSIX and Windows. + , setCwd + , setEnv + {- + , setStdin + , setStdout + , setStderr + -} + , closeFiles + , newProcessGroup + , setSession + + -- * Posix Only Options + -- | These options have no effect on Windows. + , parentIgnoresInterrupt + , setUserId + , setGroupId + + -- * Windows Only Options + -- | These options have no effect on Posix. + , waitForChildTree + + -- * Internal + , inheritStdin + , inheritStdout + , pipeStdErr + -- * Exceptions , ProcessFailure (..) -- * Generation + -- | stdout of the process is redirected to output stream. , toBytes - , toBytes' , toChunks - , toChunks' + , toChunksWith + , toChars + , toLines + , toString + , toStdout + , toNull -- * Transformation + -- | The input stream is redirected to the stdin of the process, stdout of + -- the process is redirected to the output stream. + , pipeBytes + , pipeChunks + , pipeChunksWith + , pipeChars + + -- * Stderr + -- | Like other "Generation" routines but along with stdout, stderr is also + -- included in the output stream. stdout is converted to 'Right' values in + -- the output stream and stderr is converted to 'Left' values. + , toBytesEither + , toChunksEither + , toChunksEitherWith + , pipeBytesEither + , pipeChunksEither + , pipeChunksEitherWith + + -- * Standalone Processes + , standalone + , interactive + , daemon + + -- * Deprecated , processBytes - , processBytes' - , processChunksWith , processChunks - , processChunks'With - , processChunks' ) where --- #define USE_NATIVE - +import Control.Concurrent (forkIO) +import Control.Exception (Exception(..), catch, throwIO) +import Control.Monad (void, unless) 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 Data.Function ((&)) +import Data.Word (Word8, Word32) +import Foreign.C.Error (Errno(..), ePIPE) +import GHC.IO.Exception (IOException(..), IOErrorType(..)) +import Streamly.Data.Array (Array) +import Streamly.Data.Fold (Fold) +import Streamly.Data.Stream.Prelude (MonadAsync, Stream) import System.Exit (ExitCode(..)) import System.IO (hClose, Handle) +#if !defined(mingw32_HOST_OS) +import System.Posix.Types (CUid (..), CGid (..)) +#endif #ifdef USE_NATIVE -import Control.Exception (Exception(..), catch, throwIO, SomeException) +import Control.Exception (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(..) @@ -96,32 +152,35 @@ , waitForProcess , CmdSpec(..) , terminateProcess + , withCreateProcess ) #endif -import qualified Streamly.Data.Array.Foreign as Array -import qualified Streamly.Prelude as Stream +import qualified Streamly.Data.Array as Array +import qualified Streamly.Data.Fold as Fold -- Internal imports import Streamly.Internal.System.IO (defaultChunkSize) -import qualified Streamly.Internal.Data.Array.Stream.Foreign - as ArrayStream (arraysOf) -import qualified Streamly.Internal.Data.Stream.IsStream as Stream (bracket') +import qualified Streamly.Internal.Console.Stdio as Stdio (putChunks) +import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Internal.Data.Unfold as Unfold (either) import qualified Streamly.Internal.FileSystem.Handle - as Handle (toChunks, putChunks) + as Handle (readChunks, putChunks) +import qualified Streamly.Unicode.Stream as Unicode +import qualified Streamly.Internal.Unicode.Stream as Unicode (lines) -- $setup -- >>> :set -XFlexibleContexts -- >>> import Data.Char (toUpper) -- >>> import Data.Function ((&)) --- >>> import qualified Streamly.Console.Stdio as Stdio +-- >>> import qualified Streamly.Internal.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.Data.Stream.Prelude as Stream +-- >>> import qualified Streamly.Internal.System.Process as Process -- >>> import qualified Streamly.Unicode.Stream as Unicode --- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream +-- >>> import qualified Streamly.Internal.Data.Stream as Stream +-- >>> import qualified Streamly.Internal.Unicode.Stream as Unicode ------------------------------------------------------------------------------- -- Config @@ -133,13 +192,19 @@ -- from the parent process: -- -- * Current working directory --- * Environment +-- * Environment variables -- * Open file descriptors -- * Process group +-- * Terminal session +-- +-- On POSIX: +-- -- * Process uid and gid -- * Signal handlers --- * Terminal (Session) -- +-- On Windows by default the parent process waits for the entire child process +-- tree to finish. +-- #ifdef USE_NATIVE type ProcessHandle = Process @@ -154,7 +219,15 @@ pipeStdErr :: Config -> Config pipeStdErr (Config _) = Config True + +inheritStdin :: Config -> Config +inheritStdin (Config _) = Config True + +inheritStdout :: Config -> Config +inheritStdout (Config _) = Config True + #else + newtype Config = Config CreateProcess -- | Create a default process configuration from an executable file path and @@ -178,7 +251,9 @@ , child_group = Nothing -- Posix only -- Signals (Posix only) behavior - -- Reset SIGINT (Ctrl-C) and SIGQUIT (Ctrl-\) to default handlers. + -- Ignore SIGINT (Ctrl-C) and SIGQUIT (Ctrl-\) in the parent process until + -- the child exits i.e. let the child handle it. See + -- https://www.cons.org/cracauer/sigint.html . , delegate_ctlc = False -- Terminal behavior @@ -190,8 +265,183 @@ , use_process_jobs = True -- Windows only } +-- XXX use osPath + +-- | Set the current working directory of the new process. When 'Nothing', the +-- working directory is inherited from the parent process. +-- +-- Default is 'Nothing' - inherited from the parent process. +setCwd :: Maybe (FilePath) -> Config -> Config +setCwd path (Config cfg) = Config $ cfg { cwd = path } + +-- | Set the environment variables for the new process. When 'Nothing', the +-- environment is inherited from the parent process. +-- +-- Default is 'Nothing' - inherited from the parent process. +setEnv :: Maybe [(String, String)] -> Config -> Config +setEnv e (Config cfg) = Config $ cfg { env = e } + +{- +-- XXX We should allow setting only those stdio streams which are not used for +-- piping. We can either close those or inherit from parent. +-- +-- * In a source we have to close stdin and use stdout +-- * In a pipe we have to use both +-- * In a sink we have to close stdout and use stdin. +-- +-- Only stderr may be left for setting - either pipe it to merge it with stdout +-- or inherit or close. Closing may lead to bad behavior in most cases. So it +-- is either inherit or merge with stdout. Merge with stdout can be achieved by +-- using the either combinators. So there is nothing really left to set here +-- for any std stream. + +-- UseProc, UsePipe? +-- | What to do with the stdin, stdout, stderr streams of the process. +data Stdio = + ToHaskell -- ^ Pipe to Haskell Streamly + | ToProcess -- ^ Connect to the parent process + +toStdStream :: Stdio -> StdStream +toStdStream x = + case x of + ToProcess -> Inherit + ToHaskell -> CreatePipe + +-- | What to do with the @stdin@ stream of the process. +setStdin :: Stdio -> Config -> Config +setStdin x (Config cfg) = Config $ cfg { std_in = toStdStream x } + +-- | What to do with the @stdout@ stream of the process. +setStdout :: Stdio -> Config -> Config +setStdout x (Config cfg) = Config $ cfg { std_out = toStdStream x } + +-- | What to do with the @stderr@ stream of the process. +setStderr :: Stdio -> Config -> Config +setStderr x (Config cfg) = Config $ cfg { std_err = toStdStream x } +-} + +-- | Close all open file descriptors inherited from the parent process. Note, +-- this does not apply to stdio descriptors - the behavior of those is determined +-- by other configuration settings. +-- +-- Default is 'False'. +-- +-- Note: if the number of open descriptors is large, it may take a while +-- closing them. +closeFiles :: Bool -> Config -> Config +closeFiles x (Config cfg) = Config $ cfg { close_fds = x } + +-- XXX Do these details apply to Windows as well? + +-- | If 'True' the new process starts a new process group, becomes a process +-- group leader, its pid becoming the process group id. +-- +-- See the POSIX @setpgid@ man page. +-- +-- Default is 'False', the new process belongs to the parent's process group. +newProcessGroup :: Bool -> Config -> Config +newProcessGroup x (Config cfg) = Config $ cfg { create_group = x } + +-- | 'InheritSession' makes the new process inherit the terminal session from the +-- parent process. This is the default. +-- +-- 'NewSession' makes the new process start with a new session without a +-- controlling terminal. On POSIX, @setsid@ is used to create a new process +-- group and session, the pid of the new process is the session id and process +-- group id as well. On Windows @DETACHED_PROCESS@ flag is used to detach the +-- process from inherited console session. +-- +-- 'NewConsole' creates a new terminal and attaches the process to the new +-- terminal on Windows, using the CREATE_NEW_CONSOLE flag. On POSIX this does +-- nothing. +-- +-- For Windows see +-- * https://learn.microsoft.com/en-us/windows/win32/procthread/process-creation-flags +-- * https://learn.microsoft.com/en-us/windows/console/creation-of-a-console . +-- +-- For POSIX see, @setsid@ man page. +data Session = + InheritSession -- ^ Inherit the parent session + | NewSession -- ^ Detach process from the current session + | NewConsole -- ^ Windows only, CREATE_NEW_CONSOLE flag + +-- | Define the terminal session behavior for the new process. +-- +-- Default is 'InheritSession'. +setSession :: Session -> Config -> Config +setSession x (Config cfg) = + Config $ + case x of + InheritSession -> cfg + NewSession -> cfg { new_session = True} + NewConsole -> cfg {create_new_console = True} + +-- | Use the POSIX @setuid@ call to set the user id of the new process before +-- executing the command. The parent process must have sufficient privileges to +-- set the user id. +-- +-- POSIX only. See the POSIX @setuid@ man page. +-- +-- Default is 'Nothing' - inherit from the parent. +setUserId :: Maybe Word32 -> Config -> Config +#if defined(mingw32_HOST_OS) +setUserId _ (Config cfg) = + Config cfg +#else +setUserId x (Config cfg) = + Config $ cfg { child_user = CUid <$> x } +#endif + +-- | Use the POSIX @setgid@ call to set the group id of the new process before +-- executing the command. The parent process must have sufficient privileges to +-- set the group id. +-- +-- POSIX only. See the POSIX @setgid@ man page. +-- +-- Default is 'Nothing' - inherit from the parent. +setGroupId :: Maybe Word32 -> Config -> Config +#if defined(mingw32_HOST_OS) +setGroupId _ (Config cfg) = + Config cfg +#else +setGroupId x (Config cfg) = + Config $ cfg { child_group = CGid <$> x } +#endif + +-- See https://www.cons.org/cracauer/sigint.html for more details on signal +-- handling by interactive processes. + +-- | When this is 'True', the parent process ignores user interrupt signals +-- @SIGINT@ and @SIGQUIT@ delivered to it until the child process exits. If +-- multiple child processes are started then the default handling in the parent +-- is restored only after the last one exits. +-- +-- When a user presses CTRL-C or CTRL-\ on the terminal, a SIGINT or SIGQUIT is +-- sent to all the foreground processes in the terminal session, this includes +-- both the child and the parent. By default, on receiving these signals, the +-- parent process would cleanup and exit, to avoid that and let the child +-- handle these signals we can choose to ignore these signals in the parent +-- until the child exits. +-- +-- POSIX only. Default is 'False'. +parentIgnoresInterrupt :: Bool -> Config -> Config +parentIgnoresInterrupt x (Config cfg) = Config $ cfg { delegate_ctlc = x } + +-- | On Windows, the parent waits for the entire tree of process i.e. including +-- processes that are spawned by the child process. +-- +-- Default is 'True'. +waitForChildTree :: Bool -> Config -> Config +waitForChildTree x (Config cfg) = Config $ cfg { use_process_jobs = x } + pipeStdErr :: Config -> Config pipeStdErr (Config cfg) = Config $ cfg { std_err = CreatePipe } + +inheritStdin :: Config -> Config +inheritStdin (Config cfg) = Config $ cfg { std_in = Inherit } + +inheritStdout :: Config -> Config +inheritStdout (Config cfg) = Config $ cfg { std_out = Inherit } #endif ------------------------------------------------------------------------------- @@ -213,6 +463,9 @@ displayException (ProcessFailure exitCode) = "Process failed with exit code: " ++ show exitCode +parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a +parallel s1 s2 = Stream.parList (Stream.eager True) [s1, s2] + ------------------------------------------------------------------------------- -- Transformation ------------------------------------------------------------------------------- @@ -221,9 +474,9 @@ -- 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 +cleanupNormal :: + (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> IO () +cleanupNormal (_, _, _, procHandle) = do #ifdef USE_NATIVE -- liftIO $ putStrLn "cleanupNormal waiting" status <- wait procHandle @@ -247,11 +500,15 @@ -- 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 +cleanupException :: + (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> IO () +cleanupException (Just stdinH, Just stdoutH, stderrMaybe, ph) = do -- Send a SIGTERM to the process +#ifdef USE_NATIVE + terminate ph +#else terminateProcess ph +#endif -- 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 @@ -261,7 +518,11 @@ whenJust hClose stderrMaybe -- Non-blocking wait for the process to go away +#ifdef USE_NATIVE + void $ forkIO (void $ wait ph) +#else void $ forkIO (void $ waitForProcess ph) +#endif where @@ -307,62 +568,62 @@ Config cfg = modCfg $ mkConfig path args {-# INLINE putChunksClose #-} -putChunksClose :: (MonadIO m, IsStream t) => - Handle -> t m (Array Word8) -> t m a +putChunksClose :: MonadIO m => + Handle -> Stream m (Array Word8) -> Stream m a putChunksClose h input = Stream.before - (Handle.putChunks h (adapt input) >> liftIO (hClose h)) + (Handle.putChunks h 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) +toChunksClose :: MonadAsync m => Handle -> Stream m (Array Word8) +toChunksClose h = Stream.afterIO (hClose h) (Handle.readChunks h) -{-# INLINE processChunksWithAction #-} -processChunksWithAction :: - (IsStream t, MonadCatch m, MonadAsync m) - => ((Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> t m a) +{-# INLINE pipeChunksWithAction #-} +pipeChunksWithAction :: + (MonadCatch m, MonadAsync m) + => ((Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> Stream m a) -> (Config -> Config) -> FilePath -- ^ Path to Executable -> [String] -- ^ Arguments - -> t m a -- ^ Output stream -processChunksWithAction run modCfg path args = - Stream.bracket' + -> Stream m a -- ^ Output stream +pipeChunksWithAction run modCfg path args = + Stream.bracketIO3 alloc cleanupNormal cleanupException cleanupException run where - alloc = liftIO $ createProc' modCfg path args + alloc = createProc' modCfg path args -{-# INLINE processChunks'With #-} -processChunks'With :: - (IsStream t, MonadCatch m, MonadAsync m) +{-# INLINE pipeChunksEitherWith #-} +pipeChunksEitherWith :: + (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 + -> Stream m (Array Word8) -- ^ Input stream + -> Stream m (Either (Array Word8) (Array Word8)) -- ^ Output stream +pipeChunksEitherWith modifier path args input = + pipeChunksWithAction 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" + `parallel` fmap Left (toChunksClose stderrH) + `parallel` fmap Right (toChunksClose stdoutH) + run _ = error "pipeChunksEitherWith: Not reachable" -{-# INLINE processChunks' #-} -processChunks' :: - (IsStream t, MonadCatch m, MonadAsync m) +{-# INLINE pipeChunksEither #-} +pipeChunksEither :: + (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 + -> Stream m (Array Word8) -- ^ Input stream + -> Stream m (Either (Array Word8) (Array Word8)) -- ^ Output stream +pipeChunksEither = pipeChunksEitherWith id --- | @processBytes' path args input@ runs the executable at @path@ using @args@ +-- | @pipeBytesEither 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. @@ -373,45 +634,47 @@ -- world" | tr [:lower:] [:upper:]@: -- -- >>> :{ --- processBytes' "echo" ["hello world"] Stream.nil --- & Stream.rights --- & processBytes' "tr" ["[:lower:]", "[:upper:]"] --- & Stream.rights +-- pipeBytesEither "echo" ["hello world"] Stream.nil +-- & Stream.catRights +-- & pipeBytesEither "tr" ["[:lower:]", "[:upper:]"] +-- & Stream.catRights -- & Stream.fold Stdio.write -- :} --HELLO WORLD -- -- @since 0.1.0 -{-# INLINE processBytes' #-} -processBytes' :: - (IsStream t, MonadCatch m, MonadAsync m) +{-# INLINE pipeBytesEither #-} +pipeBytesEither :: + (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 Stream.unfoldMany (Unfold.either Array.read) output + -> Stream m Word8 -- ^ Input Stream + -> Stream m (Either Word8 Word8) -- ^ Output Stream +pipeBytesEither path args input = + let input1 = Stream.chunksOf defaultChunkSize input + output = pipeChunksEither path args input1 + leftRdr = fmap Left Array.reader + rightRdr = fmap Right Array.reader + in Stream.unfoldMany (Unfold.either leftRdr rightRdr) output -{-# INLINE processChunksWith #-} -processChunksWith :: - (IsStream t, MonadCatch m, MonadAsync m) +{-# INLINE pipeChunksWith #-} +pipeChunksWith :: + (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 + -> Stream m (Array Word8) -- ^ Input stream + -> Stream m (Array Word8) -- ^ Output stream +pipeChunksWith modifier path args input = + pipeChunksWithAction run modifier path args where run (Just stdinH, Just stdoutH, _, _) = putChunksClose stdinH input `parallel` toChunksClose stdoutH - run _ = error "processChunks: Not reachable" + run _ = error "pipeChunksWith: Not reachable" --- | @processChunks file args input@ runs the executable @file@ specified by +-- | @pipeChunks 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. -- @@ -430,53 +693,136 @@ -- -- >>> :{ -- Process.toChunks "echo" ["hello world"] --- & Process.processChunks "tr" ["[a-z]", "[A-Z]"] +-- & Process.pipeChunks "tr" ["[a-z]", "[A-Z]"] -- & Stream.fold Stdio.writeChunks -- :} --HELLO WORLD -- --- @since 0.1.0 +-- /pre-release/ +{-# INLINE pipeChunks #-} +pipeChunks :: + (MonadCatch m, MonadAsync m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> Stream m (Array Word8) -- ^ Input stream + -> Stream m (Array Word8) -- ^ Output stream +pipeChunks = pipeChunksWith id + +{-# DEPRECATED processChunks "Please use pipeChunks instead." #-} {-# INLINE processChunks #-} processChunks :: - (IsStream t, MonadCatch m, MonadAsync m) + (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 + -> Stream m (Array Word8) -- ^ Input stream + -> Stream m (Array Word8) -- ^ Output stream +processChunks = pipeChunks --- | Like 'processChunks' except that it works on a stream of bytes instead of +-- | Like 'pipeChunks' 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: +-- We can write the example in 'pipeChunks' as follows. -- -- >>> :{ -- Process.toBytes "echo" ["hello world"] --- & Unicode.decodeLatin1 & Stream.map toUpper & Unicode.encodeLatin1 +-- & Process.pipeBytes "tr" ["[a-z]", "[A-Z]"] -- & Stream.fold Stdio.write -- :} --HELLO WORLD -- --- @since 0.1.0 +-- /pre-release/ +{-# INLINE pipeBytes #-} +pipeBytes :: + (MonadCatch m, MonadAsync m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> Stream m Word8 -- ^ Input Stream + -> Stream m Word8 -- ^ Output Stream +pipeBytes path args input = -- rights . pipeBytesEither path args + let input1 = Stream.chunksOf defaultChunkSize input + output = pipeChunks path args input1 + in Stream.unfoldMany Array.reader output + +{-# DEPRECATED processBytes "Please use pipeBytes instead." #-} {-# INLINE processBytes #-} processBytes :: - (IsStream t, MonadCatch m, MonadAsync m) + (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 + -> Stream m Word8 -- ^ Input Stream + -> Stream m Word8 -- ^ Output Stream +processBytes = pipeBytes +-- | Like 'pipeChunks' except that it works on a stream of chars instead of +-- a stream of chunks. +-- +-- >>> :{ +-- Process.toChars "echo" ["hello world"] +-- & Process.pipeChars "tr" ["[a-z]", "[A-Z]"] +-- & Stdio.putChars +-- :} +--HELLO WORLD +-- +-- We can seamlessly replace the @tr@ process with the Haskell @toUpper@ +-- function: +-- +-- >>> :{ +-- Process.toChars "echo" ["hello world"] +-- & fmap toUpper +-- & Stdio.putChars +-- :} +--HELLO WORLD +-- +-- /pre-release/ +{-# INLINE pipeChars #-} +pipeChars :: + (MonadCatch m, MonadAsync m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> Stream m Char -- ^ Input Stream + -> Stream m Char -- ^ Output Stream +pipeChars path args input = + Unicode.encodeUtf8 input + & pipeBytes path args + & Unicode.decodeUtf8 + ------------------------------------------------------------------------------- -- Generation ------------------------------------------------------------------------------- --- --- | @toBytes' path args@ runs the executable at @path@ using @args@ as + +{-# INLINE toChunksEitherWith #-} +toChunksEitherWith :: + (MonadCatch m, MonadAsync m) + => (Config -> Config) -- ^ Config modifier + -> FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> Stream m (Either (Array Word8) (Array Word8)) -- ^ Output stream +toChunksEitherWith modifier path args = + pipeChunksWithAction run (modifier . inheritStdin . pipeStdErr) path args + + where + + run (_, Just stdoutH, Just stderrH, _) = + fmap Left (toChunksClose stderrH) + `parallel` fmap Right (toChunksClose stdoutH) + run _ = error "toChunksEitherWith: Not reachable" + +{-# INLINE toChunksWith #-} +toChunksWith :: + (MonadCatch m, MonadAsync m) + => (Config -> Config) -- ^ Config modifier + -> FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> Stream m (Array Word8) -- ^ Output stream +toChunksWith modifier path args = + pipeChunksWithAction run (modifier . inheritStdin) path args + + where + + run (_, Just stdoutH, _, _) = toChunksClose stdoutH + run _ = error "toChunksWith: Not reachable" + +-- | @toBytesEither 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. -- @@ -486,31 +832,29 @@ -- 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"] +-- Process.toBytesEither "/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) +{-# INLINE toBytesEither #-} +toBytesEither :: + (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 + -> Stream m (Either Word8 Word8) -- ^ Output Stream +toBytesEither path args = + let output = toChunksEither path args + leftRdr = fmap Left Array.reader + rightRdr = fmap Right Array.reader + in Stream.unfoldMany (Unfold.either leftRdr rightRdr) output --- | 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"@: +-- | The following code is equivalent to the shell command @echo "hello +-- world"@: -- -- >>> :{ -- Process.toBytes "echo" ["hello world"] @@ -521,41 +865,40 @@ -- @since 0.1.0 {-# INLINE toBytes #-} toBytes :: - (IsStream t, MonadAsync m, MonadCatch m) + (MonadAsync m, MonadCatch m) => FilePath -- ^ Executable name or path -> [String] -- ^ Arguments - -> t m Word8 -- ^ Output Stream -toBytes path args = processBytes path args Stream.nil + -> Stream m Word8 -- ^ Output Stream +toBytes path args = + let output = toChunks path args + in Stream.unfoldMany Array.reader output -- | Like 'toBytes' but generates a stream of @Array Word8@ instead of a stream -- of @Word8@. -- -- >>> :{ --- toChunks' "bash" ["-c", "echo 'hello'; echo 'world' 1>&2"] +-- toChunksEither "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 +-- >>> toChunksEither = toChunksEitherWith id -- --- Prefer 'toChunks' over 'toBytes' when performance matters. +-- Prefer 'toChunksEither over 'toBytesEither when performance matters. -- --- @since 0.1.0 -{-# INLINE toChunks' #-} -toChunks' :: - (IsStream t, MonadAsync m, MonadCatch m) +-- /Pre-release/ +{-# INLINE toChunksEither #-} +toChunksEither :: + (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 + -> Stream m (Either (Array Word8) (Array Word8)) -- ^ Output Stream +toChunksEither = toChunksEitherWith id --- | 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"@: +-- | The following code is equivalent to the shell command @echo "hello +-- world"@: -- -- >>> :{ -- Process.toChunks "echo" ["hello world"] @@ -563,11 +906,154 @@ -- :} --hello world -- +-- >>> toChunks = toChunksWith id +-- -- @since 0.1.0 {-# INLINE toChunks #-} toChunks :: - (IsStream t, MonadAsync m, MonadCatch m) + (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 + -> Stream m (Array Word8) -- ^ Output Stream +toChunks = toChunksWith id + +-- | +-- >>> toChars path args = toBytes path args & Unicode.decodeUtf8 +-- +{-# INLINE toChars #-} +toChars :: + (MonadAsync m, MonadCatch m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> Stream m Char -- ^ Output Stream +toChars path args = toBytes path args & Unicode.decodeUtf8 + +-- | +-- >>> toLines path args f = toChars path args & Unicode.lines f +-- +{-# INLINE toLines #-} +toLines :: + (MonadAsync m, MonadCatch m) + => Fold m Char a + -> FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> Stream m a -- ^ Output Stream +toLines f path args = toChars path args & Unicode.lines f + +-- | +-- >>> toString path args = toChars path args & Stream.fold Fold.toList +-- +{-# INLINE toString #-} +toString :: + (MonadAsync m, MonadCatch m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> m String +toString path args = toChars path args & Stream.fold Fold.toList + +-- | +-- >>> toStdout path args = toChunks path args & Stdio.putChunks +-- +{-# INLINE toStdout #-} +toStdout :: + (MonadAsync m, MonadCatch m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> m () +toStdout path args = toChunks path args & Stdio.putChunks +{- +-- Directly inherits stdout instead. +toStdout path args = do + _ <- liftIO $ createProc' (inheritStdin . inheritStdout) path args + return () +-} + +-- | +-- >>> toNull path args = toChunks path args & Stream.fold Fold.drain +-- +{-# INLINE toNull #-} +toNull :: + (MonadAsync m, MonadCatch m) + => FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> m () +toNull path args = toChunks path args & Stream.fold Fold.drain + +------------------------------------------------------------------------------- +-- Process not interacting with the parent process +------------------------------------------------------------------------------- + +{-# INLINE standalone #-} +standalone :: + Bool -- ^ Wait for process to finish? + -> (Bool, Bool, Bool) -- ^ close (stdin, stdout, stderr) + -> (Config -> Config) + -> FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> IO (Either ExitCode ProcessHandle) +standalone wait (close_stdin, close_stdout, close_stderr) modCfg path args = + withCreateProcess cfg postCreate + + where + + postCreate _ _ _ procHandle = + if wait + then fmap Left $ waitForProcess procHandle + else return $ Right procHandle + + cfg = + let Config c = modCfg $ mkConfig path args + s_in = if close_stdin then NoStream else Inherit + s_out = if close_stdout then NoStream else Inherit + s_err = if close_stderr then NoStream else Inherit + in c {std_in = s_in, std_out = s_out, std_err = s_err} + +-- | Inherits stdin, stdout, and stderr from the parent, so that the user can +-- interact with the process, user interrupts are handled by the child process, +-- the parent waits for the child process to exit. +-- +-- This is same as the common @system@ function found in other libraries used +-- to execute commands. +-- +-- On Windows you can pass @setSession NewConsole@ to create a new console. +-- +{-# INLINE interactive #-} +interactive :: + (Config -> Config) + -> FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> IO ExitCode +interactive modCfg path args = + withCreateProcess cfg (\_ _ _ p -> waitForProcess p) + + where + + -- let child handle SIGINT/QUIT + modCfg1 = (parentIgnoresInterrupt True . modCfg) + + cfg = + let Config c = modCfg1 $ mkConfig path args + in c {std_in = Inherit, std_out = Inherit, std_err = Inherit} + +-- XXX ProcessHandle can be used to terminate the process. re-export +-- terminateProcess? + +-- | Closes stdin, stdout and stderr, creates a new session, detached from the +-- terminal, the parent does not wait for the process to finish. +-- +{-# INLINE daemon #-} +daemon :: + (Config -> Config) + -> FilePath -- ^ Executable name or path + -> [String] -- ^ Arguments + -> IO ProcessHandle +daemon modCfg path args = withCreateProcess cfg (\_ _ _ p -> return p) + + where + + -- Detach terminal + modCfg1 = (setSession NewSession . modCfg) + + cfg = + let Config c = modCfg1 $ mkConfig path args + in c {std_in = NoStream, std_out = NoStream, std_err = NoStream} diff --git a/src/Streamly/Internal/System/Process/Posix.hs b/src/Streamly/Internal/System/Process/Posix.hs --- a/src/Streamly/Internal/System/Process/Posix.hs +++ b/src/Streamly/Internal/System/Process/Posix.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE Safe #-} -- | -- Module : Streamly.Internal.System.Process.Posix -- Copyright : (c) 2020 Composewell Technologies @@ -17,6 +18,7 @@ , newProcess , wait , getStatus + , terminate -- * IPC , mkPipe @@ -38,6 +40,7 @@ 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.Signals (signalProcess, sigTERM) import System.Posix.Internals (fdGetMode) import qualified GHC.IO.FD as FD @@ -315,3 +318,6 @@ if isDoesNotExistError e then return (Nothing, Nothing) else throwIO e + +terminate :: Process -> IO () +terminate (Process pid _ _) = signalProcess sigTERM pid diff --git a/src/Streamly/System/Command.hs b/src/Streamly/System/Command.hs new file mode 100644 --- /dev/null +++ b/src/Streamly/System/Command.hs @@ -0,0 +1,158 @@ +-- | +-- Module : Streamly.System.Command +-- Copyright : (c) 2023 Composewell Technologies +-- License : Apache-2.0 +-- Maintainer : streamly@composewell.com +-- Stability : experimental +-- Portability : GHC +-- +-- Use command strings to execute OS processes. These processes can be used +-- just like native Haskell functions - to generate, transform or consume +-- streams. It provides a powerful way to write high-level Haskell scripts to +-- perform tasks similar to shell scripts without requiring the shell. +-- Moreover, the Haskell scripts provide C-like performance. +-- +-- This module is a wrapper over the "Streamly.System.Process" module. +-- +-- See also: "Streamly.Internal.System.Command". +-- +module Streamly.System.Command + ( + -- * Setup + -- | To execute the code examples provided in this module in ghci, please + -- run the following commands first. + -- + -- $setup + + -- * Overview + -- $overview + + -- * Types + ProcessFailure (..) + + -- * Generation + , toBytes + , toChunks + , toChars + , toLines + + -- * Effects + , toString + , toStdout + , toNull + + -- * Transformation + , pipeBytes + , pipeChars + , pipeChunks + + -- -- * Helpers + -- , runWith + -- , streamWith + -- , pipeWith + ) +where + +import Streamly.Internal.System.Command +import Streamly.Internal.System.Process (ProcessFailure (..)) + +-- Keep it synced with the Internal module + +-- $setup +-- >>> :set -XFlexibleContexts +-- >>> :set -XQuasiQuotes +-- >>> import Data.Char (toUpper) +-- >>> import Data.Function ((&)) +-- >>> import Streamly.Unicode.String (str) +-- >>> import qualified Streamly.Data.Array as Array +-- >>> import qualified Streamly.Console.Stdio as Stdio +-- >>> import qualified Streamly.Data.Fold as Fold +-- >>> import qualified Streamly.Data.Stream.Prelude as Stream +-- >>> import qualified Streamly.System.Command as Command +-- >>> import qualified Streamly.Unicode.Stream as Unicode +-- +-- >>> import qualified Streamly.Internal.System.Process as Process +-- >>> import qualified Streamly.Internal.Console.Stdio as Stdio +-- >>> import qualified Streamly.Internal.FileSystem.Dir as Dir + +-- Note: Commands are not executed using shell +-- +-- You can use this module to execute system commands and compose them with +-- Haskell. It does not use the system shell to execute commands, they are +-- executed as independent processes. This provides a convenient and powerful +-- way to replace shell scripting with Haskell. Instead of composing commands +-- using shell you can use Haskell Streamly streaming APIs to compose them with +-- better efficiency and type safety. +-- +-- Normally, you should not need the system shell but if you want to use shell +-- scripts in your program then you can take a look at the @streamly-shell@ +-- package which provides convenient wrapper over "Streamly.System.Process" to +-- execute shell scripts, commands. + +-- $overview +-- +-- Please see the "Streamly.System.Process" for basics. +-- +-- "Streamly.System.Process" module requires specifying the command executable +-- name and its arguments separately (e.g. "ls" "-al") whereas using this +-- module we can specify the executable and its arguments more conveniently as +-- a single command string e.g. we can execute "ls -al". +-- +-- A command string is parsed in the same way as a posix shell would parse it. +-- A command string consists of whitespace separated tokens with the first +-- token treated as the executable name and the rest as arguments. Whitespace +-- can be escaped using @\\@. Alternatively, double quotes or single quotes can +-- be used to enclose tokens with whitespaces. Quotes can be escaped using @\\@. +-- Single quotes inside double quotes or vice-versa are treated as normal +-- characters. +-- +-- You can use the string quasiquoter 'Streamly.Unicode.String.str' to write +-- commands conveniently, it allows Haskell variable expansion as well e.g.: +-- +-- >>> f = "file name" +-- >>> [str|ls -al "#{f} with spaces"|] +-- "ls -al \"file name with spaces\"" +-- +-- With the "Streamly.System.Command" module you can write the examples in the +-- "Streamly.System.Process" module more conveniently. +-- +-- = Executables as functions +-- +-- The shell command @echo "hello world" | tr [a-z] [A-Z]@ can be written as +-- follows using this module: +-- +-- >>> :{ +-- Command.toBytes [str|echo "hello world"|] +-- & Command.pipeBytes [str|tr [a-z] [A-Z]|] +-- & Stream.fold Stdio.write +-- :} +-- HELLO WORLD +-- +-- = Shell commands as functions +-- +-- If you want to execute the same command using the shell: +-- +-- >>> :{ +-- Command.toBytes [str|sh "-c" "echo 'hello world' | tr [a-z] [A-Z]"|] +-- & Stream.fold Stdio.write +-- :} +-- HELLO WORLD +-- +-- = Running Commands Concurrently +-- +-- Running @grep@ concurrently on many files: +-- +-- >>> :{ +-- grep file = +-- Command.toBytes [str|grep -H "pattern" #{file}|] +-- & Stream.handle (\(_ :: Command.ProcessFailure) -> Stream.nil) +-- & Stream.foldMany (Fold.takeEndBy (== 10) Array.write) +-- :} +-- +-- >>> :{ +-- pgrep = +-- Dir.readFiles "." +-- & Stream.parConcatMap id grep +-- & Stream.fold Stdio.writeChunks +-- :} +-- diff --git a/src/Streamly/System/Process.hs b/src/Streamly/System/Process.hs --- a/src/Streamly/System/Process.hs +++ b/src/Streamly/System/Process.hs @@ -6,45 +6,107 @@ -- 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. +-- Use OS processes just like native Haskell functions - to generate, transform +-- or consume streams. -- --- = Imports for examples +-- See "Streamly.System.Command" module for a higher level wrapper over this +-- module. -- --- Use the following imports for examples below. +-- See also: "Streamly.Internal.System.Process" for unreleased functions. -- +module Streamly.System.Process + ( + -- * Setup + -- | To execute the code examples provided in this module in ghci, please + -- run the following commands first. + -- + -- $setup + + -- * Overview + -- $overview + + -- * Exceptions + -- | Since we are composing using Streamly's 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. + ProcessFailure (..) + + -- * Process Configuration + , Config + + {- + -- ** Common Config Options + -- | These options apply to both POSIX and Windows. + , setCwd + , setEnv + , closeFiles + , newProcessGroup + , setSession + + -- * Posix Only Options + -- | These options have no effect on Windows. + , parentIgnoresInterrupt + , setUserId + , setGroupId + + -- * Windows Only Options + -- | These options have no effect on Posix. + , waitForChildTree + -} + + -- * Generation + , toChunks + , toBytes + + -- * Transformation + , pipeChunks + , pipeBytes + + -- * Deprecated + , 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.Array as Array -- >>> import qualified Streamly.Data.Fold as Fold --- >>> import qualified Streamly.Prelude as Stream +-- >>> import qualified Streamly.Data.Stream.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 +-- >>> import qualified Streamly.Internal.Data.Stream as Stream + +-- $overview -- +-- 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 using Haskell functions with Streamly threads for performing +-- tasks whenever possible. This approach offers a simpler programming model +-- compared to system processes, which also have a larger performance overhead. +-- -- = 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: +-- Processes can be composed in a streaming pipeline just like a Posix shell +-- command pipeline. 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]"] +-- & Process.pipeBytes "tr" ["[a-z]", "[A-Z]"] -- & Stream.fold Stdio.write -- :} -- HELLO WORLD @@ -53,7 +115,7 @@ -- -- >>> :{ -- Process.toBytes "echo" ["hello world"] --- & Unicode.decodeLatin1 & Stream.map toUpper & Unicode.encodeLatin1 +-- & Unicode.decodeLatin1 & fmap toUpper & Unicode.encodeLatin1 -- & Stream.fold Stdio.write -- :} -- HELLO WORLD @@ -69,7 +131,7 @@ -- :} -- HELLO -- --- = Concurrent Processing +-- = Running Commands Concurrently -- -- We can run executables or commands concurrently as we would run any other -- functions in Streamly. For example, the following program greps the word @@ -77,63 +139,14 @@ -- -- >>> :{ -- grep file = --- Process.toBytes "grep" ["-H", "to", file] +-- Process.toBytes "grep" ["-H", "pattern", file] -- & Stream.handle (\(_ :: Process.ProcessFailure) -> Stream.nil) --- & Stream.splitWithSuffix (== 10) Array.write +-- & Stream.foldMany (Fold.takeEndBy (== 10) Array.write) -- :} -- -- >>> :{ -- pgrep = --- Dir.toFiles "." --- & Stream.concatMapWith Stream.async grep +-- Dir.readFiles "." +-- & Stream.parConcatMap id 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 --- a/streamly-process.cabal +++ b/streamly-process.cabal @@ -1,11 +1,14 @@ cabal-version: 2.2 name: streamly-process -version: 0.2.0.1 +version: 0.3.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. + Use operating system (OS) commands in Haskell programs as if they were + native Haskell functions, by treating their inputs and outputs as + Haskell streams. This allows you to write high-level Haskell scripts + that can perform tasks similar to shell scripts, but with C-like + performance, and with strong safety guarantees, refactorability, and + modularity. homepage: https://streamly.composewell.com bug-reports: https://github.com/composewell/streamly-process/issues license: Apache-2.0 @@ -15,12 +18,12 @@ copyright: Composewell Technologies category: Streamly, Streaming, System stability: Experimental -tested-with: GHC==9.2.1 - , GHC==9.0.1 - , GHC==8.10.5 +tested-with: GHC==9.4.4 + , GHC==9.2.7 + , GHC==9.0.2 + , GHC==8.10.7 , GHC==8.8.4 , GHC==8.6.5 - , GHC==8.4.4 build-type: Simple extra-source-files: CHANGELOG.md @@ -43,22 +46,29 @@ manual: True default: False -flag use-gauge - description: Use gauge instead of tasty-bench for benchmarking +flag use-native + description: Do not depend on the process package 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 + -Weverything + -Wno-implicit-prelude + -Wno-missing-deriving-strategies + -Wno-missing-exported-signatures + -Wno-missing-import-lists + -Wno-missing-local-signatures + -Wno-missing-safe-haskell-mode + -Wno-missed-specialisations + -Wno-all-missed-specialisations + -Wno-monomorphism-restriction + -Wno-prepositive-qualified-module + -Wno-unsafe + if !impl(ghc < 9.2) + ghc-options: + -Wno-missing-kind-signatures common optimization-options ghc-options: @@ -66,34 +76,32 @@ -fdicts-strict -fspec-constr-recursive=16 -fmax-worker-args=16 + if flag(use-native) + cpp-options: -DUSE_NATIVE library import: compile-options, optimization-options hs-source-dirs: src exposed-modules: Streamly.System.Process + Streamly.System.Command Streamly.Internal.System.Process - if !os(windows) + Streamly.Internal.System.Command + if flag (use-native) && !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.1.* || == 0.8.2.* , exceptions >= 0.8 && < 0.11 - if !os(windows) - build-depends: - unix >= 2.5 && < 2.8 + -- Uses internal APIs + , streamly == 0.9.0.* + , streamly-core == 0.1.0 + if !flag(use-native) + build-depends: process >= 1.0 && < 1.7 + else + if !os(windows) + build-depends: + unix >= 2.5 && < 2.8 ------------------------------------------------------------------------------- -- Benchmarks @@ -114,21 +122,13 @@ streamly-process , base >= 4.8 && < 5 , directory >= 1.2.2 && < 1.4 - , process >= 1.0 && < 1.7 -- Uses internal APIs - , streamly == 0.8.1.* || == 0.8.2.* + , streamly-core == 0.1.0.* + , tasty-bench >= 0.2.5 && < 0.4 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 @@ -144,7 +144,6 @@ , 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.1.* || == 0.8.2.* + , streamly-core == 0.1.0.* diff --git a/test/Streamly/System/Process.hs b/test/Streamly/System/Process.hs --- a/test/Streamly/System/Process.hs +++ b/test/Streamly/System/Process.hs @@ -1,6 +1,6 @@ {-# LANGUAGE CPP #-} -module Main where +module Main (main) where import Data.Function ((&)) import Data.List ((\\)) @@ -10,8 +10,11 @@ import Control.Monad (unless) import Control.Monad.Catch (throwM, catch) import Control.Monad.IO.Class (MonadIO(..)) +import Streamly.Data.Array (Array, Unbox) +import Streamly.Data.Stream (Stream) import Streamly.System.Process (ProcessFailure (..)) -import System.Directory (removeFile, findExecutable) +import System.Directory (removeFile, findExecutable, doesFileExist) +import System.Exit (exitSuccess) import System.IO (IOMode(..), openFile, hClose) import Test.Hspec (hspec, describe) import Test.Hspec.QuickCheck @@ -24,16 +27,14 @@ ) import Test.QuickCheck.Monadic (monadicIO, PropertyM, assert, monitor, run) +import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold -import qualified Streamly.Prelude as S +import qualified Streamly.Data.Stream 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.FileSystem.Handle as FH (putBytes, read) import qualified Streamly.Internal.System.Process as Proc + (pipeChunksEither, pipeBytesEither, toChunksEither, toBytesEither) newtype SimpleError = SimpleError String deriving Show @@ -68,18 +69,12 @@ maxBlockCount :: Int maxBlockCount = 100 -minNumChar :: Int -minNumChar = 1 - -maxNumChar :: Int -maxNumChar = 100 * 1024 - arrayChunkSize :: Int arrayChunkSize = 100 interpreterFile :: FilePath interpreterArg :: String -#if mingw32_HOST_OS == 1 +#ifdef mingw32_HOST_OS interpreterFile = "cmd.exe" interpreterArg = "/c" #else @@ -88,21 +83,21 @@ #endif executableFile :: FilePath -#if mingw32_HOST_OS == 1 +#ifdef mingw32_HOST_OS executableFile = "./test/data/writeTrToError.bat" #else executableFile = "./test/data/writeTrToError.sh" #endif executableFileFail :: FilePath -#if mingw32_HOST_OS == 1 +#ifdef mingw32_HOST_OS executableFileFail = "./test/data/failExec.bat" #else executableFileFail = "./test/data/failExec.sh" #endif executableFilePass :: FilePath -#if mingw32_HOST_OS == 1 +#ifdef mingw32_HOST_OS executableFilePass = "./test/data/passExec.bat" #else executableFilePass = "./test/data/passExec.sh" @@ -147,17 +142,17 @@ monadicIO $ do catBinary <- run $ which "cat" let - genStrm = S.rights $ Proc.toBytes' catBinary [charFile] + genStrm = S.catRights $ Proc.toBytesEither catBinary [charFile] ioByteStrm = do handle <- openFile charFile ReadMode - let strm = FH.toBytes handle - return $ S.finally (hClose handle) strm + let strm = FH.read handle + return $ S.finallyIO (hClose handle) strm run $ generateCharFile numBlock byteStrm <- run ioByteStrm - genList <- run $ S.toList genStrm - byteList <- run $ S.toList byteStrm + genList <- run $ S.fold Fold.toList genStrm + byteList <- run $ S.fold Fold.toList byteStrm run $ removeFile charFile listEquals (==) genList byteList @@ -166,8 +161,8 @@ where action = do - S.drain $ - Proc.toBytes' interpreterFile [interpreterArg, executableFileFail] + S.fold Fold.drain $ + Proc.toBytesEither interpreterFile [interpreterArg, executableFileFail] return False failAction (ProcessFailure exitCode) = @@ -175,23 +170,27 @@ checkFailAction = catch action failAction +{-# INLINE concatArr #-} +concatArr :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a +concatArr = S.unfoldMany Array.reader + toChunks1 :: Property toChunks1 = forAll (choose (minBlockCount, maxBlockCount)) $ \numBlock -> monadicIO $ do catBinary <- run $ which "cat" let - genStrm = S.rights $ Proc.toChunks' catBinary [charFile] + genStrm = S.catRights $ Proc.toChunksEither catBinary [charFile] ioByteStrm = do handle <- openFile charFile ReadMode - let strm = FH.toBytes handle - return $ S.finally (hClose handle) strm + let strm = FH.read handle + return $ S.finallyIO (hClose handle) strm run $ generateCharFile numBlock byteStrm <- run ioByteStrm - genList <- run $ S.toList (AS.concat genStrm) - byteList <- run $ S.toList byteStrm + genList <- run $ S.fold Fold.toList (concatArr genStrm) + byteList <- run $ S.fold Fold.toList byteStrm run $ removeFile charFile listEquals (==) genList byteList @@ -200,8 +199,8 @@ where action = do - S.drain $ - Proc.toChunks' interpreterFile [interpreterArg, executableFileFail] + S.fold Fold.drain $ + Proc.toChunksEither interpreterFile [interpreterArg, executableFileFail] return False failAction (ProcessFailure exitCode) = @@ -209,30 +208,30 @@ checkFailAction = catch action failAction -processBytes1 :: Property -processBytes1 = +pipeBytes1 :: Property +pipeBytes1 = forAll (listOf (choose(_a, _z))) $ \ls -> monadicIO $ do trBinary <- run $ which "tr" let inputStream = S.fromList ls - genStrm = Proc.processBytes + genStrm = Proc.pipeBytes trBinary ["[a-z]", "[A-Z]"] inputStream - charUpperStrm = S.map toUpper inputStream + charUpperStrm = fmap toUpper inputStream - genList <- run $ S.toList genStrm - charList <- run $ S.toList charUpperStrm + genList <- run $ S.fold Fold.toList genStrm + charList <- run $ S.fold Fold.toList charUpperStrm listEquals (==) genList charList -processBytes2 :: Property -processBytes2 = monadicIO $ run checkFailAction +pipeBytes2 :: Property +pipeBytes2 = monadicIO $ run checkFailAction where action = do - S.drain $ - Proc.processBytes + S.fold Fold.drain $ + Proc.pipeBytes interpreterFile [interpreterArg, executableFileFail] S.nil return False @@ -241,13 +240,13 @@ checkFailAction = catch action failAction -processBytes3 :: Property -processBytes3 = monadicIO $ run checkFailAction +pipeBytes3 :: Property +pipeBytes3 = monadicIO $ run checkFailAction where action = do - S.drain $ - Proc.processBytes + S.fold Fold.drain $ + Proc.pipeBytes interpreterFile [interpreterArg, executableFilePass] (S.nilM $ throwM (SimpleError failErrorMessage)) @@ -267,16 +266,16 @@ let inputStream = S.fromList ls - genStrm = AS.concat $ - Proc.processChunks + genStrm = concatArr $ + Proc.pipeChunks trBinary ["[a-z]", "[A-Z]"] - (AS.arraysOf arrayChunkSize inputStream) + (S.chunksOf arrayChunkSize inputStream) - charUpperStrm = S.map toUpper inputStream + charUpperStrm = fmap toUpper inputStream - genList <- run $ S.toList genStrm - charList <- run $ S.toList charUpperStrm + genList <- run $ S.fold Fold.toList genStrm + charList <- run $ S.fold Fold.toList charUpperStrm listEquals (==) genList charList processChunksConsumePartialInput :: Property @@ -287,17 +286,19 @@ let inputStream = S.fromList ls - procOutput = AS.concat $ - Proc.processChunks + procOutput = concatArr $ + Proc.pipeChunks path ["-n", "1"] - (AS.arraysOf arrayChunkSize inputStream) + (S.chunksOf arrayChunkSize inputStream) headLine = - S.splitWithSuffix (== 10) Fold.toList inputStream - & S.head + S.foldMany + (Fold.takeEndBy (== 10) Fold.toList) + inputStream + & S.fold Fold.one - procList <- run $ S.toList procOutput + procList <- run $ S.fold Fold.toList procOutput expectedList <- run $ fmap (fromMaybe []) headLine listEquals (==) procList expectedList @@ -307,8 +308,8 @@ where runProcess = do - S.drain $ - Proc.processChunks + S.fold Fold.drain $ + Proc.pipeChunks interpreterFile [interpreterArg, executableFileFail] S.nil return False @@ -320,8 +321,8 @@ where runProcess = do - S.drain $ - Proc.processChunks + S.fold Fold.drain $ + Proc.pipeChunks interpreterFile [interpreterArg, executableFilePass] (S.nilM $ throwM (SimpleError failErrorMessage)) @@ -329,48 +330,48 @@ checkException (SimpleError err) = return (err == failErrorMessage) -processBytes'1 :: Property -processBytes'1 = +pipeBytesEither1 :: Property +pipeBytesEither1 = forAll (listOf (choose(_a, _z))) $ \ls -> monadicIO $ do trBinary <- run $ which "tr" let inputStream = S.fromList ls - genStrm = S.rights $ Proc.processBytes' + genStrm = S.catRights $ Proc.pipeBytesEither trBinary ["[a-z]", "[A-Z]"] inputStream - charUpperStrm = S.map toUpper inputStream + charUpperStrm = fmap toUpper inputStream - genList <- run $ S.toList genStrm - charList <- run $ S.toList charUpperStrm + genList <- run $ S.fold Fold.toList genStrm + charList <- run $ S.fold Fold.toList charUpperStrm listEquals (==) genList charList -processBytes'2 :: Property -processBytes'2 = +pipeBytesEither2 :: Property +pipeBytesEither2 = forAll (listOf (choose(_a, _z))) $ \ls -> monadicIO $ do let inputStream = S.fromList ls - outStream = S.lefts $ - Proc.processBytes' + outStream = S.catLefts $ + Proc.pipeBytesEither interpreterFile [interpreterArg, executableFile, "[a-z]", "[A-Z]"] inputStream - charUpperStrm = S.map toUpper inputStream + charUpperStrm = fmap toUpper inputStream - genList <- run $ S.toList outStream - charList <- run $ S.toList charUpperStrm + genList <- run $ S.fold Fold.toList outStream + charList <- run $ S.fold Fold.toList charUpperStrm listEquals (==) genList charList -processBytes'3 :: Property -processBytes'3 = monadicIO $ run checkFailAction +pipeBytesEither3 :: Property +pipeBytesEither3 = monadicIO $ run checkFailAction where action = do - S.drain $ - Proc.processBytes' + S.fold Fold.drain $ + Proc.pipeBytesEither interpreterFile [interpreterArg, executableFileFail] S.nil return False @@ -379,13 +380,13 @@ checkFailAction = catch action failAction -processBytes'4 :: Property -processBytes'4 = monadicIO $ run checkFailAction +pipeBytesEither4 :: Property +pipeBytesEither4 = monadicIO $ run checkFailAction where action = do - S.drain $ - Proc.processBytes' + S.fold Fold.drain $ + Proc.pipeBytesEither interpreterFile [interpreterArg, executableFilePass] (S.nilM $ throwM (SimpleError failErrorMessage)) @@ -396,50 +397,50 @@ checkFailAction = catch action failAction -processChunks'1 :: Property -processChunks'1 = +pipeChunksEither1 :: Property +pipeChunksEither1 = forAll (listOf (choose(_a, _z))) $ \ls -> monadicIO $ do trBinary <- run $ which "tr" let inputStream = S.fromList ls - genStrm = AS.concat $ S.rights $ - Proc.processChunks' + genStrm = concatArr $ S.catRights $ + Proc.pipeChunksEither trBinary ["[a-z]", "[A-Z]"] - (AS.arraysOf arrayChunkSize inputStream) + (S.chunksOf arrayChunkSize inputStream) - charUpperStrm = S.map toUpper inputStream + charUpperStrm = fmap toUpper inputStream - genList <- run $ S.toList genStrm - charList <- run $ S.toList charUpperStrm + genList <- run $ S.fold Fold.toList genStrm + charList <- run $ S.fold Fold.toList charUpperStrm listEquals (==) genList charList -processChunks'2 :: Property -processChunks'2 = +pipeChunksEither2 :: Property +pipeChunksEither2 = forAll (listOf (choose(_a, _z))) $ \ls -> monadicIO $ do let inputStream = S.fromList ls - outStream = AS.concat $ S.lefts $ - Proc.processChunks' + outStream = concatArr $ S.catLefts $ + Proc.pipeChunksEither interpreterFile [interpreterArg, executableFile, "[a-z]", "[A-Z]"] - (AS.arraysOf arrayChunkSize inputStream) + (S.chunksOf arrayChunkSize inputStream) - charUpperStrm = S.map toUpper inputStream + charUpperStrm = fmap toUpper inputStream - genList <- run $ S.toList outStream - charList <- run $ S.toList charUpperStrm + genList <- run $ S.fold Fold.toList outStream + charList <- run $ S.fold Fold.toList charUpperStrm listEquals (==) genList charList -processChunks'3 :: Property -processChunks'3 = monadicIO $ run checkFailAction +pipeChunksEither3 :: Property +pipeChunksEither3 = monadicIO $ run checkFailAction where action = do - S.drain $ Proc.processChunks' + S.fold Fold.drain $ Proc.pipeChunksEither interpreterFile [interpreterArg, executableFileFail] S.nil return False @@ -448,13 +449,13 @@ checkFailAction = catch action failAction -processChunks'4 :: Property -processChunks'4 = monadicIO $ run checkFailAction +pipeChunksEither4 :: Property +pipeChunksEither4 = monadicIO $ run checkFailAction where action = do - S.drain $ - Proc.processChunks' + S.fold Fold.drain $ + Proc.pipeChunksEither interpreterFile [interpreterArg, executableFilePass] (S.nilM $ throwM (SimpleError failErrorMessage)) @@ -467,6 +468,15 @@ main :: IO () main = do + -- Nix does not have "/usr/bin/env", so the execution of test executables + -- fails. We can create the executables as temporary files to avoid this + -- issue. Even creating Haskell executables does not work because of + -- https://github.com/haskell/cabal/issues/7577 . + r <- doesFileExist "/usr/bin/env" + unless r $ do + putStrLn $ "/usr/bin/env does not exist, skipping tests." + exitSuccess + hspec $ do describe "Streamly.System.Process" $ do -- XXX Add a test for garbage collection case. Also check whether @@ -474,15 +484,15 @@ -- -- Keep the tests in dependency order so that we test the basic -- things first. - describe "processChunks'" $ do + describe "pipeChunksEither" $ do prop - "AS.concat $ processChunks' tr = map toUpper" - processChunks'1 + "concatArr $ pipeChunksEither tr = map toUpper" + pipeChunksEither1 prop - "error stream of processChunks' tr = map toUpper" - processChunks'2 - prop "processChunks' on failing executable" processChunks'3 - prop "processChunks' using error stream" processChunks'4 + "error stream of pipeChunksEither tr = map toUpper" + pipeChunksEither2 + prop "pipeChunksEither on failing executable" pipeChunksEither3 + prop "pipeChunksEither using error stream" pipeChunksEither4 describe "processChunks" $ do prop "consumeAllInput" processChunksConsumeAllInput @@ -491,24 +501,24 @@ prop "inputFailure" processChunksInputFailure -- based on processChunks - describe "processBytes'" $ do - prop "processBytes' tr = map toUpper" processBytes'1 + describe "pipeBytesEither" $ do + prop "pipeBytesEither tr = map toUpper" pipeBytesEither1 prop - "error stream of processBytes' tr = map toUpper" - processBytes'2 - prop "processBytes' on failing executable" processBytes'3 - prop "processBytes' using error stream" processBytes'4 + "error stream of pipeBytesEither tr = map toUpper" + pipeBytesEither2 + prop "pipeBytesEither on failing executable" pipeBytesEither3 + prop "pipeBytesEither using error stream" pipeBytesEither4 - describe "processBytes" $ do - prop "processBytes tr = map toUpper" processBytes1 - prop "processBytes on failing executable" processBytes2 - prop "processBytes using error stream" processBytes3 + describe "pipeBytes" $ do + prop "pipeBytes tr = map toUpper" pipeBytes1 + prop "pipeBytes on failing executable" pipeBytes2 + prop "pipeBytes using error stream" pipeBytes3 - -- Based on processBytes/Chunks - describe "toChunks'" $ do - prop "toChunks' cat = FH.toChunks" toChunks1 - prop "toChunks' on failing executable" toChunks2 + -- Based on pipeBytes/Chunks + describe "toChunksEither" $ do + prop "toChunksEither cat = FH.toChunks" toChunks1 + prop "toChunksEither on failing executable" toChunks2 - describe "toBytes'" $ do - prop "toBytes' cat = FH.toBytes" toBytes1 - prop "toBytes' on failing executable" toBytes2 + describe "toBytesEither" $ do + prop "toBytesEither cat = FH.toBytes" toBytes1 + prop "toBytesEither on failing executable" toBytes2