diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,7 @@ +Version 0.0.3.0 2017-08-30 by luispedro + * Much improved mergeC2 function + * Add removeRepeatsC conduit + Version 0.0.2.0 2017-08-08 by luispedro * Fix haddock code generation. * Export mergeC2 function. diff --git a/Data/Conduit/Algorithms.hs b/Data/Conduit/Algorithms.hs --- a/Data/Conduit/Algorithms.hs +++ b/Data/Conduit/Algorithms.hs @@ -12,12 +12,13 @@ module Data.Conduit.Algorithms ( uniqueOnC , uniqueC + , removeRepeatsC , mergeC , mergeC2 ) where import qualified Data.Conduit as C -import qualified Data.Conduit.Combinators as CC +import qualified Data.Conduit.Internal as CI import qualified Data.Set as S import Control.Monad.Trans.Class (lift) @@ -26,9 +27,13 @@ -- | Unique conduit. -- --- Note that this conduit **does not** assume that the input is sorted. Instead +-- For each element, it checks its key (using the @a -> b@ key function) and +-- yields it if it has not seen it before. +-- +-- Note that this conduit /does not/ assume that the input is sorted. Instead -- it uses a 'Data.Set' to store previously seen elements. Thus, memory usage --- is O(N). +-- is O(N) and time is O(N log N). If the input is sorted, you can use +-- 'removeRepeatsC' uniqueOnC :: (Ord b, Monad m) => (a -> b) -> C.Conduit a m a uniqueOnC f = checkU (S.empty :: S.Set b) where @@ -38,51 +43,68 @@ else do C.yield val checkU (S.insert (f val) cur) --- | See 'uniqueOnC' +-- | Unique conduit +-- +-- See 'uniqueOnC' and 'removeRepeatsC' uniqueC :: (Ord a, Monad m) => C.Conduit a m a uniqueC = uniqueOnC id --- | Merge a list of sorted sources +-- | Removes repeated elements -- +-- @ +-- yieldMany [0, 0, 1, 1, 1, 2, 2, 0] .| removeRepeatsC .| consume +-- @ +-- +-- is equivalent to @[0, 1, 2, 0]@ +-- +-- See 'uniqueC' and 'uniqueOnC' +removeRepeatsC :: (Eq a, Monad m) => C.Conduit a m a +removeRepeatsC = awaitJust removeRepeatsC' + where + removeRepeatsC' prev = C.await >>= \case + Nothing -> C.yield prev + Just next + | next == prev -> removeRepeatsC' prev + | otherwise -> do + C.yield prev + removeRepeatsC' next + + +-- | Merge a list of sorted sources to produce a single (sorted) source +-- +-- This takes a list of sorted sources and produces a 'C.Source' which outputs +-- all elements in sorted order. +-- -- See 'mergeC2' mergeC :: (Ord a, Monad m) => [C.Source m a] -> C.Source m a mergeC [] = return () mergeC [s] = s mergeC [a,b] = mergeC2 a b -mergeC args = let (a,b) = split2 args in mergeC2 (mergeC a) (mergeC b) +mergeC args = mergeC2 (mergeC right) (mergeC left) where - split2 :: [a] -> ([a],[a]) - split2 [] = ([], []) - split2 [a] = ([a], []) - split2 [a,b] = ([a], [b]) - split2 (x:y:rs) = let (xs,ys) = split2 rs in (x:xs, y:ys) + right = take n args + left = drop n args + n = (length args) `div` 2 -- | Take two sorted sources and merge them. -- -- See 'mergeC' mergeC2 :: (Ord a, Monad m) => C.Source m a -> C.Source m a -> C.Source m a -mergeC2 s1 s2 = do - (c1', e1) <- lift $ s1 C.$$+ CC.head - (c2', e2) <- lift $ s2 C.$$+ CC.head - continue c1' c2' e1 e2 - where - continue :: (Monad m, Ord a) => C.ResumableSource m a -> C.ResumableSource m a -> Maybe a -> Maybe a -> C.Source m a - continue _ _ Nothing Nothing = return () - continue _ c Nothing e = continue c undefined e Nothing - continue c _ (Just e) Nothing = do - C.yield e - yieldAll c - continue c1 c2 je1@(Just e1) je2@(Just e2) - | compare e1 e2 == GT = continue c2 c1 je2 je1 - | otherwise = do - C.yield e1 - (c1', e1') <- lift $ c1 C.$$++ CC.head - continue c1' c2 e1' (Just e2) - yieldAll :: (Monad m) => C.ResumableSource m a -> C.Source m a - yieldAll rs = do - (rs',v) <- lift $ rs C.$$++ CC.head - case v of - Just v' -> do - C.yield v' - yieldAll rs' - Nothing -> return () +mergeC2 (CI.ConduitM s1) (CI.ConduitM s2) = CI.ConduitM $ \rest -> let + go right@(CI.HaveOutput s1' f1 v1) left@(CI.HaveOutput s2' f2 v2) + | compare v1 v2 /= GT = CI.HaveOutput (go s1' left) (f1 >> f2) v1 + | otherwise = CI.HaveOutput (go right s2') (f1 >> f2) v2 + go right CI.Done{} = right + go CI.Done{} left = left + go (CI.PipeM p) left = do + next <- lift p + go next left + go right (CI.PipeM p) = do + next <- lift p + go right next + go (CI.NeedInput _ next) left = go (next ()) left + go right (CI.NeedInput _ next) = go right (next ()) + go (CI.Leftover next ()) left = go next left + go right (CI.Leftover next ()) = go right next + in go (s1 rest) (s2 rest) + diff --git a/Data/Conduit/Algorithms/Async.hs b/Data/Conduit/Algorithms/Async.hs --- a/Data/Conduit/Algorithms/Async.hs +++ b/Data/Conduit/Algorithms/Async.hs @@ -50,16 +50,31 @@ --- | This is like Data.Conduit.List.map, except that each element is processed --- in a separate thread (up to maxSize can be queued up at any one time). --- Results are evaluated to normal form (not WHNF!) to ensure that the --- computation is fully evaluated before being yielded to the next conduit. -asyncMapC :: forall a m b . (MonadIO m, NFData b) => Int -> (a -> b) -> C.Conduit a m b -asyncMapC maxSize f = initLoop (0 :: Int) (Seq.empty :: Seq.Seq (A.Async b)) +-- | This is like 'Data.Conduit.List.map', except that each element is processed +-- in a separate thread (up to 'maxThreads' can be queued up at any one time). +-- Results are evaluated to normal form (not weak-head normal form!, i.e., the +-- structure is deeply evaluated) to ensure that the computation is fully +-- evaluated in the worker thread. +-- +-- Note that there is some overhead in threading. It is often a good idea to +-- build larger chunks of input before passing it to 'asyncMapC' to amortize +-- the costs. That is, when @f@ is not a lot of work, instead of @asyncMapC f@, +-- it is sometimes better to do +-- +-- @ +-- CC.conduitVector 4096 .| asyncMapC (V.map f) .| CC.concat +-- @ +-- +-- where @CC@ refers to 'Data.Conduit.Combinators' +asyncMapC :: forall a m b . (MonadIO m, NFData b) => + Int -- ^ Maximum number of worker threads + -> (a -> b) -- ^ Function to execute + -> C.Conduit a m b +asyncMapC maxThreads f = initLoop (0 :: Int) (Seq.empty :: Seq.Seq (A.Async b)) where initLoop :: Int -> Seq.Seq (A.Async b) -> C.Conduit a m b initLoop size q - | size == maxSize = loop q + | size == maxThreads = loop q | otherwise = C.await >>= \case Nothing -> yAll q Just v -> do @@ -89,13 +104,15 @@ yieldOrCleanup q = flip C.yieldOr (cleanup q) --- | asyncMapC with error handling. The inner function can now return an error --- (as a 'Left'). When the first error is seen, it 'throwError's in the main --- monad. Note that 'f' may be evaluated for arguments beyond the first error --- (as some threads may be running in the background and already processing --- elements after the first error). +-- | 'asyncMapC' with error handling. The inner function can now return an +-- error (as a 'Left'). When the first error is seen, it 'throwError's in the +-- main monad. Note that 'f' may be evaluated for arguments beyond the first +-- error (as some threads may be running in the background and already +-- processing elements after the first error). +-- +-- See 'asyncMapC' asyncMapEitherC :: forall a m b e . (MonadIO m, NFData b, NFData e, MonadError e m) => Int -> (a -> Either e b) -> C.Conduit a m b -asyncMapEitherC maxSize f = asyncMapC maxSize f .| (C.awaitForever $ \case +asyncMapEitherC maxThreads f = asyncMapC maxThreads f .| (C.awaitForever $ \case Right v -> C.yield v Left err -> throwError err) @@ -126,7 +143,8 @@ untilNothing _ -> return () --- | A simple sink which performs gzip in a separate thread and writes the results to `h`. +-- | A simple sink which performs gzip compression in a separate thread and +-- writes the results to `h`. -- -- See also 'asyncGzipToFile' asyncGzipTo :: forall m. (MonadIO m, MonadBaseControl IO m) => Handle -> C.Sink B.ByteString m () @@ -179,6 +197,8 @@ -- | If the filename indicates a gzipped file (or, on Unix, also a bz2 file), -- then it reads it and uncompresses it. +-- +-- On Windows, attempting to read from a bzip2 file, results in 'error'. -- -- For the case of gzip, 'asyncGzipFromFile' is used. conduitPossiblyCompressedFile :: (MonadBaseControl IO m, MonadResource m) => FilePath -> C.Source m B.ByteString diff --git a/Data/Conduit/Algorithms/Tests.hs b/Data/Conduit/Algorithms/Tests.hs --- a/Data/Conduit/Algorithms/Tests.hs +++ b/Data/Conduit/Algorithms/Tests.hs @@ -12,11 +12,12 @@ import qualified Data.Conduit as C import qualified Data.Conduit.Combinators as CC import Data.Conduit ((.|)) +import Data.List (sort) +import System.Directory (removeFile) import qualified Data.Conduit.Algorithms as CAlg import qualified Data.Conduit.Algorithms.Utils as CAlg import qualified Data.Conduit.Algorithms.Async as CAlg -import System.Directory (removeFile) main :: IO () main = $(defaultMainGenerator) @@ -28,19 +29,46 @@ extractIO c = C.runConduitRes (c .| CC.sinkList) +shouldProduce values cond = extract cond @?= values + case_uniqueC = extract (CC.yieldMany [1,2,3,1,1,2,3] .| CAlg.uniqueC) @=? [1,2,3 :: Int] -case_mergeC = extract (CAlg.mergeC [CC.yieldMany [0, 2, 4], CC.yieldMany [1,3,4,5]]) @=? [0,1,2,3,4,4,5 :: Int] -case_groupC = extract (CC.yieldMany [0..10] .| CAlg.groupC 3) @=? [[0,1,2], [3,4,5], [6,7,8], [9, 10 :: Int]] +case_mergeC = shouldProduce expected $ + CAlg.mergeC + [ CC.yieldMany i1 + , CC.yieldMany i2 + , CC.yieldMany i3 + ] + where + expected = sort (concat [i1, i2, i3]) + i1 = [ 0, 2, 4 :: Int] + i2 = [ 1, 3, 4, 5] + i3 = [-1, 0, 7] +case_mergeC2 = shouldProduce [0, 1, 1, 2, 3, 5 :: Int] $ + CAlg.mergeC2 + (CC.yieldMany [0, 1, 2]) + (CC.yieldMany [1, 3, 5]) + +case_mergeC2same = shouldProduce [0, 0, 1, 1, 2, 2 :: Int] $ + CAlg.mergeC2 + (CC.yieldMany [0, 1, 2]) + (CC.yieldMany [0, 1, 2]) + +case_groupC = shouldProduce [[0,1,2], [3,4,5], [6,7,8], [9, 10 :: Int]] $ + CC.yieldMany [0..10] .| CAlg.groupC 3 + +case_removeRepeatsC = shouldProduce [0,1,2,3,4,5,6,7,8,9, 10 :: Int] $ + CC.yieldMany [0,0,0,1,1,1,2,2,3,4,5,6,6,6,6,7,7,8,9,10,10] .| CAlg.removeRepeatsC + case_asyncMap :: IO () case_asyncMap = do vals <- extractIO (CC.yieldMany [0..10] .| CAlg.asyncMapC 3 (+ (1:: Int))) - (vals @=? [1..11]) + (vals @?= [1..11]) case_asyncGzip :: IO () case_asyncGzip = do C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncGzipToFile testingFileNameGZ) r <- B.concat <$> (extractIO (CAlg.asyncGzipFromFile testingFileNameGZ)) - r @=? "Hello World" + r @?= "Hello World" removeFile testingFileNameGZ diff --git a/Data/Conduit/Algorithms/Utils.hs b/Data/Conduit/Algorithms/Utils.hs --- a/Data/Conduit/Algorithms/Utils.hs +++ b/Data/Conduit/Algorithms/Utils.hs @@ -4,7 +4,7 @@ License : MIT Maintainer : luis@luispedro.org -A few miscellaneous set of conduit utilities +A few miscellaneous conduit utils -} module Data.Conduit.Algorithms.Utils ( awaitJust @@ -15,13 +15,31 @@ import Data.Maybe (maybe) import Control.Monad (unless) --- | This is a simple utility adapted from +-- | Act on the next input (do nothing if no input). @awaitJust f@ is equivalent to +-- +-- +-- @ do +-- next <- C.await +-- case next of +-- Just val -> f val +-- Nothing -> return () +-- @ +-- +-- This is a simple utility adapted from -- http://neilmitchell.blogspot.de/2015/07/thoughts-on-conduits.html awaitJust :: Monad m => (a -> C.Conduit a m b) -> C.Conduit a m b awaitJust f = C.await >>= maybe (return ()) f -- | groupC yields the input as groups of 'n' elements. If the input is not a -- multiple of 'n', the last element will be incomplete +-- +-- Example: +-- +-- @ +-- CC.yieldMany [0..10] .| groupC 3 .| CC.consumeList +-- @ +-- +-- results in @[ [0,1,2], [3,4,5], [6,7,8], [9, 10] ]@ groupC :: (Monad m) => Int -> C.Conduit a m [a] groupC n = loop n [] where diff --git a/conduit-algorithms.cabal b/conduit-algorithms.cabal --- a/conduit-algorithms.cabal +++ b/conduit-algorithms.cabal @@ -1,5 +1,5 @@ name: conduit-algorithms -version: 0.0.2.0 +version: 0.0.3.0 synopsis: Conduit-based algorithms description: Algorithms on Conduits, including higher level asynchronous processing and some other utilities.