diff options
-rw-r--r-- | Assistant.hs | 6 | ||||
-rw-r--r-- | Assistant/DaemonStatus.hs | 50 | ||||
-rw-r--r-- | Assistant/Threads/Pusher.hs | 3 | ||||
-rw-r--r-- | Assistant/Threads/SanityChecker.hs | 24 | ||||
-rw-r--r-- | Assistant/Threads/TransferWatcher.hs | 9 | ||||
-rw-r--r-- | Assistant/Threads/Transferrer.hs | 4 | ||||
-rw-r--r-- | Assistant/Threads/Watcher.hs | 9 | ||||
-rw-r--r-- | Assistant/Threads/WebApp.hs | 2 | ||||
-rw-r--r-- | Assistant/TransferQueue.hs | 2 |
9 files changed, 49 insertions, 60 deletions
diff --git a/Assistant.hs b/Assistant.hs index ca428988f..6b25c3c6f 100644 --- a/Assistant.hs +++ b/Assistant.hs @@ -65,10 +65,8 @@ - Annex monad in IO actions run by the watcher and committer - threads. Thus, a single state is shared amoung the threads, and - only one at a time can access it. - - DaemonStatusHandle: (MVar) - - The daemon's current status. This MVar should only be manipulated - - from inside the Annex monad, which ensures it's accessed only - - after the ThreadState MVar. + - DaemonStatusHandle: (STM TMVar) + - The daemon's current status. - ChangeChan: (STM TChan) - Changes are indicated by writing to this channel. The committer - reads from it. diff --git a/Assistant/DaemonStatus.hs b/Assistant/DaemonStatus.hs index 84a3662f0..52165138e 100644 --- a/Assistant/DaemonStatus.hs +++ b/Assistant/DaemonStatus.hs @@ -15,7 +15,7 @@ import Utility.NotificationBroadcaster import Logs.Transfer import qualified Command.Sync -import Control.Concurrent +import Control.Concurrent.STM import System.Posix.Types import Data.Time.Clock.POSIX import Data.Time @@ -41,7 +41,8 @@ data DaemonStatus = DaemonStatus type TransferMap = M.Map Transfer TransferInfo -type DaemonStatusHandle = MVar DaemonStatus +{- This TMVar is never left empty, so accessing it will never block. -} +type DaemonStatusHandle = TMVar DaemonStatus newDaemonStatus :: IO DaemonStatus newDaemonStatus = do @@ -56,21 +57,19 @@ newDaemonStatus = do , notificationBroadcaster = nb } -getDaemonStatus :: DaemonStatusHandle -> Annex DaemonStatus -getDaemonStatus = liftIO . readMVar +getDaemonStatus :: DaemonStatusHandle -> IO DaemonStatus +getDaemonStatus = atomically . readTMVar -modifyDaemonStatus_ :: DaemonStatusHandle -> (DaemonStatus -> DaemonStatus) -> Annex () -modifyDaemonStatus_ handle a = do - nb <- liftIO $ modifyMVar handle $ \s -> return - (a s, notificationBroadcaster s) - liftIO $ sendNotification nb +modifyDaemonStatus_ :: DaemonStatusHandle -> (DaemonStatus -> DaemonStatus) -> IO () +modifyDaemonStatus_ handle a = modifyDaemonStatus handle $ \s -> (a s, ()) -modifyDaemonStatus :: DaemonStatusHandle -> (DaemonStatus -> (DaemonStatus, b)) -> Annex b +modifyDaemonStatus :: DaemonStatusHandle -> (DaemonStatus -> (DaemonStatus, b)) -> IO b modifyDaemonStatus handle a = do - (b, nb) <- liftIO $ modifyMVar handle $ \s -> do - let (s', b) = a s - return $ (s', (b, notificationBroadcaster s)) - liftIO $ sendNotification nb + (b, nb) <- atomically $ do + (s, b) <- a <$> takeTMVar handle + putTMVar handle s + return $ (b, notificationBroadcaster s) + sendNotification nb return b {- Updates the cached ordered list of remotes from the list in Annex @@ -78,10 +77,10 @@ modifyDaemonStatus handle a = do updateKnownRemotes :: DaemonStatusHandle -> Annex () updateKnownRemotes dstatus = do remotes <- Command.Sync.syncRemotes [] - modifyDaemonStatus_ dstatus $ + liftIO $ modifyDaemonStatus_ dstatus $ \s -> s { knownRemotes = remotes } -{- Load any previous daemon status file, and store it in the MVar for this +{- Load any previous daemon status file, and store it in a MVar for this - process to use as its DaemonStatus. Also gets current transfer status. -} startDaemonStatus :: Annex DaemonStatusHandle startDaemonStatus = do @@ -90,7 +89,7 @@ startDaemonStatus = do catchDefaultIO (readDaemonStatusFile file) =<< newDaemonStatus transfers <- M.fromList <$> getTransfers remotes <- Command.Sync.syncRemotes [] - liftIO $ newMVar status + liftIO $ atomically $ newTMVar status { scanComplete = False , sanityCheckRunning = False , currentTransfers = transfers @@ -102,18 +101,17 @@ startDaemonStatus = do -} daemonStatusThread :: ThreadState -> DaemonStatusHandle -> IO () daemonStatusThread st handle = do - bhandle <- runThreadState st $ - liftIO . newNotificationHandle - =<< notificationBroadcaster <$> getDaemonStatus handle + bhandle <- newNotificationHandle + =<< notificationBroadcaster <$> getDaemonStatus handle checkpoint runEvery (Seconds tenMinutes) $ do - liftIO $ waitNotification bhandle + waitNotification bhandle checkpoint where - checkpoint = runThreadState st $ do - file <- fromRepo gitAnnexDaemonStatusFile + checkpoint = do status <- getDaemonStatus handle - liftIO $ writeDaemonStatusFile file status + file <- runThreadState st $ fromRepo gitAnnexDaemonStatusFile + writeDaemonStatusFile file status {- Don't just dump out the structure, because it will change over time, - and parts of it are not relevant. -} @@ -167,12 +165,12 @@ tenMinutes :: Int tenMinutes = 10 * 60 {- Mutates the transfer map. -} -adjustTransfers :: DaemonStatusHandle -> (TransferMap -> TransferMap) -> Annex () +adjustTransfers :: DaemonStatusHandle -> (TransferMap -> TransferMap) -> IO () adjustTransfers dstatus a = modifyDaemonStatus_ dstatus $ \s -> s { currentTransfers = a (currentTransfers s) } {- Removes a transfer from the map, and returns its info. -} -removeTransfer :: DaemonStatusHandle -> Transfer -> Annex (Maybe TransferInfo) +removeTransfer :: DaemonStatusHandle -> Transfer -> IO (Maybe TransferInfo) removeTransfer dstatus t = modifyDaemonStatus dstatus go where go s = diff --git a/Assistant/Threads/Pusher.hs b/Assistant/Threads/Pusher.hs index cba53af23..3762c4836 100644 --- a/Assistant/Threads/Pusher.hs +++ b/Assistant/Threads/Pusher.hs @@ -51,8 +51,7 @@ pushThread st daemonstatus commitchan pushmap = do now <- getCurrentTime if shouldPush now commits then do - remotes <- runThreadState st $ - knownRemotes <$> getDaemonStatus daemonstatus + remotes <- knownRemotes <$> getDaemonStatus daemonstatus pushToRemotes thisThread now st (Just pushmap) remotes else do debug thisThread diff --git a/Assistant/Threads/SanityChecker.hs b/Assistant/Threads/SanityChecker.hs index 09aee0797..5e27246a0 100644 --- a/Assistant/Threads/SanityChecker.hs +++ b/Assistant/Threads/SanityChecker.hs @@ -26,32 +26,28 @@ thisThread = "SanityChecker" {- This thread wakes up occasionally to make sure the tree is in good shape. -} sanityCheckerThread :: ThreadState -> DaemonStatusHandle -> TransferQueue -> ChangeChan -> IO () sanityCheckerThread st status transferqueue changechan = forever $ do - waitForNextCheck st status + waitForNextCheck status debug thisThread ["starting sanity check"] - runThreadState st $ - modifyDaemonStatus_ status $ \s -> s - { sanityCheckRunning = True } + modifyDaemonStatus_ status $ \s -> s + { sanityCheckRunning = True } now <- getPOSIXTime -- before check started catchIO (check st status transferqueue changechan) (runThreadState st . warning . show) - runThreadState st $ do - modifyDaemonStatus_ status $ \s -> s - { sanityCheckRunning = False - , lastSanityCheck = Just now - } + modifyDaemonStatus_ status $ \s -> s + { sanityCheckRunning = False + , lastSanityCheck = Just now + } debug thisThread ["sanity check complete"] - {- 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 +waitForNextCheck :: DaemonStatusHandle -> IO () +waitForNextCheck status = do + v <- lastSanityCheck <$> getDaemonStatus status now <- getPOSIXTime threadDelaySeconds $ Seconds $ calcdelay now v where diff --git a/Assistant/Threads/TransferWatcher.hs b/Assistant/Threads/TransferWatcher.hs index be520aaf9..447ff2264 100644 --- a/Assistant/Threads/TransferWatcher.hs +++ b/Assistant/Threads/TransferWatcher.hs @@ -55,12 +55,11 @@ onErr _ _ msg _ = error msg onAdd :: Handler onAdd st dstatus file _ = case parseTransferFile file of Nothing -> noop - Just t -> do - runThreadState st $ go t =<< checkTransfer t + Just t -> go t =<< runThreadState st (checkTransfer t) where go _ Nothing = noop -- transfer already finished go t (Just info) = do - liftIO $ debug thisThread + debug thisThread [ "transfer starting:" , show t ] @@ -71,11 +70,11 @@ onAdd st dstatus file _ = case parseTransferFile file of {- Called when a transfer information file is removed. -} onDel :: Handler -onDel st dstatus file _ = case parseTransferFile file of +onDel _ dstatus file _ = case parseTransferFile file of Nothing -> noop Just t -> do debug thisThread [ "transfer finishing:" , show t ] - void $ runThreadState st $ removeTransfer dstatus t + void $ removeTransfer dstatus t diff --git a/Assistant/Threads/Transferrer.hs b/Assistant/Threads/Transferrer.hs index d8a146948..30802f742 100644 --- a/Assistant/Threads/Transferrer.hs +++ b/Assistant/Threads/Transferrer.hs @@ -48,7 +48,7 @@ transfererThread st dstatus transferqueue slots = go - being uploaded to isn't known to have the file. -} shouldTransfer :: DaemonStatusHandle -> Transfer -> TransferInfo -> Annex Bool shouldTransfer dstatus t info = - go =<< currentTransfers <$> getDaemonStatus dstatus + go =<< currentTransfers <$> liftIO (getDaemonStatus dstatus) where go m | M.member t m = return False @@ -84,7 +84,7 @@ transferThread st dstatus slots t info = case (transferRemote info, associatedFi tid <- inTransferSlot slots st $ transferprocess remote file now <- getCurrentTime - runThreadState st $ adjustTransfers dstatus $ + adjustTransfers dstatus $ M.insertWith' const t info { startedTime = Just $ utcTimeToPOSIXSeconds now , transferTid = Just tid diff --git a/Assistant/Threads/Watcher.hs b/Assistant/Threads/Watcher.hs index 31025361b..ab57bf04a 100644 --- a/Assistant/Threads/Watcher.hs +++ b/Assistant/Threads/Watcher.hs @@ -76,8 +76,7 @@ statupScan st dstatus scanner = do runThreadState st $ showAction "scanning" r <- scanner - runThreadState st $ - modifyDaemonStatus_ dstatus $ \s -> s { scanComplete = True } + modifyDaemonStatus_ dstatus $ \s -> s { scanComplete = True } -- Notice any files that were deleted before watching was started. runThreadState st $ do @@ -132,7 +131,7 @@ runHandler threadname st dstatus transferqueue changechan handler file filestatu onAdd :: Handler onAdd threadname file filestatus dstatus _ | maybe False isRegularFile filestatus = do - ifM (scanComplete <$> getDaemonStatus dstatus) + ifM (scanComplete <$> liftIO (getDaemonStatus dstatus)) ( go , ifM (null <$> inRepo (Git.LsFiles.notInRepo False [file])) ( noChange @@ -156,7 +155,7 @@ onAddSymlink threadname file filestatus dstatus transferqueue = go =<< Backend.l link <- calcGitLink file key ifM ((==) link <$> liftIO (readSymbolicLink file)) ( do - s <- getDaemonStatus dstatus + s <- liftIO $ getDaemonStatus dstatus checkcontent key s ensurestaged link s , do @@ -167,7 +166,7 @@ onAddSymlink threadname file filestatus dstatus transferqueue = go =<< Backend.l ) go Nothing = do -- other symlink link <- liftIO (readSymbolicLink file) - ensurestaged link =<< getDaemonStatus dstatus + ensurestaged link =<< liftIO (getDaemonStatus dstatus) {- This is often called on symlinks that are already - staged correctly. A symlink may have been deleted diff --git a/Assistant/Threads/WebApp.hs b/Assistant/Threads/WebApp.hs index 92f7ff253..6e895ccf6 100644 --- a/Assistant/Threads/WebApp.hs +++ b/Assistant/Threads/WebApp.hs @@ -115,7 +115,7 @@ statusDisplay = do current <- liftIO $ runThreadState (threadState webapp) $ M.toList . currentTransfers - <$> getDaemonStatus (daemonStatus webapp) + <$> liftIO (getDaemonStatus $ daemonStatus webapp) queued <- liftIO $ getTransferQueue $ transferQueue webapp let transfers = current ++ queued diff --git a/Assistant/TransferQueue.hs b/Assistant/TransferQueue.hs index 414a1f9be..2f09813eb 100644 --- a/Assistant/TransferQueue.hs +++ b/Assistant/TransferQueue.hs @@ -59,7 +59,7 @@ stubInfo f r = TransferInfo {- Adds transfers to queue for some of the known remotes. -} queueTransfers :: Schedule -> TransferQueue -> DaemonStatusHandle -> Key -> AssociatedFile -> Direction -> Annex () queueTransfers schedule q daemonstatus k f direction = do - rs <- knownRemotes <$> getDaemonStatus daemonstatus + rs <- knownRemotes <$> liftIO (getDaemonStatus daemonstatus) mapM_ go =<< sufficientremotes rs where sufficientremotes rs |