diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -177,7 +177,7 @@ ### Things that will likely change the API -I think the fundamental are right but the following needs tweaking: +I think the fundamentals are right but the following needs tweaking: - 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 @@ -208,6 +208,3 @@ CPU/latency-wise if the concurrency is high - consider most performant functions for HTTP2.HPACK encoding/decoding * currently using naive function 'Network.HPACK.encodeHeader' -- consider using a typeclass or a way to change the frame client -- we need a story for IO exceptions -- we need a story for instrumentation diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -2,19 +2,19 @@ {-# LANGUAGE RecordWildCards #-} module Main where -import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent (forkIO, myThreadId, throwTo, threadDelay) import Control.Concurrent.Async (async, waitAnyCancel) import Control.Monad (forever, when, void) import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as ByteString import Data.Default.Class (def) import Data.Maybe (fromMaybe) +import Data.Monoid ((<>)) import Data.Time.Clock (diffUTCTime, getCurrentTime) import qualified Network.HTTP2 as HTTP2 import qualified Network.TLS as TLS import qualified Network.TLS.Extra.Cipher as TLS import Options.Applicative -import Data.Monoid ((<>)) import System.IO import Network.HTTP2.Client @@ -182,50 +182,51 @@ return hdrFrame } } - conn <- case _verboseDebug of - Verbose -> - wrapFrameClient wrappedFrameConn _encoderBufsize _decoderBufsize conf - NonVerbose -> - wrapFrameClient frameConn _encoderBufsize _decoderBufsize conf - linkHttp2Client conn - _addCredit (_incomingFlowControl conn) _initialWindowKick - _ <- forkIO $ forever $ do - updated <- _updateWindow $ _incomingFlowControl conn - when updated $ timePrint ("sending flow-control update" :: String) - threadDelay _interFlowControlUpdates + parentThread <- myThreadId + let withConn = case _verboseDebug of + Verbose -> + runHttp2Client wrappedFrameConn _encoderBufsize _decoderBufsize conf defaultGoAwayHandler printUnhandledFrame + NonVerbose -> + runHttp2Client frameConn _encoderBufsize _decoderBufsize conf (throwTo parentThread) ignoreFallbackHandler - _ <- forkIO $ when (_interPingDelay > 0) $ forever $ do - threadDelay _interPingDelay - (t0, t1, pingReply) <- ping conn _pingTimeout "pingpong" - timePrint $ ("ping-reply:" :: String, pingReply, diffUTCTime t1 t0) + withConn $ \conn -> do + linkAsyncs conn + _addCredit (_incomingFlowControl conn) _initialWindowKick + _ <- forkIO $ forever $ do + updated <- _updateWindow $ _incomingFlowControl conn + when updated $ timePrint ("sending flow-control update" :: String) + threadDelay _interFlowControlUpdates - let go 0 idx = timePrint $ "done worker: " <> show idx - go n idx = do - _ <- (withHttp2Stream conn $ \stream -> - let initStream = do - _ <- async $ onPushPromise stream (ppHandler n idx) - headers stream headersPairs headersFlags - handler streamINFlowControl streamOUTFlowControl = do - timePrint $ "stream started " <> show (idx, n) - _ <- async $ dataPostFunction conn - (_outgoingFlowControl conn) - stream - streamOUTFlowControl - ret <- fromStreamResult <$> waitStream stream streamINFlowControl - either (\e -> timePrint e) (dump MainFile _path n idx _downloadPrefix) ret - timePrint $ "stream ended " <> show (idx, n) - in StreamDefinition initStream handler) - go (n - 1) idx + _ <- forkIO $ when (_interPingDelay > 0) $ forever $ do + threadDelay _interPingDelay + (t0, t1, pingReply) <- ping conn _pingTimeout "pingpong" + timePrint $ ("ping-reply:" :: String, pingReply, diffUTCTime t1 t0) - _ <- waitAnyCancel =<< traverse (async . go _numberQueries) [1 .. _concurrentQueriesCount] + let go 0 idx = timePrint $ "done worker: " <> show idx + go n idx = do + _ <- (withHttp2Stream conn $ \stream -> + let initStream = do + _ <- async $ onPushPromise stream (ppHandler n idx) + headers stream headersPairs headersFlags + handler streamINFlowControl streamOUTFlowControl = do + timePrint $ "stream started " <> show (idx, n) + _ <- async $ dataPostFunction conn + (_outgoingFlowControl conn) + stream + streamOUTFlowControl + ret <- fromStreamResult <$> waitStream stream streamINFlowControl + either (\e -> timePrint e) (dump MainFile _path n idx _downloadPrefix) ret + timePrint $ "stream ended " <> show (idx, n) + in StreamDefinition initStream handler) + go (n - 1) idx - when (_finalDelay > 0) (threadDelay _finalDelay) + _ <- waitAnyCancel =<< traverse (async . go _numberQueries) [1 .. _concurrentQueriesCount] - _gtfo conn HTTP2.NoError _finalMessage + when (_finalDelay > 0) (threadDelay _finalDelay) - _close conn - return () + _gtfo conn HTTP2.NoError _finalMessage + _close conn where tlsParams = TLS.ClientParams { TLS.clientWantSessionResume = Nothing @@ -273,3 +274,6 @@ timePrint x = do tst <- getCurrentTime ByteString.hPutStrLn stderr $ ByteString.pack $ show (tst, x) + +printUnhandledFrame :: FallBackFrameHandler +printUnhandledFrame (fh,fp) = timePrint ("UNHANDLED:"::String, fh, fp) diff --git a/http2-client.cabal b/http2-client.cabal --- a/http2-client.cabal +++ b/http2-client.cabal @@ -1,5 +1,5 @@ name: http2-client -version: 0.3.0.2 +version: 0.4.0.0 synopsis: A native HTTP2 client library. description: Please read the README.md at the homepage. homepage: https://github.com/lucasdicioccio/http2-client @@ -33,7 +33,7 @@ executable http2-client-exe hs-source-dirs: app main-is: Main.hs - ghc-options: -threaded -rtsopts -with-rtsopts=-N + ghc-options: -threaded -rtsopts -with-rtsopts=-N -O2 build-depends: base , async , bytestring 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 @@ -8,7 +8,7 @@ -- For higher-level primitives, please refer to Network.HTTP2.Client.Helpers . module Network.HTTP2.Client ( -- * Basics - newHttp2Client + runHttp2Client , withHttp2Stream , headers , sendData @@ -25,11 +25,15 @@ , IncomingFlowControl(..) , OutgoingFlowControl(..) -- * Exceptions - , linkHttp2Client + , linkAsyncs + , RemoteSentGoAwayFrame(..) + , GoAwayHandler + , defaultGoAwayHandler -- * Misc. + , FallBackFrameHandler + , ignoreFallbackHandler , FlagSetter , Http2ClientAsyncs(..) - , wrapFrameClient , _gtfo -- * Convenience re-exports , module Network.HTTP2.Client.FrameConnection @@ -37,8 +41,8 @@ , module Network.TLS ) where -import Control.Exception (bracket, throw) -import Control.Concurrent.Async (Async, async, race, link) +import Control.Concurrent.Async (Async, race, withAsync, link) +import Control.Exception (bracket, throwIO, SomeException, catch) import Control.Concurrent.MVar (newMVar, takeMVar, putMVar) import Control.Concurrent (threadDelay) import Control.Concurrent.Chan (Chan, newChan, dupChan, readChan, writeChan) @@ -47,7 +51,6 @@ import qualified Data.ByteString as ByteString import Data.IORef (IORef, newIORef, atomicModifyIORef', readIORef) import Data.Maybe (fromMaybe) -import Data.Monoid ((<>)) import GHC.Exception (Exception) import Network.HPACK as HPACK import Network.HTTP2 as HTTP2 @@ -176,8 +179,9 @@ -- | Set of Async threads running an Http2Client. -- --- If you add other asyncs here, you should also modify --- 'linkHttp2Client'. +-- This asyncs are linked to the thread where the Http2Client is created. +-- If you modify this structure to add more Async, please also modify +-- 'linkAsyncs' accordingly. data Http2ClientAsyncs = Http2ClientAsyncs { _waitSettingsAsync :: Async (FrameHeader, FramePayload) -- ^ Async waiting for the initial settings ACK. @@ -196,18 +200,15 @@ -- 'incomingFramesLoop'. } --- | Links all asyncs threads of a client to the current thread. --- --- See 'link'. -linkHttp2Client :: Http2Client -> IO () -linkHttp2Client client = - let Http2ClientAsyncs{..} = _asyncs client - in do - link _waitSettingsAsync - link _creditFlowsAsync - link _HPACKAsync - link _controlFramesAsync - link _incomingFramesAsync +-- | Links all client's asyncs to current thread. +linkAsyncs :: Http2Client -> IO () +linkAsyncs client = + let Http2ClientAsyncs{..} = _asyncs client in do + link _waitSettingsAsync + link _creditFlowsAsync + link _HPACKAsync + link _controlFramesAsync + link _incomingFramesAsync -- | Couples client and server settings together. data ConnectionSettings = ConnectionSettings { @@ -277,25 +278,6 @@ , _applySettings :: Int -> IO () } --- | Starts a new Http2Client with a remote Host/Port. TLS ClientParams are --- mandatory because we only support TLS-protected streams for now. -newHttp2Client :: HostName - -- ^ Host to connect to. - -> PortNumber - -- ^ Port number to connect to (usually 443 on the web). - -> Int - -- ^ The buffersize for the Network.HPACK encoder. - -> Int - -- ^ The buffersize for the Network.HPACK decoder. - -> ClientParams - -- ^ The TLS client parameters (e.g., to allow some certificates). - -> SettingsList - -- ^ Initial SETTINGS that are sent as first frame. - -> IO Http2Client -newHttp2Client host port encoderBufSize decoderBufSize tlsParams initSettings = do - conn <- newHttp2FrameConnection host port tlsParams - wrapFrameClient conn encoderBufSize decoderBufSize initSettings - -- | Starts a new stream (i.e., one HTTP request + server-pushes). -- -- You will typically call the returned 'StreamStarter' immediately to define @@ -334,12 +316,12 @@ headers :: Http2Stream -> HeaderList -> FlagSetter -> IO StreamThread headers = _headers --- | Prepares a client around a frame connection. +-- | Starts a new Http2Client around a frame connection. -- -- This is mostly useful if you want to muck around the Http2FrameConnection -- (e.g., for tracing frames) as well as for unit testing purposes. If you want -- to create a new connection you should likely use 'newHttp2Client' instead. -wrapFrameClient +runHttp2Client :: Http2FrameConnection -- ^ A frame connection. -> Int @@ -348,8 +330,16 @@ -- ^ The buffersize for the Network.HPACK decoder. -> SettingsList -- ^ Initial SETTINGS that are sent as first frame. - -> IO Http2Client -wrapFrameClient conn encoderBufSize decoderBufSize initSettings = do + -> GoAwayHandler + -- ^ Actions to run when the remote sends a GoAwayFrame + -> FallBackFrameHandler + -- ^ Actions to run when a control frame is not yet handled in http2-client + -- lib (e.g., PRIORITY frames). + -> (Http2Client -> IO a) + -- ^ Actions to run on the client. + -> IO a +runHttp2Client conn encoderBufSize decoderBufSize initSettings goAwayHandler fallbackHandler mainHandler = do + let _close = closeConnection conn -- prepare hpack contexts hpackEncoder <- do let strategy = (HPACK.defaultEncodeStrategy { HPACK.useHuffman = True }) @@ -371,87 +361,82 @@ -- Initial thread receiving server frames. maxReceivedStreamId <- newIORef 0 serverFrames <- newChan - aIncoming <- async $ incomingFramesLoop conn serverFrames maxReceivedStreamId - - -- Thread handling control frames. - settings <- newIORef defaultConnectionSettings - controlFrames <- dupChan serverFrames - aControl <- async $ incomingControlFramesLoop controlFrames settings hpackEncoder ackPing ackSettings - - -- Thread handling push-promises and headers frames serializing the buffers. - serverStreamFrames <- dupChan serverFrames - serverHeaders <- newChan - hpackDecoder <- do - dt <- newDynamicTableForDecoding HPACK.defaultDynamicTableSize decoderBufSize - return dt - - creditFrames <- dupChan serverFrames - - _outgoingFlowControl <- newOutgoingFlowControl settings 0 creditFrames - _incomingFlowControl <- newIncomingFlowControl settings controlStream - - serverPushPromises <- newChan - - aHPACK <- async $ incomingHPACKFramesLoop serverStreamFrames - serverHeaders - serverPushPromises - hpackDecoder + let incomingLoop = incomingFramesLoop conn serverFrames maxReceivedStreamId + withAsync incomingLoop $ \aIncoming -> do - dataFrames <- dupChan serverFrames - aCredit <- async $ creditDataFramesLoop _incomingFlowControl dataFrames + -- Thread handling control frames. + settings <- newIORef defaultConnectionSettings + controlFrames <- dupChan serverFrames + let controlLoop = incomingControlFramesLoop controlFrames settings hpackEncoder ackPing ackSettings goAwayHandler fallbackHandler + withAsync controlLoop $ \aControl -> do + -- Thread handling push-promises and headers frames serializing the buffers. + serverStreamFrames <- dupChan serverFrames + serverHeaders <- newChan + hpackDecoder <- do + dt <- newDynamicTableForDecoding HPACK.defaultDynamicTableSize decoderBufSize + return dt - conccurentStreams <- newIORef 0 - let _startStream getWork = do - maxConcurrency <- fromMaybe 100 . maxConcurrentStreams . _serverSettings - <$> readIORef settings - roomNeeded <- atomicModifyIORef' conccurentStreams - (\n -> if n < maxConcurrency then (n + 1, 0) else (n, 1 + n - maxConcurrency)) - if roomNeeded > 0 - then - return $ Left $ TooMuchConcurrency roomNeeded - else do - cont <- withClientStreamId $ \sid -> do - initializeStream conn - settings - serverFrames - serverHeaders - serverPushPromises - hpackEncoder - sid - getWork - v <- cont - atomicModifyIORef' conccurentStreams (\n -> (n - 1, ())) - return $ Right v + creditFrames <- dupChan serverFrames - let _ping dat = do - -- Need to dupChan before sending the query to avoid missing a fast - -- answer if the network is fast. - pingFrames <- dupChan serverFrames - sendPingFrame controlStream id dat - return $ waitFrame (isPingReply dat) pingFrames - let _settings settslist = do - -- 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 + _outgoingFlowControl <- newOutgoingFlowControl settings 0 creditFrames + _incomingFlowControl <- newIncomingFlowControl settings controlStream - let _paylodSplitter = settingsPayloadSplitter <$> readIORef settings + serverPushPromises <- newChan - aSettings <- async =<< _settings initSettings + let hpackLoop = incomingHPACKFramesLoop serverStreamFrames serverHeaders serverPushPromises hpackDecoder + withAsync hpackLoop $ \aHPACK -> do + dataFrames <- dupChan serverFrames + let creditLoop = creditDataFramesLoop _incomingFlowControl dataFrames + withAsync creditLoop $ \aCredit -> do + conccurentStreams <- newIORef 0 + let _startStream getWork = do + maxConcurrency <- fromMaybe 100 . maxConcurrentStreams . _serverSettings + <$> readIORef settings + roomNeeded <- atomicModifyIORef' conccurentStreams + (\n -> if n < maxConcurrency then (n + 1, 0) else (n, 1 + n - maxConcurrency)) + if roomNeeded > 0 + then + return $ Left $ TooMuchConcurrency roomNeeded + else do + cont <- withClientStreamId $ \sid -> do + initializeStream conn + settings + serverFrames + serverHeaders + serverPushPromises + hpackEncoder + sid + getWork + v <- cont + atomicModifyIORef' conccurentStreams (\n -> (n - 1, ())) + return $ Right v - let _asyncs = Http2ClientAsyncs aSettings aCredit aHPACK aControl aIncoming + let _ping dat = do + -- Need to dupChan before sending the query to avoid missing a fast + -- answer if the network is fast. + pingFrames <- dupChan serverFrames + sendPingFrame controlStream id dat + return $ waitFrame (isPingReply dat) pingFrames + let _settings settslist = do + -- 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 _close = closeConnection conn + let _paylodSplitter = settingsPayloadSplitter <$> readIORef settings - return $ Http2Client{..} + settsIO <- _settings initSettings + withAsync settsIO $ \aSettings -> do + let _asyncs = Http2ClientAsyncs aSettings aCredit aHPACK aControl aIncoming + mainHandler $ Http2Client{..} initializeStream :: Exception e @@ -521,12 +506,31 @@ -> Chan (FrameHeader, Either HTTP2Error FramePayload) -> IORef StreamId -> IO () -incomingFramesLoop conn frames maxReceivedStreamId = forever $ do +incomingFramesLoop conn frames maxReceivedStreamId = delayException $ forever $ do frame@(fh, _) <- next conn -- Remember highest streamId. atomicModifyIORef' maxReceivedStreamId (\n -> (max n (streamId fh), ())) writeChan frames frame +-- | Runs an action, rethrowing exception 50ms later. +-- +-- In a context where asynchronous are likely to occur this function gives a +-- chance to other threads to do some work before Async linking reaps them all. +-- +-- In particular, servers are likely to close their TCP connection soon after +-- sending a GoAwayFrame and we want to give a better chance to clients to +-- observe the GoAwayFrame in their handlers to distinguish GoAwayFrames +-- followed by TCP disconnection and plain TCP resets. As a result, this +-- function is mostly used to delay 'incomingFramesLoop'. A more involved +-- and future design will be to inline the various loop-processes for +-- incomingFramesLoop and GoAwayHandlers in a same thread (e.g., using +-- pipe/conduit to retain composability). +delayException :: IO a -> IO a +delayException act = act `catch` slowdown + where + slowdown :: SomeException -> IO a + slowdown e = threadDelay 50000 >> throwIO e + incomingControlFramesLoop :: Exception e => Chan (FrameHeader, Either e FramePayload) @@ -534,8 +538,10 @@ -> HpackEncoderContext -> (ByteString -> IO ()) -> IO () + -> GoAwayHandler + -> FallBackFrameHandler -> IO () -incomingControlFramesLoop frames settings hpackEncoder ackPing ackSettings = forever $ do +incomingControlFramesLoop frames settings hpackEncoder ackPing ackSettings goAwayHandler fallbackHandler = forever $ do controlFrame@(fh, payload) <- waitFrameWithStreamId 0 frames case payload of (SettingsFrame settsList) @@ -556,12 +562,44 @@ ignore "PingFrame replies waited for in the requestor thread" (WindowUpdateFrame _ ) -> ignore "connection-wide WindowUpdateFrame waited for in OutgoingFlowControl threads" - _ -> putStrLn ("UNHANDLED frame: " <> show controlFrame) + (GoAwayFrame lastSid errCode reason) -> + goAwayHandler $ RemoteSentGoAwayFrame lastSid errCode reason + _ -> + fallbackHandler controlFrame + where ignore :: String -> IO () ignore _ = return () +-- | A fallback handler for frames. +type FallBackFrameHandler = (FrameHeader, FramePayload) -> IO () + +-- | Default FallBackFrameHandler that ignores frames. +ignoreFallbackHandler :: FallBackFrameHandler +ignoreFallbackHandler = const $ pure () + +-- | A Handler for exceptional circumstances. +type GoAwayHandler = RemoteSentGoAwayFrame -> IO () + +-- | Default GoAwayHandler throws a 'RemoteSentGoAwayFrame' in the current +-- thread. +-- +-- A probably sharper handler if you want to abruptly stop any operation is to +-- get the 'ThreadId' of the main client thread and using +-- 'Control.Exception.Base.throwTo'. +-- +-- There's an inherent race condition when receiving a GoAway frame because the +-- server will likely close the connection which will lead to TCP errors as +-- well. +defaultGoAwayHandler :: GoAwayHandler +defaultGoAwayHandler = throwIO + +-- | An exception thrown when the server sends a GoAwayFrame. +data RemoteSentGoAwayFrame = RemoteSentGoAwayFrame !StreamId !ErrorCodeId !ByteString + deriving Show +instance Exception RemoteSentGoAwayFrame + -- | We currently need a specific loop for crediting streams because a client -- user may programmatically reset and stop listening for a stream and stop -- calling waitData (which credits streams). @@ -874,7 +912,7 @@ where loop = do (fHead, fPayload) <- readChan chan - let dat = either throw id fPayload + dat <- either throwIO pure fPayload if test fHead dat then return (fHead, dat) else loop