conduit-algorithms 0.0.8.0 → 0.0.8.1
raw patch · 6 files changed
+161/−71 lines, 6 filesdep +criteriondep +pqueuePVP ok
version bump matches the API change (PVP)
Dependencies added: criterion, pqueue
API changes (from Hackage documentation)
Files
- ChangeLog +3/−0
- Data/Conduit/Algorithms.hs +14/−22
- Data/Conduit/Algorithms/Async.hs +17/−22
- bench/Bench.hs +61/−0
- conduit-algorithms.cabal +52/−19
- tests/Tests.hs +14/−8
ChangeLog view
@@ -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)
Data/Conduit/Algorithms.hs view
@@ -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. --
Data/Conduit/Algorithms/Async.hs view
@@ -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
+ bench/Bench.hs view
@@ -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)+ ]+ ]+
conduit-algorithms.cabal view
@@ -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
tests/Tests.hs view
@@ -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