master..711dee1 · curryer

git.iko.soy/curryer

on 711dee1 help sign in

Changelog.markdowndeleted−4

since splitsnapshots

unifiedsplit

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

LICENSEdeleted−25

@@ -1,25 +0,0 @@

1−This is free and unencumbered software released into the public domain.

2−

3−Anyone is free to copy, modify, publish, use, compile, sell, or

4−distribute this software, either in source code form or as a compiled

5−binary, for any purpose, commercial or non-commercial, and by any

6−means.

7−

8−In jurisdictions that recognize copyright laws, the author or authors

9−of this software dedicate any and all copyright interest in the

10−software to the public domain. We make this dedication for the benefit

11−of the public at large and to the detriment of our heirs and

12−successors. We intend this dedication to be an overt act of

13−relinquishment in perpetuity of all present and future rights to this

14−software under copyright law.

15−

16−THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,

17−EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF

18−MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.

19−IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR

20−OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,

21−ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR

22−OTHER DEALINGS IN THE SOFTWARE.

23−

24−For more information, please refer to <http://unlicense.org>

25−

README.markdowndeleted−78

@@ -1,78 +0,0 @@

1−# Curryer - Fast Haskell-to-Haskell RPC

2−

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

4−

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.

6−

7−## Features

8−

9−* blocking and non-blocking remote function calls

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

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

bench.shdeleted−1

@@ -1 +0,0 @@

1−cabal run perf -- +RTS -p && ~/.cabal/bin/ghc-prof-flamegraph < perf.prof > perf.html

bench/Basic.hs+13−43

@@ -9,70 +9,40 @@ import Codec.Winery

9import Control.Concurrent.Async

10import Control.Concurrent

11import Control.Monad

12−import qualified Data.ByteString as BS

14main :: IO ()

13main :: IO ()

15main = do

14main = do

16 --start server shared across benchmarks- gather stats on round-trip requests-responses

15 --start server shared across benchmarks- gather stats on round-trip requests-responses

17 portReadyMVar <- newEmptyMVar

16 portReadyMVar <- newEmptyMVar

18− server <- async (serve benchmarkServerRequestHandlers () localHostAddr 0 (Just portReadyMVar))

17+ server <- async (serve (pure Nothing) benchmarkServerMessageHandler localHostAddr 0 (Just portReadyMVar))

19 --wait for server to be ready

18 --wait for server to be ready

20 (SockAddrInet port _) <- takeMVar portReadyMVar

19 (SockAddrInet port _) <- takeMVar portReadyMVar

21− clientConn <- connect [] localHostAddr port

20+ clientConn <- connect (const (pure ())) localHostAddr port

23 _ <- syncWaitRequest clientConn 100

22 _ <- syncWaitRequest clientConn 100

24− let syncWait ms = nfIO (syncWaitRequest clientConn ms)

25− syncByteString bs = nfIO (syncByteStringRequest clientConn bs)

26− bs0 = BS.empty

27− bs100 = BS.pack (replicate 100 3)

28− bs1000 = BS.pack (replicate 1000 3)

29− bs1M = BS.pack (replicate 1000000 3)

30− defaultMain [bgroup "wait sync" [bench "0 ms" (syncWait 0),

23+ defaultMain [bgroup "wait sync" [bench "100 ms" (nfIO $ syncWaitRequest clientConn 100)]]

31− bench "10 ms" (syncWait 10),

32− bench "100 ms" (syncWait 100)],

33− bgroup "bytestring roundtrip" [bench "0 bytes" (syncByteString bs0)

34− ,bench "100 bytes" (syncByteString bs100)

35− ,bench "1000 bytes" (syncByteString bs1000)

36− ,bench "1000000 bytes" (syncByteString bs1M)

37− ]]

38−

39− --TODO: bench for > PIPE_BUF bytes

25+ putStrLn "benchmark complete"

41 close clientConn

26 close clientConn

42 cancel server

27 cancel server

44−data WaitMillisecondsReq = WaitMillisecondsReq Int --ask the server to respond in X milliseconds

29+data BenchRequest = WaitMillisecondsReq Int

45− deriving (Generic, Show)

46− deriving Serialise via WineryVariant WaitMillisecondsReq

47−

48−data WaitByteStringReq = WaitByteStringReq BS.ByteString

49 deriving (Generic, Show)

30 deriving (Generic, Show)

50− deriving Serialise via WineryVariant WaitByteStringReq

31+ deriving Serialise via WineryVariant BenchRequest

52−data WaitMillisecondsResp = WaitMillisecondsResp

33+data BenchResponse = WaitMillisecondsResp

53 deriving (Generic, Show, Eq)

34 deriving (Generic, Show, Eq)

54− deriving Serialise via WineryVariant WaitMillisecondsResp

35+ deriving Serialise via WineryVariant BenchResponse

56−benchmarkServerRequestHandlers :: RequestHandlers a

37+benchmarkServerMessageHandler :: BenchRequest -> IO (HandlerResponse BenchResponse)

57−benchmarkServerRequestHandlers =

