From e43359910474d5f6922fdfc092baa49911ab974a Mon Sep 17 00:00:00 2001 From: Michal Kozakiewicz Date: Sun, 2 Aug 2026 09:55:30 +0000 Subject: [PATCH 1/3] Unpack the remote cache with several tar processes Restoring a cache bundle is dominated by per-file filesystem work, not by the download. The diagnostic added in the previous commit measured, across seven tasks on CI, 13.6s of downloading against 420.7s of unpacking - 97% of the time. The cost is per entry rather than per byte: a representative bundle is 80 MiB holding 56,845 entries, 77% of them under 4 KiB, and it takes ~23s to unpack, i.e. ~400us each. That is filesystem latency, so it parallelises. ParallelUnpack splits the tar stream across several concurrent `tar -x` processes. It parses only enough of each header to find where the entry ends and forwards the bytes verbatim, so permissions, mtimes, symlinks, hardlinks, long names and sparse files are all still handled by real tar rather than reimplemented. The archive format does not change, so bundles saved by any previous version still restore. Two things cannot be done concurrently and are held back for a final sequential pass: hardlinks, which need their target to exist, and directory entries, whose metadata must be applied after the files inside them. Separately, the workers must never create a directory themselves - two tar processes racing to auto-create the same parent silently loses files - so the splitter creates every directory before dispatching the entry that needs it. Verified against a real production bundle (56,845 entries) and against an archive built to cover the awkward cases: contents, modes, mtimes, symlink targets, hardlink inode sharing and sparse allocation all match a plain `tar -x` exactly. The worker count defaults to the processor count clamped to 4-8 and can be set with TASKRUNNER_UNPACK_WORKERS; 1 restores the previous single-process path. Co-Authored-By: Claude Opus 5 (1M context) --- src/App.hs | 7 + src/ParallelUnpack.hs | 308 ++++++++++++++++++++++++ src/RemoteCache.hs | 16 +- src/Types.hs | 1 + taskrunner.cabal | 1 + test/t/remote-cache-parallel-unpack.out | 74 ++++++ test/t/remote-cache-parallel-unpack.txt | 70 ++++++ 7 files changed, 476 insertions(+), 1 deletion(-) create mode 100644 src/ParallelUnpack.hs create mode 100644 test/t/remote-cache-parallel-unpack.out create mode 100644 test/t/remote-cache-parallel-unpack.txt diff --git a/src/App.hs b/src/App.hs index f6d43b6..185699d 100644 --- a/src/App.hs +++ b/src/App.hs @@ -30,6 +30,7 @@ import SnapshotCliArgs qualified import Data.Text qualified as Text import qualified Data.Text.IO as Text import GHC.IO.Exception (ExitCode(..)) +import GHC.Conc (getNumProcessors) import Crypto.Hash qualified as H import Data.Containers.ListUtils (nubOrdOn) import Prelude (read) @@ -63,6 +64,11 @@ getSettings = do mainBranch <- map toText <$> lookupEnv "TASKRUNNER_MAIN_BRANCH" quietMode <- (==Just "1") <$> lookupEnv "TASKRUNNER_QUIET" githubTokenRefreshThresholdSeconds <- maybe 300 read <$> lookupEnv "TASKRUNNER_GITHUB_TOKEN_REFRESH_THRESHOLD_SECONDS" + -- Unpacking a cache bundle is bound by per-file filesystem latency rather + -- than by CPU, so several tar processes help even on a machine with few + -- cores. 1 disables the parallel path entirely. + defaultUnpackWorkers <- max 4 . min 8 <$> getNumProcessors + unpackWorkers <- maybe defaultUnpackWorkers read <$> lookupEnv "TASKRUNNER_UNPACK_WORKERS" pure Settings { stateDirectory , rootDirectory @@ -81,6 +87,7 @@ getSettings = do , githubTokenRefreshThresholdSeconds , trace = False , traceFiles = False + , unpackWorkers } main :: IO () diff --git a/src/ParallelUnpack.hs b/src/ParallelUnpack.hs new file mode 100644 index 0000000..ae53ceb --- /dev/null +++ b/src/ParallelUnpack.hs @@ -0,0 +1,308 @@ +{-# LANGUAGE BangPatterns #-} +-- | Extracting a cache bundle is dominated by per-file filesystem work, not by +-- the download or by decompression: on CI a 80 MiB bundle holding ~57k mostly +-- tiny files takes ~0.5s to download and ~23s to unpack. That cost is latency +-- per entry, so it parallelises well. +-- +-- This module splits a tar stream across several concurrent @tar -x@ processes. +-- We parse only enough of each header to find where the entry ends; the bytes +-- themselves are forwarded verbatim, so permissions, mtimes, symlinks, long +-- names and every other GNU tar semantic are still handled by real tar. The +-- archive format is untouched, so existing cache bundles work unchanged. +module ParallelUnpack (unpackTarParallel) where + +import Universum + +import Data.Bits ((.&.)) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Char8 as B8 +import Data.Conduit (ConduitT, bracketP) +import qualified Data.Conduit as C +import Data.List ((!!)) +import qualified Data.Set as Set +import Control.Monad.Trans.Resource (MonadResource) +import System.Directory (createDirectoryIfMissing) +import System.Exit (ExitCode (..)) +import System.FilePath (()) +import System.IO (hSetBuffering, BufferMode (BlockBuffering)) +import System.Process (CreateProcess (..), StdStream (..), cleanupProcess, createProcess_, proc, waitForProcess) +import Types +import Utils (bail, formatSeconds, logDebug, timed) + +-- | Everything in a tar archive is a multiple of this. +blockSize :: Int +blockSize = 512 + +-- | A tar header, parsed just far enough to find the next one. The name stays +-- as raw bytes: it is only needed to create parent directories, which is rare +-- compared to the number of entries. +data Header = Header + { name :: BS.ByteString + , size :: Int + , typeflag :: Char + } + +-- | Extract a (decompressed) tar stream using @workerCount@ concurrent @tar -x@ +-- processes, all with cwd @workdir@. +unpackTarParallel :: MonadResource m => AppState -> Handle -> FilePath -> Int -> ConduitT BS.ByteString Void m () +unpackTarParallel appState stderrHandle workdir workerCount = do + liftIO $ logDebug appState $ "Running " <> show workerCount <> " subprocesses: " + <> show (tarCmd:tarArgs) <> " in cwd " <> show workdir + + bracketP (replicateM workerCount startTar) (mapM_ cleanupProcess) \procs -> do + pipes <- forM procs \case + (Just stdinPipe, _, _, _) -> do + -- Entries are mostly small, so the default buffer would turn this into + -- one write syscall per few entries. + liftIO $ hSetBuffering stdinPipe (BlockBuffering (Just pipeBufferSize)) + pure stdinPipe + _ -> liftIO $ error "unable to obtain stdin pipe" + + -- Directories we have already created. Two tar processes racing to + -- auto-create the same parent directory corrupts the tree (files end up + -- missing), so the workers must never have to create one: we create every + -- directory here, single-threaded, before dispatching the entry that needs + -- it. + madeRef <- newIORef Set.empty + -- Directory and hardlink entries, replayed sequentially once the workers + -- are done: a hardlink needs its target to exist, and directory metadata + -- must be applied after the files inside them have been written. Both are + -- header-only, so holding them costs 512 bytes per directory. + deferredRef <- newIORef [] + entriesRef <- newIORef (0 :: Int) + -- Entries arrive depth first, so consecutive ones nearly always share a + -- parent directory; remembering the last one skips most set lookups. + lastDirRef <- newIORef BS.empty + + let + ensureDir dir = do + lastDir <- readIORef lastDirRef + unless (dir == lastDir || BS.null dir || dir == ".") do + made <- readIORef madeRef + unless (Set.member dir made) do + createDirectoryIfMissing True (workdir decodePath dir) + modifyIORef' madeRef (Set.insert dir) + writeIORef lastDirRef dir + + -- @pending@ holds extended headers ('L', 'K', 'x'), which describe the + -- entry that follows and so must reach the same worker as it. + loop !next !pending !longName = do + block <- readExactly blockSize + if + -- An archive ends with zero blocks. Testing the first byte first + -- keeps the full scan off the path every real header takes. + | BS.null block || (BS.head block == 0 && BS.all (== 0) block) -> + pass + | BS.length block < blockSize -> + liftIO $ bail "truncated tar archive" + | otherwise -> do + header <- either (liftIO . bail) pure $ parseHeader block + let entryName = fromMaybe header.name longName + padded = ((header.size + blockSize - 1) `div` blockSize) * blockSize + case header.typeflag of + -- Extended headers describe the entry that follows, so they + -- have to reach the same worker as it. + t | t == 'L' || t == 'K' || t == 'x' || t == 'g' -> do + body <- readExactly padded + let longName' = case t of + 'L' -> Just (BS.takeWhile (/= 0) body) + 'x' -> paxPath body <|> longName + _ -> longName + -- Newest first; reversed again when written out. + loop next (body : block : pending) longName' + t | t == '5' || t == '1' -> do + body <- readExactly padded + when (t == '5') $ liftIO $ ensureDir (dropTrailingSlash entryName) + liftIO do + modifyIORef' deferredRef (BS.concat (reverse (body : block : pending)) :) + modifyIORef' entriesRef (+ 1) + loop next [] Nothing + _ -> do + -- Round-robin: assignment stays balanced without a slow + -- worker attracting more work, which a least-loaded + -- policy would do once its pipe backs up. + let pipe = pipes !! (next `mod` workerCount) + liftIO do + ensureDir (parentDir entryName) + mapM_ (BS.hPut pipe) (reverse (block : pending)) + modifyIORef' entriesRef (+ 1) + copyExactly pipe padded + loop (next + 1) [] Nothing + + (_, splitSeconds) <- timed $ loop (0 :: Int) [] Nothing + + liftIO do + (_, drainSeconds) <- timed do + forM_ pipes \pipe -> BS.hPut pipe endOfArchive >> hClose pipe + forM_ procs \(_, _, _, process) -> checkExit =<< waitForProcess process + + deferred <- reverse <$> readIORef deferredRef + (_, deferredSeconds) <- timed $ unless (null deferred) $ runDeferredPass deferred + + entries <- readIORef entriesRef + logDebug appState $ "Unpacked " <> show entries <> " entries using " + <> show workerCount <> " tar processes - splitting " <> formatSeconds splitSeconds + <> ", draining " <> formatSeconds drainSeconds + <> ", " <> show (length deferred) <> " directories and hardlinks in " + <> formatSeconds deferredSeconds + where + tarCmd = "tar" + tarArgs = ["-x"] + + startTar = createProcess_ "createProcess_" + (proc tarCmd tarArgs) + { std_in = CreatePipe + , std_err = UseHandle stderrHandle + , cwd = Just workdir + } + + -- One last tar for the entries that could not be extracted concurrently. + runDeferredPass blocks = + bracket startTar cleanupProcess \case + (Just stdinPipe, _, _, process) -> do + mapM_ (BS.hPut stdinPipe) blocks + BS.hPut stdinPipe endOfArchive + hClose stdinPipe + checkExit =<< waitForProcess process + _ -> + error "unable to obtain stdin pipe" + + checkExit exitCode = + when (exitCode /= ExitSuccess) do + bail $ "tar unpack command failed with code: " <> show exitCode + +-- | Big enough that a run of small entries becomes one write syscall. +pipeBufferSize :: Int +pipeBufferSize = 256 * 1024 + +-- | tar marks the end of an archive with two zero blocks. Without them tar +-- reports an unexpected EOF and exits nonzero. +endOfArchive :: BS.ByteString +endOfArchive = BS.replicate (2 * blockSize) 0 + +-- | Read exactly @n@ bytes, returning fewer only at end of input. +readExactly :: Monad m => Int -> ConduitT BS.ByteString o m BS.ByteString +readExactly = go [] + where + -- Upstream chunks are far bigger than a header, so the usual case is one + -- slice of the current chunk and no copying at all. + join1 [] chunk = chunk + join1 acc chunk = BS.concat (reverse (chunk : acc)) + + go [] 0 = pure BS.empty + go acc 0 = pure $ BS.concat (reverse acc) + go acc n = + C.await >>= \case + Nothing -> pure $ BS.concat (reverse acc) + Just chunk + | BS.length chunk <= n -> go (chunk : acc) (n - BS.length chunk) + | otherwise -> do + let (wanted, rest) = BS.splitAt n chunk + C.leftover rest + pure $ join1 acc wanted + +-- | Copy exactly @n@ bytes from upstream to a handle, without holding the whole +-- entry in memory - archives may contain individually huge files. +copyExactly :: MonadIO m => Handle -> Int -> ConduitT BS.ByteString o m () +copyExactly pipe = go + where + go 0 = pass + go n = + C.await >>= \case + Nothing -> liftIO $ bail "truncated tar archive" + Just chunk + | BS.length chunk <= n -> do + liftIO $ BS.hPut pipe chunk + go (n - BS.length chunk) + | otherwise -> do + let (wanted, rest) = BS.splitAt n chunk + liftIO $ BS.hPut pipe wanted + C.leftover rest + +parseHeader :: BS.ByteString -> Either String Header +parseHeader block = do + unless (checksumMatches block) $ + Left "tar header checksum mismatch - corrupt archive?" + size <- maybeToRight "invalid size field in tar header" $ parseNumeric (field 124 12) + pure Header + { name = fullName + , size + , typeflag = B8.head (field 156 1) + } + where + field offset len = BS.take len (BS.drop offset block) + nulTerminated = BS.takeWhile (/= 0) + + -- The prefix field only holds part of the path in POSIX ustar archives; in + -- the GNU format those bytes mean something else entirely, and long names + -- arrive as a separate 'L' entry instead. + name_ = nulTerminated (field 0 100) + prefix = nulTerminated (field 345 155) + fullName + | field 257 6 == "ustar\0", not (BS.null prefix) = prefix <> "/" <> name_ + | otherwise = name_ + +-- | Guards against silently desynchronising from the stream: if a size field +-- were misread we would slice the archive at the wrong offset and hand every +-- worker garbage. +-- +-- This runs on every header, so it sums the bytes in place rather than +-- materialising a blanked-out copy of the block. +checksumMatches :: BS.ByteString -> Bool +checksumMatches block = + case parseNumeric checksumField of + Nothing -> False + -- The checksum is defined over the header with its own field read as + -- spaces. Some historic tars summed the bytes as signed, so accept either. + Just expected -> expected == unsignedSum || expected == signedSum + where + checksumField = BS.take 8 (BS.drop 148 block) + blankedOut = 8 * 32 + sumWith f bs = BS.foldl' (\acc byte -> acc + f byte) (0 :: Int) bs + unsigned = fromIntegral :: Word8 -> Int + signed byte = if byte > 127 then fromIntegral byte - 256 else fromIntegral byte + unsignedSum = sumWith unsigned block - sumWith unsigned checksumField + blankedOut + signedSum = sumWith signed block - sumWith signed checksumField + blankedOut + +-- | Header numbers are octal, except that large values use a base-256 escape +-- flagged by the top bit of the first byte. +parseNumeric :: BS.ByteString -> Maybe Int +parseNumeric bs + | BS.null bs = Just 0 + | BS.head bs .&. 0x80 /= 0 = Just $ BS.foldl' (\acc b -> acc * 256 + fromIntegral b) 0 (BS.tail bs) + | BS.null digits = Just 0 + | otherwise = Just $ BS.foldl' (\acc c -> acc * 8 + fromIntegral (c - 0x30)) 0 digits + where + digits = BS.takeWhile (\c -> 0x30 <= c && c <= 0x37) (BS.dropWhile (== 0x20) bs) + +-- | The @path@ record of a PAX extended header, which overrides the name in the +-- header that follows. Records look like @" =\n"@, where +-- @len@ counts the whole record. +paxPath :: BS.ByteString -> Maybe BS.ByteString +paxPath body = find (const True) [ value | Just value <- map (BS.stripPrefix "path=") (records body) ] + where + records bs = case B8.readInt bs of + Just (len, afterDigits) + | len > 0, len <= BS.length bs, contentLen >= 0 -> + BS.take contentLen (BS.drop 1 afterDigits) : records (BS.drop len bs) + where + -- Drop the length digits, the separating space and the trailing newline. + contentLen = len - (BS.length bs - BS.length afterDigits) - 2 + _ -> + [] + +-- | Names are only used to create parent directories; tar itself receives the +-- raw bytes, so a name we cannot decode does not corrupt the extracted file. +decodePath :: BS.ByteString -> FilePath +decodePath = toString . decodeUtf8 @Text + +-- | Everything up to the last @/@, i.e. the directory the entry lives in. +-- Empty for a top-level entry. Written the same way as a directory entry's own +-- name so both hit the same cache. +parentDir :: BS.ByteString -> BS.ByteString +parentDir = dropTrailingSlash . fst . B8.breakEnd (== '/') + +dropTrailingSlash :: BS.ByteString -> BS.ByteString +dropTrailingSlash path + | not (BS.null path), B8.last path == '/' = BS.init path + | otherwise = path diff --git a/src/RemoteCache.hs b/src/RemoteCache.hs index 1325031..bd98caf 100644 --- a/src/RemoteCache.hs +++ b/src/RemoteCache.hs @@ -36,6 +36,7 @@ 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) +import ParallelUnpack (unpackTarParallel) packTar :: MonadResource m => AppState -> Handle -> FilePath -> [FilePath] -> ConduitT () BS.ByteString m () @@ -58,6 +59,19 @@ packTar appState stderrHandle workdir files = do _ -> error "unable to obtain stdout pipe" +-- | Unpack a compressed archive into @workdir@. +-- +-- With more than one unpack worker the stream is decompressed here and split +-- across concurrent tar processes, which is much faster for the many-small-files +-- trees that caches usually hold. The archive format is the same either way, so +-- this reads bundles saved by any version. +unpack :: MonadResource m => AppState -> Handle -> FilePath -> ConduitT BS.ByteString Void m () +unpack appState stderrHandle workdir + | appState.settings.unpackWorkers > 1 = + Zstd.decompress .| unpackTarParallel appState stderrHandle workdir appState.settings.unpackWorkers + | otherwise = + unpackTar appState stderrHandle workdir + unpackTar :: MonadResource m => AppState -> Handle -> FilePath -> ConduitT BS.ByteString Void m () unpackTar appState stderrHandle workdir = do let cmd = "tar" @@ -257,7 +271,7 @@ restoreCache appState settings cacheRoot archiveName logMode = do liftIO $ logInfo appState $ "Found remote cache " <> archiveName <> ", restoring" response.body.body .| measureTransfer statsRef - .| unpackTar appState stderrHandle cacheRoot + .| unpack appState stderrHandle cacheRoot stats <- readIORef statsRef -- The size is that of the compressed archive, and the rate covers the whole diff --git a/src/Types.hs b/src/Types.hs index 84151cf..9c3870f 100644 --- a/src/Types.hs +++ b/src/Types.hs @@ -24,6 +24,7 @@ data Settings = Settings , githubTokenRefreshThresholdSeconds :: Int , trace :: Bool , traceFiles :: Bool + , unpackWorkers :: Int } deriving (Show) type JobName = String diff --git a/taskrunner.cabal b/taskrunner.cabal index 0a74fc6..e1398bc 100644 --- a/taskrunner.cabal +++ b/taskrunner.cabal @@ -29,6 +29,7 @@ library CliArgs CommitStatus Control.Monad.EarlyReturn + ParallelUnpack RemoteCache SnapshotCliArgs Trace diff --git a/test/t/remote-cache-parallel-unpack.out b/test/t/remote-cache-parallel-unpack.out new file mode 100644 index 0000000..9e06aff --- /dev/null +++ b/test/t/remote-cache-parallel-unpack.out @@ -0,0 +1,74 @@ +-- output: +[mytask] info | Inputs changed, running task +[mytask] stdout | Expensive computation +[mytask] info | success +--- saved from a: +d 500 981173106.0000000000 nested/deep +d 755 981173106.0000000000 +d 755 981173106.0000000000 dir with spaces +d 755 981173106.0000000000 empty-dir +d 755 981173106.0000000000 nested +f 644 981173106.0000000000 dir with spaces/file two.txt +f 644 981173106.0000000000 nested/deep/one.txt +f 644 981173106.0000000000 nested/f1.txt +f 644 981173106.0000000000 nested/f2.txt +f 644 981173106.0000000000 nested/f3.txt +f 644 981173106.0000000000 nested/f4.txt +f 644 981173106.0000000000 nested/f5.txt +l 777 981173106.0000000000 link + symlink link -> nested/deep/one.txt + out/dir with spaces/file two.txt = two + out/nested/deep/one.txt = one + out/nested/f1.txt = contents 1 + out/nested/f2.txt = contents 2 + out/nested/f3.txt = contents 3 + out/nested/f4.txt = contents 4 + out/nested/f5.txt = contents 5 +*** restoring in b with 4 unpack workers +[mytask] info | Found remote cache mytask-6b36308659163b577fd1c832107dc46ca3aa659b.tar.zst, restoring +[mytask] info | Restored from remote cache +d 500 981173106.0000000000 nested/deep +d 755 981173106.0000000000 +d 755 981173106.0000000000 dir with spaces +d 755 981173106.0000000000 empty-dir +d 755 981173106.0000000000 nested +f 644 981173106.0000000000 dir with spaces/file two.txt +f 644 981173106.0000000000 nested/deep/one.txt +f 644 981173106.0000000000 nested/f1.txt +f 644 981173106.0000000000 nested/f2.txt +f 644 981173106.0000000000 nested/f3.txt +f 644 981173106.0000000000 nested/f4.txt +f 644 981173106.0000000000 nested/f5.txt +l 777 981173106.0000000000 link + symlink link -> nested/deep/one.txt + out/dir with spaces/file two.txt = two + out/nested/deep/one.txt = one + out/nested/f1.txt = contents 1 + out/nested/f2.txt = contents 2 + out/nested/f3.txt = contents 3 + out/nested/f4.txt = contents 4 + out/nested/f5.txt = contents 5 +*** restoring in c with 1 unpack worker, i.e. the single-process path +[mytask] info | Found remote cache mytask-6b36308659163b577fd1c832107dc46ca3aa659b.tar.zst, restoring +[mytask] info | Restored from remote cache +d 500 981173106.0000000000 nested/deep +d 755 981173106.0000000000 +d 755 981173106.0000000000 dir with spaces +d 755 981173106.0000000000 empty-dir +d 755 981173106.0000000000 nested +f 644 981173106.0000000000 dir with spaces/file two.txt +f 644 981173106.0000000000 nested/deep/one.txt +f 644 981173106.0000000000 nested/f1.txt +f 644 981173106.0000000000 nested/f2.txt +f 644 981173106.0000000000 nested/f3.txt +f 644 981173106.0000000000 nested/f4.txt +f 644 981173106.0000000000 nested/f5.txt +l 777 981173106.0000000000 link + symlink link -> nested/deep/one.txt + out/dir with spaces/file two.txt = two + out/nested/deep/one.txt = one + out/nested/f1.txt = contents 1 + out/nested/f2.txt = contents 2 + out/nested/f3.txt = contents 3 + out/nested/f4.txt = contents 4 + out/nested/f5.txt = contents 5 diff --git a/test/t/remote-cache-parallel-unpack.txt b/test/t/remote-cache-parallel-unpack.txt new file mode 100644 index 0000000..b522979 --- /dev/null +++ b/test/t/remote-cache-parallel-unpack.txt @@ -0,0 +1,70 @@ +# no toplevel +# s3 + +# The remote cache is unpacked by several tar processes at once (see +# ParallelUnpack). Check that a tree containing the awkward cases - nested and +# empty directories, a symlink, spaces in names, a directory that is not +# writable - comes back exactly as it was saved, and identically to what the +# single-process path produces. + +export TASKRUNNER_SAVE_REMOTE_CACHE=1 + +mkdir a + +( + cd a + echo foo > input.txt + git init -q + git add input.txt + git commit -qm "Add input.txt" +) + +cp -r a b +cp -r a c + +go() { + # Note: we want separate workdir for separate repos + TASKRUNNER_STATE_DIRECTORY="$(pwd)" taskrunner -n mytask bash -e -c ' + snapshot input.txt --outputs out + echo "Expensive computation" + mkdir -p out/nested/deep "out/dir with spaces" out/empty-dir + echo one > out/nested/deep/one.txt + echo two > "out/dir with spaces/file two.txt" + for i in 1 2 3 4 5; do echo "contents $i" > "out/nested/f$i.txt"; done + ln -s nested/deep/one.txt out/link + # Fixed modes and timestamps so the listing below does not depend on the + # umask or the clock. A read-only directory also has to be restorable. + chmod -R u=rwX,go=rX out + chmod 500 out/nested/deep + find out -exec touch -h -d "2001-02-03T04:05:06Z" {} + + ' +} + +describe() { + find out -printf "%y %m %T@ %P\n" | LC_ALL=C sort + find out -type l -printf " symlink %P -> %l\n" + find out -type f | LC_ALL=C sort | while read -r f; do printf " %s = " "$f"; cat "$f"; done +} + +( + cd a + go + echo "--- saved from a:" + describe +) + +( + echo "*** restoring in b with 4 unpack workers" + cd b + export TASKRUNNER_SAVE_REMOTE_CACHE=0 TASKRUNNER_UNPACK_WORKERS=4 + go + describe +) + +( + echo "*** restoring in c with 1 unpack worker, i.e. the single-process path" + cd c + export TASKRUNNER_SAVE_REMOTE_CACHE=0 TASKRUNNER_UNPACK_WORKERS=1 + go + describe +) From 5a2b18f2453f62df4f633629a72ef2faad9a0337 Mon Sep 17 00:00:00 2001 From: Michal Kozakiewicz Date: Sun, 2 Aug 2026 11:05:06 +0000 Subject: [PATCH 2/3] v0.18.0.12 --- package.yaml | 2 +- taskrunner.cabal | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/package.yaml b/package.yaml index 16e2d82..a3f9a87 100644 --- a/package.yaml +++ b/package.yaml @@ -1,5 +1,5 @@ name: taskrunner -version: 0.18.0.11 +version: 0.18.0.12 github: "githubuser/taskrunner" license: BSD-3-Clause author: "Author name here" diff --git a/taskrunner.cabal b/taskrunner.cabal index e1398bc..ff8c1ff 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.11 +version: 0.18.0.12 description: Please see the README on GitHub at homepage: https://github.com/githubuser/taskrunner#readme bug-reports: https://github.com/githubuser/taskrunner/issues From b2b3d395d56f4dbb18f6b4dbd37431fbb4fed016 Mon Sep 17 00:00:00 2001 From: Michal Kozakiewicz Date: Sun, 2 Aug 2026 17:47:52 +0000 Subject: [PATCH 3/3] Attribute restore three ways, and account for the tar tail Follows the review on #20. Decompressing in-process for the parallel path means a tap can go between zstd and tar, so restore now separates network from decompression from the file writes - which #20 could not do, since there zstd runs inside tar. As on the save path the taps nest rather than partition, so decompression is the difference between them. Two corrections to the previous numbers: - Dropped the feed loop's own timing. It blocks whenever a worker's pipe is full, so its wall clock conflated this thread's work with waiting for tar, and a large value read as "the splitter is slow" when it actually meant "the workers are the bottleneck". The tap either side of zstd attributes that correctly. - Report how long tar keeps extracting after the last byte is handed over. That tail is outside the pipeline's accounting, so without it the figures visibly failed to add up: a 700 MB single-file archive reported 0.00s of stalling against 0.63s elapsed, because all the work happened after feeding finished. All four combinations now sum to the elapsed time: 20k small files, 8 workers download 0.00s decompress 0.01s unpack 0.61s (0.63s) 20k small files, 1 worker download 0.00s decompress+unpack 0.42s tar tail 0.49s (0.91s) one 700 MB file, 8 workers download 0.00s decompress 0.04s unpack 0.57s (0.62s) one 700 MB file, 1 worker download 0.00s decompress+unpack 0.00s tar tail 0.62s (0.63s) Co-Authored-By: Claude Opus 5 (1M context) --- src/ParallelUnpack.hs | 13 ++++++---- src/RemoteCache.hs | 56 ++++++++++++++++++++++++++++++++----------- 2 files changed, 50 insertions(+), 19 deletions(-) diff --git a/src/ParallelUnpack.hs b/src/ParallelUnpack.hs index ae53ceb..9b39132 100644 --- a/src/ParallelUnpack.hs +++ b/src/ParallelUnpack.hs @@ -129,7 +129,7 @@ unpackTarParallel appState stderrHandle workdir workerCount = do copyExactly pipe padded loop (next + 1) [] Nothing - (_, splitSeconds) <- timed $ loop (0 :: Int) [] Nothing + loop (0 :: Int) [] Nothing liftIO do (_, drainSeconds) <- timed do @@ -140,11 +140,14 @@ unpackTarParallel appState stderrHandle workdir workerCount = do (_, deferredSeconds) <- timed $ unless (null deferred) $ runDeferredPass deferred entries <- readIORef entriesRef + -- No figure for the feed loop itself: it blocks whenever a worker's pipe + -- is full, so its wall clock conflated this thread's work with waiting for + -- tar. 'restoreCache' attributes that properly, via a tap either side of + -- zstd. logDebug appState $ "Unpacked " <> show entries <> " entries using " - <> show workerCount <> " tar processes - splitting " <> formatSeconds splitSeconds - <> ", draining " <> formatSeconds drainSeconds - <> ", " <> show (length deferred) <> " directories and hardlinks in " - <> formatSeconds deferredSeconds + <> show workerCount <> " tar processes (" <> show (length deferred) + <> " directories and hardlinks in " <> formatSeconds deferredSeconds + <> ", workers drained in " <> formatSeconds drainSeconds <> ")" where tarCmd = "tar" tarArgs = ["-x"] diff --git a/src/RemoteCache.hs b/src/RemoteCache.hs index 767527f..b396e4f 100644 --- a/src/RemoteCache.hs +++ b/src/RemoteCache.hs @@ -65,10 +65,20 @@ packTar appState stderrHandle workdir files = do -- across concurrent tar processes, which is much faster for the many-small-files -- trees that caches usually hold. The archive format is the same either way, so -- this reads bundles saved by any version. -unpack :: MonadResource m => AppState -> Handle -> FilePath -> ConduitT BS.ByteString Void m () -unpack appState stderrHandle workdir +-- +-- Decompressing here also means a tap can go between zstd and tar, which is what +-- lets 'restoreCache' tell decompression apart from the file writes. The +-- single-process path cannot: there zstd runs inside tar. +unpack + :: MonadResource m + => AppState -> Handle -> FilePath + -> IORef TransferStats -- ^ Tap on the decompressed stream, parallel path only + -> ConduitT BS.ByteString Void m () +unpack appState stderrHandle workdir decompressedStatsRef | appState.settings.unpackWorkers > 1 = - Zstd.decompress .| unpackTarParallel appState stderrHandle workdir appState.settings.unpackWorkers + Zstd.decompress + .| measureTransfer decompressedStatsRef + .| unpackTarParallel appState stderrHandle workdir appState.settings.unpackWorkers | otherwise = unpackTar appState stderrHandle workdir @@ -86,10 +96,16 @@ unpackTar appState stderrHandle workdir = do ) cleanupProcess \case (Just stdinPipe, _, _, process) -> do sinkHandle stdinPipe - hClose stdinPipe - exitCode <- liftIO $ waitForProcess process - when (exitCode /= ExitSuccess) do - liftIO $ bail $ "tar unpack command failed with code: " <> show exitCode + -- tar is still extracting after we hand over the last byte, and that + -- tail is outside the pipeline's own accounting, so report it too. + -- Otherwise the figures visibly fail to add up to the elapsed time. + (_, drainSeconds) <- timed do + hClose stdinPipe + exitCode <- liftIO $ waitForProcess process + when (exitCode /= ExitSuccess) do + liftIO $ bail $ "tar unpack command failed with code: " <> show exitCode + liftIO $ logDebug appState $ "tar drained in " <> formatSeconds drainSeconds + <> " after the last byte" _ -> error "unable to obtain stdin pipe" @@ -275,6 +291,7 @@ restoreCache appState settings cacheRoot archiveName logMode = do handling _NoSuchKey onNoSuchKey $ withStderrPipe appState \stderrHandle -> do statsRef <- newIORef emptyTransferStats + decompressedStatsRef <- newIORef emptyTransferStats (_, elapsed) <- timed $ runConduitRes do response <- AWS.send env $ newGetObject (BucketName bucket) (ObjectKey objectKey) @@ -282,17 +299,28 @@ restoreCache appState settings cacheRoot archiveName logMode = do liftIO $ logInfo appState $ "Found remote cache " <> archiveName <> ", restoring" response.body.body .| measureTransfer statsRef - .| unpack appState stderrHandle cacheRoot + .| unpack appState stderrHandle cacheRoot decompressedStatsRef 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 + decompressedStats <- readIORef decompressedStatsRef + -- The size is that of the compressed archive. As on the save path the taps + -- nest rather than partition, so decompression is the difference between + -- them; the figures then add up to the elapsed time. Note the download side + -- excludes connection setup, which happened above in AWS.send, and see -- 'measureTransfer' for why it is a lower bound. + let downloadSeconds = stats.producingSeconds + unpackAttribution + -- The single-process path has no tap between zstd and tar, so the two + -- cannot be told apart there. + | appState.settings.unpackWorkers > 1 = + ", on decompression " + <> formatSeconds (max 0 (decompressedStats.producingSeconds - downloadSeconds)) + <> ", on unpacking " <> formatSeconds decompressedStats.consumingSeconds + | otherwise = + ", on decompression and unpacking " <> formatSeconds stats.consumingSeconds logDebug appState $ "Downloaded and unpacked " <> transferSummary stats.bytes elapsed - <> " - blocked on download " <> formatSeconds stats.producingSeconds - <> ", on decompression and unpacking " <> formatSeconds stats.consumingSeconds + <> " - blocked on download " <> formatSeconds downloadSeconds + <> unpackAttribution pure True