mirrors/git-annex
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

converted 6 more threads

Browse source
This commit is contained in:
Joey Hess 2012-10-29 11:40:22 -04:00
parent bad88e404a
commit 76768ad977
8 changed files with 350 additions and 370 deletions
Download patch file Download diff file Expand all files Collapse all files

16
Assistant.hs
View file

@ -182,26 +182,24 @@ startAssistant assistant daemonize webappwaiter = withThreadState $ \st -> do
dstatus <- getAssistant daemonStatusHandle dstatus <- getAssistant daemonStatusHandle
changechan <- getAssistant changeChan changechan <- getAssistant changeChan
commitchan <- getAssistant commitChan commitchan <- getAssistant commitChan
pushmap <- getAssistant failedPushMap
transferqueue <- getAssistant transferQueue transferqueue <- getAssistant transferQueue
transferslots <- getAssistant transferSlots transferslots <- getAssistant transferSlots
scanremotes <- getAssistant scanRemoteMap scanremotes <- getAssistant scanRemoteMap
branchhandle <- getAssistant branchChangeHandle
pushnotifier <- getAssistant pushNotifier pushnotifier <- getAssistant pushNotifier
#ifdef WITH_WEBAPP #ifdef WITH_WEBAPP
urlrenderer <- liftIO newUrlRenderer urlrenderer <- liftIO newUrlRenderer
#endif #endif
mapM_ (startthread d) mapM_ (startthread d)
[ watch $ commitThread st changechan commitchan transferqueue dstatus [ watch $ commitThread
#ifdef WITH_WEBAPP #ifdef WITH_WEBAPP
, assist $ webAppThread d urlrenderer False Nothing webappwaiter , assist $ webAppThread d urlrenderer False Nothing webappwaiter
#ifdef WITH_PAIRING #ifdef WITH_PAIRING
, assist $ pairListenerThread st dstatus scanremotes urlrenderer , assist $ pairListenerThread st dstatus scanremotes urlrenderer
#endif #endif
#endif #endif
, assist $ pushThread st dstatus commitchan pushmap pushnotifier , assist $ pushThread
, assist $ pushRetryThread st dstatus pushmap pushnotifier , assist $ pushRetryThread
, assist $ mergeThread st dstatus transferqueue branchhandle , assist $ mergeThread
, assist $ transferWatcherThread st dstatus transferqueue , assist $ transferWatcherThread st dstatus transferqueue
, assist $ transferPollerThread , assist $ transferPollerThread
, assist $ transfererThread st dstatus transferqueue transferslots commitchan , assist $ transfererThread st dstatus transferqueue transferslots commitchan
@ -210,10 +208,10 @@ startAssistant assistant daemonize webappwaiter = withThreadState $ \st -> do
, assist $ mountWatcherThread st dstatus scanremotes pushnotifier , assist $ mountWatcherThread st dstatus scanremotes pushnotifier
, assist $ netWatcherThread , assist $ netWatcherThread
, assist $ netWatcherFallbackThread , assist $ netWatcherFallbackThread
, assist $ transferScannerThread st dstatus scanremotes transferqueue , assist $ transferScannerThread
, assist $ configMonitorThread st dstatus branchhandle commitchan , assist $ configMonitorThread
#ifdef WITH_XMPP #ifdef WITH_XMPP
, assist $ pushNotifierThread st dstatus pushnotifier , assist $ pushNotifierThread
#endif #endif
, watch $ watchThread , watch $ watchThread
] ]

215
Assistant/Threads/Committer.hs
View file

