pending adds now retried for kqueue

Rethought how to keep track of pending adds that need to be retried later.
The commit thread already run up every second when there are changes,
so let's keep pending adds queued as changes until they're safe to add.

Also, the committer is now smarter about avoiding empty commits when
all the adds are currently unsafe, or in the rare case that an add event
for a symlink is not received in time. It may avoid them entirely.

This seems to work as before for inotify, and is untested for kqueue.

(Actually commit batching seems to be improved for inotify, although I'm
not sure why. I'm seeing only two commits made during large batch
operations, and the first of those is the non-batch mode commit.)

Assistant.hs

@@ -75,8 +75,8 @@ startDaemon foreground
 				-- begin adding files and having them
 				-- committed, even while the startup scan
 				-- is taking place.
+				_ <- forkIO $ commitThread st changechan
 				_ <- forkIO $ daemonStatusThread st dstatus
-				_ <- forkIO $ commitThread st dstatus changechan
 				_ <- forkIO $ sanityCheckerThread st dstatus changechan
 				-- Does not return.
 				watchThread st dstatus changechan

Assistant/Changes.hs

@@ -7,20 +7,26 @@ module Assistant.Changes where
 
 import Common.Annex
 import qualified Annex.Queue
+import Types.KeySource
 
 import Control.Concurrent.STM
 import Data.Time.Clock
 
-data ChangeType = PendingAddChange | LinkChange | RmChange | RmDirChange
+data ChangeType = AddChange | LinkChange | RmChange | RmDirChange
 	deriving (Show, Eq)
 
 type ChangeChan = TChan Change
 
-data Change = Change
+data Change
-	{ changeTime :: UTCTime
+	= Change 
-	, changeFile :: FilePath
+		{ changeTime :: UTCTime
-	, changeType :: ChangeType
+		, changeFile :: FilePath
-	}
+		, changeType :: ChangeType
+		}
+	| PendingAddChange
+		{ changeTime ::UTCTime
+		, keySource :: KeySource
+		}
 	deriving (Show)
 
 runChangeChan :: STM a -> IO a
@@ -33,13 +39,29 @@ newChangeChan = atomically newTChan
 madeChange :: FilePath -> ChangeType -> Annex (Maybe Change)
 madeChange f t = do
 	-- Just in case the commit thread is not flushing the queue fast enough.
-	when (t /= PendingAddChange) $
+	Annex.Queue.flushWhenFull
-		Annex.Queue.flushWhenFull
 	liftIO $ Just <$> (Change <$> getCurrentTime <*> pure f <*> pure t)
 
 noChange :: Annex (Maybe Change)
 noChange = return Nothing
 
+{- Indicates an add is in progress. -}
+pendingAddChange :: KeySource -> Annex (Maybe Change)
+pendingAddChange ks =
+	liftIO $ Just <$> (PendingAddChange <$> getCurrentTime <*> pure ks)
+
+isPendingAddChange :: Change -> Bool
+isPendingAddChange (PendingAddChange {}) = True
+isPendingAddChange _ = False
+
+finishedChange :: Change -> Change
+finishedChange c@(PendingAddChange { keySource = ks }) = Change
+	{ changeTime = changeTime c
+	, changeFile = keyFilename ks
+	, changeType = AddChange
+	}
+finishedChange c = c
+
 {- Gets all unhandled changes.
  - Blocks until at least one change is made. -}
 getChanges :: ChangeChan -> IO [Change]

Assistant/Committer.hs

@@ -7,7 +7,6 @@ module Assistant.Committer where
 
 import Common.Annex
 import Assistant.Changes
-import Assistant.DaemonStatus
 import Assistant.ThreadedMonad
 import Assistant.Watcher
 import qualified Annex
@@ -24,20 +23,25 @@ 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 -> DaemonStatusHandle -> ChangeChan -> IO ()
+commitThread :: ThreadState -> ChangeChan -> IO ()
-commitThread st dstatus changechan = runEvery (Seconds 1) $ do
+commitThread st changechan = runEvery (Seconds 1) $ do
 	-- We already waited one second as a simple rate limiter.
-	-- Next, wait until at least one change has been made.
+	-- Next, wait until at least one change is available for
-	cs <- getChanges changechan
+	-- processing.
+	changes <- getChanges changechan
 	-- Now see if now's a good time to commit.
 	time <- getCurrentTime
-	if shouldCommit time cs
+	if shouldCommit time changes
 		then do
