diff --git a/Streaming.hs b/Streaming.hs --- a/Streaming.hs +++ b/Streaming.hs @@ -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 diff --git a/Streaming/Internal.hs b/Streaming/Internal.hs --- a/Streaming/Internal.hs +++ b/Streaming/Internal.hs @@ -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 diff --git a/Streaming/Prelude.hs b/Streaming/Prelude.hs --- a/Streaming/Prelude.hs +++ b/Streaming/Prelude.hs @@ -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 diff --git a/streaming.cabal b/streaming.cabal --- a/streaming.cabal +++ b/streaming.cabal @@ -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 + + . + > 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 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 - 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 . @@ -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 @/@ below: + The simplest interoperation with regular Haskell lists is provided by, say . - > >>> main - > 1 - > streaming says: 1 - > pipes says: 1 - > conduit says: 1 - > io-streams says: 1 - > 2 - > streaming says: 2 - > pipes says: 2 - > conduit says: 2 - > io-streams says: 2 - > 3 - > 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 - - 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 @@ -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