master..26ff225 · curryer

git.iko.soy/curryer

on 26ff225 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−

README.markdown+2−68

@@ -2,77 +2,11 @@

3Curryer (pun intended) is a fast, Haskell-exclusive RPC (remote procedure call) library. By using the latest Haskell serialization and streaming libraries, curryer aims to be the fastest and easiest means of communicating between Haskell-based processes.

5−Curryer is inspired by the now unmaintained [distributed-process](https://hackage.haskell.org/package/distributed-process) library, 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.

7## Features

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

12+* utilizes versionable shared ADT for communication instead of shared function `StaticPtr`s

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/Basic.hs+21−25

@@ -15,10 +15,10 @@ main :: IO ()

15main = do

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

17 portReadyMVar <- newEmptyMVar

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

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

19 --wait for server to be ready

20 (SockAddrInet port _) <- takeMVar portReadyMVar

21− clientConn <- connect [] localHostAddr port

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

23 _ <- syncWaitRequest clientConn 100

24 let syncWait ms = nfIO (syncWaitRequest clientConn ms)

@@ -30,10 +30,10 @@ main = do

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

31 bench "10 ms" (syncWait 10),

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

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

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

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

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

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

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

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

36+ bench "1000000 bytes" (syncByteString bs1M)

37 ]]

39 --TODO: bench for > PIPE_BUF bytes

@@ -41,38 +41,34 @@ main = do

41 close clientConn

42 cancel server

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

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

45+ | WaitByteStringReq BS.ByteString --ask the server to round trip arbitrary bytes

45 deriving (Generic, Show)

46 deriving (Generic, Show)

46− deriving Serialise via WineryVariant WaitMillisecondsReq

47+ deriving Serialise via WineryVariant BenchRequest

48−data WaitByteStringReq = WaitByteStringReq BS.ByteString

49+data BenchResponse = WaitMillisecondsResp

49− deriving (Generic, Show)

50+ | WaitByteStringResp BS.ByteString

50− deriving Serialise via WineryVariant WaitByteStringReq

51−

52−data WaitMillisecondsResp = WaitMillisecondsResp

53 deriving (Generic, Show, Eq)

51 deriving (Generic, Show, Eq)

54− deriving Serialise via WineryVariant WaitMillisecondsResp

52+ deriving Serialise via WineryVariant BenchResponse

56−benchmarkServerRequestHandlers :: RequestHandlers a

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

57−benchmarkServerRequestHandlers =

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

55+benchmarkServerMessageHandler (WaitMillisecondsReq ms) = do

59− threadDelay (1000 * ms)

56+ threadDelay (1000 * ms)

60− pure WaitMillisecondsResp,

57+ pure (HandlerResponse WaitMillisecondsResp)

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

58+benchmarkServerMessageHandler (WaitByteStringReq bs) = pure (HandlerResponse (WaitByteStringResp bs))

62− pure bs

63− ]

65--perform 10000 small synchronous requests

60--perform 10000 small synchronous requests

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

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

67syncWaitRequest conn ms = do

62syncWaitRequest conn ms = do

68 ret <- call conn (WaitMillisecondsReq ms)

63 ret <- call conn (WaitMillisecondsReq ms)

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

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

70 pure ms

65 pure ms

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

67+syncByteStringRequest :: Connection BenchResponse -> BS.ByteString -> IO BS.ByteString

73syncByteStringRequest conn bs = do

68syncByteStringRequest conn bs = do

74 ret <- call conn (WaitByteStringReq bs)

69 ret <- call conn (WaitByteStringReq bs)

75 case ret of

70 case ret of

76− Right bs' -> pure bs'

