c95cb38 · curryer · git.iko.soy

git.iko.soy/curryer

add server-side timeout

commitc95cb384aaf0cb095eb6eb3f98893c39d57e6dda authorAgentM <agentm@themactionfaction.com> dateSun 11 Oct 2020, 13:43 (6 years ago) parent9f87d95

2 files changed+70−40

compare with master

unifiedsplit

src/Network/RPC/Curryer/Client.hs+20−11 src/Network/RPC/Curryer/Server.hs+50−29

src/Network/RPC/Curryer/Client.hs+20−11

@@ -11,7 +11,7 @@ import GHC.Conc

1111import Data.Time.Clock

1212import System.Timeout

1313

14−type SyncMap a = STMMap.Map UUID (MVar a, UTCTime)

14+type SyncMap a = STMMap.Map UUID (MVar (Either ConnectionError a), UTCTime)

1515-- the request map holds

1616data Connection a = Connection Socket (Async ()) (SyncMap a)

1717

@@ -39,21 +39,30 @@ clientAsync :: Serialise b =>

3939 IO ()

4040clientAsync sock syncmap asyncHandler = do

4141 -- ping proper thread to continue

42+ let findRequest requestId =

43+ atomically $ do

44+ val' <- STMMap.lookup requestId syncmap

45+ STMMap.delete requestId syncmap

46+ pure val'

47+

4248 let responseHandler responseMsg = do

4349 putStrLn "client-side message handler"

4450 case responseMsg of

4551 Response requestId val -> do

46− varval <- atomically $ do

47− val' <- STMMap.lookup requestId syncmap

48− STMMap.delete requestId syncmap

49− pure val'

52+ varval <- findRequest requestId

5053 case varval of

5154 Nothing -> error "dumped unrequested response"

52− Just (mVar,_) -> putMVar mVar val

55+ Just (mVar,_) -> putMVar mVar (Right val)

5356 AsyncRequest _ asyncMsg ->

5457 asyncHandler asyncMsg

55− ResponseExpectedRequest _ _ -> error "dumped response expected request"

58+ ResponseExpectedRequest{} -> error "dumped response expected request"

5659 ExceptionResponse _ -> error "TODO Exception"

60+ TimedOutResponse requestId -> do

61+ varval <- findRequest requestId

62+ case varval of

63+ Nothing -> error "dumped unrequested timeout response"

64+ Just (mVar,_) -> putMVar mVar (Left TimeoutError)

65+

5766 drainSocketMessages sock responseHandler

5867

5968call :: (Serialise request, Serialise response) => Connection response -> request -> IO (Either ConnectionError response)

@@ -67,18 +76,18 @@ callTimeout mTimeout (Connection sock _ mVarMap) msg = do

6776 responseMVar <- newEmptyMVar

6877 now <- getCurrentTime

6978 atomically $ STMMap.insert (responseMVar, now) requestID mVarMap

70− sendMessage (ResponseExpectedRequest requestID msg) sock

79+ sendMessage (ResponseExpectedRequest requestID mTimeout msg) sock

7180 let timeoutMicroseconds =

7281 case mTimeout of

73− Just timeout' -> timeout'

82+ Just timeout' -> timeout' + 100 --add 100 ms to account for unknown network latency

7483 Nothing -> -1

7584 mResponse <- timeout timeoutMicroseconds (takeMVar responseMVar)

7685 atomically $ STMMap.delete requestID mVarMap

7786 case mResponse of

7887 --timeout

79− Nothing -> do

88+ Nothing ->

8089 pure (Left TimeoutError)

81− Just response -> pure (Right response)

90+ Just res -> pure res

8291

8392asyncCall :: (Serialise request, Serialise response) => Connection response -> request -> IO (Either ConnectionError ())

8493asyncCall (Connection sock _ _) msg = do

src/Network/RPC/Curryer/Server.hs+50−29

@@ -1,10 +1,10 @@

1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving #-}

1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications #-}

22{- HLINT ignore "Use lambda-case" -}

33module Network.RPC.Curryer.Server where

44import Streamly

55import qualified Streamly.Prelude as S

66import Streamly.Network.Socket

7−import Streamly.Internal.Network.Socket (handleWithM, writeChunk)

7+import Streamly.Internal.Network.Socket (handleWithM)

88import Network.Socket as Socket

99import Network.Socket.ByteString as Socket

1010import Streamly.Internal.Data.Parser.ParserD as PD

@@ -30,6 +30,7 @@ import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA

3030import qualified Network.RPC.Curryer.StreamlyAdditions as SA

3131import Control.Monad

3232import Data.Hashable

33+import System.Timeout

3334

3435-- for toArrayS conversion

3536import qualified Data.ByteString.Internal as BSI

