http2-client 0.3.0.2 → 0.4.0.0

raw patch · 4 files changed

+204/−165 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README.md +1/−4
• app/Main.hs +43/−39
• http2-client.cabal +2/−2
• src/Network/HTTP2/Client.hs +158/−120

@@ -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

@@ -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)

@@ -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

@@ -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