71+ Right (WaitByteStringResp bs') -> pure bs'

72+ Right _ -> error "invalid resp"

77 Left err -> error (show err)

73 Left err -> error (show err)

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

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

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

33 , hashable

34 , time

35 , network-byte-order

36− , stm

37 Hs-Source-Dirs: ./src

36 Hs-Source-Dirs: ./src

38 Default-Language: Haskell2010

37 Default-Language: Haskell2010

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

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

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

51 build-depends: tasty

50 build-depends: tasty

52 , tasty-hunit

51 , tasty-hunit

53 , base

52 , base

54− , curryer-rpc

53+ , curryer

55 , winery

54 , winery

56 , network

55 , network

57 , async

56 , async

58− , stm

59 other-modules: Curryer.Test.Basic

57 other-modules: Curryer.Test.Basic

61Benchmark perf

59Benchmark perf

62 Default-Language: Haskell2010

60 Default-Language: Haskell2010

63 Default-Extensions: OverloadedStrings

61 Default-Extensions: OverloadedStrings

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

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

66 Main-Is: Basic.hs

64 Main-Is: Basic.hs

67 Type: exitcode-stdio-1.0

65 Type: exitcode-stdio-1.0

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

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

69− -- -fprof-auto

70 HS-Source-Dirs: ./bench

67 HS-Source-Dirs: ./bench

72−Executable SimpleKeyValueServer

69+

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

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

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

1+{-# LANGUAGE TypeApplications, RankNTypes, ScopedTypeVariables #-}

2module Network.RPC.Curryer.Client where

3import Network.RPC.Curryer.Server

4import Network.Socket as Socket

@@ -11,32 +11,24 @@ import Control.Concurrent.MVar

11import GHC.Conc

12import Data.Time.Clock

13import System.Timeout

14−import Control.Monad

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

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

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

17+-- the request map holds

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

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

20− _conn_asyncThread :: Async (),

19+ _conn_asyncThread :: Async (),

21− _conn_syncmap :: SyncMap

20+ _conn_syncmap :: SyncMap

22− }

21+ }

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.

31connect ::

23connect ::

32− ClientAsyncRequestHandlers ->

24+ --AsyncMessageHandler ->

33− HostAddr ->

25+ HostAddr ->

34− PortNumber ->

26+ PortNumber ->

35− IO Connection

27+ IO Connection

36−connect asyncHandlers hostAddr portNum = do

28+connect hostAddr portNum = do

37 sock <- TCP.connect hostAddr portNum

29 sock <- TCP.connect hostAddr portNum

38 syncmap <- STMMap.newIO

30 syncmap <- STMMap.newIO

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

31+ asyncThread <- async (clientAsync sock syncmap)

40 sockLock <- newLock sock

32 sockLock <- newLock sock

41 pure (Connection {

33 pure (Connection {

42 _conn_sockLock = sockLock,

34 _conn_sockLock = sockLock,

@@ -44,75 +36,50 @@ connect asyncHandlers hostAddr portNum = do

44 _conn_syncmap = syncmap

36 _conn_syncmap = syncmap

45 })

37 })

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

48close :: Connection -> IO ()

39close :: Connection -> IO ()

49close conn = do

40close conn = do

50 withLock (_conn_sockLock conn) $ \sock ->

41 withLock (_conn_sockLock conn) $ \sock ->

51 Socket.close sock

42 Socket.close sock

52 cancel (_conn_asyncThread conn)

43 cancel (_conn_asyncThread conn)

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

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

55clientAsync ::

46clientAsync ::

56 Socket ->

47 Socket ->

57 SyncMap ->

48 SyncMap ->

58− ClientAsyncRequestHandlers ->

49+ --AsyncMessageHandler ->

59 IO ()

50 IO ()

60−clientAsync sock syncmap asyncHandlers = do

51+clientAsync sock syncmap = do

61 lsock <- newLock sock

52 lsock <- newLock sock

62− drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)

53+ drainSocketMessages sock (clientEnvelopeHandler lsock syncmap)

64−consumeResponse :: UUID -> STMMap.Map UUID (MVar a, b) -> a -> IO ()

55+--handles envelope responses from server

65−consumeResponse msgId syncMap val = do

56+clientEnvelopeHandler ::

57+ Locking Socket

58+ -> SyncMap

59+ -> Envelope

60+ -> IO ()

61+clientEnvelopeHandler _ _ (Envelope _ RequestMessage _ _) = error "client received request"

62+clientEnvelopeHandler _ syncMap (Envelope _ ResponseMessage msgId binaryMessage) = do

63+ --find a matching response item

66 match <- atomically $ do

64 match <- atomically $ do

67 val' <- STMMap.lookup msgId syncMap

65 val' <- STMMap.lookup msgId syncMap

68 STMMap.delete msgId syncMap

66 STMMap.delete msgId syncMap

69 pure val'

67 pure val'

70 case match of

68 case match of

71− Nothing -> pure () -- drop message

69+ Nothing -> error ("dropping unrequested response " <> show msgId)

