master..88a59ff · curryer

git.iko.soy/curryer

10 files changed+212−403

unifiedsplit

Changelog.markdown−4 README.markdown+2−68 bench/Basic.hs+6−6 cabal.project+1−1 curryer-rpc.cabal → curryer.cabal+7−21 examples/SimpleKeyValueClient.hs−39 examples/SimpleKeyValueServer.hs−30 src/Network/RPC/Curryer/Client.hs+31−36 src/Network/RPC/Curryer/Server.hs+127−113 test/Curryer/Test/Basic.hs+38−85

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

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

@@ -33,7 +33,6 @@ Library

3333 , hashable

3434 , time

3535 , network-byte-order

36− , stm

3736 Hs-Source-Dirs: ./src

3837 Default-Language: Haskell2010

3938 ghc-options: -Wall -fwarn-unused-binds -fwarn-unused-imports

@@ -51,33 +50,20 @@ Test-Suite test

5150 build-depends: tasty

5251 , tasty-hunit

5352 , base

54− , curryer-rpc

53+ , curryer

5554 , winery

5655 , network

5756 , async

58− , stm

5957 other-modules: Curryer.Test.Basic

6058

6159Benchmark perf

6260 Default-Language: Haskell2010

6361 Default-Extensions: OverloadedStrings

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

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

6563

6664 Main-Is: Basic.hs

6765 Type: exitcode-stdio-1.0

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

69− -- -fprof-auto

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

7067 HS-Source-Dirs: ./bench

7168

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−

69+

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

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

@@ -80,29 +68,38 @@ clientEnvelopeHandler ::

8068 -> IO ()

8169clientEnvelopeHandler handlers _ _ envelope@(Envelope _ (RequestMessage _) _ _) = do

8270 --should this run off on another green thread?

83− let firstMatcher Nothing (ClientAsyncRequestHandler (dispatchf :: a -> IO ())) = do

84− case openEnvelope envelope of

71+ let firstMatcher Nothing (ClientAsyncRequestHandler dispatchf) = do

72+ case matchEnvelope envelope dispatchf of

8573 Nothing -> pure Nothing

86− Just decoded -> do

87− dispatchf decoded