38+benchmarkServerMessageHandler (WaitMillisecondsReq ms) = do

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

59− threadDelay (1000 * ms)

39+ threadDelay (1000 * ms)

60− pure WaitMillisecondsResp,

40+ pure (HandlerResponse WaitMillisecondsResp)

61− RequestHandler $ \_ (WaitByteStringReq bs) ->

62− pure bs

63− ]

65--perform 10000 small synchronous requests

42--perform 10000 small synchronous requests

66−syncWaitRequest :: Connection -> Int -> IO Int

43+syncWaitRequest :: Connection BenchResponse -> Int -> IO Int

67syncWaitRequest conn ms = do

44syncWaitRequest conn ms = do

68 ret <- call conn (WaitMillisecondsReq ms)

45 ret <- call conn (WaitMillisecondsReq ms)

69 when (ret /= Right WaitMillisecondsResp) (error "ret failed")

46 when (ret /= Right WaitMillisecondsResp) (error "ret failed")

70 pure ms

47 pure ms

72−syncByteStringRequest :: Connection -> BS.ByteString -> IO BS.ByteString

73−syncByteStringRequest conn bs = do

74− ret <- call conn (WaitByteStringReq bs)

75− case ret of

76− Right bs' -> pure bs'

77− Left err -> error (show err)

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

6+ tag: e218fb9d8f4263fd69d32a2ff318bac0e179c642

curryer-rpc.cabal → curryer.cabalrenamed+6−24

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

1−Name: curryer-rpc

1+Name: curryer

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

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

16Source-Repository head

17 Type: git

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

20Library

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

21+ Build-Depends: base >= 4.8

22 , winery

23 , bytestring

24 , streamly >= 0.7.2

@@ -31,9 +31,6 @@ Library

31 , containers

32 , stm-containers

33 , hashable

34− , time

35− , network-byte-order

36− , stm

37 Hs-Source-Dirs: ./src

34 Hs-Source-Dirs: ./src

38 Default-Language: Haskell2010

35 Default-Language: Haskell2010

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

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

@@ -51,33 +48,18 @@ Test-Suite test

51 build-depends: tasty

48 build-depends: tasty

52 , tasty-hunit

49 , tasty-hunit

53 , base

50 , base

54− , curryer-rpc

51+ , curryer

55 , winery

52 , winery

56 , network

53 , network

57 , async

54 , async

58− , stm

59 other-modules: Curryer.Test.Basic

55 other-modules: Curryer.Test.Basic

61Benchmark perf

57Benchmark perf

62 Default-Language: Haskell2010

58 Default-Language: Haskell2010

63 Default-Extensions: OverloadedStrings

59 Default-Extensions: OverloadedStrings

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

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

65−

66 Main-Is: Basic.hs

61 Main-Is: Basic.hs

67 Type: exitcode-stdio-1.0

62 Type: exitcode-stdio-1.0

68 GHC-Options: -Wall -threaded -rtsopts

63 GHC-Options: -Wall -threaded -rtsopts

69− -- -fprof-auto

70 HS-Source-Dirs: ./bench

64 HS-Source-Dirs: ./bench

71−

65+

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−

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+46−122

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

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

2module Network.RPC.Curryer.Client where

1module Network.RPC.Curryer.Client where

3import Network.RPC.Curryer.Server

2import Network.RPC.Curryer.Server

4import Network.Socket as Socket

3import Network.Socket as Socket

@@ -9,139 +8,64 @@ import qualified Data.UUID.V4 as UUIDBase

9import qualified StmContainers.Map as STMMap

8import qualified StmContainers.Map as STMMap

10import Control.Concurrent.MVar

9import Control.Concurrent.MVar

11import GHC.Conc

10import GHC.Conc

12−import Data.Time.Clock

13−import System.Timeout

14−import Control.Monad

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

12+type SyncMap a = STMMap.Map UUID (MVar a)

13+-- the request map holds

14+data Connection a = Connection Socket (Async ()) (SyncMap a)

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

16+connect :: (Serialise msg) =>

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

17+ AsyncMessageHandler msg ->

20− _conn_asyncThread :: Async (),

21− _conn_syncmap :: SyncMap

22− }

23−

24−-- | Function handlers run on the client, triggered by the server- useful for asynchronous callbacks.

25−data ClientAsyncRequestHandler where

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

27−

28−type ClientAsyncRequestHandlers = [ClientAsyncRequestHandler]

29−

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

31−connect ::

32− ClientAsyncRequestHandlers ->

33− HostAddr ->

18+ HostAddr ->

34− PortNumber ->

19+ PortNumber ->

35− IO Connection

20+ IO (Connection msg)

36−connect asyncHandlers hostAddr portNum = do

21+connect notificationCallback hostAddr portNum = do

37 sock <- TCP.connect hostAddr portNum

22 sock <- TCP.connect hostAddr portNum

38 syncmap <- STMMap.newIO

23 syncmap <- STMMap.newIO

39− asyncThread <- async (clientAsync sock syncmap asyncHandlers)

24+ asyncThread <- async (clientAsync sock syncmap notificationCallback)