72− Just (mVar, _) -> putMVar mVar val

70+ Just (mVar, _) ->

71+ putMVar mVar (Right binaryMessage)

72+

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.

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

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

103call = callTimeout Nothing

75call = 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.

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

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

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

107callTimeout mTimeout conn msg = do

79callTimeout mTimeout conn msg = do

108 requestID <- UUID <$> UUIDBase.nextRandom

80 requestID <- UUID <$> UUIDBase.nextRandom

109 let mVarMap = _conn_syncmap conn

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

82+ envelope = Envelope fprint RequestMessage requestID (serialise msg)

116 fprint = fingerprint msg

83 fprint = fingerprint msg

117 -- setup mvar to wait for response

84 -- setup mvar to wait for response

118 responseMVar <- newEmptyMVar

85 responseMVar <- newEmptyMVar

@@ -129,19 +96,17 @@ callTimeout mTimeout conn msg = do

129 --timeout

96 --timeout

130 Nothing ->

97 Nothing ->

131 pure (Left TimeoutError)

98 pure (Left TimeoutError)

132− Just (Left exc) ->

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

133− pure (Left exc)

134 Just (Right binmsg) ->

100 Just (Right binmsg) ->

135 --TODO use decoder instead

101 --TODO use decoder instead

136 case deserialise binmsg of

102 case deserialise binmsg of

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

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

138 Right m -> pure (Right m)

104 Right m -> pure (Right m)

139−

105+{-

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

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

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

142asyncCall conn msg = do

107asyncCall conn msg = do

143 requestID <- UUID <$> UUIDBase.nextRandom

108 requestID <- UUID <$> UUIDBase.nextRandom

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

109+ sendMessage (AsyncRequest requestID msg) (_conn_sockLock conn)

145− fprint = fingerprint msg

146− sendEnvelope envelope (_conn_sockLock conn)

147 pure (Right ())

110 pure (Right ())

111+

112+-}

src/Network/RPC/Curryer/Server.hs+115−150

@@ -1,7 +1,8 @@

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

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

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

4module Network.RPC.Curryer.Server where

5+import Streamly

5import qualified Streamly.Prelude as S

6import qualified Streamly.Prelude as S

6import Streamly.Network.Socket

7import Streamly.Network.Socket

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

8import Streamly.Internal.Network.Socket (handleWithM)

@@ -25,24 +26,21 @@ import Streamly.Data.Fold as FL

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

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

26import qualified Data.Binary as B

27import qualified Data.Binary as B

27import qualified Data.UUID as UUIDBase

28import qualified Data.UUID as UUIDBase

28−import qualified Data.UUID.V4 as UUIDBase

29−import Control.Monad

30−import Data.Functor

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

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

33--import Control.Monad

31--import Control.Monad

34import Data.Hashable

32import Data.Hashable

35−import System.Timeout

33+--import System.Timeout

36import qualified Network.ByteOrder as BO

34import qualified Network.ByteOrder as BO

38−

39-- for toArrayS conversion

36-- for toArrayS conversion

40−import qualified Data.ByteString.Internal as BSI

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

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

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

39+import Foreign.ForeignPtr (plusForeignPtr)

42import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

40import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

43−import GHC.Ptr (minusPtr)

41+import GHC.Ptr (plusPtr)

44−

42+-}

45−-- define CURRYER_SHOW_BYTES 1

43+-- #define CURRYER_SHOW_BYTES 1

46#if CURRYER_SHOW_BYTES == 1

44#if CURRYER_SHOW_BYTES == 1

47import Debug.Trace

45import Debug.Trace

48#endif

46#endif

@@ -68,58 +66,45 @@ withLock (Locking mvar v) m =

68lockless :: Locking a -> a

66lockless :: Locking a -> a

69lockless (Locking _ a) = a

67lockless (Locking _ a) = a

71−type Timeout = Word32

69+type Timeout = Int

73type BinaryMessage = BS.ByteString

71type BinaryMessage = BS.ByteString

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

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

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

77−data Envelope = Envelope {

75+data Envelope = Envelope !Fingerprint !MessageType !UUID !BinaryMessage

78− envFingerprint :: !Fingerprint,

79− envMessageType :: !MessageType,

80− envMsgId :: !UUID,

81− envPayload :: !BinaryMessage

82− }

83 deriving (Generic, Show)

