streamly-archive 0.2.0 → 0.3.0
raw patch · 8 files changed
+684/−270 lines, 8 filesdep +asyncdep +containersdep +splitdep ~bytestringdep ~directorydep ~streamly
Dependencies added: async, containers, split
Dependency ranges changed: bytestring, directory, streamly, streamly-core, tasty-hunit
Files
- ChangeLog.md +8/−0
- LICENSE +1/−1
- README.md +29/−44
- src/Streamly/External/Archive.hs +306/−73
- src/Streamly/External/Archive/Internal/Foreign.hs +29/−31
- streamly-archive.cabal +18/−13
- test/ReadmeMain.hs +47/−0
- test/Streamly/External/Archive/Tests.hs +246/−108
ChangeLog.md view
@@ -1,3 +1,11 @@+## 0.3.0++* Updated for Streamly 0.10.0.+* Eliminated the `Void` seed for `readArchive`.+* Added the ability to map over headers and filter out entries during `readArchive`.+* Generalized `groupByHeader` to `groupByLeft`.+* Added utility functions: `eitherByLeft`, `chunkOn`, `chunkOnFold`.+ ## 0.2.0 * Updated for Streamly 0.9.0.
LICENSE view
@@ -1,4 +1,4 @@-Copyright Shlok Datye (c) 2023+Copyright Shlok Datye (c) 2024 All rights reserved.
README.md view
@@ -20,69 +20,54 @@ module Main where -import Crypto.Hash (hashFinalize, hashInit, hashUpdate)-import Crypto.Hash.Algorithms (SHA256)+import Crypto.Hash import Data.ByteString (ByteString)-import Data.Function ((&))-import Data.Maybe (fromJust, fromMaybe)-import Data.Void (Void)+import Data.Function+import Data.Functor+import Data.Maybe+import Streamly.Data.Fold (Fold) import qualified Streamly.Data.Fold as F import qualified Streamly.Data.Stream.Prelude as S import Streamly.External.Archive- ( Header,- groupByHeader,- headerPathName,- readArchive,- )-import Streamly.Internal.Data.Fold.Type (Fold (Fold), Step (Partial))-import Streamly.Internal.Data.Unfold.Type (Unfold) main :: IO () main = do- -- Obtain an unfold for the archive.- -- For each entry in the archive, we will get a Header followed- -- by zero or more ByteStrings containing chunks of file data.- let unf :: Unfold IO Void (Either Header ByteString) =- readArchive "/path/to/archive.tar.gz"-- -- Create a fold for converting each entry (which, as we saw- -- above, is a Left followed by zero or more Rights) into a- -- path and corresponding SHA-256 hash (Nothing for no data).+ -- A fold for converting each archive entry (which is a Header followed by+ -- zero or more ByteStrings) into a path and corresponding SHA-256 hash+ -- (Nothing for no data). let entryFold :: Fold IO (Either Header ByteString) (String, Maybe String) =- Fold+ F.foldlM' ( \(mpath, mctx) e -> case e of Left h -> do mpath' <- headerPathName h- return $ Partial (mpath', mctx)+ return (mpath', mctx) Right bs ->- return $- Partial- ( mpath,- Just . (`hashUpdate` bs) $- fromMaybe (hashInit @SHA256) mctx- )- )- (return $ Partial (Nothing, Nothing))- ( \(mpath, mctx) ->- return- ( show $ fromJust mpath,- show . hashFinalize <$> mctx- )+ return+ ( mpath,+ Just . (`hashUpdate` bs) $+ fromMaybe (hashInit @SHA256) mctx+ ) )+ (return (Nothing, Nothing))+ <&> ( \(mpath, mctx) ->+ ( show $ fromMaybe (error "path expected") mpath,+ show . hashFinalize <$> mctx+ )+ ) - -- Execute the stream, grouping at the headers (the Lefts) using the- -- above fold, and output the paths and SHA-256 hashes along the way.- S.unfold unf undefined- & groupByHeader entryFold+ -- Execute the stream, grouping at the headers (the Lefts) using the above+ -- fold, and output the paths and SHA-256 hashes along the way.+ S.unfold readArchive (id, "/path/to/archive.tar.gz")+ & groupByLeft entryFold & S.mapM print & S.fold F.drain ``` ## Benchmarks -See `./bench/README.md`. We find on our machine<sup>†</sup> that (1) reading an archive using this library is just as fast as using plain Haskell `IO` code; and that (2) both are somewhere between 1.7x (large files) and 2.5x (many 1-byte files) slower than C.--The former fulfills the promise of [streamly](https://hackage.haskell.org/package/streamly) and stream fusion. The differences to C are presumably explained by the marshalling of data into the Haskell world and are currently small enough for our purposes.+See `./bench/README.md`. Summary (with rough figures from our machine<sup>†</sup>):+ * For 1-byte files, this library has roughly a 70 ns/byte overhead compared to plain Haskell `IO` code, which has roughly a 895 ns/byte overhead compared to plain C.+ * For larger (> 10 KB) files, this library performs just as good as plain Haskell `IO` code, which has roughly a 0.15 ns/byte overhead compared to plain C. -<sup>†</sup> April 2023; [Linode](https://linode.com); Debian 11, Dedicated 32GB: 16 CPU, 640GB Storage, 32GB RAM.+<sup>†</sup> July 2024; NixOS 22.11; Intel i7-12700K (3.6 GHz, 12 cores); Corsair VENGEANCE LPX DDR4 RAM 64GB (2 x 32GB) 3200MHz; Samsung 970 EVO Plus SSD 2TB (M.2 NVMe).
src/Streamly/External/Archive.hs view
@@ -1,12 +1,25 @@+{-# LANGUAGE BangPatterns #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} module Streamly.External.Archive ( -- ** Read readArchive,- groupByHeader, + -- *** Read options+ ReadOptions,+ mapHeaderMaybe,++ -- ** Utility functions++ -- | Various utility functions that some might find useful.+ groupByLeft,+ eitherByLeft,+ chunkOn,+ chunkOnFold,+ -- ** Header Header, FileType (..),@@ -17,39 +30,26 @@ ) where -import Control.Exception (mask_)-import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Exception+import Control.Monad.IO.Class import Data.ByteString (ByteString) import qualified Data.ByteString as B import Data.Either+import Data.Foldable import Data.Function-import Data.Int (Int64)-import Data.Void (Void)-import Foreign (Ptr, free, malloc)-import Foreign.C.Types (CChar, CSize)+import qualified Data.Sequence as Seq+import Foreign+import Foreign.C.Types import Streamly.Data.Fold (Fold) import qualified Streamly.Data.Parser as P import Streamly.Data.Stream.Prelude (Stream) import qualified Streamly.Data.Stream.Prelude as S-import Streamly.Data.Unfold (lmap)+import Streamly.Data.Unfold import Streamly.External.Archive.Internal.Foreign- ( Entry,- FileType (..),- archive_entry_filetype,- archive_entry_pathname,- archive_entry_pathname_utf8,- archive_entry_size,- archive_read_data_block,- archive_read_free,- archive_read_new,- archive_read_next_header,- archive_read_open_filename,- archive_read_support_filter_all,- archive_read_support_format_all,- )-import Streamly.Internal.Data.IOFinalizer (newIOFinalizer, runIOFinalizer)-import Streamly.Internal.Data.Stream.StreamD.Type (Step (..))-import Streamly.Internal.Data.Unfold.Type (Unfold (..))+import qualified Streamly.Internal.Data.Fold as F+import Streamly.Internal.Data.IOFinalizer+import qualified Streamly.Internal.Data.Stream as S+import qualified Streamly.Internal.Data.Unfold as U -- | Header information for an entry in the archive. newtype Header = Header Entry@@ -66,20 +66,95 @@ headerPathNameUtf8 :: Header -> IO (Maybe ByteString) headerPathNameUtf8 (Header e) = archive_entry_pathname_utf8 e +-- | Returns the file size of the entry, if it has been set; returns 'Nothing' otherwise. {-# INLINE headerSize #-} headerSize :: Header -> IO (Maybe Int) headerSize (Header e) = archive_entry_size e --- | A convenience function for grouping @Either Header ByteString@s, usually obtained with--- 'readArchive', by the headers. The input @Fold@ processes a single entry (a 'Header' followed by--- zero or more @ByteString@s).-{-# INLINE groupByHeader #-}-groupByHeader ::+-- | Creates an unfold with which we can stream data out of the given archive.+--+-- By default (with 'id' as the read options modifier), we get for each entry in the archive a+-- 'Header' followed by zero or more @ByteString@s containing chunks of file data.+--+-- To modify the read options, one can use function composition.+{-# INLINE readArchive #-}+readArchive ::+ (MonadIO m) =>+ Unfold m (ReadOptions m Header -> ReadOptions m a, FilePath) (Either a ByteString)+readArchive =+ U.Unfold+ ( \(ropts, arch, buf, sz, offs, pos, ref, readHeader) ->+ if readHeader+ then do+ me <- liftIO $ archive_read_next_header arch+ case me of+ Nothing -> do+ liftIO $ runIOFinalizer ref+ return U.Stop+ Just e -> do+ let hdr = Header e+ m <- _mapHeaderMaybe ropts hdr+ return $ case m of+ Nothing -> U.Skip (ropts, arch, buf, sz, offs, 0, ref, True)+ Just a -> U.Yield (Left a) (ropts, arch, buf, sz, offs, 0, ref, False)+ else do+ (bs, done) <- liftIO $ archive_read_data_block arch buf sz offs pos+ return $+ if B.length bs > 0+ then+ U.Yield+ (Right bs)+ (ropts, arch, buf, sz, offs, pos + fromIntegral (B.length bs), ref, done)+ else U.Skip (ropts, arch, buf, sz, offs, pos, ref, done)+ )+ ( \(modifier, fp) -> do+ (arch, buf, sz, offs, ref) <- liftIO . mask_ $ do+ arch <- liftIO archive_read_new+ buf :: Ptr (Ptr CChar) <- liftIO malloc+ sz :: Ptr CSize <- liftIO malloc+ offs :: Ptr Int64 <- liftIO malloc+ ref <- newIOFinalizer $ archive_read_free arch >> free buf >> free sz >> free offs+ return (arch, buf, sz, offs, ref)+ liftIO $ archive_read_support_filter_all arch+ liftIO $ archive_read_support_format_all arch+ liftIO $ archive_read_open_filename arch fp++ -- + We ended up with functions instead of records to avoid an error about an ambiguous+ -- monad type for defaultReadOptions when the user sets the headerFilter record.+ -- + (A dummy Proxy record worked too, but partially exporting records breaks Haddock.)+ let ropts = modifier _defaultReadOptions++ return (ropts, arch, buf, sz, offs, 0, ref, True)+ )++newtype ReadOptions m a = ReadOptions+ { _mapHeaderMaybe :: Header -> m (Maybe a)+ }++_defaultReadOptions :: (Monad m) => ReadOptions m Header+_defaultReadOptions =+ ReadOptions+ { _mapHeaderMaybe = return . Just+ }++-- | If this returns @Just@ for a header, that header (mapped to a different value if desired) and+-- any following @ByteString@ chunks are included in the 'readArchive' unfold. If this returns+-- @Nothing@ for a header, that header and any following @ByteString@ chunks are excluded from the+-- 'readArchive' unfold.+--+-- By default, all entries are included with unaltered headers.+mapHeaderMaybe :: (Header -> m (Maybe a)) -> ReadOptions m Header -> ReadOptions m a+mapHeaderMaybe x o = o {_mapHeaderMaybe = x}++-- | Groups a stream of @Either@s by the @Left@s. The provided @Fold@ processes a single @Left@+-- followed by any subsequent (zero or more) @Right@s.+{-# INLINE groupByLeft #-}+groupByLeft :: (Monad m) =>- Fold m (Either Header ByteString) b ->- Stream m (Either Header ByteString) ->- Stream m b-groupByHeader itemFold str =+ Fold m (Either a b) c ->+ Stream m (Either a b) ->+ Stream m c+groupByLeft itemFold str = str & S.parseMany (P.groupBy (\_ e -> isRight e) itemFold) & fmap@@ -87,45 +162,203 @@ Left _ -> -- groupBy is documented to never fail. error "unexpected parseMany/groupBy error"- Right b -> b+ Right c -> c ) --- | Creates an unfold with which we can stream data out of the given archive.-{-# INLINE readArchive #-}-readArchive :: (MonadIO m) => FilePath -> Unfold m Void (Either Header ByteString)-readArchive fp =- (lmap . const) () $- Unfold- ( \(arch, buf, sz, offs, pos, ref, readHeader) ->- if readHeader- then do- me <- liftIO $ archive_read_next_header arch- case me of- Nothing -> do- liftIO $ runIOFinalizer ref- return Stop- Just e -> do- return $ Yield (Left $ Header e) (arch, buf, sz, offs, 0, ref, False)- else do- (bs, done) <- liftIO $ archive_read_data_block arch buf sz offs pos- return $- if B.length bs > 0- then- Yield- (Right bs)- (arch, buf, sz, offs, pos + fromIntegral (B.length bs), ref, done)- else Skip (arch, buf, sz, offs, pos, ref, done)- )- ( \() -> do- (arch, buf, sz, offs, ref) <- liftIO . mask_ $ do- arch <- liftIO archive_read_new- buf :: Ptr (Ptr CChar) <- liftIO malloc- sz :: Ptr CSize <- liftIO malloc- offs :: Ptr Int64 <- liftIO malloc- ref <- newIOFinalizer $ archive_read_free arch >> free buf >> free sz >> free offs- return (arch, buf, sz, offs, ref)- liftIO $ archive_read_support_filter_all arch- liftIO $ archive_read_support_format_all arch- liftIO $ archive_read_open_filename arch fp- return (arch, buf, sz, offs, 0, ref, True)+-- | Associates each @Right@ in a stream with the latest @Left@ that came before it.+--+-- >>> l = [Right 10, Left "a", Right 1, Right 2, Left "b", Left "c", Right 20]+-- >>> S.fold F.toList . eitherByLeft . S.fromList $ l+-- [("a",1),("a",2),("c",20)]+eitherByLeft :: (Monad m) => Stream m (Either a b) -> Stream m (a, b)+eitherByLeft s =+ S.scanl'+ ( \(curra, _) e ->+ case e of+ Left newa -> (Just newa, Nothing)+ Right newb -> (curra, Just newb)+ )+ (Nothing, Nothing)+ s+ & S.mapMaybe+ ( \(ma, mb) -> case (ma, mb) of+ (Just a, Just b) -> Just (a, b)+ _ -> Nothing )++-- | The state of the chunkOn stream.+data ChunkOnState_ is h+ = -- | The initial state; or a header is done being yielded.+ COInitOrYieldHeader_+ | -- | A bytestring not containing splitWd is being built up.+ COResidue_ !ByteString+ | -- | Chunks are being processed.+ COProcessChunks_ ![ByteString] !ByteString+ | -- | A stop has been asked for.+ COStop_+ | -- | A header yield has been asked for.+ COYieldHeader_ !h !is++-- | Chunks up the bytestrings following each @Left@ by the given word, discarding the given word.+-- (For instance, the word could be @10@ (newline), which gives us lines as the chunks.) The+-- bytestrings in the resulting stream are the desired chunks.+{-# INLINE chunkOn #-}+chunkOn ::+ (Monad m) =>+ Word8 ->+ Stream m (Either a ByteString) ->+ Stream m (Either a ByteString)+chunkOn splitWd (S.Stream istep isinit) =+ -- "i": input.+ S.Stream step' (isinit, COInitOrYieldHeader_)+ where+ -- A utility function to obtain (chunks, next residue) from the previous residue and the latest+ -- incoming bytestring.+ {-# INLINE toChunks #-}+ toChunks residue newbs =+ -- Non-empty newbs expected.+ let tentativeChunks = Seq.fromList . B.split splitWd $ residue `B.append` newbs+ in case tentativeChunks of+ Seq.Empty -> (Seq.empty, "")+ init' Seq.:|> last' ->+ -- Note: This logic works also when newbs ends with splitWd because then the last+ -- chunk is the empty bytestring.+ (init', last')++ -- Processes chunks obtained with toChunks.+ {-# INLINE processChunks #-}+ processChunks is [] residue =+ return $ S.Skip (is, COResidue_ residue)+ processChunks is (chunk : chunks) residue =+ return $ S.Yield (Right chunk) (is, COProcessChunks_ chunks residue)++ {-# INLINE step' #-}+ -- "is": state of the input stream.+ -- "gst": "global" state? (Inspired by '_compactOnByteCustom' in streamly-0.10.1.)+ step' gst (is, s) = case s of+ COInitOrYieldHeader_ -> do+ istep' <- istep gst is+ case istep' of+ S.Stop -> return S.Stop+ S.Skip is' -> return $ S.Skip (is', COInitOrYieldHeader_)+ S.Yield e is' -> case e of+ Left hdr -> return $ S.Yield (Left hdr) (is', COInitOrYieldHeader_)+ Right newbs -> do+ -- Note: In the initial case (and not just the yield header case), this is possible.+ -- Although a bytestring appearing initially without any preceding header is not what+ -- we have in mind for streamly-archive, we want this function to focus only on the+ -- bytestring splitting.+ let (chunks, residue') = toChunks "" newbs+ return $ S.Skip (is', COProcessChunks_ (toList chunks) residue')+ COResidue_ !residue -> do+ istep' <- istep gst is+ case istep' of+ S.Stop -> return $ S.Yield (Right residue) (is, COStop_)+ S.Skip is' -> return $ S.Skip (is', COResidue_ residue)+ S.Yield e is' -> case e of+ Left hdr -> return $ S.Yield (Right residue) (is', COYieldHeader_ hdr is')+ Right newbs -> do+ let (chunks, residue') = toChunks residue newbs+ return $ S.Skip (is', COProcessChunks_ (toList chunks) residue')+ COStop_ -> return S.Stop+ COYieldHeader_ !hdr !is' -> return $ S.Yield (Left hdr) (is', COInitOrYieldHeader_)+ COProcessChunks_ !chunks !residue ->+ processChunks is chunks residue++-- | The state of the outer 'chunkOnFold' fold.+data ChunkOnFoldState_+ = -- | The initialization of the fold is complete. This state occurs only once (in the beginning).+ Init_+ | -- | The processing of a header is complete.+ Header_+ | -- | The processing of chunks is complete, and a residue (possibly empty) has been made+ -- available.+ Chunks_ !ByteString++-- | Chunks up the bytestrings following each @Left@ by the given word, discarding the given word.+-- (For instance, the word could be @10@ (newline), which gives us lines as the chunks.) The+-- bytestrings in the provided fold are the desired chunks.+{-# INLINE chunkOnFold #-}+chunkOnFold ::+ (Monad m) =>+ Word8 ->+ Fold m (Either a ByteString) b ->+ Fold m (Either a ByteString) b+chunkOnFold splitWd (F.Fold chstep chinit chextr chfinal) =+ -- "ch": chunk.+ let -- A utility function to consume all the chunks available in the same iteration.+ {-# INLINE go #-}+ go chs [] = return $ F.Partial chs -- "chs": state of the chunk fold.+ go chs (chbs : chbss) = do+ chstep' <- chstep chs (Right chbs)+ case chstep' of+ F.Done a -> return $ F.Done a+ F.Partial chs' -> go chs' chbss+ -- A utility function to obtain (chunks, next residue) from the previous residue and the+ -- latest incoming bytestring.+ {-# INLINE toChunks #-}+ toChunks residue newbs =+ -- Non-empty newbs expected.+ let tentativeChunks = Seq.fromList . B.split splitWd $ residue `B.append` newbs+ in case tentativeChunks of+ Seq.Empty -> (Seq.empty, "")+ init' Seq.:|> last' ->+ -- Note: This logic works also when newbs ends with splitWd because then the last+ -- chunk is the empty bytestring.+ (init', last')++ {-# INLINE processHeader #-}+ processHeader chs hdr = do+ chstep' <- chstep chs (Left hdr)+ case chstep' of+ F.Done a -> return $ F.Done a+ F.Partial chs' -> return $ F.Partial (chs', Header_)+ {-# INLINE processBytestring #-}+ processBytestring chs residue chbs = do+ let (chunks, residue') = toChunks residue chbs+ chstep' <- go chs (toList chunks)+ case chstep' of+ F.Done a -> return $ F.Done a+ F.Partial chs' -> return $ F.Partial (chs', Chunks_ residue')+ in -- Note: If a file ends with "\n", we want to include the last empty line.+ F.Fold+ ( \(chs, s) e -> case s of+ Init_ -> case e of+ Left hdr -> do+ processHeader chs hdr+ Right newbs ->+ -- This case is possible. Although a bytestring appearing initially without any+ -- preceding header is not what we have in mind for streamly-archive, we want this+ -- fold to focus only on the bytestring splitting.+ processBytestring chs "" newbs+ Header_ -> case e of+ Left hdr -> do+ -- No bytestrings followed the previous header.+ processHeader chs hdr+ Right newbs ->+ processBytestring chs "" newbs+ Chunks_ residue -> case e of+ Left hdr -> do+ chstep' <- chstep chs (Right residue)+ case chstep' of+ F.Done a -> return $ F.Done a+ F.Partial chs' -> do+ processHeader chs' hdr+ Right newbs ->+ processBytestring chs residue newbs+ )+ ( do+ chstep' <- chinit+ case chstep' of+ F.Done a -> return $ F.Done a+ F.Partial chs' -> return $ F.Partial (chs', Init_)+ )+ (\(chs, _) -> chextr chs)+ ( \(chs, s) -> case s of+ Chunks_ residue -> do+ chstep' <- chstep chs (Right residue)+ case chstep' of+ F.Done a -> return a+ F.Partial chs' -> chfinal chs'+ _ -> chfinal chs+ )
src/Streamly/External/Archive/Internal/Foreign.hs view
@@ -20,77 +20,75 @@ ) where -import Control.Exception (Exception, mask_, throw)-import Control.Monad (when)-import Data.Bits ((.&.))-import Data.ByteString (ByteString, packCString, packCStringLen)+import Control.Exception+import Control.Monad+import Data.Bits+import Data.ByteString import qualified Data.ByteString as B-import Data.Int (Int64)-import Foreign (FunPtr, Ptr, nullPtr, peek)-import Foreign.C.String (CString, peekCString, withCString)-import Foreign.C.Types (CChar, CInt (CInt), CSize (CSize))-import Foreign.ForeignPtr (ForeignPtr, newForeignPtr, withForeignPtr)-import Foreign.Marshal.Alloc (mallocBytes)-import System.Posix.Types (CMode (CMode), CSsize (CSsize))+import Data.Int+import Foreign+import Foreign.C.String+import Foreign.C.Types+import System.Posix.Types data CArchive data CEntry -foreign import ccall unsafe "archive.h archive_errno"+foreign import ccall safe "archive.h archive_errno" c_archive_errno :: Ptr CArchive -> IO CInt -foreign import ccall unsafe "archive.h archive_error_string"+foreign import ccall safe "archive.h archive_error_string" c_archive_error_string :: Ptr CArchive -> IO CString -foreign import ccall unsafe "archive.h archive_read_new"+foreign import ccall safe "archive.h archive_read_new" c_archive_read_new :: IO (Ptr CArchive) -foreign import ccall unsafe "archive.h archive_read_support_filter_all"+foreign import ccall safe "archive.h archive_read_support_filter_all" c_archive_read_support_filter_all :: Ptr CArchive -> IO CInt -foreign import ccall unsafe "archive.h archive_read_support_format_all"+foreign import ccall safe "archive.h archive_read_support_format_all" c_archive_read_support_format_all :: Ptr CArchive -> IO CInt -foreign import ccall unsafe "archive.h archive_read_support_format_gnutar"+foreign import ccall safe "archive.h archive_read_support_format_gnutar" c_archive_read_support_format_gnutar :: Ptr CArchive -> IO CInt -foreign import ccall unsafe "archive.h archive_read_open_filename"+foreign import ccall safe "archive.h archive_read_open_filename" c_archive_read_open_filename :: Ptr CArchive -> CString -> CSize -> IO CInt -foreign import ccall unsafe "archive.h archive_read_next_header2"+foreign import ccall safe "archive.h archive_read_next_header2" c_archive_read_next_header2 :: Ptr CArchive -> Ptr CEntry -> IO CInt -foreign import ccall unsafe "archive.h archive_read_data"+foreign import ccall safe "archive.h archive_read_data" -- Todo: Think about la_ssize_t on non-POSIX. c_archive_read_data :: Ptr CArchive -> Ptr CChar -> CSize -> IO CSsize -foreign import ccall unsafe "archive.h archive_read_data_block"+foreign import ccall safe "archive.h archive_read_data_block" c_archive_read_data_block :: Ptr CArchive -> Ptr (Ptr CChar) -> Ptr CSize -> Ptr Int64 -> IO CInt -foreign import ccall unsafe "archive.h archive_read_free"+foreign import ccall safe "archive.h archive_read_free" c_archive_read_free :: Ptr CArchive -> IO CInt -foreign import ccall unsafe "archive_entry.h archive_entry_filetype"+foreign import ccall safe "archive_entry.h archive_entry_filetype" c_archive_entry_filetype :: Ptr CEntry -> IO CMode -- Todo: Think about type on non-POSIX. -foreign import ccall unsafe "archive_entry.h archive_entry_new"+foreign import ccall safe "archive_entry.h archive_entry_new" c_archive_entry_new :: IO (Ptr CEntry) -- Similar to c_free_finalizer from ByteString.-foreign import ccall unsafe "static archive_entry.h &archive_entry_free"+foreign import ccall safe "static archive_entry.h &archive_entry_free" c_archive_entry_free_finalizer :: FunPtr (Ptr CEntry -> IO ()) -foreign import ccall unsafe "archive_entry.h archive_entry_pathname"+foreign import ccall safe "archive_entry.h archive_entry_pathname" c_archive_entry_pathname :: Ptr CEntry -> IO CString -foreign import ccall unsafe "archive_entry.h archive_entry_pathname_utf8"+foreign import ccall safe "archive_entry.h archive_entry_pathname_utf8" c_archive_entry_pathname_utf8 :: Ptr CEntry -> IO CString -foreign import ccall unsafe "archive_entry.h archive_entry_size"+foreign import ccall safe "archive_entry.h archive_entry_size" c_archive_entry_size :: Ptr CEntry -> IO Int64 -foreign import ccall unsafe "archive_entry.h archive_entry_size_is_set"+foreign import ccall safe "archive_entry.h archive_entry_size_is_set" c_archive_entry_size_is_set :: Ptr CEntry -> IO CInt -- Documented libarchive return codes.@@ -256,8 +254,8 @@ alloc_archive_read_data_buffer :: IO (Ptr CChar) alloc_archive_read_data_buffer = mallocBytes blockSize --- | Returns 'Nothing' if there is no more data for the current entry.--- Pass in a buffer allocated with 'alloc_archive_read_data_buffer'.+-- | Returns 'Nothing' if there is no more data for the current entry. Pass in a buffer allocated+-- with 'alloc_archive_read_data_buffer'. {-# INLINE archive_read_data #-} archive_read_data :: Archive -> Ptr CChar -> IO (Maybe ByteString) archive_read_data (Archive aptr) buf = do
streamly-archive.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: streamly-archive-version: 0.2.0+version: 0.3.0 synopsis: Stream data from archives using the streamly library. description: Please see the README on GitHub at <https://github.com/shlok/streamly-archive#readme> category: Archive, Codec, Streaming, Streamly@@ -8,7 +8,7 @@ bug-reports: https://github.com/shlok/streamly-archive/issues author: Shlok Datye maintainer: sd-haskell@quant.is-copyright: 2023 Shlok Datye+copyright: 2024 Shlok Datye license: BSD-3-Clause license-file: LICENSE build-type: Simple@@ -37,17 +37,19 @@ archive build-depends: base >=4.7 && <5- , bytestring ==0.11.*- , streamly ==0.9.*- , streamly-core ==0.1.*+ , bytestring >=0.10.10.0 && <0.12+ , containers >=0.6.2.1 && <0.8+ , streamly >=0.10.0 && <0.11+ , streamly-core >=0.2.0 && <0.3 default-language: Haskell2010 test-suite streamly-archive-test type: exitcode-stdio-1.0 main-is: TestSuite.hs other-modules:- Streamly.External.Archive.Tests+ ReadmeMain Paths_streamly_archive+ Streamly.External.Archive.Tests autogen-modules: Paths_streamly_archive hs-source-dirs:@@ -56,19 +58,22 @@ extra-libraries: archive build-depends:- QuickCheck >=2.13.2 && <2.15+ async >=2.2.2 && <2.3 , base >=4.7 && <5- , bytestring ==0.11.*+ , bytestring >=0.10.10.0 && <0.12+ , containers >=0.6.2.1 && <0.8 , cryptonite >=0.26- , directory >=1.3.6.0 && <1.4+ , directory >=1.3.1 && <1.4 , filepath >=1.4.2.1 && <1.5- , streamly ==0.9.*- , streamly-core ==0.1.*+ , QuickCheck >=2.13.2 && <2.15+ , split >=0.2.3.5 && <0.3+ , streamly >=0.10.0 && <0.11 , streamly-archive+ , streamly-core >=0.2.0 && <0.3 , tar >=0.5.1.1 && <0.6 , tasty >=1.2.3 && <1.5- , tasty-hunit >=0.10.0.2 && <0.11+ , tasty-hunit >=0.10.1 && <0.11 , tasty-quickcheck >=0.10.1.1 && <0.11- , temporary ==1.3.*+ , temporary >=1.3 && <1.4 , zlib >=0.6.2.1 && <0.7 default-language: Haskell2010
+ test/ReadmeMain.hs view
@@ -0,0 +1,47 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++module ReadmeMain where++import Crypto.Hash+import Data.ByteString (ByteString)+import Data.Function+import Data.Functor+import Data.Maybe+import Streamly.Data.Fold (Fold)+import qualified Streamly.Data.Fold as F+import qualified Streamly.Data.Stream.Prelude as S+import Streamly.External.Archive++main :: IO ()+main = do+ -- A fold for converting each archive entry (which is a Header followed by+ -- zero or more ByteStrings) into a path and corresponding SHA-256 hash+ -- (Nothing for no data).+ let entryFold :: Fold IO (Either Header ByteString) (String, Maybe String) =+ F.foldlM'+ ( \(mpath, mctx) e ->+ case e of+ Left h -> do+ mpath' <- headerPathName h+ return (mpath', mctx)+ Right bs ->+ return+ ( mpath,+ Just . (`hashUpdate` bs) $+ fromMaybe (hashInit @SHA256) mctx+ )+ )+ (return (Nothing, Nothing))+ <&> ( \(mpath, mctx) ->+ ( show $ fromMaybe (error "path expected") mpath,+ show . hashFinalize <$> mctx+ )+ )++ -- Execute the stream, grouping at the headers (the Lefts) using the above+ -- fold, and output the paths and SHA-256 hashes along the way.+ S.unfold readArchive (id, "/path/to/archive.tar.gz")+ & groupByLeft entryFold+ & S.mapM print+ & S.fold F.drain
test/Streamly/External/Archive/Tests.hs view
@@ -1,164 +1,303 @@ {-# LANGUAGE NumericUnderscores #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-} module Streamly.External.Archive.Tests (tests) where import qualified Codec.Archive.Tar as Tar-import Codec.Compression.GZip (compress)-import Control.Monad (forM)-import Crypto.Random.Entropy (getEntropy)-import Data.Bifunctor (first)-import Data.ByteString (ByteString, append)+import Codec.Compression.GZip+import Control.Concurrent.Async+import Control.Monad+import Crypto.Random.Entropy+import Data.Bifunctor+import Data.ByteString (ByteString) import qualified Data.ByteString as B-import Data.ByteString.Char8 (unpack)+import qualified Data.ByteString.Char8 as BC import qualified Data.ByteString.Lazy as LB-import Data.Char (chr, ord)-import Data.Function ((&))-import Data.List (nub, sort)-import Data.Maybe (fromJust)+import Data.Char+import Data.Function+import Data.Functor+import Data.List+import Data.List.Split+import Data.Maybe+import qualified Data.Set as Set+import Data.Word+import qualified Streamly.Data.Fold as F import qualified Streamly.Data.Stream.Prelude as S import Streamly.External.Archive-import Streamly.External.Archive.Internal.Foreign (blockSize)-import Streamly.Internal.Data.Fold.Type (Fold (Fold), Step (Partial))-import System.Directory (createDirectoryIfMissing)-import System.FilePath (addTrailingPathSeparator, hasTrailingPathSeparator, joinPath, takeDirectory)-import System.IO.Temp (withSystemTempDirectory)-import Test.QuickCheck (Gen, choose, frequency, vectorOf)-import Test.QuickCheck.Monadic (monadicIO, pick, run)-import Test.Tasty (TestTree)-import Test.Tasty.HUnit (assertBool, assertEqual, testCase)-import Test.Tasty.QuickCheck (testProperty)+import Streamly.External.Archive.Internal.Foreign+import System.Directory+import System.FilePath+import System.IO.Temp+import Test.QuickCheck+import Test.QuickCheck.Monadic+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck tests :: [TestTree] tests = [ testTar False, testTar True,- testSparse+ testSparse,+ testChunkOnAndChunkOnFold,+ testEitherByLeft ] --- | Use other libraries to create a tar (or tar.gz) file containing random data,--- read the file back using our library, and check if the results are as expected.+-- | Use other libraries to create a tar (or tar.gz) file containing random data, read the file back+-- using our library, and check if the results are as expected. testTar :: Bool -> TestTree testTar gz = testProperty ("tar (" ++ (if gz then "gz" else "no gz") ++ ")") $ monadicIO $ do- -- Generate a random file system hierarchy.- hierarchy <- pick $ randomHierarchy "" 4 4 5+ -- Generate a random file system hierarchy for writing to disk.+ hierarchyToWrite <- pick $ randomHierarchy "" 4 4 5 - -- Create a new temporary directory, write our hierarchy into- -- a "files" subdirectory of the temporary directory, use other- -- libraries to create files.tar (or files.tar.gz), read the file+ numThreads <- pick $ chooseInt (1, 4)++ -- Of the hierarchy we wrote, sometimes we only read some of them back.+ readSome :: Bool <- pick arbitrary+ readSomePaths <-+ pick $+ sublistOf hierarchyToWrite+ <&> Set.fromList+ . map (("files/" ++) . fst) -- Make comparable to what our library reads back.++ -- Create a new temporary directory, write our hierarchy into a "files" subdirectory of the+ -- temporary directory, use other libraries to create files.tar (or files.tar.gz), read the file -- back using our library, and check if the results are as expected. run . withSystemTempDirectory "archive-streaming-testZip" $ \tmpDir -> do let filesDir = joinPath [tmpDir, "files"] createDirectoryIfMissing True filesDir- pathsAndByteStrings <- writeHierarchy filesDir hierarchy+ writePathsAndByteStrings <- writeHierarchy filesDir hierarchyToWrite let archFile = joinPath [tmpDir, "files.tar" ++ (if gz then ".gz" else "")] LB.writeFile archFile . (if gz then compress else id) . Tar.write =<< Tar.pack tmpDir ["files"] let fileFold =- Fold+ F.foldlM' ( \(mfp, mtyp, msz, mbs) e -> case e of Left h -> do mfp_ <- headerPathName h mtyp_ <- headerFileType h msz_ <- headerSize h- return $ Partial (unpack <$> mfp_, mtyp_, msz_, mbs)+ return (BC.unpack <$> mfp_, mtyp_, msz_, mbs) Right bs ->- return $- Partial- ( mfp,- mtyp,- msz,- case mbs of- Nothing -> Just bs- Just bs' -> Just $ bs' `append` bs- )+ return+ ( mfp,+ mtyp,+ msz,+ case mbs of+ Nothing -> Just bs+ Just bs' -> Just $ bs' `BC.append` bs+ ) )- (return $ Partial (Nothing, Nothing, Nothing, Nothing))- return+ (return (Nothing, Nothing, Nothing, Nothing)) - pathsFileTypesSizesAndByteStrings <-- S.unfold (readArchive archFile) undefined- & groupByHeader fileFold- & fmap (\(mfp, mtyp, msz, mbs) -> (fromJust mfp, fromJust mtyp, msz, mbs))- & S.toList+ readPathsFileTypesSizesAndByteStringss <-+ replicateConcurrently numThreads $+ S.unfold+ readArchive+ ( if readSome+ then+ mapHeaderMaybe+ ( \h -> do+ p <- fromJust <$> headerPathName h+ return $+ if BC.unpack p `Set.member` readSomePaths+ then Just h+ else Nothing+ )+ else+ id,+ archFile+ )+ & groupByLeft fileFold+ & fmap (\(mfp, mtyp, msz, mbs) -> (fromJust mfp, fromJust mtyp, msz, mbs))+ & S.toList - -- Make the file paths and ByteStrings comparable and compare them.- let pathsAndByteStrings_ =- sort $ map (first ("files/" ++)) (("", Nothing) : pathsAndByteStrings)- let pathAndByteStrings2_ =- sort . map (\(x, _, _, y) -> (x, y)) $ pathsFileTypesSizesAndByteStrings- let samePathsAndByteStrings = pathsAndByteStrings_ == pathAndByteStrings2_+ threadResults <- forM readPathsFileTypesSizesAndByteStringss $+ \readPathsFileTypesSizesAndByteStrings -> do+ let readPathAndByteStrings =+ sort . map (\(x, _, _, y) -> (x, y)) $ readPathsFileTypesSizesAndByteStrings - -- Check FileType.- let fileTypesCorrect =- all- ( \(fp, typ, _, _) ->- if hasTrailingPathSeparator fp- then typ == FileTypeDirectory- else typ == FileTypeRegular- )- pathsFileTypesSizesAndByteStrings+ let writePathsAndByteStrings2 =+ sort+ . (if readSome then filter (\(x, _) -> x `Set.member` readSomePaths) else id)+ . map (first ("files/" ++)) -- Make comparable to what our library reads back.+ $ ("", Nothing) : writePathsAndByteStrings - -- Check header file size.- let fileSizeCorrect =- all- ( \(_, _, msz, mbs) ->- case (msz, mbs) of- (Nothing, _) -> False -- The size is always available.- (Just sz, Nothing) -> sz == 0 -- File or directory.- (Just sz, Just bs) -> fromIntegral sz == B.length bs- )- pathsFileTypesSizesAndByteStrings+ let samePathsAndByteStrings = writePathsAndByteStrings2 == readPathAndByteStrings - return $ samePathsAndByteStrings && fileTypesCorrect && fileSizeCorrect+ -- Check FileType.+ let fileTypesCorrect =+ all+ ( \(fp, typ, _, _) ->+ if hasTrailingPathSeparator fp+ then typ == FileTypeDirectory+ else typ == FileTypeRegular+ )+ readPathsFileTypesSizesAndByteStrings + -- Check header file size.+ let fileSizeCorrect =+ all+ ( \(_, _, msz, mbs) ->+ case (msz, mbs) of+ (Nothing, _) -> False -- The size is always available.+ (Just sz, Nothing) -> sz == 0 -- File or directory.+ (Just sz, Just bs) -> fromIntegral sz == B.length bs+ )+ readPathsFileTypesSizesAndByteStrings++ return $ samePathsAndByteStrings && fileTypesCorrect && fileSizeCorrect++ return $ and threadResults+ -- | Read a fixed sparse file (sparse.tar) using our library and make sure the results are as -- expected. (The file was created manually on Linux with "cp --sparse=always" to create the sparse -- files and "tar -Scvf" to create the archive. We were unable to do the equivalent thing on macOS -- Mojave / APFS.) testSparse :: TestTree-testSparse = testCase "sparse" $ do+testSparse = testProperty "sparse" $ monadicIO $ do let fileFold =- Fold+ F.foldlM' ( \(mfp, mbs) e -> case e of Left h -> do mfp_ <- headerPathName h- return $ Partial (unpack <$> mfp_, mbs)+ return (BC.unpack <$> mfp_, mbs) Right bs ->- return $- Partial- ( mfp,- case mbs of- Nothing -> Just bs- Just bs' -> Just $ bs' `append` bs- )+ return+ ( mfp,+ case mbs of+ Nothing -> Just bs+ Just bs' -> Just $ bs' `BC.append` bs+ ) )- (return $ Partial (Nothing, Nothing))- return+ (return (Nothing, Nothing)) - archive <-- S.unfold (readArchive "test/data/sparse.tar") undefined- & groupByHeader fileFold- & fmap (\(mfp, mbs) -> (fromJust mfp, fromJust mbs))- & S.toList+ numThreads <- pick $ chooseInt (1, 4) - assertEqual "" (map fst archive) ["zero", "zeroZero", "zeroAsdf", "asdfZero"]+ archives <-+ run $+ replicateConcurrently numThreads $+ S.unfold readArchive (id, "test/data/sparse.tar")+ & groupByLeft fileFold+ & fmap (\(mfp, mbs) -> (fromJust mfp, fromJust mbs))+ & S.toList - let tenMb = 10_000_000- let zero = B.replicate tenMb 0- let asdf = "asdf"+ threadResults <- forM archives $ \archive -> do+ let validPaths = map fst archive == ["zero", "zeroZero", "zeroAsdf", "asdfZero"] - assertBool "unexpected bytestring (1)" $ snd (head archive) == zero- assertBool "unexpected bytestring (2)" $ snd (archive !! 1) == zero `B.append` zero- assertBool "unexpected bytestring (3)" $ snd (archive !! 2) == zero `B.append` asdf- assertBool "unexpected bytestring (4)" $ snd (archive !! 3) == asdf `B.append` zero+ let tenMb = 10_000_000+ let zero = B.replicate tenMb 0+ let asdf = "asdf" --- | Writes a given hierarchy of relative paths (created with 'randomHierarchy') to disk--- in the specified directory and returns the same hierarchy except with actual ByteStrings--- instead of lengths. Note: The original relative paths are returned back unaltered.+ let validByteString1 = snd (head archive) == zero+ let validByteString2 = snd (archive !! 1) == zero `B.append` zero+ let validByteString3 = snd (archive !! 2) == zero `B.append` asdf+ let validByteString4 = snd (archive !! 3) == asdf `B.append` zero++ return $+ and+ [ validPaths,+ validByteString1,+ validByteString2,+ validByteString3,+ validByteString4+ ]++ return $ and threadResults++testChunkOnAndChunkOnFold :: TestTree+testChunkOnAndChunkOnFold = testProperty "chunkOn/chunkOnFold" $ monadicIO $ do+ -- Although we say “lines,” our splitWd is an arbitrary non-printable ASCII character.+ splitWd :: Word8 <- pick $ elements [0 .. 31]++ let maxLineLen = 50+ maxChunkSz = 100++ genLine = do+ lineLen <- choose (0, maxLineLen)+ B.pack+ <$>+ -- For easier debug visualization, lines have only a to z.+ replicateM lineLen (elements [97 .. 122])++ genLines = do+ numLines <- pick $ choose (0, 10)+ if numLines <= 2+ then+ replicateM numLines $ pick genLine+ else do+ lines' <- replicateM (numLines - 2) $ pick genLine+ -- Make it a bit more likely we begin/end with an empty line+ begLine <- pick $ frequency [(5, return ""), (95, genLine)]+ endLine <- pick $ frequency [(5, return ""), (95, genLine)]+ return $ [begLine] ++ lines' ++ [endLine]++ chunkSz <- pick $ choose (1, maxChunkSz)++ let linesToChunks [""] = [""] -- Special case.+ linesToChunks lns =+ map B.pack+ . chunksOf chunkSz+ . B.unpack+ -- ["line1", "\n" , "line2"] -> "line1\nline2"+ . B.concat+ -- ["line1", "line2"] -> ["line1", "\n" , "line2"]+ . intersperse (B.singleton splitWd)+ $ lns++ -- Most users of streamly-archive will probably have none of these, but we test this case+ -- nonetheless.+ linesBeforeFirstFile <- genLines++ numFiles <- pick $ choose (0, 5)++ files <- replicateM numFiles $ do+ fileLines <- genLines+ fileName :: Int <- pick arbitrary+ return (fileName, fileLines)++ let chunks =+ map Right (linesToChunks linesBeforeFirstFile)+ ++ concatMap+ (\(fileName, fileLines) -> Left fileName : map Right (linesToChunks fileLines))+ files+ expectedChunkOnResult =+ map Right linesBeforeFirstFile+ ++ concatMap+ (\(fileName, fileLines) -> Left fileName : map Right fileLines)+ files++ chunkOnResult <-+ S.fromList chunks+ & chunkOn splitWd+ & S.fold F.toList++ chunkOnFoldResult <-+ S.fromList chunks+ & S.fold (chunkOnFold splitWd F.toList)++ return $+ chunkOnResult == expectedChunkOnResult+ && chunkOnFoldResult == expectedChunkOnResult++testEitherByLeft :: TestTree+testEitherByLeft = testCase "eitherByLeft" $ do+ let getRes = S.fold F.toList . eitherByLeft . S.fromList++ res1 <- getRes [Right 10, Left "a", Right 1, Right 2, Left "b", Left "c", Right 20]+ res1 @?= ([("a", 1), ("a", 2), ("c", 20)] :: [(String, Int)])++ res2 <- getRes []+ res2 @?= ([] :: [(String, Int)])++-- | Writes a given hierarchy of relative paths (created with 'randomHierarchy') to disk in the+-- specified directory and returns the same hierarchy except with actual ByteStrings instead of+-- lengths. Note: The original relative paths are returned back unaltered. writeHierarchy :: FilePath -> [(FilePath, Maybe Int)] -> IO [(FilePath, Maybe ByteString)] writeHierarchy writeDir = mapM $ \(p, mBsLen) -> let fullp = joinPath [writeDir, p]@@ -175,10 +314,9 @@ ) Nothing -> createDirectoryIfMissing True fullp >> return (p, Nothing) --- | Recursively generates a random hierarchy of relative paths to files and--- directories. (Nothing is written to disk; only the paths are returned.)--- The initial dirPath should be "". A random bytestring length is--- provided in case of a file; 'Nothing' in the case of a directory.+-- | Recursively generates a random hierarchy of relative paths to files and directories. (Nothing+-- is written to disk; only the paths are returned.) The initial dirPath should be "". A random+-- bytestring length is provided in case of a file; 'Nothing' in the case of a directory. randomHierarchy :: FilePath -> Int -> Int -> Int -> Gen [(FilePath, Maybe Int)] randomHierarchy dirPath maxFiles maxDirs maxDepth = do numFiles <- choose (0, maxFiles)@@ -210,10 +348,10 @@ randomHierarchy dirPath' (maxFiles `div` 2) (maxDirs `div` 2) (maxDepth - 1) ) - return $ zip filePaths bsLengths ++ zip dirPaths (repeat Nothing) ++ recursion+ return $ zip filePaths bsLengths ++ map (,Nothing) dirPaths ++ recursion --- | Generates a random path component of length between 1 and 10, e.g., "HO53UVKQ".--- For compatibility with case-insensitive file systems, uses only one case.+-- | Generates a random path component of length between 1 and 10, e.g., "HO53UVKQ". For+-- compatibility with case-insensitive file systems, uses only one case. pathComponent :: Gen String pathComponent = do len <- choose (1, 10)