custom protocol for transferrer

Rather than using Read/Show, which would force me to preserve data types
into the future.

I considered just deriving json and sending that, but I don't much like
deriving json with data types that have named constructors (like Key
does) because again it locks in data type details.

So instead, used SimpleProtocol, with a fairly complex and unreadable
protocol. But it is as efficient as the p2p protocol at least, and as
future proof.

(Writing my own custom json instances would have worked but I thought
of it too late and don't want to do all the work twice. The only real
benefit might be that aeson could be faster.)

Note that, when a new protocol request type is added later, git-annex
trying to use it will cause the git-annex transferrer to display a
protocol error message. That seems ok; it would only happen if a new
git-annex found an old version of itself in PATH or the program
file. So it's unlikely, and all it can do anyway is display an error.
(The error message could perhaps be improved..)

This commit was sponsored by Jack Hill on Patreon.
This commit is contained in:
Joey Hess 2020-12-09 15:44:00 -04:00
parent 447d798987
commit 04c12aa6df
No known key found for this signature in database
GPG key ID: DB12DB0FF05F8F38
6 changed files with 194 additions and 53 deletions

View file

@ -34,7 +34,6 @@ import Utility.Metered
import Utility.ThreadScheduler import Utility.ThreadScheduler
import Annex.LockPool import Annex.LockPool
import Types.Key import Types.Key
import Types.Transferrer
import qualified Types.Remote as Remote import qualified Types.Remote as Remote
import Types.Concurrency import Types.Concurrency
import Annex.Concurrent.Utility import Annex.Concurrent.Utility

View file

