c95cb38..master · curryer

git.iko.soy/curryer

on master help sign in

Changelog.markdownnew file+4

since splitsnapshots

unifiedsplit

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

1+# v0.1 (2020-12-27)

3+* initial release to support [Project:M36](https://github.com/agentm/project-m36)

README.markdown+68−2

@@ -1,12 +1,78 @@

1# Curryer - Fast Haskell-to-Haskell RPC

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 the fastest and easiest means of communicating between Haskell-based processes.

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.

5−Curryer is inspired by the now unmaintained [distributed-process](https://hackage.haskell.org/package/distributed-process) library.

5+Curryer is inspired by the now unmaintained [distributed-process](https://hackage.haskell.org/package/distributed-process) library, but is lighter-weight and uses a higher-performance serialization package.

7## Features

9* blocking and non-blocking remote function calls

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

11* timeouts

12+* leverages [winery](https://hackage.haskell.org/package/winery) for high-performance serialization

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.shnew file+1

@@ -0,0 +1 @@

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

bench/Basic.hs+43−13

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

9import Control.Concurrent.Async

10import Control.Concurrent

11import Control.Monad

12+import qualified Data.ByteString as BS

13main :: IO ()

14main :: IO ()

14main = do

15main = do

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

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

16 portReadyMVar <- newEmptyMVar

17 portReadyMVar <- newEmptyMVar

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

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

18 --wait for server to be ready

19 --wait for server to be ready

19 (SockAddrInet port _) <- takeMVar portReadyMVar

20 (SockAddrInet port _) <- takeMVar portReadyMVar

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

21+ clientConn <- connect [] localHostAddr port

22 _ <- syncWaitRequest clientConn 100

23 _ <- 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)

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

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)

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"

26 close clientConn

41 close clientConn

27 cancel server

42 cancel server

29−data BenchRequest = WaitMillisecondsReq Int

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

45+ deriving (Generic, Show)

46+ deriving Serialise via WineryVariant WaitMillisecondsReq

47+

48+data WaitByteStringReq = WaitByteStringReq BS.ByteString

30 deriving (Generic, Show)

49 deriving (Generic, Show)

31− deriving Serialise via WineryVariant BenchRequest

50+ deriving Serialise via WineryVariant WaitByteStringReq

33−data BenchResponse = WaitMillisecondsResp

52+data WaitMillisecondsResp = WaitMillisecondsResp

34 deriving (Generic, Show, Eq)

53 deriving (Generic, Show, Eq)

35− deriving Serialise via WineryVariant BenchResponse

54+ deriving Serialise via WineryVariant WaitMillisecondsResp

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

56+benchmarkServerRequestHandlers :: RequestHandlers a

57+benchmarkServerRequestHandlers =

38−benchmarkServerMessageHandler (WaitMillisecondsReq ms) = do

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

39− threadDelay (1000 * ms)

59+ threadDelay (1000 * ms)

40− pure (HandlerResponse WaitMillisecondsResp)

60+ pure WaitMillisecondsResp,

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

62+ pure bs

63+ ]

42--perform 10000 small synchronous requests

65--perform 10000 small synchronous requests

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

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

44syncWaitRequest conn ms = do

67syncWaitRequest conn ms = do

45 ret <- call conn (WaitMillisecondsReq ms)

68 ret <- call conn (WaitMillisecondsReq ms)

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

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

47 pure ms

70 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: e218fb9d8f4263fd69d32a2ff318bac0e179c642

6+ tag: 4e48b52e8709390a47e99c107ebed2d1c9420076

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

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

1−Name: curryer

1+Name: curryer-rpc

2Version: 0.1

3License: PublicDomain

4Build-Type: Simple

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

11Cabal-Version: >= 1.10

12Synopsis: Fast, Haskell RPC

13Description: Haskell-to-Haskell RPC using Winery serialization.

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

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

16Source-Repository head

17 Type: git

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

20Library

21− Build-Depends: base >= 4.8

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

22 , winery

23 , bytestring

24 , streamly >= 0.7.2

@@ -32,6 +32,8 @@ Library

32 , stm-containers

33 , hashable

34 , time

35+ , network-byte-order

36+ , stm

35 Hs-Source-Dirs: ./src

37 Hs-Source-Dirs: ./src

36 Default-Language: Haskell2010

38 Default-Language: Haskell2010

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

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

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

49 build-depends: tasty

51 build-depends: tasty

50 , tasty-hunit

52 , tasty-hunit

51 , base

53 , base

52− , curryer

54+ , curryer-rpc

53 , winery

55 , winery

54 , network

56 , network

55 , async

57 , async

58+ , stm

56 other-modules: Curryer.Test.Basic

59 other-modules: Curryer.Test.Basic

58Benchmark perf

61Benchmark perf

59 Default-Language: Haskell2010

62 Default-Language: Haskell2010

60 Default-Extensions: OverloadedStrings

63 Default-Extensions: OverloadedStrings

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

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

65+

62 Main-Is: Basic.hs

66 Main-Is: Basic.hs

63 Type: exitcode-stdio-1.0

67 Type: exitcode-stdio-1.0

64 GHC-Options: -Wall -threaded -rtsopts

68 GHC-Options: -Wall -threaded -rtsopts

69+ -- -fprof-auto

65 HS-Source-Dirs: ./bench

70 HS-Source-Dirs: ./bench

66−

71+

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.hsnew file+39

@@ -0,0 +1,39 @@

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

2+import GHC.Generics

3+import Network.RPC.Curryer.Client

4+import Network.RPC.Curryer.Server (localHostAddr)

5+import Options.Generic

6+import Codec.Winery

8+data SetKey = SetKey String String

9+ deriving (Generic, Show)

10+ deriving Serialise via WineryVariant SetKey

11+

12+data GetKey = GetKey String

13+ deriving (Generic, Show)

14+ deriving Serialise via WineryVariant GetKey

15+

16+data CommandOptions = Get {name :: String}

17+ | Set {name :: String, value :: String}

18+ deriving (Generic, Show)

19+

20+instance ParseRecord CommandOptions

21+

22+main :: IO ()

23+main = do

24+ opts <- getRecord "SimpleKeyValueClient"

25+ -- connect to the remote server (in this case on the localhost address)

26+ conn <- connect [] localHostAddr 8765

27+ case opts of

28+ Get k -> do

29+ --call the remote function and validate the result

30+ eRet <- call conn (GetKey k)

31+ case eRet of

32+ Left err -> error (show err)

33+ Right (Just val) -> putStrLn val

34+ Right Nothing -> error "no such key"

35+ Set k v -> do

36+ eRet <- call conn (SetKey k v)

37+ case eRet of

38+ Left err -> error (show err)

39+ Right () -> pure ()

examples/SimpleKeyValueServer.hsnew file+30

@@ -0,0 +1,30 @@

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

2+import Network.RPC.Curryer.Server

3+import Control.Concurrent.STM

4+import StmContainers.Map as M

5+import GHC.Generics

6+import Codec.Winery

7+import Data.Functor

9+-- create data structures to represent remote function calls

10+data SetKey = SetKey String String

11+ deriving (Generic, Show)

12+ deriving Serialise via WineryVariant SetKey

13+

14+data GetKey = GetKey String

15+ deriving (Generic, Show)

16+ deriving Serialise via WineryVariant GetKey

17+

18+--call the `serve` function to handle incoming requests

19+main :: IO ()

20+main = do

21+ kvmap <- M.newIO

22+ void $ serve kvRequestHandlers kvmap localHostAddr 8765 Nothing

23+

24+-- setup incoming request handlers to operate on the server's state

25+kvRequestHandlers :: RequestHandlers (M.Map String String)

26+kvRequestHandlers = [ RequestHandler $ \state (SetKey k v) ->

27+ atomically $ M.insert v k (connectionServerState state)

28+ , RequestHandler $ \state (GetKey k) ->

29+ atomically $ M.lookup k (connectionServerState state)

30+ ]

src/Network/RPC/Curryer/Client.hs+107−57

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

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

1module Network.RPC.Curryer.Client where

2module Network.RPC.Curryer.Client where

2import Network.RPC.Curryer.Server

3import Network.RPC.Curryer.Server

3import Network.Socket as Socket

4import Network.Socket as Socket

@@ -10,73 +11,114 @@ import Control.Concurrent.MVar

10import GHC.Conc

11import GHC.Conc

11import Data.Time.Clock

12import Data.Time.Clock

12import System.Timeout

13import System.Timeout

14+import Control.Monad

14−type SyncMap a = STMMap.Map UUID (MVar (Either ConnectionError a), UTCTime)

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

15−-- the request map holds

16−data Connection a = Connection Socket (Async ()) (SyncMap a)

18−connect :: (Serialise msg) =>

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

19− AsyncMessageHandler msg ->

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

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

20− HostAddr ->

33+ HostAddr ->

21− PortNumber ->

34+ PortNumber ->

22− IO (Connection msg)

35+ IO Connection

23−connect notificationCallback hostAddr portNum = do

36+connect asyncHandlers hostAddr portNum = do

24 sock <- TCP.connect hostAddr portNum

37 sock <- TCP.connect hostAddr portNum

25 syncmap <- STMMap.newIO

38 syncmap <- STMMap.newIO

26− asyncThread <- async (clientAsync sock syncmap notificationCallback)

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

40+ sockLock <- newLock sock

27− pure (Connection sock asyncThread syncmap)

41+ pure (Connection {

42+ _conn_sockLock = sockLock,

43+ _conn_asyncThread = asyncThread,

44+ _conn_syncmap = syncmap

45+ })

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

29−close :: Connection a -> IO ()

48+close :: Connection -> IO ()

30−close (Connection sock asyncThread _) = do

49+close conn = do

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

31− Socket.close sock

51+ Socket.close sock

32− cancel asyncThread

52+ cancel (_conn_asyncThread conn)

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

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

35−clientAsync :: Serialise b =>

55+clientAsync ::

36 Socket ->

56 Socket ->

37− SyncMap b ->

57+ SyncMap ->

38− AsyncMessageHandler b ->

58+ ClientAsyncRequestHandlers ->

39 IO ()

59 IO ()

40−clientAsync sock syncmap asyncHandler = do

60+clientAsync sock syncmap asyncHandlers = do

41− -- ping proper thread to continue

61+ lsock <- newLock sock

42− let findRequest requestId =

43− atomically $ do

44− val' <- STMMap.lookup requestId syncmap

45− STMMap.delete requestId syncmap

46− pure val'

47−

48− let responseHandler responseMsg = do

49− putStrLn "client-side message handler"

50− case responseMsg of

51− Response requestId val -> do

52− varval <- findRequest requestId

53− case varval of

54− Nothing -> error "dumped unrequested response"

55− Just (mVar,_) -> putMVar mVar (Right val)

56− AsyncRequest _ asyncMsg ->

57− asyncHandler asyncMsg

58− ResponseExpectedRequest{} -> error "dumped response expected request"

59− ExceptionResponse _ -> error "TODO Exception"

60− TimedOutResponse requestId -> do

61− varval <- findRequest requestId

62− case varval of

63− Nothing -> error "dumped unrequested timeout response"

64− Just (mVar,_) -> putMVar mVar (Left TimeoutError)

65−

66− drainSocketMessages sock responseHandler

62+ drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)

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