40− sockLock <- newLock sock

41− pure (Connection {

25+ pure (Connection sock asyncThread syncmap)

42− _conn_sockLock = sockLock,

43− _conn_asyncThread = asyncThread,

44− _conn_syncmap = syncmap

45− })

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

48−close :: Connection -> IO ()

27+close :: Connection a -> IO ()

49−close conn = do

28+close (Connection sock asyncThread _) = do

50− withLock (_conn_sockLock conn) $ \sock ->

51− Socket.close sock

29+ Socket.close sock

52− cancel (_conn_asyncThread conn)

30+ cancel asyncThread

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

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

55−clientAsync ::

33+clientAsync :: Serialise b =>

56 Socket ->

34 Socket ->

57− SyncMap ->

35+ SyncMap b ->

58− ClientAsyncRequestHandlers ->

36+ AsyncMessageHandler b ->

59 IO ()

37 IO ()

60−clientAsync sock syncmap asyncHandlers = do

38+clientAsync sock syncmap asyncHandler = do

61− lsock <- newLock sock

39+ -- ping proper thread to continue

40+ let responseHandler responseMsg = do

41+ putStrLn "clientAsync handler"

42+ case responseMsg of

43+ Response requestId val -> do

44+ mMVar <- atomically $ STMMap.lookup requestId syncmap

45+ case mMVar of

46+ Nothing -> error "dumped unrequested response"

47+ Just mVar -> putMVar mVar val

48+ AsyncRequest _ asyncMsg ->

49+ asyncHandler asyncMsg

50+ ResponseExpectedRequest _ _ -> error "dumped response expected request"

51+ ExceptionResponse _ -> error "TODO Exception"

62− drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)

52+ drainSocketMessages sock responseHandler

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

75−clientEnvelopeHandler ::

76− ClientAsyncRequestHandlers

77− -> Locking Socket

78− -> SyncMap

79− -> Envelope

80− -> IO ()

81−clientEnvelopeHandler handlers _ _ envelope@(Envelope _ (RequestMessage _) _ _) = do

82− --should this run off on another green thread?

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

84− case openEnvelope envelope of

85− Nothing -> pure Nothing

86− Just decoded -> do

87− dispatchf decoded

88− pure (Just ())

89− firstMatcher acc _ = pure acc

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

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

103−call = callTimeout Nothing

104−

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

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

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

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

107−callTimeout mTimeout conn msg = do

56+call (Connection sock _ mVarMap) msg = do

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

57+ requestID <- UUID <$> UUIDBase.nextRandom

109− let mVarMap = _conn_syncmap conn

110− timeoutms = case mTimeout of

111− Nothing -> 0

112− Just tm | tm < 0 -> 0

113− Just tm -> fromIntegral tm

114−

115− envelope = Envelope fprint (RequestMessage timeoutms) requestID (serialise msg)

116− fprint = fingerprint msg

117 -- setup mvar to wait for response

58 -- setup mvar to wait for response

118 responseMVar <- newEmptyMVar

59 responseMVar <- newEmptyMVar

119− now <- getCurrentTime

120− atomically $ STMMap.insert (responseMVar, now) requestID mVarMap

60+ atomically $ STMMap.insert responseMVar requestID mVarMap

121− sendEnvelope envelope (_conn_sockLock conn)

61+ sendMessage (ResponseExpectedRequest requestID msg) sock

122− let timeoutMicroseconds =

123− case mTimeout of

124− Just timeout' -> timeout' + 100 --add 100 ms to account for unknown network latency

125− Nothing -> -1

126− mResponse <- timeout timeoutMicroseconds (takeMVar responseMVar)

62+ response <- takeMVar responseMVar

127 atomically $ STMMap.delete requestID mVarMap

63 atomically $ STMMap.delete requestID mVarMap

128− case mResponse of

64+ pure (Right response)

129− --timeout

130− Nothing ->

131− pure (Left TimeoutError)

132− Just (Left exc) ->

133− pure (Left exc)

134− Just (Right binmsg) ->

135− --TODO use decoder instead

136− case deserialise binmsg of

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

138− Right m -> pure (Right m)

139

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

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

66+asyncCall :: (Serialise request, Serialise response) => Connection response -> request -> IO (Either ConnectionError ())

142−asyncCall conn msg = do

67+asyncCall (Connection sock _ _) msg = do

143 requestID <- UUID <$> UUIDBase.nextRandom

68 requestID <- UUID <$> UUIDBase.nextRandom

144− let envelope = Envelope fprint (RequestMessage 0) requestID (serialise msg)

69+ sendMessage (AsyncRequest requestID msg) sock

145− fprint = fingerprint msg

146− sendEnvelope envelope (_conn_sockLock conn)

147 pure (Right ())

70 pure (Right ())

71+

src/Network/RPC/Curryer/Server.hs+133−267

@@ -1,20 +1,17 @@

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

1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving #-}

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

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

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

4module Network.RPC.Curryer.Server where

3module Network.RPC.Curryer.Server where

4+import Streamly

