master..804b0ba · curryer

git.iko.soy/curryer

10 files changed+106−293

unifiedsplit

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

Changelog.markdowndeleted−4

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

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

2−

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

4−

README.markdown+2−68

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

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, but is lighter-weight and uses a higher-performance serialization package.

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

77## Features

99* blocking and non-blocking remote function calls

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

1111* timeouts

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

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

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 benchmarkServerRequestHandlers () localHostAddr 0 (Just portReadyMVar))

18+ server <- async (serve benchmarkServerMessageHandlers 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−benchmarkServerRequestHandlers :: RequestHandlers a

57−benchmarkServerRequestHandlers =

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

56+benchmarkServerMessageHandlers :: MessageHandlers

57+benchmarkServerMessageHandlers =

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: 4e48b52e8709390a47e99c107ebed2d1c9420076

6+ tag: ac3af8749194f1788704dda8667d0b3807075cc2

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

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

1−Name: curryer-rpc

1+Name: curryer

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 cabal.project

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

1515

1616Source-Repository head

1717 Type: git

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

1919

2020Library

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

21+ Build-Depends: base >= 4.8

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

54+ , curryer

5555 , winery

5656 , network

5757 , async

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

6161Benchmark perf

6262 Default-Language: Haskell2010

6363 Default-Extensions: OverloadedStrings

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

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

6565

6666 Main-Is: Basic.hs

6767 Type: exitcode-stdio-1.0

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

69− -- -fprof-auto

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

7069 HS-Source-Dirs: ./bench

7170

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−

71+

examples/SimpleKeyValueClient.hsdeleted−39

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

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

7−

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.hsdeleted−30

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

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

8−

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+36−27

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

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

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

22module Network.RPC.Curryer.Client where

33import Network.RPC.Curryer.Server

44import Network.Socket as Socket

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

1515

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

1717

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

18+-- the request map holds

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.

3130connect ::

3231 ClientAsyncRequestHandlers ->

3332 HostAddr ->

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

4443 _conn_syncmap = syncmap

4544 })

4645

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

4846close :: Connection -> IO ()

4947close conn = do

5048 withLock (_conn_sockLock conn) $ \sock ->

5149 Socket.close sock

5250 cancel (_conn_asyncThread conn)

5351

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

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

5553clientAsync ::

5654 Socket ->

5755 SyncMap ->

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

6159 lsock <- newLock sock

6260 drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)

6361

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

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

7563clientEnvelopeHandler ::

7664 ClientAsyncRequestHandlers

7765 -> Locking Socket

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

8876 pure (Just ())

8977 firstMatcher acc _ = pure acc

9078 foldM_ firstMatcher Nothing handlers

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

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

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

98108 Right excStr ->

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

109+ putMVar mVar (Left (ExceptionError excStr))

100110

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

111+

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

103113call = callTimeout Nothing

104114

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

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

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

107117callTimeout mTimeout conn msg = do

108118 requestID <- UUID <$> UUIDBase.nextRandom

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

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

138148 Right m -> pure (Right m)

139149

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

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

142151asyncCall conn msg = do

143152 requestID <- UUID <$> UUIDBase.nextRandom

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

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

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

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

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

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

44module Network.RPC.Curryer.Server where

@@ -27,7 +27,6 @@ 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

3130

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

3332--import Control.Monad

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

3534import System.Timeout

3635import qualified Network.ByteOrder as BO

3736

38−

3937-- for toArrayS conversion

40−import qualified Data.ByteString.Internal as BSI

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

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)

4241import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

43−import GHC.Ptr (minusPtr)

44−

45−-- define CURRYER_SHOW_BYTES 1

42+import GHC.Ptr (plusPtr)

43+-}

44+-- #define CURRYER_SHOW_BYTES 1

4645#if CURRYER_SHOW_BYTES == 1

4746import Debug.Trace

4847#endif

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

8786deriving instance Generic Fingerprint

8887deriving via WineryVariant Fingerprint instance Serialise Fingerprint

8988

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

9189data MessageType = RequestMessage TimeoutMicroseconds

9290 | ResponseMessage

9391 | TimeoutResponseMessage

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

9593 deriving (Generic, Show)

9694 deriving Serialise via WineryVariant MessageType

9795

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.

96+type MessageHandlers serverState = [RequestHandler serverState]

97+

10298data RequestHandler serverState where

10399 -- | create a request handler with a response

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

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

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

107103

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

104+-- | Passed to

109105data ConnectionState a = ConnectionState {

110106 connectionServerState :: a,

111107 connectionSocket :: Locking Socket

112108 }

113109

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

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

117113 requestID <- UUID <$> UUIDBase.nextRandom

118114 let env =

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

120− timeout' = 0

121− sendEnvelope env lockSock

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

116+ sendEnvelope env (connectionSocket sState)

122117

123118--avoid orphan instance

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

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

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

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

140135

141−-- | Errors from remote calls.

136+

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

143138 | TimeoutError

144139 | ExceptionError String

145140 deriving (Generic, Show, Eq)

146141 deriving Serialise via WineryVariant ConnectionError

147142

148−data TimeoutException = TimeoutException

149− deriving Show

150−

151−instance Exception TimeoutException

152−

153143type HostAddr = (Word8, Word8, Word8, Word8)

154144

155145allHostAddrs,localHostAddr :: HostAddr

156146allHostAddrs = (0,0,0,0)

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

158148

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

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

160150envelopeP :: Parser IO Word8 Envelope

161151envelopeP = do

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

152+ let s = FL.toList

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

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

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

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

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

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

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

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

167158 lenPrefixedByteStringP = do

168159 c <- fromIntegral <$> word32P

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)

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

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