@@ -53,9 +54,12 @@ data ServerHelloMessage = ServerHelloMessage { serverServices :: [String],

5354--client-side timeout

5455--server-to-client and client-to-server async request (no response requested)

5556

57+type Timeout = Int

58+

5659data Message a = Response UUID a

5760 | AsyncRequest UUID a

58− | ResponseExpectedRequest UUID a

61+ | ResponseExpectedRequest UUID (Maybe Timeout) a

62+ | TimedOutResponse UUID

5963 -- | Services

6064 | ExceptionResponse String

6165 deriving (Generic, Show)

@@ -103,12 +107,12 @@ messageBoundaryP :: Parser IO Word8 BS.ByteString

103107messageBoundaryP = do

104108 let x = FL.toList

105109 w4x8 <- PD.take 4 x

106− traceShowM ("w4x8", w4x8)

110+ traceShowM ("w4x8"::String, w4x8)

107111 let c = fromIntegral (fromOctets w4x8)

108− traceShowM ("c", c)

112+ traceShowM ("c"::String, c)

109113 vals <- PD.take c x

110114 let bytes = BS.pack vals

111− traceShowM ("parsedBytes", c, BS.length bytes, bytes)

115+ traceShowM ("parsedBytes"::String, c, BS.length bytes, bytes)

112116 pure bytes

113117

114118type NewConnectionHandler msg = IO (Maybe msg)

@@ -122,25 +126,9 @@ serve :: (Show msg, Serialise msg, Serialise resp) =>

122126 PortNumber ->

123127 Maybe (MVar SockAddr) ->

124128 IO Bool

125−serve connhandler msghandler hostaddr port mSockLock = do

129+serve connhandler userMsgHandler hostaddr port mSockLock = do

126130 let handleSock sock = do

127131 putStrLn "handleSock"

128− let messageHandler msg = do

129− putStrLn $ "GOT MSG " ++ show msg

130− case msg of

131− Response{} -> putStrLn "client sent response"

132− ResponseExpectedRequest requestID val -> do

133− putStrLn "server received ResponseExpectedRequest"

134− resp <- msghandler val

135− case resp of

136− HandlerResponse responseVal -> sendMessage (Response requestID responseVal) sock

137− NoResponse -> error "attempt to return non-response to expected response message"

138− HandlerException _ -> error "TODO HandlerException"

139− AsyncRequest _ val -> do

140− putStrLn "AsyncReq"

141− void $ msghandler val

142− --no response necessaryxs

143− ExceptionResponse{} -> putStrLn "client sent exception response"

144132 -- allow the server to send an async welcome message to the new client, if necessary

145133 mResp <- connhandler

146134 case mResp of

@@ -148,15 +136,48 @@ serve connhandler msghandler hostaddr port mSockLock = do

148136 Just resp -> do

149137 requestID <- UUID <$> UUIDBase.nextRandom

150138 sendMessage (AsyncRequest requestID resp) sock

151− drainSocketMessages sock messageHandler

139+ drainSocketMessages sock (serverMessageHandler sock userMsgHandler)

152140 serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & serially . S.mapM (handleWithM handleSock) & S.drain

153141 pure True

154142

143+serverMessageHandler :: forall req resp. (Show req,

144+ Serialise req,

145+ Serialise resp)

146+ => Socket

147+ -> NewMessageHandler req resp

148+ -> Message req

149+ -> IO ()

150+serverMessageHandler sock requestMessageHandler msg = do

151+ putStrLn $ "GOT MSG " ++ show msg

152+ let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))

153+ runTimeout mTimeout m = case mTimeout of

154+ Nothing -> Just <$> m

155+ Just timeoutMicroseconds -> timeout timeoutMicroseconds m

156+ case msg of

157+ Response{} -> putStrLn "client sent response"

158+ ResponseExpectedRequest requestID mTimeout val -> do

159+ putStrLn "server received ResponseExpectedRequest"

160+ resp <- runTimeout mTimeout (requestMessageHandler val)

161+ case resp of

162+ Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sock

163+ Just NoResponse -> error "attempt to return non-response to expected response message"

164+ Just (HandlerException _) -> error "TODO HandlerException"

165+ Nothing ->

166+ sendMessage (TimedOutResponse @(Message resp) requestID) sock

167+ AsyncRequest _ val -> do

168+ putStrLn "AsyncReq"

169+ void $ requestMessageHandler val

170+ --no response necessary

171+ ExceptionResponse{} -> putStrLn "client sent exception response"

172+ TimedOutResponse{} -> putStrLn "client sent timed out response"

173+

155174--add callback to allow for responses via socket

156175type MessageHandler a = Message a -> IO ()

157176

158177type AsyncMessageHandler a = a -> IO ()

159178

179+

180+

160181drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()

161182drainSocketMessages sock msgHandler = do

162183 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)

@@ -182,10 +203,10 @@ octets w =

182203 ]

183204

184205--send length-tagged bytestring, perhaps should be in network byte order?

185−sendMessage :: Serialise a => a -> Socket -> IO ()

206+sendMessage :: Serialise a => Message a -> Socket -> IO ()

186207sendMessage msg socket' = do

187− let byteArray = toArraySlow fullbytes

188− dbgByteString = BS.pack (SA.toList byteArray)

208+ let -- byteArray = toArraySlow fullbytes

209+ --dbgByteString = BS.pack (SA.toList byteArray)

189210 msgbytes = serialise msg

190211 fullbytes = lenbytes <> msgbytes

191212 len = BS.length msgbytes

@@ -194,7 +215,7 @@ sendMessage msg socket' = do

194215

195216 byteCount <- Socket.send socket' fullbytes

196217 when (byteCount /= BS.length fullbytes) (error "bytes sent mismatch")

197− traceShowM ("sent bytes:", byteCount, fullbytes)

218+ traceShowM ("sent bytes:" :: String, byteCount, fullbytes)

198219

199220

200221

@@ -212,6 +233,6 @@ toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr

212233 endPtr = unsafeForeignPtrToPtr nfp `plusPtr` len `plusPtr` 1 -- ? why +1?

213234

214235toArraySlow :: BS.ByteString -> SA.Array Word8

215−toArraySlow bs = traceShow ("arr", SA.length arr) arr

236+toArraySlow bs = traceShow ("arr" :: String, SA.length arr) arr

216237 where

217238 arr = SA.fromList (BS.unpack bs)