5import qualified Streamly.Prelude as S

6import Streamly.Network.Socket

7−import Streamly.Internal.Network.Socket (handleWithM)

7+import Streamly.Internal.Network.Socket (handleWithM, writeChunk)

8import Network.Socket as Socket

9−import Network.Socket.ByteString as Socket

10−import Streamly.Internal.Data.Parser as P hiding (concatMap)

9+import Streamly.Internal.Data.Parser.ParserD as PD

11import Codec.Winery

10import Codec.Winery

12import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)

11import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)

13import Codec.Winery.Class (mkExtractor)

12import Codec.Winery.Class (mkExtractor)

14import GHC.Generics

13import GHC.Generics

15−import GHC.Fingerprint

16−import Data.Typeable

17−import Control.Concurrent.MVar (MVar, newMVar, withMVar)

14+import Control.Concurrent.MVar (MVar)

18import Control.Exception

15import Control.Exception

19import Data.Function ((&))

16import Data.Function ((&))

20import Data.Word

17import Data.Word

@@ -23,103 +20,53 @@ import qualified Data.ByteString.Lazy as BSL

23import qualified Data.ByteString.FastBuilder as BB

20import qualified Data.ByteString.FastBuilder as BB

24import Streamly.Data.Fold as FL

21import Streamly.Data.Fold as FL

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

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

23+import Data.Foldable

24+import Data.Bits

26import qualified Data.Binary as B

25import qualified Data.Binary as B

27import qualified Data.UUID as UUIDBase

26import qualified Data.UUID as UUIDBase

28import qualified Data.UUID.V4 as UUIDBase

27import qualified Data.UUID.V4 as UUIDBase

29−import Control.Monad

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

30−import Data.Functor

31−

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

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

33−--import Control.Monad

30+import Control.Monad

34import Data.Hashable

31import Data.Hashable

35−import System.Timeout

36−import qualified Network.ByteOrder as BO

37−

39-- for toArrayS conversion

33-- for toArrayS conversion

40import qualified Data.ByteString.Internal as BSI

34import qualified Data.ByteString.Internal as BSI

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

35+import Foreign.ForeignPtr (plusForeignPtr)

42import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

36import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

43−import GHC.Ptr (minusPtr)

37+import GHC.Ptr (plusPtr)

45−-- define CURRYER_SHOW_BYTES 1

46−#if CURRYER_SHOW_BYTES == 1

47import Debug.Trace

39import Debug.Trace

48−#endif

49−

50−traceBytes :: Applicative f => String -> BS.ByteString -> f ()

51−#if CURRYER_SHOW_BYTES == 1

52−traceBytes msg bs = traceShowM (msg, BS.length bs, bs)

53−#else

54−traceBytes _ _ = pure ()

55−#endif

56−

57−data Locking a = Locking (MVar ()) a

58−

59−newLock :: a -> IO (Locking a)

60−newLock x = do

61− lock <- newMVar ()

62− pure (Locking lock x)

63−

64−withLock :: Locking a -> (a -> IO b) -> IO b

65−withLock (Locking mvar v) m =

66− withMVar mvar $ \_ -> m v

67−

68−lockless :: Locking a -> a

69−lockless (Locking _ a) = a

70−

71−type Timeout = Word32

73−type BinaryMessage = BS.ByteString

41+data ClientHelloMessage = ClientHelloMessage Int Int

42+ deriving Generic

43+ deriving Serialise via WineryVariant ClientHelloMessage

44+

45+data ServerHelloMessage = ServerHelloMessage { serverServices :: [String],

46+ serverInfo :: String

47+ }

48+ deriving Generic

49+ deriving Serialise via WineryRecord ServerHelloMessage

50+

51+--request-response token

52+--client-side timeout

53+--server-to-client and client-to-server async request (no response requested)

54+

55+data Message a = Response UUID a

56+ | AsyncRequest UUID a

57+ | ResponseExpectedRequest UUID a

58+ -- | Services

59+ | ExceptionResponse String

60+ deriving (Generic, Show)

61+ deriving Serialise via WineryVariant (Message a)

62+

63+-- | The response type

64+data HandlerResponse a = NoResponse

65+ | HandlerResponse a

66+ | HandlerException a

67+ deriving Generic

68+ deriving Serialise via WineryVariant (HandlerResponse a)

75−--includes the fingerprint of the incoming data type (held in the BinaryMessage) to determine how to dispatch the message.

76−--add another envelope type for unencoded binary messages for any easy optimization for in-process communication

77−data Envelope = Envelope {

78− envFingerprint :: !Fingerprint,

79− envMessageType :: !MessageType,

80− envMsgId :: !UUID,

81− envPayload :: !BinaryMessage

82− }

83− deriving (Generic, Show)

84−

85−type TimeoutMicroseconds = Int

86−

87−deriving instance Generic Fingerprint

88−deriving via WineryVariant Fingerprint instance Serialise Fingerprint

89−

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

91−data MessageType = RequestMessage TimeoutMicroseconds

92− | ResponseMessage

93− | TimeoutResponseMessage

