905595c...master · curryer

git.iko.soy/curryer

on master help sign in

Changelog.markdownnew file+4

since splitsnapshots

unifiedsplit

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

3Curryer (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.

7## Features

9* blocking and non-blocking remote function calls

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

11* 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: .

3source-repository-package

4 type: git

5 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

2Version: 0.1

3License: PublicDomain

4Build-Type: Simple

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

11Cabal-Version: >= 1.10

12Synopsis: Fast, Haskell RPC

13Description: 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

16Source-Repository head

17 Type: git

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

20Library

21− Build-Depends: base >= 4.8

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

22 , winery

23 , bytestring

24 , streamly >= 0.7.2

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

51 build-depends: tasty

52 , tasty-hunit

53 , base

54− , curryer

54+ , curryer-rpc

55 , winery

56 , network

57 , async

@@ -61,21 +61,22 @@ Test-Suite test

61Benchmark perf

62 Default-Language: Haskell2010

63 Default-Extensions: OverloadedStrings

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

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

66 Main-Is: Basic.hs

67 Type: exitcode-stdio-1.0

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

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

69+ -- -fprof-auto

69 HS-Source-Dirs: ./bench

70 HS-Source-Dirs: ./bench

71Executable SimpleKeyValueServer

72Executable SimpleKeyValueServer

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

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

73 Main-Is: examples/SimpleKeyValueServer.hs

74 Main-Is: examples/SimpleKeyValueServer.hs

74 GHC-Options: -Wall -threaded

75 GHC-Options: -Wall -threaded

75 Default-Language: Haskell2010

76 Default-Language: Haskell2010

77Executable SimpleKeyValueClient

78Executable SimpleKeyValueClient

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

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

79 Main-Is: examples/SimpleKeyValueClient.hs

80 Main-Is: examples/SimpleKeyValueClient.hs

80 GHC-Options: -Wall -threaded

81 GHC-Options: -Wall -threaded

81 Default-Language: Haskell2010

82 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 #-}

2module Network.RPC.Curryer.Client where

3import Network.RPC.Curryer.Server

4import Network.Socket as Socket

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

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

18−-- the request map holds

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

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

20 _conn_asyncThread :: Async (),

21 _conn_syncmap :: SyncMap

22 }

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.

25data ClientAsyncRequestHandler where

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

28type ClientAsyncRequestHandlers = [ClientAsyncRequestHandler]

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

30connect ::

31connect ::

31 ClientAsyncRequestHandlers ->

32 ClientAsyncRequestHandlers ->

32 HostAddr ->

33 HostAddr ->

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

43 _conn_syncmap = syncmap

44 _conn_syncmap = syncmap

44 })

45 })

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

46close :: Connection -> IO ()

48close :: Connection -> IO ()

47close conn = do

49close conn = do

48 withLock (_conn_sockLock conn) $ \sock ->

50 withLock (_conn_sockLock conn) $ \sock ->

49 Socket.close sock

51 Socket.close sock

50 cancel (_conn_asyncThread conn)

52 cancel (_conn_asyncThread conn)

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

53clientAsync ::

55clientAsync ::

54 Socket ->

56 Socket ->

55 SyncMap ->

57 SyncMap ->

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

59 lsock <- newLock sock

61 lsock <- newLock sock

60 drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)

62 drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)

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+

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

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

63clientEnvelopeHandler ::

75clientEnvelopeHandler ::

64 ClientAsyncRequestHandlers

76 ClientAsyncRequestHandlers

65 -> Locking Socket

77 -> Locking Socket

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

76 pure (Just ())

88 pure (Just ())

77 firstMatcher acc _ = pure acc

89 firstMatcher acc _ = pure acc

78 foldM_ firstMatcher Nothing handlers

90 foldM_ firstMatcher Nothing handlers

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

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

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)

92+ consumeResponse msgId syncMap (Right binaryMessage)

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

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

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)

94+ consumeResponse msgId syncMap (Left TimeoutError)

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

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

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

96+ case deserialise excPayload of

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

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

108 Right excStr ->

98 Right excStr ->

109− putMVar mVar (Left (ExceptionError excStr))

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