@ -13,7 +13,6 @@ import Assistant.Common
import Assistant.Changes import Assistant.Changes
import Assistant.Commits import Assistant.Commits
import Assistant.Alert import Assistant.Alert
import Assistant.ThreadedMonad
import Assistant.Threads.Watcher import Assistant.Threads.Watcher
import Assistant.TransferQueue import Assistant.TransferQueue
import Assistant.DaemonStatus import Assistant.DaemonStatus
@ -37,48 +36,40 @@ import Data.Tuple.Utils
import qualified Data.Set as S import qualified Data.Set as S
import Data.Either import Data.Either
thisThread :: ThreadName
thisThread = "Committer"
{- This thread makes git commits at appropriate times. -} {- This thread makes git commits at appropriate times. -}
commitThread :: ThreadState -> ChangeChan -> CommitChan -> TransferQueue -> DaemonStatusHandle -> NamedThread commitThread :: NamedThread
commitThread st changechan commitchan transferqueue dstatus = thread $ liftIO $ do commitThread = NamedThread "Committer" $ do
delayadd <- runThreadState st $ delayadd <- liftAnnex $
maybe delayaddDefault (Just . Seconds) . readish maybe delayaddDefault (Just . Seconds) . readish
<$> getConfig (annexConfig "delayadd") "" <$> getConfig (annexConfig "delayadd") ""
runEvery (Seconds 1) $ do runEvery (Seconds 1) <~> do
-- We already waited one second as a simple rate limiter. -- We already waited one second as a simple rate limiter.
-- Next, wait until at least one change is available for -- Next, wait until at least one change is available for
-- processing. -- processing.
changes <- getChanges changechan changes <- getChanges <<~ changeChan
-- Now see if now's a good time to commit. -- Now see if now's a good time to commit.
time <- getCurrentTime time <- liftIO getCurrentTime
if shouldCommit time changes if shouldCommit time changes
then do then do
readychanges <- handleAdds delayadd st changechan transferqueue dstatus changes readychanges <- handleAdds delayadd changes
if shouldCommit time readychanges if shouldCommit time readychanges
then do then do
brokendebug thisThread debug
[ "committing" [ "committing"
, show (length readychanges) , show (length readychanges)
, "changes" , "changes"
] ]
void $ alertWhile dstatus commitAlert $ dstatus <- getAssistant daemonStatusHandle
runThreadState st commitStaged void $ alertWhile dstatus commitAlert <~>
recordCommit commitchan liftAnnex commitStaged
recordCommit <<~ commitChan
else refill readychanges else refill readychanges
else refill changes else refill changes
where where
thread = NamedThread thisThread refill [] = noop
refill [] = noop refill cs = do
refill cs = do debug ["delaying commit of", show (length cs), "changes"]
brokendebug thisThread flip refillChanges cs <<~ changeChan
[ "delaying commit of"
, show (length cs)
, "changes"
]
refillChanges changechan cs
commitStaged :: Annex Bool commitStaged :: Annex Bool
commitStaged = do commitStaged = do
@ -99,12 +90,12 @@ commitStaged = do
- each other out, etc. Git returns nonzero on those, - each other out, etc. Git returns nonzero on those,
- so don't propigate out commit failures. -} - so don't propigate out commit failures. -}
return True return True
where where
nomessage ps nomessage ps
| Git.Version.older "1.7.2" = Param "-m" | Git.Version.older "1.7.2" = Param "-m"
: Param "autocommit" : ps : Param "autocommit" : ps
| otherwise = Param "--allow-empty-message" | otherwise = Param "--allow-empty-message"
: Param "-m" : Param "" : ps : Param "-m" : Param "" : ps
{- Decide if now is a good time to make a commit. {- Decide if now is a good time to make a commit.
- Note that the list of change times has an undefined order. - Note that the list of change times has an undefined order.
@ -118,9 +109,9 @@ shouldCommit now changes
| len > 10000 = True -- avoid bloating queue too much | len > 10000 = True -- avoid bloating queue too much
| length (filter thisSecond changes) < 10 = True | length (filter thisSecond changes) < 10 = True
| otherwise = False -- batch activity | otherwise = False -- batch activity
where where
len = length changes len = length changes
thisSecond c = now `diffUTCTime` changeTime c <= 1 thisSecond c = now `diffUTCTime` changeTime c <= 1
{- OSX needs a short delay after a file is added before locking it down, {- OSX needs a short delay after a file is added before locking it down,
- as pasting a file seems to try to set file permissions or otherwise - as pasting a file seems to try to set file permissions or otherwise
@ -152,77 +143,82 @@ delayaddDefault = Nothing
- Any pending adds that are not ready yet are put back into the ChangeChan, - Any pending adds that are not ready yet are put back into the ChangeChan,
- where they will be retried later. - where they will be retried later.
-} -}
handleAdds :: Maybe Seconds -> ThreadState -> ChangeChan -> TransferQueue -> DaemonStatusHandle -> [Change] -> IO [Change] handleAdds :: Maybe Seconds -> [Change] -> Assistant [Change]
handleAdds delayadd st changechan transferqueue dstatus cs = returnWhen (null incomplete) $ do handleAdds delayadd cs = returnWhen (null incomplete) $ do
let (pending, inprocess) = partition isPendingAddChange incomplete let (pending, inprocess) = partition isPendingAddChange incomplete
pending' <- findnew pending pending' <- findnew pending
(postponed, toadd) <- partitionEithers <$> safeToAdd delayadd st pending' inprocess (postponed, toadd) <- partitionEithers <$> safeToAdd delayadd pending' inprocess
unless (null postponed) $ unless (null postponed) $
refillChanges changechan postponed flip refillChanges postponed <<~ changeChan
returnWhen (null toadd) $ do returnWhen (null toadd) $ do
added <- catMaybes <$> forM toadd add added <- catMaybes <$> forM toadd add
if DirWatcher.eventsCoalesce || null added if DirWatcher.eventsCoalesce || null added
then return $ added ++ otherchanges then return $ added ++ otherchanges
else do else do
r <- handleAdds delayadd st changechan transferqueue dstatus r <- handleAdds delayadd
=<< getChanges changechan =<< getChanges <<~ changeChan
return $ r ++ added ++ otherchanges return $ r ++ added ++ otherchanges
where where
(incomplete, otherchanges) = partition (\c -> isPendingAddChange c || isInProcessAddChange c) cs (incomplete, otherchanges) = partition (\c -> isPendingAddChange c || isInProcessAddChange c) cs
findnew [] = return [] findnew [] = return []
findnew pending@(exemplar:_) = do findnew pending@(exemplar:_) = do
(!newfiles, cleanup) <- runThreadState st $ (!newfiles, cleanup) <- liftAnnex $
inRepo (Git.LsFiles.notInRepo False $ map changeFile pending) inRepo (Git.LsFiles.notInRepo False $ map changeFile pending)
void cleanup void $ liftIO cleanup
-- note: timestamp info is lost here -- note: timestamp info is lost here
let ts = changeTime exemplar let ts = changeTime exemplar
return $ map (PendingAddChange ts) newfiles return $ map (PendingAddChange ts) newfiles
returnWhen c a returnWhen c a
| c = return otherchanges | c = return otherchanges
| otherwise = a | otherwise = a
add :: Change -> IO (Maybe Change) add :: Change -> Assistant (Maybe Change)
add change@(InProcessAddChange { keySource = ks }) = add change@(InProcessAddChange { keySource = ks }) = do
alertWhile' dstatus (addFileAlert $ keyFilename ks) $ dstatus <- getAssistant daemonStatusHandle
liftM ret $ catchMaybeIO $ alertWhile' dstatus (addFileAlert $ keyFilename ks) <~> add' change ks
sanitycheck ks $ runThreadState st $ do add _ = return Nothing
showStart "add" $ keyFilename ks
key <- Command.Add.ingest ks
done (finishedChange change) (keyFilename ks) key
where
{- Add errors tend to be transient and will
- be automatically dealt with, so don't
- pass to the alert code. -}
ret (Just j@(Just _)) = (True, j)
ret _ = (True, Nothing)
add _ = return Nothing
done _ _ Nothing = do add' change ks = liftM ret $ catchMaybeIO <~> do
showEndFail sanitycheck ks $ do
return Nothing key <- liftAnnex $ do
done change file (Just key) = do showStart "add" $ keyFilename ks
link <- Command.Add.link file key True Command.Add.ingest ks
when DirWatcher.eventsCoalesce $ do done (finishedChange change) (keyFilename ks) key
where
{- Add errors tend to be transient and will be automatically
- dealt with, so don't pass to the alert code. -}
ret (Just j@(Just _)) = (True, j)
ret _ = (True, Nothing)
done _ _ Nothing = do
liftAnnex showEndFail
return Nothing
done change file (Just key) = do
link <- liftAnnex $ Command.Add.link file key True
when DirWatcher.eventsCoalesce $
liftAnnex $ do
sha <- inRepo $ sha <- inRepo $
Git.HashObject.hashObject BlobObject link Git.HashObject.hashObject BlobObject link
stageSymlink file sha stageSymlink file sha
queueTransfers Next transferqueue dstatus key (Just file) Upload showEndOk
showEndOk transferqueue <- getAssistant transferQueue
return $ Just change dstatus <- getAssistant daemonStatusHandle
liftAnnex $ queueTransfers Next transferqueue dstatus key (Just file) Upload
return $ Just change
{- Check that the keysource's keyFilename still exists, {- Check that the keysource's keyFilename still exists,
- and is still a hard link to its contentLocation, - and is still a hard link to its contentLocation,
- before ingesting it. -} - before ingesting it. -}
sanitycheck keysource a = do sanitycheck keysource a = do
fs <- getSymbolicLinkStatus $ keyFilename keysource fs <- liftIO $ getSymbolicLinkStatus $ keyFilename keysource
ks <- getSymbolicLinkStatus $ contentLocation keysource ks <- liftIO $ getSymbolicLinkStatus $ contentLocation keysource
if deviceID ks == deviceID fs && fileID ks == fileID fs if deviceID ks == deviceID fs && fileID ks == fileID fs
then a then a
else return Nothing else return Nothing
{- Files can Either be Right to be added now, {- Files can Either be Right to be added now,
- or are unsafe, and must be Left for later. - or are unsafe, and must be Left for later.
@ -230,11 +226,11 @@ handleAdds delayadd st changechan transferqueue dstatus cs = returnWhen (null in
- Check by running lsof on the temp directory, which - Check by running lsof on the temp directory, which
- the KeySources are locked down in. - the KeySources are locked down in.
-} -}
safeToAdd :: Maybe Seconds -> ThreadState -> [Change] -> [Change] -> IO [Either Change Change] safeToAdd :: Maybe Seconds -> [Change] -> [Change] -> Assistant [Either Change Change]
safeToAdd _ _ [] [] = return [] safeToAdd _ [] [] = return []
safeToAdd delayadd st pending inprocess = do safeToAdd delayadd pending inprocess = do
maybe noop threadDelaySeconds delayadd maybe noop (liftIO . threadDelaySeconds) delayadd
runThreadState st $ do liftAnnex $ do
keysources <- mapM Command.Add.lockDown (map changeFile pending) keysources <- mapM Command.Add.lockDown (map changeFile pending)
let inprocess' = map mkinprocess (zip pending keysources) let inprocess' = map mkinprocess (zip pending keysources)
tmpdir <- fromRepo gitAnnexTmpDir tmpdir <- fromRepo gitAnnexTmpDir
@ -250,25 +246,24 @@ safeToAdd delayadd st pending inprocess = do
mapM_ canceladd $ lefts checked mapM_ canceladd $ lefts checked
allRight $ rights checked allRight $ rights checked
else return checked else return checked
where where
check openfiles change@(InProcessAddChange { keySource = ks }) check openfiles change@(InProcessAddChange { keySource = ks })
| S.member (contentLocation ks) openfiles = Left change | S.member (contentLocation ks) openfiles = Left change
check _ change = Right change check _ change = Right change
mkinprocess (c, ks) = InProcessAddChange mkinprocess (c, ks) = InProcessAddChange
{ changeTime = changeTime c { changeTime = changeTime c
, keySource = ks , keySource = ks
} }
canceladd (InProcessAddChange { keySource = ks }) = do canceladd (InProcessAddChange { keySource = ks }) = do
warning $ keyFilename ks warning $ keyFilename ks
++ " still has writers, not adding" ++ " still has writers, not adding"
-- remove the hard link -- remove the hard link
void $ liftIO $ tryIO $ void $ liftIO $ tryIO $ removeFile $ contentLocation ks
removeFile $ contentLocation ks canceladd _ = noop
canceladd _ = noop
openwrite (_file, mode, _pid) = openwrite (_file, mode, _pid) =
mode == Lsof.OpenWriteOnly || mode == Lsof.OpenReadWrite mode == Lsof.OpenWriteOnly || mode == Lsof.OpenReadWrite
allRight = return . map Right allRight = return . map Right

72
Assistant/Threads/ConfigMonitor.hs
View file

@ -9,7 +9,6 @@ module Assistant.Threads.ConfigMonitor where
import Assistant.Common import Assistant.Common
import Assistant.BranchChange import Assistant.BranchChange
import Assistant.ThreadedMonad
import Assistant.DaemonStatus import Assistant.DaemonStatus
import Assistant.Commits import Assistant.Commits
import Utility.ThreadScheduler import Utility.ThreadScheduler
@ -19,10 +18,8 @@ import Logs.Remote
import Logs.PreferredContent import Logs.PreferredContent
import Logs.Group import Logs.Group
import Remote.List (remoteListRefresh) import Remote.List (remoteListRefresh)
import qualified Git
import qualified Git.LsTree as LsTree import qualified Git.LsTree as LsTree
import qualified Annex.Branch import qualified Annex.Branch
import qualified Annex
import qualified Data.Set as S import qualified Data.Set as S
@ -37,26 +34,22 @@ thisThread = "ConfigMonitor"
- if the branch has not changed in a while, configuration changes will - if the branch has not changed in a while, configuration changes will
- be detected immediately. - be detected immediately.
-} -}
configMonitorThread :: ThreadState -> DaemonStatusHandle -> BranchChangeHandle -> CommitChan -> NamedThread configMonitorThread :: NamedThread
configMonitorThread st dstatus branchhandle commitchan = thread $ liftIO $ do configMonitorThread = NamedThread "ConfigMonitor" $ loop =<< getConfigs
r <- runThreadState st Annex.gitRepo where
go r =<< getConfigs r loop old = do
where liftIO $ threadDelaySeconds (Seconds 60)
thread = NamedThread thisThread waitBranchChange <<~ branchChangeHandle
new <- getConfigs
go r old = do when (old /= new) $ do
threadDelaySeconds (Seconds 60) let changedconfigs = new `S.difference` old
waitBranchChange branchhandle debug $ "reloading config" :
new <- getConfigs r map fst (S.toList changedconfigs)
when (old /= new) $ do reloadConfigs new
let changedconfigs = new `S.difference` old {- Record a commit to get this config
brokendebug thisThread $ "reloading config" : - change pushed out to remotes. -}
map fst (S.toList changedconfigs) recordCommit <<~ commitChan
reloadConfigs st dstatus changedconfigs loop new
{- Record a commit to get this config
- change pushed out to remotes. -}
recordCommit commitchan
go r new
{- Config files, and their checksums. -} {- Config files, and their checksums. -}
type Configs = S.Set (FilePath, String) type Configs = S.Set (FilePath, String)
@ -73,22 +66,23 @@ configFilesActions =
, (preferredContentLog, noop) , (preferredContentLog, noop)
] ]
reloadConfigs :: ThreadState -> DaemonStatusHandle -> Configs -> IO () reloadConfigs :: Configs -> Assistant ()
reloadConfigs st dstatus changedconfigs = runThreadState st $ do reloadConfigs changedconfigs = do
sequence_ as liftAnnex $ do
void preferredContentMapLoad sequence_ as
void preferredContentMapLoad
{- Changes to the remote log, or the trust log, can affect the {- Changes to the remote log, or the trust log, can affect the
- syncRemotes list -} - syncRemotes list -}
when (Logs.Remote.remoteLog `elem` fs || Logs.Trust.trustLog `elem` fs) $ when (Logs.Remote.remoteLog `elem` fs || Logs.Trust.trustLog `elem` fs) $
updateSyncRemotes dstatus liftAnnex . updateSyncRemotes =<< getAssistant daemonStatusHandle
where where
(fs, as) = unzip $ filter (flip S.member changedfiles . fst) (fs, as) = unzip $ filter (flip S.member changedfiles . fst)
configFilesActions configFilesActions
changedfiles = S.map fst changedconfigs changedfiles = S.map fst changedconfigs
getConfigs :: Git.Repo -> IO Configs getConfigs :: Assistant Configs
getConfigs r = S.fromList . map extract getConfigs = S.fromList . map extract
<$> LsTree.lsTreeFiles Annex.Branch.fullname files r <$> liftAnnex (inRepo $ LsTree.lsTreeFiles Annex.Branch.fullname files)
where where
files = map fst configFilesActions files = map fst configFilesActions
extract treeitem = (LsTree.file treeitem, LsTree.sha treeitem) extract treeitem = (LsTree.file treeitem, LsTree.sha treeitem)

