From 06a4ab39fae86e6a509d4ec330559dd5f5ccb8f4 Mon Sep 17 00:00:00 2001 From: Joey Hess Date: Wed, 12 Aug 2020 15:17:53 -0400 Subject: [PATCH] wip external remote async protocol extension --- Remote/External.hs | 49 +++++++++++-- Remote/External/AsyncExtension.hs | 110 ++++++++++++++++++++++++++++++ Remote/External/Types.hs | 42 +++++++++--- git-annex.cabal | 1 + 4 files changed, 187 insertions(+), 15 deletions(-) create mode 100644 Remote/External/AsyncExtension.hs diff --git a/Remote/External.hs b/Remote/External.hs index 2eb3bcfd5b..5fabac8502 100644 --- a/Remote/External.hs +++ b/Remote/External.hs @@ -10,6 +10,7 @@ module Remote.External (remote) where import Remote.External.Types +import Remote.External.AsyncExtension import qualified Annex import Annex.Common import qualified Annex.ExternalAddonProcess as AddonProcess @@ -601,9 +602,37 @@ withExternalState external a = do put st = liftIO $ atomically $ modifyTVar' v (st:) {- Starts an external remote process running, and checks VERSION and - - exchanges EXTENSIONS. -} + - exchanges EXTENSIONS. + - + - When the ASYNC extension is negotiated, a single process is used, + - and this constructs a external state that communicates with a thread + - that relays to it. + -} startExternal :: External -> Annex ExternalState -startExternal external = do +startExternal external = + liftIO (atomically $ takeTMVar (externalAsync external)) >>= \case + UncheckedExternalAsync -> do + (st, extensions) <- startExternal' external + if asyncExtensionEnabled extensions + then do + v <- liftIO $ runRelayToExternalAsync st + st' <- liftIO $ relayToExternalAsync v + store (ExternalAsync v) + return st' + else do + store NoExternalAsync + return st + v@NoExternalAsync -> do + store v + fst <$> startExternal' external + v@(ExternalAsync ExternalAsyncRelay) -> do + store v + liftIO $ relayToExternalAsync v + where + store = liftIO . atomically . putTMVar (externalAsync external) + +startExternal' :: External -> Annex (ExternalState, ExtensionList) +startExternal' external = do pid <- liftIO $ atomically $ do n <- succ <$> readTVar (externalLastPid external) writeTVar (externalLastPid external) n @@ -632,8 +661,8 @@ startExternal external = do , externalConfig = cv , externalConfigChanges = ccv } - startproto st - return st + extensions <- startproto st + return (st, extensions) where basecmd = "git-annex-remote-" ++ externalType external startproto st = do @@ -645,14 +674,20 @@ startExternal external = do -- It responds with a EXTENSIONS_RESPONSE; that extensions -- list is reserved for future expansion. UNSUPPORTED_REQUEST -- is also accepted. - receiveMessage st external + exwanted <- receiveMessage st external (\resp -> case resp of - EXTENSIONS_RESPONSE _ -> result () - UNSUPPORTED_REQUEST -> result () + EXTENSIONS_RESPONSE l -> result l + UNSUPPORTED_REQUEST -> result [] _ -> Nothing ) (const Nothing) (const Nothing) + case filter (`notElem` fromExtensionList supportedExtensionList) (fromExtensionList exwanted) of + [] -> return exwanted + exrest -> giveup $ unwords $ + [ basecmd + , "requested extensions that this version of git-annex does not support:" + ] ++ exrest stopExternal :: External -> Annex () stopExternal external = liftIO $ do diff --git a/Remote/External/AsyncExtension.hs b/Remote/External/AsyncExtension.hs new file mode 100644 index 0000000000..fa4b535fc0 --- /dev/null +++ b/Remote/External/AsyncExtension.hs @@ -0,0 +1,110 @@ +{- External remote protocol async extension. + - + - Copyright 2013-2020 Joey Hess + - + - Licensed under the GNU AGPL version 3 or higher. + -} + +{-# LANGUAGE BangPatterns #-} + +module Remote.External.AsyncExtension where + +import Common.Annex +import Remote.External.Types + +import Control.Concurrent.Async +import Control.Concurrent.STM +import Control.Concurrent.STM.TBMChan +import qualified Data.Map.Strict as M + +-- | Constructs an ExternalState that can be used to communicate with +-- the external process via the relay. +relayToExternalAsync :: ExternalAsyncRelay -> IO ExternalState +relayToExternalAsync relay = do + n <- liftIO $ atomically $ do + v <- readTVar (asyncRelayLast relay) + let !n = succ v + writeTVar (asyncRelayLast relay) n + return n + return $ asyncRelayExternalState n + +-- | Starts a thread that will handle all communication with the external +-- process. The input ExternalState communicates directly with the external +-- process. +runRelayToExternalAsync :: ExternalState -> IO ExternalAsyncRelay +runRelayToExternalAsync st = do + startcomm <- runRelayToExternalAsync' st + pv <- atomically $ newTVar 1 + return $ ExternalAsyncRelay + { asyncRelayLastId = pv + , asyncRelayExternalState = relaystate startcomm + } + where + relaystate startcomm n = do + (sendh, receiveh, shutdownh) <- startcomm n + ExternalState + { externalSend = atomically . writeTBMChan sendh + , externalReceive = atomically . readTBMChan receiveh + , externalShutdown = atomically . writeTBMChan shutdownh + -- These three TVars are shared amoung all + -- ExternalStates that use this relay; they're + -- common state about the external process. + -- TODO: ALL code using these in Remote.External + -- has to be made async-safe. + , externalPrepared = externalPrepared st + , externalConfig = externalConfig st + , externalConfigChanges = externalConfigChanges st + } + +newtype ClientId = ClientId Int + deriving (Show, Eq, Ord) + +runRelayToExternalAsync' + :: ExternalState + -> IO (ClientId -> IO (TBMChan String, TBMChan (Maybe String), TBMChan Bool)) +runRelayToExternalAsync' st = do + let startcomm n = + sendt <- async sendloop + void $ async (receiveloop [] Nothing sendt) + return startcomm + where + receiveloop newreqs currjid sendt = externalReceive st >>= \case + Just l -> case parseMessage l :: Maybe AsyncMessage of + Just (START_ASYNC jid) -> case newreqs of + [] -> giveup "async special remote protocol error: unexpected START-ASYNC" + (c:newreqs') -> do + let !receiverjids' = M.insert jid c receiverjids + receiveloop newreqs' Nothing receiverjids' sendt + Just (END_ASYNC jid) -> do + let !receiverjids' = M.delete jid receiverjids + receiveloop newreqs (Just jid) receiverjids' sendt + Just (UPDATE_ASYNC jid) -> + receiveloop newreqs (Just jid) receiverjids sendt + Nothing -> case currjid of + Just jid -> + -- + Nothing -> case newreqs of + [] -> giveup "async special remote protocol error: unexpected non-async message" + (c:_) -> do + case M.lookup c receivers of + Just c -> atomically $ writeTBMChan c l + Nothing -> return () + receiveloop newreqs Nothing sendt + Nothing -> do + -- Unable to receive anything more from the + -- process, so it's not usable any longer. + -- So close all chans, stop the process, + -- and avoid any new ExternalStates from being + -- created using it. + atomically $ do + void $ tryTakeTMVar (externalAsync external) + putTMVar (externalAsync external) + UncheckedExternalAsync + forM_ (M.toList receivers) $ + atomically . closeTBMChan + forM_ (M.toList senders) $ + atomically . closeTBMChan + externalShutdown st True + cancel sendt + + sendloop = do diff --git a/Remote/External/Types.hs b/Remote/External/Types.hs index 46f8575839..a635550675 100644 --- a/Remote/External/Types.hs +++ b/Remote/External/Types.hs @@ -15,6 +15,7 @@ module Remote.External.Types ( ExternalState(..), PrepareStatus(..), supportedExtensionList, + asyncExtensionEnabled, Proto.parseMessage, Proto.Sendable(..), Proto.Receivable(..), @@ -60,6 +61,7 @@ data External = External , externalDefaultConfig :: ParsedRemoteConfig , externalGitConfig :: Maybe RemoteGitConfig , externalRemoteStateHandle :: Maybe RemoteStateHandle + , externalAsync :: TMVar ExternalAsync } newExternal :: ExternalType -> Maybe UUID -> ParsedRemoteConfig -> Maybe RemoteGitConfig -> Maybe RemoteStateHandle -> Annex External @@ -71,6 +73,7 @@ newExternal externaltype u c gc rs = liftIO $ External <*> pure c <*> pure gc <*> pure rs + <*> atomically (newTMVar UncheckedExternalAsync) type ExternalType = String @@ -87,11 +90,29 @@ data ExternalState type PID = Int -- List of extensions to the protocol. -newtype ExtensionList = ExtensionList [String] +newtype ExtensionList = ExtensionList { fromExtensionList :: [String] } deriving (Show) supportedExtensionList :: ExtensionList -supportedExtensionList = ExtensionList ["INFO", "ASYNC"] +supportedExtensionList = ExtensionList ["INFO", asyncExtension] + +asyncExtension :: String +asyncExtension = "ASYNC" + +asyncExtensionEnabled :: ExtensionList -> Bool +asyncExtensionEnabled l = asyncExtension `elem` fromExtensionList l + +-- When the async extension is in use, a single external process +-- is started and used for all requests. +data ExternalAsync + = ExternalAsync ExternalAsyncRelay + | NoExternalAsync + | UncheckedExternalAsync + +data ExternalAsyncRelay = ExternalAsyncRelay + { asyncRelayLastId :: TVar Int + , asyncRelayExternalState :: Int -> IO ExternalState + } data PrepareStatus = Unprepared | Prepared | FailedPrepare ErrorMsg @@ -337,14 +358,18 @@ instance Proto.Receivable ExceptionalMessage where -- Messages used by the async protocol extension. data AsyncMessage - = START_ASYNC JobId - | END_ASYNC JobId - | UPDATE_ASYNC JobId + = START_ASYNC JobId WrappedMsg + | END_ASYNC JobId WrappedMsg + | RESULT_ASYNC WrappedMsg + | ASYNC JobId WrappedMsg + | REPLY_ASYNC JobId WrappedMsg instance Proto.Receivable AsyncMessage where - parseCommand "START-ASYNC" = Proto.parse1 START_ASYNC - parseCommand "END-ASYNC" = Proto.parse1 END_ASYNC - parseCommand "UPDATE-ASYNC" = Proto.parse1 UPDATE_ASYNC + parseCommand "START-ASYNC" = Proto.parse2 START_ASYNC + parseCommand "END-ASYNC" = Proto.parse2 END_ASYNC + parseCommand "RESULT-ASYNC" = Proto.parse1 RESULT_ASYNC + parseCommand "ASYNC" = Proto.parse2 ASYNC + parseCommand "REPLY-ASYNC" = Proto.parse2 REPLY_ASYNC parseCommand _ = Proto.parseFail -- Data types used for parameters when communicating with the remote. @@ -355,6 +380,7 @@ type Description = String type ProtocolVersion = Int type Size = Maybe Integer type JobId = String +type WrappedMsg = String supportedProtocolVersions :: [ProtocolVersion] supportedProtocolVersions = [1] diff --git a/git-annex.cabal b/git-annex.cabal index 6709feb713..05ef916a5a 100644 --- a/git-annex.cabal +++ b/git-annex.cabal @@ -961,6 +961,7 @@ Executable git-annex Remote.Directory Remote.Directory.LegacyChunked Remote.External + Remote.External.AsyncExtension Remote.External.Types Remote.GCrypt Remote.Git -- 2.30.2