diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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. diff --git a/Network/HTTP2/Arch/Manager.hs b/Network/HTTP2/Arch/Manager.hs --- a/Network/HTTP2/Arch/Manager.hs +++ b/Network/HTTP2/Arch/Manager.hs @@ -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 diff --git a/Network/HTTP2/Arch/Queue.hs b/Network/HTTP2/Arch/Queue.hs --- a/Network/HTTP2/Arch/Queue.hs +++ b/Network/HTTP2/Arch/Queue.hs @@ -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 () diff --git a/Network/HTTP2/Arch/Receiver.hs b/Network/HTTP2/Arch/Receiver.hs --- a/Network/HTTP2/Arch/Receiver.hs +++ b/Network/HTTP2/Arch/Receiver.hs @@ -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 diff --git a/Network/HTTP2/Arch/Types.hs b/Network/HTTP2/Arch/Types.hs --- a/Network/HTTP2/Arch/Types.hs +++ b/Network/HTTP2/Arch/Types.hs @@ -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] diff --git a/Network/HTTP2/Client/Run.hs b/Network/HTTP2/Client/Run.hs --- a/Network/HTTP2/Client/Run.hs +++ b/Network/HTTP2/Client/Run.hs @@ -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 diff --git a/Network/HTTP2/Server/Worker.hs b/Network/HTTP2/Server/Worker.hs --- a/Network/HTTP2/Server/Worker.hs +++ b/Network/HTTP2/Server/Worker.hs @@ -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 diff --git a/http2.cabal b/http2.cabal --- a/http2.cabal +++ b/http2.cabal @@ -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 @@ -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, diff --git a/test/HTTP2/ClientSpec.hs b/test/HTTP2/ClientSpec.hs --- a/test/HTTP2/ClientSpec.hs +++ b/test/HTTP2/ClientSpec.hs @@ -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 diff --git a/util/client.hs b/util/client.hs --- a/util/client.hs +++ b/util/client.hs @@ -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 " + 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) diff --git a/util/server.hs b/util/server.hs --- a/util/server.hs +++ b/util/server.hs @@ -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