94− | ExceptionResponseMessage

95− deriving (Generic, Show)

96− deriving Serialise via WineryVariant MessageType

97−

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

99−type RequestHandlers serverState = [RequestHandler serverState]

100−

101−-- | Data types for server-side request handlers, in synchronous (client waits for return value) and asynchronous (client does not wait for return value) forms.

102−data RequestHandler serverState where

103− -- | create a request handler with a response

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

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

122−

123--avoid orphan instance

70--avoid orphan instance

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

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

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

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

@@ -138,18 +85,12 @@ instance Serialise UUID where

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

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

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

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

140

141−-- | Errors from remote calls.

88+

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

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

143 | TimeoutError

90 | TimeoutError

144− | ExceptionError String

145 deriving (Generic, Show, Eq)

91 deriving (Generic, Show, Eq)

146 deriving Serialise via WineryVariant ConnectionError

92 deriving Serialise via WineryVariant ConnectionError

147

148−data TimeoutException = TimeoutException

149− deriving Show

150−

151−instance Exception TimeoutException

152−

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

94type HostAddr = (Word8, Word8, Word8, Word8)

154

155allHostAddrs,localHostAddr :: HostAddr

96allHostAddrs,localHostAddr :: HostAddr

@@ -157,186 +98,111 @@ allHostAddrs = (0,0,0,0)

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

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

158

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

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

160−envelopeP :: Parser IO Word8 Envelope

101+messageBoundaryP :: Parser IO Word8 BS.ByteString

161−envelopeP = do

102+messageBoundaryP = do

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

103+ --traceShowM "message!"

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

104+ w4x8 <- PD.take 4 FL.toList

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

105+ --traceShowM ("w4x8",w4x8)

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

106+ let c = fromIntegral (fromOctets w4x8)

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

107+ --traceShowM ("c", c)

167− lenPrefixedByteStringP = do

108+ vals <- PD.take c FL.toList

168− c <- fromIntegral <$> word32P

109+ let bytes = BS.pack vals

169− --streamly can't handle takeEQ 0, so add special handling

110+ traceShowM ("vals", BS.length bytes, bytes)

170− if c == 0 then

111+ pure bytes

171− pure BS.empty

172− else

173− fromArray <$> P.takeEQ c (Arr.writeN c)

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

175−

176−encodeEnvelope :: Envelope -> BS.ByteString

177−encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =

178− fingerprintBs <> msgTypeBs <> msgIdBs <> lenPrefixedBs

179− where

180− fingerprintBs = BO.bytestring64 fp1 <> BO.bytestring64 fp2

181− msgTypeBs = case msgType of

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

183− ResponseMessage -> BS.singleton 1

184− TimeoutResponseMessage -> BS.singleton 2

185− ExceptionResponseMessage -> BS.singleton 3

186− msgIdBs =

187− case UUIDBase.toWords (_unUUID msgId) of

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

189− msgLen = fromIntegral (BS.length bs)

190− lenPrefixedBs = BO.bytestring32 msgLen <> bs

191−

192−fingerprintP :: Parser IO Word8 Fingerprint

193−fingerprintP =

194− Fingerprint <$> word64P <*> word64P

195−

196−word64P :: Parser IO Word8 Word64

197−word64P = do

198− let s = FL.toList

199− b <- P.takeEQ 8 s

200− pure (BO.word64 (BS.pack b))

201−

202−--parse a 32-bit integer from network byte order

203−word32P :: Parser IO Word8 Word32

204−word32P = do

205− let s = FL.toList

206− w4x8 <- P.takeEQ 4 s

207− pure (BO.word32 (BS.pack w4x8))

208−

209−-- uuid is encode as 4 32-bit words because of its convenient 32-bit tuple encoding

210−uuidP :: Parser IO Word8 UUID

211−uuidP = do

212− u1 <- word32P

213− u2 <- word32P

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

219

112

220type NewConnectionHandler msg = IO (Maybe msg)

113type NewConnectionHandler msg = IO (Maybe msg)

221

114

222−type NewMessageHandler req resp = req -> IO resp

115+type NewMessageHandler msg resp = msg -> IO (HandlerResponse resp)

223−

116+

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.

117+serve :: (Show msg, Serialise msg, Serialise resp) =>

225−serve ::

118+ NewConnectionHandler resp ->

226− RequestHandlers s->

119+ NewMessageHandler msg resp ->

227− s ->

228 HostAddr ->

120 HostAddr ->

229 PortNumber ->

121 PortNumber ->

230 Maybe (MVar SockAddr) ->

122 Maybe (MVar SockAddr) ->

231 IO Bool

123 IO Bool

232−serve userMsgHandlers serverState hostaddr port mSockLock = do

124+serve connhandler msghandler hostaddr port mSockLock = do

233− let

234− handleSock sock = do

125+ let handleSock sock = do

235− lockingSocket <- newLock sock

126+ putStrLn "handleSock"

236− drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)

127+ let messageHandler msg = do

237−

128+ putStrLn $ "GOT MSG " ++ show msg