65+consumeResponse msgId syncMap val = do

66+ match <- atomically $ do

67+ val' <- STMMap.lookup msgId syncMap

68+ STMMap.delete msgId syncMap

69+ pure val'

70+ case match of

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

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

73+

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.

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

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

69call = callTimeout Nothing

103call = callTimeout Nothing

104

71−-- | Send a request to the remote server and returns a response.

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

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

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

73−callTimeout mTimeout (Connection sock _ mVarMap) msg = do

107+callTimeout mTimeout conn msg = do

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

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

75 -- setup mvar to wait for response

117 -- setup mvar to wait for response

76 responseMVar <- newEmptyMVar

118 responseMVar <- newEmptyMVar

77 now <- getCurrentTime

119 now <- getCurrentTime

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

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

79− sendMessage (ResponseExpectedRequest requestID mTimeout msg) sock

121+ sendEnvelope envelope (_conn_sockLock conn)

80 let timeoutMicroseconds =

122 let timeoutMicroseconds =

81 case mTimeout of

123 case mTimeout of

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

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

@@ -87,11 +129,19 @@ callTimeout mTimeout (Connection sock _ mVarMap) msg = do

87 --timeout

129 --timeout

88 Nothing ->

130 Nothing ->

89 pure (Left TimeoutError)

