process-streaming 0.5.0.0 → 0.5.0.1
raw patch · 3 files changed
+84/−104 lines, 3 files
Files
- CHANGELOG +8/−9
- process-streaming.cabal +2/−2
- src/System/Process/Streaming.hs +74/−93
CHANGELOG view
@@ -1,19 +1,18 @@+0.5.0.1 +------- + +- Changes in the internals to reduce the number of spawned threads. + 0.5.0.0 ------- - Now the constructors for "PipingPolicy" take "Siphons" directly, instead of -continuation functions. - + continuation functions. - Removed "separated" and "combined" functions, added new "PipingPolicy" -constructors in their place. - + constructors in their place. - Removed "LeftoverPolicy", its function is now performed by Siphons. - -- Removed "surely", "safely", "monoidally" functions with confusing -signatures. - +- Removed "surely", "safely", "monoidally" functions with confusing signatures. - Removed all occurrences of unbounded buffers in the code. - - Implemented support for branching pipelines of processes. 0.3.0.0
process-streaming.cabal view
@@ -1,5 +1,5 @@ name: process-streaming -version: 0.5.0.0 +version: 0.5.0.1 license: BSD3 license-file: LICENSE data-files: @@ -9,7 +9,7 @@ build-type: Simple cabal-version: >= 1.10 Synopsis: Streaming interface to system processes. -Description: Concurrent, buffered, streaming access to the input and outputs of system processes. +Description: Concurrent, streaming access to the input and outputs of system processes. Extra-Source-Files: README.md
src/System/Process/Streaming.hs view
@@ -202,12 +202,12 @@ data PipingPolicy e a = PPNone a | PPOutput (Producer ByteString IO () -> IO (Either e a)) - | PPError (Producer ByteString IO () -> IO (Either e a)) + | PPError (Producer ByteString IO () -> IO (Either e a)) | PPOutputError ((Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a)) | PPInput ((Consumer ByteString IO (), IO ()) -> IO (Either e a)) | PPInputOutput ((Consumer ByteString IO (), IO (),Producer ByteString IO ()) -> IO (Either e a)) - | PPInputError ((Consumer ByteString IO (), IO (), Producer ByteString IO ()) -> IO (Either e a)) - | PPInputOutputError ((Consumer ByteString IO (),IO (),Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a)) + | PPInputError ((Consumer ByteString IO (), IO (), Producer ByteString IO ()) -> IO (Either e a)) + | PPInputOutputError ((Consumer ByteString IO (),IO (),Producer ByteString IO (),Producer ByteString IO ()) -> IO (Either e a)) deriving (Functor) instance Bifunctor PipingPolicy where @@ -231,26 +231,26 @@ Pipe @stdout@. -} pipeo :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a -pipeo (runSiphon -> siphonout) = PPOutput $ siphonout +pipeo (halting -> siphonout) = PPOutput $ siphonout {-| Pipe @stderr@. -} pipee :: (Show e,Typeable e) => Siphon ByteString e a -> PipingPolicy e a -pipee (runSiphon -> siphonout) = PPError $ siphonout +pipee (halting -> siphonout) = PPError $ siphonout {-| Pipe @stdout@ and @stderr@. -} pipeoe :: (Show e,Typeable e) => Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (a,b) -pipeoe (runSiphon -> siphonout) (runSiphon -> siphonerr) = +pipeoe (halting -> siphonout) (halting -> siphonerr) = PPOutputError $ uncurry $ separated siphonout siphonerr {-| Pipe @stdout@ and @stderr@ and consume them combined as 'Text'. -} pipeoec :: (Show e,Typeable e) => LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e a -pipeoec policy1 policy2 (runSiphon -> siphon) = +pipeoec policy1 policy2 (halting -> siphon) = PPOutputError $ uncurry $ combined policy1 policy2 siphon {-| @@ -264,7 +264,7 @@ -} pipeio :: (Show e, Typeable e) => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a) -pipeio (Pump feeder) (runSiphon -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) -> +pipeio (Pump feeder) (halting -> siphonout) = PPInputOutput $ \(consumer,cleanup,producer) -> (conceit (feeder consumer `finally` cleanup) (siphonout producer)) {-| @@ -272,7 +272,7 @@ -} pipeie :: (Show e, Typeable e) => Pump ByteString e i -> Siphon ByteString e a -> PipingPolicy e (i,a) -pipeie (Pump feeder) (runSiphon -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) -> +pipeie (Pump feeder) (halting -> siphonerr) = PPInputError $ \(consumer,cleanup,producer) -> (conceit (feeder consumer `finally` cleanup) (siphonerr producer)) {-| @@ -280,7 +280,7 @@ -} pipeioe :: (Show e, Typeable e) => Pump ByteString e i -> Siphon ByteString e a -> Siphon ByteString e b -> PipingPolicy e (i,a,b) -pipeioe (Pump feeder) (runSiphon -> siphonout) (runSiphon -> siphonerr) = fmap flattenTuple $ PPInputOutputError $ +pipeioe (Pump feeder) (halting -> siphonout) (halting -> siphonerr) = fmap flattenTuple $ PPInputOutputError $ \(consumer,cleanup,outprod,errprod) -> (conceit (feeder consumer `finally` cleanup) (separated siphonout siphonerr outprod errprod)) @@ -292,7 +292,7 @@ -} pipeioec :: (Show e, Typeable e) => Pump ByteString e i -> LinePolicy e -> LinePolicy e -> Siphon Text e a -> PipingPolicy e (i,a) -pipeioec (Pump feeder) policy1 policy2 (runSiphon -> siphon) = PPInputOutputError $ +pipeioec (Pump feeder) policy1 policy2 (halting -> siphon) = PPInputOutputError $ \(consumer,cleanup,outprod,errprod) -> (conceit (feeder consumer `finally` cleanup) (combined policy1 policy2 siphon outprod errprod)) @@ -341,7 +341,7 @@ . decoder $ producer viewLines = getConst . T.lines Const - teardown freeLines >>= runSiphon lopo + teardown freeLines >>= halting lopo -- http://unix.stackexchange.com/questions/114182/can-redirecting-stdout-and-stderr-to-the-same-file-mangle-lines here combined :: (Show e, Typeable e) @@ -369,14 +369,14 @@ join $ withMVar mvar $ \output -> do runEffect $ (textProducer <* P.yield (singleton '\n')) >-> (toOutput output >> P.drain) -fromProducer :: Producer b IO () -> Pump b e () -fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (producer >-> consumer) +fromProducer :: Producer b IO r -> Pump b e () +fromProducer producer = Pump $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) -fromSafeProducer :: Producer b (SafeT IO) () -> Pump b e () -fromSafeProducer producer = Pump $ safely $ \consumer -> fmap pure $ runEffect (producer >-> consumer) +fromSafeProducer :: Producer b (SafeT IO) r -> Pump b e () +fromSafeProducer producer = Pump $ safely $ \consumer -> fmap pure $ runEffect (mute producer >-> consumer) -fromFallibleProducer :: Producer b (ExceptT e IO) () -> Pump b e () -fromFallibleProducer producer = Pump $ \consumer -> runExceptT $ runEffect (producer >-> hoist lift consumer) +fromFallibleProducer :: Producer b (ExceptT e IO) r -> Pump b e () +fromFallibleProducer producer = Pump $ \consumer -> runExceptT $ runEffect (mute producer >-> hoist lift consumer) {-| Useful when we want to plug in a handler that does its work in the 'SafeT' @@ -394,7 +394,7 @@ type DecodingFunction bytes text = forall r. Producer bytes IO r -> Producer text IO (Producer bytes IO r) {-| - Constructs a 'Siphon' that works on undecoded values out of a 'Siphon' that + Constructs a 'Siphon' that works on encoded values out of a 'Siphon' that works on decoded values. The two first arguments are a decoding function and a 'Siphon' that @@ -406,23 +406,12 @@ -> Siphon bytes e (a -> b) -> Siphon text e a -> Siphon bytes e b -encoded decoder policy activity = Halting $ \producer -> buffer policy activity $ decoder producer - where - buffer :: (Show e, Typeable e) - => Siphon bytes e (a -> b) - -> Siphon text e a - -> Producer text IO (Producer bytes IO ()) -> IO (Either e b) - buffer policy activity producer = do - (outbox,inbox,seal) <- spawn' Single - r <- conceit - (do feeding <- async $ runEffect $ - producer >-> (toOutput outbox >> P.drain) - Right <$> wait feeding `finally` atomically seal - ) - (runSiphon activity (fromInput inbox) `finally` atomically seal) - case r of - Left e -> return $ Left e - Right (leftovers,a) -> runSiphon (fmap ($a) policy) leftovers +encoded decoder policy activity = + Unhalting $ \producer -> + runExceptT $ do + (a,leftovers) <- ExceptT $ unhalting activity $ decoder producer + (f,r) <- ExceptT $ unhalting policy leftovers + pure (f a,r) data WrappedError e = WrappedError e deriving (Show, Typeable) @@ -501,7 +490,7 @@ (fromInput inbox1 >> fromInput inbox2) >-> consumer) `finally` atomically seal1 `finally` atomically seal2 - return $ pure () + runExceptT $ pure () ) instance (Show e, Typeable e, Monoid a) => Monoid (Pump b e a) where @@ -534,53 +523,65 @@ pure = Trivial s1 <*> s2 = case (s1,s2) of - (Trivial f, s2') -> fmap f s2' - (s1', Trivial a) -> fmap ($ a) s1' - (Halting fs, Halting as) -> fork fs as - (Halting fs, Unhalting as) -> fork fs (halting as) - (Unhalting fs, Halting as) -> fork (halting fs) as - (Unhalting fs, Unhalting as) -> fork (halting fs) (halting as) + (Trivial f,_) -> fmap f s2 + (_,Trivial a) -> fmap ($ a) s1 + (_,_) -> bifurcate (halting s1) (halting s2) where - fork fs as = - Halting $ \producer -> do + bifurcate fs as = + Unhalting $ \producer -> do (outbox1,inbox1,seal1) <- spawn' Single (outbox2,inbox2,seal2) <- spawn' Single runConceit $ - Conceit (do - -- mmm who cancels these asyncs ?? - feeding <- async $ runEffect $ - producer >-> P.tee (toOutput outbox1 >> P.drain) - >-> (toOutput outbox2 >> P.drain) - -- is these async neccessary ?? - sealing <- async $ wait feeding `finally` atomically seal1 - `finally` atomically seal2 - return $ pure () - ) - *> + (,) + <$> Conceit (fmap (uncurry ($)) <$> conceit ((fs $ fromInput inbox1) `finally` atomically seal1) ((as $ fromInput inbox2) `finally` atomically seal2) ) + <*> + Conceit ((fmap pure $ runEffect $ + producer >-> P.tee (toOutput outbox1 >> P.drain) + >-> (toOutput outbox2 >> P.drain)) + `finally` atomically seal1 `finally` atomically seal2 + ) +halting :: (Show e, Typeable e) => Siphon b e a -> Producer b IO () -> IO (Either e a) +halting s = case s of + a@(Trivial _) -> halting $ Unhalting $ unhalting a + Unhalting u -> \producer -> liftM (fmap fst) $ u producer + Halting h -> h -halting :: (forall r. Producer b IO r -> IO (Either e (a,r))) - -> (Producer b IO () -> IO (Either e a)) -halting polyfunc = \producer -> - liftM (fmap fst) $ polyfunc producer +unhalting :: (Show e, Typeable e) => Siphon b e a -> Producer b IO r -> IO (Either e (a,r)) +unhalting s = case s of + Trivial a -> \producer -> do + r <- (runEffect $ producer >-> P.drain) + pure . pure $ (a,r) + Unhalting u -> u + Halting activity -> \producer -> do + (outbox,inbox,seal) <- spawn' Single + runConceit $ + (,) + <$> + Conceit (activity (fromInput inbox) `finally` atomically seal) + <*> + Conceit ((fmap pure $ runEffect $ + producer >-> (toOutput outbox >> P.drain)) + `finally` atomically seal + ) instance (Show e, Typeable e, Monoid a) => Monoid (Siphon b e a) where mempty = pure mempty mappend s1 s2 = (<>) <$> s1 <*> s2 -fromConsumer :: Consumer b IO () -> Siphon b e () -fromConsumer consumer = siphon $ \producer -> fmap pure $ runEffect $ producer >-> consumer +fromConsumer :: Consumer b IO r -> Siphon b e () +fromConsumer consumer = siphon $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer -fromSafeConsumer :: Consumer b (SafeT IO) () -> Siphon b e () -fromSafeConsumer consumer = siphon $ safely $ \producer -> fmap pure $ runEffect $ producer >-> consumer +fromSafeConsumer :: Consumer b (SafeT IO) r -> Siphon b e () +fromSafeConsumer consumer = siphon $ safely $ \producer -> fmap pure $ runEffect $ producer >-> mute consumer -fromFallibleConsumer :: Consumer b (ExceptT e IO) () -> Siphon b e () -fromFallibleConsumer consumer = siphon $ \producer -> runExceptT $ runEffect (hoist lift producer >-> consumer) +fromFallibleConsumer :: Consumer b (ExceptT e IO) r -> Siphon b e () +fromFallibleConsumer consumer = siphon $ \producer -> runExceptT $ runEffect (hoist lift producer >-> mute consumer) {-| Turn a 'Parser' from @pipes-parse@ into a 'Sihpon'. @@ -588,14 +589,6 @@ fromParser :: Parser b IO (Either e a) -> Siphon b e a fromParser parser = siphon $ Pipes.Parse.evalStateT parser -runSiphon :: (Show e, Typeable e) - => Siphon b e a - -> (Producer b IO () -> IO (Either e a)) -runSiphon s = case s of - Trivial a -> \producer -> (runEffect $ producer >-> P.drain) >> (pure . pure $ a) - Unhalting u -> halting u -- no need to re-buffer - Halting h -> buffer_ h - {-| Builds a 'Siphon' out of a computation that does something with a 'Producer', but may fail with an error of type @e@. @@ -615,19 +608,6 @@ siphon' :: (forall r. Producer b IO r -> IO (Either e (a,r))) -> Siphon b e a siphon' = Unhalting -buffer_ :: (Show e, Typeable e) - => (Producer b IO () -> IO (Either e a)) - -> Producer b IO () -> IO (Either e a) -buffer_ activity producer = do - (outbox,inbox,seal) <- spawn' Single - runConceit $ - Conceit (do feeding <- async $ runEffect $ - producer >-> (toOutput outbox >> P.drain) - Right <$> wait feeding `finally` atomically seal - ) - *> - Conceit (activity (fromInput inbox) `finally` atomically seal) - fromFold :: (Producer b IO () -> IO a) -> Siphon b e a fromFold aFold = siphon $ fmap (fmap pure) $ aFold @@ -645,10 +625,10 @@ 'Producer' produces anything. -} unwanted :: a -> Siphon b b a -unwanted a = Halting $ \producer -> do - r <- next producer - return $ case r of - Left () -> Right a +unwanted a = Unhalting $ \producer -> do + n <- next producer + return $ case n of + Left r -> Right (a,r) Right (b,_) -> Left b executePipeline :: PipingPolicy Void a -> CreatePipeline Void -> IO a @@ -781,8 +761,6 @@ `finally` atomically iseal `finally` atomically oseal `finally` atomically eseal ) where - mute = fmap (const ()) - errorSiphonUTF8 :: MVar (Output ByteString) -> LinePolicy e -> Siphon ByteString e () errorSiphonUTF8 mvar (LinePolicy fun) = Halting $ fun iterTLines where @@ -792,6 +770,9 @@ runEffect $ (textProducer <* P.yield (singleton '\n')) >-> P.map Data.Text.Encoding.encodeUtf8 >-> (toOutput output >> P.drain) + +mute :: Functor f => f a -> f () +mute = fmap (const ()) {-| An individual stage in a process pipeline.