9b7209e...master · curryer

git.iko.soy/curryer

10 files changed+322−112

since splitsnapshots

unifiedsplit

Changelog.markdown+4 README.markdown+68−2 bench/Basic.hs+6−6 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+70−46 test/Curryer/Test/Basic.hs+58−14

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+6−6

@@ -15,10 +15,10 @@ 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

21− clientConn <- connect localHostAddr port

21+ clientConn <- connect [] localHostAddr port

2222

2323 _ <- syncWaitRequest clientConn 100

2424 let syncWait ms = nfIO (syncWaitRequest clientConn ms)

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

5353 deriving (Generic, Show, Eq)

5454 deriving Serialise via WineryVariant WaitMillisecondsResp

5555

56−benchmarkServerMessageHandlers :: MessageHandlers

57−benchmarkServerMessageHandlers =

58− [RequestHandler $ \(WaitMillisecondsReq ms) -> do

56+benchmarkServerRequestHandlers :: RequestHandlers a

57+benchmarkServerRequestHandlers =

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

5959 threadDelay (1000 * ms)

6060 pure WaitMillisecondsResp,

61− RequestHandler $ \(WaitByteStringReq bs) ->

61+ RequestHandler $ \_ (WaitByteStringReq bs) ->

6262 pure bs

6363 ]

6464

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+70−46

@@ -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,7 +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 Control.Concurrent.STM

30+import Data.Functor

3131

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

3333--import Control.Monad

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

3535import System.Timeout

3636import qualified Network.ByteOrder as BO

3737

38+

3839-- for toArrayS conversion

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

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

41−import Foreign.ForeignPtr (plusForeignPtr)

40+import qualified Data.ByteString.Internal as BSI

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

4242import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

43−import GHC.Ptr (plusPtr)

44−-}

45−-- #define CURRYER_SHOW_BYTES 1

43+import GHC.Ptr (minusPtr)

44+

45+-- define CURRYER_SHOW_BYTES 1

4646#if CURRYER_SHOW_BYTES == 1

4747import Debug.Trace

4848#endif

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

8787deriving instance Generic Fingerprint

8888deriving via WineryVariant Fingerprint instance Serialise Fingerprint

8989

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

9091data MessageType = RequestMessage TimeoutMicroseconds

9192 | ResponseMessage

9293 | TimeoutResponseMessage

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

9495 deriving (Generic, Show)

9596 deriving Serialise via WineryVariant MessageType

9697

97−type MessageHandlers serverState = [RequestHandler serverState]

98−

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.

99102data RequestHandler serverState where

100103 -- | create a request handler with a response

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

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

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

104107

105−-- | Passed to

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

106109data ConnectionState a = ConnectionState {

107− connectionServerState :: STM a,

110+ connectionServerState :: a,

108111 connectionSocket :: Locking Socket

109112 }

110113

111−-- | 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.

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

113−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

114117 requestID <- UUID <$> UUIDBase.nextRandom

115118 let env =

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

117− sendEnvelope env (connectionSocket sState)

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

120+ timeout' = 0

121+ sendEnvelope env lockSock

118122

119123--avoid orphan instance

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

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

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

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

136140

137−

141+-- | Errors from remote calls.

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

139143 | TimeoutError

140144 | ExceptionError String

141145 deriving (Generic, Show, Eq)

142146 deriving Serialise via WineryVariant ConnectionError

143147

148+data TimeoutException = TimeoutException

149+ deriving Show

150+

151+instance Exception TimeoutException

152+

144153type HostAddr = (Word8, Word8, Word8, Word8)

145154

146155allHostAddrs,localHostAddr :: HostAddr

147156allHostAddrs = (0,0,0,0)

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

149158

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

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

151160envelopeP :: Parser IO Word8 Envelope

152161envelopeP = do

153− let s = FL.toList

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

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

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

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

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

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

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

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

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

159167 lenPrefixedByteStringP = do

160168 c <- fromIntegral <$> word32P

161− 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)

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

163175

164176encodeEnvelope :: Envelope -> BS.ByteString

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

173185 ExceptionResponseMessage -> BS.singleton 3

174186 msgIdBs =

175187 case UUIDBase.toWords (_unUUID msgId) of

176− (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]

177189 msgLen = fromIntegral (BS.length bs)

178190 lenPrefixedBs = BO.bytestring32 msgLen <> bs

179191

180192fingerprintP :: Parser IO Word8 Fingerprint

181−fingerprintP = do

182− f1 <- word64P

183− f2 <- word64P

184− pure (Fingerprint f1 f2)

193+fingerprintP =

194+ Fingerprint <$> word64P <*> word64P

185195

186196word64P :: Parser IO Word8 Word64

187197word64P = do

@@ -202,16 +212,19 @@ uuidP = do

202212 u1 <- word32P

203213 u2 <- word32P

204214 u3 <- word32P

205− u4 <- word32P

206− 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

207219

208220type NewConnectionHandler msg = IO (Maybe msg)

209221

210222type NewMessageHandler req resp = req -> IO resp

211−

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.

212225serve ::

213− MessageHandlers s->

214− STM s ->

226+ RequestHandlers s->

227+ s ->

215228 HostAddr ->

216229 PortNumber ->

217230 Maybe (MVar SockAddr) ->

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

226239 pure True

227240

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

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

242+openEnvelope (Envelope eprint _ _ bytes) =

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

231244 case deserialise bytes of

232245 Left err -> error (show err)

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

243256 Nothing -> Nothing

244257 Just decoded -> Just (dispatchf, decoded)