131 pure (Left TimeoutError)

90− Just res -> pure res

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.

92−asyncCall :: (Serialise request, Serialise response) => Connection response -> request -> IO (Either ConnectionError ())

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

93−asyncCall (Connection sock _ _) msg = do

142+asyncCall conn msg = do

94 requestID <- UUID <$> UUIDBase.nextRandom

143 requestID <- UUID <$> UUIDBase.nextRandom

95− sendMessage (AsyncRequest requestID msg) sock

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

145+ fprint = fingerprint msg

146+ sendEnvelope envelope (_conn_sockLock conn)

96 pure (Right ())

147 pure (Right ())

97−

src/Network/RPC/Curryer/Server.hs+262−158

@@ -1,18 +1,20 @@

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

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

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

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

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

3module Network.RPC.Curryer.Server where

4module Network.RPC.Curryer.Server where

4−import Streamly

5import qualified Streamly.Prelude as S

6import Streamly.Network.Socket

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

8import Network.Socket as Socket

9import Network.Socket.ByteString as Socket

10−import Streamly.Internal.Data.Parser.ParserD as PD

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

11import Codec.Winery

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

13import Codec.Winery.Class (mkExtractor)

14import GHC.Generics

15+import GHC.Fingerprint

16+import Data.Typeable

15−import Control.Concurrent.MVar (MVar)

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