129+ case msg of

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

131+ ResponseExpectedRequest requestID val -> do

132+ putStrLn "ResponseExp"

133+ resp <- msghandler val

134+ case resp of

135+ HandlerResponse responseVal -> sendMessage (Response requestID responseVal) sock

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

137+ HandlerException _ -> error "TODO HandlerException"

138+ AsyncRequest _ val -> do

139+ putStrLn "AsyncReq"

140+ void $ msghandler val

141+ --no response necessaryxs

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

143+ -- allow the server to send an async welcome message to the new client, if necessary

144+ mResp <- connhandler

145+ case mResp of

146+ Nothing -> pure ()

147+ Just resp -> do

148+ requestID <- UUID <$> UUIDBase.nextRandom

149+ sendMessage (AsyncRequest requestID resp) sock

150+ drainSocketMessages sock messageHandler

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

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

239 pure True

152 pure True

240

153

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

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

242−openEnvelope (Envelope eprint _ _ bytes) =

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

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

244− case deserialise bytes of

245− Left err -> error (show err)

246− Right v -> Just v

247− else

248− Nothing

249

156

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

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

251− Envelope ->

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

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

254−matchEnvelope envelope dispatchf =

255− case openEnvelope envelope :: Maybe a of

256− Nothing -> Nothing

257− Just decoded -> Just (dispatchf, decoded)

258

158

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

159+drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()

260−serverEnvelopeHandler ::

261− Locking Socket

262− -> RequestHandlers s

263− -> s

264− -> Envelope

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

270− --find first matching handler

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

272− runTimeout m =

273− 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 =

287− case matchEnvelope envelope msghandler of

288− Nothing -> pure Nothing

289− Just (dispatchf, decoded) -> do

290− --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 =

301− case matchEnvelope envelope msghandler of

302− Nothing -> pure Nothing

303− Just (dispatchf, decoded) -> do

304− _ <- dispatchf sState decoded

305− pure (Just ())

306− 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 ()

313−

314−

315−type EnvelopeHandler = Envelope -> IO ()

316−

317−drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()

318−drainSocketMessages sock envelopeHandler = do

160+drainSocketMessages sock msgHandler = do

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

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

320− S.drain $ S.serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler

162+ handler bs = do

321−

163+ --print ("drain handler", bs)

322−--send length-tagged bytestring, perhaps should be in network byte order?

164+ case deserialise bs of

323−sendEnvelope :: Envelope -> Locking Socket -> IO ()

165+ Left err ->

324−sendEnvelope envelope sockLock = do

166+ print err

325− let envelopebytes = encodeEnvelope envelope

167+ Right val -> do

326− fullbytes = envelopebytes

168+ --putStrLn "deserialized value for msgHandler"

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

169+ msgHandler val --add response function

328− withLock sockLock $ \socket' ->

170+ S.drain $ S.parseManyD messageBoundaryP sockStream & S.mapM handler

329− Socket.sendAll socket' fullbytes

171+

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

172+fromOctets :: [Word8] -> Word32

331− traceBytes "sendEnvelope" fullbytes

173+fromOctets = foldl' accum 0

174+ where

175+ accum a o = (a `shiftL` 8) .|. fromIntegral o

332

176

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

177+octets :: Word32 -> [Word8]

334−fingerprint = typeRepFingerprint . typeOf

178+octets w =

179+ [ fromIntegral (w `shiftR` 24)

180+ , fromIntegral (w `shiftR` 16)

181+ , fromIntegral (w `shiftR` 8)

182+ , fromIntegral w

183+ ]

335

184

336−fromArray :: Arr.Array Word8 -> BSI.ByteString

185+--send length-tagged bytestring, perhaps should be in network byte order?

337−fromArray Arr.Array {..}

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

338− | aLen == 0 = mempty

187+sendMessage msg socket' = do

339− | otherwise = BSI.PS aStart 0 aLen

188+ putStrLn ("sending bytes: " ++ show (blen <> bytes))

189+ writeChunk socket' (toArrayS (blen <> bytes))

340 where

190 where

191+ bytes = serialise msg

192+ len = BS.length bytes

193+ blen = BS.pack (octets (fromIntegral len))

194+

195+

196+--from streamly-bytestring

197+-- | Convert a 'ByteString' to an array of 'Word8'. This function unwraps the

198+-- 'ByteString' and wraps it with 'Array' constructors and hence the operation

199+-- is performed in constant time.