110

100

111−

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

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

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

113call = callTimeout Nothing

103call = callTimeout Nothing

114

104

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.

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

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

117callTimeout mTimeout conn msg = do

107callTimeout mTimeout conn msg = do

118 requestID <- UUID <$> UUIDBase.nextRandom

108 requestID <- UUID <$> UUIDBase.nextRandom

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

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

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

148 Right m -> pure (Right m)

138 Right m -> pure (Right m)

149

139

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

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

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

151asyncCall conn msg = do

142asyncCall conn msg = do

152 requestID <- UUID <$> UUIDBase.nextRandom

143 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 #-}

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

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

4module Network.RPC.Curryer.Server where

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

27import qualified Data.UUID as UUIDBase

28import qualified Data.UUID.V4 as UUIDBase

29import Control.Monad

30+import Data.Functor

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

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

32--import Control.Monad

33--import Control.Monad

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

34import System.Timeout

35import System.Timeout

35import qualified Network.ByteOrder as BO

36import qualified Network.ByteOrder as BO

38+

37-- for toArrayS conversion

39-- for toArrayS conversion

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

40+import qualified Data.ByteString.Internal as BSI

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

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

40−import Foreign.ForeignPtr (plusForeignPtr)

41import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

42import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

42−import GHC.Ptr (plusPtr)

43+import GHC.Ptr (minusPtr)

43−-}

44+

44−-- #define CURRYER_SHOW_BYTES 1

45+-- define CURRYER_SHOW_BYTES 1

45#if CURRYER_SHOW_BYTES == 1

46#if CURRYER_SHOW_BYTES == 1

46import Debug.Trace

47import Debug.Trace

47#endif

48#endif

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

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

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

98type RequestHandlers serverState = [RequestHandler serverState]

99type RequestHandlers serverState = [RequestHandler serverState]

100

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.

101data RequestHandler serverState where

102data RequestHandler serverState where

102 -- | create a request handler with a response

103 -- | create a request handler with a response

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

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

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

110 connectionSocket :: Locking Socket

111 connectionSocket :: Locking Socket

111 }

112 }

112

113

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

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

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

115−sendMessage sState msg = do

116+sendMessage lockSock msg = do

116 requestID <- UUID <$> UUIDBase.nextRandom

117 requestID <- UUID <$> UUIDBase.nextRandom

117 let env =

118 let env =

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

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

120+ timeout' = 0

119− sendEnvelope env (connectionSocket sState)

121+ sendEnvelope env lockSock

120

122

121--avoid orphan instance

123--avoid orphan instance

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

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

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

143 deriving (Generic, Show, Eq)

145 deriving (Generic, Show, Eq)

144 deriving Serialise via WineryVariant ConnectionError

146 deriving Serialise via WineryVariant ConnectionError

145

147

148+data TimeoutException = TimeoutException

149+ deriving Show

150+

151+instance Exception TimeoutException

152+

146type HostAddr = (Word8, Word8, Word8, Word8)

153type HostAddr = (Word8, Word8, Word8, Word8)

147

154

148allHostAddrs,localHostAddr :: HostAddr

155allHostAddrs,localHostAddr :: HostAddr

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

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

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

153envelopeP :: Parser IO Word8 Envelope

160envelopeP :: Parser IO Word8 Envelope

154envelopeP = do

161envelopeP = do

155− let s = FL.toList

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

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

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

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

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

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

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

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

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

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

161 lenPrefixedByteStringP = do

167 lenPrefixedByteStringP = do

162 c <- fromIntegral <$> word32P

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

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

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

165

175

166encodeEnvelope :: Envelope -> BS.ByteString

176encodeEnvelope :: Envelope -> BS.ByteString

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

175 ExceptionResponseMessage -> BS.singleton 3

185 ExceptionResponseMessage -> BS.singleton 3

176 msgIdBs =

186 msgIdBs =

177 case UUIDBase.toWords (_unUUID msgId) of

187 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]

179 msgLen = fromIntegral (BS.length bs)

189 msgLen = fromIntegral (BS.length bs)

180 lenPrefixedBs = BO.bytestring32 msgLen <> bs