175162

176163encodeEnvelope :: Envelope -> BS.ByteString

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

185172 ExceptionResponseMessage -> BS.singleton 3

186173 msgIdBs =

187174 case UUIDBase.toWords (_unUUID msgId) of

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

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

189176 msgLen = fromIntegral (BS.length bs)

190177 lenPrefixedBs = BO.bytestring32 msgLen <> bs

191178

192179fingerprintP :: Parser IO Word8 Fingerprint

193−fingerprintP =

194− Fingerprint <$> word64P <*> word64P

180+fingerprintP = do

181+ f1 <- word64P

182+ f2 <- word64P

183+ pure (Fingerprint f1 f2)

195184

196185word64P :: Parser IO Word8 Word64

197186word64P = do

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

212201 u1 <- word32P

213202 u2 <- word32P

214203 u3 <- word32P

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

204+ u4 <- word32P

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

219206

220207type NewConnectionHandler msg = IO (Maybe msg)

221208

222209type NewMessageHandler req resp = req -> IO resp

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.

210+

225211serve ::

226− RequestHandlers s->

212+ MessageHandlers s->

227213 s ->

228214 HostAddr ->

229215 PortNumber ->

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

239225 pure True

240226

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

242−openEnvelope (Envelope eprint _ _ bytes) =

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

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

244230 case deserialise bytes of

245231 Left err -> error (show err)

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

256242 Nothing -> Nothing

257243 Just decoded -> Just (dispatchf, decoded)

258244

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

260245serverEnvelopeHandler ::

261246 Locking Socket

262− -> RequestHandlers s

247+ -> MessageHandlers s

263248 -> s

264249 -> Envelope

265250 -> IO ()

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

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

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

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

270254 --find first matching handler

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

272256 runTimeout m =

273257 if timeoutms == 0 then

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

258+ Just <$> m

275259 else

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−

260+ timeout (fromIntegral timeoutms) m

261+

281262 sState = ConnectionState {

282263 connectionServerState = serverState,

283264 connectionSocket = sockLock

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

293274 case mResponse of

294275 Just response ->

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

296− Nothing ->

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

277+ Nothing ->

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

298279 sendEnvelope envelopeResponse sockLock

299280 pure (Just ())

300281 firstMatcher (AsyncRequestHandler msghandler) Nothing =

@@ -310,10 +291,12 @@ serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (Req

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

311292 sendEnvelope env sockLock

312293 Right () -> pure ()

313−

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

314295

315296type EnvelopeHandler = Envelope -> IO ()

316297

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

299+

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

318301drainSocketMessages sock envelopeHandler = do

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

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

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

328311 withLock sockLock $ \socket' ->

329312 Socket.sendAll socket' fullbytes

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

331313 traceBytes "sendEnvelope" fullbytes

332314

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

334316fingerprint = typeRepFingerprint . typeOf

335317

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+12−32

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

1010import Control.Monad

1111import Control.Concurrent

1212import Control.Concurrent.STM

13−import Control.Exception

1413import Data.List

1514

1615import Network.RPC.Curryer.Server

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

2625 ,testCase "server-side exception" testSyncException

2726 ,testCase "multi-threaded client" testMultithreadedClient

2827 ,testCase "server state" testServerState

29− ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout

3028 ]

3129

3230

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

6361 deriving (Generic, Show)

6462 deriving Serialise via WineryVariant AsyncHelloReq

6563

66−data ThrowTimeout = ThrowTimeout

67− deriving (Generic, Show)

68− deriving Serialise via WineryVariant ThrowTimeout

69−

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

71−testServerRequestHandlers mAsyncMVar =

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

65+testServerMessageHandlers mAsyncMVar =

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

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

7468 case mAsyncMVar of

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

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

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

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

81− sendMessage (connectionSocket sState) (AsyncHelloReq s)

75+ sendMessage sState (AsyncHelloReq s)

8276

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

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

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

9488testSimpleCall = do

9589 readyVar <- newEmptyMVar

9690

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

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

9892 --wait for server to be ready

9993 (SockAddrInet port _) <- takeMVar readyVar

10094 conn <- connect [] localHostAddr port

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

113107 let clientAsyncHandlers =

114108 [ClientAsyncRequestHandler (\(AsyncHelloReq s) ->

115109 putMVar receivedAsyncMessageVar s)]

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

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

117111 (SockAddrInet port _) <- takeMVar portReadyVar

118112 conn <- connect clientAsyncHandlers localHostAddr port

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

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

132126 portReadyVar <- newEmptyMVar

133127 receivedAsyncMessageVar <- newEmptyMVar

134128

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

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

136130 (SockAddrInet port _) <- takeMVar portReadyVar

137131

138132 conn <- connect [] localHostAddr port

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

147141testSyncClientCallTimeout = do

148142 readyVar <- newEmptyMVar

149143

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

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

151145 --wait for server to be ready

152146 (SockAddrInet port _) <- takeMVar readyVar

153147 conn <- connect [] localHostAddr port

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

160154testSyncException = do

161155 readyVar <- newEmptyMVar

162156

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

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

164158 (SockAddrInet port _) <- takeMVar readyVar

165159

166160 conn <- connect [] localHostAddr port

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

179173testMultithreadedClient = do

180174 readyVar <- newEmptyMVar

181175

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

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

183177 (SockAddrInet port _) <- takeMVar readyVar

178+ {-let clientHandler :: AsyncMessageHandler TestResponse

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

184180 conn <- connect [] localHostAddr port

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

186182 replicateM_ 10 $ do

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

208204 assertEqual "server state" 2 serverState'

209205 close conn

210206 cancel server

207+

211208

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−