76 deriving (Generic, Show)

84−

77+ deriving Serialise via WineryVariant Envelope

85−type TimeoutMicroseconds = Int

87deriving instance Generic Fingerprint

79deriving instance Generic Fingerprint

88deriving via WineryVariant Fingerprint instance Serialise Fingerprint

80deriving via WineryVariant Fingerprint instance Serialise Fingerprint

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

91−data MessageType = RequestMessage TimeoutMicroseconds

82+data MessageType = RequestMessage

92 | ResponseMessage

83 | ResponseMessage

93− | TimeoutResponseMessage

94− | ExceptionResponseMessage

95 deriving (Generic, Show)

84 deriving (Generic, Show)

96 deriving Serialise via WineryVariant MessageType

85 deriving Serialise via WineryVariant MessageType

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

87+data MessageHandlers =

99−type RequestHandlers serverState = [RequestHandler serverState]

88+ forall a b. (Show a, Serialise a, Serialise b) => MessageHandlers [RequestHandler a b]

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

90+type RequestHandler a b = a -> IO b

91+

92+{-data Message a = Response UUID a

93+ | AsyncRequest UUID a

94+ | ResponseExpectedRequest UUID (Maybe Timeout) a

95+ | TimedOutResponse UUID

96+ | ExceptionResponse UUID String

97+ deriving (Generic, Show)

98+ deriving Serialise via WineryVariant (Message a)

99+

100+-- | The response type

101+

102+data HandlerResponse a = NoResponse

103+ | HandlerResponse a

104+ | HandlerException String

105+ deriving Generic

106+ deriving Serialise via WineryVariant (HandlerResponse a)

107+-}

123--avoid orphan instance

108--avoid orphan instance

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

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

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

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

@@ -138,18 +123,13 @@ instance Serialise UUID where

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

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

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

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

140

125

141−-- | Errors from remote calls.

126+

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

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

143 | TimeoutError

128 | TimeoutError

144 | ExceptionError String

129 | ExceptionError String

145 deriving (Generic, Show, Eq)

130 deriving (Generic, Show, Eq)

146 deriving Serialise via WineryVariant ConnectionError

131 deriving Serialise via WineryVariant ConnectionError

147

132

148−data TimeoutException = TimeoutException

149− deriving Show

150−

151−instance Exception TimeoutException

152−

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

133type HostAddr = (Word8, Word8, Word8, Word8)

154

134

155allHostAddrs,localHostAddr :: HostAddr

135allHostAddrs,localHostAddr :: HostAddr

@@ -159,18 +139,12 @@ localHostAddr = (127,0,0,1)

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

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

160envelopeP :: Parser IO Word8 Envelope

140envelopeP :: Parser IO Word8 Envelope

161envelopeP = do

141envelopeP = do

142+ let s = FL.toList

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

143+ msgTypeP = (P.satisfy (== 0) *> pure RequestMessage) `P.alt`

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

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

144+ (P.satisfy (== 1) *> pure ResponseMessage)

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

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

167 lenPrefixedByteStringP = do

145 lenPrefixedByteStringP = do

168 c <- fromIntegral <$> word32P

146 c <- fromIntegral <$> word32P

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

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

170− if c == 0 then

171− pure BS.empty

172− else

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

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

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

175

149

176encodeEnvelope :: Envelope -> BS.ByteString

150encodeEnvelope :: Envelope -> BS.ByteString

@@ -179,19 +153,19 @@ encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =

179 where

153 where

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

154 fingerprintBs = BO.bytestring64 fp1 <> BO.bytestring64 fp2

181 msgTypeBs = case msgType of

155 msgTypeBs = case msgType of

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

156+ RequestMessage -> BS.singleton 0

183 ResponseMessage -> BS.singleton 1

157 ResponseMessage -> BS.singleton 1

184− TimeoutResponseMessage -> BS.singleton 2

185− ExceptionResponseMessage -> BS.singleton 3

186 msgIdBs =

158 msgIdBs =

187 case UUIDBase.toWords (_unUUID msgId) of

159 case UUIDBase.toWords (_unUUID msgId) of

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

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

189 msgLen = fromIntegral (BS.length bs)

161 msgLen = fromIntegral (BS.length bs)

190 lenPrefixedBs = BO.bytestring32 msgLen <> bs

