maintain set of files pendingAdd
Kqueue needs to remember which files failed to be added due to being open, and retry them. This commit gets the data in place for such a retry thread. Broke KeySource out into its own file, and added Eq and Ord instances so it can be stored in a Set.
This commit is contained in:
parent
ad11de94e5
commit
e0fdfb2e70
11 changed files with 89 additions and 30 deletions
|
@ -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
|
||||
|
|
|
@ -7,6 +7,7 @@ module Assistant.Committer where
|
|||
|
||||
import Common.Annex
|
||||
import Assistant.Changes
|
||||
import Assistant.DaemonStatus
|
||||
import Assistant.ThreadedMonad
|
||||
import Assistant.Watcher
|
||||
import qualified Annex
|
||||
|
@ -18,15 +19,15 @@ import qualified Command.Add
|
|||
import Utility.ThreadScheduler
|
||||
import qualified Utility.Lsof as Lsof
|
||||
import qualified Utility.DirWatcher as DirWatcher
|
||||
import Types.Backend
|
||||
import Types.KeySource
|
||||
|
||||
import Data.Time.Clock
|
||||
import Data.Tuple.Utils
|
||||
import qualified Data.Set as S
|
||||
|
||||
{- This thread makes git commits at appropriate times. -}
|
||||
commitThread :: ThreadState -> ChangeChan -> IO ()
|
||||
commitThread st changechan = runEvery (Seconds 1) $ do
|
||||
commitThread :: ThreadState -> DaemonStatusHandle -> ChangeChan -> IO ()
|
||||
commitThread st dstatus 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.
|
||||
cs <- getChanges changechan
|
||||
|
@ -34,7 +35,7 @@ commitThread st changechan = runEvery (Seconds 1) $ do
|
|||
time <- getCurrentTime
|
||||
if shouldCommit time cs
|
||||
then do
|
||||
handleAdds st changechan cs
|
||||
handleAdds st dstatus changechan cs
|
||||
void $ tryIO $ runThreadState st commitStaged
|
||||
else refillChanges changechan cs
|
||||
|
||||
|
@ -79,19 +80,20 @@ shouldCommit now changes
|
|||
-
|
||||
- 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.
|
||||
- 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.
|
||||
-}
|
||||
handleAdds :: ThreadState -> ChangeChan -> [Change] -> IO ()
|
||||
handleAdds st changechan cs
|
||||
handleAdds :: ThreadState -> DaemonStatusHandle -> ChangeChan -> [Change] -> IO ()
|
||||
handleAdds st dstatus changechan cs
|
||||
| null toadd = noop
|
||||
| otherwise = do
|
||||
toadd' <- safeToAdd st toadd
|
||||
toadd' <- safeToAdd st dstatus toadd
|
||||
unless (null toadd') $ do
|
||||
added <- filter id <$> forM toadd' add
|
||||
unless (DirWatcher.eventsCoalesce || null added) $
|
||||
handleAdds st changechan
|
||||
handleAdds st dstatus changechan
|
||||
=<< getChanges changechan
|
||||
where
|
||||
toadd = map changeFile $ filter isPendingAdd cs
|
||||
|
@ -122,8 +124,8 @@ handleAdds st changechan cs
|
|||
- opened for write, so lsof is run on the temp directory
|
||||
- to check them.
|
||||
-}
|
||||
safeToAdd :: ThreadState -> [FilePath] -> IO [KeySource]
|
||||
safeToAdd st files = do
|
||||
safeToAdd :: ThreadState -> DaemonStatusHandle -> [FilePath] -> IO [KeySource]
|
||||
safeToAdd st dstatus files = do
|
||||
locked <- catMaybes <$> lockdown files
|
||||
runThreadState st $ ifM (Annex.getState Annex.force)
|
||||
( return locked -- force bypasses lsof check
|
||||
|
@ -134,16 +136,33 @@ safeToAdd st files = do
|
|||
catMaybes <$> forM locked (go open)
|
||||
)
|
||||
where
|
||||
{- When a file is still open, it can be put into pendingAdd
|
||||
- to be checked again later. However when closingTracked
|
||||
- is supported, another event will be received once it's
|
||||
- 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"
|
||||
-- remove the hard link
|
||||
--_ <- liftIO $ tryIO $
|
||||
-- removeFile $ contentLocation keysource
|
||||
void $ liftIO $ canceladd keysource
|
||||
else void $ addpending keysource
|
||||
return Nothing
|
||||
| otherwise = return $ Just keysource
|
||||
|
||||
canceladd keysource = tryIO $
|
||||
-- remove the hard link
|
||||
removeFile $ contentLocation keysource
|
||||
|
||||
{- The same file (or a file with the same name)
|
||||
- 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
|
||||
|
|
|
@ -9,12 +9,14 @@ 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
|
||||
|
@ -25,6 +27,8 @@ 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)
|
||||
|
||||
|
@ -36,13 +40,17 @@ 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 = liftIO $ modifyMVar_ handle (return . a)
|
||||
modifyDaemonStatus handle a = modifyDaemonStatusM 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. -}
|
||||
|
|
|
@ -6,7 +6,6 @@
|
|||
-}
|
||||
|
||||
module Backend (
|
||||
B.KeySource(..),
|
||||
list,
|
||||
orderedList,
|
||||
genKey,
|
||||
|
@ -23,6 +22,7 @@ import Config
|
|||
import qualified Annex
|
||||
import Annex.CheckAttr
|
||||
import Types.Key
|
||||
import Types.KeySource
|
||||
import qualified Types.Backend as B
|
||||
|
||||
-- When adding a new backend, import it here and add it to the list.
|
||||
|
@ -54,12 +54,12 @@ orderedList = do
|
|||
{- Generates a key for a file, trying each backend in turn until one
|
||||
- accepts it.
|
||||
-}
|
||||
genKey :: B.KeySource -> Maybe Backend -> Annex (Maybe (Key, Backend))
|
||||
genKey :: KeySource -> Maybe Backend -> Annex (Maybe (Key, Backend))
|
||||
genKey source trybackend = do
|
||||
bs <- orderedList
|
||||
let bs' = maybe bs (: bs) trybackend
|
||||
genKey' bs' source
|
||||
genKey' :: [Backend] -> B.KeySource -> Annex (Maybe (Key, Backend))
|
||||
genKey' :: [Backend] -> KeySource -> Annex (Maybe (Key, Backend))
|
||||
genKey' [] _ = return Nothing
|
||||
genKey' (b:bs) source = do
|
||||
r <- B.getKey b source
|
||||
|
|
|
@ -11,6 +11,7 @@ import Common.Annex
|
|||
import qualified Annex
|
||||
import Types.Backend
|
||||
import Types.Key
|
||||
import Types.KeySource
|
||||
import qualified Build.SysConfig as SysConfig
|
||||
|
||||
type SHASize = Int
|
||||
|
|
|
@ -10,6 +10,7 @@ module Backend.WORM (backends) where
|
|||
import Common.Annex
|
||||
import Types.Backend
|
||||
import Types.Key
|
||||
import Types.KeySource
|
||||
|
||||
backends :: [Backend]
|
||||
backends = [backend]
|
||||
|
|
|
@ -12,6 +12,7 @@ import Annex.Exception
|
|||
import Command
|
||||
import qualified Annex
|
||||
import qualified Annex.Queue
|
||||
import Types.KeySource
|
||||
import Backend
|
||||
import Logs.Location
|
||||
import Annex.Content
|
||||
|
|
|
@ -20,6 +20,7 @@ import Annex.Content
|
|||
import Logs.Web
|
||||
import qualified Option
|
||||
import Types.Key
|
||||
import Types.KeySource
|
||||
import Config
|
||||
|
||||
def :: [Command]
|
||||
|
|
|
@ -11,6 +11,7 @@ import Common.Annex
|
|||
import Command
|
||||
import Backend
|
||||
import qualified Types.Key
|
||||
import Types.KeySource
|
||||
import Annex.Content
|
||||
import qualified Command.ReKey
|
||||
|
||||
|
|
|
@ -10,13 +10,7 @@
|
|||
module Types.Backend where
|
||||
|
||||
import Types.Key
|
||||
|
||||
{- The source used to generate a key. The location of the content
|
||||
- may be different from the filename associated with the key. -}
|
||||
data KeySource = KeySource
|
||||
{ keyFilename :: FilePath
|
||||
, contentLocation :: FilePath
|
||||
}
|
||||
import Types.KeySource
|
||||
|
||||
data BackendA a = Backend
|
||||
{ name :: String
|
||||
|
|
33
Types/KeySource.hs
Normal file
33
Types/KeySource.hs
Normal file
|
@ -0,0 +1,33 @@
|
|||
{- KeySource data type
|
||||
-
|
||||
- Copyright 2012 Joey Hess <joey@kitenet.net>
|
||||
-
|
||||
- Licensed under the GNU GPL version 3 or higher.
|
||||
-}
|
||||
|
||||
module Types.KeySource where
|
||||
|
||||
import Data.Ord
|
||||
|
||||
{- When content is in the process of being added to the annex,
|
||||
- and a Key generated from it, this data type is used.
|
||||
-
|
||||
- The contentLocation may be different from the filename
|
||||
- associated with the key. For example, the add command
|
||||
- temporarily puts the content into a lockdown directory
|
||||
- for checking. The migrate command uses the content
|
||||
- of a different Key. -}
|
||||
data KeySource = KeySource
|
||||
{ keyFilename :: FilePath
|
||||
, contentLocation :: FilePath
|
||||
}
|
||||
deriving (Show)
|
||||
|
||||
{- KeySources are assumed to be equal when the same filename is associated
|
||||
- with the key. The contentLocation can be a random temp file.
|
||||
-}
|
||||
instance Eq KeySource where
|
||||
x == y = keyFilename x == keyFilename y
|
||||
|
||||
instance Ord KeySource where
|
||||
compare = comparing keyFilename
|
Loading…
Reference in a new issue