diff --git a/package.yaml b/package.yaml index 9c0f0f2..16e2d82 100644 --- a/package.yaml +++ b/package.yaml @@ -1,5 +1,5 @@ name: taskrunner -version: 0.18.0.10 +version: 0.18.0.11 github: "githubuser/taskrunner" license: BSD-3-Clause author: "Author name here" diff --git a/src/RemoteCache.hs b/src/RemoteCache.hs index 48fa670..bf5c16d 100644 --- a/src/RemoteCache.hs +++ b/src/RemoteCache.hs @@ -26,7 +26,7 @@ import Network.URI (parseURI, URI (..), URIAuth(..)) import System.Directory (makeAbsolute, canonicalizePath) import System.FilePath (makeRelative) import qualified System.FilePath as FP -import Utils (bail, logDebug, logFileName, logInfo, withStderrPipe) +import Utils (bail, bytesfmt, formatSeconds, logDebug, logFileName, logInfo, timed, transferSummary, withStderrPipe) import qualified Amazonka as AWS import Control.Exception.Lens (handling) import System.Exit (ExitCode(..)) @@ -35,6 +35,7 @@ import qualified Data.Conduit.Text as CT import qualified Data.Text.Lazy as TL import qualified Data.Text.Lazy.Builder as TLB import Amazonka.S3.PutObject (newPutObject, PutObject(..)) +import GHC.Clock (getMonotonicTime) packTar :: MonadResource m => AppState -> Handle -> FilePath -> [FilePath] -> ConduitT () BS.ByteString m () @@ -114,7 +115,6 @@ parseEndpoint s = do . (\svc -> svc { s3AddressingStyle = S3AddressingStylePath }) -- TODO: --- - report speed, size etc. -- - integrate amazonka logging -- - handle errors saveCache @@ -145,13 +145,20 @@ saveCache appState settings relativeCacheRoot files archiveName = do logDebug appState $ "Uploading to s3://" <> bucket <> "/" <> objectKey - withStderrPipe appState \stderrHandle -> + -- Taps either side of zstd, so a slow save can be pinned on reading the + -- files, on compressing them, or on the network. + packStatsRef <- newIORef emptyTransferStats + uploadStatsRef <- newIORef emptyTransferStats + + (_, elapsed) <- timed $ withStderrPipe appState \stderrHandle -> runConduitRes do let multipartUpload = (newCreateMultipartUpload (BucketName bucket) (ObjectKey objectKey) :: CreateMultipartUpload) { storageClass = Just StorageClass_REDUCED_REDUNDANCY } result <- packTar appState stderrHandle cacheRoot filesRelativeToCacheRoot + .| measureTransfer packStatsRef .| Zstd.compress 3 + .| measureTransfer uploadStatsRef .| streamUpload env Nothing multipartUpload case result of Left (_, err) -> @@ -160,6 +167,77 @@ saveCache appState settings relativeCacheRoot files archiveName = do liftIO $ logDebug appState "Upload success" pure () + packStats <- readIORef packStatsRef + uploadStats <- readIORef uploadStatsRef + -- The two taps nest rather than partition: when zstd wants input it pulls + -- through the upstream tap, so the time the upload tap spent waiting for + -- zstd already contains the time spent waiting for tar. Subtracting leaves + -- compression on its own, and the three then add up to the elapsed time. + let readingSeconds = packStats.producingSeconds + compressionSeconds = max 0 (uploadStats.producingSeconds - packStats.producingSeconds) + uploadSeconds = uploadStats.consumingSeconds + -- Largest figure is the bottleneck. See 'measureTransfer' for why the + -- network one is a lower bound. + logDebug appState $ "Packed and uploaded " <> transferSummary uploadStats.bytes elapsed + <> ", compressed from " <> toText (bytesfmt "%.2f" packStats.bytes) + <> " - blocked on reading files " <> formatSeconds readingSeconds + <> ", on compression " <> formatSeconds compressionSeconds + <> ", on upload " <> formatSeconds uploadSeconds + +-- | Bytes seen, and how the wall clock divided between producing them and +-- consuming them. +data TransferStats = TransferStats + { bytes :: !Int + , producingSeconds :: !Double + , consumingSeconds :: !Double + } + +emptyTransferStats :: TransferStats +emptyTransferStats = TransferStats + { bytes = 0 + , producingSeconds = 0 + , consumingSeconds = 0 + } + +-- | Pass data through unchanged, recording how long the pipeline sat blocked +-- waiting for upstream to hand over a chunk versus blocked waiting for +-- downstream to accept one. +-- +-- Conduit runs the two strictly alternately - 'C.await' returns once upstream +-- has a chunk, and 'C.yield' returns once downstream wants the next one - so +-- together these account for the pipeline's whole wall clock. +-- +-- These are *stall* times, not time spent doing the work, and for I/O the two +-- differ. A @write@ to a socket returns as soon as the data is copied into the +-- kernel's send buffer; the kernel then transmits it while the next chunk is +-- being compressed. So the transfer overlaps the rest of the pipeline and is +-- undercounted here - if the pipeline is CPU-bound the writes cost almost +-- nothing. Reads are the mirror image: data accumulates in the receive buffer +-- while we are busy, so an 'await' that finds it already there costs nothing. +-- +-- Read these numbers as "where did the pipeline stall", which is what identifies +-- the bottleneck. Do not read them as "how long the bytes spent in transit". +measureTransfer :: MonadIO m => IORef TransferStats -> ConduitT BS.ByteString BS.ByteString m () +measureTransfer ref = loop + where + loop = do + beforeAwait <- liftIO getMonotonicTime + m_chunk <- C.await + afterAwait <- liftIO getMonotonicTime + case m_chunk of + Nothing -> + liftIO $ modifyIORef' ref \stats -> stats + { producingSeconds = stats.producingSeconds + (afterAwait - beforeAwait) } + Just chunk -> do + C.yield chunk + afterYield <- liftIO getMonotonicTime + liftIO $ modifyIORef' ref \stats -> stats + { bytes = stats.bytes + BS.length chunk + , producingSeconds = stats.producingSeconds + (afterAwait - beforeAwait) + , consumingSeconds = stats.consumingSeconds + (afterYield - afterAwait) + } + loop + data LogMode = NoLog | Log deriving (Eq, Show) restoreCache @@ -181,14 +259,28 @@ restoreCache appState settings cacheRoot archiveName logMode = do logDebug appState $ "Remote cache archive not found s3://" <> bucket <> "/" <> objectKey pure False - handling _NoSuchKey onNoSuchKey $ withStderrPipe appState \stderrHandle -> - runConduitRes do + handling _NoSuchKey onNoSuchKey $ withStderrPipe appState \stderrHandle -> do + statsRef <- newIORef emptyTransferStats + + (_, elapsed) <- timed $ runConduitRes do response <- AWS.send env $ newGetObject (BucketName bucket) (ObjectKey objectKey) when (logMode == Log) do liftIO $ logInfo appState $ "Found remote cache " <> archiveName <> ", restoring" response.body.body + .| measureTransfer statsRef .| unpackTar appState stderrHandle cacheRoot - pure True + + stats <- readIORef statsRef + -- The size is that of the compressed archive. Only a two-way split is + -- available here: zstd runs inside tar, so decompression cannot be + -- distinguished from the file writes. Note the download side excludes + -- connection setup, which happened above in AWS.send, and see + -- 'measureTransfer' for why it is a lower bound. + logDebug appState $ "Downloaded and unpacked " <> transferSummary stats.bytes elapsed + <> " - blocked on download " <> formatSeconds stats.producingSeconds + <> ", on decompression and unpacking " <> formatSeconds stats.consumingSeconds + + pure True getLatestBuildHash :: AppState diff --git a/src/Utils.hs b/src/Utils.hs index 30a67eb..c8ef005 100644 --- a/src/Utils.hs +++ b/src/Utils.hs @@ -17,6 +17,7 @@ import GHC.IO.Handle (hIsClosed) import System.FilePath (()) import Control.Concurrent.Async (async, wait) import System.Timeout (timeout) +import GHC.Clock (getMonotonicTime) outputLine :: AppState -> Handle -> ByteString -> ByteString -> IO () outputLine appState toplevelOutput streamName line = do @@ -90,6 +91,34 @@ bytesfmt formatter bs = printf (formatter <> " %s") bytesSuffixes = ["B", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB"] bytesSuffix = bytesSuffixes !! i +-- | Run an action, also returning how long it took, in seconds. +timed :: MonadIO m => m a -> m (a, Double) +timed action = do + start <- liftIO getMonotonicTime + result <- action + end <- liftIO getMonotonicTime + pure (result, end - start) + +-- | Format a duration in seconds, e.g. @"1.50s"@. +formatSeconds :: Double -> Text +formatSeconds seconds = toText (printf "%.2fs" seconds :: String) + +-- | Describe a transfer of @bytes@ bytes taking @seconds@ seconds. +-- +-- >>> transferSummary 12345678 1.5 +-- "11.77 MiB in 1.50s (7.85 MiB/s)" +transferSummary :: Int -> Double -> Text +transferSummary bytes seconds = + toText (bytesfmt "%.2f" bytes) <> " in " <> formatSeconds seconds <> rate + where + rate + -- Below that the rate is mostly measurement noise, and dividing by a very + -- small elapsed time gives an absurd number. + | seconds >= 0.01 = + " (" <> toText (bytesfmt "%.2f" (round (fromIntegral bytes / seconds) :: Int)) <> "/s)" + | otherwise = + "" + -- | Create a per-subprocess stderr pipe that prefixes output with the job name. -- The pipe is fully drained before returning. withStderrPipe :: AppState -> (Handle -> IO a) -> IO a diff --git a/taskrunner.cabal b/taskrunner.cabal index 710f969..0a74fc6 100644 --- a/taskrunner.cabal +++ b/taskrunner.cabal @@ -5,7 +5,7 @@ cabal-version: 2.2 -- see: https://github.com/sol/hpack name: taskrunner -version: 0.18.0.10 +version: 0.18.0.11 description: Please see the README on GitHub at homepage: https://github.com/githubuser/taskrunner#readme bug-reports: https://github.com/githubuser/taskrunner/issues