packages feed

streamly-process 0.1.0 → 0.2.0

raw patch · 10 files changed

+57/−105 lines, 10 filesdep ~streamlynew-uploader

Dependency ranges changed: streamly

Files

CHANGELOG.md view
@@ -1,5 +1,13 @@ # Changelog +## 0.2.0 (Dec 2021)++* The process in following APIs is now terminated with SIGTERM if an exception+  is thrown or if the process is garbage collected before it could terminate.+  - `toBytes`+  - `processChunks`+  - `processBytes`+ ## 0.1.0 (Jul 2021)  * Initial version.
src/Streamly/Internal/System/Process.hs view
@@ -103,20 +103,14 @@ 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 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.Data.Unfold as Unfold (either) 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@@ -129,43 +123,6 @@ -- >>> 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 -------------------------------------------------------------------------------@@ -283,15 +240,16 @@     case exitCode of         ExitSuccess -> return ()         ExitFailure code -> throwM $ ProcessFailure code+#endif  -- | 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 =>+cleanupException :: MonadIO m =>     (Maybe Handle, Maybe Handle, Maybe Handle, ProcessHandle) -> m ()-_cleanupException (Just stdinH, Just stdoutH, stderrMaybe, ph) = liftIO $ do+cleanupException (Just stdinH, Just stdoutH, stderrMaybe, ph) = liftIO $ do     -- Send a SIGTERM to the process     terminateProcess ph @@ -318,8 +276,7 @@             _ -> False      eatSIGPIPE e = unless (isSIGPIPE e) $ throwIO e-_cleanupException _ = error "cleanupProcess: Not reachable"-#endif+cleanupException _ = error "cleanupProcess: Not reachable"  -- | Creates a system process from an executable path and arguments. For the -- default attributes used to create the process see 'mkConfig'.@@ -370,9 +327,8 @@     -> [String]             -- ^ Arguments     -> t m a     -- ^ Output stream processChunksWithAction run modCfg path args =-    -- Stream.bracket'-    --      alloc cleanupNormal _cleanupException _cleanupException run-    Stream.bracket alloc cleanupNormal run+    Stream.bracket'+          alloc cleanupNormal cleanupException cleanupException run      where @@ -436,7 +392,7 @@ processBytes' path args input =     let input1 = ArrayStream.arraysOf defaultChunkSize input         output = processChunks' path args input1-     in unfoldManyEither Array.read output+     in Stream.unfoldMany (Unfold.either Array.read) output  {-# INLINE processChunksWith #-} processChunksWith ::@@ -464,8 +420,7 @@ -- 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.+-- 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.
streamly-process.cabal view
@@ -1,6 +1,6 @@ cabal-version:       2.2 name:                streamly-process-version:             0.1.0+version:             0.2.0 synopsis:            Use OS processes as stream transformation functions description:     Run operating system processes as stream source, sink or transformation@@ -15,19 +15,33 @@ 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+tested-with:         GHC==9.2.1+                   , 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+  test/data/failExec.bat+  test/data/failExec.sh+  test/data/passExec.bat+  test/data/passExec.sh+  test/data/writeTrToError.bat+  test/data/writeTrToError.sh+ source-repository head   type: git   location: https://github.com/composewell/streamly-process  flag fusion-plugin   description: Use fusion plugin for benchmarks and executables+  manual: True+  default: False  flag use-gauge   description: Use gauge instead of tasty-bench for benchmarking@@ -75,7 +89,7 @@       base              >= 4.8   && < 5     , process           >= 1.0   && < 1.7     -- Uses internal APIs-    , streamly          >= 0.8   && < 0.8.1+    , streamly          >= 0.8.1 && < 0.8.2     , exceptions        >= 0.8   && < 0.11   if !os(windows)     build-depends:@@ -102,7 +116,7 @@     , directory         >= 1.2.2 && < 1.4     , process           >= 1.0   && < 1.7     -- Uses internal APIs-    , streamly          >= 0.8   && < 0.8.1+    , streamly          >= 0.8.1 && < 0.8.2    if flag(fusion-plugin) && !impl(ghc < 8.6)     build-depends:@@ -133,4 +147,4 @@     , process           >= 1.0   && < 1.7     , QuickCheck        >= 2.10  && < 2.15     -- Uses internal APIs-    , streamly          >= 0.8   && < 0.8.1+    , streamly          >= 0.8.1 && < 0.8.2
test/Streamly/System/Process.hs view
@@ -13,7 +13,6 @@ 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@@ -78,61 +77,27 @@ 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"+executableFile = "./test/data/writeTrToError.bat" #else-executableFileContent =-    "tr [a-z] [A-Z] >&2"+executableFile = "./test/data/writeTrToError.sh" #endif  executableFileFail :: FilePath #if mingw32_HOST_OS == 1-executableFileFail = "./failExec.bat"+executableFileFail = "./test/data/failExec.bat" #else-executableFileFail = "./failExec.sh"+executableFileFail = "./test/data/failExec.sh" #endif -executableFileFailContent :: String-executableFileFailContent =-    "exit 1"- executableFilePass :: FilePath #if mingw32_HOST_OS == 1-executableFilePass = "./passExec.bat"+executableFilePass = "./test/data/passExec.bat" #else-executableFilePass = "./passExec.sh"+executableFilePass = "./test/data/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@@ -483,7 +448,6 @@  main :: IO () main = do-    createExecutables     hspec $ do         describe "Streamly.System.Process" $ do             -- XXX Add a test for garbage collection case. Also check whether@@ -529,5 +493,3 @@             describe "toBytes'" $ do                 prop "toBytes' cat = FH.toBytes" toBytes1                 prop "toBytes' on failing executable" toBytes2--    removeExecutables
+ test/data/failExec.bat view
@@ -0,0 +1,1 @@+exit 1
+ test/data/failExec.sh view
@@ -0,0 +1,3 @@+#!/usr/bin/env sh++exit 1
+ test/data/passExec.bat view
@@ -0,0 +1,1 @@+exit 0
+ test/data/passExec.sh view
@@ -0,0 +1,3 @@+#!/usr/bin/env sh++exit 0
+ test/data/writeTrToError.bat view
@@ -0,0 +1,2 @@+@echo off
+tr [a-z] [A-Z] >&2
+ test/data/writeTrToError.sh view
@@ -0,0 +1,3 @@+#!/usr/bin/env sh++tr [a-z] [A-Z] >&2