@@ -1,18 +1,20 @@
1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications, CPP #-}
1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, CPP, ExistentialQuantification, StandaloneDeriving, GADTs, RecordWildCards #-}
2+{-# OPTIONS_GHC -fno-warn-orphans #-}
23{- HLINT ignore "Use lambda-case" -}
34module Network.RPC.Curryer.Server where
4−import Streamly
55import qualified Streamly.Prelude as S
66import Streamly.Network.Socket
77import Streamly.Internal.Network.Socket (handleWithM)
88import Network.Socket as Socket
99import Network.Socket.ByteString as Socket
10−import Streamly.Internal.Data.Parser as P
10+import Streamly.Internal.Data.Parser as P hiding (concatMap)
1111import Codec.Winery
1212import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)
1313import Codec.Winery.Class (mkExtractor)
1414import GHC.Generics
15−import Control.Concurrent.MVar (MVar)
15+import GHC.Fingerprint
16+import Data.Typeable
17+import Control.Concurrent.MVar (MVar, newMVar, withMVar)
1618import Control.Exception
1719import Data.Function ((&))
1820import Data.Word
@@ -24,21 +26,23 @@ import qualified Streamly.Internal.Data.Stream.IsStream as S
2426import qualified Data.Binary as B
2527import qualified Data.UUID as UUIDBase
2628import qualified Data.UUID.V4 as UUIDBase
29+import Control.Monad
30+import Data.Functor
2731
2832import qualified Network.RPC.Curryer.StreamlyAdditions as SA
29−import Control.Monad
33+--import Control.Monad
3034import Data.Hashable
3135import System.Timeout
3236import qualified Network.ByteOrder as BO
33−import Data.Proxy
37+
3438
3539-- for toArrayS conversion
36−{-import qualified Data.ByteString.Internal as BSI
37−import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA
38−import Foreign.ForeignPtr (plusForeignPtr)
40+import qualified Data.ByteString.Internal as BSI
41+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as Arr
3942import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
40−import GHC.Ptr (plusPtr)
41−-}
43+import GHC.Ptr (minusPtr)
44+
45+-- define CURRYER_SHOW_BYTES 1
4246#if CURRYER_SHOW_BYTES == 1
4347import Debug.Trace
4448#endif
@@ -50,23 +54,72 @@ traceBytes msg bs = traceShowM (msg, BS.length bs, bs)
5054traceBytes _ _ = pure ()
5155#endif
5256
53−type Timeout = Int
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
72+
73+type BinaryMessage = BS.ByteString
5474
55−data Message a = Response UUID a
56− | AsyncRequest UUID a
57− | ResponseExpectedRequest UUID (Maybe Timeout) a
58− | TimedOutResponse UUID
59− | ExceptionResponse UUID String
60− deriving (Generic, Show)
61− deriving Serialise via WineryVariant (Message a)
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)
6284
63−-- | The response type
64−data HandlerResponse a = NoResponse
65− | HandlerResponse a
66− | HandlerException String
67− deriving Generic
68− deriving Serialise via WineryVariant (HandlerResponse a)
85+type TimeoutMicroseconds = Int
6986
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+
70123--avoid orphan instance
71124newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
72125 deriving (Show, Eq, B.Binary, Hashable)
@@ -85,13 +138,18 @@ instance Serialise UUID where
85138 x -> error $ "invalid schema element " <> show x
86139 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)
87140
88−
141+-- | Errors from remote calls.
89142data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps
90143 | TimeoutError
91144 | ExceptionError String
92145 deriving (Generic, Show, Eq)
93146 deriving Serialise via WineryVariant ConnectionError
94147
148+data TimeoutException = TimeoutException
149+ deriving Show
150+
151+instance Exception TimeoutException
152+
95153type HostAddr = (Word8, Word8, Word8, Word8)
96154
97155allHostAddrs,localHostAddr :: HostAddr
@@ -99,129 +157,186 @@ allHostAddrs = (0,0,0,0)
99157localHostAddr = (127,0,0,1)
100158
101159-- Each message is length-prefixed by a 32-bit unsigned length.
102−messageBoundaryP :: Parser IO Word8 BS.ByteString
103−messageBoundaryP = do
160+envelopeP :: Parser IO Word8 Envelope
161+envelopeP = do
162+ let msgTypeP = (P.satisfy (== 0) *>
163+ (RequestMessage . fromIntegral <$> word32P)) `P.alt`
164+ (P.satisfy (== 1) $> ResponseMessage) `P.alt`
165+ (P.satisfy (== 2) $> TimeoutResponseMessage) `P.alt`
166+ (P.satisfy (== 3) $> ExceptionResponseMessage)
167+ lenPrefixedByteStringP = do
168+ c <- fromIntegral <$> word32P
169+ --streamly can't handle takeEQ 0, so add special handling
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
104198 let s = FL.toList
105− w4x8 <- P.take 4 s
106− --traceShowM ("w4x8"::String, w4x8)
107− let c = fromIntegral (BO.word32 (BS.pack w4x8))
108− --traceShowM ("c"::String, c)
109− vals <- P.take c s
110− let bytes = BS.pack vals
111− --traceShowM ("parsedBytes"::String, c, BS.length bytes, bytes)
112− pure bytes
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
113219
114220type NewConnectionHandler msg = IO (Maybe msg)
115221
116−type NewMessageHandler req resp = req -> IO (HandlerResponse resp)
117−
118−serve :: forall req resp. (Show req, Serialise req, Serialise resp) =>
119− NewConnectionHandler resp ->
120− NewMessageHandler req resp ->
222+type NewMessageHandler req resp = req -> IO resp
223+
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.
225+serve ::
226+ RequestHandlers s->
227+ s ->
121228 HostAddr ->
122229 PortNumber ->
123230 Maybe (MVar SockAddr) ->
124231 IO Bool
125−serve connhandler userMsgHandler hostaddr port mSockLock = do
232+serve userMsgHandlers serverState hostaddr port mSockLock = do
126233 let
127− decoder :: NewMessageHandler req resp -> Decoder (Message req)
128− decoder _ =
129− case getDecoder (schema (Proxy @(Message req))) of
130− Left exc -> error (show exc)
131− Right dec -> dec
132234 handleSock sock = do
133− mResp <- connhandler
134− case mResp of
135− Nothing -> pure ()
136− Just resp -> do
137− requestID <- UUID <$> UUIDBase.nextRandom
138− sendMessage (AsyncRequest requestID resp) sock
139− drainSocketMessages sock (decoder userMsgHandler) (serverMessageHandler sock userMsgHandler)
140− serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
235+ lockingSocket <- newLock sock
236+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)
237+
238+ S.serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.parallely . S.mapM (handleWithM handleSock) & S.drain
141239 pure True
142240
143−serverMessageHandler :: forall req resp. (Show req,
144− Serialise req,
145− Serialise resp)
146− => Socket
147− -> NewMessageHandler req resp
148− -> Message req
241+openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s
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
249+
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
149265 -> IO ()
150−serverMessageHandler sock requestMessageHandler msg = do
151− --putStrLn $ "GOT MSG " ++ show msg
152− let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))
153− runTimeout mTimeout m = case mTimeout of
154− Nothing -> Just <$> m
155− Just timeoutMicroseconds -> timeout timeoutMicroseconds m
156− case msg of
157− Response{} -> putStrLn "client sent response"
158− ResponseExpectedRequest requestID mTimeout val -> do
159− let normalResponder = do
160− resp <- runTimeout mTimeout (requestMessageHandler val)
161− case resp of
162− Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sock
163− Just NoResponse -> error "attempt to return non-response to expected response message"
164− Just (HandlerException exc) -> sendMessage (ExceptionResponse @(Message resp) requestID exc) sock
165− Nothing ->
166− sendMessage (TimedOutResponse @(Message resp) requestID) sock
167− excHandler :: SomeException -> IO ()
168− excHandler e = do
169− --send exception to client
170− sendMessage (ExceptionResponse @(Message resp) requestID (show e)) sock
171− throwIO e
172− catch normalResponder excHandler
173− AsyncRequest _ val -> do
174− void $ requestMessageHandler val
175− --no response necessary
176− ExceptionResponse{} -> putStrLn "client sent exception response"
177− TimedOutResponse{} -> putStrLn "client sent timed out response"
178−
179−--add callback to allow for responses via socket
180−type MessageHandler a = Message a -> IO ()
181−
182−type AsyncMessageHandler a = a -> IO ()
183−
184−
185−
186−drainSocketMessages :: Serialise msg => Socket -> Decoder (Message msg) -> MessageHandler msg -> IO ()
187−drainSocketMessages sock decoder msgHandler = do
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 ()
313+
314+
315+type EnvelopeHandler = Envelope -> IO ()
316+
317+drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()
318+drainSocketMessages sock envelopeHandler = do
188319 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
189− handler bs = do
190− let decoded = evalDecoder decoder bs
191− msgHandler decoded
192− S.drain $ serially $ S.parseMany messageBoundaryP sockStream & S.mapM handler
320+ S.drain $ S.serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler
193321
194322--send length-tagged bytestring, perhaps should be in network byte order?
195−sendMessage :: Serialise a => Message a -> Socket -> IO ()
196−sendMessage msg socket' = do
197− let
198− msgbytes = serialiseOnly msg
199− fullbytes = lenbytes <> msgbytes
200− len = BS.length msgbytes
201− lenbytes = BO.bytestring32 (fromIntegral len)
202−
203− byteCount <- Socket.send socket' fullbytes
204− when (byteCount /= BS.length fullbytes) (error "bytes sent mismatch")
205− traceBytes "sent bytes" fullbytes
206−
207−
208−
209−--from streamly-bytestring
210−-- | Convert a 'ByteString' to an array of 'Word8'. This function unwraps the
211−-- 'ByteString' and wraps it with 'Array' constructors and hence the operation
212−-- is performed in constant time.
213−{-
214−{-# INLINE toArrayS #-}
215−toArrayS :: BS.ByteString -> SA.Array Word8
216−--slow path
217−--toArrayS = SA.fromList . BS.unpack
218−toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr
219− where
220− nfp = fp `plusForeignPtr` off
221− endPtr = unsafeForeignPtrToPtr nfp `plusPtr` len `plusPtr` 1 -- ? why +1?
323+sendEnvelope :: Envelope -> Locking Socket -> IO ()
324+sendEnvelope envelope sockLock = do
325+ let envelopebytes = encodeEnvelope envelope
326+ fullbytes = envelopebytes
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
328+ withLock sockLock $ \socket' ->
329+ Socket.sendAll socket' fullbytes
330+ --traceShowM ("sendEnvelope", envelope)
331+ traceBytes "sendEnvelope" fullbytes
332+
333+fingerprint :: Typeable a => a -> Fingerprint
334+fingerprint = typeRepFingerprint . typeOf
222335
223−toArraySlow :: BS.ByteString -> SA.Array Word8
224−toArraySlow bs = arr
336+fromArray :: Arr.Array Word8 -> BSI.ByteString
337+fromArray Arr.Array {..}
338+ | aLen == 0 = mempty
339+ | otherwise = BSI.PS aStart 0 aLen
225340 where
226− arr = SA.fromList (BS.unpack bs)
227−-}
341+ aStartPtr = unsafeForeignPtrToPtr aStart
342+ aLen = aEnd `minusPtr` aStartPtr