curryer-rpc (empty) → 0.1
raw patch · 12 files changed
+1104/−0 lines, 12 filesdep +asyncdep +basedep +binary
Dependencies added: async, base, binary, bytestring, containers, criterion, curryer-rpc, exceptions, fast-builder, hashable, network, network-byte-order, optparse-generic, stm, stm-containers, streamly, tasty, tasty-hunit, time, uuid, winery
Files
- Changelog.markdown +4/−0
- README.markdown +78/−0
- bench/Basic.hs +78/−0
- cabal.project +6/−0
- curryer-rpc.cabal +83/−0
- examples/SimpleKeyValueClient.hs +39/−0
- examples/SimpleKeyValueServer.hs +30/−0
- src/Network/RPC/Curryer/Client.hs +147/−0
- src/Network/RPC/Curryer/Server.hs +336/−0
- src/Network/RPC/Curryer/StreamlyAdditions.hs +66/−0
- test/Curryer/Test/Basic.hs +228/−0
- test/Driver.hs +9/−0
+ Changelog.markdown view
@@ -0,0 +1,4 @@+# v0.1 (2020-12-27)++* initial release to support [Project:M36](https://github.com/agentm/project-m36)+
+ README.markdown view
@@ -0,0 +1,78 @@+# Curryer - Fast Haskell-to-Haskell RPC++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.++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.++## Features++* blocking and non-blocking remote function calls+* asynchronous server-to-client callbacks (for server-initiated notifications)+* timeouts+* leverages [winery](https://hackage.haskell.org/package/winery) for high-performance serialization++## Requirements++* GHC 8.6+++## Code Example++[Server](https://github.com/agentm/curryer/examples/SimpleKeyValueServer.hs):++```haskell+data SetKey = SetKey String String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant SetKey++data GetKey = GetKey String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant GetKey++main :: IO ()+main = do+ kvmap <- M.newIO+ void $ serve kvRequestHandlers kvmap localHostAddr 8765 Nothing+ +kvRequestHandlers :: RequestHandlers (M.Map String String)+kvRequestHandlers = [ RequestHandler $ \state (SetKey k v) ->+ atomically $ M.insert v k (connectionServerState state)+ , RequestHandler $ \state (GetKey k) ->+ atomically $ M.lookup k (connectionServerState state)+ ]+```++[Client](https://github.com/agentm/curryer/examples/SimpleKeyValueClient.hs):++```haskell+data SetKey = SetKey String String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant SetKey++data GetKey = GetKey String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant GetKey++data CommandOptions = Get {name :: String}+ | Set {name :: String, value :: String}+ deriving (Generic, Show)++instance ParseRecord CommandOptions+ +main :: IO ()+main = do+ opts <- getRecord "SimpleKeyValueClient"+ conn <- connect [] localHostAddr 8765+ case opts of+ Get k -> do+ eRet <- call conn (GetKey k)+ case eRet of+ Left err -> error (show err)+ Right (Just val) -> putStrLn val+ Right Nothing -> error "no such key"+ Set k v -> do+ eRet <- call conn (SetKey k v)+ case eRet of+ Left err -> error (show err)+ Right () -> pure ()++```
+ bench/Basic.hs view
@@ -0,0 +1,78 @@+{-# LANGUAGE DerivingStrategies, DeriveGeneric, DerivingVia #-}+import Network.RPC.Curryer.Server+import Network.RPC.Curryer.Client+import Criterion.Main+import GHC.Generics+import Control.Concurrent.MVar+import Network.Socket (SockAddr(..))+import Codec.Winery+import Control.Concurrent.Async+import Control.Concurrent+import Control.Monad+import qualified Data.ByteString as BS++main :: IO ()+main = do+ --start server shared across benchmarks- gather stats on round-trip requests-responses+ portReadyMVar <- newEmptyMVar+ server <- async (serve benchmarkServerRequestHandlers () localHostAddr 0 (Just portReadyMVar))+ --wait for server to be ready+ (SockAddrInet port _) <- takeMVar portReadyMVar+ clientConn <- connect [] localHostAddr port++ _ <- syncWaitRequest clientConn 100+ let syncWait ms = nfIO (syncWaitRequest clientConn ms)+ syncByteString bs = nfIO (syncByteStringRequest clientConn bs)+ bs0 = BS.empty + bs100 = BS.pack (replicate 100 3)+ bs1000 = BS.pack (replicate 1000 3)+ bs1M = BS.pack (replicate 1000000 3)+ defaultMain [bgroup "wait sync" [bench "0 ms" (syncWait 0),+ bench "10 ms" (syncWait 10),+ bench "100 ms" (syncWait 100)],+ bgroup "bytestring roundtrip" [bench "0 bytes" (syncByteString bs0)+ ,bench "100 bytes" (syncByteString bs100)+ ,bench "1000 bytes" (syncByteString bs1000)+ ,bench "1000000 bytes" (syncByteString bs1M)+ ]]++ --TODO: bench for > PIPE_BUF bytes++ close clientConn + cancel server++data WaitMillisecondsReq = WaitMillisecondsReq Int --ask the server to respond in X milliseconds+ deriving (Generic, Show)+ deriving Serialise via WineryVariant WaitMillisecondsReq++data WaitByteStringReq = WaitByteStringReq BS.ByteString+ deriving (Generic, Show)+ deriving Serialise via WineryVariant WaitByteStringReq++data WaitMillisecondsResp = WaitMillisecondsResp+ deriving (Generic, Show, Eq)+ deriving Serialise via WineryVariant WaitMillisecondsResp++benchmarkServerRequestHandlers :: RequestHandlers a+benchmarkServerRequestHandlers =+ [RequestHandler $ \_ (WaitMillisecondsReq ms) -> do+ threadDelay (1000 * ms)+ pure WaitMillisecondsResp,+ RequestHandler $ \_ (WaitByteStringReq bs) ->+ pure bs+ ]++--perform 10000 small synchronous requests+syncWaitRequest :: Connection -> Int -> IO Int+syncWaitRequest conn ms = do+ ret <- call conn (WaitMillisecondsReq ms)+ when (ret /= Right WaitMillisecondsResp) (error "ret failed")+ pure ms++syncByteStringRequest :: Connection -> BS.ByteString -> IO BS.ByteString+syncByteStringRequest conn bs = do+ ret <- call conn (WaitByteStringReq bs)+ case ret of+ Right bs' -> pure bs'+ Left err -> error (show err)+
+ cabal.project view
@@ -0,0 +1,6 @@+packages: .++source-repository-package+ type: git+ location: https://github.com/composewell/streamly.git+ tag: ac3af8749194f1788704dda8667d0b3807075cc2
+ curryer-rpc.cabal view
@@ -0,0 +1,83 @@+Name: curryer-rpc+Version: 0.1+License: PublicDomain+Build-Type: Simple+Homepage: https://github.com/agentm/curryer+Bug-Reports: https://github.com/agentm/curryer+Author: AgentM+Stability: experimental+Category: RPC+Maintainer: agentm@themactionfaction.com+Cabal-Version: >= 1.10+Synopsis: Fast, Haskell RPC+Description: Haskell-to-Haskell RPC using Winery serialization.+Extra-Source-Files: Changelog.markdown README.markdown cabal.project++Source-Repository head+ Type: git+ location: https://github.com/agentm/curryer++Library+ Build-Depends: base >= 4.12 && < 4.15+ , winery+ , bytestring+ , streamly >= 0.7.2+ , network+ , exceptions+ , async+ , uuid+ , fast-builder+ , binary+ , containers+ , stm-containers+ , hashable+ , time+ , network-byte-order+ , stm+ Hs-Source-Dirs: ./src+ Default-Language: Haskell2010+ ghc-options: -Wall -fwarn-unused-binds -fwarn-unused-imports + Exposed-Modules:+ Network.RPC.Curryer.Server+ Network.RPC.Curryer.Client+ Network.RPC.Curryer.StreamlyAdditions++Test-Suite test+ type: exitcode-stdio-1.0+ main-is: Driver.hs+ hs-source-dirs: test+ default-language: Haskell2010+ ghc-options: -Wall -fwarn-unused-binds -fwarn-unused-imports+ build-depends: tasty+ , tasty-hunit+ , base+ , curryer-rpc+ , winery+ , network+ , async+ , stm + other-modules: Curryer.Test.Basic++Benchmark perf+ Default-Language: Haskell2010+ Default-Extensions: OverloadedStrings+ Build-Depends: base, criterion, curryer-rpc, network, winery, async, bytestring++ Main-Is: Basic.hs+ Type: exitcode-stdio-1.0+ GHC-Options: -Wall -threaded -rtsopts+ -- -fprof-auto+ HS-Source-Dirs: ./bench+ +Executable SimpleKeyValueServer+ Build-Depends: stm-containers, stm, base, curryer-rpc, winery+ Main-Is: examples/SimpleKeyValueServer.hs+ GHC-Options: -Wall -threaded+ Default-Language: Haskell2010++Executable SimpleKeyValueClient+ Build-Depends: stm, base, curryer-rpc, winery, optparse-generic+ Main-Is: examples/SimpleKeyValueClient.hs+ GHC-Options: -Wall -threaded+ Default-Language: Haskell2010+
+ examples/SimpleKeyValueClient.hs view
@@ -0,0 +1,39 @@+{-# LANGUAGE DerivingVia, DeriveGeneric, OverloadedStrings #-}+import GHC.Generics+import Network.RPC.Curryer.Client+import Network.RPC.Curryer.Server (localHostAddr)+import Options.Generic+import Codec.Winery++data SetKey = SetKey String String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant SetKey++data GetKey = GetKey String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant GetKey++data CommandOptions = Get {name :: String}+ | Set {name :: String, value :: String}+ deriving (Generic, Show)++instance ParseRecord CommandOptions+ +main :: IO ()+main = do+ opts <- getRecord "SimpleKeyValueClient"+ -- connect to the remote server (in this case on the localhost address)+ conn <- connect [] localHostAddr 8765+ case opts of+ Get k -> do+ --call the remote function and validate the result+ eRet <- call conn (GetKey k)+ case eRet of+ Left err -> error (show err)+ Right (Just val) -> putStrLn val+ Right Nothing -> error "no such key"+ Set k v -> do+ eRet <- call conn (SetKey k v)+ case eRet of+ Left err -> error (show err)+ Right () -> pure ()
+ examples/SimpleKeyValueServer.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE DerivingVia, DeriveGeneric #-}+import Network.RPC.Curryer.Server+import Control.Concurrent.STM+import StmContainers.Map as M+import GHC.Generics+import Codec.Winery+import Data.Functor++-- create data structures to represent remote function calls+data SetKey = SetKey String String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant SetKey++data GetKey = GetKey String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant GetKey++--call the `serve` function to handle incoming requests+main :: IO ()+main = do+ kvmap <- M.newIO+ void $ serve kvRequestHandlers kvmap localHostAddr 8765 Nothing++-- setup incoming request handlers to operate on the server's state+kvRequestHandlers :: RequestHandlers (M.Map String String)+kvRequestHandlers = [ RequestHandler $ \state (SetKey k v) ->+ atomically $ M.insert v k (connectionServerState state)+ , RequestHandler $ \state (GetKey k) ->+ atomically $ M.lookup k (connectionServerState state)+ ]
+ src/Network/RPC/Curryer/Client.hs view
@@ -0,0 +1,147 @@+{-# LANGUAGE RankNTypes, ScopedTypeVariables, GADTs #-}+module Network.RPC.Curryer.Client where+import Network.RPC.Curryer.Server+import Network.Socket as Socket+import qualified Streamly.Network.Inet.TCP as TCP+import Codec.Winery+import Control.Concurrent.Async+import qualified Data.UUID.V4 as UUIDBase+import qualified StmContainers.Map as STMMap+import Control.Concurrent.MVar+import GHC.Conc+import Data.Time.Clock+import System.Timeout+import Control.Monad++type SyncMap = STMMap.Map UUID (MVar (Either ConnectionError BinaryMessage), UTCTime)++-- | Represents a remote connection to server.+data Connection = Connection { _conn_sockLock :: Locking Socket,+ _conn_asyncThread :: Async (),+ _conn_syncmap :: SyncMap+ }++-- | Function handlers run on the client, triggered by the server- useful for asynchronous callbacks.+data ClientAsyncRequestHandler where+ ClientAsyncRequestHandler :: forall a. Serialise a => (a -> IO ()) -> ClientAsyncRequestHandler++type ClientAsyncRequestHandlers = [ClientAsyncRequestHandler]++-- | Connects to a remote server with specific async callbacks registered.+connect :: + ClientAsyncRequestHandlers ->+ HostAddr ->+ PortNumber ->+ IO Connection+connect asyncHandlers hostAddr portNum = do+ sock <- TCP.connect hostAddr portNum+ syncmap <- STMMap.newIO+ asyncThread <- async (clientAsync sock syncmap asyncHandlers)+ sockLock <- newLock sock+ pure (Connection {+ _conn_sockLock = sockLock,+ _conn_asyncThread = asyncThread,+ _conn_syncmap = syncmap+ })++-- | Close the connection and release all connection resources.+close :: Connection -> IO ()+close conn = do+ withLock (_conn_sockLock conn) $ \sock ->+ Socket.close sock+ cancel (_conn_asyncThread conn)++-- | async thread for handling client-side incoming messages- dispatch to proper waiting thread or asynchronous notifications handler+clientAsync :: + Socket ->+ SyncMap ->+ ClientAsyncRequestHandlers ->+ IO ()+clientAsync sock syncmap asyncHandlers = do+ lsock <- newLock sock+ drainSocketMessages sock (clientEnvelopeHandler asyncHandlers lsock syncmap)++consumeResponse :: UUID -> STMMap.Map UUID (MVar a, b) -> a -> IO ()+consumeResponse msgId syncMap val = do+ match <- atomically $ do+ val' <- STMMap.lookup msgId syncMap+ STMMap.delete msgId syncMap+ pure val'+ case match of+ Nothing -> pure () -- drop message+ Just (mVar, _) -> putMVar mVar val++-- | 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+clientEnvelopeHandler ::+ ClientAsyncRequestHandlers+ -> Locking Socket+ -> SyncMap+ -> Envelope+ -> IO ()+clientEnvelopeHandler handlers _ _ envelope@(Envelope _ (RequestMessage _) _ _) = do+ --should this run off on another green thread?+ let firstMatcher Nothing (ClientAsyncRequestHandler (dispatchf :: a -> IO ())) = do+ case openEnvelope envelope of+ Nothing -> pure Nothing+ Just decoded -> do+ dispatchf decoded+ pure (Just ())+ firstMatcher acc _ = pure acc+ foldM_ firstMatcher Nothing handlers+clientEnvelopeHandler _ _ syncMap (Envelope _ ResponseMessage msgId binaryMessage) =+ consumeResponse msgId syncMap (Right binaryMessage)+clientEnvelopeHandler _ _ syncMap (Envelope _ TimeoutResponseMessage msgId _) =+ consumeResponse msgId syncMap (Left TimeoutError)+clientEnvelopeHandler _ _ syncMap (Envelope _ ExceptionResponseMessage msgId excPayload) = + case deserialise excPayload of+ Left err -> error ("failed to deserialise exception string" <> show err)+ Right excStr ->+ consumeResponse msgId syncMap (Left (ExceptionError excStr))+ +-- | Basic remote function call via data type and return value.+call :: (Serialise request, Serialise response) => Connection -> request -> IO (Either ConnectionError response)+call = callTimeout Nothing++-- | Send a request to the remote server and returns a response but with the possibility of a timeout after n microseconds.+callTimeout :: (Serialise request, Serialise response) => Maybe Int -> Connection -> request -> IO (Either ConnectionError response)+callTimeout mTimeout conn msg = do+ requestID <- UUID <$> UUIDBase.nextRandom + let mVarMap = _conn_syncmap conn+ timeoutms = case mTimeout of+ Nothing -> 0+ Just tm | tm < 0 -> 0+ Just tm -> fromIntegral tm+ + envelope = Envelope fprint (RequestMessage timeoutms) requestID (serialise msg)+ fprint = fingerprint msg+ -- setup mvar to wait for response+ responseMVar <- newEmptyMVar+ now <- getCurrentTime+ atomically $ STMMap.insert (responseMVar, now) requestID mVarMap+ sendEnvelope envelope (_conn_sockLock conn)+ let timeoutMicroseconds =+ case mTimeout of+ Just timeout' -> timeout' + 100 --add 100 ms to account for unknown network latency+ Nothing -> -1+ mResponse <- timeout timeoutMicroseconds (takeMVar responseMVar)+ atomically $ STMMap.delete requestID mVarMap+ case mResponse of+ --timeout+ Nothing ->+ pure (Left TimeoutError)+ Just (Left exc) ->+ pure (Left exc)+ Just (Right binmsg) ->+ --TODO use decoder instead+ case deserialise binmsg of+ Left err -> error ("deserialise client error " <> show err)+ Right m -> pure (Right m)++-- | Call a remote function but do not expect a response from the server.+asyncCall :: Serialise request => Connection -> request -> IO (Either ConnectionError ())+asyncCall conn msg = do+ requestID <- UUID <$> UUIDBase.nextRandom+ let envelope = Envelope fprint (RequestMessage 0) requestID (serialise msg)+ fprint = fingerprint msg+ sendEnvelope envelope (_conn_sockLock conn)+ pure (Right ())
+ src/Network/RPC/Curryer/Server.hs view
@@ -0,0 +1,336 @@+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, CPP, ExistentialQuantification, StandaloneDeriving, GADTs #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+{- HLINT ignore "Use lambda-case" -}+module Network.RPC.Curryer.Server where+import qualified Streamly.Prelude as S+import Streamly.Network.Socket+import Streamly.Internal.Network.Socket (handleWithM)+import Network.Socket as Socket+import Network.Socket.ByteString as Socket+import Streamly.Internal.Data.Parser as P hiding (concatMap)+import Codec.Winery+import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)+import Codec.Winery.Class (mkExtractor)+import GHC.Generics+import GHC.Fingerprint+import Data.Typeable+import Control.Concurrent.MVar (MVar, newMVar, withMVar)+import Control.Exception+import Data.Function ((&))+import Data.Word+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BSL+import qualified Data.ByteString.FastBuilder as BB+import Streamly.Data.Fold as FL+import qualified Streamly.Internal.Data.Stream.IsStream as S+import qualified Data.Binary as B+import qualified Data.UUID as UUIDBase+import qualified Data.UUID.V4 as UUIDBase+import Control.Monad+import Data.Functor++import qualified Network.RPC.Curryer.StreamlyAdditions as SA+--import Control.Monad+import Data.Hashable+import System.Timeout+import qualified Network.ByteOrder as BO++-- for toArrayS conversion+{-import qualified Data.ByteString.Internal as BSI+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA+import Foreign.ForeignPtr (plusForeignPtr)+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)+import GHC.Ptr (plusPtr)+-}+-- define CURRYER_SHOW_BYTES 1+#if CURRYER_SHOW_BYTES == 1+import Debug.Trace+#endif++traceBytes :: Applicative f => String -> BS.ByteString -> f () +#if CURRYER_SHOW_BYTES == 1+traceBytes msg bs = traceShowM (msg, BS.length bs, bs)+#else+traceBytes _ _ = pure ()+#endif++data Locking a = Locking (MVar ()) a++newLock :: a -> IO (Locking a)+newLock x = do+ lock <- newMVar ()+ pure (Locking lock x)+ +withLock :: Locking a -> (a -> IO b) -> IO b+withLock (Locking mvar v) m =+ withMVar mvar $ \_ -> m v++lockless :: Locking a -> a+lockless (Locking _ a) = a++type Timeout = Word32++type BinaryMessage = BS.ByteString++--includes the fingerprint of the incoming data type (held in the BinaryMessage) to determine how to dispatch the message.+--add another envelope type for unencoded binary messages for any easy optimization for in-process communication+data Envelope = Envelope {+ envFingerprint :: !Fingerprint,+ envMessageType :: !MessageType,+ envMsgId :: !UUID,+ envPayload :: !BinaryMessage+ }+ deriving (Generic, Show)++type TimeoutMicroseconds = Int++deriving instance Generic Fingerprint+deriving via WineryVariant Fingerprint instance Serialise Fingerprint++-- | Internal type used to mark envelope types.+data MessageType = RequestMessage TimeoutMicroseconds+ | ResponseMessage+ | TimeoutResponseMessage+ | ExceptionResponseMessage+ deriving (Generic, Show)+ deriving Serialise via WineryVariant MessageType++-- | A list of `RequestHandler`s.+type RequestHandlers serverState = [RequestHandler serverState]++-- | Data types for server-side request handlers, in synchronous (client waits for return value) and asynchronous (client does not wait for return value) forms.+data RequestHandler serverState where+ -- | create a request handler with a response+ RequestHandler :: forall a b serverState. (Serialise a, Serialise b) => (ConnectionState serverState -> a -> IO b) -> RequestHandler serverState+ -- | create an asynchronous request handler where the client does not expect nor await a response+ AsyncRequestHandler :: forall a serverState. Serialise a => (ConnectionState serverState -> a -> IO ()) -> RequestHandler serverState++-- | Server state sent in via `serve` and passed to `RequestHandler`s.+data ConnectionState a = ConnectionState {+ connectionServerState :: a,+ connectionSocket :: Locking Socket+ }++-- | 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.+sendMessage :: Serialise a => Locking Socket -> a -> IO ()+sendMessage lockSock msg = do+ requestID <- UUID <$> UUIDBase.nextRandom+ let env =+ Envelope (fingerprint msg) (RequestMessage timeout') requestID (serialise msg)+ timeout' = 0+ sendEnvelope env lockSock+ +--avoid orphan instance+newtype UUID = UUID { _unUUID :: UUIDBase.UUID }+ deriving (Show, Eq, B.Binary, Hashable)++instance Serialise UUID where+ schemaGen _ = pure (STag (TagStr "Data.UUID") SBytes)+ toBuilder uuid = let bytes = BSL.toStrict (B.encode uuid) in+ varInt (BS.length bytes) <> BB.byteString bytes+ {-# INLINE toBuilder #-}+ extractor = mkExtractor $+ \schema' -> case schema' of+ STag (TagStr "Data.UUID") SBytes ->+ pure $ \term -> case term of+ TBytes bs -> B.decode (BSL.fromStrict bs)+ term' -> throw (InvalidTerm term')+ x -> error $ "invalid schema element " <> show x+ decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)++-- | Errors from remote calls.+data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps+ | TimeoutError+ | ExceptionError String+ deriving (Generic, Show, Eq)+ deriving Serialise via WineryVariant ConnectionError++data TimeoutException = TimeoutException+ deriving Show++instance Exception TimeoutException ++type HostAddr = (Word8, Word8, Word8, Word8)++allHostAddrs,localHostAddr :: HostAddr+allHostAddrs = (0,0,0,0)+localHostAddr = (127,0,0,1)++-- Each message is length-prefixed by a 32-bit unsigned length.+envelopeP :: Parser IO Word8 Envelope+envelopeP = do+ let s = FL.toList+ msgTypeP = (P.satisfy (== 0) *>+ (RequestMessage . fromIntegral <$> word32P)) `P.alt`+ (P.satisfy (== 1) $> ResponseMessage) `P.alt`+ (P.satisfy (== 2) $> TimeoutResponseMessage) `P.alt`+ (P.satisfy (== 3) $> ExceptionResponseMessage)+ lenPrefixedByteStringP = do+ c <- fromIntegral <$> word32P+ --streamly can't handle takeEQ 0, so add special handling+ if c == 0 then+ pure BS.empty+ else+ BS.pack <$> P.takeEQ c s+ Envelope <$> fingerprintP <*> msgTypeP <*> uuidP <*> lenPrefixedByteStringP++encodeEnvelope :: Envelope -> BS.ByteString+encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =+ fingerprintBs <> msgTypeBs <> msgIdBs <> lenPrefixedBs+ where+ fingerprintBs = BO.bytestring64 fp1 <> BO.bytestring64 fp2+ msgTypeBs = case msgType of+ RequestMessage timeoutms -> BS.singleton 0 <> BO.bytestring32 (fromIntegral timeoutms)+ ResponseMessage -> BS.singleton 1+ TimeoutResponseMessage -> BS.singleton 2+ ExceptionResponseMessage -> BS.singleton 3+ msgIdBs =+ case UUIDBase.toWords (_unUUID msgId) of+ (u1, u2, u3, u4) -> foldr ((<>) . BO.bytestring32) BS.empty [u1, u2, u3, u4]+ msgLen = fromIntegral (BS.length bs)+ lenPrefixedBs = BO.bytestring32 msgLen <> bs++fingerprintP :: Parser IO Word8 Fingerprint+fingerprintP =+ Fingerprint <$> word64P <*> word64P++word64P :: Parser IO Word8 Word64+word64P = do+ let s = FL.toList+ b <- P.takeEQ 8 s+ pure (BO.word64 (BS.pack b))++--parse a 32-bit integer from network byte order+word32P :: Parser IO Word8 Word32+word32P = do+ let s = FL.toList+ w4x8 <- P.takeEQ 4 s + pure (BO.word32 (BS.pack w4x8))++-- uuid is encode as 4 32-bit words because of its convenient 32-bit tuple encoding+uuidP :: Parser IO Word8 UUID+uuidP = do+ u1 <- word32P+ u2 <- word32P+ u3 <- word32P+ --u4 <- word32P+ --pure (UUID (UUIDBase.fromWords u1 u2 u3 u4))-}+ --(UUID . UUIDBase.fromWords) <$> word32P <*> word32P <*> word32P <*> word32P+ UUID . UUIDBase.fromWords u1 u2 u3 <$> word32P++type NewConnectionHandler msg = IO (Maybe msg)++type NewMessageHandler req resp = req -> IO resp++-- | 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.+serve :: + RequestHandlers s->+ s ->+ HostAddr ->+ PortNumber ->+ Maybe (MVar SockAddr) ->+ IO Bool+serve userMsgHandlers serverState hostaddr port mSockLock = do+ let+ handleSock sock = do+ lockingSocket <- newLock sock+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)+ + S.serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.parallely . S.mapM (handleWithM handleSock) & S.drain+ pure True++openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s+openEnvelope (Envelope eprint _ _ bytes) =+ if eprint == fingerprint (undefined :: s) then+ case deserialise bytes of+ Left err -> error (show err)+ Right v -> Just v+ else+ Nothing++matchEnvelope :: forall a b s. (Serialise a, Serialise b, Typeable b) =>+ Envelope -> + (ConnectionState s -> a -> IO b) ->+ Maybe (ConnectionState s -> a -> IO b, a)+matchEnvelope envelope dispatchf =+ case openEnvelope envelope :: Maybe a of+ Nothing -> Nothing+ Just decoded -> Just (dispatchf, decoded)++-- | Called by `serve` to process incoming envelope requests. Never returns, so use `async` to spin it off on another thread.+serverEnvelopeHandler :: + Locking Socket+ -> RequestHandlers s+ -> s + -> Envelope+ -> IO ()+serverEnvelopeHandler _ _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()+serverEnvelopeHandler _ _ _ (Envelope _ ExceptionResponseMessage _ _) = pure ()+serverEnvelopeHandler _ _ _ (Envelope _ ResponseMessage _ _) = pure ()+serverEnvelopeHandler sockLock msgHandlers serverState envelope@(Envelope _ (RequestMessage timeoutms) msgId _) = do+ --find first matching handler+ let runTimeout :: IO b -> IO (Maybe b)+ runTimeout m = + if timeoutms == 0 then+ (Just <$> m) `catch` timeoutExcHandler+ else+ (timeout (fromIntegral timeoutms) m) `catch` timeoutExcHandler+ --allow server-side function to throw TimeoutError which is caught here and becomes TimeoutError value+ timeoutExcHandler :: TimeoutException -> IO (Maybe b)+ timeoutExcHandler _ = pure Nothing+ + sState = ConnectionState {+ connectionServerState = serverState,+ connectionSocket = sockLock+ }+ + firstMatcher (RequestHandler msghandler) Nothing =+ case matchEnvelope envelope msghandler of+ Nothing -> pure Nothing+ Just (dispatchf, decoded) -> do+ --TODO add exception handling+ mResponse <- runTimeout (dispatchf sState decoded)+ let envelopeResponse =+ case mResponse of+ Just response ->+ Envelope (fingerprint response) ResponseMessage msgId (serialise response)+ Nothing -> + Envelope (fingerprint TimeoutError) TimeoutResponseMessage msgId BS.empty+ sendEnvelope envelopeResponse sockLock+ pure (Just ())+ firstMatcher (AsyncRequestHandler msghandler) Nothing = + case matchEnvelope envelope msghandler of+ Nothing -> pure Nothing+ Just (dispatchf, decoded) -> do+ _ <- dispatchf sState decoded+ pure (Just ())+ firstMatcher _ acc = pure acc+ eExc <- try $ foldM_ (flip firstMatcher) Nothing msgHandlers :: IO (Either SomeException ())+ case eExc of+ Left exc ->+ let env = Envelope (fingerprint (show exc)) ExceptionResponseMessage msgId (serialise (show exc)) in+ sendEnvelope env sockLock+ Right () -> pure ()+++type EnvelopeHandler = Envelope -> IO ()++drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()+drainSocketMessages sock envelopeHandler = do+ let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)+ S.drain $ S.serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler++--send length-tagged bytestring, perhaps should be in network byte order?+sendEnvelope :: Envelope -> Locking Socket -> IO ()+sendEnvelope envelope sockLock = do+ let envelopebytes = encodeEnvelope envelope+ fullbytes = envelopebytes+ --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+ withLock sockLock $ \socket' ->+ Socket.sendAll socket' fullbytes+ --traceShowM ("sendEnvelope", envelope)+ traceBytes "sendEnvelope" fullbytes++fingerprint :: Typeable a => a -> Fingerprint+fingerprint = typeRepFingerprint . typeOf+
+ src/Network/RPC/Curryer/StreamlyAdditions.hs view
@@ -0,0 +1,66 @@+module Network.RPC.Curryer.StreamlyAdditions where+import Control.Monad.IO.Class+import Network.Socket (Socket, PortNumber, SocketOption, SockAddr(..), maxListenQueue, Family(..), SocketType(..), defaultProtocol, tupleToHostAddress, withSocketsDo, socket, setSocketOption, bind, getSocketName)+import qualified Network.Socket as Net+import Control.Exception (onException)+import Control.Concurrent.MVar+import Data.Word+import qualified Streamly.Internal.Data.Unfold as UF+import Streamly.Network.Socket hiding (accept)+import qualified Streamly.Internal.Data.Stream.StreamD.Type as D+import Streamly.Internal.Data.Unfold.Types (Unfold(..))++acceptOnAddrWith+ :: MonadIO m+ => [(SocketOption, Int)]+ -> Maybe (MVar SockAddr)+ -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket+acceptOnAddrWith opts mSockLock = UF.lmap f (accept mSockLock)+ where+ f (addr, port) =+ (maxListenQueue+ , SockSpec+ { sockFamily = AF_INET+ , sockType = Stream+ , sockProto = defaultProtocol -- TCP+ , sockOpts = opts+ }+ , SockAddrInet port (tupleToHostAddress addr)+ )++accept :: MonadIO m => Maybe (MVar SockAddr) -> Unfold m (Int, SockSpec, SockAddr) Socket+accept mSockLock = UF.map fst (listenTuples mSockLock)++initListener :: Int -> SockSpec -> SockAddr -> IO Socket+initListener listenQLen sockSpec addr =+ withSocketsDo $ do+ sock <- socket (sockFamily sockSpec) (sockType sockSpec) (sockProto sockSpec)+ use sock `onException` Net.close sock+ return sock++ where++ use sock = do+ mapM_ (uncurry (setSocketOption sock)) (sockOpts sockSpec)+ bind sock addr+ Net.listen sock listenQLen ++listenTuples :: MonadIO m+ => Maybe (MVar SockAddr)+ -> Unfold m (Int, SockSpec, SockAddr) (Socket, SockAddr)+listenTuples mSockLock = Unfold step inject+ where+ inject (listenQLen, spec, addr) =+ liftIO $ do+ sock <- initListener listenQLen spec addr+ sockAddr <- getSocketName sock+ case mSockLock of+ Just mvar ->+ putMVar mvar sockAddr+ Nothing -> pure ()+ pure sock++ step listener = do+ r <- liftIO (Net.accept listener `onException` Net.close listener)+ return $ D.Yield r listener+
+ test/Curryer/Test/Basic.hs view
@@ -0,0 +1,228 @@+{-# LANGUAGE DerivingVia, DeriveGeneric, TypeApplications, ExistentialQuantification #-}+module Curryer.Test.Basic where+import Test.Tasty+import Test.Tasty.HUnit+import Codec.Winery+import GHC.Generics+import Control.Concurrent.MVar+import Network.Socket (SockAddr(..))+import Control.Concurrent.Async+import Control.Monad+import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception+import Data.List++import Network.RPC.Curryer.Server+import Network.RPC.Curryer.Client++-- TODO: add test for nested calls++testTree :: TestTree+testTree = testGroup "basic" [testCase "simple" testSimpleCall+ ,testCase "client async" testAsyncServerCall+ ,testCase "server async" testAsyncClientCall+ ,testCase "client sync timeout" testSyncClientCallTimeout+ ,testCase "server-side exception" testSyncException+ ,testCase "multi-threaded client" testMultithreadedClient+ ,testCase "server state" testServerState+ ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout+ ]+++data AddTwoNumbersReq = AddTwoNumbersReq Int Int+ deriving (Generic, Show)+ deriving Serialise via WineryVariant AddTwoNumbersReq++data TestCallMeBackReq = TestCallMeBackReq String+ deriving (Generic, Show, Eq)+ deriving Serialise via WineryVariant TestCallMeBackReq++data TestAsyncReq = TestAsyncReq String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant TestAsyncReq++data DelayMicrosecondsReq = DelayMicrosecondsReq Int+ deriving (Generic, Show)+ deriving Serialise via WineryVariant DelayMicrosecondsReq++data RoundtripStringReq = RoundtripStringReq String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant RoundtripStringReq++data ThrowServerSideExceptionReq = ThrowServerSideExceptionReq+ deriving (Generic, Show)+ deriving Serialise via WineryVariant ThrowServerSideExceptionReq++data ChangeServerState = ChangeServerState+ deriving (Generic, Show)+ deriving Serialise via WineryVariant ChangeServerState++--used to server -> client async request+data AsyncHelloReq = AsyncHelloReq String+ deriving (Generic, Show)+ deriving Serialise via WineryVariant AsyncHelloReq++data ThrowTimeout = ThrowTimeout+ deriving (Generic, Show)+ deriving Serialise via WineryVariant ThrowTimeout++testServerRequestHandlers :: Maybe (MVar String) -> RequestHandlers ()+testServerRequestHandlers mAsyncMVar =+ [ RequestHandler $ \_ (AddTwoNumbersReq x y) -> pure (x + y)+ , RequestHandler $ \_ (TestCallMeBackReq s) ->+ case mAsyncMVar of+ Nothing -> pure ()+ Just mvar -> putMVar mvar s+ , AsyncRequestHandler $ \_ (TestAsyncReq v) ->+ maybe (pure ()) (\mvar -> putMVar mvar v) mAsyncMVar+ -- an async hello to the server generates an async hello to the client + , AsyncRequestHandler $ \sState (AsyncHelloReq s) -> do+ sendMessage (connectionSocket sState) (AsyncHelloReq s)+ + , RequestHandler $ \_ (RoundtripStringReq s) -> pure s+ , RequestHandler $ \_ (DelayMicrosecondsReq ms) -> do+ threadDelay ms+ pure ()+ , RequestHandler $ \_ ThrowServerSideExceptionReq -> do+ _ <- error "test server exception"+ pure ()+ ]+ +-- test a simple client-to-server round-trip function execution+testSimpleCall :: Assertion+testSimpleCall = do+ readyVar <- newEmptyMVar+ + server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+ --wait for server to be ready+ (SockAddrInet port _) <- takeMVar readyVar+ conn <- connect [] localHostAddr port+ replicateM_ 5 $ do --make five AddTwo calls to shake out parallelism bugs+ x <- call conn (AddTwoNumbersReq 1 1)+ assertEqual "server request+response" (Right (2 :: Int)) x+ close conn+ cancel server+++--test that the client can proces a server-initiated asynchronous callback from the server+testAsyncServerCall :: Assertion+testAsyncServerCall = do+ portReadyVar <- newEmptyMVar+ receivedAsyncMessageVar <- newEmptyMVar+ let clientAsyncHandlers =+ [ClientAsyncRequestHandler (\(AsyncHelloReq s) ->+ putMVar receivedAsyncMessageVar s)]+ server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))+ (SockAddrInet port _) <- takeMVar portReadyVar+ conn <- connect clientAsyncHandlers localHostAddr port+ Right () <- asyncCall conn (AsyncHelloReq "welcome")+ asyncMessage <- takeMVar receivedAsyncMessageVar+ assertEqual "async message" "welcome" asyncMessage+ close conn+ cancel server+++emptyServerState :: ()+emptyServerState = ()++--test that the client can make a non-blocking call+testAsyncClientCall :: Assertion+testAsyncClientCall = do+ portReadyVar <- newEmptyMVar+ receivedAsyncMessageVar <- newEmptyMVar+ + server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))+ (SockAddrInet port _) <- takeMVar portReadyVar++ conn <- connect [] localHostAddr port+ --send an async message, wait for an async response to confirm receipt+ Right () <- asyncCall conn (TestCallMeBackReq "hi server")+ asyncMessage <- takeMVar receivedAsyncMessageVar+ assertEqual "async message" "hi server" asyncMessage + close conn+ cancel server++testSyncClientCallTimeout :: Assertion+testSyncClientCallTimeout = do+ readyVar <- newEmptyMVar+ + server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+ --wait for server to be ready+ (SockAddrInet port _) <- takeMVar readyVar+ conn <- connect [] localHostAddr port+ x <- callTimeout @_ @Int (Just 500) conn (DelayMicrosecondsReq 1000)+ assertEqual "client sync timeout" (Left TimeoutError) x+ close conn+ cancel server++testSyncException :: Assertion+testSyncException = do+ readyVar <- newEmptyMVar+ + server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+ (SockAddrInet port _) <- takeMVar readyVar++ conn <- connect [] localHostAddr port+ ret <- call conn ThrowServerSideExceptionReq+ case ret of+ Left (ExceptionError actualExc) ->+ assertBool "server-side exception" ("test server exception" `isPrefixOf` actualExc)+ Left otherErr ->+ assertFailure ("server-side error " <> show otherErr)+ Right () -> assertFailure "missed exception"+ close conn+ cancel server++--throw large messages (> PIPE_BUF) at the server from multiple client connections to exercise the socket lock+testMultithreadedClient :: Assertion+testMultithreadedClient = do+ readyVar <- newEmptyMVar+ + server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+ (SockAddrInet port _) <- takeMVar readyVar+ conn <- connect [] localHostAddr port+ let bigString = replicate (1024 * 1000) 'x'+ replicateM_ 10 $ do+ ret <- call conn (RoundtripStringReq bigString)+ assertEqual "big string multithread" (Right bigString) ret+ --putStrLn "plus one"+ close conn+ cancel server++testServerState :: Assertion+testServerState = do+ readyVar <- newEmptyMVar++ let serverHandlers = [RequestHandler (\sState ChangeServerState -> + atomically $+ modifyTVar (connectionServerState sState) (+ 1))+ ]+ serverState <- newTVarIO @Int 1+ server <- async (serve serverHandlers serverState localHostAddr 0 (Just readyVar))+ (SockAddrInet port _) <- takeMVar readyVar+ conn <- connect [] localHostAddr port+ ret <- call conn ChangeServerState+ assertEqual "server ret" (Right ()) ret+ serverState' <- readTVarIO serverState+ assertEqual "server state" 2 serverState'+ close conn+ cancel server++--test that the request handler can throw a TimeoutError exception which is converted into a TimeoutError response+testRequestHandlerThrowTimeout :: Assertion+testRequestHandlerThrowTimeout = do+ readyVar <- newEmptyMVar++ let serverHandlers = [RequestHandler (\_ ThrowTimeout ->+ throw TimeoutException >> pure (1 :: Int)+ )+ ]+ server <- async (serve serverHandlers () localHostAddr 0 (Just readyVar))+ (SockAddrInet port _) <- takeMVar readyVar+ conn <- connect [] localHostAddr port+ ret <- call @_ @Int conn ThrowTimeout+ assertEqual "handler timeout exception" (Left TimeoutError) ret+ close conn+ cancel server+
+ test/Driver.hs view
@@ -0,0 +1,9 @@+import Test.Tasty+import Curryer.Test.Basic as Basic++main :: IO ()+main = defaultMain curryerTestTree++curryerTestTree :: TestTree+curryerTestTree = testGroup "curryer" [Basic.testTree]+