diff --git a/Changelog.markdown b/Changelog.markdown --- a/Changelog.markdown +++ b/Changelog.markdown @@ -1,3 +1,7 @@ +# v0.2 (2021-12-13) + +* update for streamly 0.8.0 (we are no longer pegged to a pre-release streamly) + # v0.1 (2020-12-27) * initial release to support [Project:M36](https://github.com/agentm/project-m36) diff --git a/cabal.project b/cabal.project deleted file mode 100644 --- a/cabal.project +++ /dev/null @@ -1,6 +0,0 @@ -packages: . - -source-repository-package - type: git - location: https://github.com/composewell/streamly.git - tag: ac3af8749194f1788704dda8667d0b3807075cc2 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.1 +Version: 0.2 License: PublicDomain Build-Type: Simple Homepage: https://github.com/agentm/curryer @@ -11,7 +11,7 @@ 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 +Extra-Source-Files: Changelog.markdown README.markdown Source-Repository head Type: git @@ -21,7 +21,7 @@ Build-Depends: base >= 4.12 && < 4.15 , winery , bytestring - , streamly >= 0.7.2 + , streamly >= 0.8.0 , network , exceptions , async @@ -36,7 +36,7 @@ , stm Hs-Source-Dirs: ./src Default-Language: Haskell2010 - ghc-options: -Wall -fwarn-unused-binds -fwarn-unused-imports + ghc-options: -Wall -fwarn-unused-binds -fwarn-unused-imports Exposed-Modules: Network.RPC.Curryer.Server Network.RPC.Curryer.Client 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 #-} +{-# LANGUAGE RankNTypes, ScopedTypeVariables, GADTs, TypeApplications #-} module Network.RPC.Curryer.Client where import Network.RPC.Curryer.Server import Network.Socket as Socket @@ -93,7 +93,7 @@ clientEnvelopeHandler _ _ syncMap (Envelope _ TimeoutResponseMessage msgId _) = consumeResponse msgId syncMap (Left TimeoutError) clientEnvelopeHandler _ _ syncMap (Envelope _ ExceptionResponseMessage msgId excPayload) = - case deserialise excPayload of + case msgDeserialise excPayload of Left err -> error ("failed to deserialise exception string" <> show err) Right excStr -> consumeResponse msgId syncMap (Left (ExceptionError excStr)) @@ -112,7 +112,7 @@ Just tm | tm < 0 -> 0 Just tm -> fromIntegral tm - envelope = Envelope fprint (RequestMessage timeoutms) requestID (serialise msg) + envelope = Envelope fprint (RequestMessage timeoutms) requestID (msgSerialise msg) fprint = fingerprint msg -- setup mvar to wait for response responseMVar <- newEmptyMVar @@ -132,16 +132,15 @@ Just (Left exc) -> pure (Left exc) Just (Right binmsg) -> - --TODO use decoder instead - case deserialise binmsg of + case msgDeserialise binmsg of Left err -> error ("deserialise client error " <> show err) - Right m -> pure (Right m) + Right v -> pure (Right v) -- | 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) + let envelope = Envelope fprint (RequestMessage 0) requestID (msgSerialise msg) fprint = fingerprint msg sendEnvelope envelope (_conn_sockLock conn) pure (Right ()) 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 @@ -3,14 +3,12 @@ {- 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 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 Codec.Winery import Codec.Winery.Internal (varInt, decodeVarInt, getBytes) -import Codec.Winery.Class (mkExtractor) import GHC.Generics import GHC.Fingerprint import Data.Typeable @@ -21,28 +19,30 @@ 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 Streamly.Data.Fold as FL hiding (foldr) +import qualified Streamly.Internal.Data.Stream.IsStream as P 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 Control.Applicative 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 +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 + #if CURRYER_SHOW_BYTES == 1 import Debug.Trace #endif @@ -54,6 +54,13 @@ traceBytes _ _ = pure () #endif +-- a level of indirection to be able to switch between serialising with and without the winery schema +msgSerialise :: Serialise a => a -> BS.ByteString +msgSerialise = serialiseOnly + +msgDeserialise :: forall s. Serialise s => BS.ByteString -> Either WineryException s +msgDeserialise = deserialiseOnly + data Locking a = Locking (MVar ()) a newLock :: a -> IO (Locking a) @@ -116,7 +123,7 @@ sendMessage lockSock msg = do requestID <- UUID <$> UUIDBase.nextRandom let env = - Envelope (fingerprint msg) (RequestMessage timeout') requestID (serialise msg) + Envelope (fingerprint msg) (RequestMessage timeout') requestID (msgSerialise msg) timeout' = 0 sendEnvelope env lockSock @@ -156,28 +163,38 @@ allHostAddrs = (0,0,0,0) localHostAddr = (127,0,0,1) +msgTypeP :: Parser IO Word8 MessageType +msgTypeP = (P.satisfy (== 0) *> + (RequestMessage . fromIntegral <$> word32P)) <|> + (P.satisfy (== 1) $> ResponseMessage) <|> + (P.satisfy (== 2) $> TimeoutResponseMessage) <|> + (P.satisfy (== 3) $> ExceptionResponseMessage) + -- 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 + let lenPrefixedByteStringP = do c <- fromIntegral <$> word32P --streamly can't handle takeEQ 0, so add special handling +-- traceShowM ("envelopeP payload byteCount"::String, c) if c == 0 then pure BS.empty else - BS.pack <$> P.takeEQ c s + fromArray <$> P.takeEQ c (Arr.writeN c) 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 encodeEnvelope :: Envelope -> BS.ByteString encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) = - fingerprintBs <> msgTypeBs <> msgIdBs <> lenPrefixedBs +{- traceShow ("encodeEnvelope"::String, + ("fingerprint len"::String, BS.length fingerprintBs), + ("msgtype length"::String,BS.length msgTypeBs), + ("id len"::String, BS.length msgIdBs), + ("payload len"::String, payloadLen), + ("complete len"::String, BS.length completeMessage)) $-} + completeMessage where + completeMessage = fingerprintBs <> msgTypeBs <> msgIdBs <> lenPrefixedBs fingerprintBs = BO.bytestring64 fp1 <> BO.bytestring64 fp2 msgTypeBs = case msgType of RequestMessage timeoutms -> BS.singleton 0 <> BO.bytestring32 (fromIntegral timeoutms) @@ -187,8 +204,10 @@ 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 + lenPrefixedBs = BO.bytestring32 payloadLen <> bs + payloadLen = fromIntegral (BS.length bs) + + fingerprintP :: Parser IO Word8 Fingerprint fingerprintP = @@ -204,7 +223,8 @@ word32P :: Parser IO Word8 Word32 word32P = do let s = FL.toList - w4x8 <- P.takeEQ 4 s + w4x8 <- P.takeEQ 4 s +-- traceShowM ("w4x8"::String, BO.word32 (BS.pack w4x8)) pure (BO.word32 (BS.pack w4x8)) -- uuid is encode as 4 32-bit words because of its convenient 32-bit tuple encoding @@ -235,19 +255,26 @@ 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 + + S.fromSerial (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.fromParallel . S.mapM (forSocketM 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 + case msgDeserialise bytes of + Left _e -> {-traceShow ("openEnv error"::String, e) $-} Nothing + Right decoded -> Just decoded else Nothing +--use winery to decode only the data structure and skip the schema +deserialiseOnly :: forall s. Serialise s => BS.ByteString -> Either WineryException s +deserialiseOnly bytes = do + dec <- getDecoder (schema (Proxy :: Proxy s)) + pure (evalDecoder dec bytes) + + matchEnvelope :: forall a b s. (Serialise a, Serialise b, Typeable b) => Envelope -> (ConnectionState s -> a -> IO b) -> @@ -293,7 +320,7 @@ let envelopeResponse = case mResponse of Just response -> - Envelope (fingerprint response) ResponseMessage msgId (serialise response) + Envelope (fingerprint response) ResponseMessage msgId (msgSerialise response) Nothing -> Envelope (fingerprint TimeoutError) TimeoutResponseMessage msgId BS.empty sendEnvelope envelopeResponse sockLock @@ -308,7 +335,7 @@ 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 + let env = Envelope (fingerprint (show exc)) ExceptionResponseMessage msgId (msgSerialise (show exc)) in sendEnvelope env sockLock Right () -> pure () @@ -317,20 +344,35 @@ 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 + S.unfold SSock.read sock + & P.parseMany envelopeP + & S.mapM envelopeHandler + & S.fromAsync + & S.drain --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 + withLock sockLock $ \socket' -> do + {-traceShowM ("sendEnvelope"::String, + ("type"::String, envMessageType envelope), + socket', ("env len"::String, BS.length envelopebytes), + "payloadbytes"::String, envPayload envelope)-} + Socket.sendAll socket' 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,12 +3,13 @@ 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 qualified Streamly.Internal.Data.Stream.StreamD.Type as D -import Streamly.Internal.Data.Unfold.Types (Unfold(..)) +import Streamly.Internal.Data.Unfold.Type (Unfold(..)) acceptOnAddrWith :: MonadIO m @@ -64,3 +65,5 @@ 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)