1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
{- git-annex assistant git pushing thread
-
- Copyright 2012 Joey Hess <joey@kitenet.net>
-
- Licensed under the GNU GPL version 3 or higher.
-}
module Assistant.Threads.Pusher where
import Assistant.Common
import Assistant.Commits
import Assistant.Pushes
import Assistant.Alert
import Assistant.ThreadedMonad
import Assistant.Threads.Merger
import Assistant.DaemonStatus
import qualified Command.Sync
import Utility.ThreadScheduler
import Utility.Parallel
import qualified Remote
import Data.Time.Clock
import qualified Data.Map as M
thisThread :: ThreadName
thisThread = "Pusher"
{- This thread retries pushes that failed before. -}
pushRetryThread :: ThreadState -> DaemonStatusHandle -> FailedPushMap -> IO ()
pushRetryThread st dstatus pushmap = runEvery (Seconds halfhour) $ do
-- We already waited half an hour, now wait until there are failed
-- pushes to retry.
topush <- getFailedPushesBefore pushmap (fromIntegral halfhour)
unless (null topush) $ do
debug thisThread
[ "retrying"
, show (length topush)
, "failed pushes"
]
now <- getCurrentTime
alertWhile dstatus (alert topush) $
pushToRemotes thisThread now st (Just pushmap) topush
where
halfhour = 1800
alert rs = activityAlert (Just "Retrying sync") $
"with " ++ unwords (map Remote.name rs) ++ ", which failed earlier."
{- This thread pushes git commits out to remotes soon after they are made. -}
pushThread :: ThreadState -> DaemonStatusHandle -> CommitChan -> FailedPushMap -> IO ()
pushThread st dstatus commitchan pushmap = do
runEvery (Seconds 2) $ do
-- We already waited two seconds as a simple rate limiter.
-- Next, wait until at least one commit has been made
commits <- getCommits commitchan
-- Now see if now's a good time to push.
now <- getCurrentTime
if shouldPush now commits
then do
remotes <- knownRemotes <$> getDaemonStatus dstatus
alertWhile dstatus (syncalert remotes) $
pushToRemotes thisThread now st (Just pushmap) remotes
else do
debug thisThread
[ "delaying push of"
, show (length commits)
, "commits"
]
refillCommits commitchan commits
where
syncalert rs = activityAlert Nothing $
"Syncing with " ++ unwords (map Remote.name rs)
{- Decide if now is a good time to push to remotes.
-
- Current strategy: Immediately push all commits. The commit machinery
- already determines batches of changes, so we can't easily determine
- batches better.
-}
shouldPush :: UTCTime -> [Commit] -> Bool
shouldPush _now commits
| not (null commits) = True
| otherwise = False
{- Updates the local sync branch, then pushes it to all remotes, in
- parallel.
-
- Avoids running possibly long-duration commands in the Annex monad, so
- as not to block other threads. -}
pushToRemotes :: ThreadName -> UTCTime -> ThreadState -> (Maybe FailedPushMap) -> [Remote] -> IO ()
pushToRemotes threadname now st mpushmap remotes = do
(g, branch) <- runThreadState st $
(,) <$> fromRepo id <*> Command.Sync.currentBranch
go True branch g remotes
where
go shouldretry branch g rs = do
debug threadname
[ "pushing to"
, show rs
]
Command.Sync.updateBranch (Command.Sync.syncBranch branch) g
(succeeded, failed) <- inParallel (push g branch) rs
case mpushmap of
Nothing -> noop
Just pushmap ->
changeFailedPushMap pushmap $ \m ->
M.union (makemap failed) $
M.difference m (makemap succeeded)
unless (null failed) $
debug threadname
[ "failed to push to"
, show failed
]
unless (null failed || not shouldretry) $
retry branch g failed
makemap l = M.fromList $ zip l (repeat now)
push g branch remote =
ifM (Command.Sync.pushBranch remote branch g)
( exitSuccess, exitFailure)
retry branch g rs = do
debug threadname [ "trying manual pull to resolve failed pushes" ]
runThreadState st $ manualPull branch rs
go False branch g rs
|
