diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,6 @@ +Version 0.0.8.1 2018-05-29 by luispedro + * faster mergeC (use a priority queue instead of an ordered list) + Version 0.0.8.0 2018-03-12 by luispedro * Update for conduit 1.3 (issue #6) diff --git a/Data/Conduit/Algorithms.hs b/Data/Conduit/Algorithms.hs --- a/Data/Conduit/Algorithms.hs +++ b/Data/Conduit/Algorithms.hs @@ -6,8 +6,6 @@ Simple algorithms packaged as Conduits -} - - {-# LANGUAGE Rank2Types #-} module Data.Conduit.Algorithms ( uniqueOnC @@ -20,8 +18,9 @@ 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 qualified Data.PQueue.Prio.Min as PQ import Control.Monad.Trans.Class (lift) +import Control.Monad (foldM) import Data.Conduit.Algorithms.Utils (awaitJust) @@ -81,27 +80,20 @@ mergeC [a] = a mergeC [a,b] = mergeC2 a b mergeC cs = CI.ConduitT $ \rest -> let - go [] = rest () - go (CI.HaveOutput c_next v:larger) = - CI.HaveOutput (norm1 c_next >>= go . insert1 larger) 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)" + go q = case PQ.minView q of + Nothing -> rest() + Just (CI.HaveOutput c_next v, q') -> CI.HaveOutput (norm1insert q' c_next >>= go) v + _ -> error "This situation should have been impossible (mergeC/go)" + norm1insert :: (Monad m, Ord o) => PQ.MinPQueue o (CI.Pipe () i o () m ()) -> CI.Pipe () i o () m () -> CI.Pipe () i o () m (PQ.MinPQueue o (CI.Pipe () i o () m ())) + norm1insert q c@(CI.HaveOutput _ v) = return (PQ.insert v c q) + norm1insert q c@CI.Done{} = return q + norm1insert q (CI.PipeM p) = lift p >>= norm1insert q + norm1insert q (CI.NeedInput _ next) = norm1insert q (next ()) + norm1insert q (CI.Leftover next ()) = norm1insert q next in do let st = map (($ CI.Done) . CI.unConduitT) cs - st' <- mapM norm1 st - go . sortBy compareHO . filter isHO $ st' - + init <- foldM norm1insert PQ.empty st + go init -- | Take two sorted sources and merge them. -- 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 @@ -39,9 +39,7 @@ import qualified Data.Conduit.TQueue as CA import qualified Data.Conduit.List as CL import qualified Data.Conduit.Zlib as CZ -#ifndef WINDOWS import qualified Data.Conduit.Lzma as CX -#endif import qualified Data.Streaming.Zlib as SZ import qualified Data.Conduit.BZlib as CZ import qualified Data.Conduit as C @@ -120,9 +118,12 @@ -- | 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) >>= C.yield) >> yAll rest + yAll q + | Seq.null q = return () + | otherwise = do + (r, q') <- liftIO $ retrieveResult q + C.yield r + yAll q' loop :: Seq.Seq (A.Async b) -> C.Conduit a m b loop q = C.await >>= \case @@ -238,7 +239,7 @@ -- | A simple sink which performs bzip2 compression in a separate thread and -- writes the results to `h`. -- --- See also 'asyncGzipToFile' +-- See also 'asyncBzip2ToFile' asyncBzip2To :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.Sink B.ByteString m () asyncBzip2To h = do let drain q = C.runConduit $ @@ -253,7 +254,7 @@ -- | Compresses the output and writes to the given file with compression being -- performed in a separate thread. -- --- See also 'asyncGzipTo' +-- See also 'asyncBzip2To' asyncBzip2ToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.Sink B.ByteString m () asyncBzip2ToFile fname = C.bracketP (openFile fname WriteMode) @@ -264,7 +265,7 @@ -- 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' +-- See also 'asyncBzip2FromFile' asyncBzip2From :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.Source m B.ByteString asyncBzip2From h = do let prod q = do @@ -279,7 +280,7 @@ -- | Open and read a bzip2 file with the uncompression being performed in a -- separate thread. -- --- See also 'asyncGzipFrom' +-- See also 'asyncBzip2From' asyncBzip2FromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.Source m B.ByteString asyncBzip2FromFile fname = C.bracketP (openFile fname ReadMode) @@ -289,18 +290,14 @@ -- | A simple sink which performs lzma/xz compression in a separate thread and -- writes the results to `h`. -- --- See also 'asyncGzipToFile' +-- See also 'asyncXzToFile' asyncXzTo :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.Sink B.ByteString m () asyncXzTo h = do let drain q = C.runConduit $ CA.sourceTBQueue q .| untilNothing .| CL.map (B.concat . reverse) -#ifndef WINDOWS .| CX.compress Nothing -#else - .| error "lzma/xz compression is not available on Windows" -#endif .| C.sinkHandle h bsConcatTo ((2 :: Int) ^ (15 :: Int)) .| CA.drainTo 8 drain @@ -308,7 +305,7 @@ -- | Compresses the output and writes to the given file with compression being -- performed in a separate thread. -- --- See also 'asyncGzipTo' +-- See also 'asyncXzTo' asyncXzToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.Sink B.ByteString m () asyncXzToFile fname = C.bracketP (openFile fname WriteMode) @@ -319,18 +316,14 @@ -- 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' +-- See also 'asyncXzFromFile' asyncXzFrom :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m, MonadThrow m) => Handle -> C.Source m B.ByteString asyncXzFrom h = do let oneGBmembuffer = Just $ 1024 ^ (3 :: Integer) prod q = do C.runConduit $ C.sourceHandle h -#ifndef WINDOWS .| CZ.multiple (CX.decompress oneGBmembuffer) -#else - .| error "lzma/xz decompression is not available on Windows" -#endif .| CL.map Just .| CA.sinkTBQueue q liftIO $ atomically (TQ.writeTBQueue q Nothing) @@ -351,8 +344,6 @@ -- 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 :: (MonadUnliftIO m, MonadResource m, MonadThrow m) => FilePath -> C.Source m B.ByteString conduitPossiblyCompressedFile fname | ".gz" `isSuffixOf` fname = asyncGzipFromFile fname @@ -360,6 +351,10 @@ | ".bz2" `isSuffixOf` fname = asyncBzip2FromFile fname | otherwise = C.sourceFile fname +-- | If the filename indicates a gzipped file (or, on Unix, also a bz2 file), +-- then it compresses and write with the algorithm matching the filename +-- +-- On Windows, attempting to write to a bzip2 file, results in 'error'. conduitPossiblyCompressedToFile :: (MonadUnliftIO m, MonadResource m) => FilePath -> C.Sink B.ByteString m () conduitPossiblyCompressedToFile fname | ".gz" `isSuffixOf` fname = asyncGzipToFile fname diff --git a/bench/Bench.hs b/bench/Bench.hs new file mode 100644 --- /dev/null +++ b/bench/Bench.hs @@ -0,0 +1,61 @@ +{- Copyright 2018 + - Licence: MIT -} +import Criterion.Main (defaultMain, nfIO, bench, bgroup, Benchmarkable) +import Control.Monad.Trans.Resource + +import qualified Data.ByteString as B +import qualified Data.ByteString.Char8 as B8 +import Control.DeepSeq (NFData) +import qualified Data.Conduit.List as CL +import qualified Data.Conduit.Binary as CB +import qualified Data.Conduit as C +import Data.Conduit ((.|)) + + +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 + +progress :: Monad m => Int -> Int -> C.ConduitT () Int m () +progress _ 0 = return () +progress n i = C.yield n >> progress (n + 3) (i - 1) + +nfConduit :: NFData a => C.ConduitT () C.Void (ResourceT IO) a -> Benchmarkable +nfConduit = nfIO . C.runConduitRes + +countC = countC' (0 :: Int) + where + countC' !n = C.await >>= \case + Nothing -> return n + Just _ -> countC' (n+1) +sumC = sumC' (0 :: Int) + where + sumC' !n = C.await >>= \case + Nothing -> return n + Just v -> sumC' (n+v) + +main = defaultMain + [ bgroup "merge" + [ bench "mergeC2" $ nfConduit (CAlg.mergeC2 (progress 1 1000) (progress 30 2000) .| CL.fold (+) (0 :: Int)) + , bench "mergeC_3" $ nfConduit (CAlg.mergeC [progress 1 1000, progress 30 2000, progress 7 9] .| CL.fold (+) (0 :: Int)) + , bench "mergeC_1000" $ nfConduit (CAlg.mergeC [progress 1 n | n <- [1 .. 2000]] .| CL.fold (+) (0 :: Int)) + ] + , bgroup "async-compress" + [ bench "asyncGzipToFile" $ nfConduit (CB.sourceFile "test_data/input.txt" .| CAlg.asyncGzipToFile "test_data/output.txt.gz") + , bench "asyncBzip2ToFile" $ nfConduit (CB.sourceFile "test_data/input.txt" .| CAlg.asyncBzip2ToFile "test_data/output.txt.gz") + , bench "asyncXzToFile" $ nfConduit (CB.sourceFile "test_data/input.txt" .| CAlg.asyncXzToFile "test_data/output.txt.gz") + ] + , bgroup "async-uncompress" + [ bench "baseline" $ nfConduit (CB.sourceFile "test_data/input.txt" .| CB.sinkFile "test_data/output.txt") + , bench "asyncGzipFromFile" $ nfConduit (CAlg.conduitPossiblyCompressedFile "test_data/input.txt.gz" .| CB.sinkFile "test_data/output.txt") + , bench "asyncBzip2FromFile" $ nfConduit (CAlg.conduitPossiblyCompressedFile "test_data/input.txt.bz2" .| CB.sinkFile "test_data/output.txt") + ] + , bgroup "async-map" + [ bench "filterlines" $ nfConduit (CB.sourceFile "test_data/input.txt" .| CB.lines .| CL.filter (\line -> (read . B8.unpack $ line) > (1000 :: Int)) .| countC) + , bench "asyncFilterLinesC" $ nfConduit (CB.sourceFile "test_data/input.txt" .| CAlg.asyncFilterLinesC 8 (\line -> (read . B8.unpack $ line) > (1000 :: Int)) .| countC) + , bench "asyncMapLineGroupsC" $ nfConduit (CB.sourceFile "test_data/input.txt" .| CAlg.asyncMapLineGroupsC 8 (length . filter (\line -> (read . B8.unpack $ line) > (1000 :: Int))) .| sumC) + ] + ] + diff --git a/conduit-algorithms.cabal b/conduit-algorithms.cabal --- a/conduit-algorithms.cabal +++ b/conduit-algorithms.cabal @@ -1,11 +1,11 @@ --- This file has been generated from package.yaml by hpack version 0.20.0. +-- This file has been generated from package.yaml by hpack version 0.21.2. -- -- see: https://github.com/sol/hpack -- --- hash: e408a4353cbeef37b0ef19abf3d1c40d0939cc1dc51ce2f917d36e620eb29a89 +-- hash: c74053b52f17daacb3df2612ca8127e9e24189040509592d94be61a097cfaf0e name: conduit-algorithms -version: 0.0.8.0 +version: 0.0.8.1 synopsis: Conduit-based algorithms description: Algorithms on Conduits, including higher level asynchronous processing and some other utilities. category: Conduit @@ -27,8 +27,14 @@ location: https://github.com/luispedro/conduit-algorithms library + exposed-modules: + Data.Conduit.Algorithms + Data.Conduit.Algorithms.Utils + Data.Conduit.Algorithms.Async + Data.Conduit.Algorithms.Async.ByteString + Data.Conduit.Algorithms.Storable default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections - ghc-options: -Wall + ghc-options: -Wall -O2 build-depends: async , base >4.8 && <5 @@ -40,8 +46,10 @@ , containers , deepseq , exceptions + , lzma-conduit , monad-control , mtl + , pqueue , resourcet , stm , stm-conduit >=4.0 @@ -51,24 +59,17 @@ , vector if os(windows) cpp-options: -DWINDOWS - else - build-depends: - lzma-conduit - 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: Tests.hs + other-modules: + Paths_conduit_algorithms hs-source-dirs: ./tests default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections - ghc-options: -Wall + ghc-options: -Wall -O2 build-depends: HUnit , async @@ -83,8 +84,10 @@ , deepseq , directory , exceptions + , lzma-conduit , monad-control , mtl + , pqueue , resourcet , stm , stm-conduit >=4.0 @@ -97,9 +100,39 @@ , vector if os(windows) cpp-options: -DWINDOWS - else - build-depends: - lzma-conduit - other-modules: - Paths_conduit_algorithms + default-language: Haskell2010 + +benchmark conduit-algorithms-bench + type: exitcode-stdio-1.0 + main-is: Bench.hs + hs-source-dirs: + ./bench + default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections + ghc-options: -Wall -O2 -Wall -fwarn-tabs -fno-warn-missing-signatures -threaded -rtsopts "-with-rtsopts=-A64m -n4m -H" + build-depends: + async + , base >4.8 && <5 + , bytestring + , bzlib-conduit + , conduit >=1.3 + , conduit-algorithms + , conduit-combinators >=1.1.2 + , conduit-extra + , containers + , criterion + , deepseq + , exceptions + , lzma-conduit + , monad-control + , mtl + , pqueue + , resourcet + , stm + , stm-conduit >=4.0 + , streaming-commons + , transformers + , unliftio-core + , vector + if os(windows) + cpp-options: -DWINDOWS default-language: Haskell2010 diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -87,6 +87,20 @@ i2 = [ 1, 4, 4, 5] i3 = [-1, 0, 7] +case_mergeCrepeated :: Assertion +case_mergeCrepeated = 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, 3, 4 :: Int] + i2 = i1 + i3 = i1 + case_mergeCmonad :: Assertion case_mergeCmonad = shouldProduce expected $ CAlg.mergeC @@ -166,11 +180,7 @@ removeFile testingFileNameBZ2 case_asyncXz :: IO () -#ifdef WINDOWS -case_asyncXz = assertError $ do -#else case_asyncXz = do -#endif C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncXzToFile testingFileNameXZ) r <- B.concat <$> extractIO (CAlg.asyncXzFromFile testingFileNameXZ) r @?= "Hello World" @@ -211,11 +221,7 @@ removeFile testingFileNameBZ22 case_async_xz_to_from :: IO () -#ifdef WINDOWS -case_async_xz_to_from = assertError $ do -#else case_async_xz_to_from = do -#endif let testdata = [0 :: Int .. 12] C.runConduitRes $ CC.yieldMany testdata