-			handleAdds st dstatus changechan cs
+			readychanges <- handleAdds st changechan changes
-			void $ tryIO $ runThreadState st commitStaged
+			if shouldCommit time readychanges
-		else refillChanges changechan cs
+				then do
+					void $ tryIO $ runThreadState st commitStaged
+				else refillChanges changechan readychanges
+		else refillChanges changechan changes
 
 commitStaged :: Annex ()
 commitStaged = do
@@ -83,95 +87,99 @@ shouldCommit now changes
  - staged before returning, and will be committed immediately.
  -
  - OTOH, for kqueue, eventsCoalesce, so instead the symlink is directly
- - created and staged, if the file is not open.
+ - 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 -> DaemonStatusHandle -> ChangeChan -> [Change] -> IO ()
+handleAdds :: ThreadState -> ChangeChan -> [Change] -> IO [Change]
-handleAdds st dstatus changechan cs
+handleAdds st changechan cs = returnWhen (null pendingadds) $ do
-	| null toadd = noop
+	(postponed, toadd) <- partitionEithers <$>
-	| otherwise = do
+		safeToAdd st pendingadds
-		toadd' <- safeToAdd st dstatus toadd
+
-		unless (null toadd') $ do
+	unless (null postponed) $
-			added <- filter id <$> forM toadd' add
+		refillChanges changechan postponed
-			unless (DirWatcher.eventsCoalesce || null added) $
+
-				handleAdds st dstatus changechan
+	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
-		toadd = map changeFile $ filter isPendingAdd cs
+		(pendingadds, otherchanges) = partition isPendingAddChange cs
 
-		isPendingAdd (Change { changeType = PendingAddChange }) = True
+		returnWhen c a
-		isPendingAdd _ = False
+			| c = return otherchanges
+			| otherwise = a
 
-		add keysource = catchBoolIO $ runThreadState st $ do
+		add :: Change -> IO (Maybe Change)
-			showStart "add" $ keyFilename keysource
+		add change@(PendingAddChange { keySource = ks }) = do
-			handle (keyFilename keysource)
+			r <- catchMaybeIO $ runThreadState st $ do
-				=<< Command.Add.ingest keysource
+				showStart "add" $ keyFilename ks
+				handle (finishedChange change) (keyFilename ks)
+					=<< Command.Add.ingest ks
+			return $ maybeMaybe r
+		add _ = return Nothing
 
-		handle _ Nothing = do
+		maybeMaybe (Just j@(Just _)) = j
+		maybeMaybe _ = Nothing
+
+		handle _ _ Nothing = do
 			showEndFail
-			return False
+			return Nothing
-		handle file (Just key) = do
+		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 True
+			return $ Just change
 
-{- Checks which of a set of files can safely be added.
+{- PendingAddChanges can Either be Right to be added now,
- - Files are locked down as hard links in a temp directory,
+ - or are unsafe, and must be Left for later.
- - with their write bits disabled. But some may still be
+ -
- - opened for write, so lsof is run on the temp directory
+ - Check by running lsof on the temp directory, which
- - to check them.
+ - the KeySources are locked down in.
  -}
-safeToAdd :: ThreadState -> DaemonStatusHandle -> [FilePath] -> IO [KeySource]
+safeToAdd :: ThreadState -> [Change] -> IO [Either Change Change]
-safeToAdd st dstatus files = do
+safeToAdd st changes = runThreadState st $
-	locked <- catMaybes <$> lockdown files
+	ifM (Annex.getState Annex.force)
-	runThreadState st $ ifM (Annex.getState Annex.force)
+		( allRight changes -- force bypasses lsof check
-		( return locked -- force bypasses lsof check
 		, do
 			tmpdir <- fromRepo gitAnnexTmpDir
-			open <- S.fromList . map fst3 . filter openwrite <$>
+			openfiles <- S.fromList . map fst3 . filter openwrite <$>
 				liftIO (Lsof.queryDir tmpdir)
-			catMaybes <$> forM locked (go open)
+			
+			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
- 		{- When a file is still open, it can be put into pendingAdd
+		check openfiles change@(PendingAddChange { keySource = ks })
-		 - to be checked again later. However when closingTracked
+			| S.member (contentLocation ks) openfiles = Left change
-		 - is supported, another event will be received once it's
+		check _ change = Right change
-		 - closed, so there's no point in doing so. -}
-		go open keysource
-			| S.member (contentLocation keysource) open = do
-				if DirWatcher.closingTracked
-					then do
-						warning $ keyFilename keysource
-							++ " still has writers, not adding"
-						void $ liftIO $ canceladd keysource
-					else void $ addpending keysource
-				return Nothing
-			| otherwise = return $ Just keysource
 
-		canceladd keysource = tryIO $
+		canceladd (PendingAddChange { keySource = ks }) = do
+			warning $ keyFilename ks
+				++ " still has writers, not adding"
 			-- remove the hard link
-			removeFile $ contentLocation keysource
+			void $ liftIO $ tryIO $
-
+				removeFile $ contentLocation ks
-		{- The same file (or a file with the same name)
+		canceladd _ = noop
-		 - could already be pending add; if so this KeySource
-		 - superscedes the old one. -}
-		addpending keysource = modifyDaemonStatusM dstatus $ \s -> do
-			let set = pendingAdd s
-			mapM_ canceladd $ S.toList $ S.filter (== keysource) set
-			return $ s { pendingAdd = S.insert keysource set }
-
-		lockdown = mapM $ \file -> do
-			ms <- catchMaybeIO $ getSymbolicLinkStatus file
-			case ms of
-				Just s
-					| isRegularFile s -> 
-						catchMaybeIO $ runThreadState st $
-							Command.Add.lockDown file
-				_ -> return Nothing
-	
 
 		openwrite (_file, mode, _pid) =
 			mode == Lsof.OpenWriteOnly || mode == Lsof.OpenReadWrite
+
+		allRight = return . map Right

Assistant/DaemonStatus.hs

@@ -9,14 +9,12 @@ import Common.Annex
 import Assistant.ThreadedMonad
 import Utility.ThreadScheduler
 import Utility.TempFile
-import Types.KeySource
 
 import Control.Concurrent
 import System.Posix.Types
 import Data.Time.Clock.POSIX
 import Data.Time
 import System.Locale
-import qualified Data.Set as S
 
 data DaemonStatus = DaemonStatus
 	-- False when the daemon is performing its startup scan
@@ -27,8 +25,6 @@ data DaemonStatus = DaemonStatus
 	, sanityCheckRunning :: Bool
 	-- Last time the sanity checker ran
 	, lastSanityCheck :: Maybe POSIXTime
-	-- Files that are in the process of being added to the annex.
-	, pendingAdd :: S.Set KeySource
 	}
 	deriving (Show)
 
@@ -40,17 +36,13 @@ newDaemonStatus = DaemonStatus
 	, lastRunning = Nothing
 	, sanityCheckRunning = False
 	, lastSanityCheck = Nothing
-	, pendingAdd = S.empty
 	}
 
 getDaemonStatus :: DaemonStatusHandle -> Annex DaemonStatus
 getDaemonStatus = liftIO . readMVar
 
 modifyDaemonStatus :: DaemonStatusHandle -> (DaemonStatus -> DaemonStatus) -> Annex ()
-modifyDaemonStatus handle a = modifyDaemonStatusM handle (return . a)
+modifyDaemonStatus handle a = liftIO $ modifyMVar_ handle (return . a)
-
-modifyDaemonStatusM :: DaemonStatusHandle -> (DaemonStatus -> IO DaemonStatus) -> Annex ()
-modifyDaemonStatusM handle a = liftIO $ modifyMVar_ handle a
 
 {- Load any previous daemon status file, and store it in the MVar for this
  - process to use as its DaemonStatus. -}

Assistant/Watcher.hs

@@ -15,13 +15,14 @@ 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 Annex
+import qualified Command.Add
 import Annex.Content
 import Annex.CatFile
 import Git.Types
@@ -110,22 +111,27 @@ runHandler st dstatus changechan handler file filestatus = void $ do
  - 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 writer.
+ - We could run lsof on the file here to check for other writers.
- - But, that's slow. Instead, a Change is returned that indicates this file
+ - But, that's slow, and even if there is currently a writer, we will want
- - still needs to be added. The committer will handle bundles of these
+ - to add the file *eventually*. Instead, the file is locked down as a hard
- - Changes at once.
+ - 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 = do
+onAdd file filestatus dstatus
-	ifM (scanComplete <$> getDaemonStatus dstatus)
+	| maybe False isRegularFile filestatus = do
-		( go
+		ifM (scanComplete <$> getDaemonStatus dstatus)
-		, ifM (null <$> inRepo (Git.LsFiles.notInRepo False [file]))
+			( go
-			( noChange
+			, ifM (null <$> inRepo (Git.LsFiles.notInRepo False [file]))
-			, go
+				( noChange
+				, go
+				)
 			)
-		)
+	| otherwise = noChange
 	where
-		go = madeChange file PendingAddChange
+		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