72
Assistant/Threads/Merger.hs
View file

@ -8,8 +8,6 @@
module Assistant.Threads.Merger where module Assistant.Threads.Merger where
import Assistant.Common import Assistant.Common
import Assistant.ThreadedMonad
import Assistant.DaemonStatus
import Assistant.TransferQueue import Assistant.TransferQueue
import Assistant.BranchChange import Assistant.BranchChange
import Utility.DirWatcher import Utility.DirWatcher
@ -24,36 +22,34 @@ thisThread = "Merger"
{- This thread watches for changes to .git/refs/, and handles incoming {- This thread watches for changes to .git/refs/, and handles incoming
- pushes. -} - pushes. -}
mergeThread :: ThreadState -> DaemonStatusHandle -> TransferQueue -> BranchChangeHandle -> NamedThread mergeThread :: NamedThread
mergeThread st dstatus transferqueue branchchange = thread $ liftIO $ do mergeThread = NamedThread "Merger" $ do
g <- runThreadState st gitRepo g <- liftAnnex gitRepo
let dir = Git.localGitDir g </> "refs" let dir = Git.localGitDir g </> "refs"
createDirectoryIfMissing True dir liftIO $ createDirectoryIfMissing True dir
let hook a = Just $ runHandler st dstatus transferqueue branchchange a let hook a = Just <$> asIO2 (runHandler a)
addhook <- hook onAdd
errhook <- hook onErr
let hooks = mkWatchHooks let hooks = mkWatchHooks
{ addHook = hook onAdd { addHook = addhook
, errHook = hook onErr , errHook = errhook
} }
void $ watchDir dir (const False) hooks id void $ liftIO $ watchDir dir (const False) hooks id
brokendebug thisThread ["watching", dir] debug ["watching", dir]
where
thread = NamedThread thisThread
type Handler = ThreadState -> DaemonStatusHandle -> TransferQueue -> BranchChangeHandle -> FilePath -> Maybe FileStatus -> IO () type Handler = FilePath -> Assistant ()
{- Runs an action handler. {- Runs an action handler.
- -
- Exceptions are ignored, otherwise a whole thread could be crashed. - Exceptions are ignored, otherwise a whole thread could be crashed.
-} -}
runHandler :: ThreadState -> DaemonStatusHandle -> TransferQueue -> BranchChangeHandle -> Handler -> FilePath -> Maybe FileStatus -> IO () runHandler :: Handler -> FilePath -> Maybe FileStatus -> Assistant ()
runHandler st dstatus transferqueue branchchange handler file filestatus = void $ runHandler handler file _filestatus =
either print (const noop) =<< tryIO go either (liftIO . print) (const noop) =<< tryIO <~> handler file
where
go = handler st dstatus transferqueue branchchange file filestatus
{- Called when there's an error with inotify. -} {- Called when there's an error with inotify. -}
onErr :: Handler onErr :: Handler
onErr _ _ _ _ msg _ = error msg onErr msg = error msg
{- Called when a new branch ref is written. {- Called when a new branch ref is written.
- -
@ -67,29 +63,29 @@ onErr _ _ _ _ msg _ = error msg
- ran are merged in. - ran are merged in.
-} -}
onAdd :: Handler onAdd :: Handler
onAdd st dstatus transferqueue branchchange file _ onAdd file
| ".lock" `isSuffixOf` file = noop | ".lock" `isSuffixOf` file = noop
| isAnnexBranch file = do | isAnnexBranch file = do
branchChanged branchchange branchChanged <<~ branchChangeHandle
runThreadState st $ transferqueue <- getAssistant transferQueue
dstatus <- getAssistant daemonStatusHandle
liftAnnex $
whenM Annex.Branch.forceUpdate $ whenM Annex.Branch.forceUpdate $
queueDeferredDownloads Later transferqueue dstatus queueDeferredDownloads Later transferqueue dstatus
| "/synced/" `isInfixOf` file = runThreadState st $ do | "/synced/" `isInfixOf` file = do
mergecurrent =<< inRepo Git.Branch.current mergecurrent =<< liftAnnex (inRepo Git.Branch.current)
| otherwise = noop | otherwise = noop
where where
changedbranch = fileToBranch file changedbranch = fileToBranch file
mergecurrent (Just current) mergecurrent (Just current)
| equivBranches changedbranch current = do | equivBranches changedbranch current = do
liftIO $ brokendebug thisThread debug
[ "merging" [ "merging", show changedbranch
, show changedbranch , "into", show current
, "into" ]
, show current void $ liftAnnex $ inRepo $
] Git.Merge.mergeNonInteractive changedbranch
void $ inRepo $ mergecurrent _ = noop
Git.Merge.mergeNonInteractive changedbranch
mergecurrent _ = noop
equivBranches :: Git.Ref -> Git.Ref -> Bool equivBranches :: Git.Ref -> Git.Ref -> Bool
equivBranches x y = base x == base y equivBranches x y = base x == base y