16import Control.Exception

18import Control.Exception

17import Data.Function ((&))

19import Data.Function ((&))

18import Data.Word

20import Data.Word

@@ -21,57 +23,103 @@ import qualified Data.ByteString.Lazy as BSL

21import qualified Data.ByteString.FastBuilder as BB

23import qualified Data.ByteString.FastBuilder as BB

22import Streamly.Data.Fold as FL

24import Streamly.Data.Fold as FL

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

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

24−import Data.Foldable

25−import Data.Bits

26import qualified Data.Binary as B

27import qualified Data.UUID as UUIDBase

28import qualified Data.UUID.V4 as UUIDBase

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

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

31import Control.Monad

29import Control.Monad

30+import Data.Functor

31+

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

33+--import Control.Monad

32import Data.Hashable

34import Data.Hashable

33import System.Timeout

35import System.Timeout

36+import qualified Network.ByteOrder as BO

37+

35-- for toArrayS conversion

39-- for toArrayS conversion

36import qualified Data.ByteString.Internal as BSI

40import qualified Data.ByteString.Internal as BSI

37−import Foreign.ForeignPtr (plusForeignPtr)

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

38import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

42import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

39−import GHC.Ptr (plusPtr)

43+import GHC.Ptr (minusPtr)

45+-- define CURRYER_SHOW_BYTES 1

46+#if CURRYER_SHOW_BYTES == 1

41import Debug.Trace

47import 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

43−data ClientHelloMessage = ClientHelloMessage Int Int

73+type BinaryMessage = BS.ByteString

44− deriving Generic

45− deriving Serialise via WineryVariant ClientHelloMessage

46−

47−data ServerHelloMessage = ServerHelloMessage { serverServices :: [String],

48− serverInfo :: String

49− }

50− deriving Generic

51− deriving Serialise via WineryRecord ServerHelloMessage

52−

53−--request-response token

54−--client-side timeout

55−--server-to-client and client-to-server async request (no response requested)

56−

57−type Timeout = Int

58−

59−data Message a = Response UUID a

60− | AsyncRequest UUID a

61− | ResponseExpectedRequest UUID (Maybe Timeout) a

62− | TimedOutResponse UUID

63− -- | Services

64− | ExceptionResponse String

65− deriving (Generic, Show)

66− deriving Serialise via WineryVariant (Message a)

67−

68−-- | The response type

69−data HandlerResponse a = NoResponse

70− | HandlerResponse a

71− | HandlerException a

72− deriving Generic

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

75--avoid orphan instance

123--avoid orphan instance

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

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

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

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

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

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

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

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

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

140

93−

141+-- | Errors from remote calls.

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

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

95 | TimeoutError

143 | TimeoutError

144+ | ExceptionError String

