master..905595c · curryer

git.iko.soy/curryer

9 files changed+81−255

unifiedsplit

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

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

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

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

7271Executable SimpleKeyValueServer

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

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

7473 Main-Is: examples/SimpleKeyValueServer.hs

7574 GHC-Options: -Wall -threaded

7675 Default-Language: Haskell2010

7776

7877Executable SimpleKeyValueClient

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

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

8079 Main-Is: examples/SimpleKeyValueClient.hs

8180 GHC-Options: -Wall -threaded

8281 Default-Language: Haskell2010

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

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

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

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

9998type RequestHandlers serverState = [RequestHandler serverState]

10099

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.

100+-- |

102101data RequestHandler serverState where

103102 -- | create a request handler with a response

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

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

111110 connectionSocket :: Locking Socket

112111 }

113112

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

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

117116 requestID <- UUID <$> UUIDBase.nextRandom

118117 let env =

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

120− timeout' = 0

121− sendEnvelope env lockSock

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

119+ sendEnvelope env (connectionSocket sState)

122120

123121--avoid orphan instance

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

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

145143 deriving (Generic, Show, Eq)

146144 deriving Serialise via WineryVariant ConnectionError

147145

148−data TimeoutException = TimeoutException

149− deriving Show

150−

151−instance Exception TimeoutException

152−

153146type HostAddr = (Word8, Word8, Word8, Word8)

154147

155148allHostAddrs,localHostAddr :: HostAddr

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

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

160153envelopeP :: Parser IO Word8 Envelope

161154envelopeP = do

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

155+ let s = FL.toList

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

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

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

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

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

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

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

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

167161 lenPrefixedByteStringP = do

168162 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)

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

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

175165

176166encodeEnvelope :: Envelope -> BS.ByteString

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

185175 ExceptionResponseMessage -> BS.singleton 3

186176 msgIdBs =

187177 case UUIDBase.toWords (_unUUID msgId) of

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

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

189179 msgLen = fromIntegral (BS.length bs)

190180 lenPrefixedBs = BO.bytestring32 msgLen <> bs

191181

192182fingerprintP :: Parser IO Word8 Fingerprint

193−fingerprintP =

194− Fingerprint <$> word64P <*> word64P

183+fingerprintP = do

184+ f1 <- word64P

185+ f2 <- word64P

186+ pure (Fingerprint f1 f2)

195187

196188word64P :: Parser IO Word8 Word64

197189word64P = do

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

212204 u1 <- word32P

213205 u2 <- word32P

214206 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

207+ u4 <- word32P

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

219209

220210type NewConnectionHandler msg = IO (Maybe msg)

221211

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

239229 pure True

240230

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

242−openEnvelope (Envelope eprint _ _ bytes) =

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

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

244234 case deserialise bytes of

245235 Left err -> error (show err)

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

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

272262 runTimeout m =

273263 if timeoutms == 0 then

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

264+ Just <$> m

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

266+ timeout (fromIntegral timeoutms) m

267+

281268 sState = ConnectionState {

282269 connectionServerState = serverState,

283270 connectionSocket = sockLock

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

293280 case mResponse of

294281 Just response ->

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

296− Nothing ->

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

283+ Nothing ->

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

298285 sendEnvelope envelopeResponse sockLock

299286 pure (Just ())

300287 firstMatcher (AsyncRequestHandler msghandler) Nothing =

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

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

328315 withLock sockLock $ \socket' ->

329316 Socket.sendAll socket' fullbytes

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

331317 traceBytes "sendEnvelope" fullbytes

332318

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

334320fingerprint = typeRepFingerprint . typeOf

335321

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

@@ -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,10 +61,6 @@ 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−

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

7165testServerRequestHandlers mAsyncMVar =

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

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

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

208202 assertEqual "server state" 2 serverState'

209203 close conn

210204 cancel server

205+

211206

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−