conduit-algorithms 0.0.2.0 → 0.0.3.0
raw patch · 6 files changed
+150/−58 lines, 6 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ Data.Conduit.Algorithms: removeRepeatsC :: (Eq a, Monad m) => Conduit a m a
Files
- ChangeLog +4/−0
- Data/Conduit/Algorithms.hs +58/−36
- Data/Conduit/Algorithms/Async.hs +34/−14
- Data/Conduit/Algorithms/Tests.hs +33/−5
- Data/Conduit/Algorithms/Utils.hs +20/−2
- conduit-algorithms.cabal +1/−1
ChangeLog view
@@ -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.
Data/Conduit/Algorithms.hs view
@@ -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)+
Data/Conduit/Algorithms/Async.hs view
@@ -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
Data/Conduit/Algorithms/Tests.hs view
@@ -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
Data/Conduit/Algorithms/Utils.hs view
@@ -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
conduit-algorithms.cabal view
@@ -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.