diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,9 @@ +## 4.1.1 + +* Fixing memory-blow-up due to no flow control. + [#62][https://github.com/kazu-yamamoto/http2/issues/62] + [#66][https://github.com/kazu-yamamoto/http2/issues/66] + ## 4.1.0 * Implementing streaming from the client side. diff --git a/Network/HTTP2/Arch.hs b/Network/HTTP2/Arch.hs --- a/Network/HTTP2/Arch.hs +++ b/Network/HTTP2/Arch.hs @@ -12,6 +12,7 @@ , module Network.HTTP2.Arch.Status , module Network.HTTP2.Arch.Stream , module Network.HTTP2.Arch.Types + , module Network.HTTP2.Arch.Window ) where import Network.HTTP2.Arch.Config @@ -27,3 +28,4 @@ import Network.HTTP2.Arch.Stream import Network.HTTP2.Arch.Status import Network.HTTP2.Arch.Types +import Network.HTTP2.Arch.Window diff --git a/Network/HTTP2/Arch/Context.hs b/Network/HTTP2/Arch/Context.hs --- a/Network/HTTP2/Arch/Context.hs +++ b/Network/HTTP2/Arch/Context.hs @@ -83,7 +83,7 @@ -- the connection window for sending data , txConnectionWindow :: TVar WindowSize -- window update for receiving data - , rxConnectionInc :: IORef Int + , rxConnectionInc :: IORef WindowSize -- this is diff , pingRate :: Rate , settingsRate :: Rate , emptyFrameRate :: Rate @@ -188,7 +188,7 @@ openStream :: Context -> StreamId -> FrameType -> IO Stream openStream ctx@Context{streamTable, peerSettings} sid ftyp = do ws <- initialWindowSize <$> readIORef peerSettings - newstrm <- newStream sid $ fromIntegral ws + newstrm <- newStream sid ws when (ftyp == FrameHeaders || ftyp == FramePushPromise) $ opened ctx newstrm insert streamTable sid newstrm return newstrm 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 @@ -5,9 +5,6 @@ module Network.HTTP2.Arch.Receiver ( frameReceiver - , maxConcurrency - , myInitialAlist - , initialFrames ) where import qualified Data.ByteString as BS @@ -29,13 +26,11 @@ import Network.HTTP2.Arch.Rate import Network.HTTP2.Arch.Stream import Network.HTTP2.Arch.Types +import Network.HTTP2.Arch.Window import Network.HTTP2.Frame ---------------------------------------------------------------- -maxConcurrency :: Int -maxConcurrency = recommendedConcurrency - continuationLimit :: Int continuationLimit = 10 @@ -53,26 +48,6 @@ ---------------------------------------------------------------- -initialFrames :: SettingsList -> [ByteString] -initialFrames alist = [frame1,frame2] - where - frame1 = settingsFrame id alist - frame2 = windowUpdateFrame 0 (maxWindowSize - defaultWindowSize) - -myInitialAlist :: Config -> SettingsList -myInitialAlist Config{..} = - -- confBufferSize is the size of the write buffer. - -- But we assume that the size of the read buffer is the same size. - -- So, the size is announced to via SETTINGS_MAX_FRAME_SIZE. - [(SettingsMaxFrameSize,payloadLen) - ,(SettingsMaxConcurrentStreams,maxConcurrency) - ,(SettingsInitialWindowSize,maxWindowSize)] - where - len = confBufferSize - frameHeaderLength - payloadLen = max defaultPayloadLength len - ----------------------------------------------------------------- - frameReceiver :: Context -> Config -> IO () frameReceiver ctx@Context{..} conf@Config{..} = loop 0 `E.catch` sendGoaway where @@ -216,7 +191,7 @@ q <- newTQueueIO setStreamState ctx strm $ Open (Body q mcl bodyLength tlr) incref <- newIORef 0 - bodySource <- mkSource q $ updateWindow controlQ streamId incref + bodySource <- mkSource q $ informWindowUpdate controlQ streamId incref let inpObj = InpObj tbl mcl (readSource bodySource) tlr if isServer ctx then do let si = toServerInfo roleInfo @@ -280,7 +255,7 @@ type Payload = ByteString control :: FrameType -> FrameHeader -> Payload -> Context -> Config -> IO () -control FrameSettings header@FrameHeader{flags,streamId} bs Context{myFirstSettings,myPendingAlist,mySettings,controlQ,settingsRate} conf = do +control FrameSettings header@FrameHeader{flags,streamId} bs ctx@Context{myFirstSettings,myPendingAlist,mySettings,controlQ,settingsRate} conf = do SettingsFrame peerAlist <- guardIt $ decodeSettingsFrame header bs traverse_ E.throwIO $ checkSettingsList peerAlist if testAck flags then do @@ -304,11 +279,8 @@ enqueueControl controlQ setframe else do -- Server side only - writeIORef myFirstSettings True - let myAlist = myInitialAlist conf - writeIORef myPendingAlist $ Just myAlist - let frames = initialFrames myAlist ++ [ack] - setframe = CFrames (Just peerAlist) frames + frames <- updateMySettings conf ctx + let setframe = CFrames (Just peerAlist) (frames ++ [ack]) enqueueControl controlQ setframe control FramePing FrameHeader{flags,streamId} bs Context{controlQ,pingRate} _ = @@ -328,13 +300,9 @@ else E.throwIO $ ConnectionErrorIsReceived err sid $ Short.toShort msg -control FrameWindowUpdate header@FrameHeader{streamId} bs Context{txConnectionWindow} _ = do +control FrameWindowUpdate header@FrameHeader{streamId} bs ctx _ = do WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs - w <- atomically $ do - w0 <- readTVar txConnectionWindow - let w1 = w0 + n - writeTVar txConnectionWindow w1 - return w1 + w <- increaseConnectionWindowSize ctx n when (isWindowOverflow w) $ E.throwIO $ ConnectionErrorIsSent FlowControlError streamId "control window should be less than 2^31" control _ _ _ _ _ = @@ -403,7 +371,7 @@ _bs ctx s@(HalfClosedLocal _) _ = do - rxConnectionWindowIncrement ctx payloadLength + informConnectionWindowUpdate ctx payloadLength let endOfStream = testEndStream flags if endOfStream then do return HalfClosedRemote @@ -415,7 +383,7 @@ bs ctx@Context{emptyFrameRate} s@(Open (Body q mcl bodyLength _)) _ = do - rxConnectionWindowIncrement ctx payloadLength + informConnectionWindowUpdate ctx payloadLength DataFrame body <- guardIt $ decodeDataFrame header bs len0 <- readIORef bodyLength let len = len0 + payloadLength @@ -466,13 +434,9 @@ else return $ Open $ Continued rfrags' siz' n' endOfStream -stream FrameWindowUpdate header@FrameHeader{streamId} bs _ s Stream{streamWindow} = do +stream FrameWindowUpdate header@FrameHeader{streamId} bs _ s strm = do WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs - w <- atomically $ do - w0 <- readTVar streamWindow - let w1 = w0 + n - writeTVar streamWindow w1 - return w1 + w <- increaseStreamWindowSize strm n when (isWindowOverflow w) $ E.throwIO $ StreamErrorIsSent FlowControlError streamId return s @@ -506,29 +470,17 @@ (IORef Bool) mkSource :: TQueue ByteString -> (Int -> IO ()) -> IO Source -mkSource q update = Source update q <$> newIORef "" <*> newIORef False - -updateWindow :: TQueue Control -> StreamId -> IORef Int -> Int -> IO () -updateWindow _ _ _ 0 = return () -updateWindow controlQ sid incref len = do - w0 <- readIORef incref - let w1 = w0 + len - if w1 >= defaultWindowSize then do -- fixme - let frame = windowUpdateFrame sid w1 - enqueueControl controlQ $ CFrames Nothing [frame] - writeIORef incref 0 - else - writeIORef incref w1 +mkSource q inform = Source inform q <$> newIORef "" <*> newIORef False readSource :: Source -> IO ByteString -readSource (Source update q refBS refEOF) = do +readSource (Source inform q refBS refEOF) = do eof <- readIORef refEOF if eof then return "" else do bs <- readBS let len = BS.length bs - update len + inform len return bs where readBS = do @@ -540,16 +492,3 @@ else do writeIORef refBS "" return bs0 - ----------------------------------------------------------------- - -rxConnectionWindowIncrement :: Context -> Int -> IO () -rxConnectionWindowIncrement Context{..} len = do - w0 <- readIORef rxConnectionInc - let w1 = w0 + len - if w1 >= defaultWindowSize then do -- fixme - let frame = windowUpdateFrame 0 w1 - enqueueControl controlQ $ CFrames Nothing [frame] - writeIORef rxConnectionInc 0 - else - writeIORef rxConnectionInc w1 diff --git a/Network/HTTP2/Arch/Sender.hs b/Network/HTTP2/Arch/Sender.hs --- a/Network/HTTP2/Arch/Sender.hs +++ b/Network/HTTP2/Arch/Sender.hs @@ -14,7 +14,7 @@ import qualified Data.ByteString as BS import Data.ByteString.Builder (Builder) import qualified Data.ByteString.Builder.Extra as B -import Data.IORef (readIORef, writeIORef, modifyIORef') +import Data.IORef (readIORef, writeIORef) import Foreign.Ptr (plusPtr, minusPtr) import Network.ByteOrder import qualified UnliftIO.Exception as E @@ -31,6 +31,7 @@ import Network.HTTP2.Arch.Queue import Network.HTTP2.Arch.Stream import Network.HTTP2.Arch.Types +import Network.HTTP2.Arch.Window import Network.HTTP2.Frame ---------------------------------------------------------------- @@ -41,16 +42,6 @@ ---------------------------------------------------------------- -{-# INLINE getStreamWindowSize #-} -getStreamWindowSize :: Stream -> IO WindowSize -getStreamWindowSize Stream{streamWindow} = readTVarIO streamWindow - -{-# INLINE waitStreamWindowSize #-} -waitStreamWindowSize :: Stream -> IO () -waitStreamWindowSize Stream{streamWindow} = atomically $ do - w <- readTVar streamWindow - checkSTM (w > 0) - {-# INLINE waitStreaming #-} waitStreaming :: TBQueue a -> IO () waitStreaming tbq = atomically $ do @@ -67,7 +58,7 @@ | otherwise = E.throwIO $ BadThingHappen se frameSender :: Context -> Config -> Manager -> IO () -frameSender ctx@Context{outputQ,controlQ,txConnectionWindow,encodeDynamicTable,peerSettings,streamTable,outputBufferLimit} +frameSender ctx@Context{outputQ,controlQ,encodeDynamicTable,outputBufferLimit} Config{..} mgr = loop 0 `E.catch` wrapException where @@ -99,8 +90,7 @@ dequeue off = do isEmpty <- isEmptyTQueue controlQ if isEmpty then do - w <- readTVar txConnectionWindow - checkSTM (w > 0) + waitConnectionWindowSize ctx emp <- isEmptyTQueue outputQ if emp then if off /= 0 then return Flush else retrySTM @@ -124,11 +114,7 @@ Nothing -> return () Just peerAlist -> do -- Peer SETTINGS_INITIAL_WINDOW_SIZE - oldws <- initialWindowSize <$> readIORef peerSettings - modifyIORef' peerSettings $ \old -> updateSettings old peerAlist - newws <- initialWindowSize <$> readIORef peerSettings - let diff = newws - oldws - when (diff /= 0) $ updateAllStreamWindow (+ diff) streamTable + updatePeerSettings ctx peerAlist -- Peer SETTINGS_MAX_FRAME_SIZE case lookup SettingsMaxFrameSize peerAlist of Nothing -> return () @@ -227,7 +213,7 @@ forkAndEnqueueWhenReady (waitStreamWindowSize strm) outputQ out mgr return off else do - cws <- readTVarIO txConnectionWindow -- not 0 + cws <- getConnectionWindowSize ctx -- not 0 let lim = min cws sws output out off lim resetStream e = do @@ -283,7 +269,7 @@ -> Output Stream -> Bool -> IO Offset - fillDataHeaderEnqueueNext strm@Stream{streamWindow,streamNumber} + fillDataHeaderEnqueueNext strm@Stream{streamNumber} off datPayloadLen Nothing tlrmkr tell _ reqflush = do let buf = confWriteBuffer `plusPtr` off off' = off + frameHeaderLength + datPayloadLen @@ -297,8 +283,7 @@ off'' <- handleTrailers mtrailers off' void tell when (isServer ctx) $ halfClosedLocal ctx strm Finished - atomically $ modifyTVar' txConnectionWindow (subtract datPayloadLen) - atomically $ modifyTVar' streamWindow (subtract datPayloadLen) + decreaseWindowSize ctx strm datPayloadLen if reqflush then do flushN off'' return 0 @@ -320,14 +305,13 @@ else return off - fillDataHeaderEnqueueNext Stream{streamWindow,streamNumber} + fillDataHeaderEnqueueNext strm@Stream{streamNumber} off datPayloadLen (Just next) tlrmkr _ out reqflush = do let buf = confWriteBuffer `plusPtr` off off' = off + frameHeaderLength + datPayloadLen flag = defaultFlags fillFrameHeader FrameData datPayloadLen streamNumber flag buf - atomically $ modifyTVar' txConnectionWindow (subtract datPayloadLen) - atomically $ modifyTVar' streamWindow (subtract datPayloadLen) + decreaseWindowSize ctx strm datPayloadLen let out' = out { outputType = ONext next tlrmkr } enqueueOutput outputQ out' if reqflush then do diff --git a/Network/HTTP2/Arch/Window.hs b/Network/HTTP2/Arch/Window.hs new file mode 100644 --- /dev/null +++ b/Network/HTTP2/Arch/Window.hs @@ -0,0 +1,113 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE RecordWildCards #-} + +module Network.HTTP2.Arch.Window where + +import Data.IORef +import UnliftIO.STM + +import Imports +import Network.HTTP2.Arch.Config +import Network.HTTP2.Arch.Context +import Network.HTTP2.Arch.EncodeFrame +import Network.HTTP2.Arch.Queue +import Network.HTTP2.Arch.Stream +import Network.HTTP2.Arch.Types +import Network.HTTP2.Frame + +---------------------------------------------------------------- +-- Receiving window update + +increaseStreamWindowSize :: Stream -> Int -> IO WindowSize +increaseStreamWindowSize Stream{streamWindow} n = atomically $ do + w0 <- readTVar streamWindow + let w1 = w0 + n + writeTVar streamWindow w1 + return w1 + +increaseConnectionWindowSize :: Context -> Int -> IO WindowSize +increaseConnectionWindowSize Context{txConnectionWindow} n = atomically $ do + w0 <- readTVar txConnectionWindow + let w1 = w0 + n + writeTVar txConnectionWindow w1 + return w1 + +---------------------------------------------------------------- +-- Sending window update + +decreaseWindowSize :: Context -> Stream -> WindowSize -> IO () +decreaseWindowSize Context{txConnectionWindow} Stream{streamWindow} siz = do + atomically $ modifyTVar' txConnectionWindow (subtract siz) + atomically $ modifyTVar' streamWindow (subtract siz) + +informWindowUpdate :: TQueue Control -> StreamId -> IORef Int -> Int -> IO () +informWindowUpdate _ _ _ 0 = return () +informWindowUpdate controlQ sid incref len = do + -- incref is occupied by the receiver thread + w0 <- readIORef incref + let w1 = w0 + len + if w1 >= defaultWindowSize then do -- fixme + let frame = windowUpdateFrame sid w1 + enqueueControl controlQ $ CFrames Nothing [frame] + writeIORef incref 0 + else + writeIORef incref w1 + +informConnectionWindowUpdate :: Context -> Int -> IO () +informConnectionWindowUpdate Context{..} = + informWindowUpdate controlQ 0 rxConnectionInc + +getStreamWindowSize :: Stream -> IO WindowSize +getStreamWindowSize Stream{streamWindow} = readTVarIO streamWindow + +getConnectionWindowSize :: Context -> IO WindowSize +getConnectionWindowSize Context{txConnectionWindow} = readTVarIO txConnectionWindow + +waitStreamWindowSize :: Stream -> IO () +waitStreamWindowSize Stream{streamWindow} = atomically $ do + w <- readTVar streamWindow + checkSTM (w > 0) + +waitConnectionWindowSize :: Context -> STM () +waitConnectionWindowSize Context{txConnectionWindow} = do + w <- readTVar txConnectionWindow + checkSTM (w > 0) + +---------------------------------------------------------------- + +-- max: 2,147,483,647 (2^31-1) is too large. +-- def: 65,535 (2^16-1) it too small. +-- 1,048,575 (2^20-1) +properWindowSize :: WindowSize +properWindowSize = 1048575 + +updateMySettings :: Config -> Context -> IO [ByteString] +updateMySettings Config{..} Context{myFirstSettings,myPendingAlist} = do + writeIORef myFirstSettings True + writeIORef myPendingAlist $ Just myInitialAlist + return frames + where + len = confBufferSize - frameHeaderLength + payloadLen = max defaultPayloadLength len + myInitialAlist = + -- confBufferSize is the size of the write buffer. + -- But we assume that the size of the read buffer is the same size. + -- So, the size is announced to via SETTINGS_MAX_FRAME_SIZE. + [(SettingsMaxFrameSize,payloadLen) + ,(SettingsMaxConcurrentStreams,recommendedConcurrency) + -- Initial window size for streams + ,(SettingsInitialWindowSize,properWindowSize)] + frame1 = settingsFrame id myInitialAlist + -- Initial window update for connection + frame2 = windowUpdateFrame 0 (properWindowSize - defaultWindowSize) + frames = [frame1,frame2] + +-- Peer SETTINGS_INITIAL_WINDOW_SIZE +-- Adjusting initial window size for streams +updatePeerSettings :: Context -> SettingsList -> IO () +updatePeerSettings Context{peerSettings,streamTable} peerAlist = do + oldws <- initialWindowSize <$> readIORef peerSettings + modifyIORef' peerSettings $ \old -> updateSettings old peerAlist + newws <- initialWindowSize <$> readIORef peerSettings + let diff = newws - oldws + when (diff /= 0) $ updateAllStreamWindow (+ diff) streamTable 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 @@ -4,7 +4,6 @@ module Network.HTTP2.Client.Run where import Control.Concurrent.STM (check) -import Data.IORef (writeIORef) import UnliftIO.Async import UnliftIO.Concurrent import qualified UnliftIO.Exception as E @@ -98,10 +97,7 @@ processResponse $ Response rsp exchangeSettings :: Config -> Context -> IO () -exchangeSettings conf Context{..} = do - writeIORef myFirstSettings True - let myAlist = myInitialAlist conf - writeIORef myPendingAlist $ Just myAlist - let frames = initialFrames myAlist - setframe = CFrames Nothing (connectionPreface:frames) +exchangeSettings conf ctx@Context{..} = do + frames <- updateMySettings conf ctx + let setframe = CFrames Nothing (connectionPreface:frames) enqueueControl controlQ setframe 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.0 +version: 4.1.1 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto @@ -102,6 +102,7 @@ Network.HTTP2.Arch.Status Network.HTTP2.Arch.Stream Network.HTTP2.Arch.Types + Network.HTTP2.Arch.Window Network.HTTP2.Frame.Decode Network.HTTP2.Frame.Encode Network.HTTP2.Frame.Types