162 lenPrefixedBs = BO.bytestring32 msgLen <> bs

191

163

192fingerprintP :: Parser IO Word8 Fingerprint

164fingerprintP :: Parser IO Word8 Fingerprint

193−fingerprintP =

165+fingerprintP = do

166+ f1 <- word64P

167+ f2 <- word64P

194− Fingerprint <$> word64P <*> word64P

168+ pure (Fingerprint f1 f2)

195

169

196word64P :: Parser IO Word8 Word64

170word64P :: Parser IO Word8 Word64

197word64P = do

171word64P = do

@@ -212,112 +186,112 @@ uuidP = do

212 u1 <- word32P

186 u1 <- word32P

213 u2 <- word32P

187 u2 <- word32P

214 u3 <- word32P

188 u3 <- word32P

215− --u4 <- word32P

189+ u4 <- word32P

216− --pure (UUID (UUIDBase.fromWords u1 u2 u3 u4))-}

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

191

220type NewConnectionHandler msg = IO (Maybe msg)

192type NewConnectionHandler msg = IO (Maybe msg)

221

193

222type NewMessageHandler req resp = req -> IO resp

194type NewMessageHandler req resp = req -> IO resp

223−

195+

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.

225serve ::

196serve ::

226− RequestHandlers s->

197+ MessageHandlers ->

227− s ->

228 HostAddr ->

198 HostAddr ->

229 PortNumber ->

199 PortNumber ->

230 Maybe (MVar SockAddr) ->

200 Maybe (MVar SockAddr) ->

231 IO Bool

201 IO Bool

232−serve userMsgHandlers serverState hostaddr port mSockLock = do

202+serve userMsgHandler hostaddr port mSockLock = do

233 let

203 let

234 handleSock sock = do

204 handleSock sock = do

235 lockingSocket <- newLock sock

205 lockingSocket <- newLock sock

236− drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)

206+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandler)

237

207

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

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

239 pure True

209 pure True

240

210

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

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

213+ error "spam"

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

215+ error "spam2"

216+-}

217+

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

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

242−openEnvelope (Envelope eprint _ _ bytes) =

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

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

221+ Left err -> error (show err)

222+ Right dec -> dec-}

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

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

244 case deserialise bytes of

224 case deserialise bytes of

245 Left err -> error (show err)

225 Left err -> error (show err)

246 Right v -> Just v

226 Right v -> Just v

227+ --Just (evalDecoder decoder bytes)

247 else

228 else

248 Nothing

229 Nothing

249

230

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

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−

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

260serverEnvelopeHandler ::

231serverEnvelopeHandler ::

261 Locking Socket

232 Locking Socket

262− -> RequestHandlers s

233+ -> MessageHandlers

263− -> s

264 -> Envelope

234 -> Envelope

265 -> IO ()

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

236+serverEnvelopeHandler sockLock (MessageHandlers msgHandlers) envelope@(Envelope _ RequestMessage msgId _) = do

270 --find first matching handler

237 --find first matching handler

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

238+ let firstMatch = foldr firstMatcher Nothing msgHandlers

272− runTimeout m =

239+ firstMatcher :: forall a b. (Serialise a, Serialise b) => RequestHandler a b -> Maybe (RequestHandler a b, a) -> Maybe (RequestHandler a b, a)

273− if timeoutms == 0 then

240+ firstMatcher msghandler Nothing =

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

241+ case openEnvelope envelope :: Maybe a of

275− else

242+ Just decoded -> Just (msghandler, decoded)

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

243+ Nothing -> Nothing

277− --allow server-side function to throw TimeoutError which is caught here and becomes TimeoutError value

244+ firstMatcher _ acc = acc

278− timeoutExcHandler :: TimeoutException -> IO (Maybe b)

245+ case firstMatch of

279− timeoutExcHandler _ = pure Nothing

246+ Nothing -> error "failed to handle msg"

247+ Just (msghandler, decoded) -> do

248+ --TODO add exception handling

249+ response <- msghandler decoded

250+ sendEnvelope (Envelope (fingerprint response) ResponseMessage msgId (serialise response)) sockLock

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