74+ Just (dispatchf', decoded) -> do

75+ dispatchf' decoded

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

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

98− Right excStr ->

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

100−

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

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+

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

103100call = callTimeout Nothing

104101

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

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

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

107104callTimeout mTimeout conn msg = do

108105 requestID <- UUID <$> UUIDBase.nextRandom

@@ -129,15 +126,13 @@ callTimeout mTimeout conn msg = do

129126 --timeout

130127 Nothing ->

131128 pure (Left TimeoutError)

132− Just (Left exc) ->

133− pure (Left exc)

129+ Just (Left exc) -> error ("exception in client from server: " <> show exc)

134130 Just (Right binmsg) ->

135131 --TODO use decoder instead

136132 case deserialise binmsg of

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

138134 Right m -> pure (Right m)

139135

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

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

142137asyncCall conn msg = do

143138 requestID <- UUID <$> UUIDBase.nextRandom

src/Network/RPC/Curryer/Server.hs+127−113

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

@@ -25,9 +25,7 @@ import Streamly.Data.Fold as FL

2525import qualified Streamly.Internal.Data.Stream.IsStream as S

2626import qualified Data.Binary as B

2727import qualified Data.UUID as UUIDBase

28−import qualified Data.UUID.V4 as UUIDBase

2928import Control.Monad

30−import Data.Functor

3129

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

3331--import Control.Monad

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

3533import System.Timeout

3634import qualified Network.ByteOrder as BO

3735

38−

3936-- for toArrayS conversion

40−import qualified Data.ByteString.Internal as BSI

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

37+{-import qualified Data.ByteString.Internal as BSI

38+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA

39+import Foreign.ForeignPtr (plusForeignPtr)

4240import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

43−import GHC.Ptr (minusPtr)

44−

45−-- define CURRYER_SHOW_BYTES 1

41+import GHC.Ptr (plusPtr)

42+-}

43+-- #define CURRYER_SHOW_BYTES 1

4644#if CURRYER_SHOW_BYTES == 1

4745import Debug.Trace

4846#endif

@@ -87,39 +85,44 @@ type TimeoutMicroseconds = Int

8785deriving instance Generic Fingerprint

8886deriving via WineryVariant Fingerprint instance Serialise Fingerprint

8987

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

9188data MessageType = RequestMessage TimeoutMicroseconds

9289 | ResponseMessage

9390 | TimeoutResponseMessage

94− | ExceptionResponseMessage

9591 deriving (Generic, Show)

9692 deriving Serialise via WineryVariant MessageType

9793

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

99−type RequestHandlers serverState = [RequestHandler serverState]

94+type MessageHandlers = [RequestHandler]

95+

96+--newtype RequestHandler = RequestHandler { handler :: forall a b. (Serialise a, Serialise b) => (a -> IO b) }

10097

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.

102−data RequestHandler serverState where

98+data RequestHandler where

10399 -- | create a request handler with a response

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

100+ RequestHandler :: forall a b. (Serialise a, Serialise b) => (a -> IO b) -> RequestHandler

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

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

107−

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

109−data ConnectionState a = ConnectionState {

110− connectionServerState :: a,

111− connectionSocket :: Locking Socket

112− }

113−

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

117− requestID <- UUID <$> UUIDBase.nextRandom

118− let env =

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

120− timeout' = 0

121− sendEnvelope env lockSock

102+ AsyncRequestHandler :: forall a. Serialise a => (a -> IO ()) -> RequestHandler

122103

104+{-

105+handleCall :: forall a b. (Serialise a, Serialise b)

106+ => (a -> IO b)

107+ -> RequestHandler

108+handleCall f = RequestHandler f

109+-}

110+{-data Message a = Response UUID a

111+ | AsyncRequest UUID a

112+ | ResponseExpectedRequest UUID (Maybe Timeout) a

113+ | TimedOutResponse UUID

114+ | ExceptionResponse UUID String

115+ deriving (Generic, Show)

116+ deriving Serialise via WineryVariant (Message a)

117+

118+-- | The response type

119+

120+data HandlerResponse a = NoResponse

121+ | HandlerResponse a

122+ | HandlerException String

123+ deriving Generic

124+ deriving Serialise via WineryVariant (HandlerResponse a)

125+-}

123126--avoid orphan instance

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

125128 deriving (Show, Eq, B.Binary, Hashable)

@@ -138,39 +141,30 @@ instance Serialise UUID where

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

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

140143

141−-- | Errors from remote calls.

144+

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

143146 | TimeoutError

144147 | ExceptionError String

145148 deriving (Generic, Show, Eq)

146149 deriving Serialise via WineryVariant ConnectionError

147150

148−data TimeoutException = TimeoutException

149− deriving Show

150−

151−instance Exception TimeoutException

152−

153151type HostAddr = (Word8, Word8, Word8, Word8)

154152

155153allHostAddrs,localHostAddr :: HostAddr

156154allHostAddrs = (0,0,0,0)

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

158156

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

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

160158envelopeP :: Parser IO Word8 Envelope

161159envelopeP = do

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

160+ let s = FL.toList

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

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

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

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

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

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

164+ (P.satisfy (== 2) *> pure TimeoutResponseMessage)

167165 lenPrefixedByteStringP = do

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

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

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

175169

176170encodeEnvelope :: Envelope -> BS.ByteString

@@ -182,16 +176,17 @@ encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =

182176 RequestMessage timeoutms -> BS.singleton 0 <> BO.bytestring32 (fromIntegral timeoutms)

183177 ResponseMessage -> BS.singleton 1

184178 TimeoutResponseMessage -> BS.singleton 2

185− ExceptionResponseMessage -> BS.singleton 3

186179 msgIdBs =

187180 case UUIDBase.toWords (_unUUID msgId) of

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

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

189182 msgLen = fromIntegral (BS.length bs)

190183 lenPrefixedBs = BO.bytestring32 msgLen <> bs

191184

192185fingerprintP :: Parser IO Word8 Fingerprint

193−fingerprintP =

194− Fingerprint <$> word64P <*> word64P

186+fingerprintP = do

187+ f1 <- word64P

188+ f2 <- word64P

189+ pure (Fingerprint f1 f2)

195190

196191word64P :: Parser IO Word8 Word64

197192word64P = do

@@ -212,108 +207,135 @@ uuidP = do

212207 u1 <- word32P

213208 u2 <- word32P

214209 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

210+ u4 <- word32P

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

219212

220213type NewConnectionHandler msg = IO (Maybe msg)

221214

222215type 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.

216+

225217serve ::

226− RequestHandlers s->

227− s ->

218+ MessageHandlers ->

228219 HostAddr ->

229220 PortNumber ->

230221 Maybe (MVar SockAddr) ->

231222 IO Bool

232−serve userMsgHandlers serverState hostaddr port mSockLock = do

223+serve userMsgHandlers hostaddr port mSockLock = do

233224 let

234225 handleSock sock = do

235226 lockingSocket <- newLock sock

236− drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)

