905595c...master · curryer

git.iko.soy/curryer

9 files changed+255−81

since splitsnapshots

unifiedsplit

Changelog.markdown+4 README.markdown+68−2 cabal.project+1−1 curryer.cabal → curryer-rpc.cabal+9−8 examples/SimpleKeyValueClient.hs+39 examples/SimpleKeyValueServer.hs+30 src/Network/RPC/Curryer/Client.hs+27−36 src/Network/RPC/Curryer/Server.hs+53−32 test/Curryer/Test/Basic.hs+24−2

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

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+9−8

@@ -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,21 +61,22 @@ 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

7172Executable SimpleKeyValueServer

72− Build-Depends: stm-containers, stm, base, curryer, winery

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

7374 Main-Is: examples/SimpleKeyValueServer.hs

7475 GHC-Options: -Wall -threaded

7576 Default-Language: Haskell2010

7677

7778Executable SimpleKeyValueClient

78− Build-Depends: stm, base, curryer, winery, optparse-generic

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

7980 Main-Is: examples/SimpleKeyValueClient.hs

8081 GHC-Options: -Wall -threaded

8182 Default-Language: Haskell2010

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

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

@@ -97,7 +98,7 @@ data MessageType = RequestMessage TimeoutMicroseconds

9798-- | A list of `RequestHandler`s.

9899type RequestHandlers serverState = [RequestHandler serverState]

99100

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.

101102data RequestHandler serverState where

102103 -- | create a request handler with a response

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

@@ -110,13 +111,14 @@ data ConnectionState a = ConnectionState {

110111 connectionSocket :: Locking Socket

111112 }

112113

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

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

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

116117 requestID <- UUID <$> UUIDBase.nextRandom

117118 let env =

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

119− sendEnvelope env (connectionSocket sState)

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

120+ timeout' = 0

121+ sendEnvelope env lockSock

120122

121123--avoid orphan instance

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

@@ -143,6 +145,11 @@ data ConnectionError = CodecError String -- show of WineryException from excepti

143145 deriving (Generic, Show, Eq)

144146 deriving Serialise via WineryVariant ConnectionError

145147

148+data TimeoutException = TimeoutException

149+ deriving Show

150+

151+instance Exception TimeoutException

152+

146153type HostAddr = (Word8, Word8, Word8, Word8)

147154

148155allHostAddrs,localHostAddr :: HostAddr

@@ -152,15 +159,18 @@ localHostAddr = (127,0,0,1)

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

153160envelopeP :: Parser IO Word8 Envelope

154161envelopeP = do

155− let s = FL.toList

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

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

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

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

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

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

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

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

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

161167 lenPrefixedByteStringP = do

162168 c <- fromIntegral <$> word32P

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

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

165175

166176encodeEnvelope :: Envelope -> BS.ByteString

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

175185 ExceptionResponseMessage -> BS.singleton 3

176186 msgIdBs =

177187 case UUIDBase.toWords (_unUUID msgId) of

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

179189 msgLen = fromIntegral (BS.length bs)

180190 lenPrefixedBs = BO.bytestring32 msgLen <> bs

181191

182192fingerprintP :: Parser IO Word8 Fingerprint

183−fingerprintP = do

184− f1 <- word64P

185− f2 <- word64P

186− pure (Fingerprint f1 f2)

193+fingerprintP =

194+ Fingerprint <$> word64P <*> word64P

187195

188196word64P :: Parser IO Word8 Word64

189197word64P = do

@@ -204,8 +212,10 @@ uuidP = do

204212 u1 <- word32P

205213 u2 <- word32P

206214 u3 <- word32P

207− u4 <- word32P

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

209219

210220type NewConnectionHandler msg = IO (Maybe msg)

211221

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

229239 pure True

230240

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

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

242+openEnvelope (Envelope eprint _ _ bytes) =

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

234244 case deserialise bytes of

235245 Left err -> error (show err)

@@ -261,10 +271,13 @@ serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (Req

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

262272 runTimeout m =

263273 if timeoutms == 0 then

264− Just <$> m

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

265275 else

266− timeout (fromIntegral timeoutms) m

267−

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+

268281 sState = ConnectionState {

269282 connectionServerState = serverState,

270283 connectionSocket = sockLock

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

280293 case mResponse of

281294 Just response ->

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

283− Nothing ->

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

296+ Nothing ->

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

285298 sendEnvelope envelopeResponse sockLock

286299 pure (Just ())

287300 firstMatcher (AsyncRequestHandler msghandler) Nothing =

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

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

315328 withLock sockLock $ \socket' ->

316329 Socket.sendAll socket' fullbytes

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

317331 traceBytes "sendEnvelope" fullbytes

318332

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

320334fingerprint = typeRepFingerprint . typeOf

321335

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+24−2

@@ -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,6 +63,10 @@ data AsyncHelloReq = AsyncHelloReq String

6163 deriving (Generic, Show)

6264 deriving Serialise via WineryVariant AsyncHelloReq

6365

66+data ThrowTimeout = ThrowTimeout

67+ deriving (Generic, Show)

68+ deriving Serialise via WineryVariant ThrowTimeout

69+

6470testServerRequestHandlers :: Maybe (MVar String) -> RequestHandlers ()

6571testServerRequestHandlers mAsyncMVar =

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

@@ -72,7 +78,7 @@ testServerRequestHandlers 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

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

202208 assertEqual "server state" 2 serverState'

203209 close conn

204210 cancel server

205−

206211

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+