http2 4.1.0 → 4.1.1
raw patch · 8 files changed
+152/−111 lines, 8 files
Files
- ChangeLog.md +6/−0
- Network/HTTP2/Arch.hs +2/−0
- Network/HTTP2/Arch/Context.hs +2/−2
- Network/HTTP2/Arch/Receiver.hs +14/−75
- Network/HTTP2/Arch/Sender.hs +10/−26
- Network/HTTP2/Arch/Window.hs +113/−0
- Network/HTTP2/Client/Run.hs +3/−7
- http2.cabal +2/−1
ChangeLog.md view
@@ -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.
Network/HTTP2/Arch.hs view
@@ -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
Network/HTTP2/Arch/Context.hs view
@@ -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
Network/HTTP2/Arch/Receiver.hs view
@@ -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
Network/HTTP2/Arch/Sender.hs view
@@ -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
+ Network/HTTP2/Arch/Window.hs view
@@ -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
Network/HTTP2/Client/Run.hs view
@@ -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
http2.cabal view
@@ -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 <kazu@iij.ad.jp>@@ -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