refactor ref change watching

Added to change notification to P2P protocol.

Switched to a TBChan so that a single long-running thread can be
started, and serve perhaps intermittent requests for change
notifications, without buffering all changes in memory.

The P2P runner currently starts up a new thread each times it waits
for a change, but that should allow later reusing a thread. Although
each connection from a peer will still need a new watcher thread to run.

The dependency on stm-chans is more or less free; some stuff in yesod
uses it, so it was already indirectly pulled in when building with the
webapp.

This commit was sponsored by Francois Marier on Patreon.

1 files changed, 105 insertions, 0 deletions

diff --git a/Annex/ChangedRefs.hs b/Annex/ChangedRefs.hs
new file mode 100644
index 000000000..0dc82d3b3
--- /dev/null
+++ b/Annex/ChangedRefs.hs
@@ -0,0 +1,105 @@
+{- Waiting for changed git refs
+ -
+ - Copyright 2014-216 Joey Hess <id@joeyh.name>
+ -
+ - Licensed under the GNU GPL version 3 or higher.
+ -}
+
+module Annex.ChangedRefs
+	( ChangedRefs(..)
+	, ChangedRefsHandle
+	, waitChangedRefs
+	, drainChangedRefs
+	, stopWatchingChangedRefs
+	, watchChangedRefs
+	) where
+
+import Annex.Common
+import Utility.DirWatcher
+import Utility.DirWatcher.Types
+import qualified Git
+import Git.Sha
+import qualified Utility.SimpleProtocol as Proto
+
+import Control.Concurrent
+import Control.Concurrent.STM
+import Control.Concurrent.STM.TBMChan
+
+newtype ChangedRefs = ChangedRefs [Git.Ref]
+	deriving (Show)
+
+instance Proto.Serializable ChangedRefs where
+	serialize (ChangedRefs l) = unwords $ map Git.fromRef l
+	deserialize = Just . ChangedRefs . map Git.Ref . words
+
+data ChangedRefsHandle = ChangedRefsHandle DirWatcherHandle (TBMChan Git.Sha)
+
+-- | Wait for one or more git refs to change.
+--
+-- When possible, coalesce ref writes that occur closely together
+-- in time. Delay up to 0.05 seconds to get more ref writes.
+waitChangedRefs :: ChangedRefsHandle -> IO ChangedRefs
+waitChangedRefs (ChangedRefsHandle _ chan) = do
+	v <- atomically $ readTBMChan chan
+	case v of
+		Nothing -> return $ ChangedRefs []
+		Just r -> do
+			threadDelay 50000
+			rs <- atomically $ loop []
+			return $ ChangedRefs (r:rs)
+  where
+	loop rs = do
+		v <- tryReadTBMChan chan
+		case v of
+			Just (Just r) -> loop (r:rs)
+			_ -> return rs
+
+-- | Remove any changes that might be buffered in the channel,
+-- without waiting for any new changes.
+drainChangedRefs :: ChangedRefsHandle -> IO ()
+drainChangedRefs (ChangedRefsHandle _ chan) = atomically go
+  where
+	go = do
+		v <- tryReadTBMChan chan
+		case v of
+			Just (Just _) -> go
+			_ -> return ()
+
+stopWatchingChangedRefs :: ChangedRefsHandle -> IO ()
+stopWatchingChangedRefs h@(ChangedRefsHandle wh chan) = do
+	stopWatchDir wh
+	atomically $ closeTBMChan chan
+	drainChangedRefs h
+
+watchChangedRefs :: Annex ChangedRefsHandle
+watchChangedRefs = do
+	-- This channel is used to accumulate notifications,
+	-- because the DirWatcher might have multiple threads that find
+	-- changes at the same time. It is bounded to allow a watcher
+	-- to be started once and reused, without too many changes being
+	-- buffered in memory.
+	chan <- liftIO $ newTBMChanIO 100
+	
+	g <- gitRepo
+	let refdir = Git.localGitDir g </> "refs"
+	liftIO $ createDirectoryIfMissing True refdir
+
+	let notifyhook = Just $ notifyHook chan
+	let hooks = mkWatchHooks
+		{ addHook = notifyhook
+		, modifyHook = notifyhook
+		}
+
+	h <- liftIO $ watchDir refdir (const False) True hooks id
+	return $ ChangedRefsHandle h chan
+
+notifyHook :: TBMChan Git.Sha -> FilePath -> Maybe FileStatus -> IO ()
+notifyHook chan reffile _
+	| ".lock" `isSuffixOf` reffile = noop
+	| otherwise = void $ do
+		sha <- catchDefaultIO Nothing $
+			extractSha <$> readFile reffile
+		-- When the channel is full, there is probably no reader
+		-- running, or ref changes have been occuring very fast,
+		-- so it's ok to not write the change to it.
+		maybe noop (void . atomically . tryWriteTBMChan chan) sha