1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
|
{- git-annex command
-
- 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
import Command
import Utility.ThreadLock
import qualified Annex
import qualified Annex.Queue
import qualified Command.Add
import qualified Git.Command
import qualified Git.UpdateIndex
import qualified Backend
import Annex.Content
import Control.Concurrent
import Control.Concurrent.STM
import Data.Time.Clock
#if defined linux_HOST_OS
import Utility.Inotify
import System.INotify
#endif
type ChangeChan = TChan UTCTime
def :: [Command]
def = [command "watch" paramPaths seek "watch for changes"]
seek :: [CommandSeek]
seek = [withNothing start]
start :: CommandStart
start = notBareRepo $ do
showStart "watch" "."
watch
stop
watch :: Annex ()
#if defined linux_HOST_OS
watch = do
showAction "scanning"
inRepo $ Git.Command.run "add" [Param "--update"]
withStateMVar $ \st -> liftIO $ withINotify $ \i -> do
changechan <- atomically newTChan
_ <- forkIO $ commitThread st changechan
let hook a = Just $ runHook st changechan a
let hooks = WatchHooks
{ addHook = hook onAdd
, delHook = hook onDel
, addSymlinkHook = hook onAddSymlink
, delDirHook = hook onDelDir
, errHook = hook onErr
}
watchDir i "." (ignored . takeFileName) hooks
putStrLn "(started)"
waitForTermination
#else
watch = error "watch mode is so far only available on Linux"
#endif
ignored :: FilePath -> Bool
ignored ".git" = True
ignored ".gitignore" = True
ignored ".gitattributes" = True
ignored _ = False
{- Stores the Annex state in a MVar, so that threaded actions can access
- it.
-
- Once the action is finished, retrieves the state from the MVar.
-}
withStateMVar :: (MVar Annex.AnnexState -> Annex a) -> Annex a
withStateMVar a = do
state <- Annex.getState id
mvar <- liftIO $ newMVar state
r <- a mvar
newstate <- liftIO $ takeMVar mvar
Annex.changeState (const newstate)
return r
{- Runs an Annex action, using the state from the MVar. -}
runStateMVar :: MVar Annex.AnnexState -> Annex () -> IO ()
runStateMVar mvar a = do
startstate <- takeMVar mvar
!newstate <- Annex.exec startstate a
putMVar mvar newstate
{- Runs a hook, inside the Annex monad.
-
- Exceptions are ignored, otherwise a whole watcher thread could be crashed.
-}
runHook :: MVar Annex.AnnexState -> ChangeChan -> (FilePath -> Annex ()) -> FilePath -> IO ()
runHook st changetimes a f = handle =<< tryIO (runStateMVar st go)
where
go = do
a f
signalChange changetimes
handle (Right ()) = return ()
handle (Left e) = putStrLn $ show e
{- Adding a file is tricky; the file has to be replaced with a symlink
- but this is race prone, as the symlink could be changed immediately
- after creation. To avoid that race, git add is not used to stage the
- symlink. -}
onAdd :: FilePath -> Annex ()
onAdd file = do
showStart "add" file
Command.Add.ingest file >>= go
where
go Nothing = showEndFail
go (Just key) = do
link <- Command.Add.link file key True
stageSymlink file link
showEndOk
{- 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 :: FilePath -> Annex ()
onAddSymlink file = go =<< Backend.lookupFile file
where
go Nothing = addlink =<< liftIO (readSymbolicLink file)
go (Just (key, _)) = do
link <- calcGitLink file key
ifM ((==) link <$> liftIO (readSymbolicLink file))
( addlink link
, do
liftIO $ removeFile file
liftIO $ createSymbolicLink link file
addlink link
)
addlink link = stageSymlink file link
onDel :: FilePath -> Annex ()
onDel file = Annex.Queue.addUpdateIndex =<<
inRepo (Git.UpdateIndex.unstageFile file)
{- 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 :: FilePath -> Annex ()
onDelDir dir = Annex.Queue.addCommand "rm"
[Params "--quiet -r --cached --ignore-unmatch --"] [dir]
{- Called when there's an error with inotify. -}
onErr :: String -> Annex ()
onErr = warning
{- Adds a symlink to the index, without ever accessing the actual symlink
- on disk. -}
stageSymlink :: FilePath -> String -> Annex ()
stageSymlink file linktext =
Annex.Queue.addUpdateIndex =<<
inRepo (Git.UpdateIndex.stageSymlink file linktext)
{- Signals that a change has been made, that needs to get committed. -}
signalChange :: ChangeChan -> Annex ()
signalChange chan = do
liftIO $ (atomically . writeTChan chan) =<< getCurrentTime
-- Just in case the commit thread is not flushing
-- the queue fast enough.
Annex.Queue.flushWhenFull
{- Gets the times of all unhandled changes.
- Blocks until at least one change is made. -}
getChanges :: ChangeChan -> IO [UTCTime]
getChanges chan = atomically $ do
c <- readTChan chan
go [c]
where
go l = do
v <- tryReadTChan chan
case v of
Nothing -> return l
Just c -> go (c:l)
{- Puts unhandled changes back into the channel.
- Note: Original order is not preserved. -}
refillChanges :: ChangeChan -> [UTCTime] -> IO ()
refillChanges chan cs = atomically $ mapM_ (writeTChan chan) cs
{- This thread makes git commits. -}
commitThread :: MVar Annex.AnnexState -> ChangeChan -> IO ()
commitThread st changechan = forever $ do
-- First, a simple rate limiter.
threadDelay oneSecond
-- Next, wait until at least one change has been made.
cs <- getChanges changechan
-- Now see if now's a good time to commit.
time <- getCurrentTime
if shouldCommit time cs
then commit
else refillChanges changechan cs
where
commit = void $ tryIO $ runStateMVar st $ 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"
]
oneSecond = 1000000 -- microseconds
{- 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 commits within the past second,
- a batch activity is taking place, so wait for later.
-}
shouldCommit :: UTCTime -> [UTCTime] -> Bool
shouldCommit now changetimes
| len == 0 = False
| len > 4096 = True -- avoid bloating queue too much
| length (filter thisSecond changetimes) < 10 = True
| otherwise = False -- batch activity
where
len = length changetimes
thisSecond t = now `diffUTCTime` t <= 1
|