96 deriving (Generic, Show, Eq)

145 deriving (Generic, Show, Eq)

97 deriving Serialise via WineryVariant ConnectionError

146 deriving Serialise via WineryVariant ConnectionError

147

148+data TimeoutException = TimeoutException

149+ deriving Show

150+

151+instance Exception TimeoutException

152+

99type HostAddr = (Word8, Word8, Word8, Word8)

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

100

154

101allHostAddrs,localHostAddr :: HostAddr

155allHostAddrs,localHostAddr :: HostAddr

@@ -103,136 +157,186 @@ allHostAddrs = (0,0,0,0)

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

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

104

158

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

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

106−messageBoundaryP :: Parser IO Word8 BS.ByteString

160+envelopeP :: Parser IO Word8 Envelope

107−messageBoundaryP = do

161+envelopeP = do

108− let x = FL.toList

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

109− w4x8 <- PD.take 4 x

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

110− traceShowM ("w4x8"::String, w4x8)

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

111− let c = fromIntegral (fromOctets w4x8)

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

112− traceShowM ("c"::String, c)

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

113− vals <- PD.take c x

167+ lenPrefixedByteStringP = do

114− let bytes = BS.pack vals

168+ c <- fromIntegral <$> word32P

115− traceShowM ("parsedBytes"::String, c, BS.length bytes, bytes)

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

116− pure bytes

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

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

117

219

118type NewConnectionHandler msg = IO (Maybe msg)

220type NewConnectionHandler msg = IO (Maybe msg)

119

221

120−type NewMessageHandler msg resp = msg -> IO (HandlerResponse resp)

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

121−

223+

122−serve :: (Show msg, Serialise msg, Serialise resp) =>

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.

123− NewConnectionHandler resp ->

225+serve ::

124− NewMessageHandler msg resp ->

226+ RequestHandlers s->

227+ s ->

125 HostAddr ->

228 HostAddr ->

126 PortNumber ->

229 PortNumber ->

127 Maybe (MVar SockAddr) ->

230 Maybe (MVar SockAddr) ->

128 IO Bool

231 IO Bool

129−serve connhandler userMsgHandler hostaddr port mSockLock = do

232+serve userMsgHandlers serverState hostaddr port mSockLock = do

233+ let

130− let handleSock sock = do

234+ handleSock sock = do

131− putStrLn "handleSock"

235+ lockingSocket <- newLock sock

132− -- allow the server to send an async welcome message to the new client, if necessary

236+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)

133− mResp <- connhandler

237+

134− case mResp of

135− Nothing -> pure ()

136− Just resp -> do

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

138− sendMessage (AsyncRequest requestID resp) sock

139− drainSocketMessages sock (serverMessageHandler sock userMsgHandler)

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

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

141 pure True

239 pure True

142

240

143−serverMessageHandler :: forall req resp. (Show req,

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

144− Serialise req,

242+openEnvelope (Envelope eprint _ _ bytes) =

145− Serialise resp)

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+

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.

260+serverEnvelopeHandler ::

146− => Socket

261+ Locking Socket

147− -> NewMessageHandler req resp

262+ -> RequestHandlers s

148− -> Message req

263+ -> s

264+ -> Envelope

149 -> IO ()

265 -> IO ()

150−serverMessageHandler sock requestMessageHandler msg = do

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

151− putStrLn $ "GOT MSG " ++ show msg

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

152− let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))

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

153− runTimeout mTimeout m = case mTimeout of

272+ runTimeout m =

154− Nothing -> Just <$> m

273+ if timeoutms == 0 then

155− Just timeoutMicroseconds -> timeout timeoutMicroseconds m

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

156− case msg of

275+ else

157− Response{} -> putStrLn "client sent response"

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

158− ResponseExpectedRequest requestID mTimeout val -> do

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

159− putStrLn "server received ResponseExpectedRequest"

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

160− resp <- runTimeout mTimeout (requestMessageHandler val)

279+ timeoutExcHandler _ = pure Nothing

161− case resp of

280+

162− Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sock

281+ sState = ConnectionState {

163− Just NoResponse -> error "attempt to return non-response to expected response message"

282+ connectionServerState = serverState,

164− Just (HandlerException _) -> error "TODO HandlerException"

283+ connectionSocket = sockLock

165− Nothing ->

284+ }

