8f8eeab...master · curryer

git.iko.soy/curryer

10 files changed+290−103

since splitsnapshots

unifiedsplit

Changelog.markdown+4 README.markdown+68−2 bench/Basic.hs+3−3 cabal.project+1−1 curryer.cabal → curryer-rpc.cabal+19−7 examples/SimpleKeyValueClient.hs+39 examples/SimpleKeyValueServer.hs+30 src/Network/RPC/Curryer/Client.hs+27−36 src/Network/RPC/Curryer/Server.hs+67−42 test/Curryer/Test/Basic.hs+32−12

Changelog.markdownnew file+4

@@ -0,0 +1,4 @@

1+# v0.1 (2020-12-27)

3+* initial release to support [Project:M36](https://github.com/agentm/project-m36)

README.markdown+68−2

@@ -2,11 +2,77 @@

33Curryer (pun intended) is a fast, Haskell-exclusive RPC (remote procedure call) library. By using the latest Haskell serialization and streaming libraries, curryer aims to be the fastest and easiest means of communicating between Haskell-based processes.

5−Curryer is inspired by the now unmaintained [distributed-process](https://hackage.haskell.org/package/distributed-process) library.

5+Curryer is inspired by the now unmaintained [distributed-process](https://hackage.haskell.org/package/distributed-process) library, but is lighter-weight and uses a higher-performance serialization package.

77## Features

99* blocking and non-blocking remote function calls

1010* asynchronous server-to-client callbacks (for server-initiated notifications)

1111* timeouts

12−* utilizes versionable shared ADT for communication instead of shared function `StaticPtr`s

12+* leverages [winery](https://hackage.haskell.org/package/winery) for high-performance serialization

13+

14+## Requirements

15+

16+* GHC 8.6+

17+

18+## Code Example

19+

20+[Server](https://github.com/agentm/curryer/examples/SimpleKeyValueServer.hs):

21+

22+```haskell

23+data SetKey = SetKey String String

24+ deriving (Generic, Show)

25+ deriving Serialise via WineryVariant SetKey

26+

27+data GetKey = GetKey String

28+ deriving (Generic, Show)

29+ deriving Serialise via WineryVariant GetKey

30+

31+main :: IO ()

32+main = do

33+ kvmap <- M.newIO

34+ void $ serve kvRequestHandlers kvmap localHostAddr 8765 Nothing

35+

36+kvRequestHandlers :: RequestHandlers (M.Map String String)

37+kvRequestHandlers = [ RequestHandler $ \state (SetKey k v) ->

38+ atomically $ M.insert v k (connectionServerState state)

39+ , RequestHandler $ \state (GetKey k) ->

40+ atomically $ M.lookup k (connectionServerState state)

41+ ]

42+```

43+

44+[Client](https://github.com/agentm/curryer/examples/SimpleKeyValueClient.hs):

45+

46+```haskell

47+data SetKey = SetKey String String

48+ deriving (Generic, Show)

49+ deriving Serialise via WineryVariant SetKey

50+

51+data GetKey = GetKey String

52+ deriving (Generic, Show)

53+ deriving Serialise via WineryVariant GetKey

54+

55+data CommandOptions = Get {name :: String}

56+ | Set {name :: String, value :: String}

57+ deriving (Generic, Show)

58+

59+instance ParseRecord CommandOptions

60+

61+main :: IO ()

62+main = do

63+ opts <- getRecord "SimpleKeyValueClient"

64+ conn <- connect [] localHostAddr 8765

65+ case opts of

66+ Get k -> do

67+ eRet <- call conn (GetKey k)

68+ case eRet of

69+ Left err -> error (show err)

70+ Right (Just val) -> putStrLn val

71+ Right Nothing -> error "no such key"

72+ Set k v -> do

73+ eRet <- call conn (SetKey k v)

74+ case eRet of

75+ Left err -> error (show err)

76+ Right () -> pure ()

77+

78+```

bench/Basic.hs+3−3

@@ -15,7 +15,7 @@ main :: IO ()

1515main = do

1616 --start server shared across benchmarks- gather stats on round-trip requests-responses

1717 portReadyMVar <- newEmptyMVar

18− server <- async (serve benchmarkServerMessageHandlers () localHostAddr 0 (Just portReadyMVar))

18+ server <- async (serve benchmarkServerRequestHandlers () localHostAddr 0 (Just portReadyMVar))

1919 --wait for server to be ready

2020 (SockAddrInet port _) <- takeMVar portReadyMVar

2121 clientConn <- connect [] localHostAddr port

@@ -53,8 +53,8 @@ data WaitMillisecondsResp = WaitMillisecondsResp

5353 deriving (Generic, Show, Eq)

5454 deriving Serialise via WineryVariant WaitMillisecondsResp

5555

56−benchmarkServerMessageHandlers :: MessageHandlers a

57−benchmarkServerMessageHandlers =

56+benchmarkServerRequestHandlers :: RequestHandlers a

57+benchmarkServerRequestHandlers =

5858 [RequestHandler $ \_ (WaitMillisecondsReq ms) -> do

5959 threadDelay (1000 * ms)

6060 pure WaitMillisecondsResp,

cabal.project+1−1

@@ -3,4 +3,4 @@ packages: .

33source-repository-package

44 type: git

55 location: https://github.com/composewell/streamly.git

6− tag: ac3af8749194f1788704dda8667d0b3807075cc2

6+ tag: 4e48b52e8709390a47e99c107ebed2d1c9420076

curryer.cabal → curryer-rpc.cabalrenamed+19−7

@@ -1,4 +1,4 @@

1−Name: curryer

1+Name: curryer-rpc

22Version: 0.1

33License: PublicDomain

44Build-Type: Simple

@@ -11,14 +11,14 @@ Maintainer: agentm@themactionfaction.com

1111Cabal-Version: >= 1.10

1212Synopsis: Fast, Haskell RPC

1313Description: Haskell-to-Haskell RPC using Winery serialization.

14−Extra-Source-Files: Changelog.markdown README.markdown

14+Extra-Source-Files: Changelog.markdown README.markdown cabal.project

1515

1616Source-Repository head

1717 Type: git

1818 location: https://github.com/agentm/curryer

1919

2020Library

21− Build-Depends: base >= 4.8

21+ Build-Depends: base >= 4.12 && < 4.15

2222 , winery

2323 , bytestring

2424 , streamly >= 0.7.2

@@ -51,7 +51,7 @@ Test-Suite test

5151 build-depends: tasty

5252 , tasty-hunit

5353 , base

54− , curryer

54+ , curryer-rpc

5555 , winery

5656 , network

5757 , async

@@ -61,11 +61,23 @@ Test-Suite test

6161Benchmark perf

6262 Default-Language: Haskell2010

6363 Default-Extensions: OverloadedStrings

64− Build-Depends: base, criterion, curryer, network, winery, async, bytestring

64+ Build-Depends: base, criterion, curryer-rpc, network, winery, async, bytestring

6565

6666 Main-Is: Basic.hs

6767 Type: exitcode-stdio-1.0

68− GHC-Options: -Wall -threaded -rtsopts -fprof-auto

68+ GHC-Options: -Wall -threaded -rtsopts

69+ -- -fprof-auto

6970 HS-Source-Dirs: ./bench

7071

71−

72+Executable SimpleKeyValueServer

73+ Build-Depends: stm-containers, stm, base, curryer-rpc, winery

74+ Main-Is: examples/SimpleKeyValueServer.hs

75+ GHC-Options: -Wall -threaded

76+ Default-Language: Haskell2010

77+

78+Executable SimpleKeyValueClient

79+ Build-Depends: stm, base, curryer-rpc, winery, optparse-generic

80+ Main-Is: examples/SimpleKeyValueClient.hs

81+ GHC-Options: -Wall -threaded

82+ Default-Language: Haskell2010

83+

examples/SimpleKeyValueClient.hsnew file+39

@@ -0,0 +1,39 @@

1+{-# LANGUAGE DerivingVia, DeriveGeneric, OverloadedStrings #-}

2+import GHC.Generics

3+import Network.RPC.Curryer.Client

4+import Network.RPC.Curryer.Server (localHostAddr)

5+import Options.Generic

6+import Codec.Winery

8+data SetKey = SetKey String String

9+ deriving (Generic, Show)

10+ deriving Serialise via WineryVariant SetKey

11+

12+data GetKey = GetKey String

13+ deriving (Generic, Show)

14+ deriving Serialise via WineryVariant GetKey

15+

16+data CommandOptions = Get {name :: String}

17+ | Set {name :: String, value :: String}

18+ deriving (Generic, Show)

19+

20+instance ParseRecord CommandOptions

21+

22+main :: IO ()

23+main = do

24+ opts <- getRecord "SimpleKeyValueClient"

25+ -- connect to the remote server (in this case on the localhost address)

26+ conn <- connect [] localHostAddr 8765

27+ case opts of

28+ Get k -> do

29+ --call the remote function and validate the result

30+ eRet <- call conn (GetKey k)

31+ case eRet of

32+ Left err -> error (show err)

33+ Right (Just val) -> putStrLn val

34+ Right Nothing -> error "no such key"

35+ Set k v -> do

36+ eRet <- call conn (SetKey k v)

37+ case eRet of

38+ Left err -> error (show err)

39+ Right () -> pure ()

examples/SimpleKeyValueServer.hsnew file+30

@@ -0,0 +1,30 @@

1+{-# LANGUAGE DerivingVia, DeriveGeneric #-}

2+import Network.RPC.Curryer.Server

3+import Control.Concurrent.STM

4+import StmContainers.Map as M

5+import GHC.Generics

6+import Codec.Winery

7+import Data.Functor

9+-- create data structures to represent remote function calls

10+data SetKey = SetKey String String

11+ deriving (Generic, Show)

12+ deriving Serialise via WineryVariant SetKey

13+

14+data GetKey = GetKey String

15+ deriving (Generic, Show)

16+ deriving Serialise via WineryVariant GetKey

17+

18+--call the `serve` function to handle incoming requests

19+main :: IO ()

20+main = do

21+ kvmap <- M.newIO

22+ void $ serve kvRequestHandlers kvmap localHostAddr 8765 Nothing

23+

24+-- setup incoming request handlers to operate on the server's state

25+kvRequestHandlers :: RequestHandlers (M.Map String String)

26+kvRequestHandlers = [ RequestHandler $ \state (SetKey k v) ->

27+ atomically $ M.insert v k (connectionServerState state)

28+ , RequestHandler $ \state (GetKey k) ->

29+ atomically $ M.lookup k (connectionServerState state)

30+ ]

src/Network/RPC/Curryer/Client.hs+27−36

@@ -1,4 +1,4 @@

1−{-# LANGUAGE TypeApplications, RankNTypes, ScopedTypeVariables, GADTs #-}

1+{-# LANGUAGE RankNTypes, ScopedTypeVariables, GADTs #-}

22module Network.RPC.Curryer.Client where

33import Network.RPC.Curryer.Server

44import Network.Socket as Socket

@@ -15,18 +15,19 @@ import Control.Monad

1515

1616type SyncMap = STMMap.Map UUID (MVar (Either ConnectionError BinaryMessage), UTCTime)

1717

18−-- the request map holds

18+-- | Represents a remote connection to server.

1919data Connection = Connection { _conn_sockLock :: Locking Socket,

2020 _conn_asyncThread :: Async (),

2121 _conn_syncmap :: SyncMap

2222 }

2323

24−-- function handlers run on the client, triggered by the server- useful for asynchronous callbacks

24+-- | Function handlers run on the client, triggered by the server- useful for asynchronous callbacks.

2525data ClientAsyncRequestHandler where

2626 ClientAsyncRequestHandler :: forall a. Serialise a => (a -> IO ()) -> ClientAsyncRequestHandler

2727

2828type ClientAsyncRequestHandlers = [ClientAsyncRequestHandler]

2929

30+-- | Connects to a remote server with specific async callbacks registered.

3031connect ::

3132 ClientAsyncRequestHandlers ->

3233 HostAddr ->

@@ -43,13 +44,14 @@ connect asyncHandlers hostAddr portNum = do

4344 _conn_syncmap = syncmap

4445 })

4546

47+-- | Close the connection and release all connection resources.

4648close :: Connection -> IO ()

4749close conn = do

4850 withLock (_conn_sockLock conn) $ \sock ->

4951 Socket.close sock

5052 cancel (_conn_asyncThread conn)

5153

52−-- async thread for handling client-side incoming messages- dispatch to proper waiting thread or handler asynchronous notifications

54+-- | async thread for handling client-side incoming messages- dispatch to proper waiting thread or asynchronous notifications handler

5355clientAsync ::

5456 Socket ->

5557 SyncMap ->

@@ -59,7 +61,17 @@ clientAsync sock syncmap asyncHandlers = do

5961 lsock <- newLock sock

6062 drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)

6163

62−--handles envelope responses from server- timeout from server ignored- perhaps that's proper for trusted servers

64+consumeResponse :: UUID -> STMMap.Map UUID (MVar a, b) -> a -> IO ()

65+consumeResponse msgId syncMap val = do

66+ match <- atomically $ do

67+ val' <- STMMap.lookup msgId syncMap

68+ STMMap.delete msgId syncMap

69+ pure val'

70+ case match of

71+ Nothing -> pure () -- drop message

72+ Just (mVar, _) -> putMVar mVar val

73+

74+-- | handles envelope responses from server- timeout from ths server is ignored, but perhaps that's proper for trusted servers- the server expects the client to process all async requests

6375clientEnvelopeHandler ::

6476 ClientAsyncRequestHandlers

6577 -> Locking Socket

@@ -76,43 +88,21 @@ clientEnvelopeHandler handlers _ _ envelope@(Envelope _ (RequestMessage _) _ _)

7688 pure (Just ())

7789 firstMatcher acc _ = pure acc

7890 foldM_ firstMatcher Nothing handlers

79−clientEnvelopeHandler _ _ syncMap (Envelope _ ResponseMessage msgId binaryMessage) = do

80− --find a matching response item

81− match <- atomically $ do

82− val' <- STMMap.lookup msgId syncMap

83− STMMap.delete msgId syncMap

84− pure val'

85− case match of

86− Nothing -> error ("dropping unrequested response " <> show msgId)

87− Just (mVar, _) ->

88− putMVar mVar (Right binaryMessage)

89−clientEnvelopeHandler _ _ syncMap (Envelope _ TimeoutResponseMessage msgId _) = do

90− match <- atomically $ do

91− val' <- STMMap.lookup msgId syncMap

92− STMMap.delete msgId syncMap

93− pure val'

94− case match of

95− Nothing -> error ("dropping unrequested timeout response " <> show msgId)

96− Just (mVar, _) ->

97− putMVar mVar (Left TimeoutError)

98−clientEnvelopeHandler _ _ syncMap (Envelope _ ExceptionResponseMessage msgId excPayload) = do

99− match <- atomically $ do

100− val' <- STMMap.lookup msgId syncMap

101− STMMap.delete msgId syncMap

102− pure val'

103− case match of

104− Nothing -> error ("dropping unrequested timeout response " <> show msgId)

105− Just (mVar, _) ->

106− case deserialise excPayload of

91+clientEnvelopeHandler _ _ syncMap (Envelope _ ResponseMessage msgId binaryMessage) =

92+ consumeResponse msgId syncMap (Right binaryMessage)

93+clientEnvelopeHandler _ _ syncMap (Envelope _ TimeoutResponseMessage msgId _) =

94+ consumeResponse msgId syncMap (Left TimeoutError)

95+clientEnvelopeHandler _ _ syncMap (Envelope _ ExceptionResponseMessage msgId excPayload) =

96+ case deserialise excPayload of

10797 Left err -> error ("failed to deserialise exception string" <> show err)

10898 Right excStr ->

109− putMVar mVar (Left (ExceptionError excStr))

99+ consumeResponse msgId syncMap (Left (ExceptionError excStr))

110100

111−

101+-- | Basic remote function call via data type and return value.

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

113103call = callTimeout Nothing

114104

115−-- | Send a request to the remote server and returns a response.

105+-- | Send a request to the remote server and returns a response but with the possibility of a timeout after n microseconds.

116106callTimeout :: (Serialise request, Serialise response) => Maybe Int -> Connection -> request -> IO (Either ConnectionError response)

117107callTimeout mTimeout conn msg = do

118108 requestID <- UUID <$> UUIDBase.nextRandom

@@ -147,6 +137,7 @@ callTimeout mTimeout conn msg = do

147137 Left err -> error ("deserialise client error " <> show err)

148138 Right m -> pure (Right m)

149139

140+-- | Call a remote function but do not expect a response from the server.

150141asyncCall :: Serialise request => Connection -> request -> IO (Either ConnectionError ())

151142asyncCall conn msg = do

152143 requestID <- UUID <$> UUIDBase.nextRandom

src/Network/RPC/Curryer/Server.hs+67−42

@@ -1,4 +1,4 @@

1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications, CPP, ExistentialQuantification, StandaloneDeriving, GADTs #-}

1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, CPP, ExistentialQuantification, StandaloneDeriving, GADTs, RecordWildCards #-}

22{-# OPTIONS_GHC -fno-warn-orphans #-}

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

44module Network.RPC.Curryer.Server where

@@ -27,6 +27,7 @@ import qualified Data.Binary as B

2727import qualified Data.UUID as UUIDBase

2828import qualified Data.UUID.V4 as UUIDBase

2929import Control.Monad

30+import Data.Functor

3031

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

3233--import Control.Monad

@@ -34,14 +35,14 @@ import Data.Hashable

3435import System.Timeout

3536import qualified Network.ByteOrder as BO

3637

38+

3739-- for toArrayS conversion

38−{-import qualified Data.ByteString.Internal as BSI

39−import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA

40−import Foreign.ForeignPtr (plusForeignPtr)

40+import qualified Data.ByteString.Internal as BSI

41+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as Arr

4142import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

42−import GHC.Ptr (plusPtr)

43−-}

44−-- #define CURRYER_SHOW_BYTES 1

43+import GHC.Ptr (minusPtr)

44+

45+-- define CURRYER_SHOW_BYTES 1

4546#if CURRYER_SHOW_BYTES == 1

4647import Debug.Trace

4748#endif

@@ -86,6 +87,7 @@ type TimeoutMicroseconds = Int

8687deriving instance Generic Fingerprint

8788deriving via WineryVariant Fingerprint instance Serialise Fingerprint

8889

90+-- | Internal type used to mark envelope types.

8991data MessageType = RequestMessage TimeoutMicroseconds

9092 | ResponseMessage

9193 | TimeoutResponseMessage

@@ -93,27 +95,30 @@ data MessageType = RequestMessage TimeoutMicroseconds

9395 deriving (Generic, Show)

9496 deriving Serialise via WineryVariant MessageType

9597

96−type MessageHandlers serverState = [RequestHandler serverState]

97−

98+-- | A list of `RequestHandler`s.

99+type RequestHandlers serverState = [RequestHandler serverState]

100+

101+-- | Data types for server-side request handlers, in synchronous (client waits for return value) and asynchronous (client does not wait for return value) forms.

98102data RequestHandler serverState where

99103 -- | create a request handler with a response

100104 RequestHandler :: forall a b serverState. (Serialise a, Serialise b) => (ConnectionState serverState -> a -> IO b) -> RequestHandler serverState

101105 -- | create an asynchronous request handler where the client does not expect nor await a response

102106 AsyncRequestHandler :: forall a serverState. Serialise a => (ConnectionState serverState -> a -> IO ()) -> RequestHandler serverState

103107

104−-- | Passed to

108+-- | Server state sent in via `serve` and passed to `RequestHandler`s.

105109data ConnectionState a = ConnectionState {

106110 connectionServerState :: a,

107111 connectionSocket :: Locking Socket

108112 }

109113

110−-- | Used by server-side request handlers to send additional messages to the client. This is useful for sending asynchronous responses to the client outside of the normal request-response flow.

111−sendMessage :: Serialise a => ConnectionState s -> a -> IO ()

112−sendMessage sState msg = do

114+-- | Used by server-side request handlers to send additional messages to the client. This is useful for sending asynchronous responses to the client outside of the normal request-response flow. The locking socket can be found in the ConnectionState when a request handler is called.

115+sendMessage :: Serialise a => Locking Socket -> a -> IO ()

116+sendMessage lockSock msg = do

113117 requestID <- UUID <$> UUIDBase.nextRandom

114118 let env =

115− Envelope (fingerprint msg) (RequestMessage 0) requestID (serialise msg)

116− sendEnvelope env (connectionSocket sState)

119+ Envelope (fingerprint msg) (RequestMessage timeout') requestID (serialise msg)

120+ timeout' = 0

121+ sendEnvelope env lockSock

117122

118123--avoid orphan instance

119124newtype UUID = UUID { _unUUID :: UUIDBase.UUID }

@@ -133,31 +138,39 @@ instance Serialise UUID where

133138 x -> error $ "invalid schema element " <> show x

134139 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)

135140

136−

141+-- | Errors from remote calls.

137142data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps

138143 | TimeoutError

139144 | ExceptionError String

140145 deriving (Generic, Show, Eq)

141146 deriving Serialise via WineryVariant ConnectionError

142147

148+data TimeoutException = TimeoutException

149+ deriving Show

150+

151+instance Exception TimeoutException

152+

143153type HostAddr = (Word8, Word8, Word8, Word8)

144154

145155allHostAddrs,localHostAddr :: HostAddr

146156allHostAddrs = (0,0,0,0)

147157localHostAddr = (127,0,0,1)

148158

149−-- Each message is lxggength-prefixed by a 32-bit unsigned length.

159+-- Each message is length-prefixed by a 32-bit unsigned length.

150160envelopeP :: Parser IO Word8 Envelope

151161envelopeP = do

152− let s = FL.toList

153− msgTypeP = (P.satisfy (== 0) *>

162+ let msgTypeP = (P.satisfy (== 0) *>

154163 (RequestMessage . fromIntegral <$> word32P)) `P.alt`

155− (P.satisfy (== 1) *> pure ResponseMessage) `P.alt`

156− (P.satisfy (== 2) *> pure TimeoutResponseMessage) `P.alt`

157− (P.satisfy (== 3) *> pure ExceptionResponseMessage)

164+ (P.satisfy (== 1) $> ResponseMessage) `P.alt`

165+ (P.satisfy (== 2) $> TimeoutResponseMessage) `P.alt`

166+ (P.satisfy (== 3) $> ExceptionResponseMessage)

158167 lenPrefixedByteStringP = do

159168 c <- fromIntegral <$> word32P

160− BS.pack <$> P.takeEQ c s

169+ --streamly can't handle takeEQ 0, so add special handling

170+ if c == 0 then

171+ pure BS.empty

172+ else

173+ fromArray <$> P.takeEQ c (Arr.writeN c)

161174 Envelope <$> fingerprintP <*> msgTypeP <*> uuidP <*> lenPrefixedByteStringP

162175

163176encodeEnvelope :: Envelope -> BS.ByteString

@@ -172,15 +185,13 @@ encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =

172185 ExceptionResponseMessage -> BS.singleton 3

173186 msgIdBs =

174187 case UUIDBase.toWords (_unUUID msgId) of

175− (u1, u2, u3, u4) -> foldr (<>) BS.empty (map BO.bytestring32 [u1, u2, u3, u4])

188+ (u1, u2, u3, u4) -> foldr ((<>) . BO.bytestring32) BS.empty [u1, u2, u3, u4]

176189 msgLen = fromIntegral (BS.length bs)

177190 lenPrefixedBs = BO.bytestring32 msgLen <> bs

178191

179192fingerprintP :: Parser IO Word8 Fingerprint

180−fingerprintP = do

181− f1 <- word64P

182− f2 <- word64P

183− pure (Fingerprint f1 f2)

193+fingerprintP =

194+ Fingerprint <$> word64P <*> word64P

184195

185196word64P :: Parser IO Word8 Word64

186197word64P = do

@@ -201,15 +212,18 @@ uuidP = do

201212 u1 <- word32P

202213 u2 <- word32P

203214 u3 <- word32P

204− u4 <- word32P

205− pure (UUID (UUIDBase.fromWords u1 u2 u3 u4))

215+ --u4 <- word32P

216+ --pure (UUID (UUIDBase.fromWords u1 u2 u3 u4))-}

217+ --(UUID . UUIDBase.fromWords) <$> word32P <*> word32P <*> word32P <*> word32P

218+ UUID . UUIDBase.fromWords u1 u2 u3 <$> word32P

206219

207220type NewConnectionHandler msg = IO (Maybe msg)

208221

209222type NewMessageHandler req resp = req -> IO resp

210−

223+

224+-- | Listen for new connections and handle requests which are passed the server state 's'. The MVar SockAddr can be be optionally used to know when the server is ready for processing requests.

211225serve ::

212− MessageHandlers s->

226+ RequestHandlers s->

213227 s ->

214228 HostAddr ->

215229 PortNumber ->

@@ -225,7 +239,7 @@ serve userMsgHandlers serverState hostaddr port mSockLock = do

225239 pure True

226240

227241openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s

228−openEnvelope (Envelope eprint _ _ bytes) = do

242+openEnvelope (Envelope eprint _ _ bytes) =

229243 if eprint == fingerprint (undefined :: s) then

230244 case deserialise bytes of

231245 Left err -> error (show err)

@@ -242,23 +256,28 @@ matchEnvelope envelope dispatchf =

242256 Nothing -> Nothing

243257 Just decoded -> Just (dispatchf, decoded)

244258

259+-- | Called by `serve` to process incoming envelope requests. Never returns, so use `async` to spin it off on another thread.

245260serverEnvelopeHandler ::

246261 Locking Socket

247− -> MessageHandlers s

262+ -> RequestHandlers s

248263 -> s

249264 -> Envelope

250265 -> IO ()

251266serverEnvelopeHandler _ _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()

252267serverEnvelopeHandler _ _ _ (Envelope _ ExceptionResponseMessage _ _) = pure ()

268+serverEnvelopeHandler _ _ _ (Envelope _ ResponseMessage _ _) = pure ()

253269serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (RequestMessage timeoutms) msgId _) = do

254270 --find first matching handler

255271 let runTimeout :: IO b -> IO (Maybe b)

256272 runTimeout m =

257273 if timeoutms == 0 then

258− Just <$> m

274+ (Just <$> m) `catch` timeoutExcHandler

259275 else

260− timeout (fromIntegral timeoutms) m

261−

276+ (timeout (fromIntegral timeoutms) m) `catch` timeoutExcHandler

277+ --allow server-side function to throw TimeoutError which is caught here and becomes TimeoutError value

278+ timeoutExcHandler :: TimeoutException -> IO (Maybe b)

279+ timeoutExcHandler _ = pure Nothing

280+

262281 sState = ConnectionState {

263282 connectionServerState = serverState,

264283 connectionSocket = sockLock

@@ -274,8 +293,8 @@ serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (Req

274293 case mResponse of

275294 Just response ->

276295 Envelope (fingerprint response) ResponseMessage msgId (serialise response)

277− Nothing ->

278− Envelope (fingerprint TimeoutError) (TimeoutResponseMessage) msgId (BS.empty)

296+ Nothing ->

297+ Envelope (fingerprint TimeoutError) TimeoutResponseMessage msgId BS.empty

279298 sendEnvelope envelopeResponse sockLock

280299 pure (Just ())

281300 firstMatcher (AsyncRequestHandler msghandler) Nothing =

@@ -291,11 +310,9 @@ serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (Req

291310 let env = Envelope (fingerprint (show exc)) ExceptionResponseMessage msgId (serialise (show exc)) in

292311 sendEnvelope env sockLock

293312 Right () -> pure ()

294−serverEnvelopeHandler _ _ _ (Envelope _ ResponseMessage _ _) = error "server received response message"

295313

296−type EnvelopeHandler = Envelope -> IO ()

297314

298−type AsyncMessageHandler a = a -> IO ()

315+type EnvelopeHandler = Envelope -> IO ()

299316

300317drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()

301318drainSocketMessages sock envelopeHandler = do

@@ -310,8 +327,16 @@ sendEnvelope envelope sockLock = do

310327 --Socket.sendAll syscalls send() on a loop until all the bytes are sent, so we need socket locking here to account for serialized messages of size > PIPE_BUF

311328 withLock sockLock $ \socket' ->

312329 Socket.sendAll socket' fullbytes

330+ --traceShowM ("sendEnvelope", envelope)

313331 traceBytes "sendEnvelope" fullbytes

314332

315333fingerprint :: Typeable a => a -> Fingerprint

316334fingerprint = typeRepFingerprint . typeOf

317335

336+fromArray :: Arr.Array Word8 -> BSI.ByteString

337+fromArray Arr.Array {..}

338+ | aLen == 0 = mempty

339+ | otherwise = BSI.PS aStart 0 aLen

340+ where

341+ aStartPtr = unsafeForeignPtrToPtr aStart

342+ aLen = aEnd `minusPtr` aStartPtr

test/Curryer/Test/Basic.hs+32−12

@@ -10,6 +10,7 @@ import Control.Concurrent.Async

1010import Control.Monad

1111import Control.Concurrent

1212import Control.Concurrent.STM

13+import Control.Exception

1314import Data.List

1415

1516import Network.RPC.Curryer.Server

@@ -25,6 +26,7 @@ testTree = testGroup "basic" [testCase "simple" testSimpleCall

2526 ,testCase "server-side exception" testSyncException

2627 ,testCase "multi-threaded client" testMultithreadedClient

2728 ,testCase "server state" testServerState

29+ ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout

2830 ]

2931

3032

@@ -61,8 +63,12 @@ data AsyncHelloReq = AsyncHelloReq String

6163 deriving (Generic, Show)

6264 deriving Serialise via WineryVariant AsyncHelloReq

6365

64−testServerMessageHandlers :: Maybe (MVar String) -> MessageHandlers ()

65−testServerMessageHandlers mAsyncMVar =

66+data ThrowTimeout = ThrowTimeout

67+ deriving (Generic, Show)

68+ deriving Serialise via WineryVariant ThrowTimeout

69+

70+testServerRequestHandlers :: Maybe (MVar String) -> RequestHandlers ()

71+testServerRequestHandlers mAsyncMVar =

6672 [ RequestHandler $ \_ (AddTwoNumbersReq x y) -> pure (x + y)

6773 , RequestHandler $ \_ (TestCallMeBackReq s) ->

6874 case mAsyncMVar of

@@ -72,7 +78,7 @@ testServerMessageHandlers mAsyncMVar =

7278 maybe (pure ()) (\mvar -> putMVar mvar v) mAsyncMVar

7379 -- an async hello to the server generates an async hello to the client

7480 , AsyncRequestHandler $ \sState (AsyncHelloReq s) -> do

75− sendMessage sState (AsyncHelloReq s)

81+ sendMessage (connectionSocket sState) (AsyncHelloReq s)

7682

7783 , RequestHandler $ \_ (RoundtripStringReq s) -> pure s

7884 , RequestHandler $ \_ (DelayMicrosecondsReq ms) -> do

@@ -88,7 +94,7 @@ testSimpleCall :: Assertion

8894testSimpleCall = do

8995 readyVar <- newEmptyMVar

9096

91− server <- async (serve (testServerMessageHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

97+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

9298 --wait for server to be ready

9399 (SockAddrInet port _) <- takeMVar readyVar

94100 conn <- connect [] localHostAddr port

@@ -107,7 +113,7 @@ testAsyncServerCall = do

107113 let clientAsyncHandlers =

108114 [ClientAsyncRequestHandler (\(AsyncHelloReq s) ->

109115 putMVar receivedAsyncMessageVar s)]

110− server <- async (serve (testServerMessageHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))

116+ server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))

111117 (SockAddrInet port _) <- takeMVar portReadyVar

112118 conn <- connect clientAsyncHandlers localHostAddr port

113119 Right () <- asyncCall conn (AsyncHelloReq "welcome")

@@ -126,7 +132,7 @@ testAsyncClientCall = do

126132 portReadyVar <- newEmptyMVar

127133 receivedAsyncMessageVar <- newEmptyMVar

128134

129− server <- async (serve (testServerMessageHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))

135+ server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))

130136 (SockAddrInet port _) <- takeMVar portReadyVar

131137

132138 conn <- connect [] localHostAddr port

@@ -141,7 +147,7 @@ testSyncClientCallTimeout :: Assertion

141147testSyncClientCallTimeout = do

142148 readyVar <- newEmptyMVar

143149

144− server <- async (serve (testServerMessageHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

150+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

145151 --wait for server to be ready

146152 (SockAddrInet port _) <- takeMVar readyVar

147153 conn <- connect [] localHostAddr port

@@ -154,7 +160,7 @@ testSyncException :: Assertion

154160testSyncException = do

155161 readyVar <- newEmptyMVar

156162

157− server <- async (serve (testServerMessageHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

163+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

158164 (SockAddrInet port _) <- takeMVar readyVar

159165

160166 conn <- connect [] localHostAddr port

@@ -173,10 +179,8 @@ testMultithreadedClient :: Assertion

173179testMultithreadedClient = do

174180 readyVar <- newEmptyMVar

175181

176− server <- async (serve (testServerMessageHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

182+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))

177183 (SockAddrInet port _) <- takeMVar readyVar

178− {-let clientHandler :: AsyncMessageHandler TestResponse

179− clientHandler _ = error "async handler called"-}

180184 conn <- connect [] localHostAddr port

181185 let bigString = replicate (1024 * 1000) 'x'

182186 replicateM_ 10 $ do

@@ -204,5 +208,21 @@ testServerState = do

204208 assertEqual "server state" 2 serverState'

205209 close conn

206210 cancel server

207−

208211

212+--test that the request handler can throw a TimeoutError exception which is converted into a TimeoutError response

213+testRequestHandlerThrowTimeout :: Assertion

214+testRequestHandlerThrowTimeout = do

215+ readyVar <- newEmptyMVar

216+

217+ let serverHandlers = [RequestHandler (\_ ThrowTimeout ->

218+ throw TimeoutException >> pure (1 :: Int)

219+ )

220+ ]

221+ server <- async (serve serverHandlers () localHostAddr 0 (Just readyVar))

222+ (SockAddrInet port _) <- takeMVar readyVar

223+ conn <- connect [] localHostAddr port

224+ ret <- call @_ @Int conn ThrowTimeout

225+ assertEqual "handler timeout exception" (Left TimeoutError) ret

226+ close conn

227+ cancel server

228+