git.iko.soy/curryer
on master help sign in
swap
74ecaac25 ahead, 0 behind
Changelog.markdownnew file+4
since splitsnapshots
unifiedsplit
@@ -0,0 +1,4 @@
1+# v0.1 (2020-12-27)
2+
3+* initial release to support [Project:M36](https://github.com/agentm/project-m36)
4+
LICENSEnew file+25
@@ -0,0 +1,25 @@
1+This is free and unencumbered software released into the public domain.
2+
3+Anyone is free to copy, modify, publish, use, compile, sell, or
4+distribute this software, either in source code form or as a compiled
5+binary, for any purpose, commercial or non-commercial, and by any
6+means.
7+
8+In jurisdictions that recognize copyright laws, the author or authors
9+of this software dedicate any and all copyright interest in the
10+software to the public domain. We make this dedication for the benefit
11+of the public at large and to the detriment of our heirs and
12+successors. We intend this dedication to be an overt act of
13+relinquishment in perpetuity of all present and future rights to this
14+software under copyright law.
15+
16+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
17+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
18+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
19+IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
20+OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
21+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
22+OTHER DEALINGS IN THE SOFTWARE.
23+
24+For more information, please refer to <http://unlicense.org>
25+
README.markdownnew file+78
@@ -0,0 +1,78 @@
1+# Curryer - Fast Haskell-to-Haskell RPC
2+
3+Curryer (pun intended) is a fast, Haskell-exclusive RPC (remote procedure call) library. By using the latest Haskell serialization and streaming libraries, curryer aims to be the fastest and easiest means of communicating between Haskell-based processes.
4+
5+Curryer is inspired by the now unmaintained [distributed-process](https://hackage.haskell.org/package/distributed-process) library, but is lighter-weight and uses a higher-performance serialization package.
6+
7+## Features
8+
9+* blocking and non-blocking remote function calls
10+* asynchronous server-to-client callbacks (for server-initiated notifications)
11+* timeouts
12+* leverages [winery](https://hackage.haskell.org/package/winery) for high-performance serialization
13+
14+## Requirements
15+
16+* GHC 8.6+
17+
18+## Code Example
19+
20+[Server](https://github.com/agentm/curryer/examples/SimpleKeyValueServer.hs):
21+
22+```haskell
23+data SetKey = SetKey String String
24+ deriving (Generic, Show)
25+ deriving Serialise via WineryVariant SetKey
26+
27+data GetKey = GetKey String
28+ deriving (Generic, Show)
29+ deriving Serialise via WineryVariant GetKey
30+
31+main :: IO ()
32+main = do
33+ kvmap <- M.newIO
34+ void $ serve kvRequestHandlers kvmap localHostAddr 8765 Nothing
35+
36+kvRequestHandlers :: RequestHandlers (M.Map String String)
37+kvRequestHandlers = [ RequestHandler $ \state (SetKey k v) ->
38+ atomically $ M.insert v k (connectionServerState state)
39+ , RequestHandler $ \state (GetKey k) ->
40+ atomically $ M.lookup k (connectionServerState state)
41+ ]
42+```
43+
44+[Client](https://github.com/agentm/curryer/examples/SimpleKeyValueClient.hs):
45+
46+```haskell
47+data SetKey = SetKey String String
48+ deriving (Generic, Show)
49+ deriving Serialise via WineryVariant SetKey
50+
51+data GetKey = GetKey String
52+ deriving (Generic, Show)
53+ deriving Serialise via WineryVariant GetKey
54+
55+data CommandOptions = Get {name :: String}
56+ | Set {name :: String, value :: String}
57+ deriving (Generic, Show)
58+
59+instance ParseRecord CommandOptions
60+
61+main :: IO ()
62+main = do
63+ opts <- getRecord "SimpleKeyValueClient"
64+ conn <- connect [] localHostAddr 8765
65+ case opts of
66+ Get k -> do
67+ eRet <- call conn (GetKey k)
68+ case eRet of
69+ Left err -> error (show err)
70+ Right (Just val) -> putStrLn val
71+ Right Nothing -> error "no such key"
72+ Set k v -> do
73+ eRet <- call conn (SetKey k v)
74+ case eRet of
75+ Left err -> error (show err)
76+ Right () -> pure ()
77+
78+```
bench.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
9import Control.Concurrent.Async
10import Control.Concurrent
10import Control.Concurrent
11import Control.Monad
11import Control.Monad
12+import qualified Data.ByteString as BS
12
13
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
21
22
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
24
40
25− putStrLn "benchmark complete"
26 close clientConn
41 close clientConn
27 cancel server
42 cancel server
28
43
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
32
51
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
36
55
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+ ]
41
64
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
48
71
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
3source-repository-package
4 type: git
4 type: git
5 location: https://github.com/composewell/streamly.git
5 location: https://github.com/composewell/streamly.git
6− tag: e218fb9d8f4263fd69d32a2ff318bac0e179c642
6+ tag: 4e48b52e8709390a47e99c107ebed2d1c9420076
curryer.cabal → curryer-rpc.cabalrenamed+24−6
@@ -1,4 +1,4 @@
1−Name: curryer
1+Name: curryer-rpc
2Version: 0.1
2Version: 0.1
3License: PublicDomain
3License: PublicDomain
4Build-Type: Simple
4Build-Type: Simple
@@ -11,14 +11,14 @@ Maintainer: agentm@themactionfaction.com
11Cabal-Version: >= 1.10
11Cabal-Version: >= 1.10
12Synopsis: Fast, Haskell RPC
12Synopsis: Fast, Haskell RPC
13Description: Haskell-to-Haskell RPC using Winery serialization.
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
15
15
16Source-Repository head
16Source-Repository head
17 Type: git
17 Type: git
18 location: https://github.com/agentm/curryer
18 location: https://github.com/agentm/curryer
19
19
20Library
20Library
21− Build-Depends: base >= 4.8
21+ Build-Depends: base >= 4.12 && < 4.15
22 , winery
22 , winery
23 , bytestring
23 , bytestring
24 , streamly >= 0.7.2
24 , streamly >= 0.7.2
@@ -31,6 +31,9 @@ Library
31 , containers
31 , containers
32 , stm-containers
32 , stm-containers
33 , hashable
33 , hashable
34+ , time
35+ , network-byte-order
36+ , stm
34 Hs-Source-Dirs: ./src
37 Hs-Source-Dirs: ./src
35 Default-Language: Haskell2010
38 Default-Language: Haskell2010
36 ghc-options: -Wall -fwarn-unused-binds -fwarn-unused-imports
39 ghc-options: -Wall -fwarn-unused-binds -fwarn-unused-imports
@@ -48,18 +51,33 @@ Test-Suite test
48 build-depends: tasty
51 build-depends: tasty
49 , tasty-hunit
52 , tasty-hunit
50 , base
53 , base
51− , curryer
54+ , curryer-rpc
52 , winery
55 , winery
53 , network
56 , network
54 , async
57 , async
58+ , stm
55 other-modules: Curryer.Test.Basic
59 other-modules: Curryer.Test.Basic
56
60
57Benchmark perf
61Benchmark perf
58 Default-Language: Haskell2010
62 Default-Language: Haskell2010
59 Default-Extensions: OverloadedStrings
63 Default-Extensions: OverloadedStrings
60− Build-Depends: base, criterion, curryer, network, winery, async
64+ Build-Depends: base, criterion, curryer-rpc, network, winery, async, bytestring
65+
61 Main-Is: Basic.hs
66 Main-Is: Basic.hs
62 Type: exitcode-stdio-1.0
67 Type: exitcode-stdio-1.0
63 GHC-Options: -Wall -threaded -rtsopts
68 GHC-Options: -Wall -threaded -rtsopts
69+ -- -fprof-auto
64 HS-Source-Dirs: ./bench
70 HS-Source-Dirs: ./bench
65−
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
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.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
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+122−46
@@ -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
@@ -8,64 +9,139 @@ import qualified Data.UUID.V4 as UUIDBase
8import qualified StmContainers.Map as STMMap
9import qualified StmContainers.Map as STMMap
9import Control.Concurrent.MVar
10import Control.Concurrent.MVar
10import GHC.Conc
11import GHC.Conc
12+import Data.Time.Clock
13+import System.Timeout
14+import Control.Monad
11
15
12−type SyncMap a = STMMap.Map UUID (MVar a)
16+type SyncMap = STMMap.Map UUID (MVar (Either ConnectionError BinaryMessage), UTCTime)
13−-- the request map holds
14−data Connection a = Connection Socket (Async ()) (SyncMap a)
15
17
16−connect :: (Serialise msg) =>
18+-- | Represents a remote connection to server.
17− 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 ->
18− HostAddr ->
33+ HostAddr ->
19− PortNumber ->
34+ PortNumber ->
20− IO (Connection msg)
35+ IO Connection
21−connect notificationCallback hostAddr portNum = do
36+connect asyncHandlers hostAddr portNum = do
22 sock <- TCP.connect hostAddr portNum
37 sock <- TCP.connect hostAddr portNum
23 syncmap <- STMMap.newIO
38 syncmap <- STMMap.newIO
24− asyncThread <- async (clientAsync sock syncmap notificationCallback)
39+ asyncThread <- async (clientAsync sock syncmap asyncHandlers)
40+ sockLock <- newLock sock
25− pure (Connection sock asyncThread syncmap)
41+ pure (Connection {
42+ _conn_sockLock = sockLock,
43+ _conn_asyncThread = asyncThread,
44+ _conn_syncmap = syncmap
45+ })
26
46
47+-- | Close the connection and release all connection resources.
27−close :: Connection a -> IO ()
48+close :: Connection -> IO ()
28−close (Connection sock asyncThread _) = do
49+close conn = do
50+ withLock (_conn_sockLock conn) $ \sock ->
29− Socket.close sock
51+ Socket.close sock
30− cancel asyncThread
52+ cancel (_conn_asyncThread conn)
31
53
32−-- 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
33−clientAsync :: Serialise b =>
55+clientAsync ::
34 Socket ->
56 Socket ->
35− SyncMap b ->
57+ SyncMap ->
36− AsyncMessageHandler b ->
58+ ClientAsyncRequestHandlers ->
37 IO ()
59 IO ()
38−clientAsync sock syncmap asyncHandler = do
60+clientAsync sock syncmap asyncHandlers = do
39− -- ping proper thread to continue
61+ lsock <- newLock sock
40− let responseHandler responseMsg = do
41− putStrLn "clientAsync handler"
42− case responseMsg of
43− Response requestId val -> do
44− mMVar <- atomically $ STMMap.lookup requestId syncmap
45− case mMVar of
46− Nothing -> error "dumped unrequested response"
47− Just mVar -> putMVar mVar val
48− AsyncRequest _ asyncMsg ->
49− asyncHandler asyncMsg
50− ResponseExpectedRequest _ _ -> error "dumped response expected request"
51− ExceptionResponse _ -> error "TODO Exception"
52− drainSocketMessages sock responseHandler
62+ drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)
53
63
64+consumeResponse :: UUID -> STMMap.Map UUID (MVar a, b) -> a -> IO ()
65+consumeResponse msgId syncMap val = do
66+ match <- atomically $ do
67+ val' <- STMMap.lookup msgId syncMap
68+ STMMap.delete msgId syncMap
69+ pure val'
70+ case match of
71+ Nothing -> pure () -- drop message
72+ Just (mVar, _) -> putMVar mVar val
73+
74+-- | handles envelope responses from server- timeout from ths server is ignored, but perhaps that's proper for trusted servers- the server expects the client to process all async requests
75+clientEnvelopeHandler ::
76+ ClientAsyncRequestHandlers
77+ -> Locking Socket
78+ -> SyncMap
79+ -> Envelope
80+ -> IO ()
81+clientEnvelopeHandler handlers _ _ envelope@(Envelope _ (RequestMessage _) _ _) = do
82+ --should this run off on another green thread?
83+ let firstMatcher Nothing (ClientAsyncRequestHandler (dispatchf :: a -> IO ())) = do
84+ case openEnvelope envelope of
85+ Nothing -> pure Nothing
86+ Just decoded -> do
87+ dispatchf decoded
88+ pure (Just ())
89+ firstMatcher acc _ = pure acc
90+ foldM_ firstMatcher Nothing handlers
91+clientEnvelopeHandler _ _ syncMap (Envelope _ ResponseMessage msgId binaryMessage) =
92+ consumeResponse msgId syncMap (Right binaryMessage)
93+clientEnvelopeHandler _ _ syncMap (Envelope _ TimeoutResponseMessage msgId _) =
94+ consumeResponse msgId syncMap (Left TimeoutError)
95+clientEnvelopeHandler _ _ syncMap (Envelope _ ExceptionResponseMessage msgId excPayload) =
96+ case deserialise excPayload of
97+ Left err -> error ("failed to deserialise exception string" <> show err)
98+ Right excStr ->
99+ consumeResponse msgId syncMap (Left (ExceptionError excStr))
100+
101+-- | Basic remote function call via data type and return value.
102+call :: (Serialise request, Serialise response) => Connection -> request -> IO (Either ConnectionError response)
103+call = callTimeout Nothing
104+
54−-- | 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.
55−call :: (Serialise request, Serialise response) => Connection response -> request -> IO (Either ConnectionError response)
106+callTimeout :: (Serialise request, Serialise response) => Maybe Int -> Connection -> request -> IO (Either ConnectionError response)
56−call (Connection sock _ mVarMap) msg = do
107+callTimeout mTimeout conn msg = do
57− 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
58 -- setup mvar to wait for response
117 -- setup mvar to wait for response
59 responseMVar <- newEmptyMVar
118 responseMVar <- newEmptyMVar
119+ now <- getCurrentTime
60− atomically $ STMMap.insert responseMVar requestID mVarMap
120+ atomically $ STMMap.insert (responseMVar, now) requestID mVarMap
61− sendMessage (ResponseExpectedRequest requestID msg) sock
121+ sendEnvelope envelope (_conn_sockLock conn)
122+ let timeoutMicroseconds =
123+ case mTimeout of
124+ Just timeout' -> timeout' + 100 --add 100 ms to account for unknown network latency
125+ Nothing -> -1
62− response <- takeMVar responseMVar
126+ mResponse <- timeout timeoutMicroseconds (takeMVar responseMVar)
63 atomically $ STMMap.delete requestID mVarMap
127 atomically $ STMMap.delete requestID mVarMap
64− pure (Right response)
128+ case mResponse of
129+ --timeout
130+ Nothing ->
131+ pure (Left TimeoutError)
132+ Just (Left exc) ->
133+ pure (Left exc)
134+ Just (Right binmsg) ->
135+ --TODO use decoder instead
136+ case deserialise binmsg of
137+ Left err -> error ("deserialise client error " <> show err)
138+ Right m -> pure (Right m)
65
139
140+-- | Call a remote function but do not expect a response from the server.
66−asyncCall :: (Serialise request, Serialise response) => Connection response -> request -> IO (Either ConnectionError ())
141+asyncCall :: Serialise request => Connection -> request -> IO (Either ConnectionError ())
67−asyncCall (Connection sock _ _) msg = do
142+asyncCall conn msg = do
68 requestID <- UUID <$> UUIDBase.nextRandom
143 requestID <- UUID <$> UUIDBase.nextRandom
69− 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)
70 pure (Right ())
147 pure (Right ())
71−
src/Network/RPC/Curryer/Server.hs+267−133
@@ -1,17 +1,20 @@
1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving #-}
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
5import qualified Streamly.Prelude as S
6import Streamly.Network.Socket
6import Streamly.Network.Socket
7−import Streamly.Internal.Network.Socket (handleWithM, writeChunk)
7+import Streamly.Internal.Network.Socket (handleWithM)
8import Network.Socket as Socket
8import Network.Socket as Socket
9+import Network.Socket.ByteString as Socket
9−import Streamly.Internal.Data.Parser.ParserD as PD
10+import Streamly.Internal.Data.Parser as P hiding (concatMap)
10import Codec.Winery
11import Codec.Winery
11import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)
12import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)
12import Codec.Winery.Class (mkExtractor)
13import Codec.Winery.Class (mkExtractor)
13import GHC.Generics
14import GHC.Generics
15+import GHC.Fingerprint
16+import Data.Typeable
14−import Control.Concurrent.MVar (MVar)
17+import Control.Concurrent.MVar (MVar, newMVar, withMVar)
15import Control.Exception
18import Control.Exception
16import Data.Function ((&))
19import Data.Function ((&))
17import Data.Word
20import Data.Word
@@ -20,53 +23,103 @@ import qualified Data.ByteString.Lazy as BSL
20import qualified Data.ByteString.FastBuilder as BB
23import qualified Data.ByteString.FastBuilder as BB
21import Streamly.Data.Fold as FL
24import Streamly.Data.Fold as FL
22import qualified Streamly.Internal.Data.Stream.IsStream as S
25import qualified Streamly.Internal.Data.Stream.IsStream as S
23−import Data.Foldable
24−import Data.Bits
25import qualified Data.Binary as B
26import qualified Data.Binary as B
26import qualified Data.UUID as UUIDBase
27import qualified Data.UUID as UUIDBase
27import qualified Data.UUID.V4 as UUIDBase
28import qualified Data.UUID.V4 as UUIDBase
28−import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA
29−import qualified Network.RPC.Curryer.StreamlyAdditions as SA
30import Control.Monad
29import Control.Monad
30+import Data.Functor
31+
32+import qualified Network.RPC.Curryer.StreamlyAdditions as SA
33+--import Control.Monad
31import Data.Hashable
34import Data.Hashable
35+import System.Timeout
36+import qualified Network.ByteOrder as BO
37+
32
38
33-- for toArrayS conversion
39-- for toArrayS conversion
34import qualified Data.ByteString.Internal as BSI
40import qualified Data.ByteString.Internal as BSI
35−import Foreign.ForeignPtr (plusForeignPtr)
41+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as Arr
36import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
42import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
37−import GHC.Ptr (plusPtr)
43+import GHC.Ptr (minusPtr)
38
44
45+-- define CURRYER_SHOW_BYTES 1
46+#if CURRYER_SHOW_BYTES == 1
39import 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
40
72
41−data ClientHelloMessage = ClientHelloMessage Int Int
73+type BinaryMessage = BS.ByteString
42− deriving Generic
43− deriving Serialise via WineryVariant ClientHelloMessage
44−
45−data ServerHelloMessage = ServerHelloMessage { serverServices :: [String],
46− serverInfo :: String
47− }
48− deriving Generic
49− deriving Serialise via WineryRecord ServerHelloMessage
50−
51−--request-response token
52−--client-side timeout
53−--server-to-client and client-to-server async request (no response requested)
54−
55−data Message a = Response UUID a
56− | AsyncRequest UUID a
57− | ResponseExpectedRequest UUID a
58− -- | Services
59− | ExceptionResponse String
60− deriving (Generic, Show)
61− deriving Serialise via WineryVariant (Message a)
62−
63−-- | The response type
64−data HandlerResponse a = NoResponse
65− | HandlerResponse a
66− | HandlerException a
67− deriving Generic
68− deriving Serialise via WineryVariant (HandlerResponse a)
69
74
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+
70--avoid orphan instance
123--avoid orphan instance
71newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
124newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
72 deriving (Show, Eq, B.Binary, Hashable)
125 deriving (Show, Eq, B.Binary, Hashable)
@@ -85,12 +138,18 @@ instance Serialise UUID where
85 x -> error $ "invalid schema element " <> show x
138 x -> error $ "invalid schema element " <> show x
86 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)
139 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)
87
140
88−
141+-- | Errors from remote calls.
89data 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
90 | TimeoutError
143 | TimeoutError
144+ | ExceptionError String
91 deriving (Generic, Show, Eq)
145 deriving (Generic, Show, Eq)
92 deriving Serialise via WineryVariant ConnectionError
146 deriving Serialise via WineryVariant ConnectionError
93
147
148+data TimeoutException = TimeoutException
149+ deriving Show
150+
151+instance Exception TimeoutException
152+
94type HostAddr = (Word8, Word8, Word8, Word8)
153type HostAddr = (Word8, Word8, Word8, Word8)
95
154
96allHostAddrs,localHostAddr :: HostAddr
155allHostAddrs,localHostAddr :: HostAddr
@@ -98,111 +157,186 @@ allHostAddrs = (0,0,0,0)
98localHostAddr = (127,0,0,1)
157localHostAddr = (127,0,0,1)
99
158
100-- Each message is length-prefixed by a 32-bit unsigned length.
159-- Each message is length-prefixed by a 32-bit unsigned length.
101−messageBoundaryP :: Parser IO Word8 BS.ByteString
160+envelopeP :: Parser IO Word8 Envelope
102−messageBoundaryP = do
161+envelopeP = do
103− --traceShowM "message!"
162+ let msgTypeP = (P.satisfy (== 0) *>
104− w4x8 <- PD.take 4 FL.toList
163+ (RequestMessage . fromIntegral <$> word32P)) `P.alt`
105− --traceShowM ("w4x8",w4x8)
164+ (P.satisfy (== 1) $> ResponseMessage) `P.alt`
106− let c = fromIntegral (fromOctets w4x8)
165+ (P.satisfy (== 2) $> TimeoutResponseMessage) `P.alt`
107− --traceShowM ("c", c)
166+ (P.satisfy (== 3) $> ExceptionResponseMessage)
108− vals <- PD.take c FL.toList
167+ lenPrefixedByteStringP = do
109− let bytes = BS.pack vals
168+ c <- fromIntegral <$> word32P
110− traceShowM ("vals", BS.length bytes, bytes)
169+ --streamly can't handle takeEQ 0, so add special handling
111− 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
112
219
113type NewConnectionHandler msg = IO (Maybe msg)
220type NewConnectionHandler msg = IO (Maybe msg)
114
221
115−type NewMessageHandler msg resp = msg -> IO (HandlerResponse resp)
222+type NewMessageHandler req resp = req -> IO resp
116−
223+
117−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.
118− NewConnectionHandler resp ->
225+serve ::
119− NewMessageHandler msg resp ->
226+ RequestHandlers s->
227+ s ->
120 HostAddr ->
228 HostAddr ->
121 PortNumber ->
229 PortNumber ->
122 Maybe (MVar SockAddr) ->
230 Maybe (MVar SockAddr) ->
123 IO Bool
231 IO Bool
124−serve connhandler msghandler hostaddr port mSockLock = do
232+serve userMsgHandlers serverState hostaddr port mSockLock = do
125− let handleSock sock = do
233+ let
126− putStrLn "handleSock"
234+ handleSock sock = do
127− let messageHandler msg = do
235+ lockingSocket <- newLock sock
128− putStrLn $ "GOT MSG " ++ show msg
236+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)
129− case msg of
237+
130− Response{} -> putStrLn "client sent response"
131− ResponseExpectedRequest requestID val -> do
132− putStrLn "ResponseExp"
133− resp <- msghandler val
134− case resp of
135− HandlerResponse responseVal -> sendMessage (Response requestID responseVal) sock
136− NoResponse -> error "attempt to return non-response to expected response message"
137− HandlerException _ -> error "TODO HandlerException"
138− AsyncRequest _ val -> do
139− putStrLn "AsyncReq"
140− void $ msghandler val
141− --no response necessaryxs
142− ExceptionResponse{} -> putStrLn "client sent exception response"
143− -- allow the server to send an async welcome message to the new client, if necessary
144− mResp <- connhandler
145− case mResp of
146− Nothing -> pure ()
147− Just resp -> do
148− requestID <- UUID <$> UUIDBase.nextRandom
149− sendMessage (AsyncRequest requestID resp) sock
150− drainSocketMessages sock messageHandler
151− serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . 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
152 pure True
239 pure True
153
240
154−--add callback to allow for responses via socket
241+openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s
155−type MessageHandler a = Message a -> IO ()
242+openEnvelope (Envelope eprint _ _ bytes) =
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
156
249
157−type AsyncMessageHandler a = a -> IO ()
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 ::
261+ Locking Socket
262+ -> RequestHandlers s
263+ -> s
264+ -> Envelope
265+ -> IO ()
266+serverEnvelopeHandler _ _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()
267+serverEnvelopeHandler _ _ _ (Envelope _ ExceptionResponseMessage _ _) = pure ()
268+serverEnvelopeHandler _ _ _ (Envelope _ ResponseMessage _ _) = pure ()
269+serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (RequestMessage timeoutms) msgId _) = do
270+ --find first matching handler
271+ let runTimeout :: IO b -> IO (Maybe b)
272+ runTimeout m =
273+ if timeoutms == 0 then
274+ (Just <$> m) `catch` timeoutExcHandler
275+ else
276+ (timeout (fromIntegral timeoutms) m) `catch` timeoutExcHandler
277+ --allow server-side function to throw TimeoutError which is caught here and becomes TimeoutError value
278+ timeoutExcHandler :: TimeoutException -> IO (Maybe b)
279+ timeoutExcHandler _ = pure Nothing
280+
281+ sState = ConnectionState {
282+ connectionServerState = serverState,
283+ connectionSocket = sockLock
284+ }
285+
286+ firstMatcher (RequestHandler msghandler) Nothing =
287+ case matchEnvelope envelope msghandler of
288+ Nothing -> pure Nothing
289+ Just (dispatchf, decoded) -> do
290+ --TODO add exception handling
291+ mResponse <- runTimeout (dispatchf sState decoded)
292+ let envelopeResponse =
293+ case mResponse of
294+ Just response ->
295+ Envelope (fingerprint response) ResponseMessage msgId (serialise response)
296+ Nothing ->
297+ Envelope (fingerprint TimeoutError) TimeoutResponseMessage msgId BS.empty
298+ sendEnvelope envelopeResponse sockLock
299+ pure (Just ())
300+ firstMatcher (AsyncRequestHandler msghandler) Nothing =
301+ case matchEnvelope envelope msghandler of
302+ Nothing -> pure Nothing
303+ Just (dispatchf, decoded) -> do
304+ _ <- dispatchf sState decoded
305+ pure (Just ())
306+ firstMatcher _ acc = pure acc
307+ eExc <- try $ foldM_ (flip firstMatcher) Nothing msgHandlers :: IO (Either SomeException ())
308+ case eExc of
309+ Left exc ->
310+ let env = Envelope (fingerprint (show exc)) ExceptionResponseMessage msgId (serialise (show exc)) in
311+ sendEnvelope env sockLock
312+ Right () -> pure ()
158
313
159−drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()
160−drainSocketMessages sock msgHandler = do
161− let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
162− handler bs = do
163− --print ("drain handler", bs)
164− case deserialise bs of
165− Left err ->
166− print err
167− Right val -> do
168− --putStrLn "deserialized value for msgHandler"
169− msgHandler val --add response function
170− S.drain $ S.parseManyD messageBoundaryP sockStream & S.mapM handler
171−
172−fromOctets :: [Word8] -> Word32
173−fromOctets = foldl' accum 0
174− where
175− accum a o = (a `shiftL` 8) .|. fromIntegral o
176
314
177−octets :: Word32 -> [Word8]
315+type EnvelopeHandler = Envelope -> IO ()
178−octets w =
316+
179− [ fromIntegral (w `shiftR` 24)
317+drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()
180− , fromIntegral (w `shiftR` 16)
318+drainSocketMessages sock envelopeHandler = do
181− , fromIntegral (w `shiftR` 8)
319+ let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
182− , fromIntegral w
320+ S.drain $ S.serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler
183− ]
184
321
185--send length-tagged bytestring, perhaps should be in network byte order?
322--send length-tagged bytestring, perhaps should be in network byte order?
186−sendMessage :: Serialise a => a -> Socket -> IO ()
323+sendEnvelope :: Envelope -> Locking Socket -> IO ()
187−sendMessage msg socket' = do
324+sendEnvelope envelope sockLock = do
188− putStrLn ("sending bytes: " ++ show (blen <> bytes))
325+ let envelopebytes = encodeEnvelope envelope
189− writeChunk socket' (toArrayS (blen <> bytes))
326+ fullbytes = envelopebytes
190− where
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
191− bytes = serialise msg
328+ withLock sockLock $ \socket' ->
192− len = BS.length bytes
329+ Socket.sendAll socket' fullbytes
193− blen = BS.pack (octets (fromIntegral len))
330+ --traceShowM ("sendEnvelope", envelope)
194−
331+ traceBytes "sendEnvelope" fullbytes
195−
196−--from streamly-bytestring
197−-- | Convert a 'ByteString' to an array of 'Word8'. This function unwraps the
198−-- 'ByteString' and wraps it with 'Array' constructors and hence the operation
199−-- is performed in constant time.
200−{-# INLINE toArrayS #-}
201−toArrayS :: BS.ByteString -> SA.Array Word8
202−--slow path
203−--toArrayS = SA.fromList . BS.unpack
204−toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr
205− where
206− nfp = fp `plusForeignPtr` off
207− endPtr = unsafeForeignPtrToPtr nfp `plusPtr` len `plusPtr` 1 -- ? why +1?
208
332
333+fingerprint :: Typeable a => a -> Fingerprint
334+fingerprint = 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+178−52
@@ -1,4 +1,4 @@
1−{-# LANGUAGE DerivingVia, DeriveGeneric, TypeApplications #-}
1+{-# LANGUAGE DerivingVia, DeriveGeneric, TypeApplications, ExistentialQuantification #-}
2module Curryer.Test.Basic where
2module Curryer.Test.Basic where
3import Test.Tasty
3import Test.Tasty
4import Test.Tasty.HUnit
4import Test.Tasty.HUnit
@@ -7,83 +7,124 @@ import GHC.Generics
7import Control.Concurrent.MVar
7import Control.Concurrent.MVar
8import Network.Socket (SockAddr(..))
8import Network.Socket (SockAddr(..))
9import Control.Concurrent.Async
9import Control.Concurrent.Async
10+import Control.Monad
11+import Control.Concurrent
12+import Control.Concurrent.STM
13+import Control.Exception
14+import Data.List
10
15
11import Network.RPC.Curryer.Server
16import Network.RPC.Curryer.Server
12import Network.RPC.Curryer.Client
17import Network.RPC.Curryer.Client
13
18
19+-- TODO: add test for nested calls
20+
14testTree :: TestTree
21testTree :: TestTree
15−testTree = testGroup "basic" [testCase "simple" testSimpleCall,
22+testTree = testGroup "basic" [testCase "simple" testSimpleCall
16− testCase "client async" testAsyncServerCall,
23+ ,testCase "client async" testAsyncServerCall
17− testCase "server async" testAsyncClientCall
24+ ,testCase "server async" testAsyncClientCall
25+ ,testCase "client sync timeout" testSyncClientCallTimeout
26+ ,testCase "server-side exception" testSyncException
27+ ,testCase "multi-threaded client" testMultithreadedClient
28+ ,testCase "server state" testServerState
29+ ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout
18 ]
30 ]
19
31
20−--client-to-server request
32+
21−data TestRequest = AddTwoNumbersReq Int Int
33+data AddTwoNumbersReq = AddTwoNumbersReq Int Int
22− | TestAsyncReq String
23− | TestCallMeBackReq String
24 deriving (Generic, Show)
34 deriving (Generic, Show)
25− deriving Serialise via WineryVariant TestRequest
35+ deriving Serialise via WineryVariant AddTwoNumbersReq
26
36
27−--server responds to client
37+data TestCallMeBackReq = TestCallMeBackReq String
28−data TestResponse = AddTwoNumbersResp Int
29− | AsyncHello String
30− | TestCallMeBackResp String
31 deriving (Generic, Show, Eq)
38 deriving (Generic, Show, Eq)
32− deriving Serialise via WineryVariant TestResponse
39+ deriving Serialise via WineryVariant TestCallMeBackReq
33−
40+
34−testServerMessageHandler :: Maybe (MVar String) -> TestRequest -> IO (HandlerResponse TestResponse)
41+data TestAsyncReq = TestAsyncReq String
35−testServerMessageHandler mAsyncMVar msg = do
42+ deriving (Generic, Show)
36− --print msg
43+ deriving Serialise via WineryVariant TestAsyncReq
37− case msg of
44+
38− --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 =
39− AddTwoNumbersReq x y -> pure (HandlerResponse (AddTwoNumbersResp (x+y)))
72+ [ RequestHandler $ \_ (AddTwoNumbersReq x y) -> pure (x + y)
40− --respond to a request for an async callback
41− TestCallMeBackReq s -> do
73+ , RequestHandler $ \_ (TestCallMeBackReq s) ->
42− case mAsyncMVar of
74+ case mAsyncMVar of
43− Nothing -> pure ()
75+ Nothing -> pure ()
44− Just mvar -> putMVar mvar s
76+ Just mvar -> putMVar mvar s
77+ , AsyncRequestHandler $ \_ (TestAsyncReq v) ->
78+ maybe (pure ()) (\mvar -> putMVar mvar v) mAsyncMVar
79+ -- an async hello to the server generates an async hello to the client
80+ , AsyncRequestHandler $ \sState (AsyncHelloReq s) -> do
81+ sendMessage (connectionSocket sState) (AsyncHelloReq s)
82+
83+ , RequestHandler $ \_ (RoundtripStringReq s) -> pure s
84+ , RequestHandler $ \_ (DelayMicrosecondsReq ms) -> do
85+ threadDelay ms
86+ pure ()
87+ , RequestHandler $ \_ ThrowServerSideExceptionReq -> do
88+ _ <- error "test server exception"
45− pure NoResponse
89+ pure ()
90+ ]
46
91
47-- test a simple client-to-server round-trip function execution
92-- test a simple client-to-server round-trip function execution
48testSimpleCall :: Assertion
93testSimpleCall :: Assertion
49testSimpleCall = do
94testSimpleCall = do
50 readyVar <- newEmptyMVar
95 readyVar <- newEmptyMVar
51
96
52− server <- async (serve (pure Nothing) (testServerMessageHandler Nothing) localHostAddr 0 (Just readyVar))
97+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))
53 --wait for server to be ready
98 --wait for server to be ready
54 (SockAddrInet port _) <- takeMVar readyVar
99 (SockAddrInet port _) <- takeMVar readyVar
55− let clientHandler :: AsyncMessageHandler TestResponse
56− clientHandler _ = error "async handler called"
57− mkConn :: IO (Connection TestResponse)
58− mkConn = connect clientHandler localHostAddr port
100+ conn <- connect [] localHostAddr port
59− conn <- mkConn
101+ replicateM_ 5 $ do --make five AddTwo calls to shake out parallelism bugs
60− let c :: IO (Either ConnectionError TestResponse)
61− c = call conn (AddTwoNumbersReq 1 1)
102+ x <- call conn (AddTwoNumbersReq 1 1)
62− x <- c
63− assertEqual "server request+response" (Right (AddTwoNumbersResp 2)) x
103+ assertEqual "server request+response" (Right (2 :: Int)) x
64 close conn
104 close conn
105+ cancel server
106+
65
107
66--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
67testAsyncServerCall :: Assertion
109testAsyncServerCall :: Assertion
68testAsyncServerCall = do
110testAsyncServerCall = do
69 portReadyVar <- newEmptyMVar
111 portReadyVar <- newEmptyMVar
70 receivedAsyncMessageVar <- newEmptyMVar
112 receivedAsyncMessageVar <- newEmptyMVar
71−
113+ let clientAsyncHandlers =
72− let testServerNewConnHandler :: NewConnectionHandler TestResponse
114+ [ClientAsyncRequestHandler (\(AsyncHelloReq s) ->
73− testServerNewConnHandler = do
115+ putMVar receivedAsyncMessageVar s)]
74− pure (Just (AsyncHello "welcome"))
75−
76− server <- async (serve testServerNewConnHandler (testServerMessageHandler Nothing) localHostAddr 0 (Just portReadyVar))
116+ server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))
77 (SockAddrInet port _) <- takeMVar portReadyVar
117 (SockAddrInet port _) <- takeMVar portReadyVar
78− let clientHandler :: AsyncMessageHandler TestResponse
79− clientHandler msg = do
80− case msg of
81− AsyncHello str -> putMVar receivedAsyncMessageVar str
82− x -> error ("unexpected message " <> show x)
83− conn <- connect @TestResponse clientHandler localHostAddr port
118+ conn <- connect clientAsyncHandlers localHostAddr port
119+ Right () <- asyncCall conn (AsyncHelloReq "welcome")
84 asyncMessage <- takeMVar receivedAsyncMessageVar
120 asyncMessage <- takeMVar receivedAsyncMessageVar
85 assertEqual "async message" "welcome" asyncMessage
121 assertEqual "async message" "welcome" asyncMessage
86 close conn
122 close conn
123+ cancel server
124+
125+
126+emptyServerState :: ()
127+emptyServerState = ()
87
128
88--test that the client can make a non-blocking call
129--test that the client can make a non-blocking call
89testAsyncClientCall :: Assertion
130testAsyncClientCall :: Assertion
@@ -91,12 +132,97 @@ testAsyncClientCall = do
91 portReadyVar <- newEmptyMVar
132 portReadyVar <- newEmptyMVar
92 receivedAsyncMessageVar <- newEmptyMVar
133 receivedAsyncMessageVar <- newEmptyMVar
93
134
94− server <- async (serve (pure Nothing) (testServerMessageHandler (Just receivedAsyncMessageVar)) localHostAddr 0 (Just portReadyVar))
135+ server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))
95 (SockAddrInet port _) <- takeMVar portReadyVar
136 (SockAddrInet port _) <- takeMVar portReadyVar
96−
137+
97− conn <- connect @TestResponse (\_ -> pure ()) localHostAddr port
138+ conn <- connect [] localHostAddr port
98 --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
99 Right () <- asyncCall conn (TestCallMeBackReq "hi server")
140 Right () <- asyncCall conn (TestCallMeBackReq "hi server")
100 asyncMessage <- takeMVar receivedAsyncMessageVar
141 asyncMessage <- takeMVar receivedAsyncMessageVar
101 assertEqual "async message" "hi server" asyncMessage
142 assertEqual "async message" "hi server" asyncMessage
102 close conn
143 close conn
144+ cancel server
145+
146+testSyncClientCallTimeout :: Assertion
147+testSyncClientCallTimeout = do
148+ readyVar <- newEmptyMVar
149+
150+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))
151+ --wait for server to be ready
152+ (SockAddrInet port _) <- takeMVar readyVar
153+ conn <- connect [] localHostAddr port
154+ x <- callTimeout @_ @Int (Just 500) conn (DelayMicrosecondsReq 1000)
155+ assertEqual "client sync timeout" (Left TimeoutError) x
156+ close conn
157+ cancel server
158+
159+testSyncException :: Assertion
160+testSyncException = do
161+ readyVar <- newEmptyMVar
162+
163+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))
164+ (SockAddrInet port _) <- takeMVar readyVar
165+
166+ conn <- connect [] localHostAddr port
167+ ret <- call conn ThrowServerSideExceptionReq
168+ case ret of
169+ Left (ExceptionError actualExc) ->
170+ assertBool "server-side exception" ("test server exception" `isPrefixOf` actualExc)
171+ Left otherErr ->
172+ assertFailure ("server-side error " <> show otherErr)
173+ Right () -> assertFailure "missed exception"
174+ close conn
175+ cancel server
176+
177+--throw large messages (> PIPE_BUF) at the server from multiple client connections to exercise the socket lock
178+testMultithreadedClient :: Assertion
179+testMultithreadedClient = do
180+ readyVar <- newEmptyMVar
181+
182+ server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))
183+ (SockAddrInet port _) <- takeMVar readyVar
184+ conn <- connect [] localHostAddr port
185+ let bigString = replicate (1024 * 1000) 'x'
186+ replicateM_ 10 $ do
187+ ret <- call conn (RoundtripStringReq bigString)
188+ assertEqual "big string multithread" (Right bigString) ret
189+ --putStrLn "plus one"
190+ close conn
191+ cancel server
192+
193+testServerState :: Assertion
194+testServerState = do
195+ readyVar <- newEmptyMVar
196+
197+ let serverHandlers = [RequestHandler (\sState ChangeServerState ->
198+ atomically $
199+ modifyTVar (connectionServerState sState) (+ 1))
200+ ]
201+ serverState <- newTVarIO @Int 1
202+ server <- async (serve serverHandlers serverState localHostAddr 0 (Just readyVar))
203+ (SockAddrInet port _) <- takeMVar readyVar
204+ conn <- connect [] localHostAddr port
205+ ret <- call conn ChangeServerState
206+ assertEqual "server ret" (Right ()) ret
207+ serverState' <- readTVarIO serverState
208+ assertEqual "server state" 2 serverState'
209+ close conn
210+ cancel server
211+
212+--test that the request handler can throw a TimeoutError exception which is converted into a TimeoutError response
213+testRequestHandlerThrowTimeout :: Assertion
214+testRequestHandlerThrowTimeout = do
215+ readyVar <- newEmptyMVar
216+
217+ let serverHandlers = [RequestHandler (\_ ThrowTimeout ->
218+ throw TimeoutException >> pure (1 :: Int)
219+ )
220+ ]
221+ server <- async (serve serverHandlers () localHostAddr 0 (Just readyVar))
222+ (SockAddrInet port _) <- takeMVar readyVar
223+ conn <- connect [] localHostAddr port
224+ ret <- call @_ @Int conn ThrowTimeout
225+ assertEqual "handler timeout exception" (Left TimeoutError) ret
226+ close conn
227+ cancel server
228+