git-annex/Logs/Transfer.hs

{- git-annex transfer information files and lock files
 -
 - Copyright 2012-2021 Joey Hess <id@joeyh.name>
 -
 - Licensed under the GNU AGPL version 3 or higher.
 -}

{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE CPP #-}

module Logs.Transfer where

import Types.Transfer
import Types.ActionItem
import Annex.Common
import qualified Git
import Utility.Metered
import Utility.Percentage
import Utility.PID
import Annex.LockPool
import Utility.TimeStamp
import Logs.File
import qualified Utility.RawFilePath as R
#ifndef mingw32_HOST_OS
import Annex.Perms
#endif

import Data.Time.Clock
import Data.Time.Clock.POSIX
import Control.Concurrent.STM
import qualified Data.ByteString.Char8 as B8
import qualified System.FilePath.ByteString as P

describeTransfer :: Transfer -> TransferInfo -> String
describeTransfer t info = unwords
	[ show $ transferDirection t
	, show $ transferUUID t
	, decodeBS $ actionItemDesc $ ActionItemAssociatedFile
		(associatedFile info)
		(transferKey t)
	, show $ bytesComplete info
	]

{- Transfers that will accomplish the same task. -}
equivilantTransfer :: Transfer -> Transfer -> Bool
equivilantTransfer t1 t2
	| transferDirection t1 == Download && transferDirection t2 == Download &&
	  transferKeyData t1 == transferKeyData t2 = True
	| otherwise = t1 == t2

percentComplete :: Transfer -> TransferInfo -> Maybe Percentage
percentComplete t info =
	percentage
		<$> keySize (transferKeyData t)
		<*> Just (fromMaybe 0 $ bytesComplete info)

{- Generates a callback that can be called as transfer progresses to update
 - the transfer info file. Also returns the file it'll be updating, 
 - an action that sets up the file with appropriate permissions,
 - which should be run after locking the transfer lock file, but
 - before using the callback, and a TVar that can be used to read
 - the number of bytes processed so far. -}
mkProgressUpdater :: Transfer -> TransferInfo -> Annex (MeterUpdate, RawFilePath, Annex (), TVar (Maybe BytesProcessed))
mkProgressUpdater t info = do
	tfile <- fromRepo $ transferFile t
	let createtfile = void $ tryNonAsync $ writeTransferInfoFile info tfile
	tvar <- liftIO $ newTVarIO Nothing
	loggedtvar <- liftIO $ newTVarIO 0
	return (liftIO . updater (fromRawFilePath tfile) tvar loggedtvar, tfile, createtfile, tvar)
  where
	updater tfile tvar loggedtvar new = do
		old <- atomically $ swapTVar tvar (Just new)
		let oldbytes = maybe 0 fromBytesProcessed old
		let newbytes = fromBytesProcessed new
		when (newbytes - oldbytes >= mindelta) $ do
			let info' = info { bytesComplete = Just newbytes }
			_ <- tryIO $ updateTransferInfoFile info' tfile
			atomically $ writeTVar loggedtvar newbytes

	{- The minimum change in bytesComplete that is worth
	 - updating a transfer info file for is 1% of the total
	 - keySize, rounded down. -}
	mindelta = case keySize (transferKeyData t) of
		Just sz -> sz `div` 100
		Nothing -> 100 * 1024 -- arbitrarily, 100 kb

startTransferInfo :: AssociatedFile -> IO TransferInfo
startTransferInfo afile = TransferInfo
	<$> (Just . utcTimeToPOSIXSeconds <$> getCurrentTime)
#ifndef mingw32_HOST_OS
	<*> pure Nothing -- pid not stored in file, so omitted for speed
#else
	<*> (Just <$> getPID)
#endif
	<*> pure Nothing -- tid ditto
	<*> pure Nothing -- not 0; transfer may be resuming
	<*> pure Nothing
	<*> pure afile
	<*> pure False

{- If a transfer is still running, returns its TransferInfo.
 - 
 - If no transfer is running, attempts to clean up the stale
 - lock and info files. This can happen if a transfer process was
 - interrupted.
 -}
checkTransfer :: Transfer -> Annex (Maybe TransferInfo)
checkTransfer t = debugLocks $ do
	tfile <- fromRepo $ transferFile t
	let lck = transferLockFile tfile
	let cleanstale = do
		void $ tryIO $ R.removeLink tfile
		void $ tryIO $ R.removeLink lck
#ifndef mingw32_HOST_OS
	v <- getLockStatus lck
	case v of
		StatusLockedBy pid -> liftIO $ catchDefaultIO Nothing $
			readTransferInfoFile (Just pid) (fromRawFilePath tfile)
		_ -> do
			-- Take a non-blocking lock while deleting
			-- the stale lock file. Ignore failure
			-- due to permissions problems, races, etc.
			void $ tryIO $ do
				mode <- annexFileMode
				r <- tryLockExclusive (Just mode) lck
				case r of
					Just lockhandle -> liftIO $ do
						cleanstale
						dropLock lockhandle
					_ -> noop
			return Nothing
#else
	v <- liftIO $ lockShared lck
	liftIO $ case v of
		Nothing -> catchDefaultIO Nothing $
			readTransferInfoFile Nothing (fromRawFilePath tfile)
		Just lockhandle -> do
			dropLock lockhandle
			cleanstale
			return Nothing
#endif

{- Gets all currently running transfers. -}
getTransfers :: Annex [(Transfer, TransferInfo)]
getTransfers = getTransfers' [Download, Upload] (const True)

getTransfers' :: [Direction] -> (Key -> Bool) -> Annex [(Transfer, TransferInfo)]
getTransfers' dirs wanted = do
	transfers <- filter (wanted . transferKey)
		<$> mapMaybe parseTransferFile . concat <$> findfiles
	infos <- mapM checkTransfer transfers
	return $ mapMaybe running $ zip transfers infos
  where
	findfiles = liftIO . mapM (dirContentsRecursive . fromRawFilePath)
		=<< mapM (fromRepo . transferDir) dirs
	running (t, Just i) = Just (t, i)
	running (_, Nothing) = Nothing

{- Number of bytes remaining to download from matching downloads that are in
 - progress. -}
sizeOfDownloadsInProgress :: (Key -> Bool) -> Annex Integer
sizeOfDownloadsInProgress wanted = sum . map remaining
	<$> getTransfers' [Download] wanted
  where
	remaining (t, info) =
		case (fromKey keySize (transferKey t), bytesComplete info) of
			(Just sz, Just done) -> sz - done
			(Just sz, Nothing) -> sz
			(Nothing, _) -> 0

{- Gets failed transfers for a given remote UUID. -}
getFailedTransfers :: UUID -> Annex [(Transfer, TransferInfo)]
getFailedTransfers u = catMaybes <$> (liftIO . getpairs =<< concat <$> findfiles)
  where
	getpairs = mapM $ \f -> do
		let mt = parseTransferFile f
		mi <- readTransferInfoFile Nothing f
		return $ case (mt, mi) of
			(Just t, Just i) -> Just (t, i)
			_ -> Nothing
	findfiles = liftIO . mapM (dirContentsRecursive . fromRawFilePath)
		=<< mapM (fromRepo . failedTransferDir u) [Download, Upload]

clearFailedTransfers :: UUID -> Annex [(Transfer, TransferInfo)]
clearFailedTransfers u = do
	failed <- getFailedTransfers u
	mapM_ (removeFailedTransfer . fst) failed
	return failed

removeFailedTransfer :: Transfer -> Annex ()
removeFailedTransfer t = do
	f <- fromRepo $ failedTransferFile t
	liftIO $ void $ tryIO $ R.removeLink f

recordFailedTransfer :: Transfer -> TransferInfo -> Annex ()
recordFailedTransfer t info = do
	failedtfile <- fromRepo $ failedTransferFile t
	writeTransferInfoFile info failedtfile

{- The transfer information file to use for a given Transfer. -}
transferFile :: Transfer -> Git.Repo -> RawFilePath
transferFile (Transfer direction u kd) r =
	transferDir direction r
		P.</> B8.filter (/= '/') (fromUUID u)
		P.</> keyFile (mkKey (const kd))

{- The transfer information file to use to record a failed Transfer -}
failedTransferFile :: Transfer -> Git.Repo -> RawFilePath
failedTransferFile (Transfer direction u kd) r = 
	failedTransferDir u direction r
		P.</> keyFile (mkKey (const kd))

{- The transfer lock file corresponding to a given transfer info file. -}
transferLockFile :: RawFilePath -> RawFilePath
transferLockFile infofile = 
	let (d, f) = P.splitFileName infofile
	in P.combine d ("lck." <> f)

{- Parses a transfer information filename to a Transfer. -}
parseTransferFile :: FilePath -> Maybe Transfer
parseTransferFile file
	| "lck." `isPrefixOf` takeFileName file = Nothing
	| otherwise = case drop (length bits - 3) bits of
		[direction, u, key] -> Transfer
			<$> parseDirection direction
			<*> pure (toUUID u)
			<*> fmap (fromKey id) (fileKey (toRawFilePath key))
		_ -> Nothing
  where
	bits = splitDirectories file

writeTransferInfoFile :: TransferInfo -> RawFilePath -> Annex ()
writeTransferInfoFile info tfile = writeLogFile tfile $ writeTransferInfo info

-- The file keeps whatever permissions it has, so should be used only
-- after it's been created with the right perms by writeTransferInfoFile.
updateTransferInfoFile :: TransferInfo -> FilePath -> IO ()
updateTransferInfoFile info tfile = writeFile tfile $ writeTransferInfo info

{- File format is a header line containing the startedTime and any
 - bytesComplete value. Followed by a newline and the associatedFile.
 -
 - On unix, the transferPid is not included; instead it is obtained
 - by looking at the process that locks the file.
 -
 - On windows, the transferPid is included, as a second line.
 -}
writeTransferInfo :: TransferInfo -> String
writeTransferInfo info = unlines
	[ (maybe "" show $ startedTime info) ++
	  (maybe "" (\b -> ' ' : show b) (bytesComplete info))
#ifdef mingw32_HOST_OS
	, maybe "" show (transferPid info)
#endif
	-- comes last; arbitrary content
	, let AssociatedFile afile = associatedFile info
	  in maybe "" fromRawFilePath afile
	]

readTransferInfoFile :: Maybe PID -> FilePath -> IO (Maybe TransferInfo)
readTransferInfoFile mpid tfile = catchDefaultIO Nothing $
	readTransferInfo mpid <$> readFileStrict tfile

readTransferInfo :: Maybe PID -> String -> Maybe TransferInfo
readTransferInfo mpid s = TransferInfo
	<$> time
#ifdef mingw32_HOST_OS
	<*> pure (if isJust mpid then mpid else mpid')
#else
	<*> pure mpid
#endif
	<*> pure Nothing
	<*> pure Nothing
	<*> bytes
	<*> pure (AssociatedFile (if null filename then Nothing else Just (toRawFilePath filename)))
	<*> pure False
  where
#ifdef mingw32_HOST_OS
	(firstline, otherlines) = separate (== '\n') s
	(secondline, rest) = separate (== '\n') otherlines
	mpid' = readish secondline
#else
	(firstline, rest) = separate (== '\n') s
#endif
	filename
		| end rest == "\n" = beginning rest
		| otherwise = rest
	bits = splitc ' ' firstline
	numbits = length bits
	time = if numbits > 0
		then Just <$> parsePOSIXTime =<< headMaybe bits
		else pure Nothing -- not failure
	bytes = if numbits > 1
		then Just <$> readish =<< headMaybe (drop 1 bits)
		else pure Nothing -- not failure

{- The directory holding transfer information files for a given Direction. -}
transferDir :: Direction -> Git.Repo -> RawFilePath
transferDir direction r = gitAnnexTransferDir r P.</> formatDirection direction

{- The directory holding failed transfer information files for a given
 - Direction and UUID -}
failedTransferDir :: UUID -> Direction -> Git.Repo -> RawFilePath
failedTransferDir u direction r = gitAnnexTransferDir r
	P.</> "failed"
	P.</> formatDirection direction
	P.</> B8.filter (/= '/') (fromUUID u)

prop_read_write_transferinfo :: TransferInfo -> Bool
prop_read_write_transferinfo info
	| isJust (transferRemote info) = True -- remote not stored
	| isJust (transferTid info) = True -- tid not stored
	| otherwise = Just (info { transferPaused = False }) == info'
  where
	info' = readTransferInfo (transferPid info) (writeTransferInfo info)