new protocol for transferkeys, with message serialization
Necessarily threw out the old protocol, so if an old git-annex assistant is running, and starts a transferkeys from the new git-annex, it would fail. But, that seems unlikely; the assistant starts up transferkeys processes and then keeps them running. Still, may need to test that scenario. The new protocol is simple read/show and looks like this: TransferRequest Download (Right "origin") (Key {keyName = "f8f8766a836fb6120abf4d5328ce8761404e437529e997aaa0363bdd4fecd7bb", keyVariety = SHA2Key (HashSize 256) (HasExt True), keySize = Just 30, keyMtime = Nothing, keyChunkSize = Nothing, keyChunkNum = Nothing}) (AssociatedFile (Just "foo")) TransferOutput (ProgressMeter (Just 30) (MeterState {meterBytesProcessed = BytesProcessed 0, meterTimeStamp = 1.6070268727892535e9}) (MeterState {meterBytesProcessed = BytesProcessed 30, meterTimeStamp = 1.6070268728043e9})) TransferOutput (OutputMessage "(checksum...) ") TransferResult True Granted, this is not optimally fast, but it seems good enough, and is probably nearly as fast as the old protocol anyhow. emitSerializedOutput for ProgressMeter is not yet implemented. It needs to somehow start or update a progress meter. There may need to be a new message that allocates a progress meter, and then have ProgressMeter update it. This commit was sponsored by Ethan Aubin
This commit is contained in:
parent
82dbc4387c
commit
cad147cbbf
10 changed files with 98 additions and 96 deletions
|
@ -155,7 +155,7 @@ genTransfer t info = case transferRemote info of
|
||||||
- usual cleanup. However, first check if something else is
|
- usual cleanup. However, first check if something else is
|
||||||
- running the transfer, to avoid removing active transfers.
|
- running the transfer, to avoid removing active transfers.
|
||||||
-}
|
-}
|
||||||
go remote transferrer = ifM (liftIO $ performTransfer transferrer t info)
|
go remote transferrer = ifM (liftAnnex $ performTransfer transferrer t info)
|
||||||
( do
|
( do
|
||||||
case associatedFile info of
|
case associatedFile info of
|
||||||
AssociatedFile Nothing -> noop
|
AssociatedFile Nothing -> noop
|
||||||
|
|
|
@ -55,9 +55,9 @@ checkTransferrerPoolItem program batchmaker i = case i of
|
||||||
|
|
||||||
{- 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. -}
|
||||||
performTransfer :: Transferrer -> Transfer -> TransferInfo -> IO Bool
|
performTransfer :: Transferrer -> Transfer -> TransferInfo -> Annex Bool
|
||||||
performTransfer transferrer t info = catchBoolIO $ do
|
performTransfer transferrer t info = catchBoolIO $ do
|
||||||
T.sendRequest t info (transferrerWrite transferrer)
|
(liftIO $ T.sendRequest t info (transferrerWrite transferrer))
|
||||||
T.readResponse (transferrerRead transferrer)
|
T.readResponse (transferrerRead transferrer)
|
||||||
|
|
||||||
{- Starts a new git-annex transferkeys process, setting up handles
|
{- Starts a new git-annex transferkeys process, setting up handles
|
||||||
|
|
|
@ -1,15 +1,14 @@
|
||||||
{- git-annex command, used internally by assistant
|
{- git-annex command
|
||||||
-
|
-
|
||||||
- Copyright 2012, 2013 Joey Hess <id@joeyh.name>
|
- Copyright 2012-2020 Joey Hess <id@joeyh.name>
|
||||||
-
|
-
|
||||||
- Licensed under the GNU AGPL version 3 or higher.
|
- Licensed under the GNU AGPL version 3 or higher.
|
||||||
-}
|
-}
|
||||||
|
|
||||||
{-# LANGUAGE TypeSynonymInstances, FlexibleInstances #-}
|
|
||||||
|
|
||||||
module Command.TransferKeys where
|
module Command.TransferKeys where
|
||||||
|
|
||||||
import Command
|
import Command
|
||||||
|
import qualified Annex
|
||||||
import Annex.Content
|
import Annex.Content
|
||||||
import Logs.Location
|
import Logs.Location
|
||||||
import Annex.Transfer
|
import Annex.Transfer
|
||||||
|
@ -18,8 +17,19 @@ import Utility.SimpleProtocol (dupIoHandles)
|
||||||
import Git.Types (RemoteName)
|
import Git.Types (RemoteName)
|
||||||
import qualified Database.Keys
|
import qualified Database.Keys
|
||||||
import Annex.BranchState
|
import Annex.BranchState
|
||||||
|
import Types.Messages
|
||||||
|
import Types.Key
|
||||||
|
import Messages.Internal
|
||||||
|
|
||||||
data TransferRequest = TransferRequest Direction Remote Key AssociatedFile
|
import Text.Read (readMaybe)
|
||||||
|
|
||||||
|
data TransferRequest = TransferRequest Direction (Either UUID RemoteName) KeyData AssociatedFile
|
||||||
|
deriving (Show, Read)
|
||||||
|
|
||||||
|
data TransferResponse
|
||||||
|
= TransferOutput SerializedOutput
|
||||||
|
| TransferResult Bool
|
||||||
|
deriving (Show, Read)
|
||||||
|
|
||||||
cmd :: Command
|
cmd :: Command
|
||||||
cmd = command "transferkeys" SectionPlumbing "transfers keys"
|
cmd = command "transferkeys" SectionPlumbing "transfers keys"
|
||||||
|
@ -32,10 +42,12 @@ start :: CommandStart
|
||||||
start = do
|
start = do
|
||||||
enableInteractiveBranchAccess
|
enableInteractiveBranchAccess
|
||||||
(readh, writeh) <- liftIO dupIoHandles
|
(readh, writeh) <- liftIO dupIoHandles
|
||||||
|
Annex.setOutput $ SerializedOutput $
|
||||||
|
hPutStrLn writeh . show . TransferOutput
|
||||||
runRequests readh writeh runner
|
runRequests readh writeh runner
|
||||||
stop
|
stop
|
||||||
where
|
where
|
||||||
runner (TransferRequest direction remote key file)
|
runner (TransferRequest direction _ keydata file) remote
|
||||||
| direction == Upload = notifyTransfer direction file $
|
| 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
|
||||||
|
@ -58,82 +70,58 @@ 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
|
||||||
-> Handle
|
-> Handle
|
||||||
-> (TransferRequest -> Annex Bool)
|
-> (TransferRequest -> Remote -> Annex Bool)
|
||||||
-> Annex ()
|
-> Annex ()
|
||||||
runRequests readh writeh a = do
|
runRequests readh writeh a = go Nothing Nothing
|
||||||
liftIO $ hSetBuffering readh NoBuffering
|
|
||||||
go =<< readrequests
|
|
||||||
where
|
where
|
||||||
go (d:rn:k:f:rest) = do
|
go lastremoteoruuid lastremote = unlessM (liftIO $ hIsEOF readh) $ do
|
||||||
case (deserialize d, deserialize rn, deserialize k, deserialize f) of
|
l <- liftIO $ hGetLine readh
|
||||||
(Just direction, Just remotename, Just key, Just file) -> do
|
case readMaybe l of
|
||||||
mremote <- Remote.byName' remotename
|
Just tr@(TransferRequest _ remoteoruuid _ _) -> do
|
||||||
|
-- Often the same remote will be used
|
||||||
|
-- repeatedly, so cache the last one to
|
||||||
|
-- avoid looking up repeatedly.
|
||||||
|
mremote <- if lastremoteoruuid == Just remoteoruuid
|
||||||
|
then pure lastremote
|
||||||
|
else eitherToMaybe <$> Remote.byName'
|
||||||
|
(either fromUUID id remoteoruuid)
|
||||||
case mremote of
|
case mremote of
|
||||||
Left _ -> sendresult False
|
Just remote -> do
|
||||||
Right remote -> sendresult =<< a
|
sendresult =<< a tr remote
|
||||||
(TransferRequest direction remote key file)
|
go (Just remoteoruuid) mremote
|
||||||
_ -> sendresult False
|
Nothing -> protocolError l
|
||||||
go rest
|
Nothing -> protocolError l
|
||||||
go [] = noop
|
|
||||||
go [""] = noop
|
|
||||||
go v = error $ "transferkeys protocol error: " ++ show v
|
|
||||||
|
|
||||||
readrequests = liftIO $ split fieldSep <$> hGetContents readh
|
|
||||||
sendresult b = liftIO $ do
|
sendresult b = liftIO $ do
|
||||||
hPutStrLn writeh $ serialize b
|
hPutStrLn writeh $ show $ TransferResult b
|
||||||
hFlush writeh
|
hFlush writeh
|
||||||
|
|
||||||
sendRequest :: Transfer -> TransferInfo -> Handle -> IO ()
|
sendRequest :: Transfer -> TransferInfo -> Handle -> IO ()
|
||||||
sendRequest t tinfo h = do
|
sendRequest t tinfo h = hPutStrLn h $ show $ TransferRequest
|
||||||
hPutStr h $ intercalate fieldSep
|
(transferDirection t)
|
||||||
[ serialize (transferDirection t)
|
(maybe (Left (transferUUID t)) (Right . Remote.name) (transferRemote tinfo))
|
||||||
, maybe (serialize ((fromUUID (transferUUID t)) :: String))
|
(keyData (transferKey t))
|
||||||
(serialize . Remote.name)
|
(associatedFile tinfo)
|
||||||
(transferRemote tinfo)
|
|
||||||
, serialize (transferKey t)
|
|
||||||
, serialize (associatedFile tinfo)
|
|
||||||
, "" -- adds a trailing null
|
|
||||||
]
|
|
||||||
hFlush h
|
|
||||||
|
|
||||||
readResponse :: Handle -> IO Bool
|
-- | Read a response from this command.
|
||||||
readResponse h = fromMaybe False . deserialize <$> hGetLine h
|
--
|
||||||
|
-- Each TransferOutput line that is read before the final TransferResult
|
||||||
|
-- will be output.
|
||||||
|
readResponse :: Handle -> Annex Bool
|
||||||
|
readResponse h = do
|
||||||
|
l <- liftIO $ hGetLine h
|
||||||
|
case readMaybe l of
|
||||||
|
Just (TransferOutput so) -> do
|
||||||
|
emitSerializedOutput so
|
||||||
|
readResponse h
|
||||||
|
Just (TransferResult r) -> return r
|
||||||
|
Nothing -> protocolError l
|
||||||
|
|
||||||
fieldSep :: String
|
protocolError :: String -> a
|
||||||
fieldSep = "\0"
|
protocolError l = error $ "transferkeys protocol error: " ++ show l
|
||||||
|
|
||||||
class TCSerialized a where
|
|
||||||
serialize :: a -> String
|
|
||||||
deserialize :: String -> Maybe a
|
|
||||||
|
|
||||||
instance TCSerialized Bool where
|
|
||||||
serialize True = "1"
|
|
||||||
serialize False = "0"
|
|
||||||
deserialize "1" = Just True
|
|
||||||
deserialize "0" = Just False
|
|
||||||
deserialize _ = Nothing
|
|
||||||
|
|
||||||
instance TCSerialized Direction where
|
|
||||||
serialize Upload = "u"
|
|
||||||
serialize Download = "d"
|
|
||||||
deserialize "u" = Just Upload
|
|
||||||
deserialize "d" = Just Download
|
|
||||||
deserialize _ = Nothing
|
|
||||||
|
|
||||||
instance TCSerialized AssociatedFile where
|
|
||||||
serialize (AssociatedFile (Just f)) = fromRawFilePath f
|
|
||||||
serialize (AssociatedFile Nothing) = ""
|
|
||||||
deserialize "" = Just (AssociatedFile Nothing)
|
|
||||||
deserialize f = Just (AssociatedFile (Just (toRawFilePath f)))
|
|
||||||
|
|
||||||
instance TCSerialized RemoteName where
|
|
||||||
serialize n = n
|
|
||||||
deserialize n = Just n
|
|
||||||
|
|
||||||
instance TCSerialized Key where
|
|
||||||
serialize = serializeKey
|
|
||||||
deserialize = deserializeKey
|
|
||||||
|
|
|
@ -288,7 +288,7 @@ commandProgressDisabled = withMessageState $ \s -> return $
|
||||||
NormalOutput -> concurrentOutputEnabled s
|
NormalOutput -> concurrentOutputEnabled s
|
||||||
QuietOutput -> True
|
QuietOutput -> True
|
||||||
JSONOutput _ -> True
|
JSONOutput _ -> True
|
||||||
SerializedOutput -> True
|
SerializedOutput _ -> True
|
||||||
|
|
||||||
jsonOutputEnabled :: Annex Bool
|
jsonOutputEnabled :: Annex Bool
|
||||||
jsonOutputEnabled = withMessageState $ \s -> return $
|
jsonOutputEnabled = withMessageState $ \s -> return $
|
||||||
|
|
|
@ -29,15 +29,15 @@ outputMessage' jsonoutputter jsonbuilder msg = withMessageState $ \s -> case out
|
||||||
| otherwise -> liftIO $ flushed $ S.putStr msg
|
| otherwise -> liftIO $ flushed $ S.putStr msg
|
||||||
JSONOutput _ -> void $ jsonoutputter jsonbuilder s
|
JSONOutput _ -> void $ jsonoutputter jsonbuilder s
|
||||||
QuietOutput -> q
|
QuietOutput -> q
|
||||||
SerializedOutput -> do
|
SerializedOutput h -> do
|
||||||
liftIO $ outputSerialized $ OutputMessage msg
|
liftIO $ outputSerialized h $ OutputMessage msg
|
||||||
void $ jsonoutputter jsonbuilder s
|
void $ jsonoutputter jsonbuilder s
|
||||||
|
|
||||||
-- Buffer changes to JSON until end is reached and then emit it.
|
-- Buffer changes to JSON until end is reached and then emit it.
|
||||||
bufferJSON :: JSONBuilder -> MessageState -> Annex Bool
|
bufferJSON :: JSONBuilder -> MessageState -> Annex Bool
|
||||||
bufferJSON jsonbuilder s = case outputType s of
|
bufferJSON jsonbuilder s = case outputType s of
|
||||||
JSONOutput _ -> go (flushed . JSON.emit)
|
JSONOutput _ -> go (flushed . JSON.emit)
|
||||||
SerializedOutput -> go (outputSerialized . JSONObject . JSON.encode)
|
SerializedOutput h -> go (outputSerialized h . JSONObject . JSON.encode)
|
||||||
_ -> return False
|
_ -> return False
|
||||||
where
|
where
|
||||||
go emitter
|
go emitter
|
||||||
|
@ -63,7 +63,7 @@ bufferJSON jsonbuilder s = case outputType s of
|
||||||
outputJSON :: JSONBuilder -> MessageState -> Annex Bool
|
outputJSON :: JSONBuilder -> MessageState -> Annex Bool
|
||||||
outputJSON jsonbuilder s = case outputType s of
|
outputJSON jsonbuilder s = case outputType s of
|
||||||
JSONOutput _ -> go (flushed . JSON.emit)
|
JSONOutput _ -> go (flushed . JSON.emit)
|
||||||
SerializedOutput -> go (outputSerialized . JSONObject . JSON.encode)
|
SerializedOutput h -> go (outputSerialized h . JSONObject . JSON.encode)
|
||||||
_ -> return False
|
_ -> return False
|
||||||
where
|
where
|
||||||
go emitter = do
|
go emitter = do
|
||||||
|
@ -77,8 +77,8 @@ outputError msg = withMessageState $ \s -> case (outputType s, jsonBuffer s) of
|
||||||
let jb' = Just (JSON.addErrorMessage (lines msg) jb)
|
let jb' = Just (JSON.addErrorMessage (lines msg) jb)
|
||||||
in Annex.changeState $ \st ->
|
in Annex.changeState $ \st ->
|
||||||
st { Annex.output = s { jsonBuffer = jb' } }
|
st { Annex.output = s { jsonBuffer = jb' } }
|
||||||
(SerializedOutput, _) ->
|
(SerializedOutput h, _) ->
|
||||||
liftIO $ outputSerialized $ OutputError msg
|
liftIO $ outputSerialized h $ OutputError msg
|
||||||
_
|
_
|
||||||
| concurrentOutputEnabled s -> concurrentMessage s True msg go
|
| concurrentOutputEnabled s -> concurrentMessage s True msg go
|
||||||
| otherwise -> go
|
| otherwise -> go
|
||||||
|
@ -94,5 +94,20 @@ q = noop
|
||||||
flushed :: IO () -> IO ()
|
flushed :: IO () -> IO ()
|
||||||
flushed a = a >> hFlush stdout
|
flushed a = a >> hFlush stdout
|
||||||
|
|
||||||
outputSerialized :: SerializedOutput -> IO ()
|
outputSerialized :: (SerializedOutput -> IO ()) -> SerializedOutput -> IO ()
|
||||||
outputSerialized = print
|
outputSerialized = id
|
||||||
|
|
||||||
|
emitSerializedOutput :: SerializedOutput -> Annex ()
|
||||||
|
emitSerializedOutput (OutputMessage msg) =
|
||||||
|
outputMessage' nojsonoutputter nojsonbuilder msg
|
||||||
|
where
|
||||||
|
nojsonoutputter _ _ = return False
|
||||||
|
nojsonbuilder = id
|
||||||
|
emitSerializedOutput (OutputError msg) = outputError msg
|
||||||
|
emitSerializedOutput (ProgressMeter sz old new) = undefined -- TODO
|
||||||
|
emitSerializedOutput (JSONObject b) =
|
||||||
|
withMessageState $ \s -> case outputType s of
|
||||||
|
JSONOutput _ -> liftIO $ flushed $ JSON.emit' b
|
||||||
|
SerializedOutput h -> liftIO $
|
||||||
|
outputSerialized h $ JSONObject b
|
||||||
|
_ -> q
|
||||||
|
|
|
@ -11,6 +11,7 @@ module Messages.JSON (
|
||||||
JSONBuilder,
|
JSONBuilder,
|
||||||
JSONChunk(..),
|
JSONChunk(..),
|
||||||
emit,
|
emit,
|
||||||
|
emit',
|
||||||
encode,
|
encode,
|
||||||
none,
|
none,
|
||||||
start,
|
start,
|
||||||
|
@ -52,9 +53,12 @@ emitLock :: MVar ()
|
||||||
emitLock = unsafePerformIO $ newMVar ()
|
emitLock = unsafePerformIO $ newMVar ()
|
||||||
|
|
||||||
emit :: Object -> IO ()
|
emit :: Object -> IO ()
|
||||||
emit o = do
|
emit = emit' . encode
|
||||||
|
|
||||||
|
emit' :: L.ByteString -> IO ()
|
||||||
|
emit' b = do
|
||||||
takeMVar emitLock
|
takeMVar emitLock
|
||||||
L.hPut stdout (encode o)
|
L.hPut stdout b
|
||||||
putStr "\n"
|
putStr "\n"
|
||||||
putMVar emitLock ()
|
putMVar emitLock ()
|
||||||
|
|
||||||
|
|
|
@ -95,9 +95,9 @@ metered othermeter sizer a = withMessageState $ \st ->
|
||||||
updateMeter meter
|
updateMeter meter
|
||||||
a meter (combinemeter m)
|
a meter (combinemeter m)
|
||||||
| otherwise = nometer
|
| otherwise = nometer
|
||||||
go msize (MessageState { outputType = SerializedOutput }) = do
|
go msize (MessageState { outputType = SerializedOutput h }) = do
|
||||||
meter <- liftIO $ mkMeter msize $ \_ msize' old new ->
|
meter <- liftIO $ mkMeter msize $ \_ msize' old new ->
|
||||||
outputSerialized $ ProgressMeter msize' old new
|
outputSerialized h $ ProgressMeter msize' old new
|
||||||
m <- liftIO $ rateLimitMeterUpdate minratelimit meter $
|
m <- liftIO $ rateLimitMeterUpdate minratelimit meter $
|
||||||
updateMeter meter
|
updateMeter meter
|
||||||
a meter (combinemeter m)
|
a meter (combinemeter m)
|
||||||
|
|
|
@ -11,6 +11,7 @@ module Types.Key (
|
||||||
KeyData(..),
|
KeyData(..),
|
||||||
Key,
|
Key,
|
||||||
fromKey,
|
fromKey,
|
||||||
|
keyData,
|
||||||
mkKey,
|
mkKey,
|
||||||
alterKey,
|
alterKey,
|
||||||
isKeyPrefix,
|
isKeyPrefix,
|
||||||
|
@ -201,7 +202,7 @@ splitKeyNameExtension' keyname = S8.span (/= '.') keyname
|
||||||
|
|
||||||
{- A filename may be associated with a Key. -}
|
{- A filename may be associated with a Key. -}
|
||||||
newtype AssociatedFile = AssociatedFile (Maybe RawFilePath)
|
newtype AssociatedFile = AssociatedFile (Maybe RawFilePath)
|
||||||
deriving (Show, Eq, Ord)
|
deriving (Show, Read, Eq, Ord)
|
||||||
|
|
||||||
{- There are several different varieties of keys. -}
|
{- There are several different varieties of keys. -}
|
||||||
data KeyVariety
|
data KeyVariety
|
||||||
|
|
|
@ -19,8 +19,7 @@ data OutputType
|
||||||
= NormalOutput
|
= NormalOutput
|
||||||
| QuietOutput
|
| QuietOutput
|
||||||
| JSONOutput JSONOptions
|
| JSONOutput JSONOptions
|
||||||
| SerializedOutput
|
| SerializedOutput (SerializedOutput -> IO ())
|
||||||
deriving (Show)
|
|
||||||
|
|
||||||
data JSONOptions = JSONOptions
|
data JSONOptions = JSONOptions
|
||||||
{ jsonProgress :: Bool
|
{ jsonProgress :: Bool
|
||||||
|
|
|
@ -8,20 +8,15 @@ git annex transferkeys
|
||||||
|
|
||||||
# DESCRIPTION
|
# DESCRIPTION
|
||||||
|
|
||||||
This plumbing-level command is used by the assistant to transfer data.
|
This plumbing-level command is used to transfer data.
|
||||||
It is a long-running process, which is fed instructions about the keys
|
It is a long-running process, which is fed instructions about the keys
|
||||||
to transfer using an internal stdio protocol, which is
|
to transfer using an internal stdio protocol, which is
|
||||||
intentionally not documented (as it may change at any time).
|
intentionally not documented (as it may change at any time).
|
||||||
|
|
||||||
It's normal to have a transferkeys process running when the assistant is
|
|
||||||
running.
|
|
||||||
|
|
||||||
# SEE ALSO
|
# SEE ALSO
|
||||||
|
|
||||||
[[git-annex]](1)
|
[[git-annex]](1)
|
||||||
|
|
||||||
[[git-annex-assistant]](1)
|
|
||||||
|
|
||||||
# AUTHOR
|
# AUTHOR
|
||||||
|
|
||||||
Joey Hess <id@joeyh.name>
|
Joey Hess <id@joeyh.name>
|
||||||
|
|
Loading…
Reference in a new issue