190 lenPrefixedBs = BO.bytestring32 msgLen <> bs

181

191

182fingerprintP :: Parser IO Word8 Fingerprint

192fingerprintP :: Parser IO Word8 Fingerprint

183−fingerprintP = do

193+fingerprintP =

184− f1 <- word64P

185− f2 <- word64P

186− pure (Fingerprint f1 f2)

194+ Fingerprint <$> word64P <*> word64P

187

195

188word64P :: Parser IO Word8 Word64

196word64P :: Parser IO Word8 Word64

189word64P = do

197word64P = do

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

204 u1 <- word32P

212 u1 <- word32P

205 u2 <- word32P

213 u2 <- word32P

206 u3 <- word32P

214 u3 <- word32P

207− u4 <- word32P

215+ --u4 <- word32P

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

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

209

219

210type NewConnectionHandler msg = IO (Maybe msg)

220type NewConnectionHandler msg = IO (Maybe msg)

211

221

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

229 pure True

239 pure True

230

240

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

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

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

242+openEnvelope (Envelope eprint _ _ bytes) =

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

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

234 case deserialise bytes of

244 case deserialise bytes of

235 Left err -> error (show err)

245 Left err -> error (show err)

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

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

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

262 runTimeout m =

272 runTimeout m =

263 if timeoutms == 0 then

273 if timeoutms == 0 then

264− Just <$> m

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

265 else

275 else

266− timeout (fromIntegral timeoutms) m

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

267−

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+

268 sState = ConnectionState {

281 sState = ConnectionState {

269 connectionServerState = serverState,

282 connectionServerState = serverState,

270 connectionSocket = sockLock

283 connectionSocket = sockLock

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

280 case mResponse of

293 case mResponse of

281 Just response ->

294 Just response ->

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

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

283− Nothing ->

296+ Nothing ->

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

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

285 sendEnvelope envelopeResponse sockLock

298 sendEnvelope envelopeResponse sockLock

286 pure (Just ())

299 pure (Just ())

287 firstMatcher (AsyncRequestHandler msghandler) Nothing =

300 firstMatcher (AsyncRequestHandler msghandler) Nothing =

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

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

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

315 withLock sockLock $ \socket' ->

328 withLock sockLock $ \socket' ->

316 Socket.sendAll socket' fullbytes

329 Socket.sendAll socket' fullbytes

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

317 traceBytes "sendEnvelope" fullbytes

331 traceBytes "sendEnvelope" fullbytes

318

332

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

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

320fingerprint = typeRepFingerprint . typeOf

334fingerprint = typeRepFingerprint . typeOf

321

335

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

10import Control.Monad

11import Control.Concurrent

12import Control.Concurrent.STM

13+import Control.Exception

13import Data.List

14import Data.List

15import Network.RPC.Curryer.Server

16import Network.RPC.Curryer.Server

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

25 ,testCase "server-side exception" testSyncException

26 ,testCase "server-side exception" testSyncException

26 ,testCase "multi-threaded client" testMultithreadedClient

27 ,testCase "multi-threaded client" testMultithreadedClient

27 ,testCase "server state" testServerState

28 ,testCase "server state" testServerState

29+ ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout

28 ]

30 ]

@@ -61,6 +63,10 @@ data AsyncHelloReq = AsyncHelloReq String

61 deriving (Generic, Show)

63 deriving (Generic, Show)

62 deriving Serialise via WineryVariant AsyncHelloReq

64 deriving Serialise via WineryVariant AsyncHelloReq

66+data ThrowTimeout = ThrowTimeout

67+ deriving (Generic, Show)

68+ deriving Serialise via WineryVariant ThrowTimeout

69+

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

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

65testServerRequestHandlers mAsyncMVar =

71testServerRequestHandlers mAsyncMVar =

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

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

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

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

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

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

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

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

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

75− sendMessage sState (AsyncHelloReq s)

81+ sendMessage (connectionSocket sState) (AsyncHelloReq s)

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

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

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

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

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

202 assertEqual "server state" 2 serverState'

208 assertEqual "server state" 2 serverState'

203 close conn

209 close conn

204 cancel server

210 cancel server

205−

206

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+