diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,7 @@ +Version 0.0.9.0 2019-01-09 by luispedro + * Add Zstandard compression + * Refactor internals to avoid code repetition + Version 0.0.8.2 2018-09-27 by luispedro * Add withPossiblyCompressedFile function for prompt resource deallocation 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 @@ -1,6 +1,6 @@ {-| Module : Data.Conduit.Algorithms.Async -Copyright : 2013-2018 Luis Pedro Coelho +Copyright : 2013-2019 Luis Pedro Coelho License : MIT Maintainer : luis@luispedro.org @@ -26,6 +26,10 @@ , asyncXzToFile , asyncXzFrom , asyncXzFromFile + , asyncZstdTo + , asyncZstdToFile + , asyncZstdFrom + , asyncZstdFromFile , unorderedAsyncMapC ) where @@ -42,7 +46,9 @@ import qualified Data.Conduit.Zlib as CZ import qualified Data.Conduit.Lzma as CX import qualified Data.Streaming.Zlib as SZ -import qualified Data.Conduit.BZlib as CZ +import qualified Data.Conduit.BZlib as CBZ +import qualified Data.Conduit.Zstd as CZstd +import qualified Control.Monad.Trans.Resource as R import qualified Data.Conduit as C import Data.Conduit ((.|)) @@ -185,30 +191,59 @@ 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, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m () -asyncGzipTo h = do - let drain q = liftIO . C.runConduit $ + +genericAsyncFrom :: forall m. (MonadIO m, MonadUnliftIO m) => C.ConduitT B.ByteString B.ByteString m () -> Handle -> C.ConduitT () B.ByteString m () +genericAsyncFrom transform h = do + let prod q = do + C.runConduit $ + C.sourceHandle h + .| transform + .| CL.map Just + .| CA.sinkTBQueue q + liftIO $ atomically (TQ.writeTBQueue q Nothing) + CA.gatherFrom 8 prod .| untilNothing + +genericAsyncTo :: forall m. (MonadIO m, MonadUnliftIO m) => C.ConduitT B.ByteString B.ByteString (R.ResourceT IO) () -> Handle -> C.ConduitT B.ByteString C.Void m () +genericAsyncTo tranform h = do + let drain q = liftIO . C.runConduitRes $ CA.sourceTBQueue q .| untilNothing .| CL.map (B.concat . reverse) - .| CZ.gzip + .| tranform .| C.sinkHandle h bsConcatTo ((2 :: Int) ^ (15 :: Int)) .| CA.drainTo 8 drain + +genericFromFile :: forall m. (MonadResource m, MonadUnliftIO m) => (Handle -> C.ConduitT () B.ByteString m ()) -> FilePath -> C.ConduitT () B.ByteString m () +genericFromFile from fname = C.bracketP + (openFile fname ReadMode) + hClose + from + +genericToFile :: forall m. (MonadResource m, MonadUnliftIO m) => (Handle -> C.ConduitT B.ByteString C.Void m ()) -> FilePath -> C.ConduitT B.ByteString C.Void m () +genericToFile to fname = C.bracketP + (openFile fname WriteMode) + hClose + to + +-- | 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, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m () +asyncGzipTo h = genericAsyncTo gz h + where + gz = CZ.gzip `C.catchC` handleZLibException + handleZLibException = \(e :: SZ.ZlibException) -> + liftIO . ioError $ mkIOError userErrorType ("Error compressing gzip stream: "++displayException e) (Just h) Nothing + -- | 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, MonadUnliftIO m) => FilePath -> C.ConduitT B.ByteString C.Void m () -asyncGzipToFile fname = C.bracketP - (openFile fname WriteMode) - hClose - asyncGzipTo +asyncGzipToFile = genericToFile 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 @@ -233,35 +268,21 @@ -- -- See also 'asyncGzipFrom' asyncGzipFromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT () B.ByteString m () -asyncGzipFromFile fname = C.bracketP - (openFile fname ReadMode) - hClose - asyncGzipFrom +asyncGzipFromFile = genericFromFile asyncGzipFrom -- | A simple sink which performs bzip2 compression in a separate thread and -- writes the results to `h`. -- -- See also 'asyncBzip2ToFile' -asyncBzip2To :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m () -asyncBzip2To h = do - let drain q = C.runConduit $ - CA.sourceTBQueue q - .| untilNothing - .| CL.map (B.concat . reverse) - .| CZ.bzip2 - .| C.sinkHandle h - bsConcatTo ((2 :: Int) ^ (15 :: Int)) - .| CA.drainTo 8 drain +asyncBzip2To :: forall m. (MonadIO m, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m () +asyncBzip2To = genericAsyncTo CBZ.bzip2 -- | Compresses the output and writes to the given file with compression being -- performed in a separate thread. -- -- See also 'asyncBzip2To' asyncBzip2ToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT B.ByteString C.Void m () -asyncBzip2ToFile fname = C.bracketP - (openFile fname WriteMode) - hClose - asyncBzip2To +asyncBzip2ToFile = genericToFile asyncBzip2To -- | A source which produces the bzipped2 content from the the given handle. -- Note that this "reads ahead" so if you do not use all the input, the Handle @@ -269,50 +290,28 @@ -- -- See also 'asyncBzip2FromFile' asyncBzip2From :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.ConduitT () B.ByteString m () -asyncBzip2From h = do - let prod q = do - C.runConduit $ - C.sourceHandle h - .| CZ.multiple CZ.bunzip2 - .| CL.map Just - .| CA.sinkTBQueue q - liftIO $ atomically (TQ.writeTBQueue q Nothing) - CA.gatherFrom 8 prod .| untilNothing +asyncBzip2From = genericAsyncFrom (CZ.multiple CBZ.bunzip2) -- | Open and read a bzip2 file with the uncompression being performed in a -- separate thread. -- -- See also 'asyncBzip2From' asyncBzip2FromFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT () B.ByteString m () -asyncBzip2FromFile fname = C.bracketP - (openFile fname ReadMode) - hClose - asyncBzip2From +asyncBzip2FromFile = genericFromFile asyncBzip2From -- | A simple sink which performs lzma/xz compression in a separate thread and -- writes the results to `h`. -- -- See also 'asyncXzToFile' asyncXzTo :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m) => Handle -> C.ConduitT B.ByteString C.Void m () -asyncXzTo h = do - let drain q = C.runConduit $ - CA.sourceTBQueue q - .| untilNothing - .| CL.map (B.concat . reverse) - .| CX.compress Nothing - .| C.sinkHandle h - bsConcatTo ((2 :: Int) ^ (15 :: Int)) - .| CA.drainTo 8 drain +asyncXzTo = genericAsyncTo (CX.compress Nothing) -- | Compresses the output and writes to the given file with compression being -- performed in a separate thread. -- -- See also 'asyncXzTo' asyncXzToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT B.ByteString C.Void m () -asyncXzToFile fname = C.bracketP - (openFile fname WriteMode) - hClose - asyncXzTo +asyncXzToFile = genericToFile asyncXzTo -- | A source which produces the unxzipped content from the the given handle. -- Note that this "reads ahead" so if you do not use all the input, the Handle @@ -320,29 +319,46 @@ -- -- See also 'asyncXzFromFile' asyncXzFrom :: forall m. (MonadIO m, MonadResource m, MonadUnliftIO m, MonadThrow m) => Handle -> C.ConduitT () B.ByteString m () -asyncXzFrom h = do +asyncXzFrom = let oneGBmembuffer = Just $ 1024 ^ (3 :: Integer) - prod q = do - C.runConduit $ - C.sourceHandle h - .| CZ.multiple (CX.decompress oneGBmembuffer) - .| CL.map Just - .| CA.sinkTBQueue q - liftIO $ atomically (TQ.writeTBQueue q Nothing) - CA.gatherFrom 8 prod .| untilNothing - + in genericAsyncFrom (CX.decompress oneGBmembuffer) -- | Open and read a lzma/xz file with the uncompression being performed in a -- separate thread. -- -- See also 'asyncXzFrom' asyncXzFromFile :: forall m. (MonadResource m, MonadUnliftIO m, MonadThrow m) => FilePath -> C.ConduitT () B.ByteString m () -asyncXzFromFile fname = C.bracketP - (openFile fname ReadMode) - hClose - asyncXzFrom +asyncXzFromFile = genericFromFile asyncXzFrom +-- | Decompress ZStd format using a separate thread +-- +-- See also 'asyncZstdFromFile' +asyncZstdFrom :: forall m. (MonadIO m, MonadUnliftIO m) => Handle -> C.ConduitT () B.ByteString m () +asyncZstdFrom = genericAsyncFrom CZstd.decompress +-- | Compress in ZStd format using a separate thread and write to a file +-- See also 'asyncZstdFrom' +asyncZstdFromFile :: forall m. (MonadResource m, MonadUnliftIO m, MonadThrow m) => FilePath -> C.ConduitT () B.ByteString m () +asyncZstdFromFile = genericFromFile asyncZstdFrom + + +-- | Compress in Zstd format using a separate thread +-- +-- See also 'asyncZstdToFile' +asyncZstdTo :: forall m. (MonadIO m, MonadUnliftIO m) => + Int -- ^ compression level + -> Handle -> C.ConduitT B.ByteString C.Void m () +asyncZstdTo clevel = genericAsyncTo (CZstd.compress clevel) + + +-- | Compress in ZStd format using a separate thread and write to a file +-- +-- This will use compression level 3 as this is the default in the ZStd C API +-- +-- See also 'asyncZstdTo' +asyncZstdToFile :: forall m. (MonadResource m, MonadUnliftIO m) => FilePath -> C.ConduitT B.ByteString C.Void m () +asyncZstdToFile = genericToFile (asyncZstdTo 3) + -- | If the filename indicates a supported compressed file (gzip, xz, and, on -- Unix, bzip2), then it reads it and uncompresses it. -- @@ -368,6 +384,7 @@ | ".gz" `isSuffixOf` fname = asyncGzipFrom | ".xz" `isSuffixOf` fname = asyncXzFrom | ".bz2" `isSuffixOf` fname = asyncBzip2From + | ".zstd" `isSuffixOf` fname = asyncZstdFrom | otherwise = C.sourceHandle @@ -384,6 +401,7 @@ | ".gz" `isSuffixOf` fname = asyncGzipFromFile fname | ".xz" `isSuffixOf` fname = asyncXzFromFile fname | ".bz2" `isSuffixOf` fname = asyncBzip2FromFile fname + | ".zstd" `isSuffixOf` fname = asyncZstdFromFile fname | otherwise = C.sourceFile fname -- | If the filename indicates a gzipped file (or, on Unix, also a bz2 file), @@ -395,4 +413,5 @@ | ".gz" `isSuffixOf` fname = asyncGzipToFile fname | ".xz" `isSuffixOf` fname = asyncXzToFile fname | ".bz2" `isSuffixOf` fname = asyncBzip2ToFile fname + | ".zstd" `isSuffixOf` fname = asyncZstdToFile fname | otherwise = C.sinkFile fname diff --git a/conduit-algorithms.cabal b/conduit-algorithms.cabal --- a/conduit-algorithms.cabal +++ b/conduit-algorithms.cabal @@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 0fe3779c7b8153d9edb2c1061a1ea1ce629a6e6474a0c02fd622a91b5c237a31 +-- hash: 568a611355a3ef3c59150cc32f207c38a0d6d802487517f30c8e437c8c839ab8 name: conduit-algorithms -version: 0.0.8.2 +version: 0.0.9.0 synopsis: Conduit-based algorithms description: Algorithms on Conduits, including higher level asynchronous processing and some other utilities. category: Conduit @@ -33,7 +33,7 @@ Data.Conduit.Algorithms.Async.ByteString Data.Conduit.Algorithms.Storable default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections - ghc-options: -Wall -O2 + ghc-options: -Wall build-depends: async , base >4.8 && <5 @@ -42,6 +42,7 @@ , conduit >=1.3 , conduit-combinators >=1.1.2 , conduit-extra + , conduit-zstd , containers , deepseq , exceptions @@ -68,7 +69,7 @@ hs-source-dirs: ./tests default-extensions: BangPatterns OverloadedStrings LambdaCase TupleSections - ghc-options: -Wall -O2 + ghc-options: -Wall build-depends: HUnit , async @@ -79,6 +80,7 @@ , conduit-algorithms , conduit-combinators >=1.1.2 , conduit-extra + , conduit-zstd , containers , deepseq , directory @@ -107,7 +109,7 @@ 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" + ghc-options: -Wall -Wall -fwarn-tabs -fno-warn-missing-signatures -threaded -rtsopts "-with-rtsopts=-A64m -n4m -H" build-depends: async , base >4.8 && <5 @@ -117,6 +119,7 @@ , conduit-algorithms , conduit-combinators >=1.1.2 , conduit-extra + , conduit-zstd , containers , criterion , deepseq diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -1,4 +1,4 @@ -{- Copyright 2017-2018 Luis Pedro Coelho +{- Copyright 2017-2019 Luis Pedro Coelho - License: MIT -} {-# LANGUAGE TemplateHaskell, CPP, QuasiQuotes, FlexibleContexts, OverloadedStrings #-} @@ -46,6 +46,8 @@ testingFileNameXZ = "file_just_for_testing_delete_me_please.xz" testingFileNameXZ2 :: FilePath testingFileNameXZ2 = "file_just_for_testing_delete_me_please_2.xz" +testingFileNameZstd :: FilePath +testingFileNameZstd = "file_just_for_testing_delete_me_please.zstd" extract :: C.ConduitM () b FID.Identity () -> [b] extract c = C.runConduitPure (c .| CC.sinkList) @@ -173,6 +175,22 @@ r @?= "Hello World" removeFile testingFileNameGZ +case_asyncGzipLarge :: IO () +case_asyncGzipLarge = do + let repeats = 16384 -- large enough to cause the input to be split into blocks inside asyncGzipToFile + C.runConduitRes (CC.yieldMany (concat $ replicate repeats ["Hello" :: B.ByteString, " ", "World\n"]) .| CAlg.asyncGzipToFile testingFileNameGZ) + let checkHello !n = C.await >>= \case + Nothing -> return n + Just "Hello World\n" -> checkHello (n + 1) + _ -> return (-1) + r <- C.runConduitRes $ + CAlg.asyncGzipFromFile testingFileNameGZ + .| CC.chunksOfE 12 + .| checkHello (0 :: Int) + + removeFile testingFileNameGZ + r @?= repeats + case_asyncBzip2 :: IO () case_asyncBzip2 = do C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncBzip2ToFile testingFileNameBZ2) @@ -250,6 +268,13 @@ .| CL.map (read . B8.unpack) removeFile testingFileNameXZ removeFile testingFileNameXZ2 + +case_async_zstd_to_from :: IO () +case_async_zstd_to_from = do + C.runConduitRes (CC.yieldMany ["Hello", " ", "World"] .| CAlg.asyncZstdToFile testingFileNameZstd) + r <- B.concat <$> extractIO (CAlg.asyncZstdFromFile testingFileNameZstd) + r @?= "Hello World" + removeFile testingFileNameZstd case_asyncFilterLines :: IO () case_asyncFilterLines = do