streaming 0.1.4.1 → 0.1.4.2
raw patch · 4 files changed
+179/−123 lines, 4 filesdep +monad-controldep ~basedep ~mmorphdep ~transformers
Dependencies added: monad-control
Dependency ranges changed: base, mmorph, transformers
Files
- Streaming.hs +9/−1
- Streaming/Internal.hs +3/−3
- Streaming/Prelude.hs +83/−30
- streaming.cabal +84/−89
Streaming.hs view
@@ -1,4 +1,4 @@-{-#LANGUAGE RankNTypes #-}+{-#LANGUAGE RankNTypes, CPP, Trustworthy #-} module Streaming ( -- * An iterable streaming monad transformer@@ -79,6 +79,10 @@ MonadBase(..), ResourceT(..), runResourceT,+#if MIN_VERSION_base(4,8,0)+ Bifunctor(..),+#endif+ join, liftM, liftM2,@@ -101,6 +105,10 @@ import Control.Monad.Base import Control.Monad.Trans.Resource+#if MIN_VERSION_base(4,8,0)+import Data.Bifunctor+#endif+ {- $stream The 'Stream' data type can be used to represent any effectful
Streaming/Internal.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE RankNTypes, StandaloneDeriving,DeriveDataTypeable, BangPatterns #-} {-# LANGUAGE UndecidableInstances, CPP, FlexibleInstances, MultiParamTypeClasses #-} +{-#LANGUAGE Trustworthy #-} module Streaming.Internal ( -- * The free monad transformer -- $stream@@ -88,17 +89,16 @@ import Control.Monad.Morph import Data.Monoid (Monoid (..), (<>)) import Data.Functor.Identity-import GHC.Exts ( build ) import Data.Data ( Data, Typeable ) import Prelude hiding (splitAt) import Data.Functor.Compose import Data.Functor.Sum import Control.Concurrent (threadDelay)--- import Data.Time (getCurrentTime, diffUTCTime, picosecondsToDiffTime, addUTCTime)- import Control.Monad.Base import Control.Monad.Trans.Resource import Control.Monad.Catch (MonadCatch (..))+import Control.Monad.Trans.Control+ {- $stream
Streaming/Prelude.hs view
@@ -41,7 +41,7 @@ > -} {-# LANGUAGE RankNTypes, BangPatterns, DeriveDataTypeable, TypeFamilies,- DeriveFoldable, DeriveFunctor, DeriveTraversable #-}+ DeriveFoldable, DeriveFunctor, DeriveTraversable, CPP, Trustworthy #-} module Streaming.Prelude ( -- * Types@@ -78,7 +78,6 @@ , print , toHandle , writeFile- , first , effects , erase , drained@@ -217,6 +216,9 @@ , strictly , fst' , snd'+ , mapOf+ , _first+ , _second -- * Interoperation , reread@@ -260,12 +262,11 @@ import Data.Functor.Compose import Control.Monad.Trans.Resource -import GHC.Exts ( SpecConstrAnnotation(..) ) import GHC.Magic--data SPEC = SPEC | SPEC2 +#if MIN_VERSION_base(4,8,0)+import Data.Bifunctor+#endif -{-# ANN type SPEC ForceSpecConstr #-} -- | A left-strict pair; the base functor for streams of individual elements. data Of a b = !a :> b deriving (Data, Eq, Foldable, Ord,@@ -284,6 +285,16 @@ a <$ (b :> x) = b :> a {-#INLINE (<$) #-} +#if MIN_VERSION_base(4,8,0)+instance Bifunctor Of where+ bimap f g (a :> b) = f a :> g b + {-#INLINE bimap #-}+ first f (a :> b) = f a :> b+ {-#INLINE first #-}+ second g (a :> b) = a :> g b + {-#INLINE second #-}+#endif+ instance Monoid a => Applicative (Of a) where pure x = mempty :> x {-#INLINE pure #-}@@ -317,7 +328,7 @@ then we can restate some types as follows: -> mapOf :: (a -> b) -> Of a ~~> Of b -- bifunctor lmap+> mapOf :: (a -> b) -> Of a ~~> Of b -- Bifunctor first > lazily :: Of a ~~> (,) a > Identity . fst' :: Of a ~~> Identity a @@ -337,35 +348,77 @@ > S.map :: (a -> b) -> Stream (Of a) m ~> Stream (Of b) m - Thus we can @maps@ it in turn+ Thus we can @maps@ it in turn. -> --}+ -} lazily :: Of a b -> (a,b) lazily = \(a:>b) -> (a,b) {-# INLINE lazily #-} +{-| Convert a standard Haskell pair into a left-strict pair -} strictly :: (a,b) -> Of a b-strictly = \(a,b) -> a :> b+strictly = \(a,b) -> a :> b {-# INLINE strictly #-} +{-| @fst'@ and @snd'@ extract the first and second element of a pair++>>> S.fst' (1:>"hi")+1+>>> S.snd' (1:>"hi")+"hi"+++ They are contained in the @_first@ and @_second@ lenses, + if any lens library is in scope+ +>>> import Lens.Micro+>>> (1:>"hi") ^. S._first+1+>>> (1:>"hi") ^. S._second+"hi"++ -}+ fst' :: Of a b -> a fst' (a :> b) = a {-#INLINE fst' #-}- snd' :: Of a b -> b snd' (a :> b) = b {-#INLINE snd' #-} +{-| Map a function over the first element of an @Of@ pair++>>> S.mapOf even (1:>"hi")+False :> "hi"++ @mapOf@ is just @first@ from the @Bifunctor@ instance + +>>> first even (1:>"hi")+False :> "hi" ++ and is contained in the @_first@ lens+ +>>> import Lens.Micro+>>> over S._first even (1:>"hi")+False :> "hi"++ -}+ mapOf :: (a -> b) -> Of a r -> Of b r mapOf f (a:> b) = (f a :> b) {-#INLINE mapOf #-} -_first :: Functor f => (a -> f a1) -> Of a b -> f (Of a1 b)+{-| A lens into the first element of a left-strict pair -}+_first :: Functor f => (a -> f a') -> Of a b -> f (Of a' b) _first afb (a:>b) = fmap (\c -> (c:>b)) (afb a) {-# INLINE _first #-} +{-| A lens into the second element of a left-strict pair -}+_second :: Functor f => (b -> f b') -> Of a b -> f (Of a b')+_second afb (a:>b) = fmap (\c -> (a:>c)) (afb b)+{-#INLINABLE _second #-}+ all :: Monad m => (a -> Bool) -> Stream (Of a) m r -> m (Of Bool r) all thus = loop True where loop b str = case str of@@ -861,23 +914,23 @@ {-# INLINABLE filterM #-} --- ------------------ first--- ----------------{- | Take either the first item in a stream or the return value, if it is empty.- The typical mark of an infinite stream is a polymorphic return value; in - that case, 'first' is a sort of @safeHead@-- To iterate an action returning a 'Maybe', until it succeeds.---}-first :: Monad m => Stream (Of r) m r -> m r-first = loop where- loop str = case str of- Return r -> return r- Effect m -> m >>= loop- Step (r :> rest) -> return r-{-# INLINABLE first #-} +-- -- ---------------+-- -- first+-- -- ---------------+-- {- | Take either the first item in a stream or the return value, if it is empty.+-- The typical mark of an infinite stream is a polymorphic return value; in+-- that case, 'first' is a sort of @safeHead@+--+-- To iterate an action returning a 'Maybe', until it succeeds.+--+-- -}+-- first :: Monad m => Stream (Of r) m r -> m r+-- first = loop where+-- loop str = case str of+-- Return r -> return r+-- Effect m -> m >>= loop+-- Step (r :> rest) -> return r+-- {-# INLINABLE first #-} -- --------------- -- fold
streaming.cabal view
@@ -1,24 +1,65 @@ name: streaming-version: 0.1.4.1+version: 0.1.4.2 cabal-version: >=1.10 build-type: Simple synopsis: an elementary streaming prelude and general stream type. description: @Streaming.Prelude@ exports an elementary streaming prelude focused on a simple \"source\" or \"producer\" type, namely @Stream (Of a) m r@.- This is a sort of effectful version of @([a],r)@ in which monadic action - is interleaved between successive elements.- The main module, @Streaming@, exports a much more general type,+ @Stream (Of a) m r@ is a sort of effectful version of+ @([a],r)@ in which successive elements arise from some sort of monadic+ action. Everything is the library is organized to make + programming with this type as simple as possible+ by making it as close to @Prelude@ and @Data.List@. Thus for example+ the trivial program+ .+ > S.sum (S.take 3 (S.readLn :: Stream (Of Int) IO ()))+ .+ sums the first three valid integers from user input. Similarly,+ .+ > S.stdoutLn (S.map reverse (S.take 3 S.stdinLn)) + .+ reverses the first three lines from stdin as they arise, + and sends them to stdout. And so on,+ with filtering, mapping, breaking, chunking and so forth. + We program with streams of @Int@s or @String@s directly as + if they constituted something like a list rather than \"extracting a list from IO\",+ which is the origin of typical Haskell memory catastrophes. + Basically any case where you are + tempted to use @mapM@, @replicateM@, @traverse@ or @sequence@+ with Haskell lists, you would do better to use something like+ @Stream (Of a) m r@. The type signatures are a little fancier, but + the programs themselves are mostly the same or simpler. Thus, + the little demo program from+ <http://stackoverflow.com/questions/24068399/haskell-performance-of-iorefs this SO question>+ .+ > main = mapM newIORef [1..10^8::Int] >>= mapM readIORef >>= mapM_ print+ .+ quickly exhausts memory; this of course has nothing to do with @IORefs@ + and is cured by+ .+ > import qualified Streaming.Prelude as S+ > main = S.print (S.mapM readIORef (S.mapM newIORef (S.each [1..10^8::Int])))+ .+ which uses no more memory than @hello-world@, and is simpler anyway, since it+ doesn't involve \"extracting a list from IO\". Almost+ every use of list @mapM@, @replicateM@, @traverse@ and @sequence@ produces+ this problem on a smaller scale. People get used to it, as if it were+ characteristic of Haskell programs to use a lot of memory, when+ \"extracting a list or sequence from IO\" is just bad practice pure and simple.+ List @mapM@, @replicateM@, @traverse@ and @sequence@ make sense under certain + conditions. Similarly, @unsafePerformIO@ makes sense under certain conditions.+ .+ The @Streaming@ module exports the general type, @Stream f m r@, which can be used to stream successive distinct steps characterized by /any/- functor @f@, though we are here interested only in a limited range of- cases.- .- The streaming-IO libraries have various devices for dealing+ functor @f@, though we are mostly interested in organizing computations+ of the form @Stream (Of a) m r@. The streaming-IO libraries have + various devices for dealing with effectful variants of @[a]@ or @([a],r)@. But it is only with the general type @Stream f m r@, or some equivalent, that one can envisage (for example) the connected streaming of their- sorts of stream -- as one makes lists of lists in the Haskell+ sorts of stream - as one makes lists of lists in the Haskell @Prelude@ and @Data.List@. One needs some such type if we are to express properly streaming equivalents of e.g. .@@ -26,28 +67,41 @@ > chunksOf :: Int -> [a] -> [[a]] > lines :: [Char] -> [[Char]] -- but similarly with bytestring, etc. .- to mention a few obviously desirable operations. But once one grasps- the iterable stream concept needed to express those functions - to wit,- @Stream f m r@ or some equivalent - then one will also see that,- with it, one is already in possession of a complete+ to mention a few obviously desirable operations. (This is explained more elaborately in the <https://hackage.haskell.org/package/streaming#readme readme> below.) One could throw something+ like @Stream@ on top of a prior stream concept: this is how @pipes@ and+ @pipes-group@ (which are very much our model here) use @FreeT@.+ But once one grasps+ the iterable stream concept needed to express those functions - + here given a somewhat optimized implementation as @Stream f m r@ + (following, as usual, models derived from the @pipes@ library) - + then one will also see that,+ with it, one is /already/ in possession of a complete elementary streaming library - since one possesses @Stream ((,) a) m r@ or equivalently @Stream (Of a) m r@. This is the type of a \'generator\' or \'producer\' or whatever you call an effectful stream of items.- The present @Streaming.Prelude@ is thus the simplest streaming - library that can replicate anything like the API of the- @Prelude@ and @Data.List@. + /The present @Streaming.Prelude@ is thus the simplest streaming library that can replicate anything like the API of the @Prelude@ and @Data.List@/. . The emphasis of the library is on interoperation; for- the rest its advantages are: extreme simplicity and re-use of+ the rest its advantages are: extreme simplicity, re-use of intuitions the user has gathered from mastery of @Prelude@ and- @Data.List@. The two conceptual pre-requisites are some+ @Data.List@, and a total and systematic rejection of type synonyms. + The two conceptual pre-requisites are some comprehension of monad transformers and some familiarity- with \'rank 2 types\'. + with \'rank 2 types\'. It is hoped that experimentation with this+ simple material, starting with the ghci examples in @Streaming.Prelude@, + will give people who are new to these concepts some + intuition about their importance. The most fundamental purpose of the+ library is to express elementary streaming ideas without reliance on + a complex framework, but in a way that integrates transparently with+ the rest of Haskell, using ideas - e.g. rank 2 types, which are here+ implicit or explicit in most mapping - that the user can carry elsewhere,+ rather than binding her intelligence to a so-called streaming IO framework (as + necessary as that is for certain purposes.) . See the <https://hackage.haskell.org/package/streaming#readme readme> - below for an explanation, including the examples linked there. + below for further explanation, including the examples linked there. Elementary usage can be divined from the ghci examples in @Streaming.Prelude@ and perhaps from this rough beginning of a <https://github.com/michaelt/streaming-tutorial/blob/master/tutorial.md tutorial>.@@ -84,74 +138,15 @@ > ($$ Conduit.mapM_ Streaming.yield) . hoist lift :: Source m a -> Stream (Of a) m () . These conversions should never be more expensive than a single @>->@ or @=$=@. - .- Here is a simple example (conceptually it is a bit advanced, maybe) - that runs a single underlying stream with several- streaming-io libraries at once, superimposing their effects - without any accumulation:- .- > module Main (main) where- > import Streaming - > import Pipes - > import Data.Conduit- > import qualified Streaming.Prelude as S- > import qualified Data.Conduit.List as CL- > import qualified Pipes.Prelude as P- > import qualified System.IO.Streams as IOS- > import Data.ByteString.Char8 (pack)- > import Data.Function ((&))- >- > mkConduit = CL.unfoldM S.uncons- > mkPipe = P.unfoldr S.next- > mkIOStream = IOS.unfoldM S.uncons- >- > main = iostreamed where- > urstream = S.take 3 S.readLn :: Stream (Of Int) IO () - - > streamed = S.copy urstream & S.map (\n -> "streaming says: " ++ show n) - > & S.stdoutLn - > piped = runEffect $ - > mkPipe (S.copy streamed) >-> P.map (\n -> "pipes says: " ++ show n) - > >-> P.stdoutLn - > conduited = - > mkConduit (S.copy piped) $$ CL.map (\n -> "conduit says: " ++ show n) - > =$ CL.mapM_ (liftIO . putStrLn)- > iostreamed = do- > str0 <- mkIOStream conduited- > str1 <- IOS.map (\n -> pack $ "io-streams says: " ++ show n ++ "\n") str0 - > IOS.supply str1 IOS.stdout- .- This program successively parses three @Int@s from standard input, - and /simulaneously/ passes them to (here trivial) stream-consuming - processes from four different libraries, using the @copy@ function from- @Streaming.Prelude@. I mark my own input with @/<Enter/>@ below:+ The simplest interoperation with regular Haskell lists is provided by, say .- > >>> main- > 1 <Enter>- > streaming says: 1- > pipes says: 1- > conduit says: 1- > io-streams says: 1- > 2 <Enter>- > streaming says: 2- > pipes says: 2- > conduit says: 2- > io-streams says: 2- > 3 <Enter>- > streaming says: 3- > pipes says: 3- > conduit says: 3- > io-streams says: 3- > >>>+ > Streaming.each :: [a] -> Stream (Of a) m ()+ > Streaming.toList_ :: Stream (Of a) m r -> m [a] .- Of course, I could as well have passed the stream to several- independent conduits; and I might have derived the original- stream from a conduit @Source@ or pipes @Producer@ etc., using- one of the \'conversion\' functions above. Further- points of comparison with the going streaming-IO libraries- are discussed in the- <https://hackage.haskell.org/package/streaming#readme readme>- below.+ The latter of course accumulates the whole list in memory, and is mostly what we are trying+ to avoid. Every use of @Prelude.mapM f@ should be reconceived as using the+ composition @Streaming.toList_ . Streaming.mapM f . Streaming.each@ with a view to+ considering whether the accumulation required by @Streaming.toList_@ is really necessary. . Here are the results of some <https://gist.github.com/michaelt/f19bef01423b17f29ffd microbenchmarks>@@ -164,7 +159,6 @@ Because these are microbenchmarks for individual functions, they represent a sort of \"worst case\"; many other factors can influence the speed of a complex program.- . @@ -197,11 +191,12 @@ build-depends: base >=4.6 && <5 , mtl >=2.1 && <2.3- , mmorph >=1.0 && <1.2- , transformers >=0.4 && <0.5.2+ , mmorph >=1.0 && <1.1+ , transformers >=0.4 && <0.6 , transformers-base < 0.5 , resourcet > 1.1.0 && < 1.2 , exceptions > 0.5 && < 0.9+ , monad-control >=0.3.1 && <1.1 , time , ghc-prim