252+{-

280

253

281− sState = ConnectionState {

254+

282− connectionServerState = serverState,

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

283− connectionSocket = sockLock

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

284− }

257+ runTimeout mTimeout m = case mTimeout of

285−

286− firstMatcher (RequestHandler msghandler) Nothing =

287− case matchEnvelope envelope msghandler of

288− Nothing -> pure Nothing

258+ Nothing -> Just <$> m

289− Just (dispatchf, decoded) -> do

259+ Just timeoutMicroseconds -> timeout timeoutMicroseconds m

290− --TODO add exception handling

260+ case msg of

291− mResponse <- runTimeout (dispatchf sState decoded)

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

292− let envelopeResponse =

262+ ResponseExpectedRequest requestID mTimeout val -> do

293− case mResponse of

263+ let normalResponder = do

294− Just response ->

264+ resp <- runTimeout mTimeout (requestMessageHandler val)

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

265+ case resp of

296− Nothing ->

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

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

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

298− sendEnvelope envelopeResponse sockLock

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

299− pure (Just ())

269+ Nothing ->

300− firstMatcher (AsyncRequestHandler msghandler) Nothing =

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

301− case matchEnvelope envelope msghandler of

271+ excHandler :: SomeException -> IO ()

302− Nothing -> pure Nothing

272+ excHandler e = do

303− Just (dispatchf, decoded) -> do

273+ --send exception to client

304− _ <- dispatchf sState decoded

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

305− pure (Just ())

275+ throwIO e

306− firstMatcher _ acc = pure acc

276+ catch normalResponder excHandler

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

277+ AsyncRequest _ val ->

308− case eExc of

278+ void $ requestMessageHandler val

309− Left exc ->

279+ --no response necessary

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

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

311− sendEnvelope env sockLock

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

312− Right () -> pure ()

282+-}

313−

283+

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

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

314

286

315type EnvelopeHandler = Envelope -> IO ()

287type EnvelopeHandler = Envelope -> IO ()

316

288

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

290+

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

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

318drainSocketMessages sock envelopeHandler = do

292drainSocketMessages sock envelopeHandler = do

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

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

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

294+ S.drain $ serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler

321

295

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

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

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

297sendEnvelope :: Envelope -> Locking Socket -> IO ()

@@ -327,16 +301,7 @@ sendEnvelope envelope sockLock = do

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

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

328 withLock sockLock $ \socket' ->

302 withLock sockLock $ \socket' ->

329 Socket.sendAll socket' fullbytes

303 Socket.sendAll socket' fullbytes

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

331 traceBytes "sendEnvelope" fullbytes

304 traceBytes "sendEnvelope" fullbytes

332

305

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

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

334fingerprint = typeRepFingerprint . typeOf

307fingerprint = typeRepFingerprint . typeOf

335−

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

337−fromArray Arr.Array {..}

338− | aLen == 0 = mempty

339− | otherwise = BSI.PS aStart 0 aLen

340− where

341− aStartPtr = unsafeForeignPtrToPtr aStart

342− aLen = aEnd `minusPtr` aStartPtr

test/Curryer/Test/Basic.hs+96−135

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

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

9import Control.Concurrent.Async

10import Control.Monad

11import Control.Concurrent

12−import Control.Concurrent.STM

13−import Control.Exception

14import Data.List

12import Data.List

16import Network.RPC.Curryer.Server

14import Network.RPC.Curryer.Server

17import Network.RPC.Curryer.Client

15import Network.RPC.Curryer.Client

17+import Debug.Trace

18+

19-- TODO: add test for nested calls

21testTree :: TestTree

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

23− ,testCase "client async" testAsyncServerCall

23+ {-testCase "client async" testAsyncServerCall,

24− ,testCase "server async" testAsyncClientCall

24+ testCase "server async" testAsyncClientCall,

25− ,testCase "client sync timeout" testSyncClientCallTimeout

25+ testCase "client sync timeout" testSyncClientCallTimeout,

26− ,testCase "server-side exception" testSyncException

26+ testCase "server-side exception" testSyncException,

27− ,testCase "multi-threaded client" testMultithreadedClient

27+ testCase "multi-threaded client" testMultithreadedClient-}

28− ,testCase "server state" testServerState

29− ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout

30 ]

28 ]

32−

30+--client-to-server request

33−data AddTwoNumbersReq = AddTwoNumbersReq Int Int

31+data TestRequest = AddTwoNumbersReq Int Int

32+ | TestAsyncReq String

33+ | TestCallMeBackReq String

34+ | RoundtripStringReq String