@ -14,7 +14,6 @@ import qualified Annex
import Types.TransferrerPool import Types.TransferrerPool
import Types.Transferrer import Types.Transferrer
import Types.Transfer import Types.Transfer
import Types.Key
import qualified Types.Remote as Remote import qualified Types.Remote as Remote
import Types.StallDetection import Types.StallDetection
import Types.Messages import Types.Messages
@ -24,12 +23,12 @@ import Utility.Batch
import Utility.Metered import Utility.Metered
import Utility.HumanTime import Utility.HumanTime
import Utility.ThreadScheduler import Utility.ThreadScheduler
import qualified Utility.SimpleProtocol as Proto
import Control.Concurrent import Control.Concurrent
import Control.Concurrent.Async import Control.Concurrent.Async
import Control.Concurrent.STM hiding (check) import Control.Concurrent.STM hiding (check)
import Control.Monad.IO.Class (MonadIO) import Control.Monad.IO.Class (MonadIO)
import Text.Read (readMaybe)
import Data.Time.Clock.POSIX import Data.Time.Clock.POSIX
import System.Log.Logger (debugM) import System.Log.Logger (debugM)
@ -91,6 +90,9 @@ checkTransferrerPoolItem program batchmaker i = case i of
t <- mkTransferrer program batchmaker t <- mkTransferrer program batchmaker
return (TransferrerPoolItem (Just t) check, t) return (TransferrerPoolItem (Just t) check, t)
data TransferRequestLevel = AnnexLevel | AssistantLevel
deriving (Show)
{- Requests that a Transferrer perform a Transfer, and waits for it to {- Requests that a Transferrer perform a Transfer, and waits for it to
- finish. - finish.
- -
@ -212,17 +214,28 @@ mkTransferrer program batchmaker = do
-- | Send a request to perform a transfer. -- | Send a request to perform a transfer.
sendRequest :: TransferRequestLevel -> Transfer -> Maybe Remote -> AssociatedFile -> Handle -> IO () sendRequest :: TransferRequestLevel -> Transfer -> Maybe Remote -> AssociatedFile -> Handle -> IO ()
sendRequest level t mremote afile h = do sendRequest level t mremote afile h = do
let l = show $ TransferRequest level let tr = maybe
(transferDirection t) (TransferRemoteUUID (transferUUID t))
(maybe (Left (transferUUID t)) (Right . Remote.name) mremote) (TransferRemoteName . Remote.name)
(keyData (transferKey t)) mremote
afile let f = case (level, transferDirection t) of
(AnnexLevel, Upload) -> UploadRequest
(AnnexLevel, Download) -> DownloadRequest
(AssistantLevel, Upload) -> AssistantUploadRequest
(AssistantLevel, Download) -> AssistantDownloadRequest
let r = f tr (transferKey t) (TransferAssociatedFile afile)
let l = unwords $ Proto.formatMessage r
debugM "transfer" ("> " ++ l) debugM "transfer" ("> " ++ l)
hPutStrLn h l hPutStrLn h l
hFlush h hFlush h
sendSerializedOutputResponse :: Handle -> SerializedOutputResponse -> IO () sendSerializedOutputResponse :: Handle -> SerializedOutputResponse -> IO ()
sendSerializedOutputResponse h sor = hPutStrLn h $ show sor sendSerializedOutputResponse h sor = do
let l = unwords $ Proto.formatMessage $
TransferSerializedOutputResponse sor
debugM "transfer" ("> " ++ show l)
hPutStrLn h l
hFlush h
-- | Read a response to a transfer requests. -- | Read a response to a transfer requests.
-- --
@ -232,13 +245,13 @@ readResponse :: Handle -> IO (Either SerializedOutput Bool)
readResponse h = do readResponse h = do
l <- liftIO $ hGetLine h l <- liftIO $ hGetLine h
debugM "transfer" ("< " ++ l) debugM "transfer" ("< " ++ l)
case readMaybe l of case Proto.parseMessage l of
Just (TransferOutput so) -> return (Left so) Just (TransferOutput so) -> return (Left so)
Just (TransferResult r) -> return (Right r) Just (TransferResult r) -> return (Right r)
Nothing -> transferrerProtocolError l Nothing -> transferrerProtocolError l
transferrerProtocolError :: String -> a transferrerProtocolError :: String -> a
transferrerProtocolError l = error $ "transferrer protocol error: " ++ show l transferrerProtocolError l = giveup $ "transferrer protocol error: " ++ show l
{- Closing the fds will shut down the transferrer, but only when it's {- Closing the fds will shut down the transferrer, but only when it's
- in between transfers. -} - in between transfers. -}

View file

@ -26,7 +26,6 @@ import Assistant.Commits
import Assistant.Drop import Assistant.Drop
import Annex.Transfer (stallDetection) import Annex.Transfer (stallDetection)
import Types.Transfer import Types.Transfer
import Types.Transferrer
import Logs.Transfer import Logs.Transfer
import Logs.Location import Logs.Location
import qualified Git import qualified Git

View file

@ -19,8 +19,7 @@ import Annex.BranchState
import Types.Messages import Types.Messages
import Annex.TransferrerPool import Annex.TransferrerPool
import Types.Transferrer import Types.Transferrer
import qualified Utility.SimpleProtocol as Proto
import Text.Read (readMaybe)
cmd :: Command cmd :: Command
cmd = command "transferrer" SectionPlumbing "transfers content" cmd = command "transferrer" SectionPlumbing "transfers content"
@ -33,32 +32,36 @@ start :: CommandStart
start = do start = do
enableInteractiveBranchAccess enableInteractiveBranchAccess
(readh, writeh) <- liftIO dupIoHandles (readh, writeh) <- liftIO dupIoHandles
Annex.setOutput $ SerializedOutput let outputwriter v = do
(\v -> hPutStrLn writeh (show (TransferOutput v)) >> hFlush writeh) hPutStrLn writeh $
(readMaybe <$> hGetLine readh) unwords $ Proto.formatMessage $ TransferOutput v
hFlush writeh
let outputresponsereader = do
l <- hGetLine readh
return $ case Proto.parseMessage l of
Just (TransferSerializedOutputResponse r) -> Just r
Nothing -> Nothing
Annex.setOutput $ SerializedOutput outputwriter outputresponsereader
runRequests readh writeh runner runRequests readh writeh runner
stop stop
where where
runner (TransferRequest AnnexLevel direction _ keydata file) remote runner (UploadRequest _ key (TransferAssociatedFile file)) remote =
| direction == Upload = -- This is called by eg, Annex.Transfer.upload,
-- This is called by eg, Annex.Transfer.upload, -- so caller is responsible for doing notification,
-- so caller is responsible for doing notification, -- and for retrying.
-- and for retrying. upload' (Remote.uuid remote) key file noRetry
upload' (Remote.uuid remote) key file noRetry (Remote.action . Remote.storeKey remote key file)
(Remote.action . Remote.storeKey remote key file) noNotification
noNotification runner (DownloadRequest _ key (TransferAssociatedFile file)) remote =
| otherwise = -- This is called by eg, Annex.Transfer.download
-- This is called by eg, Annex.Transfer.download -- so caller is responsible for doing notification
-- so caller is responsible for doing notification -- and for retrying.
-- and for retrying. let go p = getViaTmp (Remote.retrievalSecurityPolicy remote) (RemoteVerify remote) key file $ \t -> do
let go p = getViaTmp (Remote.retrievalSecurityPolicy remote) (RemoteVerify remote) key file $ \t -> do Remote.verifiedAction (Remote.retrieveKeyFile remote key file (fromRawFilePath t) p)
Remote.verifiedAction (Remote.retrieveKeyFile remote key file (fromRawFilePath t) p) in download' (Remote.uuid remote) key file noRetry go
in download' (Remote.uuid remote) key file noRetry go noNotification
noNotification runner (AssistantUploadRequest _ key (TransferAssociatedFile file)) remote =
where notifyTransfer Upload file $
key = mkKey (const keydata)
runner (TransferRequest AssistantLevel direction _ keydata file) remote
| direction == Upload = notifyTransfer direction file $
upload' (Remote.uuid remote) key file stdRetry $ \p -> do upload' (Remote.uuid remote) key file stdRetry $ \p -> do
tryNonAsync (Remote.storeKey remote key file p) >>= \case tryNonAsync (Remote.storeKey remote key file p) >>= \case
Left e -> do Left e -> do
@ -67,7 +70,8 @@ start = do
Right () -> do Right () -> do
Remote.logStatus remote key InfoPresent Remote.logStatus remote key InfoPresent
return True return True
| otherwise = notifyTransfer direction file $ runner (AssistantDownloadRequest _ key (TransferAssociatedFile file)) remote =
notifyTransfer Download file $
download' (Remote.uuid remote) key file stdRetry $ \p -> download' (Remote.uuid remote) key file stdRetry $ \p ->
getViaTmp (Remote.retrievalSecurityPolicy remote) (RemoteVerify remote) key file $ \t -> do getViaTmp (Remote.retrievalSecurityPolicy remote) (RemoteVerify remote) key file $ \t -> do
r <- tryNonAsync (Remote.retrieveKeyFile remote key file (fromRawFilePath t) p) >>= \case r <- tryNonAsync (Remote.retrieveKeyFile remote key file (fromRawFilePath t) p) >>= \case
@ -80,8 +84,6 @@ start = do
-- not old cached data. -- not old cached data.
Database.Keys.closeDb Database.Keys.closeDb
return r return r
where
key = mkKey (const keydata)
runRequests runRequests
:: Handle :: Handle
@ -92,15 +94,19 @@ runRequests readh writeh a = go Nothing Nothing
where where
go lastremoteoruuid lastremote = unlessM (liftIO $ hIsEOF readh) $ do go lastremoteoruuid lastremote = unlessM (liftIO $ hIsEOF readh) $ do
l <- liftIO $ hGetLine readh l <- liftIO $ hGetLine readh
case readMaybe l of case Proto.parseMessage l of
Just tr@(TransferRequest _ _ remoteoruuid _ _) -> do Just tr -> do
let remoteoruuid = transferRequestRemote tr
-- Often the same remote will be used -- Often the same remote will be used
-- repeatedly, so cache the last one to -- repeatedly, so cache the last one to
-- avoid looking up repeatedly. -- avoid looking up repeatedly.
mremote <- if lastremoteoruuid == Just remoteoruuid mremote <- if lastremoteoruuid == Just remoteoruuid
then pure lastremote then pure lastremote
else eitherToMaybe <$> Remote.byName' else case remoteoruuid of
(either fromUUID id remoteoruuid) TransferRemoteName n ->
eitherToMaybe <$> Remote.byName' n
TransferRemoteUUID u ->
Remote.byUUID u
case mremote of case mremote of
Just remote -> do Just remote -> do
sendresult =<< a tr remote sendresult =<< a tr remote
@ -109,5 +115,6 @@ runRequests readh writeh a = go Nothing Nothing
Nothing -> transferrerProtocolError l Nothing -> transferrerProtocolError l
sendresult b = liftIO $ do sendresult b = liftIO $ do
hPutStrLn writeh $ show $ TransferResult b hPutStrLn writeh $
unwords $ Proto.formatMessage $ TransferResult b
hFlush writeh hFlush writeh

View file

@ -77,8 +77,8 @@ data SerializedOutput
| JSONObject L.ByteString | JSONObject L.ByteString
-- ^ This is always sent, it's up to the consumer to decide if it -- ^ This is always sent, it's up to the consumer to decide if it
-- wants to display JSON, or human-readable messages. -- wants to display JSON, or human-readable messages.
deriving (Show, Read) deriving (Show)
data SerializedOutputResponse data SerializedOutputResponse
= ReadyPrompt = ReadyPrompt
deriving (Eq, Show, Read) deriving (Eq, Show)

View file

@ -8,17 +8,140 @@
module Types.Transferrer where module Types.Transferrer where
import Annex.Common import Annex.Common
import Types.Transfer
import Types.Messages import Types.Messages
import Git.Types (RemoteName) import Git.Types (RemoteName)
import qualified Utility.SimpleProtocol as Proto
import Utility.Format
data TransferRequest = TransferRequest TransferRequestLevel Direction (Either UUID RemoteName) KeyData AssociatedFile import Data.Char
deriving (Show, Read)
data TransferRequestLevel = AnnexLevel | AssistantLevel -- Sent to start a transfer.
deriving (Show, Read) data TransferRequest
= UploadRequest TransferRemote Key TransferAssociatedFile
| DownloadRequest TransferRemote Key TransferAssociatedFile
| AssistantUploadRequest TransferRemote Key TransferAssociatedFile
| AssistantDownloadRequest TransferRemote Key TransferAssociatedFile
deriving (Show)
transferRequestRemote :: TransferRequest -> TransferRemote
transferRequestRemote (UploadRequest r _ _) = r
transferRequestRemote (DownloadRequest r _ _) = r
transferRequestRemote (AssistantUploadRequest r _ _) = r
transferRequestRemote (AssistantDownloadRequest r _ _) = r
data TransferRemote
= TransferRemoteUUID UUID
| TransferRemoteName RemoteName
deriving (Show, Eq)
newtype TransferAssociatedFile = TransferAssociatedFile AssociatedFile
deriving (Show)
data TransferResponse data TransferResponse
= TransferOutput SerializedOutput = TransferOutput SerializedOutput
-- ^ any number may be sent before TransferResult
| TransferResult Bool | TransferResult Bool
deriving (Show, Read) deriving (Show)
data TransferSerializedOutputResponse = TransferSerializedOutputResponse SerializedOutputResponse
deriving (Show)
instance Proto.Sendable TransferRequest where
formatMessage (UploadRequest r kd af) =
[ "u"
, Proto.serialize r
, Proto.serialize kd
, Proto.serialize af
]
formatMessage (DownloadRequest r kd af) =
[ "d"
, Proto.serialize r
, Proto.serialize kd
, Proto.serialize af
]
formatMessage (AssistantUploadRequest r kd af) =
[ "au"
, Proto.serialize r
, Proto.serialize kd
, Proto.serialize af
]
formatMessage (AssistantDownloadRequest r kd af) =
[ "ad"
, Proto.serialize r
, Proto.serialize kd
, Proto.serialize af
]
instance Proto.Receivable TransferRequest where
parseCommand "u" = Proto.parse3 UploadRequest
parseCommand "d" = Proto.parse3 DownloadRequest
parseCommand "au" = Proto.parse3 AssistantUploadRequest
parseCommand "ad" = Proto.parse3 AssistantDownloadRequest
parseCommand _ = Proto.parseFail
instance Proto.Sendable TransferResponse where
formatMessage (TransferOutput (OutputMessage m)) =
["om", Proto.serialize (encode_c (decodeBS m))]
formatMessage (TransferOutput (OutputError e)) =
["oe", Proto.serialize (encode_c e)]
formatMessage (TransferOutput (StartProgressMeter (Just n))) =
["ops", Proto.serialize n]
formatMessage (TransferOutput (StartProgressMeter Nothing)) =
["opsx"]
formatMessage (TransferOutput (UpdateProgressMeter n)) =
["op", Proto.serialize n]
formatMessage (TransferOutput EndProgressMeter) =
["ope"]
formatMessage (TransferOutput StartPrompt) =
["oprs"]
formatMessage (TransferOutput EndPrompt) =
["opre"]
formatMessage (TransferOutput (JSONObject b)) =
["oj", Proto.serialize (encode_c (decodeBL b))]
formatMessage (TransferResult True) =
["t"]
formatMessage (TransferResult False) =
["f"]
instance Proto.Receivable TransferResponse where
parseCommand "om" = Proto.parse1 (TransferOutput . OutputMessage . encodeBS . decode_c)
parseCommand "oe" = Proto.parse1 (TransferOutput . OutputError . decode_c)
parseCommand "ops" = Proto.parse1 (TransferOutput . StartProgressMeter . Just)
parseCommand "opsx" = Proto.parse0 (TransferOutput (StartProgressMeter Nothing))
parseCommand "op" = Proto.parse1 (TransferOutput . UpdateProgressMeter)
parseCommand "ope" = Proto.parse0 (TransferOutput EndProgressMeter)
parseCommand "oprs" = Proto.parse0 (TransferOutput StartPrompt)
parseCommand "opre" = Proto.parse0 (TransferOutput EndPrompt)
parseCommand "oj" = Proto.parse1 (TransferOutput . JSONObject . encodeBL . decode_c)
parseCommand "t" = Proto.parse0 (TransferResult True)
parseCommand "f" = Proto.parse0 (TransferResult False)
parseCommand _ = Proto.parseFail
instance Proto.Sendable TransferSerializedOutputResponse where
formatMessage (TransferSerializedOutputResponse ReadyPrompt) = ["opr"]
instance Proto.Receivable TransferSerializedOutputResponse where
parseCommand "opr" = Proto.parse0 (TransferSerializedOutputResponse ReadyPrompt)
parseCommand _ = Proto.parseFail
instance Proto.Serializable TransferRemote where
serialize (TransferRemoteUUID u) = 'u':fromUUID u
-- A remote name could contain whitespace or newlines, which needs
-- to be escaped for the protocol. Use C-style encoding.
serialize (TransferRemoteName r) = 'r':encode_c' isSpace r
deserialize ('u':u) = Just (TransferRemoteUUID (toUUID u))
deserialize ('r':r) = Just (TransferRemoteName (decode_c r))
deserialize _ = Nothing
instance Proto.Serializable TransferAssociatedFile where
-- Comes last, so whitespace is ok. But, in case the filename
-- contains eg a newline, escape it. Use C-style encoding.
serialize (TransferAssociatedFile (AssociatedFile (Just f))) =
encode_c (fromRawFilePath f)
serialize (TransferAssociatedFile (AssociatedFile Nothing)) = ""
deserialize "" = Just $ TransferAssociatedFile $
AssociatedFile Nothing
deserialize s = Just $ TransferAssociatedFile $
AssociatedFile $ Just $ toRawFilePath $ decode_c s