diff --git a/Changelog.markdown b/Changelog.markdown --- a/Changelog.markdown +++ b/Changelog.markdown @@ -1,3 +1,7 @@ +# v0.3.0 (2023-04-01) + +* require streamly 0.9.0+ + # v0.2.2 (2022-08-17) * add support for GHC 9.2 diff --git a/curryer-rpc.cabal b/curryer-rpc.cabal --- a/curryer-rpc.cabal +++ b/curryer-rpc.cabal @@ -1,5 +1,5 @@ Name: curryer-rpc -Version: 0.2.2 +Version: 0.3.0 License: PublicDomain Build-Type: Simple Homepage: https://github.com/agentm/curryer @@ -21,7 +21,9 @@ Build-Depends: base >= 4.12 && < 4.17 , winery , bytestring - , streamly >= 0.8.1 + , streamly >= 0.9.0 + , streamly-core >= 0.1.0 + , streamly-bytestring >= 0.2.0 , network , exceptions , async @@ -55,7 +57,10 @@ , winery , network , async - , stm + , stm + , streamly-core + , bytestring + , streamly-bytestring other-modules: Curryer.Test.Basic Benchmark perf diff --git a/src/Network/RPC/Curryer/Client.hs b/src/Network/RPC/Curryer/Client.hs --- a/src/Network/RPC/Curryer/Client.hs +++ b/src/Network/RPC/Curryer/Client.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE RankNTypes, ScopedTypeVariables, GADTs, TypeApplications #-} +{-# LANGUAGE RankNTypes, ScopedTypeVariables, GADTs #-} module Network.RPC.Curryer.Client where import Network.RPC.Curryer.Server import Network.Socket as Socket diff --git a/src/Network/RPC/Curryer/Server.hs b/src/Network/RPC/Curryer/Server.hs --- a/src/Network/RPC/Curryer/Server.hs +++ b/src/Network/RPC/Curryer/Server.hs @@ -1,12 +1,15 @@ -{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, CPP, ExistentialQuantification, StandaloneDeriving, GADTs #-} +{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, CPP, ExistentialQuantification, StandaloneDeriving, GADTs, UnboxedTuples, BangPatterns #-} {-# OPTIONS_GHC -fno-warn-orphans #-} {- HLINT ignore "Use lambda-case" -} module Network.RPC.Curryer.Server where -import qualified Streamly.Prelude as S +import qualified Streamly.Data.Stream.Prelude as SP +import Streamly.Data.Stream as Stream hiding (foldr) +import Streamly.Internal.Data.Stream.Concurrent as Stream +import Streamly.Internal.Serialize.FromBytes (word32be) import Streamly.Network.Socket as SSock import Network.Socket as Socket import Network.Socket.ByteString as Socket -import Streamly.Internal.Data.Parser as P hiding (concatMap) +import Streamly.Data.Parser as P import Codec.Winery import Codec.Winery.Internal (varInt, decodeVarInt, getBytes) import GHC.Generics @@ -20,7 +23,9 @@ import qualified Data.ByteString.Lazy as BSL import qualified Data.ByteString.FastBuilder as BB import Streamly.Data.Fold as FL hiding (foldr) -import qualified Streamly.Internal.Data.Stream.IsStream as P +--import qualified Streamly.Internal.Data.Stream.IsStream as P +import qualified Streamly.Data.Stream.Prelude as P +import qualified Streamly.External.ByteString as StreamlyBS import qualified Data.Binary as B import qualified Data.UUID as UUIDBase import qualified Data.UUID.V4 as UUIDBase @@ -32,16 +37,11 @@ import Data.Hashable import System.Timeout import qualified Network.ByteOrder as BO +import qualified Streamly.Internal.Data.Array.Type as Arr --- for toArrayS conversion -import qualified Data.ByteString.Internal as BSI -import qualified Streamly.Internal.Data.Array.Foreign.Type as Arr -import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as ArrT -import GHC.ForeignPtr (ForeignPtr(ForeignPtr)) -import GHC.Ptr (minusPtr, Ptr(..)) - ---define CURRYER_SHOW_BYTES 1 +#define CURRYER_SHOW_BYTES 0 +#define CURRYER_PASS_SCHEMA 0 #if CURRYER_SHOW_BYTES == 1 import Debug.Trace @@ -56,10 +56,18 @@ -- a level of indirection to be able to switch between serialising with and without the winery schema msgSerialise :: Serialise a => a -> BS.ByteString +#if CURRYER_PASS_SCHEMA == 1 +msgSerialise = serialise +#else msgSerialise = serialiseOnly +#endif msgDeserialise :: forall s. Serialise s => BS.ByteString -> Either WineryException s +#if CURRYER_PASS_SCHEMA == 1 +msgDeserialise = deserialise +#else msgDeserialise = deserialiseOnly +#endif data Locking a = Locking (MVar ()) a @@ -162,11 +170,14 @@ type HostAddr = (Word8, Word8, Word8, Word8) +type BParser a = Parser Word8 IO a + allHostAddrs,localHostAddr :: HostAddr allHostAddrs = (0,0,0,0) localHostAddr = (127,0,0,1) -msgTypeP :: Parser IO Word8 MessageType + +msgTypeP :: BParser MessageType msgTypeP = (P.satisfy (== 0) *> (RequestMessage . fromIntegral <$> word32P)) <|> (P.satisfy (== 1) $> ResponseMessage) <|> @@ -174,16 +185,20 @@ (P.satisfy (== 3) $> ExceptionResponseMessage) -- Each message is length-prefixed by a 32-bit unsigned length. -envelopeP :: Parser IO Word8 Envelope +envelopeP :: BParser Envelope envelopeP = do let lenPrefixedByteStringP = do - c <- fromIntegral <$> word32P + c <- fromIntegral <$> word32be --streamly can't handle takeEQ 0, so add special handling -- traceShowM ("envelopeP payload byteCount"::String, c) if c == 0 then - pure BS.empty - else - fromArray <$> P.takeEQ c (Arr.writeN c) + pure mempty + else do + ps <- P.takeEQ c (Arr.writeN c) +-- traceShowM ("envelopeP read bytes", c) + let !bs = StreamlyBS.fromArray ps +-- traceShowM ("unoptimized bs") + pure bs Envelope <$> fingerprintP <*> msgTypeP <*> uuidP <*> lenPrefixedByteStringP --overhead is fingerprint (16 bytes), msgType (1+4 optional bytes for request message), msgId (4 bytes), uuid (16 bytes) = 41 bytes per request message, 37 bytes for all others @@ -212,18 +227,18 @@ -fingerprintP :: Parser IO Word8 Fingerprint +fingerprintP :: BParser Fingerprint fingerprintP = Fingerprint <$> word64P <*> word64P -word64P :: Parser IO Word8 Word64 +word64P :: BParser 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 :: BParser Word32 word32P = do let s = FL.toList w4x8 <- P.takeEQ 4 s @@ -231,7 +246,7 @@ 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 :: BParser UUID uuidP = do u1 <- word32P u2 <- word32P @@ -254,19 +269,19 @@ Maybe (MVar SockAddr) -> IO Bool serve userMsgHandlers serverState hostaddr port mSockLock = do - let - handleSock sock = do + let handleSock sock = do lockingSocket <- newLock sock drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState) - - S.fromSerial (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.fromParallel . S.mapM (forSocketM handleSock) & S.drain + Stream.unfold (SA.acceptorOnAddr [(ReuseAddr, 1)] mSockLock) (hostaddr, port) + & Stream.parMapM id handleSock + & Stream.fold FL.drain pure True openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s openEnvelope (Envelope eprint _ _ bytes) = if eprint == fingerprint (undefined :: s) then case msgDeserialise bytes of - Left _e -> {-traceShow ("openEnv error"::String, e) $-} Nothing + Left _e -> {-traceShow ("openEnv error"::String, _e)-} Nothing Right decoded -> Just decoded else Nothing @@ -304,7 +319,7 @@ if timeoutms == 0 then (Just <$> m) `catch` timeoutExcHandler else - (timeout (fromIntegral timeoutms) m) `catch` timeoutExcHandler + 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 @@ -347,11 +362,11 @@ drainSocketMessages :: Socket -> EnvelopeHandler -> IO () drainSocketMessages sock envelopeHandler = do - S.unfold SSock.read sock + SP.unfold SSock.reader sock & P.parseMany envelopeP - & S.mapM envelopeHandler - & S.fromAsync - & S.drain + & SP.catRights + & SP.parMapM (SP.ordered False) envelopeHandler + & SP.fold FL.drain --send length-tagged bytestring, perhaps should be in network byte order? sendEnvelope :: Envelope -> Locking Socket -> IO () @@ -361,21 +376,12 @@ withLock sockLock $ \socket' -> do {-traceShowM ("sendEnvelope"::String, ("type"::String, envMessageType envelope), - socket', ("env len"::String, BS.length envelopebytes), - "payloadbytes"::String, envPayload envelope)-} + socket', + ("envelope len out"::String, BS.length envelopebytes), + "payloadbytes"::String, envPayload envelope + )-} Socket.sendAll socket' envelopebytes --- traceBytes "sendEnvelope" envelopebytes + traceBytes "sendEnvelope" envelopebytes fingerprint :: Typeable a => a -> Fingerprint fingerprint = typeRepFingerprint . typeOf - -fromArray :: Arr.Array Word8 -> BSI.ByteString -fromArray arr - | aLen == 0 = mempty - | otherwise = {-traceShow ("bsi len"::String, aLen, Arr.byteLength arr) $-} BSI.PS aStartFPtr 0 aLen - where - aStart = Arr.arrStart arr - aEnd = Arr.aEnd arr - aStartFPtr = case Arr.arrStart arr of - Ptr addr -> ForeignPtr addr (ArrT.arrayToFptrContents (Arr.arrContents arr)) - aLen = aEnd `minusPtr` aStart diff --git a/src/Network/RPC/Curryer/StreamlyAdditions.hs b/src/Network/RPC/Curryer/StreamlyAdditions.hs --- a/src/Network/RPC/Curryer/StreamlyAdditions.hs +++ b/src/Network/RPC/Curryer/StreamlyAdditions.hs @@ -3,20 +3,19 @@ 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.Monad.Catch (finally) import Control.Concurrent.MVar import Data.Word import qualified Streamly.Internal.Data.Unfold as UF -import Streamly.Network.Socket hiding (accept) +import Streamly.Network.Socket hiding (acceptor) import qualified Streamly.Internal.Data.Stream.StreamD.Type as D import Streamly.Internal.Data.Unfold.Type (Unfold(..)) -acceptOnAddrWith +acceptorOnAddr :: MonadIO m => [(SocketOption, Int)] -> Maybe (MVar SockAddr) -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket -acceptOnAddrWith opts mSockLock = UF.lmap f (accept mSockLock) +acceptorOnAddr opts mLock = UF.lmap f (acceptor mLock) where f (addr, port) = (maxListenQueue @@ -29,22 +28,8 @@ , 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 +acceptor :: MonadIO m => Maybe (MVar SockAddr) -> Unfold m (Int, SockSpec, SockAddr) Socket +acceptor mLock = UF.map fst (listenTuples mLock) listenTuples :: MonadIO m => Maybe (MVar SockAddr) @@ -56,8 +41,7 @@ sock <- initListener listenQLen spec addr sockAddr <- getSocketName sock case mSockLock of - Just mvar -> - putMVar mvar sockAddr + Just mvar -> putMVar mvar sockAddr Nothing -> pure () pure sock @@ -65,5 +49,17 @@ r <- liftIO (Net.accept listener `onException` Net.close listener) return $ D.Yield r listener -handleWithM :: (Socket -> IO ()) -> Socket -> IO () -handleWithM f sk = finally (f sk) (Net.close sk) +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 + diff --git a/test/Curryer/Test/Basic.hs b/test/Curryer/Test/Basic.hs --- a/test/Curryer/Test/Basic.hs +++ b/test/Curryer/Test/Basic.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE DerivingVia, DeriveGeneric, TypeApplications, ExistentialQuantification #-} +{-# LANGUAGE DerivingVia, DeriveGeneric, TypeApplications, ExistentialQuantification, ScopedTypeVariables #-} module Curryer.Test.Basic where import Test.Tasty import Test.Tasty.HUnit @@ -19,15 +19,16 @@ -- 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 - ] +testTree = testGroup "basic" [ + testCase "simple request and response" 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 @@ -112,7 +113,7 @@ receivedAsyncMessageVar <- newEmptyMVar let clientAsyncHandlers = [ClientAsyncRequestHandler (\(AsyncHelloReq s) -> - putMVar receivedAsyncMessageVar s)] + putMVar receivedAsyncMessageVar s)] server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar)) (SockAddrInet port _) <- takeMVar portReadyVar conn <- connect clientAsyncHandlers localHostAddr port @@ -225,4 +226,3 @@ assertEqual "handler timeout exception" (Left TimeoutError) ret close conn cancel server -