async proto basically working
Simplified the protocol by removing END-ASYNC. There's a STM crash when a non-async protocol message is sent, which needs to be fixed.
This commit is contained in:
Joey Hess
parent
c9e8cafb98
commit
7546e686a2
4 changed files with 210 additions and 146 deletions
208
Remote/External/AsyncExtension.hs
vendored
208
Remote/External/AsyncExtension.hs
vendored
|
|
@ -1,16 +1,19 @@
|
|||
{- External remote protocol async extension.
|
||||
-
|
||||
- Copyright 2013-2020 Joey Hess <id@joeyh.name>
|
||||
- Copyright 2020 Joey Hess <id@joeyh.name>
|
||||
-
|
||||
- Licensed under the GNU AGPL version 3 or higher.
|
||||
-}
|
||||
|
||||
{-# LANGUAGE RankNTypes #-}
|
||||
{-# LANGUAGE BangPatterns #-}
|
||||
|
||||
module Remote.External.AsyncExtension where
|
||||
module Remote.External.AsyncExtension (runRelayToExternalAsync) where
|
||||
|
||||
import Common
|
||||
import Messages
|
||||
import Remote.External.Types
|
||||
import Utility.SimpleProtocol as Proto
|
||||
|
||||
import Control.Concurrent.Async
|
||||
import Control.Concurrent.STM
|
||||
|
|
@ -22,14 +25,29 @@ import qualified Data.Map.Strict as M
|
|||
-- process.
|
||||
runRelayToExternalAsync :: External -> ExternalState -> IO ExternalAsyncRelay
|
||||
runRelayToExternalAsync external st = do
|
||||
startcomm <- runRelayToExternalAsync' external st
|
||||
jidmap <- newTVarIO M.empty
|
||||
mapjid <- newTVarIO M.empty
|
||||
commcounter <- newTVarIO 0
|
||||
newconns <- newTVarIO []
|
||||
sendq <- newSendQueue
|
||||
void $ async $ sendloop st newconns mapjid jidmap sendq
|
||||
void $ async $ receiveloop external st newconns jidmap mapjid sendq
|
||||
return $ ExternalAsyncRelay $ do
|
||||
(sendh, receiveh, shutdown) <- startcomm
|
||||
n <- atomically $ do
|
||||
n <- readTVar commcounter
|
||||
let n' = succ n
|
||||
writeTVar commcounter n'
|
||||
return n'
|
||||
receiveq <- newReceiveQueue
|
||||
return $ ExternalState
|
||||
{ externalSend = atomically . writeTBMChan sendh
|
||||
, externalReceive = fmap join $ atomically $ readTBMChan receiveh
|
||||
{ externalSend = \msg ->
|
||||
atomically $ writeTBMChan sendq
|
||||
( toAsyncWrapped msg
|
||||
, (n, receiveq)
|
||||
)
|
||||
, externalReceive = atomically (readTBMChan receiveq)
|
||||
-- This shuts down the whole relay.
|
||||
, externalShutdown = shutdown
|
||||
, externalShutdown = shutdown external st sendq
|
||||
-- These three TVars are shared amoung all
|
||||
-- ExternalStates that use this relay; they're
|
||||
-- common state about the external process.
|
||||
|
|
@ -40,71 +58,59 @@ runRelayToExternalAsync external st = do
|
|||
, externalConfigChanges = externalConfigChanges st
|
||||
}
|
||||
|
||||
runRelayToExternalAsync'
|
||||
:: External
|
||||
-> ExternalState
|
||||
-> IO (IO (TBMChan String, TBMChan (Maybe String), Bool -> IO ()))
|
||||
runRelayToExternalAsync' external st = do
|
||||
newreqs <- newTVarIO []
|
||||
startedcomms <- newTVarIO []
|
||||
let startcomm = do
|
||||
toq <- newTBMChanIO 10
|
||||
fromq <- newTBMChanIO 10
|
||||
let c = (toq, fromq, shutdown)
|
||||
atomically $ do
|
||||
l <- readTVar startedcomms
|
||||
-- This append is ok because the maximum size
|
||||
-- is the number of jobs that git-annex is
|
||||
-- configured to use, which is a relatively
|
||||
-- small number.
|
||||
writeTVar startedcomms (l ++ [c])
|
||||
return c
|
||||
void $ async $ sendloop newreqs startedcomms
|
||||
void $ async $ receiveloop newreqs M.empty
|
||||
return startcomm
|
||||
type ReceiveQueue = TBMChan String
|
||||
|
||||
type SendQueue = TBMChan (AsyncWrapped, Conn)
|
||||
|
||||
type ConnNum = Integer
|
||||
|
||||
type Conn = (ConnNum, ReceiveQueue)
|
||||
|
||||
type NewConns = TVar [Conn]
|
||||
|
||||
type MapJid = TVar (M.Map ConnNum JobId)
|
||||
|
||||
type JidMap = TVar (M.Map JobId Conn)
|
||||
|
||||
newReceiveQueue :: IO ReceiveQueue
|
||||
newReceiveQueue = newTBMChanIO 10
|
||||
|
||||
newSendQueue :: IO SendQueue
|
||||
newSendQueue = newTBMChanIO 10
|
||||
|
||||
receiveloop :: External -> ExternalState -> NewConns -> JidMap -> MapJid -> SendQueue -> IO ()
|
||||
receiveloop external st newconns jidmap mapjid sendq = externalReceive st >>= \case
|
||||
Just l -> case parseMessage l :: Maybe AsyncMessage of
|
||||
Just (RESULT_ASYNC msg) -> getnext newconns >>= \case
|
||||
Just (_n, c) -> do
|
||||
relayto c msg
|
||||
loop
|
||||
Nothing -> abort "unexpected RESULT-ASYNC"
|
||||
Just (START_ASYNC jid) -> getnext newconns >>= \case
|
||||
Just v@(n, _c) -> do
|
||||
atomically $ do
|
||||
modifyTVar' jidmap $ M.insert jid v
|
||||
modifyTVar' mapjid $ M.insert n jid
|
||||
loop
|
||||
Nothing -> abort "unexpected START-ASYNC"
|
||||
Just (ASYNC jid msg) -> getjid jid >>= \case
|
||||
Just (_n, c) -> do
|
||||
relayto c msg
|
||||
loop
|
||||
Nothing -> abort "ASYNC with unknown jobid"
|
||||
_ -> abort "unexpected non-async message"
|
||||
Nothing -> do
|
||||
-- Unable to receive anything more from the
|
||||
-- process, so it's not usable any longer.
|
||||
m <- readTVarIO jidmap
|
||||
forM_ (M.elems m) (closerelayto . snd)
|
||||
shutdown external st sendq True
|
||||
where
|
||||
receiveloop newreqs jidmap = externalReceive st >>= \case
|
||||
Just l -> case parseMessage l :: Maybe AsyncMessage of
|
||||
Just (RESULT_ASYNC msg) -> getnext newreqs >>= \case
|
||||
Just c -> do
|
||||
relayto c msg
|
||||
receiveloop newreqs jidmap
|
||||
Nothing -> protoerr "unexpected RESULT-ASYNC"
|
||||
Just (START_ASYNC jid msg) -> getnext newreqs >>= \case
|
||||
Just c -> do
|
||||
relayto c msg
|
||||
let !jidmap' = M.insert jid c jidmap
|
||||
receiveloop newreqs jidmap'
|
||||
Nothing -> protoerr "unexpected START-ASYNC"
|
||||
Just (END_ASYNC jid msg) -> case M.lookup jid jidmap of
|
||||
Just c -> do
|
||||
relayto c msg
|
||||
closerelayto c
|
||||
let !jidmap' = M.delete jid jidmap
|
||||
receiveloop newreqs jidmap'
|
||||
Nothing -> protoerr "END-ASYNC with unknown jobid"
|
||||
Just (ASYNC jid msg) -> case M.lookup jid jidmap of
|
||||
Just c -> do
|
||||
relayto c msg
|
||||
let !jidmap' = M.delete jid jidmap
|
||||
receiveloop newreqs jidmap'
|
||||
Nothing -> protoerr "ASYNC with unknown jobid"
|
||||
_ -> protoerr "unexpected non-async message"
|
||||
Nothing -> do
|
||||
-- Unable to receive anything more from the
|
||||
-- process, so it's not usable any longer.
|
||||
forM_ (M.elems jidmap) closerelayto
|
||||
shutdown True
|
||||
loop = receiveloop external st newconns jidmap mapjid sendq
|
||||
|
||||
sendloop newreqs startedcomms = do
|
||||
error "TODO"
|
||||
relayto q msg = atomically $ writeTBMChan q msg
|
||||
|
||||
relayto (toq, _fromq) msg =
|
||||
atomically $ writeTBMChan toq msg
|
||||
|
||||
closerelayto (toq, fromq) = do
|
||||
atomically $ closeTBMChan toq
|
||||
atomically $ closeTBMChan fromq
|
||||
closerelayto q = atomically $ closeTBMChan q
|
||||
|
||||
getnext l = atomically $ readTVar l >>= \case
|
||||
[] -> return Nothing
|
||||
|
|
@ -112,14 +118,56 @@ runRelayToExternalAsync' external st = do
|
|||
writeTVar l rest
|
||||
return (Just c)
|
||||
|
||||
shutdown b = do
|
||||
r <- atomically $ do
|
||||
r <- tryTakeTMVar (externalAsync external)
|
||||
putTMVar (externalAsync external)
|
||||
UncheckedExternalAsync
|
||||
return r
|
||||
case r of
|
||||
Just (ExternalAsync _) -> externalShutdown st b
|
||||
_ -> noop
|
||||
|
||||
protoerr s = giveup ("async special remote protocol error: " ++ s)
|
||||
getjid jid = M.lookup jid <$> readTVarIO jidmap
|
||||
|
||||
abort s = do
|
||||
warningIO (protoerr s)
|
||||
shutdown external st sendq True
|
||||
|
||||
sendloop :: ExternalState -> NewConns -> MapJid -> JidMap -> SendQueue -> IO ()
|
||||
sendloop st newconns mapjid jidmap sendq = atomically (readTBMChan sendq) >>= \case
|
||||
Just (wrappedmsg, c@(n, _)) -> do
|
||||
let newconn = atomically $ do
|
||||
-- This append is not too expensive,
|
||||
-- because the list length is limited
|
||||
-- to the maximum number of jobs.
|
||||
modifyTVar' newconns (++[c])
|
||||
M.lookup n <$> readTVar mapjid >>= \case
|
||||
Nothing -> return ()
|
||||
Just jid -> do
|
||||
modifyTVar' jidmap (M.delete jid)
|
||||
modifyTVar' mapjid (M.delete n)
|
||||
case wrappedmsg of
|
||||
AsyncWrappedRequest msg -> do
|
||||
newconn
|
||||
externalSend st msg
|
||||
AsyncWrappedExceptionalMessage msg -> do
|
||||
newconn
|
||||
externalSend st msg
|
||||
AsyncWrappedRemoteResponse msg ->
|
||||
externalSend st =<< wrapremoteresponse msg n
|
||||
AsyncWrappedAsyncReply msg ->
|
||||
externalSend st msg
|
||||
sendloop st newconns mapjid jidmap sendq
|
||||
Nothing -> return ()
|
||||
where
|
||||
wrapremoteresponse msg n =
|
||||
M.lookup n <$> readTVarIO mapjid >>= \case
|
||||
Just jid -> return $ REPLY_ASYNC jid $
|
||||
unwords $ Proto.formatMessage msg
|
||||
Nothing -> error "failed to find jobid"
|
||||
|
||||
shutdown :: External -> ExternalState -> SendQueue -> Bool -> IO ()
|
||||
shutdown external st sendq b = do
|
||||
r <- atomically $ do
|
||||
r <- tryTakeTMVar (externalAsync external)
|
||||
putTMVar (externalAsync external)
|
||||
UncheckedExternalAsync
|
||||
return r
|
||||
case r of
|
||||
Just (ExternalAsync _) -> externalShutdown st b
|
||||
_ -> noop
|
||||
atomically $ closeTBMChan sendq
|
||||
|
||||
protoerr :: String -> String
|
||||
protoerr s = "async external special remote protocol error: " ++ s
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue