add a NotificationBroadcaster to DaemonStatus
First use of it is to make the status checkpointer thread block until there is really a change to the status.
This commit is contained in:
parent
ca478b7bcb
commit
a17fde22fa
2 changed files with 50 additions and 30 deletions
|
@ -11,6 +11,7 @@ import Common.Annex
|
||||||
import Assistant.ThreadedMonad
|
import Assistant.ThreadedMonad
|
||||||
import Utility.ThreadScheduler
|
import Utility.ThreadScheduler
|
||||||
import Utility.TempFile
|
import Utility.TempFile
|
||||||
|
import Utility.NotificationBroadcaster
|
||||||
import Logs.Transfer
|
import Logs.Transfer
|
||||||
import qualified Command.Sync
|
import qualified Command.Sync
|
||||||
|
|
||||||
|
@ -34,31 +35,43 @@ data DaemonStatus = DaemonStatus
|
||||||
, currentTransfers :: TransferMap
|
, currentTransfers :: TransferMap
|
||||||
-- Ordered list of remotes to talk to.
|
-- Ordered list of remotes to talk to.
|
||||||
, knownRemotes :: [Remote]
|
, knownRemotes :: [Remote]
|
||||||
|
-- Clients can use this to wait on changes to the DaemonStatus
|
||||||
|
, notificationBroadcaster :: NotificationBroadcaster
|
||||||
}
|
}
|
||||||
deriving (Show)
|
|
||||||
|
|
||||||
type TransferMap = M.Map Transfer TransferInfo
|
type TransferMap = M.Map Transfer TransferInfo
|
||||||
|
|
||||||
type DaemonStatusHandle = MVar DaemonStatus
|
type DaemonStatusHandle = MVar DaemonStatus
|
||||||
|
|
||||||
newDaemonStatus :: DaemonStatus
|
newDaemonStatus :: IO DaemonStatus
|
||||||
newDaemonStatus = DaemonStatus
|
newDaemonStatus = do
|
||||||
|
nb <- newNotificationBroadcaster
|
||||||
|
return $ DaemonStatus
|
||||||
{ scanComplete = False
|
{ scanComplete = False
|
||||||
, lastRunning = Nothing
|
, lastRunning = Nothing
|
||||||
, sanityCheckRunning = False
|
, sanityCheckRunning = False
|
||||||
, lastSanityCheck = Nothing
|
, lastSanityCheck = Nothing
|
||||||
, currentTransfers = M.empty
|
, currentTransfers = M.empty
|
||||||
, knownRemotes = []
|
, knownRemotes = []
|
||||||
|
, notificationBroadcaster = nb
|
||||||
}
|
}
|
||||||
|
|
||||||
getDaemonStatus :: DaemonStatusHandle -> Annex DaemonStatus
|
getDaemonStatus :: DaemonStatusHandle -> Annex DaemonStatus
|
||||||
getDaemonStatus = liftIO . readMVar
|
getDaemonStatus = liftIO . readMVar
|
||||||
|
|
||||||
modifyDaemonStatus_ :: DaemonStatusHandle -> (DaemonStatus -> DaemonStatus) -> Annex ()
|
modifyDaemonStatus_ :: DaemonStatusHandle -> (DaemonStatus -> DaemonStatus) -> Annex ()
|
||||||
modifyDaemonStatus_ handle a = liftIO $ modifyMVar_ handle (return . a)
|
modifyDaemonStatus_ handle a = do
|
||||||
|
nb <- liftIO $ modifyMVar handle $ \s -> return
|
||||||
|
(a s, notificationBroadcaster s)
|
||||||
|
liftIO $ sendNotification nb
|
||||||
|
|
||||||
modifyDaemonStatus :: DaemonStatusHandle -> (DaemonStatus -> (DaemonStatus, b)) -> Annex b
|
modifyDaemonStatus :: DaemonStatusHandle -> (DaemonStatus -> (DaemonStatus, b)) -> Annex b
|
||||||
modifyDaemonStatus handle a = liftIO $ modifyMVar handle (return . a)
|
modifyDaemonStatus handle a = do
|
||||||
|
(b, nb) <- liftIO $ modifyMVar handle $ \s -> do
|
||||||
|
let (s', b) = a s
|
||||||
|
return $ (s', (b, notificationBroadcaster s))
|
||||||
|
liftIO $ sendNotification nb
|
||||||
|
return b
|
||||||
|
|
||||||
{- Updates the cached ordered list of remotes from the list in Annex
|
{- Updates the cached ordered list of remotes from the list in Annex
|
||||||
- state. -}
|
- state. -}
|
||||||
|
@ -74,7 +87,7 @@ startDaemonStatus :: Annex DaemonStatusHandle
|
||||||
startDaemonStatus = do
|
startDaemonStatus = do
|
||||||
file <- fromRepo gitAnnexDaemonStatusFile
|
file <- fromRepo gitAnnexDaemonStatusFile
|
||||||
status <- liftIO $
|
status <- liftIO $
|
||||||
catchDefaultIO (readDaemonStatusFile file) newDaemonStatus
|
catchDefaultIO (readDaemonStatusFile file) =<< newDaemonStatus
|
||||||
transfers <- M.fromList <$> getTransfers
|
transfers <- M.fromList <$> getTransfers
|
||||||
remotes <- Command.Sync.syncRemotes []
|
remotes <- Command.Sync.syncRemotes []
|
||||||
liftIO $ newMVar status
|
liftIO $ newMVar status
|
||||||
|
@ -84,11 +97,18 @@ startDaemonStatus = do
|
||||||
, knownRemotes = remotes
|
, knownRemotes = remotes
|
||||||
}
|
}
|
||||||
|
|
||||||
{- This thread wakes up periodically and writes the daemon status to disk. -}
|
{- This writes the daemon status to disk, when it changes, but no more
|
||||||
|
- frequently than once every ten minutes.
|
||||||
|
-}
|
||||||
daemonStatusThread :: ThreadState -> DaemonStatusHandle -> IO ()
|
daemonStatusThread :: ThreadState -> DaemonStatusHandle -> IO ()
|
||||||
daemonStatusThread st handle = do
|
daemonStatusThread st handle = do
|
||||||
|
bhandle <- runThreadState st $
|
||||||
|
liftIO . newNotificationHandle
|
||||||
|
=<< notificationBroadcaster <$> getDaemonStatus handle
|
||||||
|
checkpoint
|
||||||
|
runEvery (Seconds tenMinutes) $ do
|
||||||
|
liftIO $ waitNotification bhandle
|
||||||
checkpoint
|
checkpoint
|
||||||
runEvery (Seconds tenMinutes) checkpoint
|
|
||||||
where
|
where
|
||||||
checkpoint = runThreadState st $ do
|
checkpoint = runThreadState st $ do
|
||||||
file <- fromRepo gitAnnexDaemonStatusFile
|
file <- fromRepo gitAnnexDaemonStatusFile
|
||||||
|
@ -109,9 +129,9 @@ writeDaemonStatusFile file status =
|
||||||
]
|
]
|
||||||
|
|
||||||
readDaemonStatusFile :: FilePath -> IO DaemonStatus
|
readDaemonStatusFile :: FilePath -> IO DaemonStatus
|
||||||
readDaemonStatusFile file = parse <$> readFile file
|
readDaemonStatusFile file = parse <$> newDaemonStatus <*> readFile file
|
||||||
where
|
where
|
||||||
parse = foldr parseline newDaemonStatus . lines
|
parse status = foldr parseline status . lines
|
||||||
parseline line status
|
parseline line status
|
||||||
| key == "lastRunning" = parseval readtime $ \v ->
|
| key == "lastRunning" = parseval readtime $ \v ->
|
||||||
status { lastRunning = Just v }
|
status { lastRunning = Just v }
|
||||||
|
|
|
@ -11,10 +11,10 @@
|
||||||
- Licensed under the GNU GPL version 3 or higher.
|
- Licensed under the GNU GPL version 3 or higher.
|
||||||
-}
|
-}
|
||||||
|
|
||||||
module Assistant.NotificationBroadCaster (
|
module Utility.NotificationBroadcaster (
|
||||||
NotificationBroadCaster,
|
NotificationBroadcaster,
|
||||||
NotificationHandle,
|
NotificationHandle,
|
||||||
newNotificationBroadCaster,
|
newNotificationBroadcaster,
|
||||||
newNotificationHandle,
|
newNotificationHandle,
|
||||||
notificationHandleToId,
|
notificationHandleToId,
|
||||||
notificationHandleFromId,
|
notificationHandleFromId,
|
||||||
|
@ -28,21 +28,21 @@ import Control.Concurrent.STM
|
||||||
import Control.Concurrent.SampleVar
|
import Control.Concurrent.SampleVar
|
||||||
|
|
||||||
{- One SampleVar per client. The TMVar is never empty, so never blocks. -}
|
{- One SampleVar per client. The TMVar is never empty, so never blocks. -}
|
||||||
type NotificationBroadCaster = TMVar [SampleVar ()]
|
type NotificationBroadcaster = TMVar [SampleVar ()]
|
||||||
|
|
||||||
{- Handle given out to an individual client. -}
|
{- Handle given out to an individual client. -}
|
||||||
data NotificationHandle = NotificationHandle NotificationBroadCaster Int
|
data NotificationHandle = NotificationHandle NotificationBroadcaster Int
|
||||||
|
|
||||||
newNotificationBroadCaster :: IO NotificationBroadCaster
|
newNotificationBroadcaster :: IO NotificationBroadcaster
|
||||||
newNotificationBroadCaster = atomically (newTMVar [])
|
newNotificationBroadcaster = atomically (newTMVar [])
|
||||||
|
|
||||||
{- Allocates a notification handle for a client to use. -}
|
{- Allocates a notification handle for a client to use. -}
|
||||||
newNotificationHandle :: NotificationBroadCaster -> IO NotificationHandle
|
newNotificationHandle :: NotificationBroadcaster -> IO NotificationHandle
|
||||||
newNotificationHandle b = NotificationHandle
|
newNotificationHandle b = NotificationHandle
|
||||||
<$> pure b
|
<$> pure b
|
||||||
<*> addclient b
|
<*> addclient
|
||||||
where
|
where
|
||||||
addclient b = do
|
addclient = do
|
||||||
s <- newEmptySampleVar
|
s <- newEmptySampleVar
|
||||||
atomically $ do
|
atomically $ do
|
||||||
l <- readTMVar b
|
l <- readTMVar b
|
||||||
|
@ -54,13 +54,13 @@ newNotificationHandle b = NotificationHandle
|
||||||
notificationHandleToId :: NotificationHandle -> Int
|
notificationHandleToId :: NotificationHandle -> Int
|
||||||
notificationHandleToId (NotificationHandle _ i) = i
|
notificationHandleToId (NotificationHandle _ i) = i
|
||||||
|
|
||||||
{- Given a NotificationBroadCaster, and an Int identifier, recreates the
|
{- Given a NotificationBroadcaster, and an Int identifier, recreates the
|
||||||
- NotificationHandle. -}
|
- NotificationHandle. -}
|
||||||
notificationHandleFromId :: NotificationBroadCaster -> Int -> NotificationHandle
|
notificationHandleFromId :: NotificationBroadcaster -> Int -> NotificationHandle
|
||||||
notificationHandleFromId = NotificationHandle
|
notificationHandleFromId = NotificationHandle
|
||||||
|
|
||||||
{- Sends a notification to all clients. -}
|
{- Sends a notification to all clients. -}
|
||||||
sendNotification :: NotificationBroadCaster -> IO ()
|
sendNotification :: NotificationBroadcaster -> IO ()
|
||||||
sendNotification b = do
|
sendNotification b = do
|
||||||
l <- atomically $ readTMVar b
|
l <- atomically $ readTMVar b
|
||||||
mapM_ notify l
|
mapM_ notify l
|
||||||
|
|
Loading…
Add table
Reference in a new issue