diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -104,8 +104,13 @@ move in the "looser direction" (e.g., more concurrency) should be applied _after_ ACK-ing the SETTINGS frame. -For simplicity of implementation, the current design apply SETTINGS: -- (client prefs) before sending SETTINGS, irrespective of ACK -- we should change this +The current design apply SETTINGS: +- (client prefs) after receiving a ACK for sent SETTINGS, you get the choice to + wait for an ACK or wait in a thread, but you must wait for an ACK to apply + changed settings (the `_settings` function will return an IO to wait for the + ACK and apply settings). Note that the initial SETTINGS change frame is + waited for in a thread without library's user intervention (if you feel + strongly against this choice, please open a bug). - (server prefs) immediately after receiving and hence before sending ACK-SETTINGS Fortunately, changing settings mid-stream is probably a rare behavior and the @@ -126,8 +131,8 @@ A number of HTTP2 features are currently hardcoded: - PINGs are replied-to immediately (i.e., a server could hog a connection with PINGs) -- received SETTINGS are acknowledged immediately after being taken into account -- sent SETTINGS are implemented immediately before sending, and not after acking +- the initial SETTINGS frame sent to the server is waited-for in a separate + thread, settings are applied to the connection when the server ACKs the frame - flow-control from DATA frames is decremented immediately when received (in a separate thread) rather than when consumed from the client - similarly, flow-control re-increment every DATA received as soon as it is @@ -160,8 +165,6 @@ I think the fundamental are right but the following needs tweaking: -- onPushPromise handlers will be requested when opening new streams rather than - when opening new connections - function to reset a stream will likely be blocking until a RST/EndStream is received so that all DATA frames are accounted for in the flow-control system - need a way to hook custom flow-control algorithms @@ -186,11 +189,7 @@ ### Various TODO -- modify client SETTINGS only when acknowledged - * currently make use of new client settings before the server has time to - know them, this can lead to errors due to inconsistent view of settings - between sender/receiver -- consider a beter frame-subscription mechanism than broadcast wake-up +- consider a better frame-subscription mechanism than broadcast wake-up * current system of dupChan everything is prone to errors and may be costly CPU/latency-wise if the concurrency is high - consider most performant functions for HTTP2.HPACK encoding/decoding diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -111,9 +111,9 @@ , (":authority", ByteString.pack _host) ] <> _extraHeaders - let onPushPromise _ stream streamFlowControl _ = void $ forkIO $ do + let ppHandler _ stream streamFlowControl _ = void $ forkIO $ do timePrint ("push stream started" :: String) - waitStream stream streamFlowControl >>= timePrint + waitStream stream streamFlowControl >>= timePrint . fromStreamResult timePrint ("push stream ended" :: String) let conf = [ (HTTP2.SettingsMaxFrameSize, _settingsMaxFrameSize) @@ -125,7 +125,7 @@ ] timePrint conf - conn <- newHttp2Client _host _port _encoderBufsize _decoderBufsize tlsParams conf onPushPromise + conn <- newHttp2Client _host _port _encoderBufsize _decoderBufsize tlsParams conf _addCredit (_incomingFlowControl conn) _initialWindowKick _ <- forkIO $ forever $ do updated <- _updateWindow $ _incomingFlowControl conn @@ -134,7 +134,7 @@ _ <- forkIO $ when (_interPingDelay > 0) $ forever $ do threadDelay _interPingDelay - (t0, t1, pingReply) <- ping _pingTimeout "pingpong" conn + (t0, t1, pingReply) <- ping conn _pingTimeout "pingpong" timePrint $ ("ping-reply:" :: String, pingReply, diffUTCTime t1 t0) let go 0 idx = timePrint $ "done worker: " <> show idx @@ -142,8 +142,9 @@ _ <- (_startStream conn $ \stream -> let initStream = _headers stream headersPairs (HTTP2.setEndStream) handler streamFlowControl _ = do + _ <- async $ onPushPromise stream ppHandler timePrint $ "stream started " <> show (idx, n) - waitStream stream streamFlowControl >>= timePrint + waitStream stream streamFlowControl >>= timePrint . fromStreamResult timePrint $ "stream ended " <> show (idx, n) in StreamDefinition initStream handler) go (n - 1) idx diff --git a/http2-client.cabal b/http2-client.cabal --- a/http2-client.cabal +++ b/http2-client.cabal @@ -1,7 +1,7 @@ name: http2-client -version: 0.1.0.1 +version: 0.2.0.0 synopsis: A native HTTP2 client library. -description: Please read the README.ms at the homepage. +description: Please read the README.md at the homepage. homepage: https://github.com/lucasdicioccio/http2-client license: BSD3 license-file: LICENSE diff --git a/src/Network/HTTP2/Client.hs b/src/Network/HTTP2/Client.hs --- a/src/Network/HTTP2/Client.hs +++ b/src/Network/HTTP2/Client.hs @@ -10,6 +10,7 @@ -- * Starting streams , StreamDefinition(..) , StreamStarter + , TooMuchConcurrency(..) , StreamThread , Http2Stream(..) -- * Sending data for POSTs. @@ -30,7 +31,7 @@ import Control.Concurrent.MVar (newMVar, takeMVar, putMVar) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.Chan (Chan, newChan, dupChan, readChan, writeChan) -import Control.Monad (forever, when, forM_) +import Control.Monad (forever, void, when, forM_) import Data.ByteString (ByteString) import qualified Data.ByteString as ByteString import Data.IORef (IORef, newIORef, atomicModifyIORef', readIORef) @@ -139,8 +140,11 @@ -- first call to this IO will return if a reply is received. Hence we -- recommend wrapping this IO in an Async (e.g., with @race (threadDelay -- timeout)@.) - , _settings :: SettingsList -> IO () - -- ^ Sends a SETTINGS. + , _settings :: SettingsList -> IO (IO (FrameHeader, FramePayload)) + -- ^ Sends a SETTINGS. Returns an IO to wait for a settings reply. No timeout + -- is provided. Only the first call to this IO will return if a reply is + -- received. Hence we recommend wrapping this IO in an Async (e.g., with + -- @race (threadDelay timeout)@.) , _goaway :: ErrorCodeId -> ByteString -> IO () -- ^ Sends a GOAWAY. , _startStream :: forall a. StreamStarter a @@ -202,6 +206,7 @@ -- and hence does not respect the RFC if sending large blocks. Use 'sendData' -- to chunk and send naively according to server\'s preferences. This function -- can be useful if you intend to handle the framing yourself. + , _waitPushPromise :: PushPromiseHandler -> IO () } -- | Handler upon receiving a PUSH_PROMISE from the server. @@ -213,8 +218,8 @@ -- The StreamId corresponds to the parent stream as PUSH_PROMISEs are tied to a -- client-initiated stream. Longer term we may move passing this handler to the -- '_startStream' instead of 'newHttp2Client' (as it is for now). -type PushPromiseHandler a = - StreamId -> Http2Stream -> IncomingFlowControl -> OutgoingFlowControl -> IO a +type PushPromiseHandler = + StreamId -> Http2Stream -> IncomingFlowControl -> OutgoingFlowControl -> IO () -- | Helper to carry around the HPACK encoder for outgoing header blocks.. data HpackEncoderContext = HpackEncoderContext { @@ -236,10 +241,8 @@ -- ^ The TLS client parameters (e.g., to allow some certificates). -> SettingsList -- ^ Initial SETTINGS that are sent as first frame. - -> PushPromiseHandler a - -- ^ Action to perform when a server sends a PUSH_PROMISE. -> IO Http2Client -newHttp2Client host port encoderBufSize decoderBufSize tlsParams initSettings handlePPStream = do +newHttp2Client host port encoderBufSize decoderBufSize tlsParams initSettings = do -- network connection conn <- newHttp2FrameConnection host port tlsParams @@ -283,13 +286,12 @@ _outgoingFlowControl <- newOutgoingFlowControl settings 0 creditFrames _incomingFlowControl <- newIncomingFlowControl settings controlStream + serverPushPromises <- newChan + _ <- forkIO $ incomingHPACKFramesLoop serverStreamFrames serverHeaders - hpackEncoder + serverPushPromises hpackDecoder - conn - settings - handlePPStream dataFrames <- dupChan serverFrames _ <- forkIO $ creditDataFramesLoop _incomingFlowControl dataFrames @@ -309,6 +311,7 @@ settings serverFrames serverHeaders + serverPushPromises hpackEncoder sid getWork @@ -323,17 +326,22 @@ sendPingFrame controlStream id dat return $ waitFrame (isPingReply dat) pingFrames let _settings settslist = do - atomicModifyIORef' settings - (\(ConnectionSettings cli srv) -> - (ConnectionSettings (HTTP2.updateSettings cli settslist) srv, ())) + -- Much like _ping, we need to dupChan before sending the query. + pingFrames <- dupChan serverFrames sendSettingsFrame controlStream id settslist + return $ do + ret <- waitFrame isSettingsReply pingFrames + atomicModifyIORef' settings + (\(ConnectionSettings cli srv) -> + (ConnectionSettings (HTTP2.updateSettings cli settslist) srv, ())) + return ret let _goaway err errStr = do sId <- readIORef maxReceivedStreamId sendGTFOFrame controlStream sId err errStr let _paylodSplitter = settingsPayloadSplitter <$> readIORef settings - _settings initSettings + _ <- forkIO . void =<< _settings initSettings return $ Http2Client{..} @@ -343,17 +351,19 @@ -> IORef ConnectionSettings -> Chan (FrameHeader, Either e FramePayload) -> Chan (FrameHeader, StreamId, Either ErrorCode HeaderList) + -> Chan (StreamId, StreamId) -> HpackEncoderContext -> StreamId -> (Http2Stream -> StreamDefinition a) -> IO (IO a) -initializeStream conn settings serverFrames serverHeaders hpackEncoder sid getWork = do +initializeStream conn settings serverFrames serverHeaders serverPushPromises hpackEncoder sid getWork = do let frameStream = makeFrameClientStream conn sid -- Register interest in frames. frames <- dupChan serverFrames credits <- dupChan serverFrames headers <- dupChan serverHeaders + pushPromises <- dupChan serverPushPromises -- Builds a flow-control context. incomingStreamFlowControl <- newIncomingFlowControl settings frameStream @@ -377,6 +387,18 @@ let _sendDataChunk = sendDataFrame frameStream let _rst = sendResetFrame frameStream let _prio = sendPriorityFrame frameStream + let _waitPushPromise ppHandler = do + (_,ppSid) <- waitPushPromiseWithParentStreamId sid pushPromises + let mkStreamActions stream = StreamDefinition (return CST) (ppHandler sid stream) + ppCont <- initializeStream conn + settings + serverFrames + serverHeaders + serverPushPromises + hpackEncoder + ppSid + mkStreamActions + ppCont let streamActions = getWork $ Http2Stream{..} @@ -454,13 +476,10 @@ :: Exception e => Chan (FrameHeader, Either e FramePayload) -> Chan (FrameHeader, StreamId, Either ErrorCode HeaderList) - -> HpackEncoderContext + -> Chan (StreamId, StreamId) -> DynamicTable - -> Http2FrameConnection - -> IORef ConnectionSettings - -> PushPromiseHandler a -> IO () -incomingHPACKFramesLoop frames headers hpackEncoder hpackDecoder conn settings handlePPStream = forever $ do +incomingHPACKFramesLoop frames headers pushPromises hpackDecoder = forever $ do (fh, fp) <- waitFrameWithTypeId [ FrameRSTStream , FramePushPromise , FrameHeaders @@ -469,15 +488,7 @@ (sId, pattern) <- case fp of PushPromiseFrame sid hbf -> do let parentSid = HTTP2.streamId fh - let mkStreamActions stream = StreamDefinition (return CST) (handlePPStream parentSid stream) - cont <- initializeStream conn - settings - frames - headers - hpackEncoder - sid - mkStreamActions - _ <- cont -- TODO: inline + writeChan pushPromises (parentSid, sid) return (sid, Right hbf) HeadersFrame _ hbf -> -- TODO: handle priority return (HTTP2.streamId fh, Right hbf) @@ -712,6 +723,10 @@ isPingReply datSent _ (PingFrame datRcv) = datSent == datRcv isPingReply _ _ _ = False +isSettingsReply :: FrameHeader -> FramePayload -> Bool +isSettingsReply fh (SettingsFrame _) = HTTP2.testAck (flags fh) +isSettingsReply _ _ = False + waitHeadersWithStreamId :: StreamId -> Chan (FrameHeader, StreamId, t) @@ -730,4 +745,17 @@ tuple@(fH, sId, headers) <- readChan chan if test fH sId headers then return tuple + else loop + +waitPushPromiseWithParentStreamId + :: StreamId + -> Chan (StreamId, StreamId) + -> IO (StreamId, StreamId) +waitPushPromiseWithParentStreamId sid chan = + loop + where + loop = do + pair@(parentSid,_) <- readChan chan + if parentSid == sid + then return pair else loop diff --git a/src/Network/HTTP2/Client/Helpers.hs b/src/Network/HTTP2/Client/Helpers.hs --- a/src/Network/HTTP2/Client/Helpers.hs +++ b/src/Network/HTTP2/Client/Helpers.hs @@ -7,21 +7,44 @@ import Data.ByteString (ByteString) import Control.Concurrent (threadDelay) import Control.Concurrent.Async (race) +import Control.Monad (forever) import Network.HTTP2.Client -type PingReply = (UTCTime, UTCTime, Either () (HTTP2.FrameHeader, HTTP2.FramePayload)) +-- | Opaque type to express an action which timed out. +data TimedOut = TimedOut + deriving Show -ping :: Int -> ByteString -> Http2Client -> IO PingReply -ping timeout msg conn = do +-- | Result for a 'ping'. +type PingReply = (UTCTime, UTCTime, Either TimedOut (HTTP2.FrameHeader, HTTP2.FramePayload)) + +-- | Performs a 'ping' and waits for a reply up to a given timeout (in +-- microseconds). +ping :: Http2Client + -- ^ client connection + -> Int + -- ^ timeout in microseconds + -> ByteString + -- ^ 8-bytes message to uniquely identify the reply + -> IO PingReply +ping conn timeout msg = do t0 <- getCurrentTime waitPing <- _ping conn msg - pingReply <- race (threadDelay timeout) waitPing + pingReply <- race (threadDelay timeout >> return TimedOut) waitPing t1 <- getCurrentTime return $ (t0, t1, pingReply) +-- | Result containing the unpacked headers and all frames received in on a +-- stream. See 'StreamResponse' and 'fromStreamResult' to get a higher-level +-- utility. type StreamResult = (Either HTTP2.ErrorCode HPACK.HeaderList, [Either HTTP2.ErrorCode ByteString]) +-- | An HTTP2 response, once fully received, is made of headers and a payload. +type StreamResponse = (HPACK.HeaderList, ByteString) + +-- | Wait for a stream until completion. +-- +-- This function is fine if you don't want to consume results in chunks. waitStream :: Http2Stream -> IncomingFlowControl -> IO StreamResult waitStream stream streamFlowControl = do (_,_,hdrs) <- _waitHeaders stream @@ -36,3 +59,18 @@ else do _ <- _updateWindow $ streamFlowControl moredata (x:xs) + +-- | Converts a StreamResult to a StramResponse, stopping at the first error +-- using the `Either HTTP2.ErrorCode` monad. +fromStreamResult :: StreamResult -> Either HTTP2.ErrorCode StreamResponse +fromStreamResult (headersE, chunksE) = do + headers <- headersE + chunks <- sequence chunksE + return (headers, mconcat chunks) + +-- | Sequentially wait for every push-promise with a handler. +-- +-- This function runs forever and you should wrap it with 'withAsync'. +onPushPromise :: Http2Stream -> PushPromiseHandler -> IO () +onPushPromise stream handler = forever $ do + _waitPushPromise stream handler