227+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers)

237228

238229 S.serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.parallely . S.mapM (handleWithM handleSock) & S.drain

239230 pure True

240231

232+{-serverEnvelopeHandler :: Locking Socket -> MessageHandlers s -> Envelope -> IO ()

233+serverEnvelopeHandler sockLock (Envelope fprint RequestMessage msgId messagebs) = do

234+ error "spam"

235+serverEnvelopeHandler sockLock (Envelope fprint ResponseMessage msgId messagebs) = do

236+ error "spam2"

237+-}

238+

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

242−openEnvelope (Envelope eprint _ _ bytes) =

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

241+ {-let decoder = case getDecoder (schema (Proxy :: Proxy s)) of

242+ Left err -> error (show err)

243+ Right dec -> dec-}

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

244245 case deserialise bytes of

245246 Left err -> error (show err)

246247 Right v -> Just v

248+ --Just (evalDecoder decoder bytes)

247249 else

248250 Nothing

249251

250−matchEnvelope :: forall a b s. (Serialise a, Serialise b, Typeable b) =>

252+matchEnvelope :: forall a b. (Serialise a, Serialise b, Typeable b) =>

251253 Envelope ->

252− (ConnectionState s -> a -> IO b) ->

253− Maybe (ConnectionState s -> a -> IO b, a)

254+ (a -> IO b) ->

255+ Maybe (a -> IO b, a)

254256matchEnvelope envelope dispatchf =

255257 case openEnvelope envelope :: Maybe a of

256258 Nothing -> Nothing

257259 Just decoded -> Just (dispatchf, decoded)

258260

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

260261serverEnvelopeHandler ::

261262 Locking Socket

262− -> RequestHandlers s

263− -> s

263+ -> MessageHandlers

264264 -> Envelope

265265 -> IO ()

266−serverEnvelopeHandler _ _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()

267−serverEnvelopeHandler _ _ _ (Envelope _ ExceptionResponseMessage _ _) = pure ()

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

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

266+serverEnvelopeHandler _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()

267+serverEnvelopeHandler sockLock msgHandlers envelope@(Envelope _ (RequestMessage timeoutms) msgId _) = do

270268 --find first matching handler

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

272− runTimeout m =

270+ runTimeout m =

273271 if timeoutms == 0 then

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

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

281− sState = ConnectionState {

282− connectionServerState = serverState,

283− connectionSocket = sockLock

284− }

285−

286− firstMatcher (RequestHandler msghandler) Nothing =

272+ Just <$> m

273+ else

274+ timeout (fromIntegral timeoutms) m

275+ firstMatcher (RequestHandler (msghandler :: a -> IO b)) Nothing =

287276 case matchEnvelope envelope msghandler of

288277 Nothing -> pure Nothing

289− Just (dispatchf, decoded) -> do

278+ Just (dispatchf, decoded) ->

290279 --TODO add exception handling

