relayer receive loop is done

Receive loop looks right. Still need the send loop.

And, a complication is that some messages git-annex
sends need to be wrapped in REPLY_ASYNC, while others
do not. So will probably need to split externalSend
into two.

diff --git a/Remote/External.hs b/Remote/External.hs
index 5fabac8502daf445ec3cc73a9ee298b7a8d87f2e..ef9930b078555dd104bd8a34a165dc53ba28501c 100644 (file)
--- a/Remote/External.hs
+++ b/Remote/External.hs
@@ -615,7 +615,7 @@ startExternal external =
                        (st, extensions) <- startExternal' external
                        if asyncExtensionEnabled extensions
                                then do
-                                       v <- liftIO $ runRelayToExternalAsync st
+                                       v <- liftIO $ runRelayToExternalAsync external st
                                        st' <- liftIO $ relayToExternalAsync v
                                        store (ExternalAsync v)
                                        return st'
@@ -625,9 +625,9 @@ startExternal external =
                v@NoExternalAsync -> do
                        store v
                        fst <$> startExternal' external
-               v@(ExternalAsync ExternalAsyncRelay) -> do
+               v@(ExternalAsync relay) -> do
                        store v
-                       liftIO $ relayToExternalAsync v
+                       liftIO $ relayToExternalAsync relay
   where
        store = liftIO . atomically . putTMVar (externalAsync external)
 
@@ -677,7 +677,7 @@ startExternal' external = do
                exwanted <- receiveMessage st external
                        (\resp -> case resp of
                                EXTENSIONS_RESPONSE l -> result l
-                               UNSUPPORTED_REQUEST -> result []
+                               UNSUPPORTED_REQUEST -> result mempty
                                _ -> Nothing
                        )
                        (const Nothing)

diff --git a/Remote/External/AsyncExtension.hs b/Remote/External/AsyncExtension.hs
index fa4b535fc0ac032652bffc40710772305b0d9a6d..7d519a4766e8bef1b02ae12fba55b7fdb0b012b2 100644 (file)
--- a/Remote/External/AsyncExtension.hs
+++ b/Remote/External/AsyncExtension.hs
@@ -9,7 +9,7 @@
 
 module Remote.External.AsyncExtension where
 
-import Common.Annex
+import Common
 import Remote.External.Types
 
 import Control.Concurrent.Async
@@ -22,18 +22,18 @@ import qualified Data.Map.Strict as M
 relayToExternalAsync :: ExternalAsyncRelay -> IO ExternalState
 relayToExternalAsync relay = do
        n <- liftIO $ atomically $ do
-               v <- readTVar (asyncRelayLast relay)
+               v <- readTVar (asyncRelayLastId relay)
                let !n = succ v
-               writeTVar (asyncRelayLast relay) n
+               writeTVar (asyncRelayLastId relay) n
                return n
-       return $ asyncRelayExternalState n
+       asyncRelayExternalState relay 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
+runRelayToExternalAsync :: External -> ExternalState -> IO ExternalAsyncRelay
+runRelayToExternalAsync external st = do
+       startcomm <- runRelayToExternalAsync' external st
        pv <- atomically $ newTVar 1
        return $ ExternalAsyncRelay
                { asyncRelayLastId = pv
@@ -41,10 +41,10 @@ runRelayToExternalAsync st = do
                }
   where
        relaystate startcomm n = do