35+ | DelayMicrosecondsReq Int

36+ | ThrowServerSideExceptionReq

34 deriving (Generic, Show)

37 deriving (Generic, Show)

35− deriving Serialise via WineryVariant AddTwoNumbersReq

38+ deriving Serialise via WineryVariant TestRequest

37−data TestCallMeBackReq = TestCallMeBackReq String

40+--server responds to client

41+data TestResponse = AddTwoNumbersResp Int

42+ | AsyncHello String

43+ | DelayMicrosecondsResp

44+ | RoundtripStringResp String

38 deriving (Generic, Show, Eq)

45 deriving (Generic, Show, Eq)

39− deriving Serialise via WineryVariant TestCallMeBackReq

46+ deriving Serialise via WineryVariant TestResponse

40−

47+

41−data TestAsyncReq = TestAsyncReq String

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

42− deriving (Generic, Show)

49+testServerMessageHandlers mAsyncMVar =

43− deriving Serialise via WineryVariant TestAsyncReq

44−

45−data DelayMicrosecondsReq = DelayMicrosecondsReq Int

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)

50+ MessageHandlers [\(AddTwoNumbersReq x y) ->

51+ pure (AddTwoNumbersResp (x + y))]

52+{- --print msg

53+ case msg of

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

55+ AddTwoNumbersReq x y -> pure (AddTwoNumbersResp (x+y))

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

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

57+ TestCallMeBackReq s -> do

74− case mAsyncMVar of

58+ case mAsyncMVar of

75− Nothing -> pure ()

59+ Nothing -> pure ()

76− Just mvar -> putMVar mvar s

60+ Just mvar -> putMVar mvar s

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

61+ pure NoResponse

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

62+ TestAsyncReq _ -> pure NoResponse

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

63+ RoundtripStringReq s -> pure (RoundtripStringResp s)

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

64+ DelayMicrosecondsReq ms -> do

85 threadDelay ms

65 threadDelay ms

86− pure ()

66+ pure (HandlerResponse DelayMicrosecondsResp)

87− , RequestHandler $ \_ ThrowServerSideExceptionReq -> do

88− _ <- error "test server exception"

67+ ThrowServerSideExceptionReq -> error "test server exception"

89− pure ()

68+-}

90− ]

91−

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

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

93testSimpleCall :: Assertion

70testSimpleCall :: Assertion

94testSimpleCall = do

71testSimpleCall = do

95 readyVar <- newEmptyMVar

72 readyVar <- newEmptyMVar

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

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

98 --wait for server to be ready

75 --wait for server to be ready

99 (SockAddrInet port _) <- takeMVar readyVar

76 (SockAddrInet port _) <- takeMVar readyVar

77+ let clientHandler :: AsyncMessageHandler TestResponse

78+ clientHandler _ = error "async handler called"

79+ mkConn :: IO Connection

100− conn <- connect [] localHostAddr port

80+ mkConn = connect localHostAddr port

81+ conn <- mkConn

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

83+ c = call conn (AddTwoNumbersReq 1 1)

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

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

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

85+ x <- c

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

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

104 close conn

87 close conn

105 cancel server

88 cancel server

106

107−

90+{-

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

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

109testAsyncServerCall :: Assertion

92testAsyncServerCall :: Assertion

110testAsyncServerCall = do

93testAsyncServerCall = do

111 portReadyVar <- newEmptyMVar

94 portReadyVar <- newEmptyMVar

112 receivedAsyncMessageVar <- newEmptyMVar

95 receivedAsyncMessageVar <- newEmptyMVar

113− let clientAsyncHandlers =

96+

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

97+ let testServerNewConnHandler :: NewConnectionHandler TestResponse

115− putMVar receivedAsyncMessageVar s)]

98+ testServerNewConnHandler =

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

100+

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

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

117 (SockAddrInet port _) <- takeMVar portReadyVar

102 (SockAddrInet port _) <- takeMVar portReadyVar

103+ let clientHandler :: AsyncMessageHandler TestResponse

104+ clientHandler msg = do

105+ case msg of

106+ AsyncHello str -> putMVar receivedAsyncMessageVar str

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

118− conn <- connect clientAsyncHandlers localHostAddr port

108+ conn <- connect @TestResponse clientHandler localHostAddr port

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

120 asyncMessage <- takeMVar receivedAsyncMessageVar