291− mResponse <- runTimeout (dispatchf sState decoded)

292− let envelopeResponse =

293− case mResponse of

294− Just response ->

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

296− Nothing ->

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

298− sendEnvelope envelopeResponse sockLock

299− pure (Just ())

300− firstMatcher (AsyncRequestHandler msghandler) Nothing =

280+ do

281+ mResponse <- runTimeout (dispatchf decoded)

282+ let envelopeResponse =

283+ case mResponse of

284+ Just response ->

285+ Envelope (fingerprint response) ResponseMessage msgId (serialise response)

286+ Nothing -> Envelope (fingerprint TimeoutError) (TimeoutResponseMessage) msgId (BS.empty)

287+ sendEnvelope envelopeResponse sockLock

288+ pure (Just ())

289+ firstMatcher (AsyncRequestHandler (msghandler :: a -> IO ())) Nothing =

301290 case matchEnvelope envelope msghandler of

302291 Nothing -> pure Nothing

303292 Just (dispatchf, decoded) -> do

304− _ <- dispatchf sState decoded

293+ _ <- dispatchf decoded

305294 pure (Just ())

306295 firstMatcher _ acc = pure acc

307− eExc <- try $ foldM_ (flip firstMatcher) Nothing msgHandlers :: IO (Either SomeException ())

308− case eExc of

309− Left exc ->

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

311− sendEnvelope env sockLock

312− Right () -> pure ()

296+ foldM_ (flip firstMatcher) Nothing msgHandlers

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

313298

314299

300+{-

301+

302+

303+ --putStrLn $ "GOT MSG " ++ show msg

304+ let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))

305+ runTimeout mTimeout m = case mTimeout of

306+ Nothing -> Just <$> m

307+ Just timeoutMicroseconds -> timeout timeoutMicroseconds m

308+ case msg of

309+ Response{} -> putStrLn "client sent response"

310+ ResponseExpectedRequest requestID mTimeout val -> do

311+ let normalResponder = do

312+ resp <- runTimeout mTimeout (requestMessageHandler val)

313+ case resp of

314+ Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sockLock

315+ Just NoResponse -> error "attempt to return non-response to expected response message"

316+ Just (HandlerException exc) -> sendMessage (ExceptionResponse @(Message resp) requestID exc) sockLock

317+ Nothing ->

318+ sendMessage (TimedOutResponse @(Message resp) requestID) sockLock

319+ excHandler :: SomeException -> IO ()

320+ excHandler e = do

321+ --send exception to client

322+ sendMessage (ExceptionResponse @(Message resp) requestID (show e)) sockLock

323+ throwIO e

324+ catch normalResponder excHandler

325+ AsyncRequest _ val ->

326+ void $ requestMessageHandler val

327+ --no response necessary

328+ ExceptionResponse{} -> putStrLn "client sent exception response"

329+ TimedOutResponse{} -> putStrLn "client sent timed out response"

330+-}

331+

332+--add callback to allow for responses via socket

333+--type MessageHandler a = Message a -> IO ()

334+

315335type EnvelopeHandler = Envelope -> IO ()

316336

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

338+

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

318340drainSocketMessages sock envelopeHandler = do

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

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

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

328350 withLock sockLock $ \socket' ->

329351 Socket.sendAll socket' fullbytes

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

331352 traceBytes "sendEnvelope" fullbytes

332353

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

334355fingerprint = typeRepFingerprint . typeOf

335356

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+38−85

@@ -9,9 +9,6 @@ import Network.Socket (SockAddr(..))

99import Control.Concurrent.Async

1010import Control.Monad

1111import Control.Concurrent

12−import Control.Concurrent.STM

13−import Control.Exception

14−import Data.List

1512

1613import Network.RPC.Curryer.Server

1714import Network.RPC.Curryer.Client

@@ -20,13 +17,11 @@ import Network.RPC.Curryer.Client

2017

2118testTree :: TestTree