166− sendMessage (TimedOutResponse @(Message resp) requestID) sock

285+

167− AsyncRequest _ val -> do

286+ firstMatcher (RequestHandler msghandler) Nothing =

168− putStrLn "AsyncReq"

287+ case matchEnvelope envelope msghandler of

169− void $ requestMessageHandler val

288+ Nothing -> pure Nothing

170− --no response necessary

289+ Just (dispatchf, decoded) -> do

171− ExceptionResponse{} -> putStrLn "client sent exception response"

290+ --TODO add exception handling

172− TimedOutResponse{} -> putStrLn "client sent timed out response"

291+ mResponse <- runTimeout (dispatchf sState decoded)

173−

292+ let envelopeResponse =

174−--add callback to allow for responses via socket

293+ case mResponse of

175−type MessageHandler a = Message a -> IO ()

294+ Just response ->

176−

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

177−type AsyncMessageHandler a = a -> IO ()

296+ Nothing ->

178−

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

179−

298+ sendEnvelope envelopeResponse sockLock

180−

299+ pure (Just ())

181−drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()

300+ firstMatcher (AsyncRequestHandler msghandler) Nothing =

182−drainSocketMessages sock msgHandler = do

301+ case matchEnvelope envelope msghandler of

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

302+ Nothing -> pure Nothing

184− handler bs = do

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

185− case deserialise bs of

308+ case eExc of

186− Left err ->

309+ Left exc ->

187− print err

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

311+ sendEnvelope env sockLock

188− Right val ->

312+ Right () -> pure ()

189− msgHandler val --add response function

190− S.drain $ serially $ S.parseManyD messageBoundaryP sockStream & S.mapM handler

191−

192−fromOctets :: [Word8] -> Word32

193−fromOctets = foldl' accum 0

194− where

195− accum a o = (a `shiftL` 8) .|. fromIntegral o

196

313

197−octets :: Word32 -> [Word8]

198−octets w =

199− [ fromIntegral (w `shiftR` 24)

200− , fromIntegral (w `shiftR` 16)

201− , fromIntegral (w `shiftR` 8)

202− , fromIntegral w

203− ]

204

314

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

315+type EnvelopeHandler = Envelope -> IO ()

206−sendMessage :: Serialise a => Message a -> Socket -> IO ()

207−sendMessage msg socket' = do

208− let -- byteArray = toArraySlow fullbytes

209− --dbgByteString = BS.pack (SA.toList byteArray)

210− msgbytes = serialise msg

211− fullbytes = lenbytes <> msgbytes

212− len = BS.length msgbytes

213− lenbytes = BS.pack (octets (fromIntegral len))

214− --when (dbgByteString /= msgbytes) (error "mismatch!")

215−

216− byteCount <- Socket.send socket' fullbytes

217− when (byteCount /= BS.length fullbytes) (error "bytes sent mismatch")

218− traceShowM ("sent bytes:" :: String, byteCount, fullbytes)

219

316

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

318+drainSocketMessages sock envelopeHandler = do

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

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

220

321

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

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

324+sendEnvelope envelope sockLock = do

325+ let envelopebytes = encodeEnvelope envelope

326+ fullbytes = envelopebytes

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

328+ withLock sockLock $ \socket' ->

329+ Socket.sendAll socket' fullbytes

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

331+ traceBytes "sendEnvelope" fullbytes

221

332

222−--from streamly-bytestring

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

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

334+fingerprint = typeRepFingerprint . typeOf

224−-- 'ByteString' and wraps it with 'Array' constructors and hence the operation

225−-- is performed in constant time.