200+{-# INLINE toArrayS #-}

201+toArrayS :: BS.ByteString -> SA.Array Word8

202+--slow path

203+--toArrayS = SA.fromList . BS.unpack

204+toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr

205+ where

206+ nfp = fp `plusForeignPtr` off

341− aStartPtr = unsafeForeignPtrToPtr aStart

207+ endPtr = unsafeForeignPtrToPtr nfp `plusPtr` len `plusPtr` 1 -- ? why +1?

342− aLen = aEnd `minusPtr` aStartPtr

208+

test/Curryer/Test/Basic.hs+52−178

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

1−{-# LANGUAGE DerivingVia, DeriveGeneric, TypeApplications, ExistentialQuantification #-}

1+{-# LANGUAGE DerivingVia, DeriveGeneric, TypeApplications #-}

2module Curryer.Test.Basic where

3import Test.Tasty

4import Test.Tasty.HUnit

@@ -7,124 +7,83 @@ import GHC.Generics

7import Control.Concurrent.MVar

8import Network.Socket (SockAddr(..))

9import Control.Concurrent.Async

10−import Control.Monad

11−import Control.Concurrent

12−import Control.Concurrent.STM

13−import Control.Exception

14−import Data.List

16import Network.RPC.Curryer.Server

11import Network.RPC.Curryer.Server

17import Network.RPC.Curryer.Client

12import Network.RPC.Curryer.Client

19−-- TODO: add test for nested calls

20−

21testTree :: TestTree

14testTree :: TestTree

22−testTree = testGroup "basic" [testCase "simple" testSimpleCall

15+testTree = testGroup "basic" [testCase "simple" testSimpleCall,

23− ,testCase "client async" testAsyncServerCall

16+ testCase "client async" testAsyncServerCall,

24− ,testCase "server async" testAsyncClientCall

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

30 ]

18 ]

32−

20+--client-to-server request

33−data AddTwoNumbersReq = AddTwoNumbersReq Int Int

21+data TestRequest = AddTwoNumbersReq Int Int

22+ | TestAsyncReq String

23+ | TestCallMeBackReq String

34 deriving (Generic, Show)

24 deriving (Generic, Show)

35− deriving Serialise via WineryVariant AddTwoNumbersReq

25+ deriving Serialise via WineryVariant TestRequest

37−data TestCallMeBackReq = TestCallMeBackReq String

27+--server responds to client

28+data TestResponse = AddTwoNumbersResp Int

29+ | AsyncHello String

30+ | TestCallMeBackResp String

38 deriving (Generic, Show, Eq)

31 deriving (Generic, Show, Eq)

39− deriving Serialise via WineryVariant TestCallMeBackReq

32+ deriving Serialise via WineryVariant TestResponse

40−

33+

41−data TestAsyncReq = TestAsyncReq String

34+testServerMessageHandler :: Maybe (MVar String) -> TestRequest -> IO (HandlerResponse TestResponse)

42− deriving (Generic, Show)

35+testServerMessageHandler mAsyncMVar msg = do

43− deriving Serialise via WineryVariant TestAsyncReq

36+ --print msg

44−

37+ case msg of

45−data DelayMicrosecondsReq = DelayMicrosecondsReq Int

38+ --round-trip request to add two Ints

46− deriving (Generic, Show)

47− deriving Serialise via WineryVariant DelayMicrosecondsReq

48−

49−data RoundtripStringReq = RoundtripStringReq String

50− deriving (Generic, Show)

51− deriving Serialise via WineryVariant RoundtripStringReq

52−

53−data ThrowServerSideExceptionReq = ThrowServerSideExceptionReq

54− deriving (Generic, Show)

55− deriving Serialise via WineryVariant ThrowServerSideExceptionReq

56−

57−data ChangeServerState = ChangeServerState

58− deriving (Generic, Show)

59− deriving Serialise via WineryVariant ChangeServerState

60−

61−--used to server -> client async request

62−data AsyncHelloReq = AsyncHelloReq String

63− deriving (Generic, Show)

64− deriving Serialise via WineryVariant AsyncHelloReq

65−

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)

39+ AddTwoNumbersReq x y -> pure (HandlerResponse (AddTwoNumbersResp (x+y)))

40+ --respond to a request for an async callback

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

41+ TestCallMeBackReq s -> do

74− case mAsyncMVar of

42+ case mAsyncMVar of

75− Nothing -> pure ()

43+ Nothing -> pure ()

76− Just mvar -> putMVar mvar s

44+ Just mvar -> putMVar mvar s

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

45+ pure NoResponse

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

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

82−

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

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

85− threadDelay ms

86− pure ()

87− , RequestHandler $ \_ ThrowServerSideExceptionReq -> do

88− _ <- error "test server exception"

89− pure ()

90− ]

92-- test a simple client-to-server round-trip function execution

47-- test a simple client-to-server round-trip function execution

93testSimpleCall :: Assertion

48testSimpleCall :: Assertion

94testSimpleCall = do

49testSimpleCall = do

95 readyVar <- newEmptyMVar

50 readyVar <- newEmptyMVar

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

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

98 --wait for server to be ready

53 --wait for server to be ready

99 (SockAddrInet port _) <- takeMVar readyVar

54 (SockAddrInet port _) <- takeMVar readyVar

55+ let clientHandler :: AsyncMessageHandler TestResponse

56+ clientHandler _ = error "async handler called"

57+ mkConn :: IO (Connection TestResponse)

100− conn <- connect [] localHostAddr port

58+ mkConn = connect clientHandler localHostAddr port

101− replicateM_ 5 $ do --make five AddTwo calls to shake out parallelism bugs

59+ conn <- mkConn

60+ let c :: IO (Either ConnectionError TestResponse)

102− x <- call conn (AddTwoNumbersReq 1 1)

61+ c = call conn (AddTwoNumbersReq 1 1)

62+ x <- c

103− assertEqual "server request+response" (Right (2 :: Int)) x

63+ assertEqual "server request+response" (Right (AddTwoNumbersResp 2)) x

104 close conn

64 close conn

105− cancel server

106−

107

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

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

109testAsyncServerCall :: Assertion

67testAsyncServerCall :: Assertion

110testAsyncServerCall = do

68testAsyncServerCall = do

111 portReadyVar <- newEmptyMVar

69 portReadyVar <- newEmptyMVar

112 receivedAsyncMessageVar <- newEmptyMVar

70 receivedAsyncMessageVar <- newEmptyMVar

113− let clientAsyncHandlers =

71+

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

72+ let testServerNewConnHandler :: NewConnectionHandler TestResponse

115− putMVar receivedAsyncMessageVar s)]

