iteratee 0.8.5.0 → 0.8.6.0

raw patch · 9 files changed

+265/−96 lines, 9 filesdep +monad-controlPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: monad-control

API changes (from Hackage documentation)

Files

• CONTRIBUTORS +1/−0
• iteratee.cabal +3/−1
• src/Data/Iteratee/Base.hs +21/−2
• src/Data/Iteratee/Binary.hs +34/−0
• src/Data/Iteratee/Iteratee.hs +3/−34
• src/Data/Iteratee/ListLike.hs +62/−55
• src/Data/Iteratee/Parallel.hs +127/−0
• tests/benchmarks.hs +8/−2
• tests/testIteratee.hs +6/−2

@@ -1,6 +1,7 @@ Thanks to the following individuals for contributing to this project.  Oleg Kiselyov+Michael Baikov Gregory Collins Nick Ingolia Brian Lewis

@@ -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

@@ -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

@@ -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

@@ -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 #-}  -- ---------------------------------------------------

@@ -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)

@@ -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)+

@@ -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)

@@ -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 _ = "<<function>>"@@ -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