diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,6 @@ +Version 0.0.7.1 2018-01-18 by luispedro + * Fix testing in newer stack/cabal/hpack (issue #3) + Version 0.0.7.0 2018-01-17 by luispedro * Add unorderedAsyncMapC * Add Data.Conduit.Algorithms.Async.ByteString module diff --git a/Data/Conduit/Algorithms.hs b/Data/Conduit/Algorithms.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Algorithms.hs @@ -0,0 +1,130 @@ +{-| +Module : Data.Conduit.Algorithms +Copyright : 2013-2017 Luis Pedro Coelho +License : MIT +Maintainer : luis@luispedro.org + +Simple algorithms packaged as Conduits +-} + + +{-# LANGUAGE Rank2Types #-} +module Data.Conduit.Algorithms + ( uniqueOnC + , uniqueC + , removeRepeatsC + , mergeC + , mergeC2 + ) where + +import qualified Data.Conduit as C +import qualified Data.Conduit.Internal as CI +import qualified Data.Set as S +import Data.List (sortBy, insertBy) +import Control.Monad.Trans.Class (lift) + +import Data.Conduit.Algorithms.Utils (awaitJust) + + +-- | Unique conduit. +-- +-- 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) 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 + checkU cur = awaitJust $ \val -> + if f val `S.member` cur + then checkU cur + else do + C.yield val + checkU (S.insert (f val) cur) +-- | Unique conduit +-- +-- See 'uniqueOnC' and 'removeRepeatsC' +uniqueC :: (Ord a, Monad m) => C.Conduit a m a +uniqueC = uniqueOnC id + +-- | 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 [a] = a +mergeC [a,b] = mergeC2 a b +mergeC cs = CI.ConduitM $ \rest -> let + go [] = rest () + go allc@(CI.HaveOutput c_next _ v:larger) = + CI.HaveOutput (norm1 c_next >>= go . insert1 larger) (finalizeAll allc) v + go _ = error "This situation should have been impossible (mergeC/go)" + insert1 larger CI.Done{} = larger + insert1 larger c = insertBy compareHO c larger + norm1 :: Monad m => CI.Pipe () i o () m () -> CI.Pipe () i o () m (CI.Pipe () i o () m ()) + norm1 c@CI.HaveOutput{} = return c + norm1 c@CI.Done{} = return c + norm1 (CI.PipeM p) = lift p >>= norm1 + norm1 (CI.NeedInput _ next) = norm1 (next ()) + norm1 (CI.Leftover next ()) = norm1 next + isHO CI.HaveOutput{} = True + isHO _ = False + compareHO (CI.HaveOutput _ _ a) (CI.HaveOutput _ _ b) = compare a b + compareHO _ _ = error "This situation should have been impossible (mergeC/compareHO)" + finalizeAll [] = return () + finalizeAll (CI.HaveOutput _ f _ : larger) = f >> finalizeAll larger + finalizeAll (_ :larger) = finalizeAll larger + in do + let st = map (($ CI.Done) . CI.unConduitM) cs + st' <- mapM norm1 st + go . sortBy compareHO . filter isHO $ st' + + +-- | 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 (CI.ConduitM s1) (CI.ConduitM s2) = CI.ConduitM $ \rest -> let + go right@(CI.HaveOutput s1' f1 v1) left@(CI.HaveOutput s2' f2 v2) + | v1 <= v2 = CI.HaveOutput (go s1' left) (f1 >> f2) v1 + | otherwise = CI.HaveOutput (go right s2') (f1 >> f2) v2 + go right@CI.Done{} (CI.HaveOutput s f v) = CI.HaveOutput (go right s) f v + go (CI.HaveOutput s f v) left@CI.Done{} = CI.HaveOutput (go s left) f v + go CI.Done{} CI.Done{} = rest () + 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 CI.Done) (s2 CI.Done) diff --git a/Data/Conduit/Algorithms/Async.hs b/Data/Conduit/Algorithms/Async.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Algorithms/Async.hs @@ -0,0 +1,246 @@ +{-| +Module : Data.Conduit.Algorithms.Async +Copyright : 2013-2017 Luis Pedro Coelho +License : MIT +Maintainer : luis@luispedro.org + +Higher level async processing interfaces. +-} +{-# LANGUAGE ScopedTypeVariables, FlexibleContexts, CPP, TupleSections #-} + +module Data.Conduit.Algorithms.Async + ( conduitPossiblyCompressedFile + , asyncMapC + , asyncMapEitherC + , asyncGzipTo + , asyncGzipToFile + , asyncGzipFrom + , asyncGzipFromFile + , unorderedAsyncMapC + ) where + + +import qualified Data.ByteString as B +import qualified Control.Concurrent.Async as A +import qualified Control.Concurrent.STM.TBQueue as TQ +import Control.Concurrent.STM (atomically) + +import qualified Data.Conduit.Combinators as C +import qualified Data.Conduit.Async as CA +import qualified Data.Conduit.TQueue as CA +import qualified Data.Conduit.List as CL +import qualified Data.Conduit.Zlib as CZ +import qualified Data.Conduit.Lzma as CX +#ifndef WINDOWS +-- bzlib cannot compile on Windows (as of 2016/07/05) +import qualified Data.Conduit.BZlib as CZ +#endif +import qualified Data.Conduit as C +import Data.Conduit ((.|)) + +import qualified Data.Sequence as Seq +import Data.Sequence ((|>), ViewL(..)) +import Data.Foldable (toList) +import Control.Monad (forM_) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Error.Class (MonadError(..)) +import Control.Monad.Trans.Resource (MonadResource, MonadBaseControl) +import Control.Exception (evaluate) +import Control.DeepSeq +import System.IO +import Data.List (isSuffixOf) +import Data.Conduit.Algorithms.Utils (awaitJust) + + + +-- | 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' +-- +-- See 'unorderedAsyncMapC' +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 = asyncMapCHelper True + +-- | A version of 'asyncMapC' which can reorder results in the stream +-- +-- If the order of the results is not important, this function can lead to a +-- better use of resources if some of the chunks take longer to complete. +-- +-- See 'asyncMapC' +unorderedAsyncMapC :: forall a m b . (MonadIO m, NFData b) => + Int -- ^ Maximum number of worker threads + -> (a -> b) -- ^ Function to execute + -> C.Conduit a m b +unorderedAsyncMapC = asyncMapCHelper False + +asyncMapCHelper :: forall a m b . (MonadIO m, NFData b) => + Bool + -> Int -- ^ Maximum number of worker threads + -> (a -> b) -- ^ Function to execute + -> C.Conduit a m b +asyncMapCHelper isSynchronous 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 == maxThreads = loop q + | otherwise = C.await >>= \case + Nothing -> yAll q + Just v -> do + v' <- sched v + initLoop (size + 1) (q |> v') + sched :: a -> C.ConduitM a b m (A.Async b) + sched v = liftIO . A.async . evaluate . force $ f v + + -- | yield all + yAll :: Seq.Seq (A.Async b) -> C.Conduit a m b + yAll q = case Seq.viewl q of + EmptyL -> return () + v :< rest -> (liftIO (A.wait v) >>= yieldOrCleanup rest) >> yAll rest + + loop :: Seq.Seq (A.Async b) -> C.Conduit a m b + loop q = C.await >>= \case + Nothing -> yAll q + Just v -> do + v' <- sched v + (r, q') <- liftIO $ retrieveResult q + yieldOrCleanup q' r + loop (q' |> v') + cleanup :: Seq.Seq (A.Async b) -> m () + cleanup q = liftIO $ forM_ q A.cancel + yieldOrCleanup q = flip C.yieldOr (cleanup q) + + retrieveResult :: Seq.Seq (A.Async b) -> IO (b, Seq.Seq (A.Async b)) + retrieveResult q + | isSynchronous = case Seq.viewl q of + (r :< rest) -> (, rest) <$> A.wait r + _ -> error "Impossible situation" + | otherwise = do + (k, r) <- liftIO (A.waitAny (toList q)) + return (r, Seq.filter (/= k) 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). +-- +-- 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 maxThreads f = asyncMapC maxThreads f .| (C.awaitForever $ \case + Right v -> C.yield v + Left err -> throwError err) + + +-- | concatenates input into larger chunks and yields it. Its indended use is +-- to build up larger blocks from smaller ones so that they can be sent across +-- thread barriers with little overhead. +-- +-- the chunkSize parameter is a hint, not an exact element. In particular, +-- larger chunks are not split up and smaller chunks can be yielded too. +bsConcatTo :: MonadIO m => Int -- ^ chunk hint + -> C.Conduit B.ByteString m [B.ByteString] +bsConcatTo chunkSize = awaitJust start + where + start v + | B.length v >= chunkSize = C.yield [v] >> bsConcatTo chunkSize + | otherwise = continue [v] (B.length v) + continue chunks s = C.await >>= \case + Nothing -> C.yield chunks + Just v + | B.length v + s > chunkSize -> C.yield chunks >> start v + | otherwise -> continue (v:chunks) (s + B.length v) + +untilNothing :: forall m i. (Monad m) => C.Conduit (Maybe i) m i +untilNothing = C.await >>= \case + Just (Just val) -> do + C.yield val + untilNothing + _ -> return () + +-- | 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 () +asyncGzipTo h = do + let drain q = liftIO . C.runConduit $ + CA.sourceTBQueue q + .| untilNothing + .| CL.map (B.concat . reverse) + .| CZ.gzip + .| C.sinkHandle h + bsConcatTo ((2 :: Int) ^ (15 :: Int)) + .| CA.drainTo 8 drain + +-- | Compresses the output and writes to the given file with compression being +-- performed in a separate thread. +-- +-- See also 'asyncGzipTo' +asyncGzipToFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Sink B.ByteString m () +asyncGzipToFile fname = C.bracketP + (openFile fname WriteMode) + hClose + asyncGzipTo + +-- | A source which produces the ungzipped content from the the given handle. +-- Note that this "reads ahead" so if you do not use all the input, the Handle +-- will probably be left at an undefined position in the file. +-- +-- See also 'asyncGzipFromFile' +asyncGzipFrom :: forall m. (MonadIO m, MonadResource m, MonadBaseControl IO m) => Handle -> C.Source m B.ByteString +asyncGzipFrom h = do + let prod q = liftIO $ do + C.runConduit $ + C.sourceHandle h + .| CZ.multiple CZ.ungzip + .| CL.map Just + .| CA.sinkTBQueue q + atomically (TQ.writeTBQueue q Nothing) + CA.gatherFrom 8 prod + .| untilNothing + +-- | Open and read a gzip file with the uncompression being performed in a +-- separate thread. +-- +-- See also 'asyncGzipFrom' +asyncGzipFromFile :: forall m. (MonadResource m, MonadBaseControl IO m) => FilePath -> C.Source m B.ByteString +asyncGzipFromFile fname = C.bracketP + (openFile fname ReadMode) + hClose + asyncGzipFrom + +-- | 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 +conduitPossiblyCompressedFile fname + | ".gz" `isSuffixOf` fname = asyncGzipFromFile fname + | ".xz" `isSuffixOf` fname = C.sourceFile fname .| CX.decompress oneGBmembuffer +#ifndef WINDOWS + | ".bz2" `isSuffixOf` fname = C.sourceFile fname .| CZ.bunzip2 +#else + | ".bz2" `isSuffixOf` fname = error "bzip2 decompression is not available on Windows" +#endif + | otherwise = C.sourceFile fname + where oneGBmembuffer = Just $ 1024 * 1024 * 1024 + diff --git a/Data/Conduit/Algorithms/Async/ByteString.hs b/Data/Conduit/Algorithms/Async/ByteString.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Algorithms/Async/ByteString.hs @@ -0,0 +1,70 @@ +{-| +Module : Data.Conduit.Algorithms.Async.ByteString +Copyright : 2018 Luis Pedro Coelho +License : MIT +Maintainer : luis@luispedro.org + +Higher level async processing interfaces. +-} +{-# LANGUAGE ScopedTypeVariables, FlexibleContexts, TupleSections #-} + +module Data.Conduit.Algorithms.Async.ByteString + ( asyncMapLineGroupsC + , asyncFilterLinesC + ) where + + +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as BL +import qualified Data.Conduit.Algorithms.Async as CAlg +import qualified Data.Conduit.List as CL +import qualified Data.Conduit as C +import Data.Conduit ((.|)) + +import Control.Monad (unless) +import Control.Monad.IO.Class (MonadIO) +import Control.DeepSeq + + +-- | Apply a function to groups of lines +-- +-- Note that this is much more efficient than the (more or less equivalent, +-- except that the intermediate lists can be of varying sizes): +-- +-- @ +-- CB.lines .| CC.conduitVector N .| CAlg.asyncMapC nthreads (f . V.toList) +-- @ +-- +-- The reason being that splitting into lines then becomes the bottleneck and +-- processing a single line is typically a tiny chunk of work so that the +-- threading overhead overwhelms the advantage of using multiple cores. +-- Instead, 'asyncMapLineGroupsC' will pass big chunks to the worker thread and +-- perform most of the line splitting _in the worker thread_. +-- +-- Only Unix-style ASCII lines are supported (splitting at Bytes with value +-- 10, i.e., \n). When Windows lines (\r\n) are passed to this function, this +-- results in each element having an extra \r at the end. +asyncMapLineGroupsC :: (MonadIO m, NFData a) => Int -> ([B.ByteString] -> a) -> C.Conduit B.ByteString m a +asyncMapLineGroupsC nthreads f = breakAtLineBoundary .| CAlg.asyncMapC nthreads (f . asLines) + where + asLines :: BL.ByteString -> [B.ByteString] + asLines = fmap BL.toStrict . BL.split 10 + + -- The purpose is to break input blocks at a line boundary + breakAtLineBoundary :: Monad m => C.Conduit B.ByteString m BL.ByteString + breakAtLineBoundary = continue BL.empty + continue prev = C.await >>= \case + Nothing -> unless (BL.null prev) $ + C.yield prev + Just n -> case B.elemIndexEnd 10 n of + Nothing -> continue (BL.append prev (BL.fromStrict n)) + Just p -> do + let (first, rest) = B.splitAt p n + C.yield (BL.append prev (BL.fromStrict first)) + continue (BL.fromStrict $ B.drop 1 rest) -- skip \n char + +-- | Filter lines using multiple threads +asyncFilterLinesC :: MonadIO m => Int -> (B.ByteString -> Bool) -> C.Conduit B.ByteString m B.ByteString +asyncFilterLinesC n f = asyncMapLineGroupsC n (filter f) .| CL.concat +{-# INLINE asyncFilterLinesC #-} + diff --git a/Data/Conduit/Algorithms/Storable.hs b/Data/Conduit/Algorithms/Storable.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Algorithms/Storable.hs @@ -0,0 +1,68 @@ +{-| +Module : Data.Conduit.Algorithms.Storable +Copyright : 2018 Luis Pedro Coelho +License : MIT +Maintainer : luis@luispedro.org + +Higher level async processing interfaces. +-} +{-# LANGUAGE FlexibleContexts, ScopedTypeVariables #-} +module Data.Conduit.Algorithms.Storable + ( writeStorableV + , readStorableV + ) where + +import qualified Data.ByteString as B +import qualified Data.ByteString.Unsafe as BU +import qualified Data.Vector.Storable as VS +import qualified Data.Vector.Storable.Mutable as VSM +import Control.Monad.IO.Class + +import Foreign.Ptr +import Foreign.Marshal.Utils +import Foreign.Storable +import Control.Monad (when) + +import qualified Data.Conduit.List as CL +import qualified Data.Conduit.Combinators as CC +import qualified Data.Conduit as C +import Data.Conduit ((.|)) + +-- | write a Storable vector +-- +-- This uses the same format as in-memory +-- +-- See |readStorableV| +writeStorableV :: forall m a. (MonadIO m, Monad m, Storable a) => C.Conduit (VS.Vector a) m B.ByteString +writeStorableV = CL.mapM (liftIO. encodeStorable') + where + encodeStorable' :: Storable a => VS.Vector a -> IO B.ByteString + encodeStorable' v' = VS.unsafeWith v' $ \p -> + B.packCStringLen (castPtr p, VS.length v' * (sizeOf (undefined :: a))) + + +-- | read a Storable vector +-- +-- This expects the same format as the in-memory vector +-- +-- See |writeStorableV| +readStorableV :: forall m a. (MonadIO m, Storable a) => Int -> C.ConduitM B.ByteString (VS.Vector a) m () +readStorableV nelems = CC.chunksOfE blockBytes .| parseBlocks + where + blockBytes = nelems * (sizeOf a') + a' :: a + a' = undefined + + + parseBlocks :: MonadIO m => C.Conduit B.ByteString m (VS.Vector a) + parseBlocks = C.awaitForever $ \bs -> do + let (n,rest) = B.length bs `divMod` sizeOf a' + r <- liftIO $ do + v <- VSM.new n + BU.unsafeUseAsCStringLen bs $ \(p, _) -> + VSM.unsafeWith v $ \vp -> + moveBytes (castPtr vp) p (n * sizeOf a') + VS.unsafeFreeze v + C.yield r + when (rest > 0) $ do + C.leftover (B.drop (n * sizeOf a') bs) diff --git a/Data/Conduit/Algorithms/Tests.hs b/Data/Conduit/Algorithms/Tests.hs deleted file mode 100644 --- a/Data/Conduit/Algorithms/Tests.hs +++ /dev/null @@ -1,149 +0,0 @@ -{- Copyright 2017-2018 Luis Pedro Coelho - - License: MIT - -} -{-# LANGUAGE TemplateHaskell, QuasiQuotes, FlexibleContexts, OverloadedStrings #-} -module Main where - -import Test.Framework.TH -import Test.HUnit -import Test.Framework.Providers.HUnit - -import qualified Data.ByteString as B -import qualified Data.ByteString.Char8 as B8 -import qualified Data.Conduit as C -import qualified Data.Conduit.Combinators as CC -import qualified Data.Conduit.Binary as CB -import qualified Data.Conduit.List as CL -import qualified Data.Vector.Storable as VS -import Data.Conduit ((.|)) -import Data.List (sort) -import System.Directory (removeFile) -import Control.Monad (forM_) - -import qualified Data.Conduit.Algorithms as CAlg -import qualified Data.Conduit.Algorithms.Storable as CAlg -import qualified Data.Conduit.Algorithms.Utils as CAlg -import qualified Data.Conduit.Algorithms.Async as CAlg -import qualified Data.Conduit.Algorithms.Async.ByteString as CAlg - -main :: IO () -main = $(defaultMainGenerator) - -testingFileNameGZ :: FilePath -testingFileNameGZ = "file_just_for_testing_delete_me_please.gz" -testingFileNameGZ2 :: FilePath -testingFileNameGZ2 = "file_just_for_testing_delete_me_please_2.gz" - -extract c = C.runConduitPure (c .| CC.sinkList) - -extractIO c = C.runConduitRes (c .| CC.sinkList) - -shouldProduce values cond = extract cond @?= values -shouldProduceIO values cond = do - p <- extractIO cond - p @?= values - -case_uniqueC = extract (CC.yieldMany [1,2,3,1,1,2,3] .| CAlg.uniqueC) @=? [1,2,3 :: Int] -case_mergeC = shouldProduce expected $ - CAlg.mergeC - [ CC.yieldMany i1 - , CC.yieldMany i2 - , CC.yieldMany i3 - , CC.yieldMany i3 - ] - where - expected = sort (concat [i1, i2, i3, i3]) - i1 = [ 1, 2, 4 :: Int] - i2 = [ 1, 4, 4, 5] - i3 = [-1, 0, 7] - -case_mergeCmonad = shouldProduce expected $ - CAlg.mergeC - [ mYield i1 - , mYield i2 - , mYield i3 - ] - where - expected = sort (concat [i1, i2, i3]) - mYield lst = do - let lst' = map return lst - forM_ lst' $ \elem -> do - elem' <- elem - C.yield elem' - 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_mergeC2monad = shouldProduce [0, 1, 2, 2, 3, 4 :: Int] $ do - CAlg.mergeC2 - (CC.yieldMany [0, 2]) - (CC.yieldMany [1, 2]) - CC.yieldMany [3] - CC.yieldMany [4] - -case_groupC = shouldProduce [[0,1,2], [3,4,5], [6,7,8], [9, 10 :: Int]] $ - CC.yieldMany [0..10] .| CAlg.groupC 3 - -case_enumerateC = shouldProduce [(0,'z'), (1,'o'), (2,'t')] $ - CC.yieldMany ("zot" :: [Char]) .| CAlg.enumerateC - -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]) - -case_unorderedAsyncMapC :: IO () -case_unorderedAsyncMapC = do - vals <- extractIO (CC.yieldMany [0..10] .| CAlg.unorderedAsyncMapC 3 (+ (1:: Int))) - (sort 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" - removeFile testingFileNameGZ - - -case_async_gzip_to_from = do - let testdata = [0 :: Int .. 12] - C.runConduitRes $ - CC.yieldMany testdata - .| CL.map (B8.pack . (\n -> show n ++ "\n")) - .| CAlg.asyncGzipToFile testingFileNameGZ - C.runConduitRes $ - CAlg.asyncGzipFromFile testingFileNameGZ - .| CAlg.asyncGzipToFile testingFileNameGZ2 - shouldProduceIO testdata $ - CAlg.asyncGzipFromFile testingFileNameGZ2 - .| CB.lines - .| CL.map (read . B8.unpack) - removeFile testingFileNameGZ - removeFile testingFileNameGZ2 - -case_asyncFilterLines = do - vals <- extractIO (CC.yieldMany ["This is\nMy data\nBut"," sometimes","\nit is split,\n","in weird ways."] .| CAlg.asyncFilterLinesC 2 (B8.notElem ',')) - (vals @?= ["This is", "My data", "But sometimes", "in weird ways."]) - -case_asyncFilterLinesAllTrue = do - vals <- extractIO (CC.yieldMany ["This is\nMy data\nBut"," sometimes","\nit is split,\n","in weird ways."] .| CAlg.asyncFilterLinesC 2 (const True)) - (vals @?= ["This is", "My data", "But sometimes", "it is split,", "in weird ways."]) - -case_storableVector = do - let v = VS.fromList [0:: Int, 1, 2, 4, 6, 12] - vals <- extractIO (CC.yieldMany [v,v,v] .| CAlg.writeStorableV .| CAlg.readStorableV 3) - (VS.concat vals @=? VS.concat [v,v,v]) diff --git a/Data/Conduit/Algorithms/Utils.hs b/Data/Conduit/Algorithms/Utils.hs new file mode 100644 --- /dev/null +++ b/Data/Conduit/Algorithms/Utils.hs @@ -0,0 +1,63 @@ +{-| +Module : Data.Conduit.Algorithms.Utils +Copyright : 2013-2017 Luis Pedro Coelho +License : MIT +Maintainer : luis@luispedro.org + +A few miscellaneous conduit utils +-} +module Data.Conduit.Algorithms.Utils + ( awaitJust + , enumerateC + , groupC + ) where + +import qualified Data.Conduit as C +import Data.Maybe (maybe) +import Control.Monad (unless) + +-- | 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 +{-# INLINE awaitJust #-} + +-- | Conduit analogue to Python's enumerate function +enumerateC :: Monad m => C.Conduit a m (Int, a) +enumerateC = enumerateC' 0 + where + enumerateC' !i = awaitJust $ \v -> do + C.yield (i, v) + enumerateC' (i + 1) +{-# INLINE enumerateC #-} + +-- | 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] ]@ +-- +-- This function is deprecated; use 'Data.Conduit.List.chunksOf' +groupC :: (Monad m) => Int -> C.Conduit a m [a] +groupC n = loop n [] + where + loop 0 ps = C.yield (reverse ps) >> loop n [] + loop c ps = C.await >>= \case + Nothing -> unless (null ps) $ C.yield (reverse ps) + Just p -> loop (c-1) (p:ps) + diff --git a/conduit-algorithms.cabal b/conduit-algorithms.cabal --- a/conduit-algorithms.cabal +++ b/conduit-algorithms.cabal @@ -1,9 +1,11 @@ --- This file has been generated from package.yaml by hpack version 0.18.1. +-- This file has been generated from package.yaml by hpack version 0.20.0. -- -- see: https://github.com/sol/hpack +-- +-- hash: af773895c6365b46822705a9fd57b521ba0e8692c5053f7a6eba56b6cc57b07b name: conduit-algorithms -version: 0.0.7.0 +version: 0.0.7.1 synopsis: Conduit-based algorithms description: Algorithms on Conduits, including higher level asynchronous processing and some other utilities. category: Conduit @@ -28,12 +30,12 @@ default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections ghc-options: -Wall build-depends: - base > 4.8 && < 5 - , async + async + , base >4.8 && <5 , bytestring , bzlib-conduit - , conduit >= 1.0 - , conduit-combinators >= 1.1.2 + , conduit >=1.0 + , conduit-combinators >=1.1.2 , conduit-extra , containers , deepseq @@ -41,38 +43,47 @@ , mtl , resourcet , stm - , stm-conduit >= 2.7 - , vector + , stm-conduit >=2.7 , transformers - other-modules: - Paths_conduit_algorithms + , vector + exposed-modules: + Data.Conduit.Algorithms + Data.Conduit.Algorithms.Utils + Data.Conduit.Algorithms.Async + Data.Conduit.Algorithms.Async.ByteString + Data.Conduit.Algorithms.Storable default-language: Haskell2010 test-suite conduit-algorithms-test type: exitcode-stdio-1.0 - main-is: Data/Conduit/Algorithms/Tests.hs + main-is: Tests.hs + hs-source-dirs: + ./tests default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections ghc-options: -Wall build-depends: - base > 4.8 && < 5 + HUnit , async + , base >4.8 && <5 , bytestring , bzlib-conduit - , conduit >= 1.0 - , conduit-combinators >= 1.1.2 + , conduit >=1.0 + , conduit-algorithms + , conduit-combinators >=1.1.2 , conduit-extra , containers , deepseq + , directory , lzma-conduit , mtl , resourcet , stm - , stm-conduit >= 2.7 - , vector - , transformers - , directory - , HUnit + , stm-conduit >=2.7 , test-framework , test-framework-hunit , test-framework-th + , transformers + , vector + other-modules: + Paths_conduit_algorithms default-language: Haskell2010 diff --git a/tests/Tests.hs b/tests/Tests.hs new file mode 100644 --- /dev/null +++ b/tests/Tests.hs @@ -0,0 +1,149 @@ +{- Copyright 2017-2018 Luis Pedro Coelho + - License: MIT + -} +{-# LANGUAGE TemplateHaskell, QuasiQuotes, FlexibleContexts, OverloadedStrings #-} +module Main where + +import Test.Framework.TH +import Test.HUnit +import Test.Framework.Providers.HUnit + +import qualified Data.ByteString as B +import qualified Data.ByteString.Char8 as B8 +import qualified Data.Conduit as C +import qualified Data.Conduit.Combinators as CC +import qualified Data.Conduit.Binary as CB +import qualified Data.Conduit.List as CL +import qualified Data.Vector.Storable as VS +import Data.Conduit ((.|)) +import Data.List (sort) +import System.Directory (removeFile) +import Control.Monad (forM_) + +import qualified Data.Conduit.Algorithms as CAlg +import qualified Data.Conduit.Algorithms.Storable as CAlg +import qualified Data.Conduit.Algorithms.Utils as CAlg +import qualified Data.Conduit.Algorithms.Async as CAlg +import qualified Data.Conduit.Algorithms.Async.ByteString as CAlg + +main :: IO () +main = $(defaultMainGenerator) + +testingFileNameGZ :: FilePath +testingFileNameGZ = "file_just_for_testing_delete_me_please.gz" +testingFileNameGZ2 :: FilePath +testingFileNameGZ2 = "file_just_for_testing_delete_me_please_2.gz" + +extract c = C.runConduitPure (c .| CC.sinkList) + +extractIO c = C.runConduitRes (c .| CC.sinkList) + +shouldProduce values cond = extract cond @?= values +shouldProduceIO values cond = do + p <- extractIO cond + p @?= values + +case_uniqueC = extract (CC.yieldMany [1,2,3,1,1,2,3] .| CAlg.uniqueC) @=? [1,2,3 :: Int] +case_mergeC = shouldProduce expected $ + CAlg.mergeC + [ CC.yieldMany i1 + , CC.yieldMany i2 + , CC.yieldMany i3 + , CC.yieldMany i3 + ] + where + expected = sort (concat [i1, i2, i3, i3]) + i1 = [ 1, 2, 4 :: Int] + i2 = [ 1, 4, 4, 5] + i3 = [-1, 0, 7] + +case_mergeCmonad = shouldProduce expected $ + CAlg.mergeC + [ mYield i1 + , mYield i2 + , mYield i3 + ] + where + expected = sort (concat [i1, i2, i3]) + mYield lst = do + let lst' = map return lst + forM_ lst' $ \elem -> do + elem' <- elem + C.yield elem' + 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_mergeC2monad = shouldProduce [0, 1, 2, 2, 3, 4 :: Int] $ do + CAlg.mergeC2 + (CC.yieldMany [0, 2]) + (CC.yieldMany [1, 2]) + CC.yieldMany [3] + CC.yieldMany [4] + +case_groupC = shouldProduce [[0,1,2], [3,4,5], [6,7,8], [9, 10 :: Int]] $ + CC.yieldMany [0..10] .| CAlg.groupC 3 + +case_enumerateC = shouldProduce [(0,'z'), (1,'o'), (2,'t')] $ + CC.yieldMany ("zot" :: [Char]) .| CAlg.enumerateC + +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]) + +case_unorderedAsyncMapC :: IO () +case_unorderedAsyncMapC = do + vals <- extractIO (CC.yieldMany [0..10] .| CAlg.unorderedAsyncMapC 3 (+ (1:: Int))) + (sort 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" + removeFile testingFileNameGZ + + +case_async_gzip_to_from = do + let testdata = [0 :: Int .. 12] + C.runConduitRes $ + CC.yieldMany testdata + .| CL.map (B8.pack . (\n -> show n ++ "\n")) + .| CAlg.asyncGzipToFile testingFileNameGZ + C.runConduitRes $ + CAlg.asyncGzipFromFile testingFileNameGZ + .| CAlg.asyncGzipToFile testingFileNameGZ2 + shouldProduceIO testdata $ + CAlg.asyncGzipFromFile testingFileNameGZ2 + .| CB.lines + .| CL.map (read . B8.unpack) + removeFile testingFileNameGZ + removeFile testingFileNameGZ2 + +case_asyncFilterLines = do + vals <- extractIO (CC.yieldMany ["This is\nMy data\nBut"," sometimes","\nit is split,\n","in weird ways."] .| CAlg.asyncFilterLinesC 2 (B8.notElem ',')) + (vals @?= ["This is", "My data", "But sometimes", "in weird ways."]) + +case_asyncFilterLinesAllTrue = do + vals <- extractIO (CC.yieldMany ["This is\nMy data\nBut"," sometimes","\nit is split,\n","in weird ways."] .| CAlg.asyncFilterLinesC 2 (const True)) + (vals @?= ["This is", "My data", "But sometimes", "it is split,", "in weird ways."]) + +case_storableVector = do + let v = VS.fromList [0:: Int, 1, 2, 4, 6, 12] + vals <- extractIO (CC.yieldMany [v,v,v] .| CAlg.writeStorableV .| CAlg.readStorableV 3) + (VS.concat vals @=? VS.concat [v,v,v])