@@ -1,4 +1,4 @@
1−{-# LANGUAGE DerivingVia, DeriveGeneric, RankNTypes, ScopedTypeVariables, MultiParamTypeClasses, OverloadedStrings, GeneralizedNewtypeDeriving, TypeApplications, CPP, ExistentialQuantification, StandaloneDeriving, GADTs #-}
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
@@ -25,7 +25,9 @@ import Streamly.Data.Fold as FL
2525import qualified Streamly.Internal.Data.Stream.IsStream as S
2626import qualified Data.Binary as B
2727import qualified Data.UUID as UUIDBase
28+import qualified Data.UUID.V4 as UUIDBase
2829import Control.Monad
30+import Data.Functor
2931
3032import qualified Network.RPC.Curryer.StreamlyAdditions as SA
3133--import Control.Monad
@@ -33,14 +35,14 @@ import Data.Hashable
3335import 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
@@ -85,44 +87,39 @@ type TimeoutMicroseconds = Int
8587deriving instance Generic Fingerprint
8688deriving via WineryVariant Fingerprint instance Serialise Fingerprint
8789
90+-- | Internal type used to mark envelope types.
8891data MessageType = RequestMessage TimeoutMicroseconds
8992 | ResponseMessage
9093 | TimeoutResponseMessage
94+ | ExceptionResponseMessage
9195 deriving (Generic, Show)
9296 deriving Serialise via WineryVariant MessageType
9397
94−type MessageHandlers = [RequestHandler]
95−
96−--newtype RequestHandler = RequestHandler { handler :: forall a b. (Serialise a, Serialise b) => (a -> IO b) }
98+-- | A list of `RequestHandler`s.
99+type RequestHandlers serverState = [RequestHandler serverState]
97100
98−data RequestHandler where
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
99103 -- | create a request handler with a response
100− RequestHandler :: forall a b. (Serialise a, Serialise b) => (a -> IO b) -> RequestHandler
104+ RequestHandler :: forall a b serverState. (Serialise a, Serialise b) => (ConnectionState serverState -> a -> IO b) -> RequestHandler serverState
101105 -- | create an asynchronous request handler where the client does not expect nor await a response
102− AsyncRequestHandler :: forall a. Serialise a => (a -> IO ()) -> RequestHandler
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
103122
104−{-
105−handleCall :: forall a b. (Serialise a, Serialise b)
106− => (a -> IO b)
107− -> RequestHandler
108−handleCall f = RequestHandler f
109−-}
110−{-data Message a = Response UUID a
111− | AsyncRequest UUID a
112− | ResponseExpectedRequest UUID (Maybe Timeout) a
113− | TimedOutResponse UUID
114− | ExceptionResponse UUID String
115− deriving (Generic, Show)
116− deriving Serialise via WineryVariant (Message a)
117−
118−-- | The response type
119−
120−data HandlerResponse a = NoResponse
121− | HandlerResponse a
122− | HandlerException String
123− deriving Generic
124− deriving Serialise via WineryVariant (HandlerResponse a)
125−-}
126123--avoid orphan instance
127124newtype UUID = UUID { _unUUID :: UUIDBase.UUID }
128125 deriving (Show, Eq, B.Binary, Hashable)
@@ -141,30 +138,39 @@ instance Serialise UUID where
141138 x -> error $ "invalid schema element " <> show x
142139 decodeCurrent = B.decode . BSL.fromStrict <$> (decodeVarInt >>= getBytes)
143140
144−
141+-- | Errors from remote calls.
145142data ConnectionError = CodecError String -- show of WineryException from exception initiator which cannot otherwise be transmitted over a line due to dependencies on TypeReps
146143 | TimeoutError
147144 | ExceptionError String
148145 deriving (Generic, Show, Eq)
149146 deriving Serialise via WineryVariant ConnectionError
150147
148+data TimeoutException = TimeoutException
149+ deriving Show
150+
151+instance Exception TimeoutException
152+
151153type HostAddr = (Word8, Word8, Word8, Word8)
152154
153155allHostAddrs,localHostAddr :: HostAddr
154156allHostAddrs = (0,0,0,0)
155157localHostAddr = (127,0,0,1)
156158
157−-- Each message is lxggength-prefixed by a 32-bit unsigned length.
159+-- Each message is length-prefixed by a 32-bit unsigned length.
158160envelopeP :: Parser IO Word8 Envelope
159161envelopeP = do
160− let s = FL.toList
161− msgTypeP = (P.satisfy (== 0) *>
162+ let msgTypeP = (P.satisfy (== 0) *>
162163 (RequestMessage . fromIntegral <$> word32P)) `P.alt`
163− (P.satisfy (== 1) *> pure ResponseMessage) `P.alt`
164− (P.satisfy (== 2) *> pure TimeoutResponseMessage)
164+ (P.satisfy (== 1) $> ResponseMessage) `P.alt`
165+ (P.satisfy (== 2) $> TimeoutResponseMessage) `P.alt`
166+ (P.satisfy (== 3) $> ExceptionResponseMessage)
165167 lenPrefixedByteStringP = do
166168 c <- fromIntegral <$> word32P
167− 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)
168174 Envelope <$> fingerprintP <*> msgTypeP <*> uuidP <*> lenPrefixedByteStringP
169175
170176encodeEnvelope :: Envelope -> BS.ByteString
@@ -176,17 +182,16 @@ encodeEnvelope (Envelope (Fingerprint fp1 fp2) msgType msgId bs) =
176182 RequestMessage timeoutms -> BS.singleton 0 <> BO.bytestring32 (fromIntegral timeoutms)
177183 ResponseMessage -> BS.singleton 1
178184 TimeoutResponseMessage -> BS.singleton 2
185+ ExceptionResponseMessage -> BS.singleton 3
179186 msgIdBs =
180187 case UUIDBase.toWords (_unUUID msgId) of
181− (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]
182189 msgLen = fromIntegral (BS.length bs)
183190 lenPrefixedBs = BO.bytestring32 msgLen <> bs
184191
185192fingerprintP :: Parser IO Word8 Fingerprint
186−fingerprintP = do
187− f1 <- word64P
188− f2 <- word64P
189− pure (Fingerprint f1 f2)
193+fingerprintP =
194+ Fingerprint <$> word64P <*> word64P
190195
191196word64P :: Parser IO Word8 Word64
192197word64P = do
@@ -207,135 +212,108 @@ uuidP = do
207212 u1 <- word32P
208213 u2 <- word32P
209214 u3 <- word32P
210− u4 <- word32P
211− 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
212219
213220type NewConnectionHandler msg = IO (Maybe msg)
214221
215222type NewMessageHandler req resp = req -> IO resp
216−
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.
217225serve ::
218− MessageHandlers ->
226+ RequestHandlers s->
227+ s ->
219228 HostAddr ->
220229 PortNumber ->
221230 Maybe (MVar SockAddr) ->
222231 IO Bool
223−serve userMsgHandlers hostaddr port mSockLock = do
232+serve userMsgHandlers serverState hostaddr port mSockLock = do
224233 let
225234 handleSock sock = do
226235 lockingSocket <- newLock sock
227− drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers)
236+ drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)
228237
229238 S.serially (S.unfold (SA.acceptOnAddrWith [(ReuseAddr,1)] mSockLock) (hostaddr, port)) & S.parallely . S.mapM (handleWithM handleSock) & S.drain
230239 pure True
231240
232−{-serverEnvelopeHandler :: Locking Socket -> MessageHandlers s -> Envelope -> IO ()
233−serverEnvelopeHandler sockLock (Envelope fprint RequestMessage msgId messagebs) = do
234− error "spam"
235−serverEnvelopeHandler sockLock (Envelope fprint ResponseMessage msgId messagebs) = do
236− error "spam2"
237−-}
238−
239241openEnvelope :: forall s. (Serialise s, Typeable s) => Envelope -> Maybe s
240−openEnvelope (Envelope eprint _ _ bytes) = do
241− {-let decoder = case getDecoder (schema (Proxy :: Proxy s)) of
242− Left err -> error (show err)
243− Right dec -> dec-}
242+openEnvelope (Envelope eprint _ _ bytes) =
244243 if eprint == fingerprint (undefined :: s) then
245244 case deserialise bytes of
246245 Left err -> error (show err)
247246 Right v -> Just v
248− --Just (evalDecoder decoder bytes)
249247 else
250248 Nothing
251249
252−matchEnvelope :: forall a b. (Serialise a, Serialise b, Typeable b) =>
250+matchEnvelope :: forall a b s. (Serialise a, Serialise b, Typeable b) =>
253251 Envelope ->
254− (a -> IO b) ->
255− Maybe (a -> IO b, a)
252+ (ConnectionState s -> a -> IO b) ->
253+ Maybe (ConnectionState s -> a -> IO b, a)
256254matchEnvelope envelope dispatchf =
257255 case openEnvelope envelope :: Maybe a of
258256 Nothing -> Nothing
259257 Just decoded -> Just (dispatchf, decoded)
260258
259+-- | Called by `serve` to process incoming envelope requests. Never returns, so use `async` to spin it off on another thread.
261260serverEnvelopeHandler ::
262261 Locking Socket
263− -> MessageHandlers
262+ -> RequestHandlers s
263+ -> s
264264 -> Envelope
265265 -> IO ()
266−serverEnvelopeHandler _ _ (Envelope _ TimeoutResponseMessage _ _) = pure ()
267−serverEnvelopeHandler sockLock msgHandlers envelope@(Envelope _ (RequestMessage timeoutms) 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
268270 --find first matching handler
269271 let runTimeout :: IO b -> IO (Maybe b)
270− runTimeout m =
272+ runTimeout m =
271273 if timeoutms == 0 then
272− Just <$> m
273− else
274− timeout (fromIntegral timeoutms) m
275− firstMatcher (RequestHandler (msghandler :: a -> IO b)) Nothing =
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 =
276287 case matchEnvelope envelope msghandler of
277288 Nothing -> pure Nothing
278− Just (dispatchf, decoded) ->
289+ Just (dispatchf, decoded) -> do
279290 --TODO add exception handling
280− do
281− mResponse <- runTimeout (dispatchf decoded)
282− let envelopeResponse =
283− case mResponse of
284− Just response ->
285− Envelope (fingerprint response) ResponseMessage msgId (serialise response)
286− Nothing -> Envelope (fingerprint TimeoutError) (TimeoutResponseMessage) msgId (BS.empty)
287− sendEnvelope envelopeResponse sockLock
288− pure (Just ())
289− firstMatcher (AsyncRequestHandler (msghandler :: a -> IO ())) Nothing =
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 =
290301 case matchEnvelope envelope msghandler of
291302 Nothing -> pure Nothing
292303 Just (dispatchf, decoded) -> do
293− _ <- dispatchf decoded
304+ _ <- dispatchf sState decoded
294305 pure (Just ())
295306 firstMatcher _ acc = pure acc
296− foldM_ (flip firstMatcher) Nothing msgHandlers
297−serverEnvelopeHandler _ _ (Envelope _ ResponseMessage _ _) = error "server received response message"
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 ()
298313
299314
300−{-
301−
302−
303− --putStrLn $ "GOT MSG " ++ show msg
304− let --runTimeout :: IO (HandlerResponse resp) -> IO (Maybe (HandlerResponse resp))
305− runTimeout mTimeout m = case mTimeout of
306− Nothing -> Just <$> m
307− Just timeoutMicroseconds -> timeout timeoutMicroseconds m
308− case msg of
309− Response{} -> putStrLn "client sent response"
310− ResponseExpectedRequest requestID mTimeout val -> do
311− let normalResponder = do
312− resp <- runTimeout mTimeout (requestMessageHandler val)
313− case resp of
314− Just (HandlerResponse responseVal) -> sendMessage (Response requestID responseVal) sockLock
315− Just NoResponse -> error "attempt to return non-response to expected response message"
316− Just (HandlerException exc) -> sendMessage (ExceptionResponse @(Message resp) requestID exc) sockLock
317− Nothing ->
318− sendMessage (TimedOutResponse @(Message resp) requestID) sockLock
319− excHandler :: SomeException -> IO ()
320− excHandler e = do
321− --send exception to client
322− sendMessage (ExceptionResponse @(Message resp) requestID (show e)) sockLock
323− throwIO e
324− catch normalResponder excHandler
325− AsyncRequest _ val ->
326− void $ requestMessageHandler val
327− --no response necessary
328− ExceptionResponse{} -> putStrLn "client sent exception response"
329− TimedOutResponse{} -> putStrLn "client sent timed out response"
330−-}
331−
332−--add callback to allow for responses via socket
333−--type MessageHandler a = Message a -> IO ()
334−
335315type EnvelopeHandler = Envelope -> IO ()
336316
337−type AsyncMessageHandler a = a -> IO ()
338−
339317drainSocketMessages :: Socket -> EnvelopeHandler -> IO ()
340318drainSocketMessages sock envelopeHandler = do
341319 let sockStream = S.unfold readWithBufferOf (1024 * 4, sock)
@@ -349,8 +327,16 @@ sendEnvelope envelope sockLock = do
349327 --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
350328 withLock sockLock $ \socket' ->
351329 Socket.sendAll socket' fullbytes
330+ --traceShowM ("sendEnvelope", envelope)
352331 traceBytes "sendEnvelope" fullbytes
353332
354333fingerprint :: Typeable a => a -> Fingerprint
355334fingerprint = typeRepFingerprint . typeOf
356335
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