@@ -1,8 +1,7 @@
1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications, CPP, ExistentialQuantification, StandaloneDeriving #-}
1+{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, CPP, ExistentialQuantification, StandaloneDeriving, GADTs, RecordWildCards #-}
22{-# OPTIONS_GHC -fno-warn-orphans #-}
33{- HLINT ignore "Use lambda-case" -}
44module Network.RPC.Curryer.Server where
5−import Streamly
65import qualified Streamly.Prelude as S
76import Streamly.Network.Socket
87import Streamly.Internal.Network.Socket (handleWithM)
@@ -26,21 +25,24 @@ import Streamly.Data.Fold as FL
2625import qualified Streamly.Internal.Data.Stream.IsStream as S
2726import qualified Data.Binary as B
2827import qualified Data.UUID as UUIDBase
28+import qualified Data.UUID.V4 as UUIDBase
29+import Control.Monad
30+import Data.Functor
2931
3032import qualified Network.RPC.Curryer.StreamlyAdditions as SA
3133--import Control.Monad
3234import Data.Hashable
33−--import System.Timeout
35+import System.Timeout
3436import qualified Network.ByteOrder as BO
3537
38+
3639-- for toArrayS conversion
37−{-import qualified Data.ByteString.Internal as BSI
38−import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as SA
39−import Foreign.ForeignPtr (plusForeignPtr)
40+import qualified Data.ByteString.Internal as BSI
41+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as Arr
4042import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
41−import GHC.Ptr (plusPtr)
42−-}
43−-- #define CURRYER_SHOW_BYTES 1
43+import GHC.Ptr (minusPtr)
44+
45+-- define CURRYER_SHOW_BYTES 1
4446#if CURRYER_SHOW_BYTES == 1
4547import Debug.Trace
4648#endif
@@ -66,45 +68,58 @@ withLock (Locking mvar v) m =
6668lockless :: Locking a -> a
6769lockless (Locking _ a) = a
6870
69−type Timeout = Int
71+type Timeout = Word32
7072
7173type BinaryMessage = BS.ByteString
7274
7375--includes the fingerprint of the incoming data type (held in the BinaryMessage) to determine how to dispatch the message.
7476--add another envelope type for unencoded binary messages for any easy optimization for in-process communication
75−data Envelope = Envelope !Fingerprint !MessageType !UUID !BinaryMessage
77+data Envelope = Envelope {
78+ envFingerprint :: !Fingerprint,
79+ envMessageType :: !MessageType,
80+ envMsgId :: !UUID,
81+ envPayload :: !BinaryMessage
82+ }
7683 deriving (Generic, Show)
77− deriving Serialise via WineryVariant Envelope
84+
85+type TimeoutMicroseconds = Int
7886
7987deriving instance Generic Fingerprint
8088deriving via WineryVariant Fingerprint instance Serialise Fingerprint
8189
82−data MessageType = RequestMessage
90+-- | Internal type used to mark envelope types.
91+data MessageType = RequestMessage TimeoutMicroseconds
8392 | ResponseMessage
93+ | TimeoutResponseMessage
94+ | ExceptionResponseMessage
8495 deriving (Generic, Show)
8596 deriving Serialise via WineryVariant MessageType
8697
87−data MessageHandlers =
88− forall a b. (Show a, Serialise a, Serialise b) => MessageHandlers [RequestHandler a b]
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
89122
90−type RequestHandler a b = a -> IO b
91−
92−{-data Message a = Response UUID a
93− | AsyncRequest UUID a
94− | ResponseExpectedRequest UUID (Maybe Timeout) a
95− | TimedOutResponse UUID
96− | ExceptionResponse UUID String
97− deriving (Generic, Show)
98− deriving Serialise via WineryVariant (Message a)
99−
100−-- | The response type
101−
102−data HandlerResponse a = NoResponse
103− | HandlerResponse a
104− | HandlerException String
105− deriving Generic
106− deriving Serialise via WineryVariant (HandlerResponse a)
107−-}
108123--avoid orphan instance
109124newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
110125 deriving (Show, Eq, B.Binary, Hashable)
@@ -123,13 +138,18 @@ instance Serialise UUID where
123138 x -> error $ "invalid schema element " <> show x
124139 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)
125140
126−
141+-- | Errors from remote calls.
127142data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps
128143 | TimeoutError
129144 | ExceptionError String
130145 deriving (Generic, Show, Eq)
131146 deriving Serialise via WineryVariant ConnectionError
132147
148+data TimeoutException = TimeoutException
149+ deriving Show
150+
151+instance Exception TimeoutException
152+
133153type HostAddr = (Word8, Word8, Word8, Word8)
134154
135155allHostAddrs,localHostAddr :: HostAddr
@@ -139,12 +159,18 @@ localHostAddr = (127,0,0,1)
139159-- Each message is length-prefixed by a 32-bit unsigned length.
140160envelopeP :: Parser IO Word8 Envelope
141161envelopeP = do
142− let s = FL.toList
143− msgTypeP = (P.satisfy (== 0) *> pure RequestMessage) `P.alt`
144− (P.satisfy (== 1) *> pure ResponseMessage)
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)
145167 lenPrefixedByteStringP = do
146168 c <- fromIntegral <$> word32P
147− BS.pack <$> P.takeEQ c s
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)
148174 Envelope <$> fingerprintP <*> msgTypeP <*> uuidP <*> lenPrefixedByteStringP
149175
150176encodeEnvelope :: Envelope -> BS.ByteString
@@ -153,19 +179,19 @@ encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =
153179 where
154180 fingerprintBs = BO.bytestring64 fp1 <> BO.bytestring64 fp2
155181 msgTypeBs = case msgType of
156− RequestMessage -> BS.singleton 0
182+ RequestMessage timeoutms -> BS.singleton 0 <> BO.bytestring32 (fromIntegral timeoutms)
157183 ResponseMessage -> BS.singleton 1
184+ TimeoutResponseMessage -> BS.singleton 2
185+ ExceptionResponseMessage -> BS.singleton 3
158186 msgIdBs =
159187 case UUIDBase.toWords (_unUUID msgId) of
160− (u1, u2, u3, u4) -> foldr (<>) BS.empty (map BO.bytestring32 [u1, u2, u3, u4])
188+ (u1, u2, u3, u4) -> foldr ((<>) . BO.bytestring32) BS.empty [u1, u2, u3, u4]
161189 msgLen = fromIntegral (BS.length bs)
162190 lenPrefixedBs = BO.bytestring32 msgLen <> bs
163191
164192fingerprintP :: Parser IO Word8 Fingerprint
165−fingerprintP = do
166− f1 <- word64P
167− f2 <- word64P
168− pure (Fingerprint f1 f2)
193+fingerprintP =
194+ Fingerprint <$> word64P <*> word64P
169195
170196word64P :: Parser IO Word8 Word64
171197word64P = do
@@ -186,112 +212,112 @@ uuidP = do
186212 u1 <- word32P
187213 u2 <- word32P
188214 u3 <- word32P
189− u4 <- word32P
190− pure (UUID (UUIDBase.fromWords u1 u2 u3 u4))
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
191219
192220type NewConnectionHandler msg = IO (Maybe msg)
193221
194222type NewMessageHandler req resp = req -> IO resp
195−
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.
196225serve ::
197− MessageHandlers ->
226+ RequestHandlers s->
227+ s ->
198228 HostAddr ->
199229 PortNumber ->
200230 Maybe (MVar SockAddr) ->
201231 IO Bool
202−serve userMsgHandler hostaddr port mSockLock = do
232+serve userMsgHandlers serverState hostaddr port mSockLock = do
203233 let
204234 handleSock sock = do
205235 lockingSocket <- newLock sock
206− drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandler)
236+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)
207237
208− serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & parallely . S.mapM (handleWithM handleSock) & S.drain
238+ S.serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.parallely . S.mapM (handleWithM handleSock) & S.drain
209239 pure True
210240
211−{-serverEnvelopeHandler :: Locking Socket -> MessageHandlers s -> Envelope -> IO ()
212−serverEnvelopeHandler sockLock (Envelope fprint RequestMessage msgId messagebs) = do
213− error "spam"
214−serverEnvelopeHandler sockLock (Envelope fprint ResponseMessage msgId messagebs) = do
215− error "spam2"
216−-}
217−
218241openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s
219−openEnvelope (Envelope eprint _ _ bytes) = do
220− {-let decoder = case getDecoder (schema (Proxy :: Proxy s)) of
221− Left err -> error (show err)
222− Right dec -> dec-}
242+openEnvelope (Envelope eprint _ _ bytes) =
223243 if eprint == fingerprint (undefined :: s) then
224244 case deserialise bytes of
225245 Left err -> error (show err)
226246 Right v -> Just v
227− --Just (evalDecoder decoder bytes)
228247 else
229248 Nothing
230249
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.
231260serverEnvelopeHandler ::
232261 Locking Socket
233− -> MessageHandlers
262+ -> RequestHandlers s
263+ -> s
234264 -> Envelope
235265 -> IO ()
236−serverEnvelopeHandler sockLock (MessageHandlers msgHandlers) envelope@(Envelope _ RequestMessage msgId _) = 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
237270 --find first matching handler
238− let firstMatch = foldr firstMatcher Nothing msgHandlers
239− firstMatcher :: forall a b. (Serialise a, Serialise b) => RequestHandler a b -> Maybe (RequestHandler a b, a) -> Maybe (RequestHandler a b, a)
240− firstMatcher msghandler Nothing =
241− case openEnvelope envelope :: Maybe a of
242− Just decoded -> Just (msghandler, decoded)
243− Nothing -> Nothing
244− firstMatcher _ acc = acc
245− case firstMatch of
246− Nothing -> error "failed to handle msg"
247− Just (msghandler, decoded) -> do
248− --TODO add exception handling
249− response <- msghandler decoded
250− sendEnvelope (Envelope (fingerprint response) ResponseMessage msgId (serialise response)) sockLock
251−serverEnvelopeHandler _ _ (Envelope _ ResponseMessage _ _) = error "server received response message"
252−{-
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
253280
254−
255− --putStrLn $ "GOT MSG " ++ show msg
256− let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))
257− runTimeout mTimeout m = case mTimeout of
258− Nothing -> Just <$> m
259− Just timeoutMicroseconds -> timeout timeoutMicroseconds m
260− case msg of
261− Response{} -> putStrLn "client sent response"
262− ResponseExpectedRequest requestID mTimeout val -> do
263− let normalResponder = do
264− resp <- runTimeout mTimeout (requestMessageHandler val)
265− case resp of
266− Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sockLock
267− Just NoResponse -> error "attempt to return non-response to expected response message"
268− Just (HandlerException exc) -> sendMessage (ExceptionResponse @(Message resp) requestID exc) sockLock
269− Nothing ->
270− sendMessage (TimedOutResponse @(Message resp) requestID) sockLock
271− excHandler :: SomeException -> IO ()
272− excHandler e = do
273− --send exception to client
274− sendMessage (ExceptionResponse @(Message resp) requestID (show e)) sockLock
275− throwIO e
276− catch normalResponder excHandler
277− AsyncRequest _ val ->
278− void $ requestMessageHandler val
279− --no response necessary
280− ExceptionResponse{} -> putStrLn "client sent exception response"
281− TimedOutResponse{} -> putStrLn "client sent timed out response"
282−-}
283−
284−--add callback to allow for responses via socket
285−--type MessageHandler a = Message a -> IO ()
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 ()
286313
287−type EnvelopeHandler = Envelope -> IO ()
288314
289−type AsyncMessageHandler a = a -> IO ()
315+type EnvelopeHandler = Envelope -> IO ()
290316
291317drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()
292318drainSocketMessages sock envelopeHandler = do
293319 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
294− S.drain $ serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler
320+ S.drain $ S.serially $ S.parseMany envelopeP sockStream & S.mapM envelopeHandler
295321
296322--send length-tagged bytestring, perhaps should be in network byte order?
297323sendEnvelope :: Envelope -> Locking Socket -> IO ()
@@ -301,7 +327,16 @@ sendEnvelope envelope sockLock = do
301327 --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
302328 withLock sockLock $ \socket' ->
303329 Socket.sendAll socket' fullbytes
330+ --traceShowM ("sendEnvelope", envelope)
304331 traceBytes "sendEnvelope" fullbytes
305332
306333fingerprint :: Typeable a => a -> Fingerprint
307334fingerprint = 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