119
Assistant/Threads/PushNotifier.hs
View file

@ -12,7 +12,6 @@ module Assistant.Threads.PushNotifier where
import Assistant.Common import Assistant.Common
import Assistant.XMPP import Assistant.XMPP
import Assistant.ThreadedMonad
import Assistant.DaemonStatus import Assistant.DaemonStatus
import Assistant.Pushes import Assistant.Pushes
import Assistant.Sync import Assistant.Sync
@ -25,56 +24,56 @@ import qualified Data.Set as S
import qualified Git.Branch import qualified Git.Branch
import Data.Time.Clock import Data.Time.Clock
thisThread :: ThreadName pushNotifierThread :: NamedThread
thisThread = "PushNotifier" pushNotifierThread = NamedThread "PushNotifier" $ do
iodebug <- asIO debug
iopull <- asIO pull
pn <- getAssistant pushNotifier
controllerThread pn <~> xmppClient pn iodebug iopull
controllerThread :: PushNotifier -> IO () -> IO () controllerThread :: PushNotifier -> IO () -> IO ()
controllerThread pushnotifier a = forever $ do controllerThread pushnotifier xmppclient = forever $ do
tid <- forkIO a tid <- forkIO xmppclient
waitRestart pushnotifier waitRestart pushnotifier
killThread tid killThread tid
pushNotifierThread :: ThreadState -> DaemonStatusHandle -> PushNotifier -> NamedThread xmppClient :: PushNotifier -> ([String] -> IO ()) -> ([UUID] -> IO ()) -> Assistant ()
pushNotifierThread st dstatus pushnotifier = NamedThread thisThread $ liftIO $ xmppClient pushnotifier iodebug iopull = do
controllerThread pushnotifier $ do v <- liftAnnex getXMPPCreds
v <- runThreadState st $ getXMPPCreds case v of
case v of Nothing -> noop
Nothing -> noop Just c -> liftIO $ loop c =<< getCurrentTime
Just c -> loop c =<< getCurrentTime where
where loop c starttime = do
loop c starttime = do void $ connectXMPP c $ \jid -> do
void $ connectXMPP c $ \jid -> do fulljid <- bindJID jid
fulljid <- bindJID jid liftIO $ iodebug ["XMPP connected", show fulljid]
liftIO $ brokendebug thisThread ["XMPP connected", show fulljid] putStanza $ gitAnnexPresence gitAnnexSignature
putStanza $ gitAnnexPresence gitAnnexSignature s <- getSession
s <- getSession _ <- liftIO $ forkIO $ void $ runXMPP s $
_ <- liftIO $ forkIO $ void $ runXMPP s $ receivenotifications
receivenotifications sendnotifications
sendnotifications now <- getCurrentTime
now <- getCurrentTime if diffUTCTime now starttime > 300
if diffUTCTime now starttime > 300 then do
then do iodebug ["XMPP connection lost; reconnecting"]
brokendebug thisThread ["XMPP connection lost; reconnecting"] loop c now
loop c now else do
else do iodebug ["XMPP connection failed; will retry"]
brokendebug thisThread ["XMPP connection failed; will retry"] threadDelaySeconds (Seconds 300)
threadDelaySeconds (Seconds 300) loop c =<< getCurrentTime
loop c =<< getCurrentTime sendnotifications = forever $ do
us <- liftIO $ waitPush pushnotifier
sendnotifications = forever $ do putStanza $ gitAnnexPresence $ encodePushNotification us
us <- liftIO $ waitPush pushnotifier receivenotifications = forever $ do
putStanza $ gitAnnexPresence $ encodePushNotification us s <- getStanza
liftIO $ iodebug ["received XMPP:", show s]
receivenotifications = forever $ do case s of
s <- getStanza ReceivedPresence p@(Presence { presenceType = PresenceAvailable }) ->
liftIO $ brokendebug thisThread ["received XMPP:", show s] liftIO $ iopull $ concat $ catMaybes $
case s of map decodePushNotification $
ReceivedPresence p@(Presence { presenceType = PresenceAvailable }) -> presencePayloads p
liftIO $ pull st dstatus $ _ -> noop
concat $ catMaybes $
map decodePushNotification $
presencePayloads p
_ -> noop
{- We only pull from one remote out of the set listed in the push {- We only pull from one remote out of the set listed in the push
- notification, as an optimisation. - notification, as an optimisation.
@ -89,18 +88,18 @@ pushNotifierThread st dstatus pushnotifier = NamedThread thisThread $ liftIO $
- fully up-to-date. If that happens, the pushRetryThread will come along - fully up-to-date. If that happens, the pushRetryThread will come along
- and retry the push, and we'll get another notification once it succeeds, - and retry the push, and we'll get another notification once it succeeds,
- and pull again. -} - and pull again. -}
pull :: ThreadState -> DaemonStatusHandle -> [UUID] -> IO () pull :: [UUID] -> Assistant ()
pull _ _ [] = noop pull [] = noop
pull st dstatus us = do pull us = do
rs <- filter matching . syncRemotes <$> getDaemonStatus dstatus rs <- filter matching . syncRemotes <$> daemonStatus
brokendebug thisThread $ "push notification for" : debug $ "push notification for" : map (fromUUID . Remote.uuid ) rs
map (fromUUID . Remote.uuid ) rs st <- getAssistant threadState
pullone rs =<< runThreadState st (inRepo Git.Branch.current) liftIO . pullone st rs =<< liftAnnex (inRepo Git.Branch.current)
where where
matching r = Remote.uuid r `S.member` s matching r = Remote.uuid r `S.member` s
s = S.fromList us s = S.fromList us
pullone [] _ = noop pullone _ [] _ = noop
pullone (r:rs) branch = pullone st (r:rs) branch =
unlessM (all id . fst <$> manualPull st branch [r]) $ unlessM (all id . fst <$> manualPull st branch [r]) $
pullone rs branch pullone st rs branch

