streaming 0.1.2.0 → 0.1.2.2
raw patch · 4 files changed
+272/−155 lines, 4 files
Files
- README.md +87/−35
- Streaming/Internal.hs +45/−45
- Streaming/Prelude.hs +123/−68
- streaming.cabal +17/−7
README.md view
@@ -1,6 +1,16 @@ streaming ========= +1. The freely generated stream on a streamable functor+2. A freely generated stream of individual Haskell values is a Producer, Generator or Source+3. `Streaming.Prelude`+4. Mother's `Prelude` v. `Streaming.Prelude`+5. How come there's not one of those fancy "ListT done right" implementations in here?+6. Didn't I hear that free monads are a dog from the point of view of efficiency?+7. Interoperation with the streaming-io libraries+8. Where can I find examples of use?+9. Problems+10. Implementation and benchmarking notes `Stream` can be used wherever [FreeT](https://hackage.haskell.org/package/free-4.12.1/docs/Control-Monad-Trans-Free.html) is used. The compiler's standard range of optimizations work better for operations written in terms of `Stream`. `FreeT f m r` and `Stream f m r` are of course extremely general, and many functor-general combinators are exported by the general module `Streaming`. @@ -9,9 +19,11 @@ The abstraction is inevitable, though there are many ways of writing it. Once one possesses it, though, one is already in possession of an elementary streaming library, since `Stream ((,)a) m r` or its equivalent is the type of a producer, generator or source. I try to argue for this more elaborately below, bringing it into connection with the standard streaming io libraries. -The freely generated stream on a streamable functor-----------------------------------------------------+1. The freely generated stream on a streamable functor+------------------------------------------------------- +(This section is a rather abstract defense of the inevitability of the leading type we are discussing, `Stream f m r` ; it may be well to skip to the next section.)+ As soon as you consider the idea of an effectful stream of any kind whatsoever, for example, a stream of bytes from a handle, however constituted, you will inevitably be forced to contemplate the idea of a streaming *succession* of *just such streams*. Thus, for example, however you imagine your bytes streaming from a handle, you will want to consider a *succession* of *such streams* divided on newlines. @@ -55,15 +67,15 @@ splitter :: S r -> S (S r) -or, where the underlying form of effect is explicit+or splitter :: S m r -> S m (S m r) -Now we can express what we meant by 'doing one thing with the first half and another with the second', we were thinking of applying some sort of polymorphic folds, maybe with types like+Now we can express what we meant by 'doing one thing with the first half and another with the second': we were thinking of applying some sort of polymorphic folds, maybe with types like folder :: S m x -> m (a, x) -then we have+Then we would have folder . splitter :: S m r -> m (a, S m r) @@ -85,8 +97,8 @@ General combinators for working with this idea of succession __irrespective of the form of succession__ are contained in the module `Stream`. They can be used, or example, to organize a succession of io-streams `Generator`s or pipes `Producer`s or the effectful bytestreams of the [streaming-bytestring](https://hackage.haskell.org/package/streaming-bytestring) library, or whatever stream-form you can express in a Haskell functor. -A freely generated stream of individual Haskell values is a Producer, Generator or Source----------------------------------------------------------------------------------------------------+2. A freely generated stream of individual Haskell values is a Producer, Generator or Source+------------------------------------------------------------------------------------------------------ But, of course, as soon as you grasp the general form of *succession*, you are already in possession of the most basic concrete form: a simple *succession of individual Haskell values* one after another. This is just `Stream ((,) a) m r`. Here we prefer `Stream (Of a) m r`, strictifying the left element of the pair with @@ -113,8 +125,8 @@ machines: SourceT m a (= forall k. MachineT m k a) streaming: Stream (Of a) m () -`Streaming.Prelude`--------------------+3. `Streaming.Prelude`+---------------------- `Streaming.Prelude` closely follows `Pipes.Prelude`. But since it restricts itself to use only of the general idea of streaming, it cleverly *omits the pipes*: @@ -135,8 +147,8 @@ Somehow, we didn't even need a four-character operator for that, nor advice about best practices! - just ordinary Haskell common sense. -Mother's `Prelude` v. `Streaming.Prelude`------------------------------------------+4. Mother's `Prelude` v. `Streaming.Prelude`+-------------------------------------------- The effort of `Streaming.Prelude` is to leverage the intuition the user has acquired in mastering `Prelude` and `Data.List` and to elevate her understanding into a general comprehension of effectful streaming transformations. Unsurprisingly, it takes longer to type out the signatures. It cannot be emphasized enough, thought, that *the transpositions are totally mechanical*: @@ -153,8 +165,8 @@ It is easy to prove that *resistance to these types is resistance to effectful streaming itself*. I will labor this point a bit more below, but you can also find it developed, with greater skill, in the documentation for the pipes libraries. -How come there's not one of those fancy "ListT done right" implementations in here?------------------------------------------------------------------------------------+5. How come there's not one of those fancy "ListT done right" implementations in here?+--------------------------------------------------------------------------------------- The use of the final return value appears to be a complication, but in fact it is essentially contained in the idea of effectful streaming. This is why this library does not export a \_ListT done right/, which would be simple enough - following `pipes`, as usual: @@ -169,8 +181,8 @@ Note similarly that you can write a certain kind of [take](http://hackage.haskell.org/package/machines-0.5.1/docs/Data-Machine-Process.html#v:taking) and [drop](http://hackage.haskell.org/package/machines-0.5.1/docs/Data-Machine-Process.html#v:dropping) with the `machines` library - as you can even with a "`ListT` done right". But I wish you luck writing `splitAt`! Similarly you can write a [getContents](http://hackage.haskell.org/package/machines-io-0.2.0.6/docs/System-IO-Machine.html); but I wish you luck dividing the resulting bytestream on its lines. This is - as usual! - because the library was not written with the general concept of effectful succession or streaming in view. Materials for sinking some elements of a stream in one way, and others in other ways - copying each line to a different file, as it might be, but without accumulation - are documented within. So are are myriad other elementary operations of streaming io. -Didn't I hear that free monads are a dog from the point of view of efficiency?-------------------------------------------------------------------------------+6. Didn't I hear that free monads are a dog from the point of view of efficiency?+---------------------------------------------------------------------------------- We noted above that if we instantiate `Stream f m r` to `Stream ((,) a) m r` or the like, we get the standard idea of a producer or generator. If it is instantiated to `Stream f Identity m r` then we have the standard \_free monad construction/. This construction is subject to certain familiar objections from an efficiency perspective; efforts have been made to substitute exotic cps-ed implementations and so forth. It is an interesting topic. @@ -235,8 +247,8 @@ With `sequence` and `traverse`, we accumulate a pure succession of pure values from a pure succession of monadic values. Why bother if you have intrinsically monadic conception of succession or traversal? `Stream f m r` gives you an immense body of such structures and a simple discipline for working with them. Spinkle `id` freely though your program, under various names, if you get homesick for `sequence` and company. -Interoperation with the streaming-io libraries-----------------------------------------------+7. Interoperation with the streaming-io libraries+-------------------------------------------------- The simplest form of interoperation with [pipes](http://hackage.haskell.org/package/pipes) is accomplished with this isomorphism: @@ -259,26 +271,27 @@ Free.iterTM Stream.wrap :: FreeT f m a -> Stream f m a Stream.iterTM Free.wrap :: Stream f m a -> FreeT f m a -Where can I find examples of use?----------------------------------+8. Where can I find examples of use?+------------------------------------- -For some simple ghci examples, see the commentary throughout the Prelude module. For slightly more advanced usage see the commentary in the haddocks of [streaming-bytestring](https://hackage.haskell.org/package/streaming-bytestring) and e.g. [these replicas](https://gist.github.com/michaelt/6c6843e6dd8030e95d58) of shell-like programs from the io-streams tutorial. Here's a simple [streaming GET request](https://gist.github.com/michaelt/2dcea1ba32562c091357) with intrinsically streaming byte streams.+For some simple ghci examples, see the commentary throughout the Prelude module. For slightly more advanced usage see the commentary in the haddocks of [streaming-bytestring](https://hackage.haskell.org/package/streaming-bytestring) and e.g. [these replicas](https://gist.github.com/michaelt/6c6843e6dd8030e95d58) of shell-like programs from the io-streams tutorial. Here's a simple [streaming GET request](https://gist.github.com/michaelt/2dcea1ba32562c091357) with intrinsically streaming byte streams. Here is a comically simple ['high - low' game](https://gist.github.com/michaelt/242f6a23267707ad29e9) -Problems---------+9. Problems+------------ Questions about this library can be put as issues through the github site or on the [pipes mailing list](https://groups.google.com/forum/#!forum/haskell-pipes). (This library understands itself as part of the pipes "ecosystem.") -* * * * * -implementation notes +10. Implementation and benchmarking notes+------------------------------------------+ This library defines an optimized `FreeT` with an eye to use with streaming libraries, namely: data Stream f m r = Return r | Step !(f (Stream f m r))- | Delay (m (Stream f m r))+ | Effect (m (Stream f m r)) in place of the standard `FreeT` that we find in the `free` library, which is approximately: @@ -286,8 +299,6 @@ Rather than wrapping each step in a monadic 'layer', such a layer is put alongside separate 'pure' constructors for a functor 'layer' and a final return value. The maneuver is very friendly to the compiler, but requires a bit of subtlety to protect a sound monad instance. Just such an optimization is adopted internally by the `pipes` library. As in `pipes`, the constructors are here left in an `Internal` module; the main `Streaming` module exporting the type itself and various operations and instances. -There is also a still-incomplete `Prelude` of functions, some `FreeT` or `Stream` - general, some involving the functor `((,) a)` here called `Of a`. (`Stream (Of a) m r` like `FreeT ((,) a) m r` is equivalent to the `pipes` `Producer a m r` type. Similarly, `Stream (Of a) m ()` and `FreeT ((,) a) m ()` are possible implementations of `ListT done right`.- I ran a simple [benchmark](https://gist.github.com/michaelt/7f89dc8b366b30bb6acc) (adjusting a [script](https://github.com/jwiegley/streaming-tests) of John Weigly) using a very simple composition of functions: toList @@ -298,18 +309,59 @@ . filter even . each +as it interpreted by various libraries - `streaming`, `conduit`, (Weigley's) `simple-conduit`, `io-streams` and `machines`.+ The the results were fairly pleasing: - benchmarking basic/streaming- time 84.50 ms (79.81 ms .. 87.90 ms)+ benchmarking basic/stream+ time 85.45 ms (81.63 ms .. 89.32 ms)+ 0.994 R² (0.982 R² .. 0.999 R²)+ mean 86.53 ms (84.16 ms .. 90.51 ms)+ std dev 4.987 ms (2.301 ms .. 7.906 ms)+ variance introduced by outliers: 18% (moderately inflated) - benchmarking basic/iostreams- time 266.2 ms (235.6 ms .. 292.0 ms)+ benchmarking basic/conduit+ time 101.3 ms (88.77 ms .. 111.3 ms)+ 0.976 R² (0.911 R² .. 0.996 R²)+ mean 95.56 ms (84.90 ms .. 103.6 ms)+ std dev 13.76 ms (8.210 ms .. 21.79 ms)+ variance introduced by outliers: 43% (moderately inflated) + benchmarking basic/simple-conduit+ time 199.2 ms (174.1 ms .. 215.6 ms)+ 0.993 R² (0.978 R² .. 1.000 R²)+ mean 198.4 ms (190.0 ms .. 202.2 ms)+ std dev 7.091 ms (1.565 ms .. 10.000 ms)+ variance introduced by outliers: 14% (moderately inflated)+ benchmarking basic/pipes- time 232.0 ms (206.6 ms .. 246.7 ms)+ time 211.7 ms (180.8 ms .. 232.7 ms)+ 0.991 R² (0.974 R² .. 1.000 R²)+ mean 207.7 ms (199.1 ms .. 218.7 ms)+ std dev 12.34 ms (5.989 ms .. 17.67 ms)+ variance introduced by outliers: 15% (moderately inflated) - benchmarking basic/conduit- time 102.3 ms (96.24 ms .. 110.0 ms)+ benchmarking basic/data-list+ time 202.7 ms (186.5 ms .. 225.5 ms)+ 0.990 R² (0.970 R² .. 1.000 R²)+ mean 199.3 ms (188.4 ms .. 207.4 ms)+ std dev 11.67 ms (6.966 ms .. 15.11 ms)+ variance introduced by outliers: 15% (moderately inflated) -This sequence of pre-packaged combinators is, I think, very friendly to the more recent conduit fusion framework. The framework of course doesn't apply to user-defined operations, where we should expect times like those shown for pipes. Since the combinators from `streaming` is defined with naive recursion, more or less as the user might, we have reason to think the result is characteristic, but much more benchmarking is needed before anything can be said with certainty. The labor of constructor-hiding may turn up some further difficulty.+ benchmarking basic/iostreams+ time 265.7 ms (247.2 ms .. 284.8 ms)+ 0.997 R² (0.990 R² .. 1.000 R²)+ mean 265.6 ms (261.9 ms .. 272.8 ms)+ std dev 7.094 ms (146.8 μs .. 8.387 ms)+ variance introduced by outliers: 16% (moderately inflated)++ benchmarking basic/machines+ time 1.123 s (NaN s .. 1.206 s)+ 0.999 R² (0.999 R² .. 1.000 R²)+ mean 1.134 s (1.114 s .. 1.145 s)+ std dev 17.29 ms (0.0 s .. 19.07 ms)+ variance introduced by outliers: 19% (moderately inflated)++++This sequence of pre-packaged combinators is, I think, as friendly as it could possibly be to the more recent conduit fusion framework. That framework of course doesn't apply to user-defined operations; there we should expect times like those shown for pipes. Since the combinators from `streaming` are defined with naive recursion, more or less as the user might, we have reason to think this result is characteristic, but much more benchmarking is needed before anything can be said with certainty.
Streaming/Internal.hs view
@@ -80,7 +80,7 @@ succession of steps, where the form of the steps or 'commands' is specified by the first (functor) parameter. -> data Stream f m r = Step !(f (Stream f m r)) | Delay (m (Stream f m r)) | Return r+> data Stream f m r = Step !(f (Stream f m r)) | Effect (m (Stream f m r)) | Return r The /producer/ concept uses the simple functor @ (a,_) @ \- or the stricter @ Of a _ @. Then the news at each step or layer is just: an individual item of type @a@. @@ -96,7 +96,7 @@ The constructors are exported by the 'Internal' module. -} data Stream f m r = Step !(f (Stream f m r))- | Delay (m (Stream f m r))+ | Effect (m (Stream f m r)) | Return r #if __GLASGOW_HASKELL__ >= 710 deriving (Typeable)@@ -113,13 +113,13 @@ fmap f = loop where loop stream = case stream of Return r -> Return (f r)- Delay m -> Delay (do {stream' <- m; return (loop stream')})+ Effect m -> Effect (do {stream' <- m; return (loop stream')}) Step f -> Step (fmap loop f) {-# INLINABLE fmap #-} a <$ stream0 = loop stream0 where loop stream = case stream of Return r -> Return a- Delay m -> Delay (do {stream' <- m; return (loop stream')})+ Effect m -> Effect (do {stream' <- m; return (loop stream')}) Step f -> Step (fmap loop f) {-# INLINABLE (<$) #-} @@ -129,7 +129,7 @@ stream1 >> stream2 = loop stream1 where loop stream = case stream of Return _ -> stream2- Delay m -> Delay (liftM loop m)+ Effect m -> Effect (liftM loop m) Step f -> Step (fmap loop f) {-# INLINABLE (>>) #-} (>>=) = _bind@@ -137,7 +137,7 @@ -- loop stream where -- loop stream0 = case stream0 of -- Step fstr -> Step (fmap loop fstr)- -- Delay m -> Delay (liftM loop m)+ -- Effect m -> Effect (liftM loop m) -- Return r -> f r -- {-# INLINABLE (>>=) #-} @@ -152,14 +152,14 @@ _bind p0 f = go p0 where go p = case p of Step fstr -> Step (fmap go fstr)- Delay m -> Delay (m >>= \s -> return (go s))+ Effect m -> Effect (m >>= \s -> return (go s)) Return r -> f r {-# RULES "_bind (Step fstr) f" forall fstr f . _bind (Step fstr) f = Step (fmap (\p -> _bind p f) fstr);- "_bind (Delay m) f" forall m f .- _bind (Delay m) f = Delay (m >>= \p -> return (_bind p f));+ "_bind (Effect m) f" forall m f .+ _bind (Effect m) f = Effect (m >>= \p -> return (_bind p f)); "_bind (Return r) f" forall r f . _bind (Return r) f = f r; #-}@@ -174,25 +174,25 @@ -- stra0 *> strb = loop stra0 where -- loop stra = case stra of -- Return _ -> strb- -- Delay m -> Delay (do {stra' <- m ; return (stra' *> strb)})+ -- Effect m -> Effect (do {stra' <- m ; return (stra' *> strb)}) -- Step fstr -> Step (fmap (*> strb) fstr) -- {-# INLINABLE (*>) #-} -- stra <* strb0 = loop strb0 where -- loop strb = case strb of -- Return _ -> stra- -- Delay m -> Delay (do {strb' <- m ; return (stra <* strb')})+ -- Effect m -> Effect (do {strb' <- m ; return (stra <* strb')}) -- Step fstr -> Step (fmap (stra <*) fstr) -- {-# INLINABLE (<*) #-} -- instance Functor f => MonadTrans (Stream f) where- lift = Delay . liftM Return+ lift = Effect . liftM Return {-# INLINE lift #-} instance Functor f => MFunctor (Stream f) where hoist trans = loop . unexposed where loop stream = case stream of Return r -> Return r- Delay m -> Delay (trans (liftM loop m))+ Effect m -> Effect (trans (liftM loop m)) Step f -> Step (fmap loop f) {-# INLINABLE hoist #-} @@ -200,12 +200,12 @@ embed phi = loop where loop stream = case stream of Return r -> Return r- Delay m -> phi m >>= loop+ Effect m -> phi m >>= loop Step f -> Step (fmap loop f) {-# INLINABLE embed #-} instance (MonadIO m, Functor f) => MonadIO (Stream f m) where- liftIO = Delay . liftM Return . liftIO+ liftIO = Effect . liftM Return . liftIO {-# INLINE liftIO #-} @@ -229,7 +229,7 @@ destroy stream0 construct effect done = loop (unexposed stream0) where loop stream = case stream of Return r -> done r- Delay m -> effect (liftM loop m)+ Effect m -> effect (liftM loop m) Step fs -> construct (fmap loop fs) {-# INLINABLE destroy #-} @@ -268,7 +268,7 @@ -- | Reflect a church-encoded stream; cp. @GHC.Exts.build@ construct :: (forall b . (f b -> b) -> (m b -> b) -> (r -> b) -> b) -> Stream f m r-construct = \phi -> phi Step Delay Return+construct = \phi -> phi Step Effect Return {-# INLINE construct #-} @@ -284,7 +284,7 @@ inspect = loop where loop stream = case stream of Return r -> return (Left r)- Delay m -> m >>= loop+ Effect m -> m >>= loop Step fs -> return (Right fs) {-# INLINABLE inspect #-} @@ -300,7 +300,7 @@ unfold :: (Monad m, Functor f) => (s -> m (Either r (f s))) -> s -> Stream f m r unfold step = loop where- loop s0 = Delay $ do + loop s0 = Effect $ do e <- step s0 case e of Left r -> return (Return r)@@ -314,7 +314,7 @@ maps phi = loop where loop stream = case stream of Return r -> Return r- Delay m -> Delay (liftM loop m)+ Effect m -> Effect (liftM loop m) Step f -> Step (phi (fmap loop f)) {-# INLINABLE maps #-} @@ -337,7 +337,7 @@ -- mapsNT (NT phi) = loop where -- loop stream = case stream of -- Return r -> Return r--- Delay m -> Delay (liftM loop m)+-- Effect m -> Effect (liftM loop m) -- Step f -> Step (phi (fmap loop f)) {- | Map layers of one functor to another with a transformation involving the base monad@@ -350,8 +350,8 @@ mapsM phi = loop where loop stream = case stream of Return r -> Return r- Delay m -> Delay (liftM loop m)- Step f -> Delay (liftM Step (phi (fmap loop f)))+ Effect m -> Effect (liftM loop m)+ Step f -> Effect (liftM Step (phi (fmap loop f))) {-# INLINABLE mapsM #-} {-| Resort a succession of layers of the form @m (f x)@. Though @mapsM@ @@ -368,8 +368,8 @@ decompose = loop where loop stream = case stream of Return r -> Return r - Delay m -> Delay (liftM loop m)- Step (Compose mstr) -> Delay $ do+ Effect m -> Effect (liftM loop m)+ Step (Compose mstr) -> Effect $ do str <- mstr return (Step (fmap loop str)) @@ -380,7 +380,7 @@ run = loop where loop stream = case stream of Return r -> return r- Delay m -> m >>= loop+ Effect m -> m >>= loop Step mrest -> mrest >>= loop {-# INLINABLE run #-} @@ -402,13 +402,13 @@ where go0 f = case f of Return r -> return r - Delay m -> lift m >>= go0 + Effect m -> lift m >>= go0 Step fstr -> do f' <- fstr go1 f' go1 f = case f of Return r -> return r - Delay m -> lift m >>= go1+ Effect m -> lift m >>= go1 Step fstr -> do sep f' <- fstr@@ -454,7 +454,7 @@ concats = loop where loop stream = case stream of Return r -> return r- Delay m -> join $ lift (liftM loop m)+ Effect m -> join $ lift (liftM loop m) Step fs -> join (fmap loop fs) {-# INLINE concats #-} @@ -473,7 +473,7 @@ | n <= 0 = Return stream | otherwise = case stream of Return r -> Return (Return r)- Delay m -> Delay (liftM (loop n) m)+ Effect m -> Effect (liftM (loop n) m) Step fs -> case n of 0 -> Return (Step fs) _ -> Step (fmap (loop (n-1)) fs)@@ -495,7 +495,7 @@ chunksOf n0 = loop where loop stream = case stream of Return r -> Return r- Delay m -> Delay (liftM loop m)+ Effect m -> Effect (liftM loop m) Step fs -> Step $ Step $ fmap (fmap loop . splitsAt (n0-1)) fs {-# INLINABLE chunksOf #-} @@ -506,7 +506,7 @@ distribute = loop where loop stream = case stream of Return r -> lift $ Return r- Delay tmstr -> hoist lift tmstr >>= distribute+ Effect tmstr -> hoist lift tmstr >>= distribute Step fstr -> join $ lift (Step (fmap (Return . distribute) fstr)) -- | Repeat a functorial layer, command or instruction forever.@@ -517,7 +517,7 @@ -- Repeat a functorial layer, command or instruction forever. repeatsM :: (Monad m, Functor f) => m (f ()) -> Stream f m r repeatsM mf = loop where- loop = Delay $ do+ loop = Effect $ do f <- mf return $ Step $ fmap (\_ -> loop) f @@ -545,7 +545,7 @@ hoistExposed trans = loop where loop stream = case stream of Return r -> Return r- Delay m -> Delay (trans (liftM loop m))+ Effect m -> Effect (trans (liftM loop m)) Step f -> Step (fmap loop f) mapsExposed :: (Monad m, Functor f) @@ -553,15 +553,15 @@ mapsExposed phi = loop where loop stream = case stream of Return r -> Return r- Delay m -> Delay (liftM loop m)+ Effect m -> Effect (liftM loop m) Step f -> Step (phi (fmap loop f)) {-# INLINABLE mapsExposed #-} mapsMExposed phi = loop where loop stream = case stream of Return r -> Return r- Delay m -> Delay (liftM loop m)- Step f -> Delay (liftM Step (phi (fmap loop f)))+ Effect m -> Effect (liftM loop m)+ Step f -> Effect (liftM Step (phi (fmap loop f))) {-# INLINABLE mapsMExposed #-} -- Map a stream directly to its church encoding; compare @Data.List.foldr@@@ -578,7 +578,7 @@ destroyExposed stream0 construct effect done = loop stream0 where loop stream = case stream of Return r -> done r- Delay m -> effect (liftM loop m)+ Effect m -> effect (liftM loop m) Step fs -> construct (fmap loop fs) {-# INLINABLE destroyExposed #-} @@ -589,17 +589,17 @@ -} unexposed :: (Functor f, Monad m) => Stream f m r -> Stream f m r-unexposed = Delay . loop where+unexposed = Effect . loop where loop stream = case stream of Return r -> return (Return r)- Delay m -> m >>= loop- Step f -> return (Step (fmap (Delay . loop) f))+ Effect m -> m >>= loop+ Step f -> return (Step (fmap (Effect . loop) f)) {-# INLINABLE unexposed #-} mwrap :: (Monad m, Functor f ) => m (Stream f m r) -> Stream f m r-mwrap = Delay+mwrap = Effect wrap :: (Monad m, Functor f ) => f (Stream f m r) -> Stream f m r wrap = Step@@ -620,11 +620,11 @@ => (forall x y . f x -> g y -> h (x,y)) -> Stream f m r -> Stream g m r -> Stream h m r zipsWith phi = curry loop where- loop (s1, s2) = Delay $ go s1 s2+ loop (s1, s2) = Effect $ go s1 s2 go (Return r) p = return $ Return r go q (Return s) = return $ Return s- go (Delay m) p = m >>= \s -> go s p- go q (Delay m) = m >>= go q+ go (Effect m) p = m >>= \s -> go s p+ go q (Effect m) = m >>= go q go (Step f) (Step g) = return $ Step $ fmap loop (phi f g) {-# INLINABLE zipsWith #-}
Streaming/Prelude.hs view
@@ -98,6 +98,7 @@ , read , show , cons+ , duplicate -- * Splitting and inspecting streams of elements , next@@ -164,6 +165,8 @@ -- * Zips , zip , zipWith+ , zip3+ , zipWith3 -- * Pair manipulation , lazily@@ -196,7 +199,7 @@ import Prelude hiding (map, mapM, mapM_, filter, drop, dropWhile, take, mconcat, sum, product , iterate, repeat, cycle, replicate, splitAt , takeWhile, enumFrom, enumFromTo, enumFromThen, length- , print, zipWith, zip, seq, show, read+ , print, zipWith, zip, zipWith3, zip3, seq, show, read , readLn, sequence, concat, span, break) import qualified GHC.IO.Exception as G@@ -312,7 +315,7 @@ break pred = loop where loop str = case str of Return r -> Return (Return r)- Delay m -> Delay $ liftM loop m+ Effect m -> Effect $ liftM loop m Step (a :> rest) -> if (pred a) then Return (Step (a :> rest)) else Step (a :> loop rest)@@ -336,14 +339,14 @@ where loop0 x stream = case stream of Return r -> return (return r)- Delay mn -> Delay $ liftM (loop0 x) mn+ Effect mn -> Effect $ liftM (loop0 x) mn Step (a :> rest) -> loop a (step x a) rest loop a !x stream = do if pred (done x) then return (yield a >> stream) else case stream of Return r -> yield a >> return (return r)- Delay mn -> Delay $ liftM (loop a x) mn+ Effect mn -> Effect $ liftM (loop a x) mn Step (a' :> rest) -> do yield a loop a' (step x a') rest@@ -363,7 +366,7 @@ :: Monad m => (a -> Bool) -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r breaks thus = loop where- loop stream = Delay $ do+ loop stream = Effect $ do e <- next stream return $ case e of Left r -> Return r@@ -389,8 +392,8 @@ chain f = loop where loop str = case str of Return r -> return r- Delay mn -> Delay (liftM loop mn)- Step (a :> rest) -> Delay $ do+ Effect mn -> Effect (liftM loop mn)+ Step (a :> rest) -> Effect $ do f a return (Step (a :> loop rest)) {-# INLINE chain #-}@@ -473,7 +476,7 @@ cycle = forever -{-| Delay each element by the supplied number of seconds.+{-| Effect each element by the supplied number of seconds. mapM :: Monad m => (a -> m b) -> Stream (Of a) m r -> Stream (Of b) m r -}@@ -503,7 +506,7 @@ effects = loop where loop stream = case stream of Return r -> return r- Delay m -> m >>= loop + Effect m -> m >>= loop Step (_ :> rest) -> loop rest {-#INLINABLE effects #-} @@ -533,13 +536,12 @@ -- | Ignore the first n elements of a stream, but carry out the actions drop :: (Monad m) => Int -> Stream (Of a) m r -> Stream (Of a) m r drop = loop where- loop !n stream - | n <= 0 = stream- | otherwise = case stream of+ loop 0 stream = stream+ loop n stream = case stream of Return r -> Return r- Delay ma -> Delay (liftM (loop n) ma)+ Effect ma -> Effect (liftM (loop n) ma) Step (a :> as) -> loop (n-1) as-{-# INLINEABLE drop #-}+{-# INLINE drop #-} -- --------------- -- dropWhile@@ -557,11 +559,11 @@ dropWhile pred = loop where loop stream = case stream of Return r -> Return r- Delay ma -> Delay (liftM loop ma)+ Effect ma -> Effect (liftM loop ma) Step (a :> as) -> if pred a then loop as else Step (a :> as)-{-# INLINEABLE dropWhile #-}+{-# INLINE dropWhile #-} -- --------------- -- each @@ -647,13 +649,13 @@ -- | Skip elements of a stream that fail a predicate filter :: (Monad m) => (a -> Bool) -> Stream (Of a) m r -> Stream (Of a) m r filter pred = loop where- loop !str = case str of+ loop str = case str of Return r -> Return r- Delay m -> Delay (liftM loop m)+ Effect m -> Effect (liftM loop m) Step (a :> as) -> if pred a then Step (a :> loop as) else loop as-{-# INLINEABLE filter #-}+{-# INLINE filter #-} -- --------------- -- filterM@@ -664,8 +666,8 @@ filterM pred = loop where loop str = case str of Return r -> Return r- Delay m -> Delay $ liftM loop m- Step (a :> as) -> Delay $ do + Effect m -> Effect $ liftM loop m+ Step (a :> as) -> Effect $ do bool <- pred a if bool then return $ Step (a :> loop as)@@ -722,7 +724,7 @@ where loop !stream !x = case stream of Return r -> return (done x)- Delay m -> m >>= \s -> loop s x+ Effect m -> m >>= \s -> loop s x Step (a :> rest) -> loop rest (step x a) {-# INLINABLE fold_ #-} @@ -764,7 +766,7 @@ where loop stream !x = case stream of Return r -> return (done x :> r)- Delay m -> m >>= \s -> loop s x+ Effect m -> m >>= \s -> loop s x Step (a :> rest) -> loop rest (step x a) {-# INLINABLE fold #-} @@ -781,7 +783,7 @@ where loop stream !x = case stream of Return r -> done x - Delay m -> m >>= \s -> loop s x+ Effect m -> m >>= \s -> loop s x Step (a :> rest) -> do x' <- step x a loop rest x'@@ -800,7 +802,7 @@ where loop stream !x = case stream of Return r -> done x >>= \b -> return (b :> r)- Delay m -> m >>= \s -> loop s x+ Effect m -> m >>= \s -> loop s x Step (a :> rest) -> do x' <- step x a loop rest x'@@ -822,7 +824,7 @@ foldrT step = loop where loop stream = case stream of Return r -> return r- Delay m -> lift m >>= loop+ Effect m -> lift m >>= loop Step (a :> as) -> step a (loop as) {-# INLINABLE foldrT #-} @@ -835,7 +837,7 @@ foldrM step = loop where loop stream = case stream of Return r -> return r- Delay m -> m >>= loop+ Effect m -> m >>= loop Step (a :> as) -> step a (loop as) {-# INLINABLE foldrM #-} @@ -849,7 +851,7 @@ for str0 act = loop str0 where loop str = case str of Return r -> Return r - Delay m -> Delay $ liftM loop m+ Effect m -> Effect $ liftM loop m Step (a :> rest) -> do act a loop rest@@ -864,7 +866,7 @@ -> Stream (Compose (Of a) f) m r -> Stream (Stream (Compose (Of a) f) m) m r groupedBy equals = loop where- loop stream = Delay $ do+ loop stream = Effect $ do e <- inspect stream return $ case e of Left r -> Return r@@ -875,7 +877,7 @@ span' pred = loop where loop str = case str of Return r -> Return (Return r)- Delay m -> Delay $ liftM loop m+ Effect m -> Effect $ liftM loop m Step s@(Compose (a :> rest)) -> case pred a of True -> Step (Compose (a :> fmap loop rest)) False -> Return (Step s)@@ -889,7 +891,7 @@ -> Stream (Of a) m r -> Stream (Stream (Of a) m) m r groupBy equals = loop where- loop stream = Delay $ do+ loop stream = Effect $ do e <- next stream return $ case e of Left r -> Return r@@ -915,7 +917,7 @@ -- | Iterate a monadic function from a seed value, streaming the results forever iterateM :: Monad m => (a -> m a) -> m a -> Stream (Of a) m r iterateM f = loop where- loop ma = Delay $ do + loop ma = Effect $ do a <- ma return (Step (a :> loop (f a))) {-# INLINEABLE iterateM #-}@@ -957,9 +959,9 @@ map f = loop where loop stream = case stream of Return r -> Return r- Delay m -> Delay (liftM loop m)+ Effect m -> Effect (liftM loop m) Step (a :> as) -> Step (f a :> loop as)-{-# INLINEABLE map #-}+{-# INLINE map #-} -- --------------- -- mapFoldable@@ -983,8 +985,8 @@ mapM f = loop where loop str = case str of Return r -> Return r - Delay m -> Delay (liftM loop m)- Step (a :> as) -> Delay $ do + Effect m -> Effect (liftM loop m)+ Step (a :> as) -> Effect $ do a' <- f a return (Step (a' :> loop as) ) {-# INLINEABLE mapM #-}@@ -1004,7 +1006,7 @@ mapM_ f = loop where loop str = case str of Return r -> return r - Delay m -> m >>= loop+ Effect m -> m >>= loop Step (a :> as) -> do f a loop as @@ -1043,7 +1045,7 @@ next = loop where loop stream = case stream of Return r -> return (Left r)- Delay m -> m >>= loop+ Effect m -> m >>= loop Step (a :> rest) -> return (Right (a,rest)) {-# INLINABLE next #-} @@ -1059,7 +1061,7 @@ uncons = loop where loop stream = case stream of Return () -> return Nothing- Delay m -> m >>= loop+ Effect m -> m >>= loop Step (a :> rest) -> return (Just (a,rest)) {-# INLINABLE uncons #-} @@ -1142,7 +1144,7 @@ replicateM :: Monad m => Int -> m a -> Stream (Of a) m () replicateM n ma = loop n where loop 0 = Return ()- loop n = Delay $ do + loop n = Effect $ do a <- ma return (Step $ a :> loop (n-1)) {-# INLINEABLE replicateM #-}@@ -1155,7 +1157,7 @@ -} reread :: Monad m => (s -> m (Maybe a)) -> s -> Stream (Of a) m () reread step s = loop where - loop = Delay $ do + loop = Effect $ do m <- step s case m of Nothing -> return (Return ())@@ -1186,15 +1188,14 @@ scan :: Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream (Of a) m r -> Stream (Of b) m r scan step begin done = loop begin where- loop !x stream = do - yield (done x)- case stream of + loop !x stream = Step $ + done x :> case stream of Return r -> Return r- Delay m -> Delay $ liftM (loop x) m+ Effect m -> Effect $ liftM (loop x) m Step (a :> rest) -> do- let x' = step x a+ let !x' = step x a loop x' rest-{-# INLINABLE scan #-}+{-# INLINE scan #-} {-| Strict, monadic left scan @@ -1219,7 +1220,7 @@ yield b case stream of Return r -> Return r- Delay m -> Delay $ liftM (loop x) m+ Effect m -> Effect $ liftM (loop x) m Step (a :> rest) -> do x' <- lift $ step x a loop x' rest@@ -1248,7 +1249,7 @@ loop !m !x stream = do case stream of Return r -> return r- Delay mn -> Delay $ liftM (loop m x) mn+ Effect mn -> Effect $ liftM (loop m x) mn Step (a :> rest) -> do case m of Nothing' -> do @@ -1277,8 +1278,8 @@ sequence = loop where loop stream = case stream of Return r -> Return r- Delay m -> Delay $ liftM loop m- Step (ma :> rest) -> Delay $ do+ Effect m -> Effect $ liftM loop m+ Step (ma :> rest) -> Effect $ do a <- ma return (Step (a :> loop rest)) {-# INLINEABLE sequence #-}@@ -1317,7 +1318,7 @@ span pred = loop where loop str = case str of Return r -> Return (Return r)- Delay m -> Delay $ liftM loop m+ Effect m -> Effect $ liftM loop m Step (a :> rest) -> if pred a then Step (a :> loop rest) else Return (Step (a :> rest))@@ -1327,7 +1328,7 @@ {-| Split a stream of elements wherever a given element arises. The action is like that of 'Prelude.words'. ->>> S.stdoutLn $ mapsM S.toList $ split ' ' "hello world "+>>> S.stdoutLn $ mapsM S.toList $ split ' ' "hello world " hello world >>> Prelude.mapM_ Prelude.putStrLn (Prelude.words "hello world ")@@ -1380,11 +1381,12 @@ take :: (Monad m, Functor f) => Int -> Stream f m r -> Stream f m () take = loop where- loop n p = when (n > 0) $+ loop 0 p = return ()+ loop n p = case p of Step fas -> Step (fmap (loop (n-1)) fas)- Delay m -> Delay (liftM (loop n) m)+ Effect m -> Effect (liftM (loop n) m) Return r -> Return ()-{-# INLINEABLE take #-}+{-# INLINE take #-} -- --------------- -- takeWhile@@ -1396,9 +1398,9 @@ takeWhile pred = loop where loop str = case str of Step (a :> as) -> when (pred a) (Step (a :> loop as))- Delay m -> Delay (liftM loop m)+ Effect m -> Effect (liftM loop m) Return r -> Return ()-{-# INLINEABLE takeWhile #-}+{-# INLINE takeWhile #-} {- Break a stream after the designated number of seconds. @@ -1429,7 +1431,7 @@ then return str else case str of Return r -> return (return r)- Delay m -> Delay (liftM (loop utc) m)+ Effect m -> Effect (liftM (loop utc) m) Step (a:>rest) -> yield a >> loop utc rest @@ -1493,7 +1495,7 @@ unfoldr :: Monad m => (s -> m (Either r (a, s))) -> s -> Stream (Of a) m r unfoldr step = loop where- loop s0 = Delay $ do + loop s0 = Effect $ do e <- step s0 case e of Left r -> return (Return r)@@ -1552,13 +1554,49 @@ where loop str0 str1 = case str0 of Return r -> Return r- Delay m -> Delay $ liftM (\str -> loop str str1) m + Effect m -> Effect $ liftM (\str -> loop str str1) m Step (a :> rest0) -> case str1 of Return r -> Return r- Delay m -> Delay $ liftM (loop str0) m+ Effect m -> Effect $ liftM (loop str0) m Step (b :> rest1) -> Step (f a b :>loop rest0 rest1) {-# INLINABLE zipWith #-} ++-- | Zip three 'Stream's with a combining function+zipWith3 :: Monad m =>+ (a -> b -> c -> d)+ -> Stream (Of a) m r+ -> Stream (Of b) m r+ -> Stream (Of c) m r+ -> Stream (Of d) m r+zipWith3 op = loop where+ loop str0 str1 str2 = do+ e0 <- lift (next str0)+ case e0 of + Left r0 -> return r0+ Right (a0,rest0) -> do + e1 <- lift (next str1)+ case e1 of+ Left r1 -> return r1+ Right (a1,rest1) -> do + e2 <- lift (next str2)+ case e2 of+ Left r2 -> return r2+ Right (a2,rest2) -> do + yield (op a0 a1 a2)+ loop rest0 rest1 rest2+{-# INLINABLE zipWith3 #-} + + +-- | Zip three streams together +zip3 :: Monad m+ => (Stream (Of a) m r)+ -> (Stream (Of b) m r)+ -> (Stream (Of c) m r)+ -> (Stream (Of (a,b,c)) m r)+zip3 = zipWith3 (,,)+{-# INLINABLE zip3 #-}+ -- -------------- -- IO fripperies -- --------------@@ -1630,7 +1668,7 @@ toHandle handle = loop where loop str = case str of Return r -> return r- Delay m -> m >>= loop + Effect m -> m >>= loop Step (s :> rest) -> do liftIO $ IO.hPutStrLn handle s loop rest@@ -1642,7 +1680,7 @@ print = loop where loop stream = case stream of Return r -> return r - Delay m -> m >>= loop+ Effect m -> m >>= loop Step (a :> rest) -> do liftIO (Prelude.print a) loop rest@@ -1665,7 +1703,7 @@ where loop stream = case stream of Return _ -> return () - Delay m -> m >>= loop+ Effect m -> m >>= loop Step (s :> rest) -> do x <- liftIO $ try (putStrLn s) case x of@@ -1696,7 +1734,7 @@ stdoutLn' = loop where loop stream = case stream of Return r -> return r - Delay m -> m >>= loop+ Effect m -> m >>= loop Step (s :> rest) -> liftIO (putStrLn s) >> loop rest {-# INLINE stdoutLn' #-} @@ -1782,14 +1820,31 @@ eitherToSum s = case s of Left a :> r -> InL (a :> r) Right b :> r -> InR (b :> r)- +{-#INLINE eitherToSum #-} composeToSum :: Compose (Of Bool) f r -> Sum f f r composeToSum x = case x of Compose (True :> f) -> InR f Compose (False :> f) -> InL f+{-#INLINE composeToSum #-} sumToCompose :: Sum f f r -> Compose (Of Bool) f r sumToCompose x = case x of InR f -> Compose (True :> f) InL f -> Compose (False :> f)- +{-#INLINE sumToCompose #-}++duplicate+ :: Monad m =>+ Stream (Of a) m r -> Stream (Of a) (Stream (Of a) m) r+duplicate = loop where+loop str = do + e <- lift (lift (next str))+ case e of+ Left r -> return r+ Right (a, rest) -> do+ yield a + mwrap $ do+ yield a+ return (loop rest)+{-#INLINABLE duplicate#-}+
streaming.cabal view
@@ -1,5 +1,5 @@ name: streaming-version: 0.1.2.0+version: 0.1.2.2 cabal-version: >=1.10 build-type: Simple synopsis: an elementary streaming prelude and a general monad transformer for streaming applications.@@ -8,7 +8,13 @@ optimized for streaming applications and replacing @FreeT@. See the <https://hackage.haskell.org/package/streaming#readme readme> below for an explanation. Elementary usage can be divined from the ghci examples in - @Streaming.Prelude@+ @Streaming.Prelude@ and from the remarks somewhat theoretical+ <https://hackage.haskell.org/package/streaming#readme readme>+ below, including the examples linked there. Note also the + <https://hackage.haskell.org/package/streaming-bytestring streaming bytestring> + and + <https://hackage.haskell.org/package/streaming-utils streaming utils> + packages. . The simplest form of interoperation with <http://hackage.haskell.org/package/pipes pipes> is accomplished with this isomorphism:@@ -30,11 +36,15 @@ <https://hackage.haskell.org/package/streaming#readme readme> below. .- Note also the - <https://hackage.haskell.org/package/streaming-bytestring streaming bytestring> - and - <https://hackage.haskell.org/package/streaming-utils streaming utils> - packages.+ Here are some results for an + <https://gist.github.com/michaelt/f19bef01423b17f29ffd expansion> + of the little + <https://github.com/ekmett/machines/blob/master/benchmarks/Benchmarks.hs benchmarks> + included in the machines package:+ .+ <<http://i.imgur.com/sSG5MvH.png>>+ .+ license: BSD3 license-file: LICENSE