226−{-# INLINE toArrayS #-}

227−toArrayS :: BS.ByteString -> SA.Array Word8

228−--slow path

229−--toArrayS = SA.fromList . BS.unpack

230−toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr

231− where

232− nfp = fp `plusForeignPtr` off

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

234

335

235−toArraySlow :: BS.ByteString -> SA.Array Word8

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

236−toArraySlow bs = traceShow ("arr" :: String, SA.length arr) arr

337+fromArray Arr.Array {..}

338+ | aLen == 0 = mempty

339+ | otherwise = BSI.PS aStart 0 aLen

237 where

340 where

238− arr = SA.fromList (BS.unpack bs)

341+ aStartPtr = unsafeForeignPtrToPtr aStart

342+ aLen = aEnd `minusPtr` aStartPtr

test/Curryer/Test/Basic.hs+160−67

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

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

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

2module Curryer.Test.Basic where

3import Test.Tasty

4import Test.Tasty.HUnit

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

9import Control.Concurrent.Async

10import Control.Monad

11import Control.Concurrent

12+import Control.Concurrent.STM

13+import Control.Exception

14+import Data.List

13import Network.RPC.Curryer.Server

16import Network.RPC.Curryer.Server

14import Network.RPC.Curryer.Client

17import Network.RPC.Curryer.Client

19+-- TODO: add test for nested calls

20+

16testTree :: TestTree

21testTree :: TestTree

17−testTree = testGroup "basic" [testCase "simple" testSimpleCall,

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

18− testCase "client async" testAsyncServerCall,

23+ ,testCase "client async" testAsyncServerCall

19− testCase "server async" testAsyncClientCall,

24+ ,testCase "server async" testAsyncClientCall

20− testCase "client sync timeout" testSyncClientCallTimeout

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

21 ]

30 ]

23−--client-to-server request

32+

24−data TestRequest = AddTwoNumbersReq Int Int

33+data AddTwoNumbersReq = AddTwoNumbersReq Int Int

25− | TestAsyncReq String

26− | TestCallMeBackReq String

27− | DelayMicrosecondsReq Int

28 deriving (Generic, Show)

34 deriving (Generic, Show)

29− deriving Serialise via WineryVariant TestRequest

35+ deriving Serialise via WineryVariant AddTwoNumbersReq

31−--server responds to client

37+data TestCallMeBackReq = TestCallMeBackReq String

32−data TestResponse = AddTwoNumbersResp Int

33− | AsyncHello String

34− | TestCallMeBackResp String

35− | DelayMicrosecondsResp

36 deriving (Generic, Show, Eq)

38 deriving (Generic, Show, Eq)

37− deriving Serialise via WineryVariant TestResponse

39+ deriving Serialise via WineryVariant TestCallMeBackReq

38−

40+

39−testServerMessageHandler :: Maybe (MVar String) -> TestRequest -> IO (HandlerResponse TestResponse)

41+data TestAsyncReq = TestAsyncReq String

40−testServerMessageHandler mAsyncMVar msg = do

42+ deriving (Generic, Show)

41− --print msg

43+ deriving Serialise via WineryVariant TestAsyncReq

42− case msg of

44+

43− --round-trip request to add two Ints

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 =

44− AddTwoNumbersReq x y -> pure (HandlerResponse (AddTwoNumbersResp (x+y)))

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

45− --respond to a request for an async callback

46− TestCallMeBackReq s -> do

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

47− case mAsyncMVar of

74+ case mAsyncMVar of

48− Nothing -> pure ()

75+ Nothing -> pure ()

49− Just mvar -> putMVar mvar s

76+ Just mvar -> putMVar mvar s

50− pure NoResponse

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

51− TestAsyncReq _ -> 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

52− DelayMicrosecondsReq ms -> do

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

53 threadDelay ms

85 threadDelay ms

54− pure (HandlerResponse DelayMicrosecondsResp)

86+ pure ()

87+ , RequestHandler $ \_ ThrowServerSideExceptionReq -> do

88+ _ <- error "test server exception"

89+ pure ()

90+ ]

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

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

57testSimpleCall :: Assertion

93testSimpleCall :: Assertion

58testSimpleCall = do

94testSimpleCall = do

59 readyVar <- newEmptyMVar

95 readyVar <- newEmptyMVar

61− server <- async (serve (pure Nothing) (testServerMessageHandler Nothing) localHostAddr 0 (Just readyVar))

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

62 --wait for server to be ready

98 --wait for server to be ready

63 (SockAddrInet port _) <- takeMVar readyVar

99 (SockAddrInet port _) <- takeMVar readyVar

64− let clientHandler :: AsyncMessageHandler TestResponse

65− clientHandler _ = error "async handler called"

66− mkConn :: IO (Connection TestResponse)

