-- that we and the client both speak.
let protocolversion = min maxProtocolVersion clientmaxversion
selectnode <- clusterProxySelector clusteruuid protocolversion
- proxy proxydone proxymethods servermode clientside selectnode
- protocolversion othermsg protoerrhandler
+ proxy proxydone proxymethods servermode clientside
+ (fromClusterUUID clusteruuid)
+ selectnode protocolversion othermsg protoerrhandler
withclientversion Nothing = proxydone
clusterProxySelector :: ClusterUUID -> ProtocolVersion -> Annex ProxySelector
return $ ProxySelector
{ proxyCHECKPRESENT = nodecontaining remotesides
, proxyGET = nodecontaining remotesides
- , proxyPUT = \k -> error "TODO"
+ -- Send the key to every node that does not yet contain it.
+ , proxyPUT = \k -> do
+ locs <- S.fromList <$> loggedLocations k
+ return $ filter (flip S.notMember locs . remoteUUID) remotesides
, proxyREMOVE = \k -> error "TODO"
-- Content is not locked on the cluster as a whole,
-- instead it can be locked on individual nodes that are
complaining about missing tree objects.
* Tab completion of options like --from now includes special remotes,
as well as proxied remotes and clusters.
+ * P2P protocol version 2.
* Fix Windows build with Win32 2.13.4+
Thanks, Oleg Tolmatcev
closeRemoteSide remoteside
p2pDone
proxy closer proxymethods servermode clientside
+ (Remote.uuid remote)
(singleProxySelector remoteside)
protocolversion othermsg p2pErrHandler
withclientversion _ Nothing = p2pDone
defaultProtocolVersion = ProtocolVersion 0
maxProtocolVersion :: ProtocolVersion
-maxProtocolVersion = ProtocolVersion 1
+maxProtocolVersion = ProtocolVersion 2
newtype ProtoAssociatedFile = ProtoAssociatedFile AssociatedFile
deriving (Show)
| PUT ProtoAssociatedFile Key
| PUT_FROM Offset
| ALREADY_HAVE
+ | ALREADY_HAVE_PLUS [UUID]
| SUCCESS
+ | SUCCESS_PLUS [UUID]
| FAILURE
| DATA Len -- followed by bytes of data
| VALIDITY Validity
formatMessage (PUT af key) = ["PUT", Proto.serialize af, Proto.serialize key]
formatMessage (PUT_FROM offset) = ["PUT-FROM", Proto.serialize offset]
formatMessage ALREADY_HAVE = ["ALREADY-HAVE"]
+ formatMessage (ALREADY_HAVE_PLUS uuids) = ("ALREADY-HAVE-PLUS":map Proto.serialize uuids)
formatMessage SUCCESS = ["SUCCESS"]
+ formatMessage (SUCCESS_PLUS uuids) = ("SUCCESS-PLUS":map Proto.serialize uuids)
formatMessage FAILURE = ["FAILURE"]
formatMessage (VALIDITY Valid) = ["VALID"]
formatMessage (VALIDITY Invalid) = ["INVALID"]
parseCommand "PUT" = Proto.parse2 PUT
parseCommand "PUT-FROM" = Proto.parse1 PUT_FROM
parseCommand "ALREADY-HAVE" = Proto.parse0 ALREADY_HAVE
+ parseCommand "ALREADY-HAVE-PLUS" = Proto.parseList ALREADY_HAVE_PLUS
parseCommand "SUCCESS" = Proto.parse0 SUCCESS
+ parseCommand "SUCCESS-PLUS" = Proto.parseList SUCCESS_PLUS
parseCommand "FAILURE" = Proto.parse0 FAILURE
parseCommand "DATA" = Proto.parse1 DATA
parseCommand "ERROR" = Proto.parse1 ERROR
| ContentSize Key (Maybe Len -> c)
-- ^ Gets size of the content of a key, when the full content is
-- present.
- | ReadContent Key AssociatedFile Offset (L.ByteString -> Proto Validity -> Proto Bool) (Bool -> c)
+ | ReadContent Key AssociatedFile Offset (L.ByteString -> Proto Validity -> Proto (Maybe [UUID])) (Maybe [UUID] -> c)
-- ^ Reads the content of a key and sends it to the callback.
-- Must run the callback, or terminate the protocol connection.
--
sizer = fileSize dest
storer = storeContentTo dest iv
-put :: Key -> AssociatedFile -> MeterUpdate -> Proto Bool
+put :: Key -> AssociatedFile -> MeterUpdate -> Proto (Maybe [UUID])
put key af p = do
net $ sendMessage (PUT (ProtoAssociatedFile af) key)
r <- net receiveMessage
case r of
Just (PUT_FROM offset) -> sendContent key af offset p
- Just ALREADY_HAVE -> return True
+ Just ALREADY_HAVE -> return (Just [])
+ Just (ALREADY_HAVE_PLUS uuids) -> return (Just uuids)
_ -> do
net $ sendMessage (ERROR "expected PUT_FROM or ALREADY_HAVE")
- return False
+ return Nothing
data ServerHandler a
= ServerGot a
(ServeReadOnly, UploadPack) -> a Nothing
(ServeReadOnly, ReceivePack) -> a (Just sendReadOnlyError)
-sendContent :: Key -> AssociatedFile -> Offset -> MeterUpdate -> Proto Bool
+sendContent :: Key -> AssociatedFile -> Offset -> MeterUpdate -> Proto (Maybe [UUID])
sendContent key af offset@(Offset n) p = go =<< local (contentSize key)
where
go (Just (Len totallen)) = do
ver <- net getProtocolVersion
when (ver >= ProtocolVersion 1) $
net . sendMessage . VALIDITY =<< validitycheck
- checkSuccess
+ if ver >= ProtocolVersion 2
+ then checkSuccessPlus
+ else do
+ ok <- checkSuccess
+ if ok
+ then return (Just [])
+ else return Nothing
receiveContent
:: Observable t
net $ sendMessage (ERROR "expected SUCCESS or FAILURE")
return False
+checkSuccessPlus :: Proto (Maybe [UUID])
+checkSuccessPlus = do
+ ack <- net receiveMessage
+ case ack of
+ Just SUCCESS -> return (Just [])
+ Just (SUCCESS_PLUS l) -> return (Just l)
+ Just FAILURE -> return Nothing
+ _ -> do
+ net $ sendMessage (ERROR "expected SUCCESS or SUCCESS-PLUS or FAILURE")
+ return Nothing
+
sendSuccess :: Bool -> Proto ()
sendSuccess True = net $ sendMessage SUCCESS
sendSuccess False = net $ sendMessage FAILURE
-}
{-# LANGUAGE RankNTypes, FlexibleContexts, ScopedTypeVariables #-}
+{-# LANGUAGE BangPatterns #-}
module P2P.Proxy where
import Annex.Common
import P2P.Protocol
import P2P.IO
-import Utility.Metered (nullMeterUpdate)
+import Utility.Metered
+import Data.Either
import Control.Concurrent.STM
+import qualified Data.ByteString.Lazy as L
type ProtoCloser = Annex ()
, proxyUNLOCKCONTENT :: Annex (Maybe RemoteSide)
, proxyREMOVE :: Key -> Annex RemoteSide
, proxyGET :: Key -> Annex (Maybe RemoteSide)
- , proxyPUT :: Key -> Annex RemoteSide
+ , proxyPUT :: Key -> Annex [RemoteSide]
}
singleProxySelector :: RemoteSide -> ProxySelector
, proxyUNLOCKCONTENT = pure (Just r)
, proxyREMOVE = const (pure r)
, proxyGET = const (pure (Just r))
- , proxyPUT = const (pure r)
+ , proxyPUT = const (pure [r])
}
{- To keep this module limited to P2P protocol actions,
-- If the client sends a newer version than we
-- understand, reduce it; we need to parse the
-- protocol too.
- let v' = if v > maxProtocolVersion
- then maxProtocolVersion
- else v
+ let v' = min v maxProtocolVersion
in return (Just (v', Nothing))
Just othermsg -> return
(Just (defaultProtocolVersion, Just othermsg))
-> ProxyMethods
-> ServerMode
-> ClientSide
+ -> UUID
-> ProxySelector
-> ProtocolVersion
-> Maybe Message
-- ^ non-VERSION message that was received from the client when
-- negotiating protocol version, and has not been responded to yet
-> ProtoErrorHandled r
-proxy proxydone proxymethods servermode (ClientSide clientrunst clientconn) proxyselector protocolversion othermessage protoerrhandler = do
+proxy proxydone proxymethods servermode (ClientSide clientrunst clientconn) remoteuuid proxyselector (ProtocolVersion protocolversion) othermessage protoerrhandler = do
case othermessage of
Nothing -> protoerrhandler proxynextclientmessage $
- client $ net $ sendMessage $ VERSION protocolversion
+ client $ net $ sendMessage $ VERSION $ ProtocolVersion protocolversion
Just message -> proxyclientmessage (Just message)
where
client = liftIO . runNetProto clientrunst clientconn
client $ net $ sendMessage $
ERROR "content not present"
PUT _ k -> do
- remoteside <- proxyPUT proxyselector k
+ remotesides <- proxyPUT proxyselector k
servermodechecker checkPUTServerMode $
- handlePUT remoteside k message
+ handlePUT remotesides k message
-- These messages involve the git repository, not the
-- annex. So they affect the git repository of the proxy,
-- not the remote.
-- Messages that the client should only send after one of
-- the messages above.
SUCCESS -> protoerr
+ SUCCESS_PLUS _ -> protoerr
FAILURE -> protoerr
DATA _ -> protoerr
VALIDITY _ -> protoerr
AUTH_FAILURE -> protoerr
PUT_FROM _ -> protoerr
ALREADY_HAVE -> protoerr
+ ALREADY_HAVE_PLUS _ -> protoerr
-- Early messages that the client should not send now.
AUTH _ _ -> protoerr
VERSION _ -> protoerr
handleGET remoteside message = getresponse (runRemoteSide remoteside) message $
withDATA (relayGET remoteside)
- handlePUT remoteside k message =
+ handlePUT (remoteside:[]) k message =
getresponse (runRemoteSide remoteside) message $ \resp -> case resp of
ALREADY_HAVE -> protoerrhandler proxynextclientmessage $
client $ net $ sendMessage resp
+ ALREADY_HAVE_PLUS _ -> protoerrhandler proxynextclientmessage $
+ client $ net $ sendMessage resp
PUT_FROM _ ->
- getresponse client resp $ withDATA (relayPUT remoteside k)
+ getresponse client resp $
+ withDATA (relayPUT remoteside k)
_ -> protoerr
+ handlePUT [] _ _ =
+ protoerrhandler proxynextclientmessage $
+ client $ net $ sendMessage ALREADY_HAVE
+ handlePUT remotesides k message =
+ handlePutMulti remotesides k message
withDATA a message@(DATA len) = a len message
withDATA _ _ = protoerr
finished resp () = do
case resp of
SUCCESS -> addedContent proxymethods (remoteUUID remoteside) k
+ SUCCESS_PLUS us ->
+ forM_ (remoteUUID remoteside:us) $ \u ->
+ addedContent proxymethods u k
_ -> return ()
proxynextclientmessage ()
protoerrhandler (\() -> receive) $
x $ net $ sendMessage message
- relayDATACore len x y finishget = protoerrhandler send $
+ relayDATACore len x y a = protoerrhandler send $
x $ net $ receiveBytes len nullMeterUpdate
where
- send b = protoerrhandler finishget $
+ send b = protoerrhandler a $
y $ net $ sendBytes len b nullMeterUpdate
- relayDATAFinish x y sendsuccessfailure () = case protocolversion of
- ProtocolVersion 0 -> sendsuccessfailure
+ relayDATAFinish x y sendsuccessfailure ()
+ | protocolversion == 0 = sendsuccessfailure
-- Protocol version 1 has a VALID or
-- INVALID message after the data.
- _ -> relayonemessage x y (\_ () -> sendsuccessfailure)
+ | otherwise = relayonemessage x y (\_ () -> sendsuccessfailure)
+
+ handlePutMulti remotesides k message = do
+ let initiate remoteside = do
+ resp <- runRemoteSide remoteside $ net $ do
+ sendMessage message
+ receiveMessage
+ case resp of
+ Right (Just (PUT_FROM (Offset offset))) ->
+ return $ Right $
+ Right (remoteside, offset)
+ Right (Just ALREADY_HAVE) ->
+ return $ Right $ Left remoteside
+ Right (Just _) -> protoerr
+ Right Nothing -> return (Left ())
+ Left _err -> return (Left ())
+ let alreadyhave = \case
+ Right (Left _) -> True
+ _ -> False
+ l <- forM remotesides initiate
+ if all alreadyhave l
+ then if protocolversion < 2
+ then protoerrhandler proxynextclientmessage $
+ client $ net $ sendMessage ALREADY_HAVE
+ else protoerrhandler proxynextclientmessage $
+ client $ net $ sendMessage $ ALREADY_HAVE_PLUS $
+ filter (/= remoteuuid) $
+ map remoteUUID (lefts (rights l))
+ else if null (rights l)
+ -- no response from any remote
+ then proxydone
+ else do
+ let l' = rights (rights l)
+ let minoffset = minimum (map snd l')
+ getresponse client (PUT_FROM (Offset minoffset)) $
+ withDATA (relayPUTMulti minoffset l' k)
+ relayPUTMulti minoffset remotes k (Len datalen) _ = do
+ let totallen = datalen + minoffset
+ -- Tell each remote how much data to expect, depending
+ -- on the remote's offset.
+ forM_ remotes $ \(remoteside, remoteoffset) ->
+ runRemoteSide remoteside $
+ net $ sendMessage $ DATA $ Len $
+ totallen - remoteoffset
+ protoerrhandler (send remotes minoffset) $
+ client $ net $ receiveBytes (Len datalen) nullMeterUpdate
+ where
+ chunksize = fromIntegral defaultChunkSize
+
+ -- Stream the lazy bytestring out to the remotes in chunks.
+ -- Only start sending to a remote once past its desired
+ -- offset.
+ send rs n b = do
+ let (chunk, b') = L.splitAt chunksize b
+ let chunklen = fromIntegral (L.length chunk)
+ let !n' = n + chunklen
+ rs' <- forM rs $ \r@(remoteside, remoteoffset) ->
+ if n >= remoteoffset
+ then skipfailed r $ runRemoteSide remoteside $
+ net $ sendBytes (Len chunklen) chunk nullMeterUpdate
+ else if (n' <= remoteoffset)
+ then do
+ let chunkoffset = remoteoffset - n
+ let subchunklen = chunklen - chunkoffset
+ let subchunk = L.drop (fromIntegral chunkoffset) chunk
+ skipfailed r $ runRemoteSide remoteside $
+ net $ sendBytes (Len subchunklen) subchunk nullMeterUpdate
+ else return (Just r)
+ if L.null b'
+ then sent (catMaybes rs')
+ else send (catMaybes rs') n' b'
+
+ sent [] = proxydone
+ sent rs = giveup "XXX" -- XXX
+
+ skipfailed r@(remoteside, _) a = a >>= \case
+ Right _ -> return (Just r)
+ Left _ -> do
+ -- This connection to the remote is
+ -- unrecoverable at this point, so close it.
+ closeRemoteSide remoteside
+ return Nothing
)
| Git.repoIsSsh repo =
P2PHelper.store (gitconfig r)
- (Ssh.runProto r connpool (return False))
+ (Ssh.runProto r connpool (return Nothing))
key file meterupdate
| otherwise = giveup "copying to non-ssh repo not supported"
import Utility.Tuple
import Types.NumCopies
import Annex.Verify
+import Logs.Location
import Control.Concurrent
-- the pool when done.
type WithConn a c = (ClosableConnection c -> Annex (ClosableConnection c, a)) -> Annex a
-store :: RemoteGitConfig -> ProtoRunner Bool -> Key -> AssociatedFile -> MeterUpdate -> Annex ()
+store :: RemoteGitConfig -> ProtoRunner (Maybe [UUID]) -> Key -> AssociatedFile -> MeterUpdate -> Annex ()
store gc runner k af p = do
let sizer = KeySizer k (fmap (toRawFilePath . fst3) <$> prepSendAnnex k)
let bwlimit = remoteAnnexBwLimitUpload gc <|> remoteAnnexBwLimit gc
metered (Just p) sizer bwlimit $ \_ p' ->
runner (P2P.put k af p') >>= \case
- Just True -> return ()
- Just False -> giveup "Transfer failed"
+ Just (Just fanoutuuids) -> do
+ -- Storing on the remote can cause it
+ -- to be stored on additional UUIDs,
+ -- so record those.
+ forM_ fanoutuuids $ \u ->
+ logChange k u InfoPresent
+ Just Nothing -> giveup "Transfer failed"
Nothing -> remoteUnavail
retrieve :: RemoteGitConfig -> (ProtoRunner (Bool, Verification)) -> Key -> AssociatedFile -> FilePath -> MeterUpdate -> VerifyConfig -> Annex Verification
observeBool Nothing = False
observeFailure = Nothing
+instance Observable (Either e (Maybe a)) where
+ observeBool (Left _) = False
+ observeBool (Right Nothing) = False
+ observeBool (Right (Just _)) = True
+ observeFailure = Right Nothing
+
class Transferrable t where
descTransfrerrable :: t -> Maybe String
{- Simple line-based protocols.
-
- - Copyright 2013-2020 Joey Hess <id@joeyh.name>
+ - Copyright 2013-2024 Joey Hess <id@joeyh.name>
-
- License: BSD-2-clause
-}
parse3,
parse4,
parse5,
+ parseList,
dupIoHandles,
getProtocolLine,
) where
splitWord :: String -> (String, String)
splitWord = separate isSpace
+{- Only safe to use when the serialization does not include whitespace. -}
+parseList :: Serializable p => ([p] -> a) -> Parser a
+parseList mk v = mk <$> mapM deserialize (words v)
+
{- When a program speaks a simple protocol over stdio, any other output
- to stdout (or anything that attempts to read from stdin)
- will mess up the protocol. To avoid that, close stdin,
The client sends the highest protocol version it supports:
- VERSION 2
+ VERSION 3
The server responds with the highest protocol version it supports
that is less than or equal to the version the client sent:
whitespace.)
The server may respond with ALREADY-HAVE if it already
-had the conent of that key. Otherwise, it responds with:
+had the conent of that key.
+
+In protocol version 2, the server can optionally reply with
+ALREADY-HAVE-PLUS. The subsequent list of UUIDs are additional
+UUIDs where the content is stored, in addition to the UUID where
+the client was going to send it.
+
+Otherwise, it responds with:
PUT-FROM Offset
If the server successfully receives the data and stores the content,
it replies with SUCCESS. Otherwise, FAILURE.
+In protocol version 2, the server can optionally reply with SUCCESS-PLUS.
+The subsequent list of UUIDs are additional UUIDs where the content was
+stored, in addition to the UUID where the client was sending it.
+
## Getting content from the server
To get content from the server, the client sends:
If we want to send a file to multiple repositories that are behind the same
proxy, it would be wasteful to upload it through the proxy repeatedly.
-Perhaps a good user interface to this is `git-annex copy --to proxy`.
-The proxy could fan out the upload and store it in one or more nodes behind
-it. Using preferred content to select which nodes to use.
-This would need `storeKey` to be changed to allow returning a UUID (or UUIDs)
-where the content was actually stored.
-
-Alternatively, `git-annex copy --to proxy-foo` could notice that proxy-bar
-also wants the content, and fan out a copy to there. Then it could
-record in its git-annex branch that the content is present in proxy-bar.
-If the user later does `git-annex copy --to proxy-bar`, it would avoid
-another upload (and the user would learn at that point that it was in
-proxy-bar). This avoids needing to change the `storeKey` interface.
-
-Should a proxy always fanout? if `git-annex copy --to proxy` is what does
-fanout, and `git-annex copy --to proxy-foo` doesn't, then the user has
-content. But if the latter does fanout, that might be annoying to users who
-want to use proxies, but want full control over what lands where, and don't
-want to use preferred content to do it. So probably fanout should be
-configurable. But it can't be configured client side, because the fanout
-happens on the proxy. Seems like remote.name.annex-fanout could be set to
-false to prevent fanout to a specific remote. (This is analagous to a
-remote having `git-annex assistant` running on it, it might fan out uploads
-to it to other repos, and only the owner of that repo can control it.)
-
-Alternatively, fanout could be limited to clusters.
+This is certianly needed when doing `git-annex copy --to remote-cluster`,
+the cluster picks the nodes to store the content in, and it needs to report
+back some UUID that is different than the cluster UUID, in order for the
+location log to get updated. (Cluster UUIDs are not written to the location
+log.) So this will need a change to the P2P protocol to support reporting
+back additional UUIDs where the content was stored.
+
+This might also be useful for proxies. `git-annex copy --to proxy-foo`
+could notice that proxy-bar also wants the content, and fan out a copy to
+there. But that might be annoying to users, who want full control over what
+goes where when using a proxy. Seems it would need a config setting. But
+since clusters will support fanout, it seems unncessary to make proxies
+also support it.
A command like `git-annex push` would see all the instantiated remotes and
would pick ones to send content to. If fanout is done, this would
* Tab complete proxied remotes and clusters in eg --from option. (done)
-* Basic proxying to special remote support (non-streaming).
-
* Getting a key from a cluster should proxy from one of the nodes that has
it. (done)
other git-annex processes are currently using.
* Implement upload with fanout and reporting back additional UUIDs over P2P
- protocol.
+ protocol. (started, but incomplete)
+
+* On upload to cluster, send to nodes where it's preferred content, and not
+ to other nodes.
* Implement cluster drops, trying to remove from all nodes, and returning
which UUIDs it was dropped from.
And on download from a cluster, if the proxy repository has the content,
get it from there to avoid the overhead of proxying to a node.
+* Basic proxying to special remote support (non-streaming).
+
* Support proxies-of-proxies better, eg foo-bar-baz.
Currently, it does work, but have to run `git-annex updateproxy`
on foo in order for it to notice the bar-baz proxied remote exists,
projects / git-annex.git / commitdiff