pushed Assistant monad down into DaemonStatus code
Currently have three old versions of functions that more reworking is needed to remove: getDaemonStatusOld, modifyDaemonStatusOld_, and modifyDaemonStatusOld
This commit is contained in:
parent
ea8df8fe9f
commit
47d94eb9a4
20 changed files with 141 additions and 152 deletions
|
@ -31,11 +31,16 @@ getDaemonStatusOld = atomically . readTMVar
|
||||||
getDaemonStatus :: Assistant DaemonStatus
|
getDaemonStatus :: Assistant DaemonStatus
|
||||||
getDaemonStatus = (atomically . readTMVar) <<~ daemonStatusHandle
|
getDaemonStatus = (atomically . readTMVar) <<~ daemonStatusHandle
|
||||||
|
|
||||||
modifyDaemonStatus_ :: DaemonStatusHandle -> (DaemonStatus -> DaemonStatus) -> IO ()
|
-- TODO remove this
|
||||||
modifyDaemonStatus_ dstatus a = modifyDaemonStatus dstatus $ \s -> (a s, ())
|
modifyDaemonStatusOld_ :: DaemonStatusHandle -> (DaemonStatus -> DaemonStatus) -> IO ()
|
||||||
|
modifyDaemonStatusOld_ dstatus a = modifyDaemonStatusOld dstatus $ \s -> (a s, ())
|
||||||
|
|
||||||
modifyDaemonStatus :: DaemonStatusHandle -> (DaemonStatus -> (DaemonStatus, b)) -> IO b
|
modifyDaemonStatus_ :: (DaemonStatus -> DaemonStatus) -> Assistant ()
|
||||||
modifyDaemonStatus dstatus a = do
|
modifyDaemonStatus_ a = modifyDaemonStatus $ \s -> (a s, ())
|
||||||
|
|
||||||
|
-- TODO remove this
|
||||||
|
modifyDaemonStatusOld :: DaemonStatusHandle -> (DaemonStatus -> (DaemonStatus, b)) -> IO b
|
||||||
|
modifyDaemonStatusOld dstatus a = do
|
||||||
(s, b) <- atomically $ do
|
(s, b) <- atomically $ do
|
||||||
r@(s, _) <- a <$> takeTMVar dstatus
|
r@(s, _) <- a <$> takeTMVar dstatus
|
||||||
putTMVar dstatus s
|
putTMVar dstatus s
|
||||||
|
@ -43,6 +48,17 @@ modifyDaemonStatus dstatus a = do
|
||||||
sendNotification $ changeNotifier s
|
sendNotification $ changeNotifier s
|
||||||
return b
|
return b
|
||||||
|
|
||||||
|
modifyDaemonStatus :: (DaemonStatus -> (DaemonStatus, b)) -> Assistant b
|
||||||
|
modifyDaemonStatus a = do
|
||||||
|
dstatus <- getAssistant daemonStatusHandle
|
||||||
|
liftIO $ do
|
||||||
|
(s, b) <- atomically $ do
|
||||||
|
r@(s, _) <- a <$> takeTMVar dstatus
|
||||||
|
putTMVar dstatus s
|
||||||
|
return r
|
||||||
|
sendNotification $ changeNotifier s
|
||||||
|
return b
|
||||||
|
|
||||||
{- Syncable remotes ordered by cost. -}
|
{- Syncable remotes ordered by cost. -}
|
||||||
calcSyncRemotes :: Annex [Remote]
|
calcSyncRemotes :: Annex [Remote]
|
||||||
calcSyncRemotes = do
|
calcSyncRemotes = do
|
||||||
|
@ -53,11 +69,10 @@ calcSyncRemotes = do
|
||||||
return $ filter good rs
|
return $ filter good rs
|
||||||
|
|
||||||
{- Updates the sycRemotes list from the list of all remotes in Annex state. -}
|
{- Updates the sycRemotes list from the list of all remotes in Annex state. -}
|
||||||
updateSyncRemotes :: DaemonStatusHandle -> Annex ()
|
updateSyncRemotes :: Assistant ()
|
||||||
updateSyncRemotes dstatus = do
|
updateSyncRemotes = do
|
||||||
remotes <- calcSyncRemotes
|
remotes <- liftAnnex calcSyncRemotes
|
||||||
liftIO $ modifyDaemonStatus_ dstatus $
|
modifyDaemonStatus_ $ \s -> s { syncRemotes = remotes }
|
||||||
\s -> s { syncRemotes = remotes }
|
|
||||||
|
|
||||||
{- Load any previous daemon status file, and store it in a 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. -}
|
- process to use as its DaemonStatus. Also gets current transfer status. -}
|
||||||
|
@ -136,15 +151,14 @@ adjustTransfersSTM dstatus a = do
|
||||||
putTMVar dstatus $ s { currentTransfers = a (currentTransfers s) }
|
putTMVar dstatus $ s { currentTransfers = a (currentTransfers s) }
|
||||||
|
|
||||||
{- Alters a transfer's info, if the transfer is in the map. -}
|
{- Alters a transfer's info, if the transfer is in the map. -}
|
||||||
alterTransferInfo :: DaemonStatusHandle -> Transfer -> (TransferInfo -> TransferInfo) -> IO ()
|
alterTransferInfo :: Transfer -> (TransferInfo -> TransferInfo) -> Assistant ()
|
||||||
alterTransferInfo dstatus t a = updateTransferInfo' dstatus $ M.adjust a t
|
alterTransferInfo t a = updateTransferInfo' $ M.adjust a t
|
||||||
|
|
||||||
{- Updates a transfer's info. Adds the transfer to the map if necessary,
|
{- Updates a transfer's info. Adds the transfer to the map if necessary,
|
||||||
- or if already present, updates it while preserving the old transferTid,
|
- or if already present, updates it while preserving the old transferTid,
|
||||||
- transferPaused, and bytesComplete values, which are not written to disk. -}
|
- transferPaused, and bytesComplete values, which are not written to disk. -}
|
||||||
updateTransferInfo :: DaemonStatusHandle -> Transfer -> TransferInfo -> IO ()
|
updateTransferInfo :: Transfer -> TransferInfo -> Assistant ()
|
||||||
updateTransferInfo dstatus t info = updateTransferInfo' dstatus $
|
updateTransferInfo t info = updateTransferInfo' $ M.insertWith' merge t info
|
||||||
M.insertWith' merge t info
|
|
||||||
where
|
where
|
||||||
merge new old = new
|
merge new old = new
|
||||||
{ transferTid = maybe (transferTid new) Just (transferTid old)
|
{ transferTid = maybe (transferTid new) Just (transferTid old)
|
||||||
|
@ -152,52 +166,59 @@ updateTransferInfo dstatus t info = updateTransferInfo' dstatus $
|
||||||
, bytesComplete = maybe (bytesComplete new) Just (bytesComplete old)
|
, bytesComplete = maybe (bytesComplete new) Just (bytesComplete old)
|
||||||
}
|
}
|
||||||
|
|
||||||
updateTransferInfo' :: DaemonStatusHandle -> (TransferMap -> TransferMap) -> IO ()
|
updateTransferInfo' :: (TransferMap -> TransferMap) -> Assistant ()
|
||||||
updateTransferInfo' dstatus a =
|
updateTransferInfo' a = notifyTransfer `after` modifyDaemonStatus_ update
|
||||||
notifyTransfer dstatus `after` modifyDaemonStatus_ dstatus go
|
|
||||||
where
|
where
|
||||||
go s = s { currentTransfers = a (currentTransfers s) }
|
update s = s { currentTransfers = a (currentTransfers s) }
|
||||||
|
|
||||||
{- Removes a transfer from the map, and returns its info. -}
|
{- Removes a transfer from the map, and returns its info. -}
|
||||||
removeTransfer :: DaemonStatusHandle -> Transfer -> IO (Maybe TransferInfo)
|
removeTransfer :: Transfer -> Assistant (Maybe TransferInfo)
|
||||||
removeTransfer dstatus t =
|
removeTransfer t = notifyTransfer `after` modifyDaemonStatus remove
|
||||||
notifyTransfer dstatus `after` modifyDaemonStatus dstatus go
|
|
||||||
where
|
where
|
||||||
go s =
|
remove s =
|
||||||
let (info, ts) = M.updateLookupWithKey
|
let (info, ts) = M.updateLookupWithKey
|
||||||
(\_k _v -> Nothing)
|
(\_k _v -> Nothing)
|
||||||
t (currentTransfers s)
|
t (currentTransfers s)
|
||||||
in (s { currentTransfers = ts }, info)
|
in (s { currentTransfers = ts }, info)
|
||||||
|
|
||||||
{- Send a notification when a transfer is changed. -}
|
{- Send a notification when a transfer is changed. -}
|
||||||
notifyTransfer :: DaemonStatusHandle -> IO ()
|
notifyTransfer :: Assistant ()
|
||||||
notifyTransfer dstatus = sendNotification
|
notifyTransfer = do
|
||||||
|
dstatus <- getAssistant daemonStatusHandle
|
||||||
|
liftIO $ sendNotification
|
||||||
|
=<< transferNotifier <$> atomically (readTMVar dstatus)
|
||||||
|
|
||||||
|
-- TODO remove
|
||||||
|
notifyTransferOld :: DaemonStatusHandle -> IO ()
|
||||||
|
notifyTransferOld dstatus = sendNotification
|
||||||
=<< transferNotifier <$> atomically (readTMVar dstatus)
|
=<< transferNotifier <$> atomically (readTMVar dstatus)
|
||||||
|
|
||||||
{- Send a notification when alerts are changed. -}
|
{- Send a notification when alerts are changed. -}
|
||||||
notifyAlert :: DaemonStatusHandle -> IO ()
|
notifyAlert :: Assistant ()
|
||||||
notifyAlert dstatus = sendNotification
|
notifyAlert = do
|
||||||
=<< alertNotifier <$> atomically (readTMVar dstatus)
|
dstatus <- getAssistant daemonStatusHandle
|
||||||
|
liftIO $ sendNotification
|
||||||
|
=<< alertNotifier <$> atomically (readTMVar dstatus)
|
||||||
|
|
||||||
{- Returns the alert's identifier, which can be used to remove it. -}
|
{- Returns the alert's identifier, which can be used to remove it. -}
|
||||||
addAlert :: DaemonStatusHandle -> Alert -> IO AlertId
|
addAlert :: Alert -> Assistant AlertId
|
||||||
addAlert dstatus alert = notifyAlert dstatus `after` modifyDaemonStatus dstatus go
|
addAlert alert = notifyAlert `after` modifyDaemonStatus add
|
||||||
where
|
where
|
||||||
go s = (s { lastAlertId = i, alertMap = m }, i)
|
add s = (s { lastAlertId = i, alertMap = m }, i)
|
||||||
where
|
where
|
||||||
i = nextAlertId $ lastAlertId s
|
i = nextAlertId $ lastAlertId s
|
||||||
m = mergeAlert i alert (alertMap s)
|
m = mergeAlert i alert (alertMap s)
|
||||||
|
|
||||||
removeAlert :: DaemonStatusHandle -> AlertId -> IO ()
|
removeAlert :: AlertId -> Assistant ()
|
||||||
removeAlert dstatus i = updateAlert dstatus i (const Nothing)
|
removeAlert i = updateAlert i (const Nothing)
|
||||||
|
|
||||||
updateAlert :: DaemonStatusHandle -> AlertId -> (Alert -> Maybe Alert) -> IO ()
|
updateAlert :: AlertId -> (Alert -> Maybe Alert) -> Assistant ()
|
||||||
updateAlert dstatus i a = updateAlertMap dstatus $ \m -> M.update a i m
|
updateAlert i a = updateAlertMap $ \m -> M.update a i m
|
||||||
|
|
||||||
updateAlertMap :: DaemonStatusHandle -> (AlertMap -> AlertMap) -> IO ()
|
updateAlertMap :: (AlertMap -> AlertMap) -> Assistant ()
|
||||||
updateAlertMap dstatus a = notifyAlert dstatus `after` modifyDaemonStatus_ dstatus go
|
updateAlertMap a = notifyAlert `after` modifyDaemonStatus_ update
|
||||||
where
|
where
|
||||||
go s = s { alertMap = a (alertMap s) }
|
update s = s { alertMap = a (alertMap s) }
|
||||||
|
|
||||||
{- Displays an alert while performing an activity that returns True on
|
{- Displays an alert while performing an activity that returns True on
|
||||||
- success.
|
- success.
|
||||||
|
@ -213,17 +234,13 @@ alertWhile alert a = alertWhile' alert $ do
|
||||||
alertWhile' :: Alert -> Assistant (Bool, a) -> Assistant a
|
alertWhile' :: Alert -> Assistant (Bool, a) -> Assistant a
|
||||||
alertWhile' alert a = do
|
alertWhile' alert a = do
|
||||||
let alert' = alert { alertClass = Activity }
|
let alert' = alert { alertClass = Activity }
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
i <- addAlert alert'
|
||||||
i <- liftIO $ addAlert dstatus alert'
|
|
||||||
(ok, r) <- a
|
(ok, r) <- a
|
||||||
liftIO $ updateAlertMap dstatus $
|
updateAlertMap $ mergeAlert i $ makeAlertFiller ok alert'
|
||||||
mergeAlert i $ makeAlertFiller ok alert'
|
|
||||||
return r
|
return r
|
||||||
|
|
||||||
{- Displays an alert while performing an activity, then removes it. -}
|
{- Displays an alert while performing an activity, then removes it. -}
|
||||||
alertDuring :: Alert -> Assistant a -> Assistant a
|
alertDuring :: Alert -> Assistant a -> Assistant a
|
||||||
alertDuring alert a = do
|
alertDuring alert a = do
|
||||||
let alert' = alert { alertClass = Activity }
|
i <- addAlert $ alert { alertClass = Activity }
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
removeAlert i `after` a
|
||||||
i <- liftIO $ addAlert dstatus alert'
|
|
||||||
liftIO (removeAlert dstatus i) `after` a
|
|
||||||
|
|
|
@ -13,7 +13,6 @@ module Assistant.Monad (
|
||||||
newAssistantData,
|
newAssistantData,
|
||||||
runAssistant,
|
runAssistant,
|
||||||
getAssistant,
|
getAssistant,
|
||||||
withAssistant,
|
|
||||||
liftAnnex,
|
liftAnnex,
|
||||||
(<~>),
|
(<~>),
|
||||||
(<<~),
|
(<<~),
|
||||||
|
@ -110,6 +109,3 @@ asIO2 a = do
|
||||||
{- Runs an IO action on a selected field of the AssistantData. -}
|
{- Runs an IO action on a selected field of the AssistantData. -}
|
||||||
(<<~) :: (a -> IO b) -> (AssistantData -> a) -> Assistant b
|
(<<~) :: (a -> IO b) -> (AssistantData -> a) -> Assistant b
|
||||||
io <<~ v = reader v >>= liftIO . io
|
io <<~ v = reader v >>= liftIO . io
|
||||||
|
|
||||||
withAssistant :: (AssistantData -> a) -> (a -> IO b) -> Assistant b
|
|
||||||
withAssistant v io = io <<~ v
|
|
||||||
|
|
|
@ -26,5 +26,5 @@ runNamedThread (NamedThread name a) = do
|
||||||
let msg = unwords [name, "crashed:", show e]
|
let msg = unwords [name, "crashed:", show e]
|
||||||
hPutStrLn stderr msg
|
hPutStrLn stderr msg
|
||||||
-- TODO click to restart
|
-- TODO click to restart
|
||||||
void $ addAlert (daemonStatusHandle d) $
|
flip runAssistant d $ void $
|
||||||
warningAlert name msg
|
addAlert $ warningAlert name msg
|
||||||
|
|
|
@ -80,7 +80,7 @@ startSending :: DaemonStatusHandle -> PairingInProgress -> PairStage -> (PairSta
|
||||||
startSending dstatus pip stage sender = void $ forkIO $ do
|
startSending dstatus pip stage sender = void $ forkIO $ do
|
||||||
tid <- myThreadId
|
tid <- myThreadId
|
||||||
let pip' = pip { inProgressPairStage = stage, inProgressThreadId = Just tid }
|
let pip' = pip { inProgressPairStage = stage, inProgressThreadId = Just tid }
|
||||||
oldpip <- modifyDaemonStatus dstatus $
|
oldpip <- modifyDaemonStatusOld dstatus $
|
||||||
\s -> (s { pairingInProgress = Just pip' }, pairingInProgress s)
|
\s -> (s { pairingInProgress = Just pip' }, pairingInProgress s)
|
||||||
maybe noop stopold oldpip
|
maybe noop stopold oldpip
|
||||||
sender stage
|
sender stage
|
||||||
|
@ -90,7 +90,7 @@ startSending dstatus pip stage sender = void $ forkIO $ do
|
||||||
stopSending :: PairingInProgress -> DaemonStatusHandle -> IO ()
|
stopSending :: PairingInProgress -> DaemonStatusHandle -> IO ()
|
||||||
stopSending pip dstatus = do
|
stopSending pip dstatus = do
|
||||||
maybe noop killThread $ inProgressThreadId pip
|
maybe noop killThread $ inProgressThreadId pip
|
||||||
modifyDaemonStatus_ dstatus $ \s -> s { pairingInProgress = Nothing }
|
modifyDaemonStatusOld_ dstatus $ \s -> s { pairingInProgress = Nothing }
|
||||||
|
|
||||||
class ToSomeAddr a where
|
class ToSomeAddr a where
|
||||||
toSomeAddr :: a -> SomeAddr
|
toSomeAddr :: a -> SomeAddr
|
||||||
|
|
|
@ -154,6 +154,6 @@ manualPull currentbranch remotes = do
|
||||||
{- Start syncing a newly added remote, using a background thread. -}
|
{- Start syncing a newly added remote, using a background thread. -}
|
||||||
syncNewRemote :: Remote -> Assistant ()
|
syncNewRemote :: Remote -> Assistant ()
|
||||||
syncNewRemote remote = do
|
syncNewRemote remote = do
|
||||||
liftAnnex . updateSyncRemotes =<< getAssistant daemonStatusHandle
|
updateSyncRemotes
|
||||||
thread <- asIO2 reconnectRemotes
|
thread <- asIO2 reconnectRemotes
|
||||||
void $ liftIO $ forkIO $ thread False [remote]
|
void $ liftIO $ forkIO $ thread False [remote]
|
||||||
|
|
|
@ -74,7 +74,7 @@ reloadConfigs changedconfigs = do
|
||||||
{- Changes to the remote log, or the trust log, can affect the
|
{- Changes to the remote log, or the trust log, can affect the
|
||||||
- syncRemotes list -}
|
- syncRemotes list -}
|
||||||
when (Logs.Remote.remoteLog `elem` fs || Logs.Trust.trustLog `elem` fs) $
|
when (Logs.Remote.remoteLog `elem` fs || Logs.Trust.trustLog `elem` fs) $
|
||||||
liftAnnex . updateSyncRemotes =<< getAssistant daemonStatusHandle
|
updateSyncRemotes
|
||||||
where
|
where
|
||||||
(fs, as) = unzip $ filter (flip S.member changedfiles . fst)
|
(fs, as) = unzip $ filter (flip S.member changedfiles . fst)
|
||||||
configFilesActions
|
configFilesActions
|
||||||
|
|
|
@ -177,7 +177,7 @@ remotesUnder dir = do
|
||||||
let (waschanged, rs') = unzip pairs
|
let (waschanged, rs') = unzip pairs
|
||||||
when (any id waschanged) $ do
|
when (any id waschanged) $ do
|
||||||
liftAnnex $ Annex.changeState $ \s -> s { Annex.remotes = rs' }
|
liftAnnex $ Annex.changeState $ \s -> s { Annex.remotes = rs' }
|
||||||
liftAnnex . updateSyncRemotes =<< getAssistant daemonStatusHandle
|
updateSyncRemotes
|
||||||
return $ map snd $ filter fst pairs
|
return $ map snd $ filter fst pairs
|
||||||
where
|
where
|
||||||
checkremote repotop r = case Remote.localpath r of
|
checkremote repotop r = case Remote.localpath r of
|
||||||
|
|
|
@ -104,12 +104,12 @@ pairReqReceived :: Bool -> UrlRenderer -> PairMsg -> Assistant ()
|
||||||
pairReqReceived True _ _ = noop -- ignore our own PairReq
|
pairReqReceived True _ _ = noop -- ignore our own PairReq
|
||||||
pairReqReceived False urlrenderer msg = do
|
pairReqReceived False urlrenderer msg = do
|
||||||
url <- liftIO $ renderUrl urlrenderer (FinishPairR msg) []
|
url <- liftIO $ renderUrl urlrenderer (FinishPairR msg) []
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
close <- asIO removeAlert
|
||||||
liftIO $ void $ addAlert dstatus $ pairRequestReceivedAlert repo
|
void $ addAlert $ pairRequestReceivedAlert repo
|
||||||
AlertButton
|
AlertButton
|
||||||
{ buttonUrl = url
|
{ buttonUrl = url
|
||||||
, buttonLabel = T.pack "Respond"
|
, buttonLabel = T.pack "Respond"
|
||||||
, buttonAction = Just $ removeAlert dstatus
|
, buttonAction = Just close
|
||||||
}
|
}
|
||||||
where
|
where
|
||||||
repo = pairRepo msg
|
repo = pairRepo msg
|
||||||
|
|
|
@ -28,14 +28,12 @@ sanityCheckerThread = NamedThread "SanityChecker" $ forever $ do
|
||||||
debug ["sanity check complete"]
|
debug ["sanity check complete"]
|
||||||
where
|
where
|
||||||
go = do
|
go = do
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
modifyDaemonStatus_ $ \s -> s { sanityCheckRunning = True }
|
||||||
liftIO $ modifyDaemonStatus_ dstatus $ \s -> s
|
|
||||||
{ sanityCheckRunning = True }
|
|
||||||
|
|
||||||
now <- liftIO $ getPOSIXTime -- before check started
|
now <- liftIO $ getPOSIXTime -- before check started
|
||||||
r <- either showerr return =<< tryIO <~> check
|
r <- either showerr return =<< tryIO <~> check
|
||||||
|
|
||||||
liftIO $ modifyDaemonStatus_ dstatus $ \s -> s
|
modifyDaemonStatus_ $ \s -> s
|
||||||
{ sanityCheckRunning = False
|
{ sanityCheckRunning = False
|
||||||
, lastSanityCheck = Just now
|
, lastSanityCheck = Just now
|
||||||
}
|
}
|
||||||
|
@ -84,8 +82,7 @@ check = do
|
||||||
slop = fromIntegral tenMinutes
|
slop = fromIntegral tenMinutes
|
||||||
insanity msg = do
|
insanity msg = do
|
||||||
liftAnnex $ warning msg
|
liftAnnex $ warning msg
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
void $ addAlert $ sanityCheckFixAlert msg
|
||||||
liftIO $ void $ addAlert dstatus $ sanityCheckFixAlert msg
|
|
||||||
addsymlink file s = do
|
addsymlink file s = do
|
||||||
Watcher.runHandler Watcher.onAddSymlink file s
|
Watcher.runHandler Watcher.onAddSymlink file s
|
||||||
insanity $ "found unstaged symlink: " ++ file
|
insanity $ "found unstaged symlink: " ++ file
|
||||||
|
|
|
@ -52,7 +52,5 @@ transferPollerThread = NamedThread "TransferPoller" $ do
|
||||||
|
|
||||||
newsize t info sz
|
newsize t info sz
|
||||||
| bytesComplete info /= sz && isJust sz =
|
| bytesComplete info /= sz && isJust sz =
|
||||||
withAssistant daemonStatusHandle $ \h ->
|
alterTransferInfo t $ \i -> i { bytesComplete = sz }
|
||||||
alterTransferInfo h t $
|
|
||||||
\i -> i { bytesComplete = sz }
|
|
||||||
| otherwise = noop
|
| otherwise = noop
|
||||||
|
|
|
@ -64,8 +64,7 @@ onAdd file = case parseTransferFile file of
|
||||||
debug [ "transfer starting:", show t]
|
debug [ "transfer starting:", show t]
|
||||||
r <- headMaybe . filter (sameuuid t)
|
r <- headMaybe . filter (sameuuid t)
|
||||||
<$> liftAnnex Remote.remoteList
|
<$> liftAnnex Remote.remoteList
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
updateTransferInfo t info { transferRemote = r }
|
||||||
liftIO $ updateTransferInfo dstatus t info { transferRemote = r }
|
|
||||||
sameuuid t r = Remote.uuid r == transferUUID t
|
sameuuid t r = Remote.uuid r == transferUUID t
|
||||||
|
|
||||||
{- Called when a transfer information file is updated.
|
{- Called when a transfer information file is updated.
|
||||||
|
@ -79,9 +78,8 @@ onModify file = do
|
||||||
Just t -> go t =<< liftIO (readTransferInfoFile Nothing file)
|
Just t -> go t =<< liftIO (readTransferInfoFile Nothing file)
|
||||||
where
|
where
|
||||||
go _ Nothing = noop
|
go _ Nothing = noop
|
||||||
go t (Just newinfo) = withAssistant daemonStatusHandle $ \h ->
|
go t (Just newinfo) = alterTransferInfo t $
|
||||||
alterTransferInfo h t $
|
\i -> i { bytesComplete = bytesComplete newinfo }
|
||||||
\i -> i { bytesComplete = bytesComplete newinfo }
|
|
||||||
|
|
||||||
{- This thread can only watch transfer sizes when the DirWatcher supports
|
{- This thread can only watch transfer sizes when the DirWatcher supports
|
||||||
- tracking modificatons to files. -}
|
- tracking modificatons to files. -}
|
||||||
|
@ -94,7 +92,7 @@ onDel file = case parseTransferFile file of
|
||||||
Nothing -> noop
|
Nothing -> noop
|
||||||
Just t -> do
|
Just t -> do
|
||||||
debug [ "transfer finishing:", show t]
|
debug [ "transfer finishing:", show t]
|
||||||
minfo <- flip removeTransfer t <<~ daemonStatusHandle
|
minfo <- removeTransfer t
|
||||||
|
|
||||||
finished <- asIO2 finishedTransfer
|
finished <- asIO2 finishedTransfer
|
||||||
void $ liftIO $ forkIO $ do
|
void $ liftIO $ forkIO $ do
|
||||||
|
|
|
@ -32,9 +32,8 @@ transfererThread = NamedThread "Transferr" $ do
|
||||||
program <- liftIO readProgramFile
|
program <- liftIO readProgramFile
|
||||||
transferqueue <- getAssistant transferQueue
|
transferqueue <- getAssistant transferQueue
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
dstatus <- getAssistant daemonStatusHandle
|
||||||
slots <- getAssistant transferSlots
|
|
||||||
starter <- asIO2 $ startTransfer program
|
starter <- asIO2 $ startTransfer program
|
||||||
liftIO $ forever $ inTransferSlot dstatus slots $
|
forever $ inTransferSlot $ liftIO $
|
||||||
maybe (return Nothing) (uncurry starter)
|
maybe (return Nothing) (uncurry starter)
|
||||||
=<< getNextTransfer transferqueue dstatus notrunning
|
=<< getNextTransfer transferqueue dstatus notrunning
|
||||||
where
|
where
|
||||||
|
@ -48,12 +47,12 @@ startTransfer program t info = case (transferRemote info, associatedFile info) o
|
||||||
(Just remote, Just file) -> ifM (liftAnnex $ shouldTransfer t info)
|
(Just remote, Just file) -> ifM (liftAnnex $ shouldTransfer t info)
|
||||||
( do
|
( do
|
||||||
debug [ "Transferring:" , show t ]
|
debug [ "Transferring:" , show t ]
|
||||||
notifyTransfer <<~ daemonStatusHandle
|
notifyTransfer
|
||||||
tp <- asIO2 transferprocess
|
tp <- asIO2 transferprocess
|
||||||
return $ Just (t, info, tp remote file)
|
return $ Just (t, info, tp remote file)
|
||||||
, do
|
, do
|
||||||
debug [ "Skipping unnecessary transfer:" , show t ]
|
debug [ "Skipping unnecessary transfer:" , show t ]
|
||||||
void $ flip removeTransfer t <<~ daemonStatusHandle
|
void $ removeTransfer t
|
||||||
return Nothing
|
return Nothing
|
||||||
)
|
)
|
||||||
_ -> return Nothing
|
_ -> return Nothing
|
||||||
|
@ -77,10 +76,8 @@ startTransfer program t info = case (transferRemote info, associatedFile info) o
|
||||||
- in the transfer.
|
- in the transfer.
|
||||||
-}
|
-}
|
||||||
whenM (liftIO $ (==) ExitSuccess <$> waitForProcess pid) $ do
|
whenM (liftIO $ (==) ExitSuccess <$> waitForProcess pid) $ do
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
void $ addAlert $ makeAlertFiller True $
|
||||||
liftIO $ void $ addAlert dstatus $
|
transferFileAlert direction True file
|
||||||
makeAlertFiller True $
|
|
||||||
transferFileAlert direction True file
|
|
||||||
recordCommit
|
recordCommit
|
||||||
where
|
where
|
||||||
params =
|
params =
|
||||||
|
|
|
@ -85,9 +85,7 @@ startupScan scanner = do
|
||||||
inRepo $ Git.Command.run "add" [Param "--update"]
|
inRepo $ Git.Command.run "add" [Param "--update"]
|
||||||
showAction "started"
|
showAction "started"
|
||||||
|
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
modifyDaemonStatus_ $ \s -> s { scanComplete = True }
|
||||||
liftIO $ modifyDaemonStatus_ dstatus $
|
|
||||||
\s -> s { scanComplete = True }
|
|
||||||
|
|
||||||
return (True, r)
|
return (True, r)
|
||||||
|
|
||||||
|
@ -218,8 +216,7 @@ onDelDir dir _ = do
|
||||||
onErr :: Handler
|
onErr :: Handler
|
||||||
onErr msg _ = do
|
onErr msg _ = do
|
||||||
liftAnnex $ warning msg
|
liftAnnex $ warning msg
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
void $ addAlert $ warningAlert "watcher" msg
|
||||||
void $ liftIO $ addAlert dstatus $ warningAlert "watcher" msg
|
|
||||||
noChange
|
noChange
|
||||||
|
|
||||||
{- Adds a symlink to the index, without ever accessing the actual symlink
|
{- Adds a symlink to the index, without ever accessing the actual symlink
|
||||||
|
|
|
@ -123,7 +123,7 @@ enqueue schedule q dstatus t info
|
||||||
atomically $ do
|
atomically $ do
|
||||||
void $ modifyTVar' (queuesize q) succ
|
void $ modifyTVar' (queuesize q) succ
|
||||||
void $ modifyTVar' (queuelist q) modlist
|
void $ modifyTVar' (queuelist q) modlist
|
||||||
void $ notifyTransfer dstatus
|
void $ notifyTransferOld dstatus
|
||||||
|
|
||||||
{- Adds a transfer to the queue. -}
|
{- Adds a transfer to the queue. -}
|
||||||
queueTransfer :: Schedule -> TransferQueue -> DaemonStatusHandle -> AssociatedFile -> Transfer -> Remote -> IO ()
|
queueTransfer :: Schedule -> TransferQueue -> DaemonStatusHandle -> AssociatedFile -> Transfer -> Remote -> IO ()
|
||||||
|
@ -182,7 +182,7 @@ dequeueTransfers :: TransferQueue -> DaemonStatusHandle -> (Transfer -> Bool) ->
|
||||||
dequeueTransfers q dstatus c = do
|
dequeueTransfers q dstatus c = do
|
||||||
removed <- atomically $ dequeueTransfersSTM q c
|
removed <- atomically $ dequeueTransfersSTM q c
|
||||||
unless (null removed) $
|
unless (null removed) $
|
||||||
notifyTransfer dstatus
|
notifyTransferOld dstatus
|
||||||
return removed
|
return removed
|
||||||
|
|
||||||
dequeueTransfersSTM :: TransferQueue -> (Transfer -> Bool) -> STM [(Transfer, TransferInfo)]
|
dequeueTransfersSTM :: TransferQueue -> (Transfer -> Bool) -> STM [(Transfer, TransferInfo)]
|
||||||
|
|
|
@ -17,20 +17,22 @@ import qualified Control.Exception as E
|
||||||
import Control.Concurrent
|
import Control.Concurrent
|
||||||
import qualified Control.Concurrent.MSemN as MSemN
|
import qualified Control.Concurrent.MSemN as MSemN
|
||||||
|
|
||||||
|
type TransferGenerator = Assistant (Maybe (Transfer, TransferInfo, IO ()))
|
||||||
|
|
||||||
{- Waits until a transfer slot becomes available, then runs a
|
{- Waits until a transfer slot becomes available, then runs a
|
||||||
- TransferGenerator, and then runs the transfer action in its own thread.
|
- TransferGenerator, and then runs the transfer action in its own thread.
|
||||||
-}
|
-}
|
||||||
inTransferSlot :: TransferSlotRunner
|
inTransferSlot :: TransferGenerator -> Assistant ()
|
||||||
inTransferSlot dstatus s gen = do
|
inTransferSlot gen = do
|
||||||
MSemN.wait s 1
|
flip MSemN.wait 1 <<~ transferSlots
|
||||||
runTransferThread dstatus s =<< gen
|
runTransferThread =<< gen
|
||||||
|
|
||||||
{- Runs a TransferGenerator, and its transfer action,
|
{- Runs a TransferGenerator, and its transfer action,
|
||||||
- without waiting for a slot to become available. -}
|
- without waiting for a slot to become available. -}
|
||||||
inImmediateTransferSlot :: TransferSlotRunner
|
inImmediateTransferSlot :: TransferGenerator -> Assistant ()
|
||||||
inImmediateTransferSlot dstatus s gen = do
|
inImmediateTransferSlot gen = do
|
||||||
MSemN.signal s (-1)
|
flip MSemN.signal (-1) <<~ transferSlots
|
||||||
runTransferThread dstatus s =<< gen
|
runTransferThread =<< gen
|
||||||
|
|
||||||
{- Runs a transfer action, in an already allocated transfer slot.
|
{- Runs a transfer action, in an already allocated transfer slot.
|
||||||
- Once it finishes, frees the transfer slot.
|
- Once it finishes, frees the transfer slot.
|
||||||
|
@ -42,24 +44,26 @@ inImmediateTransferSlot dstatus s gen = do
|
||||||
- then pausing the thread until a ResumeTransfer exception is raised,
|
- then pausing the thread until a ResumeTransfer exception is raised,
|
||||||
- then rerunning the action.
|
- then rerunning the action.
|
||||||
-}
|
-}
|
||||||
runTransferThread :: DaemonStatusHandle -> TransferSlots -> Maybe (Transfer, TransferInfo, IO ()) -> IO ()
|
runTransferThread :: Maybe (Transfer, TransferInfo, IO ()) -> Assistant ()
|
||||||
runTransferThread _ s Nothing = MSemN.signal s 1
|
runTransferThread Nothing = flip MSemN.signal 1 <<~ transferSlots
|
||||||
runTransferThread dstatus s (Just (t, info, a)) = do
|
runTransferThread (Just (t, info, a)) = do
|
||||||
tid <- forkIO go
|
d <- getAssistant id
|
||||||
updateTransferInfo dstatus t $ info { transferTid = Just tid }
|
tid <- liftIO $ forkIO $ go d
|
||||||
|
updateTransferInfo t $ info { transferTid = Just tid }
|
||||||
where
|
where
|
||||||
go = catchPauseResume a
|
go d = catchPauseResume d a
|
||||||
pause = catchPauseResume $ runEvery (Seconds 86400) noop
|
pause d = catchPauseResume d $ runEvery (Seconds 86400) noop
|
||||||
{- Note: This must use E.try, rather than E.catch.
|
{- Note: This must use E.try, rather than E.catch.
|
||||||
- When E.catch is used, and has called go in its exception
|
- When E.catch is used, and has called go in its exception
|
||||||
- handler, Control.Concurrent.throwTo will block sometimes
|
- handler, Control.Concurrent.throwTo will block sometimes
|
||||||
- when signaling. Using E.try avoids the problem. -}
|
- when signaling. Using E.try avoids the problem. -}
|
||||||
catchPauseResume a' = do
|
catchPauseResume d a' = do
|
||||||
r <- E.try a' :: IO (Either E.SomeException ())
|
r <- E.try a' :: IO (Either E.SomeException ())
|
||||||
case r of
|
case r of
|
||||||
Left e -> case E.fromException e of
|
Left e -> case E.fromException e of
|
||||||
Just PauseTransfer -> pause
|
Just PauseTransfer -> pause d
|
||||||
Just ResumeTransfer -> go
|
Just ResumeTransfer -> go d
|
||||||
_ -> done
|
_ -> done d
|
||||||
_ -> done
|
_ -> done d
|
||||||
done = MSemN.signal s 1
|
done d = flip runAssistant d $
|
||||||
|
flip MSemN.signal 1 <<~ transferSlots
|
||||||
|
|
|
@ -9,9 +9,6 @@
|
||||||
|
|
||||||
module Assistant.Types.TransferSlots where
|
module Assistant.Types.TransferSlots where
|
||||||
|
|
||||||
import Assistant.Types.DaemonStatus
|
|
||||||
import Logs.Transfer
|
|
||||||
|
|
||||||
import qualified Control.Exception as E
|
import qualified Control.Exception as E
|
||||||
import qualified Control.Concurrent.MSemN as MSemN
|
import qualified Control.Concurrent.MSemN as MSemN
|
||||||
import Data.Typeable
|
import Data.Typeable
|
||||||
|
@ -25,9 +22,6 @@ data TransferException = PauseTransfer | ResumeTransfer
|
||||||
|
|
||||||
instance E.Exception TransferException
|
instance E.Exception TransferException
|
||||||
|
|
||||||
type TransferSlotRunner = DaemonStatusHandle -> TransferSlots -> TransferGenerator -> IO ()
|
|
||||||
type TransferGenerator = IO (Maybe (Transfer, TransferInfo, IO ()))
|
|
||||||
|
|
||||||
{- Number of concurrent transfers allowed to be run from the assistant.
|
{- Number of concurrent transfers allowed to be run from the assistant.
|
||||||
-
|
-
|
||||||
- Transfers launched by other means, including by remote assistants,
|
- Transfers launched by other means, including by remote assistants,
|
||||||
|
|
|
@ -69,7 +69,6 @@ setRepoConfig uuid mremote oldc newc = do
|
||||||
when (repoSyncable oldc /= repoSyncable newc) $
|
when (repoSyncable oldc /= repoSyncable newc) $
|
||||||
changeSyncable mremote (repoSyncable newc)
|
changeSyncable mremote (repoSyncable newc)
|
||||||
when (isJust mremote && repoName oldc /= repoName newc) $ do
|
when (isJust mremote && repoName oldc /= repoName newc) $ do
|
||||||
dstatus <- getAssistantY daemonStatusHandle
|
|
||||||
runAnnex undefined $ do
|
runAnnex undefined $ do
|
||||||
name <- fromRepo $ uniqueRemoteName (T.unpack $ repoName newc) 0
|
name <- fromRepo $ uniqueRemoteName (T.unpack $ repoName newc) 0
|
||||||
inRepo $ Git.Command.run "remote"
|
inRepo $ Git.Command.run "remote"
|
||||||
|
@ -78,7 +77,7 @@ setRepoConfig uuid mremote oldc newc = do
|
||||||
, Param name
|
, Param name
|
||||||
]
|
]
|
||||||
void $ Remote.remoteListRefresh
|
void $ Remote.remoteListRefresh
|
||||||
updateSyncRemotes dstatus
|
runAssistantY updateSyncRemotes
|
||||||
|
|
||||||
editRepositoryAForm :: RepoConfig -> AForm WebApp WebApp RepoConfig
|
editRepositoryAForm :: RepoConfig -> AForm WebApp WebApp RepoConfig
|
||||||
editRepositoryAForm def = RepoConfig
|
editRepositoryAForm def = RepoConfig
|
||||||
|
|
|
@ -34,13 +34,14 @@ import qualified Data.Text as T
|
||||||
{- Displays an alert suggesting to configure XMPP, with a button. -}
|
{- Displays an alert suggesting to configure XMPP, with a button. -}
|
||||||
xmppNeeded :: Handler ()
|
xmppNeeded :: Handler ()
|
||||||
xmppNeeded = whenM (isNothing <$> runAnnex Nothing getXMPPCreds) $ do
|
xmppNeeded = whenM (isNothing <$> runAnnex Nothing getXMPPCreds) $ do
|
||||||
dstatus <- getAssistantY daemonStatusHandle
|
|
||||||
urlrender <- getUrlRender
|
urlrender <- getUrlRender
|
||||||
void $ liftIO $ addAlert dstatus $ xmppNeededAlert $ AlertButton
|
void $ runAssistantY $ do
|
||||||
{ buttonLabel = "Configure a Jabber account"
|
close <- asIO removeAlert
|
||||||
, buttonUrl = urlrender XMPPR
|
addAlert $ xmppNeededAlert $ AlertButton
|
||||||
, buttonAction = Just $ removeAlert dstatus
|
{ buttonLabel = "Configure a Jabber account"
|
||||||
}
|
, buttonUrl = urlrender XMPPR
|
||||||
|
, buttonAction = Just close
|
||||||
|
}
|
||||||
|
|
||||||
getXMPPR :: Handler RepHtml
|
getXMPPR :: Handler RepHtml
|
||||||
#ifdef WITH_XMPP
|
#ifdef WITH_XMPP
|
||||||
|
|
|
@ -73,9 +73,7 @@ getSideBarR nid = do
|
||||||
|
|
||||||
{- Called by the client to close an alert. -}
|
{- Called by the client to close an alert. -}
|
||||||
getCloseAlert :: AlertId -> Handler ()
|
getCloseAlert :: AlertId -> Handler ()
|
||||||
getCloseAlert i = do
|
getCloseAlert = runAssistantY . removeAlert
|
||||||
dstatus <- getAssistantY daemonStatusHandle
|
|
||||||
liftIO $ removeAlert dstatus i
|
|
||||||
|
|
||||||
{- When an alert with a button is clicked on, the button takes us here. -}
|
{- When an alert with a button is clicked on, the button takes us here. -}
|
||||||
getClickAlert :: AlertId -> Handler ()
|
getClickAlert :: AlertId -> Handler ()
|
||||||
|
|
|
@ -38,7 +38,7 @@ changeSyncable (Just r) False = do
|
||||||
changeSyncFlag r False
|
changeSyncFlag r False
|
||||||
d <- getAssistantY id
|
d <- getAssistantY id
|
||||||
let dstatus = daemonStatusHandle d
|
let dstatus = daemonStatusHandle d
|
||||||
runAssistantY $ liftAnnex $ updateSyncRemotes dstatus
|
runAssistantY $ updateSyncRemotes
|
||||||
{- Stop all transfers to or from this remote.
|
{- Stop all transfers to or from this remote.
|
||||||
- XXX Can't stop any ongoing scan, or git syncs. -}
|
- XXX Can't stop any ongoing scan, or git syncs. -}
|
||||||
void $ liftIO $ dequeueTransfers (transferQueue d) dstatus tofrom
|
void $ liftIO $ dequeueTransfers (transferQueue d) dstatus tofrom
|
||||||
|
@ -67,30 +67,27 @@ pauseTransfer = cancelTransfer True
|
||||||
|
|
||||||
cancelTransfer :: Bool -> Transfer -> Handler ()
|
cancelTransfer :: Bool -> Transfer -> Handler ()
|
||||||
cancelTransfer pause t = do
|
cancelTransfer pause t = do
|
||||||
dstatus <- getAssistantY daemonStatusHandle
|
|
||||||
tq <- getAssistantY transferQueue
|
tq <- getAssistantY transferQueue
|
||||||
m <- getCurrentTransfers
|
m <- getCurrentTransfers
|
||||||
liftIO $ do
|
dstatus <- getAssistantY daemonStatusHandle
|
||||||
unless pause $
|
unless pause $ liftIO $
|
||||||
{- remove queued transfer -}
|
{- remove queued transfer -}
|
||||||
void $ dequeueTransfers tq dstatus $
|
void $ dequeueTransfers tq dstatus $
|
||||||
equivilantTransfer t
|
equivilantTransfer t
|
||||||
{- stop running transfer -}
|
{- stop running transfer -}
|
||||||
maybe noop (stop dstatus) (M.lookup t m)
|
maybe noop stop (M.lookup t m)
|
||||||
where
|
where
|
||||||
stop dstatus info = do
|
stop info = runAssistantY $ do
|
||||||
{- When there's a thread associated with the
|
{- When there's a thread associated with the
|
||||||
- transfer, it's signaled first, to avoid it
|
- transfer, it's signaled first, to avoid it
|
||||||
- displaying any alert about the transfer having
|
- displaying any alert about the transfer having
|
||||||
- failed when the transfer process is killed. -}
|
- failed when the transfer process is killed. -}
|
||||||
maybe noop signalthread $ transferTid info
|
liftIO $ maybe noop signalthread $ transferTid info
|
||||||
maybe noop killproc $ transferPid info
|
liftIO $ maybe noop killproc $ transferPid info
|
||||||
if pause
|
if pause
|
||||||
then void $
|
then void $ alterTransferInfo t $
|
||||||
alterTransferInfo dstatus t $
|
\i -> i { transferPaused = True }
|
||||||
\i -> i { transferPaused = True }
|
else void $ removeTransfer t
|
||||||
else void $
|
|
||||||
removeTransfer dstatus t
|
|
||||||
signalthread tid
|
signalthread tid
|
||||||
| pause = throwTo tid PauseTransfer
|
| pause = throwTo tid PauseTransfer
|
||||||
| otherwise = killThread tid
|
| otherwise = killThread tid
|
||||||
|
@ -115,16 +112,12 @@ startTransfer t = do
|
||||||
is <- liftIO $ map snd <$> getMatchingTransfers q dstatus (== t)
|
is <- liftIO $ map snd <$> getMatchingTransfers q dstatus (== t)
|
||||||
maybe noop start $ headMaybe is
|
maybe noop start $ headMaybe is
|
||||||
resume tid = do
|
resume tid = do
|
||||||
dstatus <- getAssistantY daemonStatusHandle
|
runAssistantY $ alterTransferInfo t $
|
||||||
liftIO $ do
|
\i -> i { transferPaused = False }
|
||||||
alterTransferInfo dstatus t $
|
liftIO $ throwTo tid ResumeTransfer
|
||||||
\i -> i { transferPaused = False }
|
|
||||||
throwTo tid ResumeTransfer
|
|
||||||
start info = runAssistantY $ do
|
start info = runAssistantY $ do
|
||||||
program <- liftIO readProgramFile
|
program <- liftIO readProgramFile
|
||||||
dstatus <- getAssistant daemonStatusHandle
|
inImmediateTransferSlot $
|
||||||
slots <- getAssistant transferSlots
|
|
||||||
inImmediateTransferSlot dstatus slots <~>
|
|
||||||
Transferrer.startTransfer program t info
|
Transferrer.startTransfer program t info
|
||||||
|
|
||||||
getCurrentTransfers :: Handler TransferMap
|
getCurrentTransfers :: Handler TransferMap
|
||||||
|
|
Loading…
Reference in a new issue