109 asyncMessage <- takeMVar receivedAsyncMessageVar

121 assertEqual "async message" "welcome" asyncMessage

110 assertEqual "async message" "welcome" asyncMessage

122 close conn

111 close conn

123 cancel server

112 cancel server

124

113

125−

126−emptyServerState :: ()

127−emptyServerState = ()

128−

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

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

130testAsyncClientCall :: Assertion

115testAsyncClientCall :: Assertion

131testAsyncClientCall = do

116testAsyncClientCall = do

132 portReadyVar <- newEmptyMVar

117 portReadyVar <- newEmptyMVar

133 receivedAsyncMessageVar <- newEmptyMVar

118 receivedAsyncMessageVar <- newEmptyMVar

134

119

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

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

136 (SockAddrInet port _) <- takeMVar portReadyVar

121 (SockAddrInet port _) <- takeMVar portReadyVar

137−

122+

138− conn <- connect [] localHostAddr port

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

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

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

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

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

141 asyncMessage <- takeMVar receivedAsyncMessageVar

126 asyncMessage <- takeMVar receivedAsyncMessageVar

@@ -147,30 +132,37 @@ testSyncClientCallTimeout :: Assertion

147testSyncClientCallTimeout = do

132testSyncClientCallTimeout = do

148 readyVar <- newEmptyMVar

133 readyVar <- newEmptyMVar

149

134

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

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

151 --wait for server to be ready

136 --wait for server to be ready

152 (SockAddrInet port _) <- takeMVar readyVar

137 (SockAddrInet port _) <- takeMVar readyVar

138+ let clientHandler :: AsyncMessageHandler TestResponse

139+ clientHandler _ = error "async handler called"

140+ mkConn :: IO (Connection TestResponse)

153− conn <- connect [] localHostAddr port

141+ mkConn = connect clientHandler localHostAddr port

142+ conn <- mkConn

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

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

144+ c = callTimeout (Just 500) conn (DelayMicrosecondsReq 1000)

145+ x <- c

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

146 assertEqual "client sync timeout" (Left TimeoutError) x

156 close conn

147 close conn

157 cancel server

148 cancel server

158−

149+

159testSyncException :: Assertion

150testSyncException :: Assertion

160testSyncException = do

151testSyncException = do

161 readyVar <- newEmptyMVar

152 readyVar <- newEmptyMVar

162

153

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

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

164 (SockAddrInet port _) <- takeMVar readyVar

155 (SockAddrInet port _) <- takeMVar readyVar

165−

156+ let clientHandler :: AsyncMessageHandler TestResponse

157+ clientHandler _ = error "async handler called"

158+ mkConn :: IO (Connection TestResponse)

166− conn <- connect [] localHostAddr port

159+ mkConn = connect clientHandler localHostAddr port

160+ conn <- mkConn

167 ret <- call conn ThrowServerSideExceptionReq

161 ret <- call conn ThrowServerSideExceptionReq

168 case ret of

162 case ret of

169 Left (ExceptionError actualExc) ->

163 Left (ExceptionError actualExc) ->

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

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

171− Left otherErr ->

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

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

165+ Right _ -> assertFailure "missed exception"

174 close conn

166 close conn

175 cancel server

167 cancel server

176

168

@@ -179,50 +171,19 @@ testMultithreadedClient :: Assertion

179testMultithreadedClient = do

171testMultithreadedClient = do

180 readyVar <- newEmptyMVar

172 readyVar <- newEmptyMVar

181

173

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

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

183 (SockAddrInet port _) <- takeMVar readyVar

175 (SockAddrInet port _) <- takeMVar readyVar

176+ let clientHandler :: AsyncMessageHandler TestResponse

177+ clientHandler _ = error "async handler called"

178+ mkConn :: IO (Connection TestResponse)

184− conn <- connect [] localHostAddr port

179+ mkConn = connect clientHandler localHostAddr port

180+ conn <- mkConn

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

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

186 replicateM_ 10 $ do

182 replicateM_ 10 $ do

187 ret <- call conn (RoundtripStringReq bigString)

183 ret <- call conn (RoundtripStringReq bigString)

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

184+ assertEqual "big string multithread" (Right (RoundtripStringResp bigString)) ret

189− --putStrLn "plus one"

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

186 close conn

227 cancel server

187 cancel server

228−

188+

189+-}