58
Assistant/Threads/Pusher.hs
View file

@ -11,7 +11,6 @@ import Assistant.Common
import Assistant.Commits import Assistant.Commits
import Assistant.Pushes import Assistant.Pushes
import Assistant.Alert import Assistant.Alert
import Assistant.ThreadedMonad
import Assistant.DaemonStatus import Assistant.DaemonStatus
import Assistant.Sync import Assistant.Sync
import Utility.ThreadScheduler import Utility.ThreadScheduler
@ -24,52 +23,49 @@ thisThread :: ThreadName
thisThread = "Pusher" thisThread = "Pusher"
{- This thread retries pushes that failed before. -} {- This thread retries pushes that failed before. -}
pushRetryThread :: ThreadState -> DaemonStatusHandle -> FailedPushMap -> PushNotifier -> NamedThread pushRetryThread :: NamedThread
pushRetryThread st dstatus pushmap pushnotifier = thread $ liftIO $ runEvery (Seconds halfhour) $ do pushRetryThread = NamedThread "PushRetrier" $ runEvery (Seconds halfhour) <~> do
-- We already waited half an hour, now wait until there are failed -- We already waited half an hour, now wait until there are failed
-- pushes to retry. -- pushes to retry.
topush <- getFailedPushesBefore pushmap (fromIntegral halfhour) pushmap <- getAssistant failedPushMap
topush <- liftIO $ getFailedPushesBefore pushmap (fromIntegral halfhour)
unless (null topush) $ do unless (null topush) $ do
brokendebug thisThread debug ["retrying", show (length topush), "failed pushes"]
[ "retrying" now <- liftIO $ getCurrentTime
, show (length topush) st <- getAssistant threadState
, "failed pushes" pushnotifier <- getAssistant pushNotifier
] dstatus <- getAssistant daemonStatusHandle
now <- getCurrentTime void $ liftIO $ alertWhile dstatus (pushRetryAlert topush) $
void $ alertWhile dstatus (pushRetryAlert topush) $
pushToRemotes thisThread now st (Just pushnotifier) (Just pushmap) topush pushToRemotes thisThread now st (Just pushnotifier) (Just pushmap) topush
where where
halfhour = 1800 halfhour = 1800
thread = NamedThread thisThread
{- This thread pushes git commits out to remotes soon after they are made. -} {- This thread pushes git commits out to remotes soon after they are made. -}
pushThread :: ThreadState -> DaemonStatusHandle -> CommitChan -> FailedPushMap -> PushNotifier -> NamedThread pushThread :: NamedThread
pushThread st dstatus commitchan pushmap pushnotifier = thread $ liftIO $ runEvery (Seconds 2) $ do pushThread = NamedThread "Pusher" $ runEvery (Seconds 2) <~> do
-- We already waited two seconds as a simple rate limiter. -- We already waited two seconds as a simple rate limiter.
-- Next, wait until at least one commit has been made -- Next, wait until at least one commit has been made
commits <- getCommits commitchan commits <- getCommits <<~ commitChan
-- Now see if now's a good time to push. -- Now see if now's a good time to push.
if shouldPush commits if shouldPush commits
then do then do
remotes <- filter pushable . syncRemotes remotes <- filter pushable . syncRemotes <$> daemonStatus
<$> getDaemonStatus dstatus
unless (null remotes) $ do unless (null remotes) $ do
now <- getCurrentTime now <- liftIO $ getCurrentTime
void $ alertWhile dstatus (pushAlert remotes) $ st <- getAssistant threadState
pushmap <- getAssistant failedPushMap
pushnotifier <- getAssistant pushNotifier
dstatus <- getAssistant daemonStatusHandle
void $ liftIO $ alertWhile dstatus (pushAlert remotes) $
pushToRemotes thisThread now st (Just pushnotifier) (Just pushmap) remotes pushToRemotes thisThread now st (Just pushnotifier) (Just pushmap) remotes
else do else do
brokendebug thisThread debug ["delaying push of", show (length commits), "commits"]
[ "delaying push of" flip refillCommits commits <<~ commitChan
, show (length commits) where
, "commits" pushable r
] | Remote.specialRemote r = False
refillCommits commitchan commits | Remote.readonly r = False
where | otherwise = True
thread = NamedThread thisThread
pushable r
| Remote.specialRemote r = False
| Remote.readonly r = False
| otherwise = True
{- Decide if now is a good time to push to remotes. {- Decide if now is a good time to push to remotes.
- -

164
Assistant/Threads/TransferScanner.hs
View file

@ -10,7 +10,6 @@ module Assistant.Threads.TransferScanner where
import Assistant.Common import Assistant.Common
import Assistant.ScanRemotes import Assistant.ScanRemotes
import Assistant.TransferQueue import Assistant.TransferQueue
import Assistant.ThreadedMonad
import Assistant.DaemonStatus import Assistant.DaemonStatus
import Assistant.Alert import Assistant.Alert
import Assistant.Drop import Assistant.Drop
@ -27,64 +26,64 @@ import Annex.Wanted
import qualified Data.Set as S import qualified Data.Set as S
thisThread :: ThreadName
thisThread = "TransferScanner"
{- This thread waits until a remote needs to be scanned, to find transfers {- This thread waits until a remote needs to be scanned, to find transfers
- that need to be made, to keep data in sync. - that need to be made, to keep data in sync.
-} -}
transferScannerThread :: ThreadState -> DaemonStatusHandle -> ScanRemoteMap -> TransferQueue -> NamedThread transferScannerThread :: NamedThread
transferScannerThread st dstatus scanremotes transferqueue = thread $ liftIO $ do transferScannerThread = NamedThread "TransferScanner" $ do
startupScan startupScan
go S.empty go S.empty
where where
thread = NamedThread thisThread go scanned = do
go scanned = do liftIO $ threadDelaySeconds (Seconds 2)
threadDelaySeconds (Seconds 2) (rs, infos) <- unzip <$> getScanRemote <<~ scanRemoteMap
(rs, infos) <- unzip <$> getScanRemote scanremotes if any fullScan infos || any (`S.notMember` scanned) rs
if any fullScan infos || any (`S.notMember` scanned) rs then do
then do expensiveScan rs
expensiveScan st dstatus transferqueue rs go $ scanned `S.union` S.fromList rs
go $ scanned `S.union` S.fromList rs else do
else do mapM_ failedTransferScan rs
mapM_ (failedTransferScan st dstatus transferqueue) rs go scanned
go scanned {- All available remotes are scanned in full on startup,
{- All available remotes are scanned in full on startup, - for multiple reasons, including:
- for multiple reasons, including: -
- - * This may be the first run, and there may be remotes
- * This may be the first run, and there may be remotes - already in place, that need to be synced.
- already in place, that need to be synced. - * We may have run before, and scanned a remote, but
- * We may have run before, and scanned a remote, but - only been in a subdirectory of the git remote, and so
- only been in a subdirectory of the git remote, and so - not synced it all.
- not synced it all. - * We may have run before, and had transfers queued,
- * We may have run before, and had transfers queued, - and then the system (or us) crashed, and that info was
- and then the system (or us) crashed, and that info was - lost.
- lost. -}
-} startupScan = do
startupScan = addScanRemotes scanremotes True scanremotes <- getAssistant scanRemoteMap
=<< syncRemotes <$> getDaemonStatus dstatus liftIO . addScanRemotes scanremotes True
=<< syncRemotes <$> daemonStatus
{- This is a cheap scan for failed transfers involving a remote. -} {- This is a cheap scan for failed transfers involving a remote. -}
failedTransferScan :: ThreadState -> DaemonStatusHandle -> TransferQueue -> Remote -> IO () failedTransferScan :: Remote -> Assistant ()
failedTransferScan st dstatus transferqueue r = do failedTransferScan r = do
failed <- runThreadState st $ getFailedTransfers (Remote.uuid r) failed <- liftAnnex $ getFailedTransfers (Remote.uuid r)
runThreadState st $ mapM_ removeFailedTransfer $ map fst failed liftAnnex $ mapM_ removeFailedTransfer $ map fst failed
mapM_ retry failed mapM_ retry failed
where where
retry (t, info) retry (t, info)
| transferDirection t == Download = do | transferDirection t == Download = do
{- Check if the remote still has the key. {- Check if the remote still has the key.
- If not, relies on the expensiveScan to - If not, relies on the expensiveScan to
- get it queued from some other remote. -} - get it queued from some other remote. -}
whenM (runThreadState st $ remoteHas r $ transferKey t) $ whenM (liftAnnex $ remoteHas r $ transferKey t) $
requeue t info
| otherwise = do
{- The Transferrer checks when uploading
- that the remote doesn't already have the
- key, so it's not redundantly checked
- here. -}
requeue t info requeue t info
requeue t info = queueTransferWhenSmall | otherwise = do
{- The Transferrer checks when uploading
- that the remote doesn't already have the
- key, so it's not redundantly checked here. -}
requeue t info
requeue t info = do
transferqueue <- getAssistant transferQueue
dstatus <- getAssistant daemonStatusHandle
liftIO $ queueTransferWhenSmall
transferqueue dstatus (associatedFile info) t r transferqueue dstatus (associatedFile info) t r
{- This is a expensive scan through the full git work tree, finding {- This is a expensive scan through the full git work tree, finding
@ -98,42 +97,45 @@ failedTransferScan st dstatus transferqueue r = do
- TODO: It would be better to first drop as much as we can, before - TODO: It would be better to first drop as much as we can, before
- transferring much, to minimise disk use. - transferring much, to minimise disk use.
-} -}
expensiveScan :: ThreadState -> DaemonStatusHandle -> TransferQueue -> [Remote] -> IO () expensiveScan :: [Remote] -> Assistant ()
expensiveScan st dstatus transferqueue rs = unless onlyweb $ do expensiveScan rs = unless onlyweb $ do
brokendebug thisThread ["starting scan of", show visiblers] debug ["starting scan of", show visiblers]
void $ alertWhile dstatus (scanAlert visiblers) $ do dstatus <- getAssistant daemonStatusHandle
g <- runThreadState st gitRepo void $ alertWhile dstatus (scanAlert visiblers) <~> do
(files, cleanup) <- LsFiles.inRepo [] g g <- liftAnnex gitRepo
(files, cleanup) <- liftIO $ LsFiles.inRepo [] g
forM_ files $ \f -> do forM_ files $ \f -> do
ts <- runThreadState st $ ts <- liftAnnex $
ifAnnexed f (findtransfers f) (return []) ifAnnexed f (findtransfers dstatus f) (return [])
mapM_ (enqueue f) ts mapM_ (enqueue f) ts
void cleanup void $ liftIO cleanup
return True return True
brokendebug thisThread ["finished scan of", show visiblers] debug ["finished scan of", show visiblers]
where where
onlyweb = all (== webUUID) $ map Remote.uuid rs onlyweb = all (== webUUID) $ map Remote.uuid rs
visiblers = let rs' = filter (not . Remote.readonly) rs visiblers = let rs' = filter (not . Remote.readonly) rs
in if null rs' then rs else rs' in if null rs' then rs else rs'
enqueue f (r, t) = do enqueue f (r, t) = do
brokendebug thisThread ["queuing", show t] debug ["queuing", show t]
queueTransferWhenSmall transferqueue dstatus (Just f) t r transferqueue <- getAssistant transferQueue
findtransfers f (key, _) = do dstatus <- getAssistant daemonStatusHandle
locs <- loggedLocations key liftIO $ queueTransferWhenSmall transferqueue dstatus (Just f) t r
{- The syncable remotes may have changed since this findtransfers dstatus f (key, _) = do
- scan began. -} locs <- loggedLocations key
syncrs <- liftIO $ syncRemotes <$> getDaemonStatus dstatus {- The syncable remotes may have changed since this
present <- inAnnex key - scan began. -}
syncrs <- liftIO $ syncRemotes <$> getDaemonStatus dstatus
present <- inAnnex key
handleDrops' locs syncrs present key (Just f) handleDrops' locs syncrs present key (Just f)
let slocs = S.fromList locs let slocs = S.fromList locs
let use a = return $ catMaybes $ map (a key slocs) syncrs let use a = return $ catMaybes $ map (a key slocs) syncrs
if present if present
then filterM (wantSend (Just f) . Remote.uuid . fst) then filterM (wantSend (Just f) . Remote.uuid . fst)
=<< use (genTransfer Upload False) =<< use (genTransfer Upload False)
else ifM (wantGet $ Just f) else ifM (wantGet $ Just f)
( use (genTransfer Download True) , return [] ) ( use (genTransfer Download True) , return [] )
genTransfer :: Direction -> Bool -> Key -> S.Set UUID -> Remote -> Maybe (Remote, Transfer) genTransfer :: Direction -> Bool -> Key -> S.Set UUID -> Remote -> Maybe (Remote, Transfer)
genTransfer direction want key slocs r genTransfer direction want key slocs r

4
Assistant/Threads/Watcher.hs
View file

@ -159,10 +159,10 @@ onAddSymlink file filestatus = go =<< liftAnnex (Backend.lookupFile file)
ensurestaged link daemonstatus ensurestaged link daemonstatus
| scanComplete daemonstatus = addlink link | scanComplete daemonstatus = addlink link
| otherwise = case filestatus of | otherwise = case filestatus of
Just s | changedrecently s -> liftIO noChange Just s | not (changedrecently s) -> liftIO noChange
_ -> addlink link _ -> addlink link
where where
changedrecently s = not $ changedrecently s =
afterLastDaemonRun (statusChangeTime s) daemonstatus afterLastDaemonRun (statusChangeTime s) daemonstatus
{- For speed, tries to reuse the existing blob for symlink target. -} {- For speed, tries to reuse the existing blob for symlink target. -}