packages feed

curryer-rpc 0.1 → 0.2

raw patch · 6 files changed

+100/−58 lines, 6 filesdep ~streamly

Dependency ranges changed: streamly

Files

Changelog.markdown view
@@ -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)
− cabal.project 
@@ -1,6 +0,0 @@-packages: .--source-repository-package-  type: git-  location: https://github.com/composewell/streamly.git-  tag: ac3af8749194f1788704dda8667d0b3807075cc2
curryer-rpc.cabal view
@@ -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
src/Network/RPC/Curryer/Client.hs view
@@ -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 ())
src/Network/RPC/Curryer/Server.hs view
@@ -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
src/Network/RPC/Curryer/StreamlyAdditions.hs view
@@ -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)