http2 4.1.3 → 4.1.4
raw patch · 11 files changed
+169/−21 lines, 11 files
Files
- ChangeLog.md +7/−0
- Network/HTTP2/Arch/Manager.hs +23/−1
- Network/HTTP2/Arch/Queue.hs +2/−8
- Network/HTTP2/Arch/Receiver.hs +40/−3
- Network/HTTP2/Arch/Types.hs +76/−0
- Network/HTTP2/Client/Run.hs +1/−3
- Network/HTTP2/Server/Worker.hs +3/−0
- http2.cabal +3/−3
- test/HTTP2/ClientSpec.hs +2/−1
- util/client.hs +10/−1
- util/server.hs +2/−1
ChangeLog.md view
@@ -1,3 +1,10 @@+## 4.1.4++* Handle RST_STREAM/NO_ERROR+ [#78](https://github.com/kazu-yamamoto/http2/issues/78)+* Fixing thread leak with forkManaged+ [#74](https://github.com/kazu-yamamoto/http2/issues/74)+ ## 4.1.3 * Using crypton instead of cryptonite.
Network/HTTP2/Arch/Manager.hs view
@@ -8,12 +8,13 @@ , setAction , stop , spawnAction- , addMyId+ , forkManaged , deleteMyId , timeoutKillThread , timeoutClose ) where +import Control.Exception import Data.Foldable import Data.IORef import Data.Set (Set)@@ -77,13 +78,34 @@ spawnAction :: Manager -> IO () spawnAction (Manager q _ _) = atomically $ writeTQueue q Spawn +----------------------------------------------------------------++-- | Fork managed thread+--+-- This guarantees that the thread ID is added to the manager's queue before+-- the thread starts, and is removed again when the thread terminates+-- (normally or abnormally).+forkManaged :: Manager -> IO () -> IO ()+forkManaged mgr io =+ void $ mask_ $ forkIOWithUnmask $ \unmask -> do+ addMyId mgr+ r <- unmask io `onException` deleteMyId mgr+ deleteMyId mgr+ return r+ -- | Adding my thread id to the kill-thread list on stopping.+--+-- This is not part of the public API; see 'forkManaged' instead. addMyId :: Manager -> IO () addMyId (Manager q _ _) = do tid <- myThreadId atomically $ writeTQueue q $ Add tid -- | Deleting my thread id from the kill-thread list on stopping.+--+-- This is /only/ necessary when you want to remove the thread's ID from+-- the manager /before/ the thread terminates (thereby assuming responsibility+-- for thread cleanup yourself). deleteMyId :: Manager -> IO () deleteMyId (Manager q _ _) = do tid <- myThreadId
Network/HTTP2/Arch/Queue.hs view
@@ -2,23 +2,17 @@ module Network.HTTP2.Arch.Queue where -import UnliftIO.Concurrent (forkIO)-import UnliftIO.Exception (bracket) import UnliftIO.STM -import Imports import Network.HTTP2.Arch.Manager import Network.HTTP2.Arch.Types {-# INLINE forkAndEnqueueWhenReady #-} forkAndEnqueueWhenReady :: IO () -> TQueue (Output Stream) -> Output Stream -> Manager -> IO ()-forkAndEnqueueWhenReady wait outQ out mgr = bracket setup teardown $ \_ ->- void . forkIO $ do+forkAndEnqueueWhenReady wait outQ out mgr =+ forkManaged mgr $ do wait enqueueOutput outQ out- where- setup = addMyId mgr- teardown _ = deleteMyId mgr {-# INLINE enqueueOutput #-} enqueueOutput :: TQueue (Output Stream) -> Output Stream -> IO ()
Network/HTTP2/Arch/Receiver.hs view
@@ -172,6 +172,8 @@ ---------------------------------------------------------------- processState :: StreamState -> Context -> Stream -> StreamId -> IO Bool++-- Transition (process1) processState (Open (NoBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} streamId = do let mcl = fst <$> (getHeaderValue tokenContentLength reqvt >>= C8.readInt) when (just mcl (/= (0 :: Int))) $ E.throwIO $ StreamErrorIsSent ProtocolError streamId "no body but content-length is not zero"@@ -184,6 +186,8 @@ else putMVar streamInput inpObj return False++-- Transition (process2) processState (Open (HasBody tbl@(_,reqvt))) ctx@Context{..} strm@Stream{streamInput} _streamId = do let mcl = fst <$> (getHeaderValue tokenContentLength reqvt >>= C8.readInt) bodyLength <- newIORef 0@@ -199,12 +203,23 @@ else putMVar streamInput inpObj return False++-- Transition (process3) processState s@(Open Continued{}) ctx strm _streamId = do setStreamState ctx strm s return True++-- Transition (process4) processState HalfClosedRemote ctx strm _streamId = do halfClosedRemote ctx strm return False++-- Transition (process5)+processState (Closed cc) ctx strm _streamId = do+ closed ctx strm cc+ return False++-- Transition (process6) processState s ctx strm _streamId = do -- Idle, Open Body, Closed setStreamState ctx strm s@@ -326,6 +341,8 @@ dep = streamDependency p stream :: FrameType -> FrameHeader -> ByteString -> Context -> StreamState -> Stream -> IO StreamState++-- Transition (stream1) stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx s@(Open JustOpened) Stream{streamNumber} = do HeadersFrame mp frag <- guardIt $ decodeHeadersFrame header bs let endOfStream = testEndStream flags@@ -351,6 +368,7 @@ let siz = BS.length frag return $ Open $ Continued [frag] siz 1 endOfStream +-- Transition (stream2) stream FrameHeaders header@FrameHeader{flags,streamId} bs ctx (Open (Body q _ _ tlr)) _ = do HeadersFrame _ frag <- guardIt $ decodeHeadersFrame header bs let endOfStream = testEndStream flags@@ -376,6 +394,7 @@ else return s +-- Transition (stream4) stream FrameData header@FrameHeader{flags,payloadLength,streamId} bs@@ -404,6 +423,7 @@ else return s +-- Transition (stream5) stream FrameContinuation FrameHeader{flags,streamId} frag ctx s@(Open (Continued rfrags siz n endOfStream)) _ = do let endOfHeader = testEndHeader flags if frag == "" && not endOfHeader then do@@ -431,17 +451,34 @@ else return $ Open $ Continued rfrags' siz' n' endOfStream +-- (No state transition) stream FrameWindowUpdate header bs _ s strm = do WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs increaseStreamWindowSize strm n return s -stream FrameRSTStream header@FrameHeader{streamId} bs ctx _ strm = do+-- Transition (stream6)+stream FrameRSTStream header@FrameHeader{streamId} bs ctx s strm = do RSTStreamFrame err <- guardIt $ decodeRSTStreamFrame header bs let cc = Reset err- closed ctx strm cc- E.throwIO $ StreamErrorIsReceived err streamId + -- The spec mandates (section 8.1):+ --+ -- > When this is true, a server MAY request that the client abort+ -- > transmission of a request without error by sending a RST_STREAM with an+ -- > error code of NO_ERROR after sending a complete response (i.e., a frame+ -- > with the END_STREAM flag).+ --+ -- We check the first part ("after sending a complete response") by checking+ -- the current stream state.+ case (s, err) of+ (HalfClosedRemote, NoError) ->+ return (Closed cc)+ _otherwise -> do+ closed ctx strm cc+ E.throwIO $ StreamErrorIsReceived err streamId++-- (No state transition) stream FramePriority header bs _ s Stream{streamNumber} = do -- ignore -- Resource Loop - CVE-2019-9513
Network/HTTP2/Arch/Types.hs view
@@ -105,6 +105,82 @@ ---------------------------------------------------------------- +{-++== Stream state++The stream state is stored in the 'streamState' field (an @IORef@) of a+'Stream'. The main place where the stream state is updated is in+'controlOrStream', which does something like this:++> state0 <- readStreamState strm+> state1 <- stream .. state0 ..+> processState .. state1 ..++where 'processState' updates the @IORef@, based on 'state1' (the state computed+by 'stream') and the /current/ state of the stream; for simplicity, we will+assume here that this must equal 'state0' (it might not, if a concurrent thread+changed the stream state).++The diagram below summarizes the stream state transitions on the client side,+omitting error cases (which result in exceptions being thrown). Each transition+is labelled with the relevant case in either the function 'stream' or the+function 'processState'.++> [Open JustOpened]+> |+> |+> HEADERS+> |+> | (stream1)+> |+> END_HEADERS?+> |+> ______/ \______+> / yes no \+> | |+> | [Open Continued] <--\+> | | |+> | CONTINUATION |+> | | |+> | | (stream5) |+> | | |+> | END_HEADERS? |+> | | |+> v yes / \ no |+> END_STREAM? <-------/ \-----------/+> | (process3)+> |+> _________/ \_________+> / yes no \+> | |+> [Open NoBody] [Open HasBody]+> | |+> | (process1) | (process2)+> | |+> [HalfClosedRemote] <--\ [Open Body] <----------------------\+> | | | |+> | | +---------------\ |+> RST_STREAM | | | |+> | | HEADERS DATA |+> | (stream6) | | | |+> | | | (stream2) | (stream4) |+> | (process5) | | | |+> | | END_STREAM? END_STREAM? |+> [Closed] | | | |+> | | yes yes / \ no |+> \--------+-------------/ \-----------/+> (process4) (process6)++Notes:++- The 'HalfClosedLocal' state is not used on the client side.+- Indeed, unless an exception is thrown, even the 'Closed' stream state is not+ used in the client; when the @IORef@ is collected, it is typically in+ 'HalfClosedRemote' state.++-}+ data OpenState = JustOpened | Continued [HeaderBlockFragment]
Network/HTTP2/Client/Run.hs view
@@ -71,9 +71,7 @@ OutBodyStreaming strmbdy -> do tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 tbqNonMmpty <- newTVarIO False- let setup = addMyId mgr- let teardown _ = deleteMyId mgr- E.bracket setup teardown $ \_ -> void $ forkIO $ do+ forkManaged mgr $ do let push b = atomically $ do writeTBQueue tbq (StreamingBuilder b) writeTVar tbqNonMmpty True
Network/HTTP2/Server/Worker.hs view
@@ -139,6 +139,9 @@ flush = atomically $ writeTBQueue tbq StreamingFlush strmbdy push flush atomically $ writeTBQueue tbq StreamingFinished+ -- Remove the thread's ID from the manager's queue, to ensure the that the+ -- manager will not terminate it before we are done. (The thread ID was+ -- added implicitly when the worker was spawned by the manager). deleteMyId mgr where (_,reqvt) = inpObjHeaders req
http2.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: http2-version: 4.1.3+version: 4.1.4 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto <kazu@iij.ad.jp>@@ -299,7 +299,7 @@ default-language: Haskell2010 default-extensions: Strict StrictData- ghc-options: -Wall+ ghc-options: -Wall -threaded build-depends: base >=4.9 && <5, async,@@ -321,7 +321,7 @@ other-modules: ServerSpec default-language: Haskell2010 default-extensions: Strict StrictData- ghc-options: -Wall+ ghc-options: -Wall -threaded build-depends: base >=4.9 && <5, bytestring,
test/HTTP2/ClientSpec.hs view
@@ -63,7 +63,8 @@ cliconf = ClientConfig sc au 20 runHTTP2Client s = E.bracket (allocSimpleConfig s 4096) freeSimpleConfig- (\conf -> run cliconf conf client)+ (\conf -> run cliconf conf $ \sendRequest ->+ client sendRequest) client sendRequest = do let req = requestNoBody methodGet "/" hd sendRequest req $ \rsp -> do
util/client.hs view
@@ -4,9 +4,12 @@ import Control.Concurrent.Async import qualified Control.Exception as E+import Control.Monad import qualified Data.ByteString.Char8 as C8 import Network.HTTP.Types import Network.Run.TCP (runTCPClient) -- network-run+import System.Environment+import System.Exit import Network.HTTP2.Client @@ -14,7 +17,13 @@ serverName = "127.0.0.1" main :: IO ()-main = runTCPClient serverName "80" $ runHTTP2Client serverName+main = do+ args <- getArgs+ when (length args /= 2) $ do+ putStrLn "client <addr> <port>"+ exitFailure+ let [host,port] = args+ runTCPClient serverName port $ runHTTP2Client host where cliconf host = ClientConfig "http" (C8.pack host) 20 runHTTP2Client host s = E.bracket (allocSimpleConfig s 4096)
util/server.hs view
@@ -14,10 +14,11 @@ import Network.HPACK import Network.HPACK.Token import Network.HTTP.Types-import Network.HTTP2.Server import Network.Run.TCP -- network-run import System.Environment import System.Exit++import Network.HTTP2.Server main :: IO () main = do