diff --git a/CONTRIBUTORS b/CONTRIBUTORS --- a/CONTRIBUTORS +++ b/CONTRIBUTORS @@ -1,6 +1,7 @@ Thanks to the following individuals for contributing to this project. Oleg Kiselyov +Michael Baikov Gregory Collins Nick Ingolia Brian Lewis diff --git a/iteratee.cabal b/iteratee.cabal --- a/iteratee.cabal +++ b/iteratee.cabal @@ -1,5 +1,5 @@ name: iteratee -version: 0.8.5.0 +version: 0.8.6.0 synopsis: Iteratee-based I/O description: The Iteratee monad provides strict, safe, and functional I/O. In addition @@ -59,6 +59,7 @@ build-depends: ListLike >= 1.0 && < 4, MonadCatchIO-transformers > 0.2 && < 0.3, + monad-control >= 0.2 && < 0.3, bytestring >= 0.9 && < 0.10, containers >= 0.2 && < 0.5, parallel >= 2 && < 4, @@ -79,6 +80,7 @@ Data.Iteratee.IO.Interact Data.Iteratee.Iteratee Data.Iteratee.ListLike + Data.Iteratee.Parallel other-modules: Data.Iteratee.IO.Base diff --git a/src/Data/Iteratee/Base.hs b/src/Data/Iteratee/Base.hs --- a/src/Data/Iteratee/Base.hs +++ b/src/Data/Iteratee/Base.hs @@ -36,14 +36,18 @@ import Prelude hiding (null, catch) import Data.Iteratee.Base.LooseMap import Data.Iteratee.Exception +import Data.Maybe +import Data.Monoid import Data.Nullable import Data.NullPoint -import Data.Monoid +import Control.Monad (liftM) import Control.Monad.IO.Class +import Control.Monad.IO.Control import Control.Monad.Trans.Class import Control.Monad.CatchIO (MonadCatchIO (..), Exception (..), catch, block, toException, fromException) +import Control.Monad.Trans.Control import Control.Applicative hiding (empty) import Control.Exception (SomeException) import qualified Control.Exception as E @@ -151,7 +155,7 @@ where self m f = Iteratee $ \onDone onCont -> let m_done a (Chunk s) - | nullC s = runIter (f a) onDone onCont + | nullC s = runIter (f a) onDone onCont m_done a stream = runIter (f a) (const . flip onDone stream) f_cont where f_cont k Nothing = runIter (k stream) onDone onCont f_cont k e = onCont k e @@ -168,6 +172,21 @@ m `catch` f = Iteratee $ \od oc -> runIter m od oc `catch` (\e -> runIter (f e) od oc) block = ilift block unblock = ilift unblock + +instance forall s. (NullPoint s, Nullable s) => MonadTransControl (Iteratee s) where + liftControl f = lift $ f $ \t -> + liftM (either (uncurry idone) + (\e -> te $ fromMaybe (iterStrExc + "iteratee: error in liftControl") e)) + $ runIter t (\x s -> return $ Left (x,s)) + (\_ e -> return $ Right e) + +te :: SomeException -> Iteratee s m a +te e = icont (const (te e)) (Just e) + +instance (Nullable s, MonadControlIO m) => MonadControlIO (Iteratee s m) where + {-# INLINE liftControlIO #-} + liftControlIO = liftLiftControlBase liftControlIO -- |Send 'EOF' to the @Iteratee@ and disregard the unconsumed part of the -- stream. If the iteratee is in an exception state, that exception is diff --git a/src/Data/Iteratee/Binary.hs b/src/Data/Iteratee/Binary.hs --- a/src/Data/Iteratee/Binary.hs +++ b/src/Data/Iteratee/Binary.hs @@ -13,6 +13,7 @@ ,endianRead3 ,endianRead3i ,endianRead4 + ,endianRead8 ) where @@ -100,6 +101,39 @@ `shiftL` 8) .|. fromIntegral c4 LSB -> return $ (((((fromIntegral c4 + `shiftL` 8) .|. fromIntegral c3) + `shiftL` 8) .|. fromIntegral c2) + `shiftL` 8) .|. fromIntegral c1 + +endianRead8 + :: (Nullable s, LL.ListLike s Word8, Monad m) => + Endian + -> Iteratee s m Word64 +endianRead8 e = do + c1 <- I.head + c2 <- I.head + c3 <- I.head + c4 <- I.head + c5 <- I.head + c6 <- I.head + c7 <- I.head + c8 <- I.head + case e of + MSB -> return $ + (((((((((((((fromIntegral c1 + `shiftL` 8) .|. fromIntegral c2) + `shiftL` 8) .|. fromIntegral c3) + `shiftL` 8) .|. fromIntegral c4) + `shiftL` 8) .|. fromIntegral c5) + `shiftL` 8) .|. fromIntegral c6) + `shiftL` 8) .|. fromIntegral c7) + `shiftL` 8) .|. fromIntegral c8 + LSB -> return $ + (((((((((((((fromIntegral c8 + `shiftL` 8) .|. fromIntegral c7) + `shiftL` 8) .|. fromIntegral c6) + `shiftL` 8) .|. fromIntegral c5) + `shiftL` 8) .|. fromIntegral c4) `shiftL` 8) .|. fromIntegral c3) `shiftL` 8) .|. fromIntegral c2) `shiftL` 8) .|. fromIntegral c1 diff --git a/src/Data/Iteratee/Iteratee.hs b/src/Data/Iteratee/Iteratee.hs --- a/src/Data/Iteratee/Iteratee.hs +++ b/src/Data/Iteratee/Iteratee.hs @@ -20,7 +20,6 @@ ,isStreamFinished -- ** Chunkwise Iteratees ,mapChunksM_ - ,mapReduce ,getChunk ,getChunks -- ** Nested iteratee combinators @@ -63,7 +62,6 @@ import Control.Exception import Control.Monad.Trans.Class -import Control.Parallel import Data.Maybe import Data.Monoid import Data.Typeable @@ -152,35 +150,6 @@ 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) - -- | Get the current chunk from the stream. getChunk :: (Monad m, Nullable s, NullPoint s) => Iteratee s m s getChunk = liftI step @@ -194,12 +163,12 @@ -- | Get a list of all chunks from the stream. getChunks :: (Monad m, Nullable s) => Iteratee s m [s] -getChunks = liftI (step []) +getChunks = liftI (step id) where step acc (Chunk xs) | nullC xs = liftI (step acc) - | otherwise = liftI (step (xs:acc)) - step acc stream = idone (reverse acc) stream + | otherwise = liftI (step $ acc . (xs:)) + step acc stream = idone (acc []) stream {-# INLINE getChunks #-} -- --------------------------------------------------- diff --git a/src/Data/Iteratee/ListLike.hs b/src/Data/Iteratee/ListLike.hs --- a/src/Data/Iteratee/ListLike.hs +++ b/src/Data/Iteratee/ListLike.hs @@ -18,6 +18,7 @@ ,dropWhile ,drop ,head + ,tryHead ,last ,heads ,peek @@ -97,23 +98,13 @@ -- |Read a stream to the end and return all of its elements as a list. -- This iteratee returns all data from the stream *strictly*. stream2list :: (Monad m, Nullable s, LL.ListLike s el) => Iteratee s m [el] -stream2list = liftI (step []) - where - step acc (Chunk ls) - | nullC ls = liftI (step acc) - | otherwise = liftI (step (acc ++ LL.toList ls)) - step acc str = idone acc str +stream2list = liftM (concatMap LL.toList) getChunks {-# INLINE stream2list #-} -- |Read a stream to the end and return all of its elements as a stream. -- This iteratee returns all data from the stream *strictly*. stream2stream :: (Monad m, Nullable s, Monoid s) => Iteratee s m s -stream2stream = icont (step mempty) Nothing - where - step acc (Chunk ls) - | nullC ls = icont (step acc) Nothing - | otherwise = icont (step (acc `mappend` ls)) Nothing - step acc str = idone acc str +stream2stream = liftM mconcat getChunks {-# INLINE stream2stream #-} @@ -146,9 +137,12 @@ -- |Attempt to read the next element of the stream and return it --- Raise a (recoverable) error if the stream is terminated +-- Raise a (recoverable) error if the stream is terminated. -- -- The analogue of @List.head@ +-- +-- Because @head@ can raise an error, it shouldn't be used when constructing +-- iteratees for @convStream@. Use @tryHead@ instead. head :: (Monad m, LL.ListLike s el) => Iteratee s m el head = liftI step where @@ -158,6 +152,17 @@ step stream = icont step (Just (setEOF stream)) {-# INLINE head #-} +-- | Similar to @head@, except it returns @Nothing@ if the stream +-- is terminated. +tryHead :: (Monad m, LL.ListLike s el) => Iteratee s m (Maybe el) +tryHead = liftI step + where + step (Chunk vec) + | LL.null vec = liftI step + | otherwise = idone (Just $ LL.head vec) (Chunk $ LL.tail vec) + step stream = idone Nothing stream +{-# INLINE tryHead #-} + -- |Attempt to read the last element of the stream and return it -- Raise a (recoverable) error if the stream is terminated -- @@ -271,7 +276,7 @@ length :: (Monad m, Num a, LL.ListLike s el) => Iteratee s m a length = liftI (step 0) where - step !i (Chunk xs) = liftI (step $! i + fromIntegral (LL.length xs)) + step !i (Chunk xs) = liftI (step $ i + fromIntegral (LL.length xs)) step !i stream = idone i stream {-# INLINE length #-} @@ -422,12 +427,7 @@ ,LooseMap s el el') => (el -> el') -> Enumeratee (s el) (s el') m a -mapStream f = eneeCheckIfDone (liftI . step) - where - step k (Chunk xs) - | LL.null xs = liftI (step k) - | otherwise = mapStream f $ k (Chunk $ lMap f xs) - step k s = idone (liftI k) s +mapStream f = mapChunks (lMap f) {-# SPECIALIZE mapStream :: Monad m => (el -> el') -> Enumeratee [el] [el'] m a #-} -- |Map the stream rigidly. @@ -439,12 +439,7 @@ :: (Monad m, LL.ListLike s el, NullPoint s) => (el -> el) -> Enumeratee s s m a -rigidMapStream f = eneeCheckIfDone (liftI . step) - where - step k (Chunk xs) - | LL.null xs = liftI (step k) - | otherwise = rigidMapStream f $ k (Chunk $ LL.rigidMap f xs) - step k s = idone (liftI k) s +rigidMapStream f = mapChunks (LL.rigidMap f) {-# SPECIALIZE rigidMapStream :: Monad m => (el -> el) -> Enumeratee [el] [el] m a #-} {-# SPECIALIZE rigidMapStream :: Monad m => (Word8 -> Word8) -> Enumeratee B.ByteString B.ByteString m a #-} @@ -454,43 +449,55 @@ -- -- The analogue of @List.filter@ filter - :: (Monad m, Nullable s, LL.ListLike s el) + :: (Monad m, Functor m, Nullable s, LL.ListLike s el) => (el -> Bool) -> Enumeratee s s m a -filter p = convStream f' - where - f' = icont step Nothing - step (Chunk xs) - | LL.null xs = f' - | otherwise = idone (LL.filter p xs) mempty - step _ = f' +filter p = convStream (LL.filter p <$> getChunk) {-# INLINE filter #-} -- |Creates an 'Enumeratee' in which elements from the stream are --- grouped into @sz@-sized blocks. The outer stream is completely --- consumed and the final block may be smaller than \sz\. +-- grouped into @sz@-sized blocks. The final block may be smaller +-- than \sz\. group :: (LL.ListLike s el, Monad m, Nullable s) => Int -- ^ size of group -> Enumeratee s [s] m a -group sz iinit = liftI $ go iinit LL.empty - where go icurr pfx (Chunk s) = case gsplit (pfx `LL.append` s) of - (full, partial) | LL.null full -> liftI $ go icurr partial - | otherwise -> do inext <- lift $ enumPure1Chunk full icurr - liftI $ go inext partial - go icurr pfx (EOF mex) - | LL.null pfx = lift . enumChunk (EOF mex) $ icurr - | otherwise = do inext <- lift $ enumPure1Chunk (LL.singleton pfx) icurr - lift . enumChunk (EOF mex) $ inext - gsplit ls = case LL.splitAt sz ls of - (g, rest) | LL.null rest -> if LL.length g == sz - then (LL.singleton g, LL.empty) - else (LL.empty, g) - | otherwise -> let (grest, leftover) = gsplit rest - g' = g `LL.cons` grest - in g' `seq` (g', leftover) -{-# INLINE group #-} +group cksz iinit = liftI (step 0 id iinit) + where + -- there are two cases to consider for performance purposes: + -- 1 - grouping lots of small chunks into bigger chunks + -- 2 - breaking large chunks into smaller pieces + -- case 2 is easier, simply split a chunk into as many pieces as necessary + -- and pass them to the inner iteratee as one list. @gsplit@ does this. + -- + -- case 1 is a bit harder, need to hold onto each chunk and coalesce them + -- after enough have been received. Currently using a difference list + -- for this, i.e ([s] -> [s]) + -- + -- not using eneeCheckIfDone because that loses final chunks at EOF + step sz pfxd icur (Chunk s) + | LL.null s = liftI (step sz pfxd icur) + | LL.length s + sz < cksz = liftI (step (sz+LL.length s) (pfxd . (s:)) icur) + | otherwise = + let (full, rest) = gsplit . mconcat $ pfxd [s] + pfxd' = if LL.null rest then id else (rest:) + in do + inext <- lift $ enumPure1Chunk full icur + liftI $ step (LL.length rest) pfxd' inext + step _ pfxd icur mErr = case pfxd [] of + [] -> idone icur mErr + rest -> do + inext <- lift $ enumPure1Chunk [mconcat rest] icur + idone inext mErr + gsplit ls = case LL.splitAt cksz ls of + (g, rest) | LL.null rest -> if LL.length g == cksz + then ([g], LL.empty) + else ([], g) + | otherwise -> let (grest, leftover) = gsplit rest + g' = g : grest + in (g', leftover) + -- | Creates an 'enumeratee' in which elements are grouped into -- contiguous blocks that are equal according to a predicate. -- @@ -522,7 +529,7 @@ xs = LL.tail l {-# INLINE groupBy #-} --- | Merge offers another way to nest iteratees: as a monad stack. +-- | @merge@ offers another way to nest iteratees: as a monad stack. -- This allows for the possibility of interleaving data from multiple -- streams. -- @@ -532,7 +539,7 @@ -- > -- > -- combine alternating lines from two sources -- > -- To see how this was derived, follow the types from --- > -- 'ileaveStream logger' and work outwards. +-- > -- 'ileaveLines logger' and work outwards. -- > run =<< enumFile 10 "file1" (joinI $ enumLinesBS $ -- > ( enumFile 10 "file2" . joinI . enumLinesBS $ joinI -- > (ileaveLines logger)) >>= run) diff --git a/src/Data/Iteratee/Parallel.hs b/src/Data/Iteratee/Parallel.hs new file mode 100644 --- /dev/null +++ b/src/Data/Iteratee/Parallel.hs @@ -0,0 +1,127 @@ +{-# LANGUAGE NoMonomorphismRestriction, BangPatterns #-} + +module Data.Iteratee.Parallel ( + psequence_ + -- ,psequence + ,parE + ,parI + ,liftParI + ,mapReduce +) + +where + +import Control.Monad.IO.Class +import Control.Monad.Trans.Class +import Data.Iteratee as I hiding (mapM_, zip, filter) +import qualified Data.ListLike as LL + +import Data.Monoid + +import Control.Concurrent +import Control.Parallel +import Control.Monad + +-- | Transform usual Iteratee into parallel composable one, introducing +-- one step extra delay. +-- +-- Ex - time spent in Enumerator working on x'th packet +-- Ix - time spent in Iteratee working on x'th packet +-- z - last packet, y = (z-1)'th packet +-- +-- regular Iteratee: E0 - I0, E1 - I1, E2 - I2 .. Ez -> Iz +-- parallel Iteratee: E0, E1, E2, .. Ez +-- \_ I0\_ I1\_ .. Iy\__ Iz +-- +parI :: (Nullable s, Monoid s) => Iteratee s IO a -> Iteratee s IO a +parI = liftI . firstStep + where + -- first step, here we fork separete thread for the next chain and at the + -- same time ask for more date from the previous chain + firstStep iter chunk = do + var <- liftIO newEmptyMVar + _ <- sideStep var chunk iter + liftI $ go var + + -- somewhere in the middle, we are getting iteratee from previous step, + -- feeding it with some new data, asking for more data and starting + -- more processing in separete thread + go var chunk@(Chunk _) = do + iter <- liftIO $ takeMVar var + _ <- sideStep var chunk iter + liftI $ go var + + -- final step - no more data, so we need to inform our consumer about it + go var e = do + iter <- liftIO $ takeMVar var + join . lift $ enumChunk e iter + + -- forks away from the main computation, return results via MVar + sideStep var chunk iter = liftIO . forkIO $ runIter iter onDone onCont + where + onDone a s = putMVar var $ idone a s + onCont k _ = runIter (k chunk) onDone onFina + onFina k e = putMVar var $ icont k e + +-- | Transform an Enumeratee into a parallel composable one, introducing +-- one step extra delay, see 'parI'. +parE :: + (Nullable s1, Nullable s2, Monoid s1) + => Enumeratee s1 s2 IO r + -> Enumeratee s1 s2 IO r +parE outer inner = parI (outer inner) + +-- | Enumerate a list of iteratees over a single stream simultaneously +-- and discard the results. Each iteratee runs in a separate forkIO thread, +-- passes all errors from iteratees up. +psequence_ :: + (LL.ListLike s el, Nullable s) + => [Iteratee s IO a] + -> Iteratee s IO () +psequence_ = I.sequence_ . map parI + + +{- +-- | Enumerate a list of iteratees over a single stream simultaneously +-- and keeps the results. Each iteratee runs in a separete forkIO thread, passes all +-- errors from iteratees up. +psequence = I.sequence . map parI +-} + +-- | A variant of 'parI' with the parallelized iteratee lifted into an +-- arbitrary MonadIO. +liftParI :: + (Nullable s, Monoid s, MonadIO m) + => Iteratee s IO a + -> Iteratee s m a +liftParI = ilift liftIO . parI + +-- | 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) + diff --git a/tests/benchmarks.hs b/tests/benchmarks.hs --- a/tests/benchmarks.hs +++ b/tests/benchmarks.hs @@ -79,6 +79,7 @@ lengthbench = makeGroup "length" listBenches takebench = makeGroup "take" $ take0 : takeBenches takeUpTobench = makeGroup "takeUpTo" takeUpToBenches +groupbench = makeGroup "group" groupBenches mapbench = makeGroup "map" $ mapBenches foldbench = makeGroup "fold" $ foldBenches convbench = makeGroup "convStream" convBenches @@ -92,15 +93,16 @@ lengthbenchbs = makeGroupBS "length" listBenches takebenchbs = makeGroupBS "take" takeBenches takeUpTobenchbs = makeGroupBS "takeUpTo" takeUpToBenches +groupbenchbs = makeGroupBS "group" groupBenches 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, foldbench, convbench, miscbench] +allListBenches = bgroup "list" [listbench, streambench, breakbench, headsbench, dropbench, lengthbench, takebench, takeUpTobench, groupbench, mapbench, foldbench, convbench, miscbench] -allByteStringBenches = bgroup "bytestring" [listbenchbs, streambenchbs, breakbenchbs, headsbenchbs, dropbenchbs, lengthbenchbs, takebenchbs, takeUpTobenchbs, mapbenchbs, foldbenchbs, convbenchbs, miscbenchbs] +allByteStringBenches = bgroup "bytestring" [listbenchbs, streambenchbs, breakbenchbs, headsbenchbs, dropbenchbs, lengthbenchbs, takebenchbs, takeUpTobenchbs, groupbenchbs, mapbenchbs, foldbenchbs, convbenchbs, miscbenchbs] list0 = makeList "list one go" deepseq list1 = BDIter1 "stream2list one go" (flip deepseq ()) stream2list @@ -167,6 +169,10 @@ takeUpTo5 = id1 "takeUpTo length long one go" (I.joinI $ I.takeUpTo 1000 I.length) takeUpTo6 = idN "takeUpTo length long chunked" (I.joinI $ I.takeUpTo 1000 I.length) takeUpToBenches = [takeUpTo1, takeUpTo2, takeUpTo3, takeUpTo4, takeUpTo5, takeUpTo6] + +group1 = id1 "group split" (I.joinI $ (I.group 24 ><> I.mapStream LL.length) I.length) +group2 = idN "group coalesce" (I.joinI $ (I.group 512 ><> I.mapStream LL.length) I.length) +groupBenches = [group1,group2] map1 = id1 "map length one go" (I.joinI $ I.rigidMapStream id I.length) map2 = idN "map length chunked" (I.joinI $ I.rigidMapStream id I.length) diff --git a/tests/testIteratee.hs b/tests/testIteratee.hs --- a/tests/testIteratee.hs +++ b/tests/testIteratee.hs @@ -21,8 +21,6 @@ import Control.Monad as CM import Control.Monad.Writer -import Text.Printf (printf) -import System.Environment (getArgs) instance Show (a -> b) where show _ = "<>" @@ -107,6 +105,11 @@ prop_head2 xs = P.length xs > 0 ==> runner1 (enumPure1Chunk xs (Iter.head >> stream2list)) == tail xs where types = xs :: [Int] +prop_tryhead xs = case xs of + [] -> runner1 (enumPure1Chunk xs tryHead) == Nothing + _ -> runner1 (enumPure1Chunk xs tryHead) == Just (P.head xs) + where types = xs :: [Int] + prop_heads xs n = n > 0 ==> runner1 (enumSpecial xs n $ heads xs) == P.length xs where types = xs :: [Int] @@ -365,6 +368,7 @@ ,testProperty "break remainder" prop_break2 ,testProperty "head" prop_head ,testProperty "head remainder" prop_head2 + ,testProperty "tryhead" prop_tryhead ,testProperty "heads" prop_heads ,testProperty "null heads" prop_heads2 ,testProperty "peek" prop_peek