245258

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

246260serverEnvelopeHandler ::

247261 Locking Socket

248− -> MessageHandlers s

249− -> STM s

262+ -> RequestHandlers s

263+ -> s

250264 -> Envelope

251265 -> IO ()

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

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

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

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

255270 --find first matching handler

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

257272 runTimeout m =

258273 if timeoutms == 0 then

259− Just <$> m

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

260275 else

261− timeout (fromIntegral timeoutms) m

262−

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+

263281 sState = ConnectionState {

264282 connectionServerState = serverState,

265283 connectionSocket = sockLock

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

275293 case mResponse of

276294 Just response ->

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

278− Nothing ->

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

296+ Nothing ->

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

280298 sendEnvelope envelopeResponse sockLock

281299 pure (Just ())

282300 firstMatcher (AsyncRequestHandler msghandler) Nothing =

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

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

293311 sendEnvelope env sockLock

294312 Right () -> pure ()

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

296313

297−type EnvelopeHandler = Envelope -> IO ()

298314

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

315+type EnvelopeHandler = Envelope -> IO ()

300316

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

302318drainSocketMessages sock envelopeHandler = do

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

311327 --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

312328 withLock sockLock $ \socket' ->

313329 Socket.sendAll socket' fullbytes

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

314331 traceBytes "sendEnvelope" fullbytes

315332

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

317334fingerprint = typeRepFingerprint . typeOf

318335

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+58−14

@@ -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

@@ -24,6 +25,8 @@ testTree = testGroup "basic" [testCase "simple" testSimpleCall

2425 ,testCase "client sync timeout" testSyncClientCallTimeout

2526 ,testCase "server-side exception" testSyncException

2627 ,testCase "multi-threaded client" testMultithreadedClient

28+ ,testCase "server state" testServerState

29+ ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout

2730 ]

2831

2932

@@ -51,13 +54,21 @@ data ThrowServerSideExceptionReq = ThrowServerSideExceptionReq

5154 deriving (Generic, Show)

5255 deriving Serialise via WineryVariant ThrowServerSideExceptionReq

5356

57+data ChangeServerState = ChangeServerState

58+ deriving (Generic, Show)

59+ deriving Serialise via WineryVariant ChangeServerState

60+

5461--used to server -> client async request

5562data AsyncHelloReq = AsyncHelloReq String

5663 deriving (Generic, Show)

5764 deriving Serialise via WineryVariant AsyncHelloReq

5865

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

60−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 =

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

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

6374 case mAsyncMVar of

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

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

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

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

70− sendMessage sState (AsyncHelloReq s)

81+ sendMessage (connectionSocket sState) (AsyncHelloReq s)

7182

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

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

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

8394testSimpleCall = do

8495 readyVar <- newEmptyMVar

8596

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

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

8798 --wait for server to be ready

8899 (SockAddrInet port _) <- takeMVar readyVar

89100 conn <- connect [] localHostAddr port

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

102113 let clientAsyncHandlers =

103114 [ClientAsyncRequestHandler (\(AsyncHelloReq s) ->

104115 putMVar receivedAsyncMessageVar s)]

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

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

106117 (SockAddrInet port _) <- takeMVar portReadyVar

107118 conn <- connect clientAsyncHandlers localHostAddr port

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

@@ -112,8 +123,8 @@ testAsyncServerCall = do

112123 cancel server

113124

114125

115−emptyServerState :: STM ()

116−emptyServerState = pure ()

126+emptyServerState :: ()

127+emptyServerState = ()

117128

118129--test that the client can make a non-blocking call

119130testAsyncClientCall :: Assertion

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

121132 portReadyVar <- newEmptyMVar

122133 receivedAsyncMessageVar <- newEmptyMVar

123134

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

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

125136 (SockAddrInet port _) <- takeMVar portReadyVar

126137

127138 conn <- connect [] localHostAddr port

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

136147testSyncClientCallTimeout = do

137148 readyVar <- newEmptyMVar

138149

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

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

140151 --wait for server to be ready

141152 (SockAddrInet port _) <- takeMVar readyVar

142153 conn <- connect [] localHostAddr port

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

149160testSyncException = do

150161 readyVar <- newEmptyMVar

151162

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

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

153164 (SockAddrInet port _) <- takeMVar readyVar

154165

155166 conn <- connect [] localHostAddr port

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

168179testMultithreadedClient = do

169180 readyVar <- newEmptyMVar

170181

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

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

172183 (SockAddrInet port _) <- takeMVar readyVar

173− {-let clientHandler :: AsyncMessageHandler TestResponse

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

175184 conn <- connect [] localHostAddr port

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

177186 replicateM_ 10 $ do

@@ -180,5 +189,40 @@ testMultithreadedClient = do

180189 --putStrLn "plus one"

181190 close conn

182191 cancel server

183−

184192

193+testServerState :: Assertion

194+testServerState = do

195+ readyVar <- newEmptyMVar

196+

197+ let serverHandlers = [RequestHandler (\sState ChangeServerState ->

198+ atomically $

199+ modifyTVar (connectionServerState sState) (+ 1))

200+ ]

201+ serverState <- newTVarIO @Int 1

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

203+ (SockAddrInet port _) <- takeMVar readyVar

204+ conn <- connect [] localHostAddr port

205+ ret <- call conn ChangeServerState

206+ assertEqual "server ret" (Right ()) ret

207+ serverState' <- readTVarIO serverState

208+ assertEqual "server state" 2 serverState'

209+ close conn

210+ cancel server

211+

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+