From 2de50f733a01ce5b282ff0f6eb8a1101bd496216 Mon Sep 17 00:00:00 2001 From: Joey Hess Date: Sun, 10 Jun 2012 16:07:48 -0400 Subject: [PATCH] smart commit thread The commit thread now has access to a channel containing the times of all uncommitted changes. This lets it be smart about detecting busy times when a batch job is running (such as rm -rf, or untarring something, etc), and avoid committing until it's done. While at the same time, instantly committing one-off changes that the user is going to expect to see immediately. I had to use STM to implement the channel, because of http://hackage.haskell.org/trac/ghc/ticket/4154 While this adds a dependency, I always wanted to use STM, so this actually makes me happy. ;) Also happy that shouldCommit is a pure function, so other commit smartness strategies can easily be played with. Although the current one seems pretty good. There is one bug, for some reason it does double commits, every time. --- Command/Watch.hs | 105 ++++++++++++++++++++++++++++++++++++----------- debian/control | 1 + git-annex.cabal | 5 ++- 3 files changed, 84 insertions(+), 27 deletions(-) diff --git a/Command/Watch.hs b/Command/Watch.hs index 7c1bc5c177..34282e46c7 100644 --- a/Command/Watch.hs +++ b/Command/Watch.hs @@ -12,9 +12,6 @@ module Command.Watch where import Common.Annex import Command -#if defined linux_HOST_OS -import Utility.Inotify -#endif import Utility.ThreadLock import qualified Annex import qualified Annex.Queue @@ -25,11 +22,16 @@ import qualified Backend import Annex.Content import Control.Concurrent +import Control.Concurrent.STM +import Data.Time.Clock #if defined linux_HOST_OS +import Utility.Inotify import System.INotify #endif +type ChangeChan = TChan UTCTime + def :: [Command] def = [command "watch" paramPaths seek "watch for changes"] @@ -42,8 +44,10 @@ start = notBareRepo $ do showStart "watch" "." showAction "scanning" inRepo $ Git.Command.run "add" [Param "--update"] - next $ next $ withStateMVar $ \mvar -> liftIO $ withINotify $ \i -> do - let hook a = Just $ runHook mvar a + next $ next $ withStateMVar $ \st -> liftIO $ withINotify $ \i -> do + changechan <- atomically newTChan + _ <- forkIO $ commitThread st changechan + let hook a = Just $ runHook st changechan a let hooks = WatchHooks { addHook = hook onAdd , delHook = hook onDel @@ -52,7 +56,6 @@ start = notBareRepo $ do , errHook = hook onErr } watchDir i "." (ignored . takeFileName) hooks - _ <- forkIO $ commitThread mvar putStrLn "(started)" waitForTermination return True @@ -91,12 +94,12 @@ runStateMVar mvar a = do - - Exceptions are ignored, otherwise a whole watcher thread could be crashed. -} -runHook :: MVar Annex.AnnexState -> (FilePath -> Annex ()) -> FilePath -> IO () -runHook mvar a f = handle =<< tryIO (runStateMVar mvar go) +runHook :: MVar Annex.AnnexState -> ChangeChan -> (FilePath -> Annex ()) -> FilePath -> IO () +runHook st changetimes a f = handle =<< tryIO (runStateMVar st go) where go = do a f - Annex.Queue.flushWhenFull + signalChange changetimes handle (Right ()) = return () handle (Left e) = putStrLn $ show e @@ -160,20 +163,72 @@ stageSymlink file linktext = Annex.Queue.addUpdateIndex =<< inRepo (Git.UpdateIndex.stageSymlink file linktext) -{- This thread wakes up periodically and makes git commits. -} -commitThread :: MVar Annex.AnnexState -> IO () -commitThread mvar = forever $ do - threadDelay 1000000 -- 1 second - commit +{- Signals that a change has been made, that needs to get committed. -} +signalChange :: ChangeChan -> Annex () +signalChange chan = do + liftIO $ (atomically . writeTChan chan) =<< getCurrentTime + -- Just in case the commit thread is not flushing + -- the queue fast enough. + Annex.Queue.flushWhenFull + +{- Gets the times of all unhandled changes. + - Blocks until at least one change is made. -} +getChanges :: ChangeChan -> IO [UTCTime] +getChanges chan = atomically $ do + c <- readTChan chan + go [c] where - commit = tryIO $ runStateMVar mvar $ - whenM ((>) <$> Annex.Queue.size <*> pure 0) $ do - Annex.Queue.flush - {- Empty commits may be made if tree - - changes cancel each other out, etc. -} - inRepo $ Git.Command.run "commit" - [ Param "--allow-empty-message" - , Param "-m", Param "" - , Param "--allow-empty" - , Param "--quiet" - ] + go l = do + v <- tryReadTChan chan + case v of + Nothing -> return l + Just c -> go (c:l) + +{- Puts unhandled changes back into the channel. + - Note: Original order is not preserved. -} +refillChanges :: ChangeChan -> [UTCTime] -> IO () +refillChanges chan cs = atomically $ mapM_ (writeTChan chan) cs + +{- This thread makes git commits. -} +commitThread :: MVar Annex.AnnexState -> ChangeChan -> IO () +commitThread st changechan = forever $ do + -- First, a simple rate limiter. + threadDelay $ oneSecond + liftIO $ putStrLn "running" + -- Next, wait until at least one change has been made. + cs <- getChanges changechan + -- Now see if now's a good time to commit. + ifM (shouldCommit <$> getCurrentTime <*> pure cs) $ + ( commit + , do + liftIO $ putStrLn $ "no commit now " ++ show (length cs) + refillChanges changechan cs + ) + where + commit = void $ tryIO $ runStateMVar st $ do + Annex.Queue.flush + {- Empty commits may be made if tree + - changes cancel each other out, etc. -} + inRepo $ Git.Command.run "commit" + [ Param "--allow-empty-message" + , Param "-m", Param "" + , Param "--allow-empty" + , Param "--quiet" + ] + oneSecond = 1000000 -- microseconds + +{- Decide if now is a good time to make a commit. + - Note that the list of change times has an undefined order. + - + - Current strategy: If there have been 10 commits within the past second, + - a batch activity is taking place, so wait for later. + -} +shouldCommit :: UTCTime -> [UTCTime] -> Bool +shouldCommit now changetimes + | len == 0 = False + | len > 4096 = True -- avoid bloating queue too much + | length (filter thisSecond changetimes) < 10 = True + | otherwise = False -- batch activity + where + len = length changetimes + thisSecond t = now `diffUTCTime` t <= 1 diff --git a/debian/control b/debian/control index bfb0017bc8..6534fef317 100644 --- a/debian/control +++ b/debian/control @@ -21,6 +21,7 @@ Build-Depends: libghc-bloomfilter-dev, libghc-edit-distance-dev, libghc-hinotify-dev, + libghc-stm-dev, ikiwiki, perlmagick, git, diff --git a/git-annex.cabal b/git-annex.cabal index 114a4069fa..e1ad164535 100644 --- a/git-annex.cabal +++ b/git-annex.cabal @@ -36,7 +36,7 @@ Executable git-annex pcre-light, extensible-exceptions, dataenc, SHA, process, json, HTTP, base == 4.5.*, monad-control, transformers-base, lifted-base, IfElse, text, QuickCheck >= 2.1, bloomfilter, edit-distance, - hinotify + hinotify, STM Other-Modules: Utility.Touch C-Sources: Utility/libdiskfree.c Extensions: CPP @@ -52,7 +52,8 @@ Test-Suite test unix, containers, utf8-string, network, mtl, bytestring, old-locale, time, pcre-light, extensible-exceptions, dataenc, SHA, process, json, HTTP, base == 4.5.*, monad-control, transformers-base, lifted-base, - IfElse, text, QuickCheck >= 2.1, bloomfilter, edit-distance + IfElse, text, QuickCheck >= 2.1, bloomfilter, edit-distance, + hinotify, STM C-Sources: Utility/libdiskfree.c Extensions: CPP