suggest when user may want annex.stalldetection
When annex.stalldetection is not enabled, and a likely stall is detected, display a suggestion to enable it. Note that the progress meter display is not taken down when displaying the message, so it will display like this: 0% 8 B 0 B/s Transfer seems to have stalled. To handle stalling transfers, configure annex.stalldetection 0% 10 B 0 B/s Although of course if it's really stalled, it will never update again after the message. Taking down the progress meter and starting a new one doesn't seem too necessary given how unusual this is, also this does help show the state it was at when it stalled. Use of uninterruptibleCancel here is ok, the thread it's canceling only does STM transactions and sleeps. The annex thread that gets forked off is separate to avoid it being canceled, so that it can be joined back at the end. A module cycle required moving from dupState the precaching of the remote list. Doing it at startConcurrency should cover all the cases where the remote list is used in concurrent actions. This commit was sponsored by Kevin Mueller on Patreon.
This commit is contained in:
Joey Hess
parent
7db4e62a90
commit
dd39e9e255
18 changed files with 160 additions and 101 deletions
|
|
@ -19,7 +19,6 @@ import Types.Concurrency
|
|||
import Types.CatFileHandles
|
||||
import Annex.CheckAttr
|
||||
import Annex.CheckIgnore
|
||||
import Remote.List
|
||||
|
||||
import qualified Data.Map as M
|
||||
|
||||
|
|
@ -72,11 +71,6 @@ forkState a = do
|
|||
-}
|
||||
dupState :: Annex AnnexState
|
||||
dupState = do
|
||||
-- Make sure that some expensive actions have been done before
|
||||
-- starting threads. This way the state has them already run,
|
||||
-- and each thread won't try to do them.
|
||||
_ <- remoteList
|
||||
|
||||
st <- Annex.getState id
|
||||
-- Make sure that concurrency is enabled, if it was not already,
|
||||
-- so the concurrency-safe resource pools are set up.
|
||||
|
|
|
|||
|
|
@ -504,7 +504,7 @@ importKeys remote importtreeconfig importcontent thirdpartypopulated importablec
|
|||
return (Just (k', ok))
|
||||
checkDiskSpaceToGet k Nothing $
|
||||
notifyTransfer Download af $
|
||||
download' (Remote.uuid remote) k af stdRetry $ \p' ->
|
||||
download' (Remote.uuid remote) k af Nothing stdRetry $ \p' ->
|
||||
withTmp k $ downloader p'
|
||||
|
||||
-- The file is small, so is added to git, so while importing
|
||||
|
|
@ -558,7 +558,7 @@ importKeys remote importtreeconfig importcontent thirdpartypopulated importablec
|
|||
return Nothing
|
||||
checkDiskSpaceToGet tmpkey Nothing $
|
||||
notifyTransfer Download af $
|
||||
download' (Remote.uuid remote) tmpkey af stdRetry $ \p ->
|
||||
download' (Remote.uuid remote) tmpkey af Nothing stdRetry $ \p ->
|
||||
withTmp tmpkey $ \tmpfile ->
|
||||
metered (Just p) tmpkey $
|
||||
const (rundownload tmpfile)
|
||||
|
|
|
|||
|
|
@ -15,9 +15,13 @@ import Utility.DataUnits
|
|||
import Utility.ThreadScheduler
|
||||
|
||||
import Control.Concurrent.STM
|
||||
import Control.Monad.IO.Class (MonadIO)
|
||||
|
||||
detectStalls :: Maybe StallDetection -> TVar (Maybe BytesProcessed) -> IO () -> IO ()
|
||||
{- This may be safely canceled (with eg uninterruptibleCancel),
|
||||
- as long as the passed action can be safely canceled. -}
|
||||
detectStalls :: (Monad m, MonadIO m) => Maybe StallDetection -> TVar (Maybe BytesProcessed) -> m () -> m ()
|
||||
detectStalls Nothing _ _ = noop
|
||||
detectStalls (Just StallDetectionDisabled) _ _ = noop
|
||||
detectStalls (Just (StallDetection minsz duration)) metervar onstall =
|
||||
detectStalls' minsz duration metervar onstall Nothing
|
||||
detectStalls (Just ProbeStallDetection) metervar onstall = do
|
||||
|
|
@ -30,7 +34,7 @@ detectStalls (Just ProbeStallDetection) metervar onstall = do
|
|||
ontimelyadvance v $ \v' -> ontimelyadvance v' $
|
||||
detectStalls' 1 duration metervar onstall
|
||||
where
|
||||
getval = atomically $ fmap fromBytesProcessed
|
||||
getval = liftIO $ atomically $ fmap fromBytesProcessed
|
||||
<$> readTVar metervar
|
||||
|
||||
duration = Duration 60
|
||||
|
|
@ -38,29 +42,30 @@ detectStalls (Just ProbeStallDetection) metervar onstall = do
|
|||
delay = Seconds (fromIntegral (durationSeconds duration) `div` 2)
|
||||
|
||||
waitforfirstupdate startval = do
|
||||
threadDelaySeconds delay
|
||||
liftIO $ threadDelaySeconds delay
|
||||
v <- getval
|
||||
if v > startval
|
||||
then return v
|
||||
else waitforfirstupdate startval
|
||||
|
||||
ontimelyadvance v cont = do
|
||||
threadDelaySeconds delay
|
||||
liftIO $ threadDelaySeconds delay
|
||||
v' <- getval
|
||||
when (v' > v) $
|
||||
cont v'
|
||||
|
||||
detectStalls'
|
||||
:: ByteSize
|
||||
:: (Monad m, MonadIO m)
|
||||
=> ByteSize
|
||||
-> Duration
|
||||
-> TVar (Maybe BytesProcessed)
|
||||
-> IO ()
|
||||
-> m ()
|
||||
-> Maybe ByteSize
|
||||
-> IO ()
|
||||
-> m ()
|
||||
detectStalls' minsz duration metervar onstall st = do
|
||||
threadDelaySeconds delay
|
||||
liftIO $ threadDelaySeconds delay
|
||||
-- Get whatever progress value was reported most recently, if any.
|
||||
v <- atomically $ fmap fromBytesProcessed
|
||||
v <- liftIO $ atomically $ fmap fromBytesProcessed
|
||||
<$> readTVar metervar
|
||||
let cont = detectStalls' minsz duration metervar onstall v
|
||||
case (st, v) of
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
{- git-annex transfers
|
||||
-
|
||||
- Copyright 2012-2020 Joey Hess <id@joeyh.name>
|
||||
- Copyright 2012-2021 Joey Hess <id@joeyh.name>
|
||||
-
|
||||
- Licensed under the GNU AGPL version 3 or higher.
|
||||
-}
|
||||
|
|
@ -37,50 +37,56 @@ import Annex.LockPool
|
|||
import Types.Key
|
||||
import qualified Types.Remote as Remote
|
||||
import Types.Concurrency
|
||||
import Annex.Concurrent.Utility
|
||||
import Annex.Concurrent
|
||||
import Types.WorkerPool
|
||||
import Annex.WorkerPool
|
||||
import Annex.TransferrerPool
|
||||
import Annex.StallDetection
|
||||
import Backend (isCryptographicallySecure)
|
||||
import Types.StallDetection
|
||||
import qualified Utility.RawFilePath as R
|
||||
|
||||
import Control.Concurrent
|
||||
import Control.Concurrent.Async
|
||||
import Control.Concurrent.STM hiding (retry)
|
||||
import qualified Data.Map.Strict as M
|
||||
import qualified System.FilePath.ByteString as P
|
||||
import Data.Ord
|
||||
|
||||
-- Upload, supporting stall detection.
|
||||
-- Upload, supporting canceling detected stalls.
|
||||
upload :: Remote -> Key -> AssociatedFile -> RetryDecider -> NotifyWitness -> Annex Bool
|
||||
upload r key f d witness = stallDetection r >>= \case
|
||||
Nothing -> upload' (Remote.uuid r) key f d go witness
|
||||
Nothing -> go (Just ProbeStallDetection)
|
||||
Just StallDetectionDisabled -> go Nothing
|
||||
Just sd -> runTransferrer sd r key f d Upload witness
|
||||
where
|
||||
go = action . Remote.storeKey r key f
|
||||
go sd = upload' (Remote.uuid r) key f sd d (action . Remote.storeKey r key f) witness
|
||||
|
||||
-- Upload, not supporting stall detection.
|
||||
upload' :: Observable v => UUID -> Key -> AssociatedFile -> RetryDecider -> (MeterUpdate -> Annex v) -> NotifyWitness -> Annex v
|
||||
upload' u key f d a _witness = guardHaveUUID u $
|
||||
runTransfer (Transfer Upload u (fromKey id key)) f d a
|
||||
-- Upload, not supporting canceling detected stalls
|
||||
upload' :: Observable v => UUID -> Key -> AssociatedFile -> Maybe StallDetection -> RetryDecider -> (MeterUpdate -> Annex v) -> NotifyWitness -> Annex v
|
||||
upload' u key f sd d a _witness = guardHaveUUID u $
|
||||
runTransfer (Transfer Upload u (fromKey id key)) f sd d a
|
||||
|
||||
alwaysUpload :: Observable v => UUID -> Key -> AssociatedFile -> RetryDecider -> (MeterUpdate -> Annex v) -> NotifyWitness -> Annex v
|
||||
alwaysUpload u key f d a _witness = guardHaveUUID u $
|
||||
alwaysRunTransfer (Transfer Upload u (fromKey id key)) f d a
|
||||
alwaysUpload :: Observable v => UUID -> Key -> AssociatedFile -> Maybe StallDetection -> RetryDecider -> (MeterUpdate -> Annex v) -> NotifyWitness -> Annex v
|
||||
alwaysUpload u key f sd d a _witness = guardHaveUUID u $
|
||||
alwaysRunTransfer (Transfer Upload u (fromKey id key)) f sd d a
|
||||
|
||||
-- Download, supporting stall detection.
|
||||
-- Download, supporting canceling detected stalls.
|
||||
download :: Remote -> Key -> AssociatedFile -> RetryDecider -> NotifyWitness -> Annex Bool
|
||||
download r key f d witness = logStatusAfter key $ stallDetection r >>= \case
|
||||
Nothing -> getViaTmp (Remote.retrievalSecurityPolicy r) (RemoteVerify r) key f $ \dest ->
|
||||
download' (Remote.uuid r) key f d (go dest) witness
|
||||
Nothing -> go (Just ProbeStallDetection)
|
||||
Just StallDetectionDisabled -> go Nothing
|
||||
Just sd -> runTransferrer sd r key f d Download witness
|
||||
where
|
||||
go dest p = verifiedAction $
|
||||
go sd = getViaTmp (Remote.retrievalSecurityPolicy r) (RemoteVerify r) key f $ \dest ->
|
||||
download' (Remote.uuid r) key f sd d (go' dest) witness
|
||||
go' dest p = verifiedAction $
|
||||
Remote.retrieveKeyFile r key f (fromRawFilePath dest) p
|
||||
|
||||
-- Download, not supporting stall detection.
|
||||
download' :: Observable v => UUID -> Key -> AssociatedFile -> RetryDecider -> (MeterUpdate -> Annex v) -> NotifyWitness -> Annex v
|
||||
download' u key f d a _witness = guardHaveUUID u $
|
||||
runTransfer (Transfer Download u (fromKey id key)) f d a
|
||||
-- Download, not supporting canceling detected stalls.
|
||||
download' :: Observable v => UUID -> Key -> AssociatedFile -> Maybe StallDetection -> RetryDecider -> (MeterUpdate -> Annex v) -> NotifyWitness -> Annex v
|
||||
download' u key f sd d a _witness = guardHaveUUID u $
|
||||
runTransfer (Transfer Download u (fromKey id key)) f sd d a
|
||||
|
||||
guardHaveUUID :: Observable v => UUID -> Annex v -> Annex v
|
||||
guardHaveUUID u a
|
||||
|
|
@ -98,34 +104,44 @@ guardHaveUUID u a
|
|||
-
|
||||
- An upload can be run from a read-only filesystem, and in this case
|
||||
- no transfer information or lock file is used.
|
||||
-
|
||||
- Cannot cancel stalls, but when a likely stall is detected,
|
||||
- suggests to the user that they enable stall detection handling.
|
||||
-}
|
||||
runTransfer :: Observable v => Transfer -> AssociatedFile -> RetryDecider -> (MeterUpdate -> Annex v) -> Annex v
|
||||
runTransfer :: Observable v => Transfer -> AssociatedFile -> Maybe StallDetection -> RetryDecider -> (MeterUpdate -> Annex v) -> Annex v
|
||||
runTransfer = runTransfer' False
|
||||
|
||||
{- Like runTransfer, but ignores any existing transfer lock file for the
|
||||
- transfer, allowing re-running a transfer that is already in progress.
|
||||
-}
|
||||
alwaysRunTransfer :: Observable v => Transfer -> AssociatedFile -> RetryDecider -> (MeterUpdate -> Annex v) -> Annex v
|
||||
alwaysRunTransfer :: Observable v => Transfer -> AssociatedFile -> Maybe StallDetection -> RetryDecider -> (MeterUpdate -> Annex v) -> Annex v
|
||||
alwaysRunTransfer = runTransfer' True
|
||||
|
||||
runTransfer' :: Observable v => Bool -> Transfer -> AssociatedFile -> RetryDecider -> (MeterUpdate -> Annex v) -> Annex v
|
||||
runTransfer' ignorelock t afile retrydecider transferaction = enteringStage TransferStage $ debugLocks $ preCheckSecureHashes (transferKey t) $ do
|
||||
info <- liftIO $ startTransferInfo afile
|
||||
(meter, tfile, createtfile, metervar) <- mkProgressUpdater t info
|
||||
mode <- annexFileMode
|
||||
(lck, inprogress) <- prep tfile createtfile mode
|
||||
if inprogress && not ignorelock
|
||||
then do
|
||||
showNote "transfer already in progress, or unable to take transfer lock"
|
||||
return observeFailure
|
||||
else do
|
||||
v <- retry 0 info metervar (transferaction meter)
|
||||
liftIO $ cleanup tfile lck
|
||||
if observeBool v
|
||||
then removeFailedTransfer t
|
||||
else recordFailedTransfer t info
|
||||
return v
|
||||
runTransfer' :: Observable v => Bool -> Transfer -> AssociatedFile -> Maybe StallDetection -> RetryDecider -> (MeterUpdate -> Annex v) -> Annex v
|
||||
runTransfer' ignorelock t afile stalldetection retrydecider transferaction =
|
||||
enteringStage TransferStage $
|
||||
debugLocks $
|
||||
preCheckSecureHashes (transferKey t) go
|
||||
where
|
||||
go = do
|
||||
info <- liftIO $ startTransferInfo afile
|
||||
(meter, tfile, createtfile, metervar) <- mkProgressUpdater t info
|
||||
mode <- annexFileMode
|
||||
(lck, inprogress) <- prep tfile createtfile mode
|
||||
if inprogress && not ignorelock
|
||||
then do
|
||||
showNote "transfer already in progress, or unable to take transfer lock"
|
||||
return observeFailure
|
||||
else do
|
||||
v <- retry 0 info metervar $
|
||||
detectStallsAndSuggestConfig stalldetection metervar $
|
||||
transferaction meter
|
||||
liftIO $ cleanup tfile lck
|
||||
if observeBool v
|
||||
then removeFailedTransfer t
|
||||
else recordFailedTransfer t info
|
||||
return v
|
||||
|
||||
prep :: RawFilePath -> Annex () -> FileMode -> Annex (Maybe LockHandle, Bool)
|
||||
#ifndef mingw32_HOST_OS
|
||||
prep tfile createtfile mode = catchPermissionDenied (const prepfailed) $ do
|
||||
|
|
@ -191,11 +207,32 @@ runTransfer' ignorelock t afile retrydecider transferaction = enteringStage Tran
|
|||
|
||||
getbytescomplete metervar
|
||||
| transferDirection t == Upload =
|
||||
liftIO $ readMVar metervar
|
||||
liftIO $ maybe 0 fromBytesProcessed
|
||||
<$> readTVarIO metervar
|
||||
| otherwise = do
|
||||
f <- fromRepo $ gitAnnexTmpObjectLocation (transferKey t)
|
||||
liftIO $ catchDefaultIO 0 $ getFileSize f
|
||||
|
||||
detectStallsAndSuggestConfig :: Maybe StallDetection -> TVar (Maybe BytesProcessed) -> Annex a -> Annex a
|
||||
detectStallsAndSuggestConfig Nothing _ a = a
|
||||
detectStallsAndSuggestConfig sd@(Just _) metervar a =
|
||||
bracket setup cleanup (const a)
|
||||
where
|
||||
setup = do
|
||||
v <- liftIO newEmptyTMVarIO
|
||||
sdt <- liftIO $ async $ detectStalls sd metervar $
|
||||
void $ atomically $ tryPutTMVar v True
|
||||
wt <- liftIO . async =<< forkState (warnonstall v)
|
||||
return (v, sdt, wt)
|
||||
cleanup (v, sdt, wt) = do
|
||||
liftIO $ uninterruptibleCancel sdt
|
||||
void $ liftIO $ atomically $ tryPutTMVar v False
|
||||
join (liftIO (wait wt))
|
||||
warnonstall v = whenM (liftIO (atomically (takeTMVar v))) $
|
||||
warning "Transfer seems to have stalled. To restart stalled transfers, configure annex.stalldetection"
|
||||
|
||||
{- Runs a transfer using a separate process, which lets detected stalls be
|
||||
- canceled. -}
|
||||
runTransferrer
|
||||
:: StallDetection
|
||||
-> Remote
|
||||
|
|
|
|||
|
|
@ -18,6 +18,8 @@ git-annex (8.20210128) UNRELEASED; urgency=medium
|
|||
* annex.stalldetection can now be set to "true" to make git-annex
|
||||
do automatic stall detection when it detects a remote is updating its
|
||||
transfer progress consistently enough.
|
||||
* When annex.stalldetection is not enabled and a likely stall is
|
||||
detected, display a suggestion to enable it.
|
||||
|
||||
-- Joey Hess <id@joeyh.name> Thu, 28 Jan 2021 12:34:32 -0400
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import Types.Concurrency
|
|||
import Messages.Concurrent
|
||||
import Types.Messages
|
||||
import Types.WorkerPool
|
||||
import Remote.List
|
||||
|
||||
import Control.Concurrent
|
||||
import Control.Concurrent.Async
|
||||
|
|
@ -251,10 +252,19 @@ startConcurrency usedstages a = do
|
|||
initworkerpool n = do
|
||||
tv <- liftIO newEmptyTMVarIO
|
||||
Annex.changeState $ \s -> s { Annex.workers = Just tv }
|
||||
prepDupState
|
||||
st <- dupState
|
||||
liftIO $ atomically $ putTMVar tv $
|
||||
allocateWorkerPool st (max n 1) usedstages
|
||||
|
||||
-- Make sure that some expensive actions have been done before
|
||||
-- starting threads. This way the state has them already run,
|
||||
-- and each thread won't try to do them.
|
||||
prepDupState :: Annex ()
|
||||
prepDupState = do
|
||||
_ <- remoteList
|
||||
return ()
|
||||
|
||||
{- Ensures that only one thread processes a key at a time.
|
||||
- Other threads will block until it's done.
|
||||
-
|
||||
|
|
|
|||
|
|
@ -332,7 +332,7 @@ downloadWeb addunlockedmatcher o url urlinfo file =
|
|||
let cleanuptmp = pruneTmpWorkDirBefore tmp (liftIO . removeWhenExistsWith R.removeLink)
|
||||
showNote "using youtube-dl"
|
||||
Transfer.notifyTransfer Transfer.Download url $
|
||||
Transfer.download' webUUID mediakey (AssociatedFile Nothing) Transfer.noRetry $ \p ->
|
||||
Transfer.download' webUUID mediakey (AssociatedFile Nothing) Nothing Transfer.noRetry $ \p ->
|
||||
youtubeDl url (fromRawFilePath workdir) p >>= \case
|
||||
Right (Just mediafile) -> do
|
||||
cleanuptmp
|
||||
|
|
@ -396,7 +396,7 @@ downloadWith' downloader dummykey u url afile =
|
|||
checkDiskSpaceToGet dummykey Nothing $ do
|
||||
tmp <- fromRepo $ gitAnnexTmpObjectLocation dummykey
|
||||
ok <- Transfer.notifyTransfer Transfer.Download url $
|
||||
Transfer.download' u dummykey afile Transfer.stdRetry $ \p -> do
|
||||
Transfer.download' u dummykey afile Nothing Transfer.stdRetry $ \p -> do
|
||||
createAnnexDirectory (parentDir tmp)
|
||||
downloader (fromRawFilePath tmp) p
|
||||
if ok
|
||||
|
|
|
|||
|
|
@ -287,7 +287,7 @@ performExport r db ek af contentsha loc allfilledvar = do
|
|||
-- could be used for more than one export
|
||||
-- location, and concurrently uploading
|
||||
-- of the content should still be allowed.
|
||||
alwaysUpload (uuid r) k af stdRetry $ \pm -> do
|
||||
alwaysUpload (uuid r) k af Nothing stdRetry $ \pm -> do
|
||||
let rollback = void $
|
||||
performUnexport r db [ek] loc
|
||||
sendAnnex k rollback $ \f ->
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ fieldTransfer direction key a = do
|
|||
<$> Fields.getField Fields.associatedFile
|
||||
ok <- maybe (a $ const noop)
|
||||
-- Using noRetry here because we're the sender.
|
||||
(\u -> runner (Transfer direction (toUUID u) (fromKey id key)) afile noRetry a)
|
||||
(\u -> runner (Transfer direction (toUUID u) (fromKey id key)) afile Nothing noRetry a)
|
||||
=<< Fields.getField Fields.remoteUUID
|
||||
liftIO $ debugM "fieldTransfer" "transfer done"
|
||||
liftIO $ exitBool ok
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ start o (_, key) = startingCustomOutput key $ case fromToOptions o of
|
|||
|
||||
toPerform :: Key -> AssociatedFile -> Remote -> CommandPerform
|
||||
toPerform key file remote = go Upload file $
|
||||
upload' (uuid remote) key file stdRetry $ \p -> do
|
||||
upload' (uuid remote) key file Nothing stdRetry $ \p -> do
|
||||
tryNonAsync (Remote.storeKey remote key file p) >>= \case
|
||||
Right () -> do
|
||||
Remote.logStatus remote key InfoPresent
|
||||
|
|
@ -62,7 +62,7 @@ toPerform key file remote = go Upload file $
|
|||
|
||||
fromPerform :: Key -> AssociatedFile -> Remote -> CommandPerform
|
||||
fromPerform key file remote = go Upload file $
|
||||
download' (uuid remote) key file stdRetry $ \p ->
|
||||
download' (uuid remote) key file Nothing stdRetry $ \p ->
|
||||
logStatusAfter key $ getViaTmp (retrievalSecurityPolicy remote) (RemoteVerify remote) key file $ \t ->
|
||||
tryNonAsync (Remote.retrieveKeyFile remote key file (fromRawFilePath t) p) >>= \case
|
||||
Right v -> return (True, v)
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@ start = do
|
|||
where
|
||||
runner (TransferRequest direction remote key file)
|
||||
| direction == Upload = notifyTransfer direction file $
|
||||
upload' (Remote.uuid remote) key file stdRetry $ \p -> do
|
||||
upload' (Remote.uuid remote) key file Nothing stdRetry $ \p -> do
|
||||
tryNonAsync (Remote.storeKey remote key file p) >>= \case
|
||||
Left e -> do
|
||||
warning (show e)
|
||||
|
|
@ -49,7 +49,7 @@ start = do
|
|||
Remote.logStatus remote key InfoPresent
|
||||
return True
|
||||
| otherwise = notifyTransfer direction file $
|
||||
download' (Remote.uuid remote) key file stdRetry $ \p ->
|
||||
download' (Remote.uuid remote) key file Nothing stdRetry $ \p ->
|
||||
logStatusAfter key $ getViaTmp (Remote.retrievalSecurityPolicy remote) (RemoteVerify remote) key file $ \t -> do
|
||||
r <- tryNonAsync (Remote.retrieveKeyFile remote key file (fromRawFilePath t) p) >>= \case
|
||||
Left e -> do
|
||||
|
|
|
|||
|
|
@ -45,21 +45,23 @@ start = do
|
|||
runner (UploadRequest _ key (TransferAssociatedFile file)) remote =
|
||||
-- This is called by eg, Annex.Transfer.upload,
|
||||
-- so caller is responsible for doing notification,
|
||||
-- and for retrying, and updating location log.
|
||||
upload' (Remote.uuid remote) key file noRetry
|
||||
-- and for retrying, and updating location log,
|
||||
-- and stall canceling.
|
||||
upload' (Remote.uuid remote) key file Nothing noRetry
|
||||
(Remote.action . Remote.storeKey remote key file)
|
||||
noNotification
|
||||
runner (DownloadRequest _ key (TransferAssociatedFile file)) remote =
|
||||
-- This is called by eg, Annex.Transfer.download
|
||||
-- so caller is responsible for doing notification
|
||||
-- and for retrying, and updating location log.
|
||||
-- and for retrying, and updating location log,
|
||||
-- and stall canceling.
|
||||
let go p = getViaTmp (Remote.retrievalSecurityPolicy remote) (RemoteVerify remote) key file $ \t -> do
|
||||
Remote.verifiedAction (Remote.retrieveKeyFile remote key file (fromRawFilePath t) p)
|
||||
in download' (Remote.uuid remote) key file noRetry go
|
||||
in download' (Remote.uuid remote) key file Nothing noRetry go
|
||||
noNotification
|
||||
runner (AssistantUploadRequest _ key (TransferAssociatedFile file)) remote =
|
||||
notifyTransfer Upload file $
|
||||
upload' (Remote.uuid remote) key file stdRetry $ \p -> do
|
||||
upload' (Remote.uuid remote) key file Nothing stdRetry $ \p -> do
|
||||
tryNonAsync (Remote.storeKey remote key file p) >>= \case
|
||||
Left e -> do
|
||||
warning (show e)
|
||||
|
|
@ -69,7 +71,7 @@ start = do
|
|||
return True
|
||||
runner (AssistantDownloadRequest _ key (TransferAssociatedFile file)) remote =
|
||||
notifyTransfer Download file $
|
||||
download' (Remote.uuid remote) key file stdRetry $ \p ->
|
||||
download' (Remote.uuid remote) key file Nothing stdRetry $ \p ->
|
||||
logStatusAfter key $ getViaTmp (Remote.retrievalSecurityPolicy remote) (RemoteVerify remote) key file $ \t -> do
|
||||
r <- tryNonAsync (Remote.retrieveKeyFile remote key file (fromRawFilePath t) p) >>= \case
|
||||
Left e -> do
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
{- git-annex transfer information files and lock files
|
||||
-
|
||||
- Copyright 2012-2019 Joey Hess <id@joeyh.name>
|
||||
- Copyright 2012-2021 Joey Hess <id@joeyh.name>
|
||||
-
|
||||
- Licensed under the GNU AGPL version 3 or higher.
|
||||
-}
|
||||
|
|
@ -27,7 +27,7 @@ import Annex.Perms
|
|||
|
||||
import Data.Time.Clock
|
||||
import Data.Time.Clock.POSIX
|
||||
import Control.Concurrent
|
||||
import Control.Concurrent.STM
|
||||
import qualified Data.ByteString.Char8 as B8
|
||||
import qualified System.FilePath.ByteString as P
|
||||
|
||||
|
|
@ -58,23 +58,25 @@ percentComplete t info =
|
|||
- 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 MVar that can be used to read
|
||||
- the number of bytesComplete. -}
|
||||
mkProgressUpdater :: Transfer -> TransferInfo -> Annex (MeterUpdate, RawFilePath, Annex (), MVar Integer)
|
||||
- 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
|
||||
mvar <- liftIO $ newMVar 0
|
||||
return (liftIO . updater (fromRawFilePath tfile) mvar, tfile, createtfile, mvar)
|
||||
tvar <- liftIO $ newTVarIO Nothing
|
||||
loggedtvar <- liftIO $ newTVarIO 0
|
||||
return (liftIO . updater (fromRawFilePath tfile) tvar loggedtvar, tfile, createtfile, tvar)
|
||||
where
|
||||
updater tfile mvar b = modifyMVar_ mvar $ \oldbytes -> do
|
||||
let newbytes = fromBytesProcessed b
|
||||
if newbytes - oldbytes >= mindelta
|
||||
then do
|
||||
let info' = info { bytesComplete = Just newbytes }
|
||||
_ <- tryIO $ updateTransferInfoFile info' tfile
|
||||
return newbytes
|
||||
else return oldbytes
|
||||
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. -}
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ runLocal runst runner a = case a of
|
|||
case v of
|
||||
Right (Just (f, checkchanged)) -> proceed $ do
|
||||
-- alwaysUpload to allow multiple uploads of the same key.
|
||||
let runtransfer ti = transfer alwaysUpload k af $ \p ->
|
||||
let runtransfer ti = transfer alwaysUpload k af Nothing $ \p ->
|
||||
sinkfile f o checkchanged sender p ti
|
||||
checktransfer runtransfer fallback
|
||||
Right Nothing -> proceed fallback
|
||||
|
|
@ -75,7 +75,7 @@ runLocal runst runner a = case a of
|
|||
let rsp = RetrievalAllKeysSecure
|
||||
v <- tryNonAsync $ do
|
||||
let runtransfer ti =
|
||||
Right <$> transfer download' k af (\p ->
|
||||
Right <$> transfer download' k af Nothing (\p ->
|
||||
logStatusAfter k $ getViaTmp rsp DefaultVerify k af $ \tmp ->
|
||||
storefile (fromRawFilePath tmp) o l getb validitycheck p ti)
|
||||
let fallback = return $ Left $
|
||||
|
|
@ -145,11 +145,11 @@ runLocal runst runner a = case a of
|
|||
runner next
|
||||
RunValidityCheck checkaction next -> runner . next =<< checkaction
|
||||
where
|
||||
transfer mk k af ta = case runst of
|
||||
transfer mk k af sd ta = case runst of
|
||||
-- Update transfer logs when serving.
|
||||
-- Using noRetry because we're the sender.
|
||||
Serving theiruuid _ _ ->
|
||||
mk theiruuid k af noRetry ta noNotification
|
||||
mk theiruuid k af sd noRetry ta noNotification
|
||||
-- Transfer logs are updated higher in the stack when
|
||||
-- a client.
|
||||
Client _ -> ta nullMeterUpdate
|
||||
|
|
|
|||
|
|
@ -543,7 +543,7 @@ copyFromRemote'' repo forcersync r st@(State connpool _ _ _ _) key file dest met
|
|||
Just (object, checksuccess) -> do
|
||||
copier <- mkCopier hardlink st params
|
||||
(ok, v) <- runTransfer (Transfer Download u (fromKey id key))
|
||||
file stdRetry $ \p ->
|
||||
file Nothing stdRetry $ \p ->
|
||||
metered (Just (combineMeterUpdate p meterupdate)) key $ \_ p' ->
|
||||
copier object dest p' checksuccess
|
||||
if ok
|
||||
|
|
@ -688,7 +688,7 @@ copyToRemote' repo r st@(State connpool duc _ _ _) key file meterupdate
|
|||
-- run copy from perspective of remote
|
||||
res <- onLocalFast st $ ifM (Annex.Content.inAnnex key)
|
||||
( return True
|
||||
, runTransfer (Transfer Download u (fromKey id key)) file stdRetry $ \p -> do
|
||||
, runTransfer (Transfer Download u (fromKey id key)) file Nothing stdRetry $ \p -> do
|
||||
copier <- mkCopier hardlink st params
|
||||
let verify = Annex.Content.RemoteVerify r
|
||||
let rsp = RetrievalAllKeysSecure
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ data StallDetection
|
|||
| ProbeStallDetection
|
||||
-- ^ Used when unsure how frequently transfer progress is updated,
|
||||
-- or how fast data can be sent.
|
||||
| StallDetectionDisabled
|
||||
deriving (Show)
|
||||
|
||||
-- Parse eg, "0KiB/60s"
|
||||
|
|
@ -38,4 +39,4 @@ parseStallDetection s = case isTrueFalse s of
|
|||
d <- parseDuration ds
|
||||
return (Just (StallDetection b d))
|
||||
Just True -> Right (Just ProbeStallDetection)
|
||||
Just False -> Right Nothing
|
||||
Just False -> Right (Just StallDetectionDisabled)
|
||||
|
|
|
|||
|
|
@ -1422,16 +1422,20 @@ Remotes are configured using these settings in `.git/config`.
|
|||
|
||||
* `remote.<name>.annex-stalldetecton`, `annex.stalldetection`
|
||||
|
||||
This lets stalled or too-slow transfers be detected, and dealt with, so
|
||||
rather than getting stuck, git-annex will cancel the stalled operation.
|
||||
When this happens, the transfer will be considered to have failed, so
|
||||
Configuring this lets stalled or too-slow transfers be detected, and
|
||||
dealt with, so rather than getting stuck, git-annex will cancel the
|
||||
stalled operation. The transfer will be considered to have failed, so
|
||||
settings like annex.retry will control what it does next.
|
||||
|
||||
Set to "true" to enable automatic stall detection. With this setting,
|
||||
if a remote does not update its progress consistently, no stall detection
|
||||
will be done. And it may take a while for git-annex to decide a remote
|
||||
is really stalled when using automatic stall detection, since it needs
|
||||
to be conservative about what looks like a stall.
|
||||
By default, git-annex detects transfers that have probably stalled,
|
||||
and suggests configuring this. If it is incorrectly detecting
|
||||
stalls, setting this to "false" will avoid that.
|
||||
|
||||
Set to "true" to enable automatic stall detection. If a remote does not
|
||||
update its progress consistently, no automatic stall detection will be
|
||||
done. And it may take a while for git-annex to decide a remote is really
|
||||
stalled when using automatic stall detection, since it needs to be
|
||||
conservative about what looks like a stall.
|
||||
|
||||
For more fine control over what constitutes a stall, set to a value in
|
||||
the form "$amount/$timeperiod" to specify how much data git-annex should
|
||||
|
|
|
|||
|
|
@ -20,3 +20,5 @@ gets it wrong and the remote does not support resuming transfers,
|
|||
defaulting to on could lead to bad waste of resources. It could
|
||||
detect stalls even when not turned on, but only display a message
|
||||
suggesting enabling the config. --[[Joey]]
|
||||
|
||||
> [[done]] --[[Joey]]
|
||||
|
|
|
|||
Loading…
Reference in a new issue