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
|
{- git-annex assistant pending transfer queue
-
- Copyright 2012-2014 Joey Hess <id@joeyh.name>
-
- Licensed under the GNU GPL version 3 or higher.
-}
{-# LANGUAGE BangPatterns #-}
module Assistant.TransferQueue (
TransferQueue,
Schedule(..),
newTransferQueue,
getTransferQueue,
queueTransfers,
queueTransfersMatching,
queueDeferredDownloads,
queueTransfer,
queueTransferAt,
queueTransferWhenSmall,
getNextTransfer,
getMatchingTransfers,
dequeueTransfers,
) where
import Assistant.Common
import Assistant.DaemonStatus
import Assistant.Types.TransferQueue
import Types.Transfer
import Logs.Transfer
import Types.Remote
import qualified Remote
import qualified Types.Remote as Remote
import Annex.Wanted
import Utility.TList
import Control.Concurrent.STM
import qualified Data.Map.Strict as M
import qualified Data.Set as S
type Reason = String
{- Reads the queue's content without blocking or changing it. -}
getTransferQueue :: Assistant [(Transfer, TransferInfo)]
getTransferQueue = (atomically . readTList . queuelist) <<~ transferQueue
stubInfo :: AssociatedFile -> Remote -> TransferInfo
stubInfo f r = stubTransferInfo
{ transferRemote = Just r
, associatedFile = f
}
{- Adds transfers to queue for some of the known remotes.
- Honors preferred content settings, only transferring wanted files. -}
queueTransfers :: Reason -> Schedule -> Key -> AssociatedFile -> Direction -> Assistant Bool
queueTransfers = queueTransfersMatching (const True)
{- Adds transfers to queue for some of the known remotes, that match a
- condition. Honors preferred content settings. -}
queueTransfersMatching :: (UUID -> Bool) -> Reason -> Schedule -> Key -> AssociatedFile -> Direction -> Assistant Bool
queueTransfersMatching matching reason schedule k f direction
| direction == Download = ifM (liftAnnex $ wantGet True (Just k) f)
( go
, return False
)
| otherwise = go
where
go = do
rs <- liftAnnex . selectremotes =<< getDaemonStatus
let matchingrs = filter (matching . Remote.uuid) rs
if null matchingrs
then do
defer
return False
else do
forM_ matchingrs $ \r ->
enqueue reason schedule (gentransfer r) (stubInfo f r)
return True
selectremotes st
{- Queue downloads from all remotes that
- have the key. The list of remotes is ordered with
- cheapest first. More expensive ones will only be tried
- if downloading from a cheap one fails. -}
| direction == Download = do
s <- locs
return $ filter (inset s) (downloadRemotes st)
{- Upload to all remotes that want the content and don't
- already have it. -}
| otherwise = do
s <- locs
filterM (wantSend True (Just k) f . Remote.uuid) $
filter (\r -> not (inset s r || Remote.readonly r))
(syncDataRemotes st)
where
locs = S.fromList . map Remote.uuid <$> Remote.keyPossibilities k
inset s r = S.member (Remote.uuid r) s
gentransfer r = Transfer
{ transferDirection = direction
, transferKey = k
, transferUUID = Remote.uuid r
}
defer
{- Defer this download, as no known remote has the key. -}
| direction == Download = do
q <- getAssistant transferQueue
void $ liftIO $ atomically $
consTList (deferreddownloads q) (k, f)
| otherwise = noop
{- Queues any deferred downloads that can now be accomplished, leaving
- any others in the list to try again later. -}
queueDeferredDownloads :: Reason -> Schedule -> Assistant ()
queueDeferredDownloads reason schedule = do
q <- getAssistant transferQueue
l <- liftIO $ atomically $ readTList (deferreddownloads q)
rs <- downloadRemotes <$> getDaemonStatus
left <- filterM (queue rs) l
unless (null left) $
liftIO $ atomically $ appendTList (deferreddownloads q) left
where
queue rs (k, f) = do
uuids <- liftAnnex $ Remote.keyLocations k
let sources = filter (\r -> uuid r `elem` uuids) rs
unless (null sources) $
forM_ sources $ \r ->
enqueue reason schedule
(gentransfer r) (stubInfo f r)
return $ null sources
where
gentransfer r = Transfer
{ transferDirection = Download
, transferKey = k
, transferUUID = Remote.uuid r
}
enqueue :: Reason -> Schedule -> Transfer -> TransferInfo -> Assistant ()
enqueue reason schedule t info
| schedule == Next = go consTList
| otherwise = go snocTList
where
go modlist = whenM (add modlist) $ do
debug [ "queued", describeTransfer t info, ": " ++ reason ]
notifyTransfer
add modlist = do
q <- getAssistant transferQueue
dstatus <- getAssistant daemonStatusHandle
liftIO $ atomically $ ifM (checkRunningTransferSTM dstatus t)
( return False
, do
l <- readTList (queuelist q)
if (t `notElem` map fst l)
then do
void $ modifyTVar' (queuesize q) succ
void $ modlist (queuelist q) (t, info)
return True
else return False
)
{- Adds a transfer to the queue. -}
queueTransfer :: Reason -> Schedule -> AssociatedFile -> Transfer -> Remote -> Assistant ()
queueTransfer reason schedule f t remote =
enqueue reason schedule t (stubInfo f remote)
{- Blocks until the queue is no larger than a given size, and then adds a
- transfer to the queue. -}
queueTransferAt :: Int -> Reason -> Schedule -> AssociatedFile -> Transfer -> Remote -> Assistant ()
queueTransferAt wantsz reason schedule f t remote = do
q <- getAssistant transferQueue
liftIO $ atomically $ do
sz <- readTVar (queuesize q)
unless (sz <= wantsz) $
retry -- blocks until queuesize changes
enqueue reason schedule t (stubInfo f remote)
queueTransferWhenSmall :: Reason -> AssociatedFile -> Transfer -> Remote -> Assistant ()
queueTransferWhenSmall reason = queueTransferAt 10 reason Later
{- Blocks until a pending transfer is available in the queue,
- and removes it.
-
- Checks that it's acceptable, before adding it to the
- currentTransfers map. If it's not acceptable, it's discarded.
-
- This is done in a single STM transaction, so there is no window
- where an observer sees an inconsistent status. -}
getNextTransfer :: (TransferInfo -> Bool) -> Assistant (Maybe (Transfer, TransferInfo))
getNextTransfer acceptable = do
q <- getAssistant transferQueue
dstatus <- getAssistant daemonStatusHandle
liftIO $ atomically $ do
sz <- readTVar (queuesize q)
if sz < 1
then retry -- blocks until queuesize changes
else readTList (queuelist q) >>= \case
[] -> retry -- blocks until something is queued
(r@(t,info):rest) -> do
void $ modifyTVar' (queuesize q) pred
setTList (queuelist q) rest
if acceptable info
then do
adjustTransfersSTM dstatus $
M.insert t info
return $ Just r
else return Nothing
{- Moves transfers matching a condition from the queue, to the
- currentTransfers map. -}
getMatchingTransfers :: (Transfer -> Bool) -> Assistant [(Transfer, TransferInfo)]
getMatchingTransfers c = do
q <- getAssistant transferQueue
dstatus <- getAssistant daemonStatusHandle
liftIO $ atomically $ do
ts <- dequeueTransfersSTM q c
unless (null ts) $
adjustTransfersSTM dstatus $ \m -> M.union m $ M.fromList ts
return ts
{- Removes transfers matching a condition from the queue, and returns the
- removed transfers. -}
dequeueTransfers :: (Transfer -> Bool) -> Assistant [(Transfer, TransferInfo)]
dequeueTransfers c = do
q <- getAssistant transferQueue
removed <- liftIO $ atomically $ dequeueTransfersSTM q c
unless (null removed) $
notifyTransfer
return removed
dequeueTransfersSTM :: TransferQueue -> (Transfer -> Bool) -> STM [(Transfer, TransferInfo)]
dequeueTransfersSTM q c = do
!(removed, ts) <- partition (c . fst) <$> readTList (queuelist q)
let !len = length ts
void $ writeTVar (queuesize q) len
setTList (queuelist q) ts
return removed
|