2219testTree = testGroup "basic" [testCase "simple" testSimpleCall

23− ,testCase "client async" testAsyncServerCall

20+ --,testCase "client async" testAsyncServerCall

2421 ,testCase "server async" testAsyncClientCall

25− ,testCase "client sync timeout" testSyncClientCallTimeout

26− ,testCase "server-side exception" testSyncException

27− ,testCase "multi-threaded client" testMultithreadedClient

28− ,testCase "server state" testServerState

29− ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout

22+ --,testCase "client sync timeout" testSyncClientCallTimeout

23+ --,testCase "server-side exception" testSyncException

24+ ,testCase "multi-threaded client" testMultithreadedClient

3025 ]

3126

3227

@@ -54,37 +49,28 @@ data ThrowServerSideExceptionReq = ThrowServerSideExceptionReq

5449 deriving (Generic, Show)

5550 deriving Serialise via WineryVariant ThrowServerSideExceptionReq

5651

57−data ChangeServerState = ChangeServerState

58− deriving (Generic, Show)

59− deriving Serialise via WineryVariant ChangeServerState

60−

6152--used to server -> client async request

62−data AsyncHelloReq = AsyncHelloReq String

53+data AsyncHelloReq = AsyncHelloReq

6354 deriving (Generic, Show)

6455 deriving Serialise via WineryVariant AsyncHelloReq

6556

66−data ThrowTimeout = ThrowTimeout

67− deriving (Generic, Show)

68− deriving Serialise via WineryVariant ThrowTimeout

69−

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

71−testServerRequestHandlers mAsyncMVar =

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

73− , RequestHandler $ \_ (TestCallMeBackReq s) ->

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

58+testServerMessageHandlers mAsyncMVar =

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

60+ , RequestHandler $ \(TestCallMeBackReq s) ->

7461 case mAsyncMVar of

7562 Nothing -> pure ()

7663 Just mvar -> putMVar mvar s

77− , AsyncRequestHandler $ \_ (TestAsyncReq v) ->

64+ , AsyncRequestHandler $ \(TestAsyncReq v) ->

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

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

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

81− sendMessage (connectionSocket sState) (AsyncHelloReq s)

67+ --, AsyncRequestHandler $ \(AsyncReq s) -> currently lacking server state to message client

8268

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

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

69+ , RequestHandler $ \(RoundtripStringReq s) -> pure s

70+ , RequestHandler $ \(DelayMicrosecondsReq ms) -> do

8571 threadDelay ms

8672 pure ()

87− , RequestHandler $ \_ ThrowServerSideExceptionReq -> do

73+ , RequestHandler $ \ThrowServerSideExceptionReq -> do

8874 _ <- error "test server exception"

8975 pure ()

9076 ]

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

9480testSimpleCall = do

9581 readyVar <- newEmptyMVar

9682

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

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

9884 --wait for server to be ready

9985 (SockAddrInet port _) <- takeMVar readyVar

10086 conn <- connect [] localHostAddr port

@@ -106,25 +92,23 @@ testSimpleCall = do

10692

10793

10894--test that the client can proces a server-initiated asynchronous callback from the server

95+{-

10996testAsyncServerCall :: Assertion

11097testAsyncServerCall = do

11198 portReadyVar <- newEmptyMVar

11299 receivedAsyncMessageVar <- newEmptyMVar

113100 let clientAsyncHandlers =

114− [ClientAsyncRequestHandler (\(AsyncHelloReq s) ->

115− putMVar receivedAsyncMessageVar s)]

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

101+ [ClientAsyncRequestHandler (\AsyncHelloReq ->

102+ putMVar receivedAsyncMessageVar "")]

103+ server <- async (serve clientAsyncHandlers (testServerMessageHandlers receivedAsyncMessageVar) localHostAddr 0 (Just portReadyVar))

117104 (SockAddrInet port _) <- takeMVar portReadyVar

118− conn <- connect clientAsyncHandlers localHostAddr port

119− Right () <- asyncCall conn (AsyncHelloReq "welcome")

105+ conn <- connect @TestResponse clientHandler localHostAddr port

106+ callAsync conn TestAsyncReq

120107 asyncMessage <- takeMVar receivedAsyncMessageVar

121108 assertEqual "async message" "welcome" asyncMessage

122109 close conn

123110 cancel server

124−

125−

126−emptyServerState :: ()

127−emptyServerState = ()

111+-}