67− mkConn = connect clientHandler localHostAddr port

100+ conn <- connect [] localHostAddr port

68− conn <- mkConn

69− let c :: IO (Either ConnectionError TestResponse)

70− c = call conn (AddTwoNumbersReq 1 1)

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

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

72− x <- c

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

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

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

74 close conn

104 close conn

75 cancel server

105 cancel server

106

107+

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

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

78testAsyncServerCall :: Assertion

109testAsyncServerCall :: Assertion

79testAsyncServerCall = do

110testAsyncServerCall = do

80 portReadyVar <- newEmptyMVar

111 portReadyVar <- newEmptyMVar

81 receivedAsyncMessageVar <- newEmptyMVar

112 receivedAsyncMessageVar <- newEmptyMVar

82−

113+ let clientAsyncHandlers =

83− let testServerNewConnHandler :: NewConnectionHandler TestResponse

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

84− testServerNewConnHandler = do

115+ putMVar receivedAsyncMessageVar s)]

85− pure (Just (AsyncHello "welcome"))

86−

87− server <- async (serve testServerNewConnHandler (testServerMessageHandler Nothing) localHostAddr 0 (Just portReadyVar))

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

88 (SockAddrInet port _) <- takeMVar portReadyVar

117 (SockAddrInet port _) <- takeMVar portReadyVar

89− let clientHandler :: AsyncMessageHandler TestResponse

90− clientHandler msg = do

91− case msg of

92− AsyncHello str -> putMVar receivedAsyncMessageVar str

93− x -> error ("unexpected message " <> show x)

94− conn <- connect @TestResponse clientHandler localHostAddr port

118+ conn <- connect clientAsyncHandlers localHostAddr port

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

95 asyncMessage <- takeMVar receivedAsyncMessageVar

120 asyncMessage <- takeMVar receivedAsyncMessageVar

96 assertEqual "async message" "welcome" asyncMessage

121 assertEqual "async message" "welcome" asyncMessage

97 close conn

122 close conn

98 cancel server

123 cancel server

124

125+

126+emptyServerState :: ()

127+emptyServerState = ()

128+

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

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

101testAsyncClientCall :: Assertion

130testAsyncClientCall :: Assertion

102testAsyncClientCall = do

131testAsyncClientCall = do

103 portReadyVar <- newEmptyMVar

132 portReadyVar <- newEmptyMVar

104 receivedAsyncMessageVar <- newEmptyMVar

133 receivedAsyncMessageVar <- newEmptyMVar

105

134

106− server <- async (serve (pure Nothing) (testServerMessageHandler (Just receivedAsyncMessageVar)) localHostAddr 0 (Just portReadyVar))

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

107 (SockAddrInet port _) <- takeMVar portReadyVar

136 (SockAddrInet port _) <- takeMVar portReadyVar

108−

137+

109− conn <- connect @TestResponse (\_ -> pure ()) localHostAddr port

138+ conn <- connect [] localHostAddr port

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

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

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

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

112 asyncMessage <- takeMVar receivedAsyncMessageVar

141 asyncMessage <- takeMVar receivedAsyncMessageVar

@@ -118,18 +147,82 @@ testSyncClientCallTimeout :: Assertion

118testSyncClientCallTimeout = do

147testSyncClientCallTimeout = do

119 readyVar <- newEmptyMVar

148 readyVar <- newEmptyMVar

120

149

121− server <- async (serve (pure Nothing) (testServerMessageHandler Nothing) localHostAddr 0 (Just readyVar))

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

122 --wait for server to be ready

151 --wait for server to be ready

123 (SockAddrInet port _) <- takeMVar readyVar

152 (SockAddrInet port _) <- takeMVar readyVar

124− let clientHandler :: AsyncMessageHandler TestResponse

125− clientHandler _ = error "async handler called"

126− mkConn :: IO (Connection TestResponse)

127− mkConn = connect clientHandler localHostAddr port

153+ conn <- connect [] localHostAddr port

128− conn <- mkConn

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

129− let c :: IO (Either ConnectionError TestResponse)

130− c = callTimeout (Just 1000) conn (DelayMicrosecondsReq 500)

131− x <- c

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

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

133 close conn

156 close conn

134 cancel server

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

135

228