iteratee 0.8.2.0 → 0.8.3.0
raw patch · 5 files changed
+116/−50 lines, 5 filesdep +parallelPVP ok
version bump matches the API change (PVP)
Dependencies added: parallel
API changes (from Hackage documentation)
+ Data.Iteratee.Iteratee: mapReduce :: (Monad m, Nullable s, Monoid b) => Int -> (s -> b) -> Iteratee s m b
Files
- iteratee.cabal +2/−1
- src/Data/Iteratee/Iteratee.hs +38/−7
- src/Data/Iteratee/ListLike.hs +25/−26
- tests/benchmarkHandle.hs +28/−4
- tests/benchmarks.hs +23/−12
iteratee.cabal view
@@ -1,5 +1,5 @@ name: iteratee-version: 0.8.2.0+version: 0.8.3.0 synopsis: Iteratee-based I/O description: The Iteratee monad provides strict, safe, and functional I/O. In addition@@ -59,6 +59,7 @@ MonadCatchIO-transformers > 0.2 && < 0.3, bytestring >= 0.9 && < 0.10, containers >= 0.2 && < 0.5,+ parallel >= 2 && < 4, transformers >= 0.2 && < 0.3 exposed-modules:
src/Data/Iteratee/Iteratee.hs view
@@ -2,6 +2,7 @@ ,RankNTypes ,FlexibleContexts ,ScopedTypeVariables+ ,BangPatterns ,DeriveDataTypeable #-} -- |Monadic and General Iteratees:@@ -19,6 +20,7 @@ ,isStreamFinished -- ** Chunkwise Iteratees ,mapChunksM_+ ,mapReduce -- ** Nested iteratee combinators ,convStream ,unfoldConvStream@@ -56,7 +58,9 @@ import Control.Exception import Control.Monad.Trans.Class+import Control.Parallel import Data.Maybe+import Data.Monoid import Data.Typeable -- exception helpers@@ -129,11 +133,11 @@ -- | Map a monadic function over the chunks of the stream and ignore the -- result. Useful for creating efficient monadic iteratee consumers, e.g.---+-- -- logger = mapChunksM_ (liftIO . putStrLn)---+-- -- these can be efficiently run in parallel with other iteratees via--- enumPair.+-- @Data.Iteratee.ListLike.zip@. mapChunksM_ :: (Monad m, Nullable s) => (s -> m b) -> Iteratee s m () mapChunksM_ f = liftI step where@@ -143,6 +147,34 @@ step s@(EOF _) = idone () s {-# INLINE mapChunksM_ #-} +-- | Perform a parallel map/reduce. The `bufsize` parameter controls+-- the maximum number of chunks to read at one time. A larger bufsize+-- allows for greater parallelism, but will require more memory.+-- +-- Implementation of `sum`+-- +-- > sum :: (Monad m, LL.ListLike s, Nullable s) => Iteratee s m Int64+-- > sum = getSum <$> mapReduce 4 (Sum . LL.sum)+mapReduce ::+ (Monad m, Nullable s, Monoid b)+ => Int -- ^ maximum number of chunks to read+ -> (s -> b) -- ^ map function+ -> Iteratee s m b+mapReduce bufsize f = liftI (step (0, []))+ where+ step a@(!buf,acc) (Chunk xs)+ | nullC xs = liftI (step a)+ | buf >= bufsize =+ let acc' = mconcat acc+ b' = f xs+ in b' `par` acc' `pseq` liftI (step (0,[b' `mappend` acc']))+ | otherwise =+ let b' = f xs+ in b' `par` liftI (step (succ buf,b':acc))+ step (_,acc) s@(EOF Nothing) =+ idone (mconcat acc) s+ step acc (EOF (Just err)) =+ throwRecoverableErr err (step acc) -- --------------------------------------------------- -- The converters show a different way of composing two iteratees:@@ -269,10 +301,9 @@ -- Run the second enumeratee within the first. In this example, stream2list -- is run within the 'take 10', which is itself run within 'take 15', resulting -- in 15 elements being consumed------ > run =<< enumPure1Chunk [1..1000 :: Int]--- > (joinI $ (I.take 15 ><> I.take 10) I.stream2list)--- > [1,2,3,4,5,6,7,8,9,10]+-- +-- >>> run =<< enumPure1Chunk [1..1000 :: Int] (joinI $ (I.take 15 ><> I.take 10) I.stream2list)+-- [1,2,3,4,5,6,7,8,9,10] -- (><>) :: (Nullable s1, Monad m)
src/Data/Iteratee/ListLike.hs view
@@ -2,7 +2,7 @@ -- |Monadic Iteratees: -- incremental input parsers, processors and transformers---+-- -- This module provides many basic iteratees from which more complicated -- iteratees can be built. In general these iteratees parallel those in -- @Data.List@, with some additions.@@ -121,11 +121,11 @@ -- predicate. -- If the stream is not terminated, the first character of the remaining stream -- satisfies the predicate.---+-- -- N.B. @breakE@ should be used in preference to @break@. -- @break@ will retain all data until the predicate is met, which may -- result in a space leak.---+-- -- The analogue of @List.break@ break :: (Monad m, LL.ListLike s el) => (el -> Bool) -> Iteratee s m s@@ -143,7 +143,7 @@ -- |Attempt to read the next element of the stream and return it -- Raise a (recoverable) error if the stream is terminated---+-- -- The analogue of @List.head@ head :: (Monad m, LL.ListLike s el) => Iteratee s m el head = liftI step@@ -156,7 +156,7 @@ -- |Attempt to read the last element of the stream and return it -- Raise a (recoverable) error if the stream is terminated---+-- -- The analogue of @List.last@ last :: (Monad m, LL.ListLike s el, Nullable s) => Iteratee s m el last = liftI (step Nothing)@@ -231,7 +231,7 @@ -- |Drop n elements of the stream, if there are that many.---+-- -- The analogue of @List.drop@ drop :: (Monad m, Nullable s, LL.ListLike s el) => Int -> Iteratee s m () drop 0 = return ()@@ -244,7 +244,7 @@ {-# INLINE drop #-} -- |Skip all elements while the predicate is true.---+-- -- The analogue of @List.dropWhile@ dropWhile :: (Monad m, LL.ListLike s el) => (el -> Bool) -> Iteratee s m () dropWhile p = liftI step@@ -260,7 +260,7 @@ -- |Return the total length of the remaining part of the stream. -- This forces evaluation of the entire stream.---+-- -- The analogue of @List.length@ length :: (Monad m, Num a, LL.ListLike s el) => Iteratee s m a length = liftI (step 0)@@ -276,10 +276,10 @@ -- |Takes an element predicate and an iteratee, running the iteratee -- on all elements of the stream until the predicate is met.---+-- -- the following rule relates @break@ to @breakE@ -- @break@ pred === @joinI@ (@breakE@ pred stream2stream)---+-- -- @breakE@ should be used in preference to @break@ whenever possible. breakE :: (Monad m, LL.ListLike s el, NullPoint s)@@ -299,7 +299,7 @@ -- |Read n elements from a stream and apply the given iteratee to the -- stream of the read elements. Unless the stream is terminated early, we -- read exactly n elements, even if the iteratee has accepted fewer.---+-- -- The analogue of @List.take@ take :: (Monad m, Nullable s, LL.ListLike s el) => Int -> Enumeratee s s m a take n' iter@@ -326,7 +326,7 @@ -- This is the variation of `take' with the early termination -- of processing of the outer stream once the processing of the inner stream -- finished early.---+-- -- N.B. If the inner iteratee finishes early, remaining data within the current -- chunk will be dropped. takeUpTo :: (Monad m, Nullable s, LL.ListLike s el) => Int -> Enumeratee s s m a@@ -353,7 +353,7 @@ -- Given the stream of elements of the type @el@ and the function @el->el'@, -- build a nested stream of elements of the type @el'@ and apply the -- given iteratee to it.---+-- -- The analog of @List.map@ mapStream :: (Monad m@@ -372,7 +372,7 @@ {-# SPECIALIZE mapStream :: Monad m => (el -> el') -> Enumeratee [el] [el'] m a #-} -- |Map the stream rigidly.---+-- -- Like 'mapStream', but the element type cannot change. -- This function is necessary for @ByteString@ and similar types -- that cannot have 'LooseMap' instances, and may be more efficient.@@ -392,7 +392,7 @@ -- |Creates an 'enumeratee' with only elements from the stream that -- satisfy the predicate function. The outer stream is completely consumed.---+-- -- The analogue of @List.filter@ filter :: (Monad m, Nullable s, LL.ListLike s el)@@ -432,11 +432,10 @@ in g' `seq` (g', leftover) {-# INLINE group #-} --- |Creates an 'enumeratee' in which elements are grouped into+-- | Creates an 'enumeratee' in which elements are grouped into -- contiguous blocks that are equal according to a predicate. -- --- The analogue of @List.groupBy#- +-- The analogue of 'List.groupBy' groupBy :: (LL.ListLike s el, Monad m, Nullable s) => (el -> el -> Bool)@@ -468,7 +467,7 @@ -- Folds -- | Left-associative fold.---+-- -- The analogue of @List.foldl@ foldl :: (Monad m, LL.ListLike s el, FLL.FoldableLL s el)@@ -486,7 +485,7 @@ -- | Left-associative fold that is strict in the accumulator. -- This function should be used in preference to 'foldl' whenever possible.---+-- -- The analogue of @List.foldl'@. foldl' :: (Monad m, LL.ListLike s el, FLL.FoldableLL s el)@@ -503,7 +502,7 @@ -- | Variant of foldl with no base case. Requires at least one element -- in the stream.---+-- -- The analogue of @List.foldl1@. foldl1 :: (Monad m, LL.ListLike s el, FLL.FoldableLL s el)@@ -561,7 +560,7 @@ -- |Enumerate two iteratees over a single stream simultaneously. -- Deprecated, use `Data.Iteratee.ListLike.zip` instead.---+-- -- Compare to @zip@. {-# DEPRECATED enumPair "use Data.Iteratee.ListLike.zip" #-} enumPair@@ -573,7 +572,7 @@ -- |Enumerate two iteratees over a single stream simultaneously.---+-- -- Compare to @zip@. zip :: (Monad m, Nullable s, LL.ListLike s el)@@ -637,9 +636,9 @@ -- is still consuming input. The second iteratee will be terminated with EOF -- when the first iteratee has completed. An example use is to determine -- how many elements an iteratee has consumed:---+-- -- > snd <$> enumWith (dropWhile (<5)) length---+-- -- Compare to @zip@ enumWith :: (Monad m, Nullable s, LL.ListLike s el)@@ -672,7 +671,7 @@ -- |Enumerate a list of iteratees over a single stream simultaneously -- and discard the results. This is a different behavior than Prelude's -- sequence_ which runs iteratees in the list one after the other.---+-- -- Compare to @sequence_@. sequence_ :: (Monad m, LL.ListLike s el, Nullable s)
tests/benchmarkHandle.hs view
@@ -6,6 +6,7 @@ import Data.ByteString (ByteString) import qualified Data.ByteString as B import Criterion.Main+import Data.Monoid import Data.Word import Data.Iteratee import Data.Iteratee.Base.ReadableChunk@@ -42,9 +43,32 @@ len :: Monad m => Iteratee ByteString m Int len = length +testFdMapReduce :: Int -> IO ()+testFdMapReduce n = fileDriverFd bufSize sum file >> return ()+ where+ sum :: Iteratee ByteString IO Word8+ sum = getSum `fmap` mapReduce n (Sum . B.foldl' (+) 0)++testFdFold :: IO ()+testFdFold = fileDriverFd bufSize sum file >> return ()+ where+ sum :: Iteratee ByteString IO Word8+ sum = foldl' (+) 0+ main = defaultMain- [ bench "Fd with String" testFdString- , bench "Hd with String" testHdString- , bench "Fd with ByteString" testFdByte- , bench "Hd with ByteString" testHdByte+ [+ bgroup "String" [+ bench "Fd" testFdString+ ,bench "Hd with String" testHdString+ ]+ ,bgroup "ByteString" [+ bench "Fd" testFdByte+ ,bench "Hd" testHdByte+ ]+ ,bgroup "folds" [+ bench "Fd/fold" testFdFold+ ,bench "Fd/mapReduce 2" $ testFdMapReduce 2+ ,bench "Fd/mapReduce 4" $ testFdMapReduce 4+ ,bench "Fd/mapReduce 8" $ testFdMapReduce 8+ ] ]
tests/benchmarks.hs view
@@ -10,10 +10,11 @@ import Data.Word import Data.Monoid import qualified Data.ByteString as BS+import Control.Applicative+import Control.DeepSeq import Control.Monad.Identity import Control.Monad import qualified Data.ListLike as LL-import Control.DeepSeq import Criterion.Main @@ -31,20 +32,23 @@ | BDIterN String Int (a -> b) (Iteratee s m a) | BDList String (s -> b) s -id1 name i = BDIter1 name id i-idN name i = BDIterN name 5 id i+id1 name i = BDIter1 name id i+idN name i = BDIterN name 5 id i+idNl name i = BDIterN name 1000 id i -makeList name f = BDList name f [1..10000]+defTotalSize = 10000 +makeList name f = BDList name f [1..defTotalSize]+ makeBench :: BD n eval [Int] Identity -> Benchmark makeBench (BDIter1 n eval i) = bench n $- proc eval runIdentity (enumPure1Chunk [1..10000]) i+ proc eval runIdentity (enumPure1Chunk [1..defTotalSize]) i makeBench (BDIterN n csize eval i) = bench n $- proc eval runIdentity (enumPureNChunk [1..10000] csize) i+ proc eval runIdentity (enumPureNChunk [1..defTotalSize] csize) i makeBench (BDList n f l) = bench n $ whnf f l packedBS :: BS.ByteString-packedBS = (BS.pack [1..10000])+packedBS = (BS.pack [1..defTotalSize]) makeBenchBS (BDIter1 n eval i) = bench n $ proc eval runIdentity (enumPure1Chunk packedBS) i@@ -60,9 +64,6 @@ -> Pure proc eval runner enum iter = whnf (eval . runner . (I.run <=< enum)) iter -defaultProc = proc id runIdentity (enumPure1Chunk [1..10000])-defaultNProc = proc id runIdentity (enumPureNChunk [1..10000] 5)- -- ------------------------------------------------------------- -- benchmark groups makeGroup n = bgroup n . map makeBench@@ -79,6 +80,7 @@ takebench = makeGroup "take" $ take0 : takeBenches takeUpTobench = makeGroup "takeUpTo" takeUpToBenches mapbench = makeGroup "map" $ mapBenches+foldbench = makeGroup "fold" $ foldBenches convbench = makeGroup "convStream" convBenches miscbench = makeGroup "other" miscBenches @@ -91,13 +93,14 @@ takebenchbs = makeGroupBS "take" takeBenches takeUpTobenchbs = makeGroupBS "takeUpTo" takeUpToBenches mapbenchbs = makeGroupBS "map" mapBenches+foldbenchbs = makeGroupBS "fold" $ foldBenches convbenchbs = makeGroupBS "convStream" convBenches miscbenchbs = makeGroupBS "other" miscBenches -allListBenches = bgroup "list" [listbench, streambench, breakbench, headsbench, dropbench, lengthbench, takebench, takeUpTobench, mapbench, convbench, miscbench]+allListBenches = bgroup "list" [listbench, streambench, breakbench, headsbench, dropbench, lengthbench, takebench, takeUpTobench, mapbench, foldbench, convbench, miscbench] -allByteStringBenches = bgroup "bytestring" [listbenchbs, streambenchbs, breakbenchbs, headsbenchbs, dropbenchbs, lengthbenchbs, takebenchbs, takeUpTobenchbs, mapbenchbs, convbenchbs, miscbenchbs]+allByteStringBenches = bgroup "bytestring" [listbenchbs, streambenchbs, breakbenchbs, headsbenchbs, dropbenchbs, lengthbenchbs, takebenchbs, takeUpTobenchbs, mapbenchbs, foldbenchbs, convbenchbs, miscbenchbs] list0 = makeList "list one go" deepseq list1 = BDIter1 "stream2list one go" (flip deepseq ()) stream2list@@ -171,6 +174,11 @@ map4 = idN "map head chunked" (I.joinI $ I.rigidMapStream id I.head) mapBenches = [map1, map2, map3, map4] +foldB1 = idNl "foldl' sum" (I.foldl' (+) 0)+foldB2 = idNl "mapReduce foldl' 2 sum" (getSum <$> I.mapReduce 2 (Sum . LL.foldl' (+) 0))+foldB3 = idNl "mapReduce foldl' 4 sum" (getSum <$> I.mapReduce 4 (Sum . LL.foldl' (+) 0))+foldBenches = [foldB1, foldB2, foldB3]+ conv1 = idN "convStream id head chunked" (I.joinI . I.convStream idChunk $ I.head) conv2 = idN "convStream id length chunked" (I.joinI . I.convStream idChunk $ I.length) idChunk = I.liftI step@@ -181,3 +189,6 @@ convBenches = [conv1, conv2] instance NFData BS.ByteString where++instance NFData a => NFData (Sum a) where+ rnf (Sum a) = rnf a