summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--Command/Watch.hs76
1 files changed, 51 insertions, 25 deletions
diff --git a/Command/Watch.hs b/Command/Watch.hs
index 76f03a097..5c354cace 100644
--- a/Command/Watch.hs
+++ b/Command/Watch.hs
@@ -1,13 +1,36 @@
-{- git-annex command
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE BangPatterns #-}
+
+{- git-annex watch daemon
+ -
+ - Overview of threads and MVars, etc:
+ -
+ - Thread 1: Parent
+ - The initial thread run, double forks to background, starts other
+ - threads, and then stops, waiting for them to terminate.
+ - Thread 2: inotify
+ - Notices new files, and calls handlers for events, queuing changes.
+ - Thread 3: inotify internal
+ - Used by haskell inotify library to ensure inotify event buffer is
+ - kept drained.
+ - Thread 4: committer
+ - Waits for changes to occur, and runs the git queue to update its
+ - index, then commits.
+ -
+ - State MVar:
+ - The Annex state is stored here, which allows recuscitating the
+ - Annex monad in IO actions run by the inotify and committer
+ - threads. Thus, a single state is shared amoung the threads, and
+ - only one at a time can access it.
+ - ChangeChan STM TChan:
+ - Changes are indicated by writing to this channel. The committer
+ - reads from it.
-
- Copyright 2012 Joey Hess <joey@kitenet.net>
-
- Licensed under the GNU GPL version 3 or higher.
-}
-{-# LANGUAGE CPP #-}
-{-# LANGUAGE BangPatterns #-}
-
module Command.Watch where
import Common.Annex
@@ -83,7 +106,7 @@ startDaemon True st = do
watch :: MVar Annex.AnnexState -> IO ()
#if defined linux_HOST_OS
watch st = withINotify $ \i -> do
- changechan <- atomically newTChan
+ changechan <- runChangeChan newTChan
let hook a = Just $ runHandler st changechan a
let hooks = WatchHooks
{ addHook = hook onAdd
@@ -131,26 +154,38 @@ withStateMVar a = do
return r
{- Runs an Annex action, using the state from the MVar. -}
-runStateMVar :: MVar Annex.AnnexState -> Annex () -> IO ()
+runStateMVar :: MVar Annex.AnnexState -> Annex a -> IO a
runStateMVar mvar a = do
+ liftIO $ putStrLn "takeMVar"
startstate <- takeMVar mvar
- !newstate <- Annex.exec startstate a
+ !(r, newstate) <- Annex.run startstate a
+ liftIO $ putStrLn "putMVar"
putMVar mvar newstate
+ return r
-{- Runs an action handler, inside the Annex monad.
+runChangeChan :: STM a -> IO a
+runChangeChan = atomically
+
+{- Runs an action handler, inside the Annex monad, and if there was a
+ - change, adds it to the ChangeChan.
-
- Exceptions are ignored, otherwise a whole watcher thread could be crashed.
-}
runHandler :: MVar Annex.AnnexState -> ChangeChan -> Handler -> FilePath -> IO ()
-runHandler st changechan handler file =
- either (putStrLn . show) return =<< tryIO (runStateMVar st go)
- where
- go = maybe noop (signalChange changechan) =<< handler file
+runHandler st changechan handler file = void $ do
+ r <- tryIO (runStateMVar st $ handler file)
+ case r of
+ Left e -> putStrLn $ show e
+ Right Nothing -> noop
+ Right (Just change) -> void $
+ runChangeChan $ writeTChan changechan change
{- Handlers call this when they made a change that needs to get committed. -}
madeChange :: FilePath -> String -> Annex (Maybe Change)
-madeChange file desc = liftIO $
- Just <$> (Change <$> getCurrentTime <*> pure file <*> pure desc)
+madeChange file desc = do
+ -- Just in case the commit thread is not flushing the queue fast enough.
+ Annex.Queue.flushWhenFull
+ liftIO $ Just <$> (Change <$> getCurrentTime <*> pure file <*> pure desc)
noChange :: Annex (Maybe Change)
noChange = return Nothing
@@ -243,19 +278,10 @@ stageSymlink file sha =
Annex.Queue.addUpdateIndex =<<
inRepo (Git.UpdateIndex.stageSymlink file sha)
-{- Signals that a change has been made, that needs to get committed. -}
-signalChange :: ChangeChan -> Change -> Annex ()
-signalChange chan change = do
- liftIO $ atomically $ writeTChan chan change
-
- -- Just in case the commit thread is not flushing
- -- the queue fast enough.
- Annex.Queue.flushWhenFull
-
{- Gets all unhandled changes.
- Blocks until at least one change is made. -}
getChanges :: ChangeChan -> IO [Change]
-getChanges chan = atomically $ do
+getChanges chan = runChangeChan $ do
c <- readTChan chan
go [c]
where
@@ -268,7 +294,7 @@ getChanges chan = atomically $ do
{- Puts unhandled changes back into the channel.
- Note: Original order is not preserved. -}
refillChanges :: ChangeChan -> [Change] -> IO ()
-refillChanges chan cs = atomically $ mapM_ (writeTChan chan) cs
+refillChanges chan cs = runChangeChan $ mapM_ (writeTChan chan) cs
{- This thread makes git commits at appropriate times. -}
commitThread :: MVar Annex.AnnexState -> ChangeChan -> IO ()