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
|
{- git-annex assistant pending transfer queue
-
- Copyright 2012 Joey Hess <joey@kitenet.net>
-
- Licensed under the GNU GPL version 3 or higher.
-}
module Assistant.TransferQueue where
import Common.Annex
import Assistant.DaemonStatus
import Logs.Transfer
import Types.Remote
import qualified Remote
import Control.Concurrent.STM
{- The transfer queue consists of a channel listing the transfers to make;
- the size of the queue is also tracked -}
data TransferQueue = TransferQueue
{ queue :: TChan (Transfer, TransferInfo)
, queuesize :: TVar Integer
}
data Schedule = Next | Later
deriving (Eq)
newTransferQueue :: IO TransferQueue
newTransferQueue = atomically $ TransferQueue <$> newTChan <*> newTVar 0
stubInfo :: AssociatedFile -> TransferInfo
stubInfo f = TransferInfo
{ startedTime = Nothing
, transferPid = Nothing
, transferTid = Nothing
, transferRemote = Nothing
, bytesComplete = Nothing
, associatedFile = f
}
{- Adds pending 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
mapM_ go =<< sufficientremotes rs
where
sufficientremotes rs
-- Queue downloads from all remotes that
-- have the key, with the cheapest ones first.
-- More expensive ones will only be tried if
-- downloading from a cheap one fails.
| direction == Download = do
uuids <- Remote.keyLocations k
return $ filter (\r -> uuid r `elem` uuids) rs
-- TODO: Determine a smaller set of remotes that
-- can be uploaded to, in order to ensure all
-- remotes can access the content. Currently,
-- send to every remote we can.
| otherwise = return rs
gentransfer r = Transfer
{ transferDirection = direction
, transferKey = k
, transferUUID = Remote.uuid r
}
go r = liftIO $ atomically $ do
let info = (stubInfo f) { transferRemote = Just r }
enqueue schedule q (gentransfer r) info
enqueue :: Schedule -> TransferQueue -> Transfer -> TransferInfo -> STM ()
enqueue schedule q t info
| schedule == Next = go unGetTChan
| otherwise = go writeTChan
where
go a = do
void $ a (queue q) (t, info)
void $ modifyTVar' (queuesize q) succ
{- Adds a transfer to the queue. -}
queueTransfer :: Schedule -> TransferQueue -> AssociatedFile -> Transfer -> IO ()
queueTransfer schedule q f t = atomically $ enqueue schedule q t (stubInfo f)
{- Blocks until the queue is no larger than a given size, and then adds a
- transfer to the queue. -}
queueTransferAt :: Integer -> Schedule -> TransferQueue -> AssociatedFile -> Transfer -> IO ()
queueTransferAt wantsz schedule q f t = atomically $ do
sz <- readTVar (queuesize q)
if sz <= wantsz
then enqueue schedule q t (stubInfo f)
else retry -- blocks until queuesize changes
{- Blocks until a pending transfer is available from the queue. -}
getNextTransfer :: TransferQueue -> IO (Transfer, TransferInfo)
getNextTransfer q = atomically $ do
void $ modifyTVar' (queuesize q) pred
readTChan (queue q)
|