73+ testServerNewConnHandler = do

74+ pure (Just (AsyncHello "welcome"))

75+

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

76+ server <- async (serve testServerNewConnHandler (testServerMessageHandler Nothing) localHostAddr 0 (Just portReadyVar))

117 (SockAddrInet port _) <- takeMVar portReadyVar

77 (SockAddrInet port _) <- takeMVar portReadyVar

78+ let clientHandler :: AsyncMessageHandler TestResponse

79+ clientHandler msg = do

80+ case msg of

81+ AsyncHello str -> putMVar receivedAsyncMessageVar str

82+ x -> error ("unexpected message " <> show x)

118− conn <- connect clientAsyncHandlers localHostAddr port

83+ conn <- connect @TestResponse clientHandler localHostAddr port

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

120 asyncMessage <- takeMVar receivedAsyncMessageVar

84 asyncMessage <- takeMVar receivedAsyncMessageVar

121 assertEqual "async message" "welcome" asyncMessage

85 assertEqual "async message" "welcome" asyncMessage

122 close conn

86 close conn

123− cancel server

124−

125−

126−emptyServerState :: ()

127−emptyServerState = ()

128

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

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

130testAsyncClientCall :: Assertion

89testAsyncClientCall :: Assertion

@@ -132,97 +91,12 @@ testAsyncClientCall = do

132 portReadyVar <- newEmptyMVar

91 portReadyVar <- newEmptyMVar

133 receivedAsyncMessageVar <- newEmptyMVar

92 receivedAsyncMessageVar <- newEmptyMVar

134

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

94+ server <- async (serve (pure Nothing) (testServerMessageHandler (Just receivedAsyncMessageVar)) localHostAddr 0 (Just portReadyVar))

136 (SockAddrInet port _) <- takeMVar portReadyVar

95 (SockAddrInet port _) <- takeMVar portReadyVar

137−

96+

138− conn <- connect [] localHostAddr port

97+ conn <- connect @TestResponse (\_ -> pure ()) localHostAddr port

139 --send an async message, wait for an async response to confirm receipt

98 --send an async message, wait for an async response to confirm receipt

140 Right () <- asyncCall conn (TestCallMeBackReq "hi server")

99 Right () <- asyncCall conn (TestCallMeBackReq "hi server")

141 asyncMessage <- takeMVar receivedAsyncMessageVar

100 asyncMessage <- takeMVar receivedAsyncMessageVar

142 assertEqual "async message" "hi server" asyncMessage

101 assertEqual "async message" "hi server" asyncMessage

143 close conn

102 close conn

144− cancel server

145−

146−testSyncClientCallTimeout :: Assertion

147−testSyncClientCallTimeout = do

148− readyVar <- newEmptyMVar

149−

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

151− --wait for server to be ready

152− (SockAddrInet port _) <- takeMVar readyVar

153− conn <- connect [] localHostAddr port

154− x <- callTimeout @_ @Int (Just 500) conn (DelayMicrosecondsReq 1000)

155− assertEqual "client sync timeout" (Left TimeoutError) x

156− close conn

157− cancel server

158−

159−testSyncException :: Assertion

160−testSyncException = do

161− readyVar <- newEmptyMVar

162−

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

164− (SockAddrInet port _) <- takeMVar readyVar

165−

166− conn <- connect [] localHostAddr port

167− ret <- call conn ThrowServerSideExceptionReq

168− case ret of

169− Left (ExceptionError actualExc) ->

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

171− Left otherErr ->

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

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

174− close conn

175− cancel server

176−

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

178−testMultithreadedClient :: Assertion

179−testMultithreadedClient = do

180− readyVar <- newEmptyMVar

181−

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

183− (SockAddrInet port _) <- takeMVar readyVar

184− conn <- connect [] localHostAddr port

185− let bigString = replicate (1024 * 1000) 'x'

186− replicateM_ 10 $ do

187− ret <- call conn (RoundtripStringReq bigString)

188− assertEqual "big string multithread" (Right bigString) ret

189− --putStrLn "plus one"

190− close conn

191− cancel server

192−

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−