-               (sendh, receiveh, shutdownh) <- startcomm n
-               ExternalState
+               (sendh, receiveh, shutdownh) <- startcomm (ClientId n)
+               return $ ExternalState
                        { externalSend = atomically . writeTBMChan sendh
-                       , externalReceive = atomically . readTBMChan receiveh
+                       , externalReceive = fmap join $ atomically $ readTBMChan receiveh
                        , externalShutdown = atomically . writeTBMChan shutdownh
                        -- These three TVars are shared amoung all
                        -- ExternalStates that use this relay; they're
@@ -60,51 +60,71 @@ newtype ClientId = ClientId Int
        deriving (Show, Eq, Ord)
 
 runRelayToExternalAsync'
-       :: ExternalState
+       :: External
+       -> ExternalState
        -> IO (ClientId -> IO (TBMChan String, TBMChan (Maybe String), TBMChan Bool))
-runRelayToExternalAsync' st = do
-       let startcomm n = 
+runRelayToExternalAsync' external st = do
+       let startcomm n = error "TODO"
        sendt <- async sendloop
-       void $ async (receiveloop [] Nothing sendt)
+       newreqs <- newTVarIO []
+       void $ async (receiveloop newreqs M.empty sendt)
        return startcomm
   where
-       receiveloop newreqs currjid sendt = externalReceive st >>= \case
+       receiveloop newreqs jidmap 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
+                       Just (RESULT_ASYNC msg) -> getnext newreqs >>= \case
+                               Just c -> do
+                                       relayto c msg
+                                       receiveloop newreqs jidmap sendt
+                               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' sendt
+                               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' sendt
+                               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' sendt
+                               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.
                        -- So close all chans, stop the process,
                        -- and avoid any new ExternalStates from being
                        -- created using it.
+                       cancel sendt
                        atomically $ do
                                void $ tryTakeTMVar (externalAsync external) 
                                putTMVar (externalAsync external)
                                        UncheckedExternalAsync
-                       forM_ (M.toList receivers) $
-                               atomically . closeTBMChan
-                       forM_ (M.toList senders) $
-                               atomically . closeTBMChan
+                       forM_ (M.elems jidmap) closerelayto
                        externalShutdown st True
-                       cancel sendt
 
        sendloop = do
+               error "TODO"
+
+       relayto (toq, _fromq) msg = atomically $ writeTBMChan toq msg
+
+       closerelayto (toq, fromq) = do
+               atomically $ closeTBMChan toq
+               atomically $ closeTBMChan fromq
+
+       getnext l = atomically $ readTVar l >>= \case
+               [] -> return Nothing
+               (c:rest) -> do
+                       writeTVar l rest
+                       return (Just c)
+       
+       protoerr s = giveup ("async special remote protocol error: " ++ s)
+

diff --git a/Remote/External/Types.hs b/Remote/External/Types.hs
index a635550675b6cb9e7d8457dc87560e17a571f70c..6b2a6dfdd764e652f0c3cda6c476bfdaff2fd7e7 100644 (file)
--- a/Remote/External/Types.hs
+++ b/Remote/External/Types.hs
@@ -6,6 +6,7 @@
  -}
 
 {-# LANGUAGE FlexibleInstances, TypeSynonymInstances #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# OPTIONS_GHC -fno-warn-orphans #-}
 
 module Remote.External.Types (
@@ -14,8 +15,11 @@ module Remote.External.Types (
        ExternalType,
        ExternalState(..),
        PrepareStatus(..),
+       ExtensionList(..),
        supportedExtensionList,
        asyncExtensionEnabled,
+       ExternalAsync(..),
+       ExternalAsyncRelay(..),
        Proto.parseMessage,
        Proto.Sendable(..),
        Proto.Receivable(..),
@@ -27,6 +31,7 @@ module Remote.External.Types (
        RemoteRequest(..),
        RemoteResponse(..),
        ExceptionalMessage(..),
+       AsyncMessage(..),
        ErrorMsg,
        Setting,
        Description,
@@ -91,7 +96,7 @@ type PID = Int
 
 -- List of extensions to the protocol.
 newtype ExtensionList = ExtensionList { fromExtensionList :: [String] }
-       deriving (Show)
+       deriving (Show, Monoid, Semigroup)
 
 supportedExtensionList :: ExtensionList
 supportedExtensionList = ExtensionList ["INFO", asyncExtension]

diff --git a/doc/design/external_special_remote_protocol/async_appendix.mdwn b/doc/design/external_special_remote_protocol/async_appendix.mdwn
index 8119846a0bae7583b8bf1e9c18ed68f7863f1577..087054e9ed718c07f26235d21474c8dd4542945a 100644 (file)
--- a/doc/design/external_special_remote_protocol/async_appendix.mdwn
+++ b/doc/design/external_special_remote_protocol/async_appendix.mdwn
@@ -111,7 +111,9 @@ Here's the details about the additions to the protocol.
 * `END-ASYNC JobId ReplyMsg`  
   Indicates that an async job is complete. The ReplyMsg indicates the result
   of the job, and is anything that would be sent as a protocol reply in the
-  non-async protocol.
+  non-async protocol.  
+  After this, the JobId is not in use, an indeed the same value could be
+  reused by a new `START-ASYNC` if desired.
 * `RESULT-ASYNC ReplyMsg`  
   This is the same as sending `START-ASYNC` immediately followed by
   `END-ASYNC`. This is often used to respond to `PREPARE`, `LISTCONFIGS`,