diff --git a/Main.hs b/Main.hs --- a/Main.hs +++ b/Main.hs @@ -1,23 +1,23 @@ -{-# LANGUAGE CPP #-} +{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} module Main where -import Data.Conduit (($$)) -import Data.Conduit.Binary (sourceHandle) -import Data.Text (pack) +import Data.Conduit ( runConduit, (.|) ) +import Data.Conduit.Binary ( sourceHandle ) +import Data.Text ( pack ) -import Network.AWS -import Network.AWS.Data.Text (fromText) -import Network.AWS.S3.CreateMultipartUpload -import Network.AWS.S3.StreamingUpload +import Network.AWS +import Network.AWS.Data.Text ( fromText ) +import Network.AWS.S3.CreateMultipartUpload +import Network.AWS.S3.StreamingUpload -import System.Environment -import System.IO (BufferMode(BlockBuffering), - hSetBuffering, stdin) +import Control.Monad.IO.Class ( liftIO ) +import System.Environment +import System.IO ( BufferMode(BlockBuffering), hSetBuffering, stdin ) #if !MIN_VERSION_base(4,8,0) -import Control.Applicative (pure, (<$>), (<*>)) +import Control.Applicative ( pure, (<$>), (<*>) ) #endif main :: IO () @@ -31,17 +31,14 @@ <*> fromText (pack key) of Right (creds,_reg,buck,ky) -> do -#if !MIN_VERSION_amazonka(1,4,4) - env <- newEnv _reg creds -#else env <- newEnv creds -#endif hSetBuffering stdin (BlockBuffering Nothing) res <- runResourceT . runAWS env $ case file of - -- Stream data from stdin - "-" -> sourceHandle stdin $$ streamUpload Nothing (createMultipartUpload buck ky) - -- Read from a file - _ -> concurrentUpload Nothing Nothing (FP file) $ createMultipartUpload buck ky + "-" -> runConduit (sourceHandle stdin .| streamUpload Nothing (createMultipartUpload buck ky)) + >>= liftIO . either print print + _ -> concurrentUpload Nothing Nothing (FP file) (createMultipartUpload buck ky) + >>= liftIO . print + print res Left err -> print err >> usage ("abort":region:profile:credfile:bucket:_) -> @@ -50,11 +47,7 @@ <*> fromText (pack bucket) of Right (creds,_reg,buck) -> do -#if !MIN_VERSION_amazonka(1,4,4) - env <- newEnv _reg creds -#else env <- newEnv creds -#endif res <- runResourceT . runAWS env . abortAllUploads $ buck print res Left err -> print err >> usage diff --git a/amazonka-s3-streaming.cabal b/amazonka-s3-streaming.cabal --- a/amazonka-s3-streaming.cabal +++ b/amazonka-s3-streaming.cabal @@ -1,5 +1,5 @@ name: amazonka-s3-streaming -version: 0.2.0.5 +version: 1.0.0.0 synopsis: Provides conduits to upload data to S3 using the Multipart API description: Provides a conduit based streaming interface and a concurrent interface to uploading data to S3 using the Multipart API. Also provides method to upload @@ -9,50 +9,35 @@ license-file: LICENSE author: Alex Mason maintainer: Alex.Mason@data61.csiro.au -copyright: Copyright (c) 2016 Commonwealth Scientific and Industrial Research Organisation (CSIRO) +copyright: Alex Mason, Copyright (c) 2016 Commonwealth Scientific and Industrial Research Organisation (CSIRO) category: Network, AWS, Cloud, Distributed Computing build-type: Simple extra-source-files: README.md, Changelog.md cabal-version: >=1.10 -tested-with: GHC == 7.8.4, GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.* +tested-with: GHC == 8.0.* || == 8.2.2 || == 8.4.* || == 8.6.* library hs-source-dirs: src exposed-modules: Network.AWS.S3.StreamingUpload default-language: Haskell2010 build-depends: base >= 4.6 && < 5 - , amazonka >= 1.3 && < 1.6 - , amazonka-core >= 1.3 && < 1.6 - , amazonka-s3 >= 1.3 && < 1.6 - , resourcet >= 1.1.7.4 && < 1.2 - , conduit >= 1.2.6.6 && < 1.3 + , amazonka >= 1.6 && < 1.7 + , amazonka-core >= 1.6 && < 1.7 + , amazonka-s3 >= 1.6 && < 1.7 + , conduit >= 1.3 && < 1.4 , bytestring >= 0.10.8.0 && < 0.11 , mmorph >= 1.0.6 && < 1.2 , lens >= 4.13 && < 5.0 , mtl >= 2.2.1 && < 2.3 - , exceptions >= 0.8.2.1 && < 0.9 + , exceptions >= 0.8.2.1 && < 0.11 , dlist >= 0.8 && < 0.9 - , lifted-async >= 0.9 && < 0.10 - , mmap >= 0.5 && < 0.6 - , deepseq >= 1.2 && < 1.5 + , async >= 2 && < 2.3 , http-client >= 0.4 && < 0.6 - if impl(ghc < 7.11) - build-depends: - semigroups >= 0.12 - , transformers >=0.4 - if impl(ghc == 7.8.*) - build-depends: - foundation == 0.0.15 - - -- -- ini 0.3.4 fails to build because it doesn't have <* - -- if impl(ghc < 7.9) - -- build-depends: - -- ini >= 0.3.5 - -Flag s3upload-exe +flag s3upload-exe Description: Whether to build the s3upload executable for uploading files using this library. - Default: False + default: False + manual: True source-repository head type: git @@ -62,7 +47,10 @@ main-is: Main.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N default-language: Haskell2010 - buildable: False + if flag(s3upload-exe) + buildable: True + else + buildable: False build-depends: base , amazonka , amazonka-core @@ -70,7 +58,5 @@ , amazonka-s3-streaming , conduit-extra , conduit - , bytestring , text - if flag(s3upload-exe) - buildable: True + diff --git a/src/Network/AWS/S3/StreamingUpload.hs b/src/Network/AWS/S3/StreamingUpload.hs --- a/src/Network/AWS/S3/StreamingUpload.hs +++ b/src/Network/AWS/S3/StreamingUpload.hs @@ -1,80 +1,66 @@ -{-# LANGUAGE BangPatterns #-} -{-# LANGUAGE CPP #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE ParallelListComp #-} -{-# LANGUAGE PatternGuards #-} -{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE ParallelListComp #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} module Network.AWS.S3.StreamingUpload -( - streamUpload - , UploadLocation(..) - , concurrentUpload - , abortAllUploads - , module Network.AWS.S3.CreateMultipartUpload - , module Network.AWS.S3.CompleteMultipartUpload - , minimumChunkSize -) where + ( streamUpload + , ChunkSize + , minimumChunkSize + , NumThreads + , concurrentUpload + , UploadLocation(..) + , abortAllUploads + , module Network.AWS.S3.CreateMultipartUpload + , module Network.AWS.S3.CompleteMultipartUpload + ) where -import Network.AWS (HasEnv(..), - LogLevel(..), MonadAWS, - getFileSize, - hashedBody, send, - toBody) +import Network.AWS + ( AWS, HasEnv(..), LogLevel(..), MonadAWS, getFileSize, hashedBody, hashedFileRange, + liftAWS, runAWS, runResourceT, send, toBody ) -import Control.Monad.Trans.AWS (AWSConstraint) -import Network.AWS.Data.Crypto (Digest, SHA256, - hashFinalize, hashInit, - hashUpdate) +import Network.AWS.Data.Crypto ( Digest, SHA256, hashFinalize, hashInit, hashUpdate ) -import Network.AWS.S3.AbortMultipartUpload -import Network.AWS.S3.CompleteMultipartUpload -import Network.AWS.S3.CreateMultipartUpload -import Network.AWS.S3.ListMultipartUploads -import Network.AWS.S3.Types (BucketName, cmuParts, completedMultipartUpload, - completedPart, muKey, - muUploadId) -import Network.AWS.S3.UploadPart +import Network.AWS.S3.AbortMultipartUpload +import Network.AWS.S3.CompleteMultipartUpload +import Network.AWS.S3.CreateMultipartUpload +import Network.AWS.S3.ListMultipartUploads +import Network.AWS.S3.Types + ( BucketName, cmuParts, completedMultipartUpload, completedPart, muKey, muUploadId ) +import Network.AWS.S3.UploadPart -import Control.Applicative -import Control.Category ((>>>)) -import Control.Monad (forM_, when, (>=>)) -import Control.Monad.IO.Class (MonadIO, liftIO) -import Control.Monad.Morph (lift) -import Control.Monad.Reader.Class (local) -import Control.Monad.Trans.Resource (MonadBaseControl, - MonadResource) +import Control.Applicative +import Control.Category ( (>>>) ) +import Control.Monad ( forM_, when, (>=>) ) +import Control.Monad.IO.Class ( MonadIO, liftIO ) +import Control.Monad.Morph ( lift ) +import Control.Monad.Reader.Class ( local ) -import Data.Conduit (Sink, await) -import Data.Conduit.List (sourceList) +import Conduit ( MonadUnliftIO(..) ) +import Data.Conduit ( ConduitT, Void, await, catchC ) +import Data.Conduit.List ( sourceList ) -import Data.ByteString (ByteString) -import qualified Data.ByteString as BS -import Data.ByteString.Builder (stringUtf8) -import System.IO.MMap (mmapFileByteString) +import Data.ByteString ( ByteString ) +import qualified Data.ByteString as BS +import Data.ByteString.Builder ( stringUtf8 ) -import Control.DeepSeq (rnf) -import qualified Data.DList as D -import Data.List (unfoldr) -import Data.List.NonEmpty (nonEmpty) +import qualified Data.DList as D +import Data.List ( unfoldr ) +import Data.List.NonEmpty ( nonEmpty ) -import Control.Lens (set, view) -import Control.Lens.Operators -import Control.Monad.Catch (onException) +import Control.Lens ( set, view ) +import Control.Lens.Operators -import Text.Printf (printf) +import Text.Printf ( printf ) -import Control.Concurrent (newQSem, signalQSem, - waitQSem) -import Control.Concurrent.Async.Lifted (forConcurrently) -import System.Mem (performGC) +import Control.Concurrent ( newQSem, signalQSem, waitQSem ) +import Control.Concurrent.Async ( forConcurrently ) +import Control.Exception.Base ( SomeException, bracket_ ) +import Control.Monad.Catch ( onException ) +import System.Mem ( performGC ) -import Network.HTTP.Client (defaultManagerSettings, - managerConnCount, - newManager) +import Network.HTTP.Client ( defaultManagerSettings, managerConnCount, newManager ) type ChunkSize = Int type NumThreads = Int @@ -86,11 +72,11 @@ {- | Given a 'CreateMultipartUpload', creates a 'Sink' which will sequentially -upload the data streamed in in chunks of at least 'chunkSize' and return -the 'CompleteMultipartUploadResponse'. +upload the data streamed in in chunks of at least 'minimumChunkSize' and return either +the 'CompleteMultipartUploadResponse', or if an exception is thrown, +`AbortMultipartUploadResponse` and the exception as `SomeException`. If aborting +the upload also fails then the exception caused by the call to abort will be thrown. -If uploading of any parts fails, an attempt is made to abort the Multipart -upload, but it is likely that if an upload fails this abort may also fail. 'Network.AWS.S3.ListMultipartUploads' can be used to list any pending uploads - it is important to abort multipart uploads because you will be charged for storage of the parts until it is completed or aborted. @@ -98,51 +84,45 @@ May throw 'Network.AWS.Error' -} -streamUpload :: (MonadResource m, AWSConstraint r m, MonadAWS m) +streamUpload :: (MonadUnliftIO m, MonadAWS m) => Maybe ChunkSize -- ^ Optional chunk size -> CreateMultipartUpload -- ^ Upload location - -> Sink ByteString m CompleteMultipartUploadResponse -streamUpload mcs cmu = do - logger <- lift $ view envLogger + -> ConduitT ByteString Void m (Either (AbortMultipartUploadResponse, SomeException) CompleteMultipartUploadResponse) +streamUpload mChunkSize multiPartUploadDesc = do + logger <- lift $ liftAWS $ view envLogger let logStr :: MonadIO m => String -> m () logStr = liftIO . logger Debug . stringUtf8 - chunkSize = maybe minimumChunkSize (max minimumChunkSize) mcs + chunkSize = maybe minimumChunkSize (max minimumChunkSize) mChunkSize - cmur <- lift $ send cmu - when (cmur ^. cmursResponseStatus /= 200) $ + multiPartUpload <- lift $ send multiPartUploadDesc + when (multiPartUpload ^. cmursResponseStatus /= 200) $ fail "Failed to create upload" logStr "\n**** Created upload\n" - let Just upId = cmur ^. cmursUploadId - bucket = cmu ^. cmuBucket - key = cmu ^. cmuKey - -- go :: Text -> Builder -> Int -> Int -> Sink ByteString m () - go !bss !bufsize !ctx !partnum !completed = Data.Conduit.await >>= \mbs -> case mbs of - Just bs | l <- BS.length bs - , bufsize + l <= chunkSize -> - go (D.snoc bss bs) (bufsize + l) (hashUpdate ctx bs) partnum completed - - | otherwise -> do - rs <- lift $ partUploader partnum (bufsize + BS.length bs) - (hashFinalize $ hashUpdate ctx bs) - (D.snoc bss bs) + let Just upId = multiPartUpload ^. cmursUploadId + bucket = multiPartUploadDesc ^. cmuBucket + key = multiPartUploadDesc ^. cmuKey + -- go :: DList ByteString -> Int -> Context SHA256 -> Int -> DList (Maybe CompletedPart) -> Sink ByteString m () + go !bss !bufsize !ctx !partnum !completed = await >>= \case + Just bs + | l <- BS.length bs, bufsize + l <= chunkSize + -> go (D.snoc bss bs) (bufsize + l) (hashUpdate ctx bs) partnum completed + | otherwise -> do + rs <- lift $ performUpload partnum (bufsize + BS.length bs) + (hashFinalize $ hashUpdate ctx bs) + (D.snoc bss bs) - logStr $ printf "\n**** Uploaded part %d size %d\n" partnum bufsize - let part = completedPart partnum <$> (rs ^. uprsETag) -#if MIN_VERSION_amazonka_s3(1,4,1) - !_ = rnf part -#endif - liftIO performGC - go empty 0 hashInit (partnum+1) . D.snoc completed $! part + logStr $ printf "\n**** Uploaded part %d size %d\n" partnum bufsize + let !part = completedPart partnum <$> (rs ^. uprsETag) + liftIO performGC + go empty 0 hashInit (partnum+1) . D.snoc completed $! part Nothing -> lift $ do prts <- if bufsize > 0 then do - rs <- partUploader partnum bufsize (hashFinalize ctx) bss - + rs <- performUpload partnum bufsize (hashFinalize ctx) bss logStr $ printf "\n**** Uploaded (final) part %d size %d\n" partnum bufsize - let allParts = D.toList $ D.snoc completed $ completedPart partnum <$> (rs ^. uprsETag) pure $ nonEmpty =<< sequence allParts else do @@ -153,8 +133,8 @@ & cMultipartUpload ?~ set cmuParts prts completedMultipartUpload - partUploader :: MonadAWS m => Int -> Int -> Digest SHA256 -> D.DList ByteString -> m UploadPartResponse - partUploader pnum size digest = + performUpload :: MonadAWS m => Int -> Int -> Digest SHA256 -> D.DList ByteString -> m UploadPartResponse + performUpload pnum size digest = D.toList >>> sourceList >>> hashedBody digest (fromIntegral size) @@ -168,8 +148,9 @@ when (upr ^. uprsResponseStatus /= 200) $ fail "Failed to upload piece" return upr - go D.empty 0 hashInit 1 D.empty `onException` lift (send (abortMultipartUpload bucket key upId)) - -- Whatever happens, we abort the upload and rethrow + (Right <$> go D.empty 0 hashInit 1 D.empty) `catchC` \(except :: SomeException) -> + Left . (,except) <$> lift (send (abortMultipartUpload bucket key upId)) + -- Whatever happens, we abort the upload and return the exception -- | Specifies whether to upload a file or 'ByteString @@ -177,10 +158,6 @@ = FP FilePath -- ^ A file to be uploaded | BS ByteString -- ^ A strict 'ByteString' - -- IO (Int -> IO (Maybe ByteString)) (Either (IO ()) (IO ())) - -- part number as input, may be called many times until Nothing is returned, - -- and a function to close either this part or all parts - {-| Allows a file or 'ByteString' to be uploaded concurrently, using the async library. The chunk size may optionally be specified, but will be at least @@ -196,78 +173,80 @@ multipart upload on any error, but this may also fail if, for example, the network connection has been broken. See `abortAllUploads` for a crude cleanup method. -} -concurrentUpload :: (AWSConstraint r m, MonadAWS m, MonadBaseControl IO m) +concurrentUpload :: (MonadAWS m) => Maybe ChunkSize -- ^ Optional chunk size -> Maybe NumThreads -- ^ Optional number of threads to upload with -> UploadLocation -- ^ Whether to upload a file on disk or a `ByteString` that's already in memory. -> CreateMultipartUpload -- ^ Description of where to upload. -> m CompleteMultipartUploadResponse -concurrentUpload mcs mnt ud cmu = do - cmur <- send cmu - when (cmur ^. cmursResponseStatus /= 200) $ - fail "Failed to create upload" - logger <- view envLogger - let logStr :: MonadIO m => String -> m () - logStr = liftIO . logger Info . stringUtf8 - let Just upId = cmur ^. cmursUploadId - bucket = cmu ^. cmuBucket - key = cmu ^. cmuKey +concurrentUpload mChunkSize mNumThreads uploadLoc multiPartUploadDesc = do + env <- liftAWS $ view environment + cmur <- send multiPartUploadDesc + when (cmur ^. cmursResponseStatus /= 200) $ + fail "Failed to create upload" - calcChunkSize :: Int -> Int - calcChunkSize len = - let chunkSize' = maybe minimumChunkSize (max minimumChunkSize) mcs - in if len `div` chunkSize' >= 10000 then len `div` 9999 else chunkSize' + let logStr :: MonadIO m => String -> m () + logStr = liftIO . (env ^. envLogger) Info . stringUtf8 + bucket = multiPartUploadDesc ^. cmuBucket + key = multiPartUploadDesc ^. cmuKey + Just upId = cmur ^. cmursUploadId - let mConnCount = managerConnCount defaultManagerSettings - nThreads = maybe mConnCount (max 1) mnt - exec run = if maybe False (> mConnCount) mnt - then do - mgr' <- liftIO $ newManager defaultManagerSettings{managerConnCount = nThreads} - local (envManager .~ mgr') run - else run - exec $ flip onException (send (abortMultipartUpload bucket key upId)) $ do - sem <- liftIO $ newQSem nThreads - umrs <- case ud of - BS bs -> - let chunkSize = calcChunkSize $ BS.length bs - in forConcurrently (zip [1..] $ chunksOf chunkSize bs) $ \(partnum, b) -> do - liftIO $ waitQSem sem - logStr $ "Starting part: " ++ show partnum - umr <- send . uploadPart bucket key partnum upId . toBody $ b - logStr $ "Finished part: " ++ show partnum - liftIO $ signalQSem sem - pure $ completedPart partnum <$> (umr ^. uprsETag) + calculateChunkSize :: Int -> Int + calculateChunkSize len = + let chunkSize' = maybe minimumChunkSize (max minimumChunkSize) mChunkSize + in if len `div` chunkSize' >= 10000 then len `div` 9999 else chunkSize' - FP fp -> do - fsize <- liftIO $ getFileSize fp - let chunkSize = calcChunkSize $ fromIntegral fsize - (count,lst) = divMod (fromIntegral fsize) chunkSize - params = [(partnum, chunkSize*offset, size) - | partnum <- [1..] - | offset <- [0..count] - | size <- (chunkSize <$ [0..count-1]) ++ [lst] - ] + mConnCount = managerConnCount defaultManagerSettings + nThreads = maybe mConnCount (max 1) mNumThreads - forConcurrently params $ \(partnum,off,size) -> do - liftIO $ waitQSem sem - b <- liftIO $ mmapFileByteString fp (Just (fromIntegral off,size)) - umr <- send . uploadPart bucket key partnum upId . toBody $ b - liftIO $ signalQSem sem - pure $ completedPart partnum <$> (umr ^. uprsETag) + exec :: MonadAWS m => AWS a -> m a + exec act = if maybe False (> mConnCount) mNumThreads + then do + mgr' <- liftIO $ newManager defaultManagerSettings{managerConnCount = nThreads} + liftAWS $ local (envManager .~ mgr') act + else liftAWS act + exec $ flip onException (send (abortMultipartUpload bucket key upId)) $ do + sem <- liftIO $ newQSem nThreads + uploadResponses <- case uploadLoc of + BS bytes -> + let chunkSize = calculateChunkSize $ BS.length bytes + in liftIO $ forConcurrently (zip [1..] $ chunksOf chunkSize bytes) $ \(partnum, chunk) -> + bracket_ (waitQSem sem) (signalQSem sem) $ do + logStr $ "Starting part: " ++ show partnum + umr <- runResourceT $ runAWS env $ send . uploadPart bucket key partnum upId . toBody $ chunk + logStr $ "Finished part: " ++ show partnum + pure $ completedPart partnum <$> (umr ^. uprsETag) - let prts = nonEmpty =<< sequence umrs - send $ completeMultipartUpload bucket key upId - & cMultipartUpload ?~ set cmuParts prts completedMultipartUpload + FP filePath -> do + fsize <- liftIO $ getFileSize filePath + let chunkSize = calculateChunkSize $ fromIntegral fsize + (count,lst) = fromIntegral fsize `divMod` chunkSize + params = [(partnum, chunkSize*offset, size) + | partnum <- [1..] + | offset <- [0..count] + | size <- (chunkSize <$ [0..count-1]) ++ [lst] + ] + liftIO $ forConcurrently params $ \(partnum,off,size) -> + bracket_ (waitQSem sem) (signalQSem sem) $ do + logStr $ "Starting file part: " ++ show partnum + chunkStream <- hashedFileRange filePath (fromIntegral off) (fromIntegral size) + uploadResp <- runResourceT $ runAWS env $ + send . uploadPart bucket key partnum upId . toBody $ chunkStream + logStr $ "Finished file part: " ++ show partnum + pure $ completedPart partnum <$> (uploadResp ^. uprsETag) + + let parts = nonEmpty =<< sequence uploadResponses + send $ completeMultipartUpload bucket key upId + & cMultipartUpload ?~ set cmuParts parts completedMultipartUpload + -- | Aborts all uploads in a given bucket - useful for cleaning up. abortAllUploads :: (MonadAWS m) => BucketName -> m () abortAllUploads bucket = do rs <- send (listMultipartUploads bucket) forM_ (rs ^. lmursUploads) $ \mu -> do let mki = (,) <$> mu ^. muKey <*> mu ^. muUploadId - case mki of - Nothing -> pure () - Just (key,uid) -> send (abortMultipartUpload bucket key uid) >> pure () + forM_ mki $ \(key,uid) -> send (abortMultipartUpload bucket key uid)