128112

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

130114testAsyncClientCall :: Assertion

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

132116 portReadyVar <- newEmptyMVar

133117 receivedAsyncMessageVar <- newEmptyMVar

134118

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

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

136120 (SockAddrInet port _) <- takeMVar portReadyVar

137121

138122 conn <- connect [] localHostAddr port

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

147131testSyncClientCallTimeout = do

148132 readyVar <- newEmptyMVar

149133

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

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

151135 --wait for server to be ready

152136 (SockAddrInet port _) <- takeMVar readyVar

153137 conn <- connect [] localHostAddr port

@@ -156,73 +140,42 @@ testSyncClientCallTimeout = do

156140 close conn

157141 cancel server

158142

143+{-

159144testSyncException :: Assertion

160145testSyncException = do

161146 readyVar <- newEmptyMVar

162147

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

148+ server <- async (serve (pure Nothing) (testServerMessageHandler Nothing) localHostAddr 0 (Just readyVar))

164149 (SockAddrInet port _) <- takeMVar readyVar

165−

166− conn <- connect [] localHostAddr port

150+ let clientHandler :: AsyncMessageHandler TestResponse

151+ clientHandler _ = error "async handler called"

152+ mkConn :: IO (Connection TestResponse)

153+ mkConn = connect clientHandler localHostAddr port

154+ conn <- mkConn

167155 ret <- call conn ThrowServerSideExceptionReq

168156 case ret of

169157 Left (ExceptionError actualExc) ->

170158 assertBool "server-side exception" ("test server exception" `isPrefixOf` actualExc)

171− Left otherErr ->

172− assertFailure ("server-side error " <> show otherErr)

173− Right () -> assertFailure "missed exception"

159+ Right _ -> assertFailure "missed exception"

174160 close conn

175161 cancel server

176−

162+-}

177163--throw large messages (> PIPE_BUF) at the server from multiple client connections to exercise the socket lock

178164testMultithreadedClient :: Assertion

179165testMultithreadedClient = do

180166 readyVar <- newEmptyMVar

181167

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

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

183169 (SockAddrInet port _) <- takeMVar readyVar

170+ {-let clientHandler :: AsyncMessageHandler TestResponse

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

184172 conn <- connect [] localHostAddr port

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

186174 replicateM_ 10 $ do

187175 ret <- call conn (RoundtripStringReq bigString)

188176 assertEqual "big string multithread" (Right bigString) ret

189− --putStrLn "plus one"

177+ putStrLn "plus one"

190178 close conn

191179 cancel server

180+

192181

193−testServerState :: Assertion

194−testServerState = do

195− readyVar <- newEmptyMVar

196−

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

198− atomically $

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

200− ]

201− serverState <- newTVarIO @Int 1

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

203− (SockAddrInet port _) <- takeMVar readyVar

204− conn <- connect [] localHostAddr port

205− ret <- call conn ChangeServerState

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

207− serverState' <- readTVarIO serverState

208− assertEqual "server state" 2 serverState'

209− close conn

210− cancel server

211−

212−--test that the request handler can throw a TimeoutError exception which is converted into a TimeoutError response

213−testRequestHandlerThrowTimeout :: Assertion

214−testRequestHandlerThrowTimeout = do

215− readyVar <- newEmptyMVar

216−

217− let serverHandlers = [RequestHandler (\_ ThrowTimeout ->

218− throw TimeoutException >> pure (1 :: Int)

219− )

220− ]

221− server <- async (serve serverHandlers () localHostAddr 0 (Just readyVar))

222− (SockAddrInet port _) <- takeMVar readyVar

223− conn <- connect [] localHostAddr port

224− ret <- call @_ @Int conn ThrowTimeout

225− assertEqual "handler timeout exception" (Left TimeoutError) ret

226− close conn

227− cancel server

228−