@@ -1,17 +1,20 @@
1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving #-}
2−{-# HLINT ignore "Use lambda-case" #-}
1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, CPP, ExistentialQuantification, StandaloneDeriving, GADTs, RecordWildCards #-}
2+{-# OPTIONS_GHC -fno-warn-orphans #-}
3+{- HLINT ignore "Use lambda-case" -}
34module Network.RPC.Curryer.Server where
4−import Streamly
55import qualified Streamly.Prelude as S
66import Streamly.Network.Socket
7−import Streamly.Internal.Network.Socket (handleWithM, writeChunk)
7+import Streamly.Internal.Network.Socket (handleWithM)
88import Network.Socket as Socket
9−import Streamly.Internal.Data.Parser.ParserD as PD
9+import Network.Socket.ByteString as Socket
10+import Streamly.Internal.Data.Parser as P hiding (concatMap)
1011import Codec.Winery
1112import Codec.Winery.Internal (varInt, decodeVarInt, getBytes)
1213import Codec.Winery.Class (mkExtractor)
1314import GHC.Generics
14−import Control.Concurrent.MVar (MVar)
15+import GHC.Fingerprint
16+import Data.Typeable
17+import Control.Concurrent.MVar (MVar, newMVar, withMVar)
1518import Control.Exception
1619import Data.Function ((&))
1720import Data.Word
@@ -20,52 +23,103 @@ import qualified Data.ByteString.Lazy as BSL
2023import qualified Data.ByteString.FastBuilder as BB
2124import Streamly.Data.Fold as FL
2225import qualified Streamly.Internal.Data.Stream.IsStream as S
23−import Data.Foldable
24−import Data.Bits
2526import qualified Data.Binary as B
2627import qualified Data.UUID as UUIDBase
27−import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA
28−import qualified Network.RPC.Curryer.StreamlyAdditions as SA
28+import qualified Data.UUID.V4 as UUIDBase
2929import Control.Monad
30+import Data.Functor
31+
32+import qualified Network.RPC.Curryer.StreamlyAdditions as SA
33+--import Control.Monad
3034import Data.Hashable
35+import System.Timeout
36+import qualified Network.ByteOrder as BO
37+
3138
3239-- for toArrayS conversion
3340import qualified Data.ByteString.Internal as BSI
34−import Foreign.ForeignPtr (plusForeignPtr)
41+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as Arr
3542import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
36−import GHC.Ptr (plusPtr)
43+import GHC.Ptr (minusPtr)
3744
45+-- define CURRYER_SHOW_BYTES 1
46+#if CURRYER_SHOW_BYTES == 1
3847import Debug.Trace
48+#endif
49+
50+traceBytes :: Applicative f => String -> BS.ByteString -> f ()
51+#if CURRYER_SHOW_BYTES == 1
52+traceBytes msg bs = traceShowM (msg, BS.length bs, bs)
53+#else
54+traceBytes _ _ = pure ()
55+#endif
56+
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
3972
40−data ClientHelloMessage = ClientHelloMessage Int Int
41− deriving Generic
42− deriving Serialise via WineryVariant ClientHelloMessage
43−
44−data ServerHelloMessage = ServerHelloMessage { serverServices :: [String],
45− serverInfo :: String
46− }
47− deriving Generic
48− deriving Serialise via WineryRecord ServerHelloMessage
49−
50−--request-response token
51−--client-side timeout
52−--server-to-client and client-to-server async request (no response requested)
53−
54−data Message a = Response UUID a
55− | AsyncRequest a
56− | ResponseExpectedRequest UUID a
57− -- | Services
58− | ExceptionResponse String
59− deriving (Generic, Show)
60− deriving Serialise via WineryVariant (Message a)
61−
62−-- | The response type
63−data HandlerResponse a = NoResponse
64− | HandlerResponse a
65− | HandlerException a
66− deriving Generic
67− deriving Serialise via WineryVariant (HandlerResponse a)
73+type BinaryMessage = BS.ByteString
6874
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)
84+
85+type TimeoutMicroseconds = Int
86+
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+
69123--avoid orphan instance
70124newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
71125 deriving (Show, Eq, B.Binary, Hashable)
@@ -84,12 +138,18 @@ instance Serialise UUID where
84138 x -> error $ "invalid schema element " <> show x
85139 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)
86140
87−
141+-- | Errors from remote calls.
88142data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps
89143 | TimeoutError
144+ | ExceptionError String
90145 deriving (Generic, Show, Eq)
91146 deriving Serialise via WineryVariant ConnectionError
92147
148+data TimeoutException = TimeoutException
149+ deriving Show
150+
151+instance Exception TimeoutException
152+
93153type HostAddr = (Word8, Word8, Word8, Word8)
94154
95155allHostAddrs,localHostAddr :: HostAddr
@@ -97,109 +157,186 @@ allHostAddrs = (0,0,0,0)
97157localHostAddr = (127,0,0,1)
98158
99159-- Each message is length-prefixed by a 32-bit unsigned length.
100−messageBoundaryP :: Parser IO Word8 BS.ByteString
101−messageBoundaryP = do
102− --traceShowM "message!"
103− w4x8 <- PD.take 4 FL.toList
104− --traceShowM ("w4x8",w4x8)
105− let c = fromIntegral (fromOctets w4x8)
106− --traceShowM ("c", c)
107− vals <- PD.take c FL.toList
108− let bytes = BS.pack vals
109− traceShowM ("vals", BS.length bytes, bytes)
110− pure bytes
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
198+ let s = FL.toList
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
111219
112220type NewConnectionHandler msg = IO (Maybe msg)
113221
114−type NewMessageHandler msg resp = msg -> IO (HandlerResponse resp)
115−
116−serve :: (Show msg, Serialise msg, Serialise resp) =>
117− NewConnectionHandler resp ->
118− NewMessageHandler msg 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 ->
119228 HostAddr ->
120229 PortNumber ->
121230 Maybe (MVar SockAddr) ->
122231 IO Bool
123−serve connhandler msghandler hostaddr port mSockLock = do
124− let handleSock sock = do
125− putStrLn "handleSock"
126− let messageHandler msg = do
127− putStrLn $ "GOT MSG " ++ show msg
128− case msg of
129− Response{} -> putStrLn "client sent response"
130− ResponseExpectedRequest requestID val -> do
131− putStrLn "ResponseExp"
132− resp <- msghandler val
133− case resp of
134− HandlerResponse responseVal -> sendMessage (Response requestID responseVal) sock
135− NoResponse -> error "attempt to return non-response to expected response message"
136− HandlerException _ -> error "TODO HandlerException"
137− AsyncRequest val -> do
138− putStrLn "AsyncReq"
139− void $ msghandler val
140− --no response necessaryxs
141− ExceptionResponse{} -> putStrLn "client sent exception response"
142− -- allow the server to send an async welcome message to the new client, if necessary
143− mResp <- connhandler
144− case mResp of
145− Nothing -> pure ()
146− Just resp -> sendMessage (AsyncRequest resp) sock
147− drainSocketMessages sock messageHandler
148− serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
232+serve userMsgHandlers serverState hostaddr port mSockLock = do
233+ let
234+ handleSock sock = do
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
149239 pure True
150240
151−--add callback to allow for responses via socket
152−type MessageHandler a = Message a -> IO ()
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
153249
154−type AsyncMessageHandler a = a -> IO ()
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
265+ -> IO ()
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 ()
155313
156−drainSocketMessages :: Serialise msg => Socket -> MessageHandler msg -> IO ()
157−drainSocketMessages sock msgHandler = do
158− let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
159− handler bs = do
160− --print ("drain handler", bs)
161− case deserialise bs of
162− Left err ->
163− print err
164− Right val -> do
165− --putStrLn "deserialized value for msgHandler"
166− msgHandler val --add response function
167− S.drain $ S.parseManyD messageBoundaryP sockStream & S.mapM handler
168−
169−fromOctets :: [Word8] -> Word32
170−fromOctets = foldl' accum 0
171− where
172− accum a o = (a `shiftL` 8) .|. fromIntegral o
173314
174−octets :: Word32 -> [Word8]
175−octets w =
176− [ fromIntegral (w `shiftR` 24)
177− , fromIntegral (w `shiftR` 16)
178− , fromIntegral (w `shiftR` 8)
179− , fromIntegral w
180− ]
315+type EnvelopeHandler = Envelope -> IO ()
316+
317+drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()
318+drainSocketMessages sock envelopeHandler = do
319+ let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
320+ S.drain $ S.serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler
181321
182322--send length-tagged bytestring, perhaps should be in network byte order?
183−sendMessage :: Serialise a => a -> Socket -> IO ()
184−sendMessage msg socket' = do
185− putStrLn ("sending bytes: " ++ show (blen <> bytes))
186− writeChunk socket' (toArrayS (blen <> bytes))
187− where
188− bytes = serialise msg
189− len = BS.length bytes
190− blen = BS.pack (octets (fromIntegral len))
191−
192−
193−--from streamly-bytestring
194−-- | Convert a 'ByteString' to an array of 'Word8'. This function unwraps the
195−-- 'ByteString' and wraps it with 'Array' constructors and hence the operation
196−-- is performed in constant time.
197−{-# INLINE toArrayS #-}
198−toArrayS :: BS.ByteString -> SA.Array Word8
199−--slow path
200−--toArrayS = SA.fromList . BS.unpack
201−toArrayS (BSI.PS fp off len) = SA.Array nfp endPtr
202− where
203− nfp = fp `plusForeignPtr` off
204− 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
205332
333+fingerprint :: Typeable a => a -> Fingerprint
334+fingerprint = typeRepFingerprint . typeOf
335+
336+fromArray :: Arr.Array Word8 -> BSI.ByteString
337+fromArray Arr.Array {..}
338+ | aLen == 0 = mempty
339+ | otherwise = BSI.PS aStart 0 aLen
340+ where
341+ aStartPtr = unsafeForeignPtrToPtr aStart
342+ aLen = aEnd `minusPtr` aStartPtr