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/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..9b39132 --- /dev/null +++ b/src/ParallelUnpack.hs @@ -0,0 +1,311 @@ +{-# 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 + + 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 + -- 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 (" <> show (length deferred) + <> " directories and hardlinks in " <> formatSeconds deferredSeconds + <> ", workers drained in " <> formatSeconds drainSeconds <> ")" + 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 bf5c16d..b396e4f 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,29 @@ 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. +-- +-- 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 + .| measureTransfer decompressedStatsRef + .| 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" @@ -72,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" @@ -261,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) @@ -268,17 +299,28 @@ 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 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 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..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 @@ -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 +)