reorg threads

5 files changed, 641 insertions, 0 deletions

diff --git a/Assistant/Threads/Committer.hs b/Assistant/Threads/Committer.hs
new file mode 100644
index 000000000..488056fa2
--- /dev/null
+++ b/Assistant/Threads/Committer.hs
@@ -0,0 +1,199 @@
+{- git-annex assistant commit thread
+ -
+ - Copyright 2012 Joey Hess <joey@kitenet.net>
+ -
+ - Licensed under the GNU GPL version 3 or higher.
+ -}
+
+module Assistant.Threads.Committer where
+
+import Common.Annex
+import Assistant.Changes
+import Assistant.Commits
+import Assistant.ThreadedMonad
+import Assistant.Threads.Watcher
+import qualified Annex
+import qualified Annex.Queue
+import qualified Git.Command
+import qualified Git.HashObject
+import Git.Types
+import qualified Command.Add
+import Utility.ThreadScheduler
+import qualified Utility.Lsof as Lsof
+import qualified Utility.DirWatcher as DirWatcher
+import Types.KeySource
+
+import Data.Time.Clock
+import Data.Tuple.Utils
+import qualified Data.Set as S
+import Data.Either
+
+{- This thread makes git commits at appropriate times. -}
+commitThread :: ThreadState -> ChangeChan -> CommitChan -> IO ()
+commitThread st changechan commitchan = runEvery (Seconds 1) $ do
+	-- We already waited one second as a simple rate limiter.
+	-- Next, wait until at least one change is available for
+	-- processing.
+	changes <- getChanges changechan
+	-- Now see if now's a good time to commit.
+	time <- getCurrentTime
+	if shouldCommit time changes
+		then do
+			readychanges <- handleAdds st changechan changes
+			if shouldCommit time readychanges
+				then do
+					void $ tryIO $ runThreadState st commitStaged
+					recordCommit commitchan (Commit time)
+				else refillChanges changechan readychanges
+		else refillChanges changechan changes
+
+commitStaged :: Annex ()
+commitStaged = do
+	Annex.Queue.flush
+	inRepo $ Git.Command.run "commit"
+		[ Param "--allow-empty-message"
+		, Param "-m", Param ""
+		-- Empty commits may be made if tree changes cancel
+		-- each other out, etc
+		, Param "--allow-empty"
+		-- Avoid running the usual git-annex pre-commit hook;
+		-- watch does the same symlink fixing, and we don't want
+		-- to deal with unlocked files in these commits.
+		, Param "--quiet"
+		]
+
+{- 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 changes within the past second,
+ - a batch activity is taking place, so wait for later.
+ -}
+shouldCommit :: UTCTime -> [Change] -> Bool
+shouldCommit now changes
+	| len == 0 = False
+	| len > 10000 = True -- avoid bloating queue too much
+	| length (filter thisSecond changes) < 10 = True
+	| otherwise = False -- batch activity
+	where
+		len = length changes
+		thisSecond c = now `diffUTCTime` changeTime c <= 1
+
+{- If there are PendingAddChanges, the files have not yet actually been
+ - added to the annex (probably), and that has to be done now, before
+ - committing.
+ -
+ - Deferring the adds to this point causes batches to be bundled together,
+ - which allows faster checking with lsof that the files are not still open
+ - for write by some other process.
+ -
+ - When a file is added, Inotify will notice the new symlink. So this waits
+ - for additional Changes to arrive, so that the symlink has hopefully been
+ - staged before returning, and will be committed immediately.
+ -
+ - OTOH, for kqueue, eventsCoalesce, so instead the symlink is directly
+ - created and staged.
+ -
+ - Returns a list of all changes that are ready to be committed.
+ - Any pending adds that are not ready yet are put back into the ChangeChan,
+ - where they will be retried later.
+ -}
+handleAdds :: ThreadState -> ChangeChan -> [Change] -> IO [Change]
+handleAdds st changechan cs = returnWhen (null pendingadds) $ do
+	(postponed, toadd) <- partitionEithers <$>
+		safeToAdd st pendingadds
+
+	unless (null postponed) $
+		refillChanges changechan postponed
+
+	returnWhen (null toadd) $ do
+		added <- catMaybes <$> forM toadd add
+		if (DirWatcher.eventsCoalesce || null added)
+			then return $ added ++ otherchanges
+			else do
+				r <- handleAdds st changechan
+					=<< getChanges changechan
+				return $ r ++ added ++ otherchanges
+	where
+		(pendingadds, otherchanges) = partition isPendingAddChange cs
+
+		returnWhen c a
+			| c = return otherchanges
+			| otherwise = a
+
+		add :: Change -> IO (Maybe Change)
+		add change@(PendingAddChange { keySource = ks }) = do
+			r <- catchMaybeIO $ sanitycheck ks $ runThreadState st $ do
+				showStart "add" $ keyFilename ks
+				handle (finishedChange change) (keyFilename ks)
+					=<< Command.Add.ingest ks
+			return $ maybeMaybe r
+		add _ = return Nothing
+
+		maybeMaybe (Just j@(Just _)) = j
+		maybeMaybe _ = Nothing
+
+		handle _ _ Nothing = do
+			showEndFail
+			return Nothing
+		handle change file (Just key) = do
+			link <- Command.Add.link file key True
+			when DirWatcher.eventsCoalesce $ do
+				sha <- inRepo $
+					Git.HashObject.hashObject BlobObject link
+				stageSymlink file sha
+			showEndOk
+			return $ Just change
+
+		{- Check that the keysource's keyFilename still exists,
+		 - and is still a hard link to its contentLocation,
+		 - before ingesting it. -}
+		sanitycheck keysource a = do
+			fs <- getSymbolicLinkStatus $ keyFilename keysource
+			ks <- getSymbolicLinkStatus $ contentLocation keysource
+			if deviceID ks == deviceID fs && fileID ks == fileID fs
+				then a
+				else return Nothing
+
+{- PendingAddChanges can Either be Right to be added now,
+ - or are unsafe, and must be Left for later.
+ -
+ - Check by running lsof on the temp directory, which
+ - the KeySources are locked down in.
+ -}
+safeToAdd :: ThreadState -> [Change] -> IO [Either Change Change]
+safeToAdd st changes = runThreadState st $
+	ifM (Annex.getState Annex.force)
+		( allRight changes -- force bypasses lsof check
+		, do
+			tmpdir <- fromRepo gitAnnexTmpDir
+			openfiles <- S.fromList . map fst3 . filter openwrite <$>
+				liftIO (Lsof.queryDir tmpdir)
+			
+			let checked = map (check openfiles) changes
+
+			{- If new events are received when files are closed,
+			 - there's no need to retry any changes that cannot
+			 - be done now. -}
+			if DirWatcher.closingTracked
+				then do
+					mapM_ canceladd $ lefts checked
+					allRight $ rights checked
+				else return checked
+		)
+	where
+		check openfiles change@(PendingAddChange { keySource = ks })
+			| S.member (contentLocation ks) openfiles = Left change
+		check _ change = Right change
+
+		canceladd (PendingAddChange { keySource = ks }) = do
+			warning $ keyFilename ks
+				++ " still has writers, not adding"
+			-- remove the hard link
+			void $ liftIO $ tryIO $
+				removeFile $ contentLocation ks
+		canceladd _ = noop
+
+		openwrite (_file, mode, _pid) =
+			mode == Lsof.OpenWriteOnly || mode == Lsof.OpenReadWrite
+
+		allRight = return . map Right
diff --git a/Assistant/Threads/Merger.hs b/Assistant/Threads/Merger.hs
new file mode 100644
index 000000000..d2c8b9b76
--- /dev/null
+++ b/Assistant/Threads/Merger.hs
@@ -0,0 +1,72 @@
+{- git-annex assistant git merge thread
+ -
+ - Copyright 2012 Joey Hess <joey@kitenet.net>
+ -
+ - Licensed under the GNU GPL version 3 or higher.
+ -}
+
+module Assistant.Threads.Merger where
+
+import Common.Annex
+import Assistant.ThreadedMonad
+import Utility.DirWatcher
+import Utility.Types.DirWatcher
+import qualified Git
+import qualified Git.Merge
+import qualified Git.Branch
+import qualified Command.Sync
+
+{- This thread watches for changes to .git/refs/heads/synced/*,
+ - which indicate incoming pushes. It merges those pushes into the
+ - currently checked out branch. -}
+mergeThread :: ThreadState -> IO ()
+mergeThread st = do
+	g <- runThreadState st $ fromRepo id
+	let dir = Git.localGitDir g </> "refs" </> "heads" </> "synced"
+	createDirectoryIfMissing True dir
+	let hook a = Just $ runHandler g a
+	let hooks = mkWatchHooks
+		{ addHook = hook onAdd
+		, errHook = hook onErr
+		}
+	watchDir dir (const False) hooks id
+	where
+
+type Handler = Git.Repo -> FilePath -> Maybe FileStatus -> IO ()
+
+{- Runs an action handler.
+ -
+ - Exceptions are ignored, otherwise a whole thread could be crashed.
+ -}
+runHandler :: Git.Repo -> Handler -> FilePath -> Maybe FileStatus -> IO ()
+runHandler g handler file filestatus = void $ do
+        either print (const noop) =<< tryIO go
+        where
+                go = handler g file filestatus
+
+{- Called when there's an error with inotify. -}
+onErr :: Handler
+onErr _ msg _ = error msg
+
+{- Called when a new branch ref is written.
+ -
+ - This relies on git's atomic method of updating branch ref files,
+ - which is to first write the new file to .lock, and then rename it
+ - over the old file. So, ignore .lock files, and the rename ensures
+ - the watcher sees a new file being added on each update.
+ -
+ - At startup, synthetic add events fire, causing this to run, but that's
+ - ok; it ensures that any changes pushed since the last time the assistant
+ - ran are merged in.
+ -}
+onAdd :: Handler
+onAdd g file _
+	| ".lock" `isSuffixOf` file = noop
+	| otherwise = do
+		let branch = Git.Ref $ "refs" </> "heads" </> takeFileName file
+		current <- Git.Branch.current g
+		when (Just branch == current) $
+			void $ mergeBranch branch g
+
+mergeBranch :: Git.Ref -> Git.Repo -> IO Bool
+mergeBranch = Git.Merge.mergeNonInteractive . Command.Sync.syncBranch
diff --git a/Assistant/Threads/Pusher.hs b/Assistant/Threads/Pusher.hs
new file mode 100644
index 000000000..de90d4e64
--- /dev/null
+++ b/Assistant/Threads/Pusher.hs
@@ -0,0 +1,69 @@
+{- 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 Common.Annex
+import Assistant.Commits
+import Assistant.ThreadedMonad
+import qualified Command.Sync
+import Utility.ThreadScheduler
+import Utility.Parallel
+
+import Data.Time.Clock
+
+data FailedPush = FailedPush 
+	{ failedRemote :: Remote
+	, failedTimeStamp :: UTCTime
+	}
+
+{- This thread pushes git commits out to remotes. -}
+pushThread :: ThreadState -> CommitChan -> IO ()
+pushThread st commitchan = do
+	remotes <- runThreadState st $ Command.Sync.syncRemotes []
+	runEveryWith (Seconds 2) [] $ \failedpushes -> 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.
+		time <- getCurrentTime
+		if shouldPush time commits failedpushes
+			then pushToRemotes time st remotes
+			else do
+				refillCommits commitchan commits
+				return failedpushes
+
+{- 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.
+ -
+ - TODO: FailedPushs are only retried the next time there's a commit.
+ - Should retry them periodically, or when a remote that was not available
+ - becomes available.
+ -}
+shouldPush :: UTCTime -> [Commit] -> [FailedPush] -> Bool
+shouldPush _now commits _failedremotes
+	| 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 :: UTCTime -> ThreadState -> [Remote] -> IO [FailedPush]
+pushToRemotes now st remotes = do
+	(g, branch) <- runThreadState st $
+		(,) <$> fromRepo id <*> Command.Sync.currentBranch
+	Command.Sync.updateBranch (Command.Sync.syncBranch branch) g
+	map (`FailedPush` now) <$> inParallel (push g branch) remotes
+	where
+		push g branch remote =
+			ifM (Command.Sync.pushBranch remote branch g)
+				( exitSuccess, exitFailure)
diff --git a/Assistant/Threads/SanityChecker.hs b/Assistant/Threads/SanityChecker.hs
new file mode 100644
index 000000000..4db2a61b2
--- /dev/null
+++ b/Assistant/Threads/SanityChecker.hs
@@ -0,0 +1,83 @@
+{- git-annex assistant sanity checker
+ -
+ - Copyright 2012 Joey Hess <joey@kitenet.net>
+ -
+ - Licensed under the GNU GPL version 3 or higher.
+ -}
+
+module Assistant.Threads.SanityChecker (
+	sanityCheckerThread
+) where
+
+import Common.Annex
+import qualified Git.LsFiles
+import Assistant.DaemonStatus
+import Assistant.ThreadedMonad
+import Assistant.Changes
+import Utility.ThreadScheduler
+import qualified Assistant.Threads.Watcher as Watcher
+
+import Data.Time.Clock.POSIX
+
+{- This thread wakes up occasionally to make sure the tree is in good shape. -}
+sanityCheckerThread :: ThreadState -> DaemonStatusHandle -> ChangeChan -> IO ()
+sanityCheckerThread st status changechan = forever $ do
+	waitForNextCheck st status
+
+	runThreadState st $
+		modifyDaemonStatus status $ \s -> s
+				{ sanityCheckRunning = True }
+
+	now <- getPOSIXTime -- before check started
+	catchIO (check st status changechan)
+		(runThreadState st . warning . show)
+
+	runThreadState st $ do
+		modifyDaemonStatus status $ \s -> s
+			{ sanityCheckRunning = False
+			, lastSanityCheck = Just now
+			}
+
+{- Only run one check per day, from the time of the last check. -}
+waitForNextCheck :: ThreadState -> DaemonStatusHandle -> IO ()
+waitForNextCheck st status = do
+	v <- runThreadState st $
+		lastSanityCheck <$> getDaemonStatus status
+	now <- getPOSIXTime
+	threadDelaySeconds $ Seconds $ calcdelay now v
+	where
+		calcdelay _ Nothing = oneDay
+		calcdelay now (Just lastcheck)
+			| lastcheck < now = max oneDay $
+				oneDay - truncate (now - lastcheck)
+			| otherwise = oneDay
+
+oneDay :: Int
+oneDay = 24 * 60 * 60
+
+{- It's important to stay out of the Annex monad as much as possible while
+ - running potentially expensive parts of this check, since remaining in it
+ - will block the watcher. -}
+check :: ThreadState -> DaemonStatusHandle -> ChangeChan -> IO () 
+check st status changechan = do
+	g <- runThreadState st $ do
+		showSideAction "Running daily check"
+		fromRepo id
+	-- Find old unstaged symlinks, and add them to git.
+	unstaged <- Git.LsFiles.notInRepo False ["."] g
+	now <- getPOSIXTime
+	forM_ unstaged $ \file -> do
+		ms <- catchMaybeIO $ getSymbolicLinkStatus file
+		case ms of
+			Just s	| toonew (statusChangeTime s) now -> noop
+				| isSymbolicLink s ->
+					addsymlink file ms
+			_ -> noop
+	where
+		toonew timestamp now = now < (realToFrac (timestamp + slop) :: POSIXTime)
+		slop = fromIntegral tenMinutes
+		insanity m = runThreadState st $ warning m
+		addsymlink file s = do
+			insanity $ "found unstaged symlink: " ++ file
+			Watcher.runHandler st status changechan
+				Watcher.onAddSymlink file s
diff --git a/Assistant/Threads/Watcher.hs b/Assistant/Threads/Watcher.hs
new file mode 100644
index 000000000..1b6ec15f1
--- /dev/null
+++ b/Assistant/Threads/Watcher.hs
@@ -0,0 +1,218 @@
+{- git-annex assistant tree watcher
+ -
+ - Copyright 2012 Joey Hess <joey@kitenet.net>
+ -
+ - Licensed under the GNU GPL version 3 or higher.
+ -}
+
+{-# LANGUAGE CPP #-}
+
+module Assistant.Threads.Watcher where
+
+import Common.Annex
+import Assistant.ThreadedMonad
+import Assistant.DaemonStatus
+import Assistant.Changes
+import Utility.DirWatcher
+import Utility.Types.DirWatcher
+import qualified Annex
+import qualified Annex.Queue
+import qualified Git.Command
+import qualified Git.UpdateIndex
+import qualified Git.HashObject
+import qualified Git.LsFiles
+import qualified Backend
+import qualified Command.Add
+import Annex.Content
+import Annex.CatFile
+import Git.Types
+
+import Data.Bits.Utils
+import qualified Data.ByteString.Lazy as L
+
+checkCanWatch :: Annex ()
+checkCanWatch
+	| canWatch = 
+		unlessM (liftIO (inPath "lsof") <||> Annex.getState Annex.force) $
+			needLsof
+	| otherwise = error "watch mode is not available on this system"
+
+needLsof :: Annex ()
+needLsof = error $ unlines
+	[ "The lsof command is needed for watch mode to be safe, and is not in PATH."
+	, "To override lsof checks to ensure that files are not open for writing"
+	, "when added to the annex, you can use --force"
+	, "Be warned: This can corrupt data in the annex, and make fsck complain."
+	]
+
+watchThread :: ThreadState -> DaemonStatusHandle -> ChangeChan -> IO ()
+watchThread st dstatus changechan = watchDir "." ignored hooks startup
+	where
+		startup = statupScan st dstatus
+		hook a = Just $ runHandler st dstatus changechan a
+		hooks = WatchHooks
+			{ addHook = hook onAdd
+			, delHook = hook onDel
+			, addSymlinkHook = hook onAddSymlink
+			, delDirHook = hook onDelDir
+			, errHook = hook onErr
+			}
+
+{- Initial scartup scan. The action should return once the scan is complete. -}
+statupScan :: ThreadState -> DaemonStatusHandle -> IO a -> IO a
+statupScan st dstatus scanner = do
+	runThreadState st $
+		showAction "scanning"
+	r <- scanner
+	runThreadState st $
+		modifyDaemonStatus dstatus $ \s -> s { scanComplete = True }
+
+	-- Notice any files that were deleted before watching was started.
+	runThreadState st $ do
+		inRepo $ Git.Command.run "add" [Param "--update"]
+		showAction "started"
+
+	return r
+
+ignored :: FilePath -> Bool
+ignored = ig . takeFileName
+	where
+	ig ".git" = True
+	ig ".gitignore" = True
+	ig ".gitattributes" = True
+	ig _ = False
+
+type Handler = FilePath -> Maybe FileStatus -> DaemonStatusHandle -> Annex (Maybe Change)
+
+{- 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 :: ThreadState -> DaemonStatusHandle -> ChangeChan -> Handler -> FilePath -> Maybe FileStatus -> IO ()
+runHandler st dstatus changechan handler file filestatus = void $ do
+	r <- tryIO go
+	case r of
+		Left e -> print e
+		Right Nothing -> noop
+		Right (Just change) -> recordChange changechan change
+	where
+		go = runThreadState st $ handler file filestatus dstatus
+
+{- During initial directory scan, this will be run for any regular files
+ - that are already checked into git. We don't want to turn those into
+ - symlinks, so do a check. This is rather expensive, but only happens
+ - during startup.
+ -
+ - It's possible for the file to still be open for write by some process.
+ - This can happen in a few ways; one is if two processes had the file open
+ - and only one has just closed it. We want to avoid adding a file to the
+ - annex that is open for write, to avoid anything being able to change it.
+ -
+ - We could run lsof on the file here to check for other writers.
+ - But, that's slow, and even if there is currently a writer, we will want
+ - to add the file *eventually*. Instead, the file is locked down as a hard
+ - link in a temp directory, with its write bits disabled, for later
+ - checking with lsof, and a Change is returned containing a KeySource
+ - using that hard link. The committer handles running lsof and finishing
+ - the add.
+ -}
+onAdd :: Handler
+onAdd file filestatus dstatus
+	| maybe False isRegularFile filestatus = do
+		ifM (scanComplete <$> getDaemonStatus dstatus)
+			( go
+			, ifM (null <$> inRepo (Git.LsFiles.notInRepo False [file]))
+				( noChange
+				, go
+				)
+			)
+	| otherwise = noChange
+	where
+		go = pendingAddChange =<< Command.Add.lockDown file
+
+{- A symlink might be an arbitrary symlink, which is just added.
+ - Or, if it is a git-annex symlink, ensure it points to the content
+ - before adding it.
+ -}
+onAddSymlink :: Handler
+onAddSymlink file filestatus dstatus = go =<< Backend.lookupFile file
+	where
+		go (Just (key, _)) = do
+			link <- calcGitLink file key
+			ifM ((==) link <$> liftIO (readSymbolicLink file))
+				( ensurestaged link =<< getDaemonStatus dstatus
+				, do
+					liftIO $ removeFile file
+					liftIO $ createSymbolicLink link file
+					addlink link
+				)
+		go Nothing = do -- other symlink
+			link <- liftIO (readSymbolicLink file)
+			ensurestaged link =<< getDaemonStatus dstatus
+
+		{- This is often called on symlinks that are already
+		 - staged correctly. A symlink may have been deleted
+		 - and being re-added, or added when the watcher was
+		 - not running. So they're normally restaged to make sure.
+		 -
+		 - As an optimisation, during the status scan, avoid
+		 - restaging everything. Only links that were created since
+		 - the last time the daemon was running are staged.
+		 - (If the daemon has never ran before, avoid staging
+		 - links too.)
+		 -}
+		ensurestaged link daemonstatus
+			| scanComplete daemonstatus = addlink link
+			| otherwise = case filestatus of
+				Just s
+					| not (afterLastDaemonRun (statusChangeTime s) daemonstatus) -> noChange
+				_ -> addlink link
+
+		{- For speed, tries to reuse the existing blob for
+		 - the symlink target. -}
+		addlink link = do
+			v <- catObjectDetails $ Ref $ ':':file
+			case v of
+				Just (currlink, sha)
+					| s2w8 link == L.unpack currlink ->
+						stageSymlink file sha
+				_ -> do
+					sha <- inRepo $
+						Git.HashObject.hashObject BlobObject link
+					stageSymlink file sha
+			madeChange file LinkChange
+
+onDel :: Handler
+onDel file _ _dstatus = do
+	Annex.Queue.addUpdateIndex =<<
+		inRepo (Git.UpdateIndex.unstageFile file)
+	madeChange file RmChange
+
+{- A directory has been deleted, or moved, so tell git to remove anything
+ - that was inside it from its cache. Since it could reappear at any time,
+ - use --cached to only delete it from the index. 
+ -
+ - Note: This could use unstageFile, but would need to run another git
+ - command to get the recursive list of files in the directory, so rm is
+ - just as good. -}
+onDelDir :: Handler
+onDelDir dir _ _dstatus = do
+	Annex.Queue.addCommand "rm"
+		[Params "--quiet -r --cached --ignore-unmatch --"] [dir]
+	madeChange dir RmDirChange
+
+{- Called when there's an error with inotify. -}
+onErr :: Handler
+onErr msg _ _dstatus = do
+	warning msg
+	return Nothing
+
+{- Adds a symlink to the index, without ever accessing the actual symlink
+ - on disk. This avoids a race if git add is used, where the symlink is
+ - changed to something else immediately after creation.
+ -}
+stageSymlink :: FilePath -> Sha -> Annex ()
+stageSymlink file sha =
+	Annex.Queue.addUpdateIndex =<<
+		inRepo (Git.UpdateIndex.stageSymlink file sha)