http2 4.0.0 → 4.1.0
raw patch · 17 files changed
+527/−329 lines, 17 filesdep −doctestPVP ok
version bump matches the API change (PVP)
Dependencies removed: doctest
API changes (from Hackage documentation)
- Network.HTTP2.Frame: [maxHeaderBlockSize] :: Settings -> Maybe Int
- Network.HTTP2.Frame: decoderstStreamFrame :: FramePayloadDecoder
- Network.HTTP2.Frame: defaultInitialWindowSize :: WindowSize
+ Network.HTTP2.Frame: [maxHeaderListSize] :: Settings -> Maybe Int
+ Network.HTTP2.Frame: decodeRSTStreamFrame :: FramePayloadDecoder
+ Network.HTTP2.Frame: defaultPayloadLength :: Int
+ Network.HTTP2.Frame: defaultWindowSize :: WindowSize
- Network.HTTP2.Internal: Next :: BytesFilled -> Maybe DynaNext -> Next
+ Network.HTTP2.Internal: Next :: BytesFilled -> Bool -> Maybe DynaNext -> Next
Files
- ChangeLog.md +20/−5
- Network/HTTP2/Arch/Config.hs +5/−1
- Network/HTTP2/Arch/Context.hs +27/−12
- Network/HTTP2/Arch/Receiver.hs +133/−84
- Network/HTTP2/Arch/Sender.hs +190/−128
- Network/HTTP2/Arch/Types.hs +7/−8
- Network/HTTP2/Client/Run.hs +49/−13
- Network/HTTP2/Frame.hs +4/−2
- Network/HTTP2/Frame/Decode.hs +8/−11
- Network/HTTP2/Frame/Types.hs +21/−14
- Network/HTTP2/Server/Run.hs +3/−3
- Network/HTTP2/Server/Types.hs +0/−1
- Network/HTTP2/Server/Worker.hs +6/−4
- http2.cabal +1/−21
- test/HTTP2/ClientSpec.hs +1/−1
- test/HTTP2/ServerSpec.hs +52/−7
- test/doctests.hs +0/−14
ChangeLog.md view
@@ -1,3 +1,18 @@+## 4.1.0++* Implementing streaming from the client side.+ [#41](https://github.com/kazu-yamamoto/http2/pull/41)+* Making use of SettingsMaxFrameSize+ [#44](https://github.com/kazu-yamamoto/http2/pull/44)+ [#57](https://github.com/kazu-yamamoto/http2/pull/57)+* Disabling flow control+ [#55](https://github.com/kazu-yamamoto/http2/pull/55)+* Fixing buffer overrun by trailers+ [#52](https://github.com/kazu-yamamoto/http2/pull/52)+* Proper use of settings+* Breaking change: the data structure of `Next` was changed.+ The `http3` package is influenced.+ ## 4.0.0 * Breaking change: `HTTP2Error` is redefined.@@ -11,18 +26,18 @@ ## 3.0.3 * Return correct status messages in HTTP2 client- (#31)[https://github.com/kazu-yamamoto/http2/pull/31]+ [#31](https://github.com/kazu-yamamoto/http2/pull/31) * Follow changes in Aeson 2- (#32)[https://github.com/kazu-yamamoto/http2/pull/32]+ [#32](https://github.com/kazu-yamamoto/http2/pull/32) * Make sure connection preface is always sent first- (#33)[https://github.com/kazu-yamamoto/http2/pull/33]+ [#33](https://github.com/kazu-yamamoto/http2/pull/33) * Avoid empty data- (#34)[https://github.com/kazu-yamamoto/http2/pull/34]+ [#34](https://github.com/kazu-yamamoto/http2/pull/34) ## 3.0.2 * Skip inserting entries that do not fit in the encoding table- (#28)[https://github.com/kazu-yamamoto/http2/pull/28]+ [#28](https://github.com/kazu-yamamoto/http2/pull/28) ## 3.0.1
Network/HTTP2/Arch/Config.hs view
@@ -13,9 +13,13 @@ -- | HTTP/2 configuration. data Config = Config {- -- confWriteBuffer is used only by frameSender.+ -- | This is used only by frameSender. -- This MUST be freed after frameSender is terminated. confWriteBuffer :: Buffer+ -- | The size of the write buffer.+ -- We assume that the read buffer is the same size.+ -- So, this value is announced via SETTINGS_MAX_FRAME_SIZE+ -- to the peer. , confBufferSize :: BufferSize , confSendAll :: ByteString -> IO () , confReadN :: Int -> IO ByteString
Network/HTTP2/Arch/Context.hs view
@@ -60,9 +60,11 @@ data Context = Context { role :: Role , roleInfo :: RoleInfo- -- HTTP/2 settings received from a browser- , http2settings :: IORef Settings- , firstSettings :: IORef Bool+ -- Settings+ , myFirstSettings :: IORef Bool+ , myPendingAlist :: IORef (Maybe SettingsList)+ , mySettings :: IORef Settings+ , peerSettings :: IORef Settings , streamTable :: StreamTable , concurrency :: IORef Int -- | RFC 9113 says "Other frames (from any stream) MUST NOT@@ -72,12 +74,16 @@ , continued :: IORef (Maybe StreamId) , myStreamId :: IORef StreamId , peerStreamId :: IORef StreamId+ , outputBufferLimit :: IORef Int , outputQ :: TQueue (Output Stream)+ , outputQStreamID :: TVar StreamId , controlQ :: TQueue Control , encodeDynamicTable :: DynamicTable , decodeDynamicTable :: DynamicTable- -- the connection window for data from a server to a browser.- , connectionWindow :: TVar WindowSize+ -- the connection window for sending data+ , txConnectionWindow :: TVar WindowSize+ -- window update for receiving data+ , rxConnectionInc :: IORef Int , pingRate :: Rate , settingsRate :: Rate , emptyFrameRate :: Rate@@ -85,21 +91,27 @@ ---------------------------------------------------------------- -newContext :: RoleInfo -> IO Context-newContext rinfo =+newContext :: RoleInfo -> BufferSize -> IO Context+newContext rinfo siz = Context rl rinfo- <$> newIORef defaultSettings- <*> newIORef False+ <$> newIORef False+ <*> newIORef Nothing+ <*> newIORef defaultSettings+ <*> newIORef defaultSettings <*> newStreamTable <*> newIORef 0 <*> newIORef Nothing <*> newIORef sid0 <*> newIORef 0+ <*> newIORef buflim <*> newTQueueIO+ <*> newTVarIO sid0 <*> newTQueueIO+ -- My SETTINGS_HEADER_TABLE_SIZE <*> newDynamicTableForEncoding defaultDynamicTableSize <*> newDynamicTableForDecoding defaultDynamicTableSize 4096- <*> newTVarIO defaultInitialWindowSize+ <*> newTVarIO defaultWindowSize+ <*> newIORef 0 <*> newRate <*> newRate <*> newRate@@ -109,6 +121,9 @@ _ -> Server sid0 | rl == Client = 1 | otherwise = 2+ dlim = defaultPayloadLength + frameHeaderLength+ buflim | siz >= dlim = dlim+ | otherwise = siz ---------------------------------------------------------------- @@ -171,8 +186,8 @@ setStreamState ctx strm (Closed cc) -- anyway openStream :: Context -> StreamId -> FrameType -> IO Stream-openStream ctx@Context{streamTable, http2settings} sid ftyp = do- ws <- initialWindowSize <$> readIORef http2settings+openStream ctx@Context{streamTable, peerSettings} sid ftyp = do+ ws <- initialWindowSize <$> readIORef peerSettings newstrm <- newStream sid $ fromIntegral ws when (ftyp == FrameHeaders || ftyp == FramePushPromise) $ opened ctx newstrm insert streamTable sid newstrm
Network/HTTP2/Arch/Receiver.hs view
@@ -6,7 +6,8 @@ module Network.HTTP2.Arch.Receiver ( frameReceiver , maxConcurrency- , initialFrame+ , myInitialAlist+ , initialFrames ) where import qualified Data.ByteString as BS@@ -20,6 +21,7 @@ import Imports hiding (delete, insert) import Network.HPACK import Network.HPACK.Token+import Network.HTTP2.Arch.Config import Network.HTTP2.Arch.Context import Network.HTTP2.Arch.EncodeFrame import Network.HTTP2.Arch.HPACK@@ -51,15 +53,28 @@ ---------------------------------------------------------------- -initialFrame :: ByteString-initialFrame = settingsFrame id [(SettingsMaxConcurrentStreams,maxConcurrency)]+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 -type RecvN = Int -> IO ByteString+---------------------------------------------------------------- -frameReceiver :: Context -> RecvN -> IO ()-frameReceiver ctx@Context{..} recvN = loop 0 `E.catch` sendGoaway+frameReceiver :: Context -> Config -> IO ()+frameReceiver ctx@Context{..} conf@Config{..} = loop 0 `E.catch` sendGoaway where loop :: Int -> IO () loop n@@ -67,55 +82,59 @@ yield loop 0 | otherwise = do- hd <- recvN frameHeaderLength+ hd <- confReadN frameHeaderLength if BS.null hd then- enqueueControl controlQ CFinish+ enqueueControl controlQ $ CFinish ConnectionIsClosed else do- processFrame ctx recvN $ decodeFrameHeader hd+ processFrame ctx conf $ decodeFrameHeader hd loop (n + 1) - sendGoaway e- | Just ConnectionIsClosed <- E.fromException e = E.throwIO ConnectionIsClosed- | Just (ConnectionErrorIsReceived _ _ _) <- E.fromException e =- E.throwIO e- | Just (ConnectionErrorIsSent err sid msg) <- E.fromException e = do+ sendGoaway se+ | Just e@ConnectionIsClosed <- E.fromException se =+ enqueueControl controlQ $ CFinish e+ | Just e@(ConnectionErrorIsReceived _ _ _) <- E.fromException se =+ enqueueControl controlQ $ CFinish e+ | Just e@(ConnectionErrorIsSent err sid msg) <- E.fromException se = do let frame = goawayFrame sid err $ Short.fromShort msg- enqueueControl controlQ $ CGoaway frame- | Just (StreamErrorIsSent err sid) <- E.fromException e = do+ enqueueControl controlQ $ CFrames Nothing [frame]+ enqueueControl controlQ $ CFinish e+ | Just e@(StreamErrorIsSent err sid) <- E.fromException se = do let frame = resetFrame err sid- enqueueControl controlQ $ CFrame frame+ enqueueControl controlQ $ CFrames Nothing [frame] let frame' = goawayFrame sid err "treat a stream error as a connection error"- enqueueControl controlQ $ CGoaway frame'- E.throwIO e- | Just (StreamErrorIsReceived err sid) <- E.fromException e = do+ enqueueControl controlQ $ CFrames Nothing [frame']+ enqueueControl controlQ $ CFinish e+ | Just e@(StreamErrorIsReceived err sid) <- E.fromException se = do let frame = goawayFrame sid err "treat a stream error as a connection error"- enqueueControl controlQ $ CGoaway frame- E.throwIO e+ enqueueControl controlQ $ CFrames Nothing [frame]+ enqueueControl controlQ $ CFinish e -- this never happens- | Just x@(BadThingHappen _) <- E.fromException e = E.throwIO x- | otherwise = E.throwIO $ BadThingHappen e+ | Just e@(BadThingHappen _) <- E.fromException se =+ enqueueControl controlQ $ CFinish e+ | otherwise =+ enqueueControl controlQ $ CFinish $ BadThingHappen se ---------------------------------------------------------------- -processFrame :: Context -> RecvN -> (FrameType, FrameHeader) -> IO ()-processFrame ctx _recvN (fid, FrameHeader{streamId})+processFrame :: Context -> Config -> (FrameType, FrameHeader) -> IO ()+processFrame ctx _conf (fid, FrameHeader{streamId}) | isServer ctx && isServerInitiated streamId && (fid `notElem` [FramePriority,FrameRSTStream,FrameWindowUpdate]) = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "stream id should be odd" -processFrame Context{..} recvN (ftyp, FrameHeader{payloadLength,streamId})+processFrame Context{..} Config{..} (ftyp, FrameHeader{payloadLength,streamId}) | ftyp > maxFrameType = do mx <- readIORef continued case mx of Nothing -> do -- ignoring unknown frame- void $ recvN payloadLength+ void $ confReadN payloadLength Just _ -> E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "unknown frame"-processFrame ctx recvN (FramePushPromise, header@FrameHeader{payloadLength,streamId})+processFrame ctx Config{..} (FramePushPromise, header@FrameHeader{payloadLength,streamId}) | isServer ctx = E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "push promise is not allowed" | otherwise = do- pl <- recvN payloadLength+ pl <- confReadN payloadLength PushPromiseFrame sid frag <- guardIt $ decodePushPromiseFrame header pl unless (isServerInitiated sid) $ E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "wrong sid for push promise"@@ -132,23 +151,25 @@ strm <- openStream ctx sid FramePushPromise insertCache method path strm $ roleInfo ctx _ -> return ()-processFrame ctx@Context{..} recvN typhdr@(ftyp, header) = do- settings <- readIORef http2settings+processFrame ctx@Context{..} conf typhdr@(ftyp, header) = do+ -- My SETTINGS_MAX_FRAME_SIZE+ -- My SETTINGS_ENABLE_PUSH+ settings <- readIORef mySettings case checkFrameHeader settings typhdr of Left (FrameDecodeError ec sid msg) -> E.throwIO $ ConnectionErrorIsSent ec sid msg- Right _ -> controlOrStream ctx recvN ftyp header+ Right _ -> controlOrStream ctx conf ftyp header ---------------------------------------------------------------- -controlOrStream :: Context -> RecvN -> FrameType -> FrameHeader -> IO ()-controlOrStream ctx@Context{..} recvN ftyp header@FrameHeader{streamId, payloadLength}+controlOrStream :: Context -> Config -> FrameType -> FrameHeader -> IO ()+controlOrStream ctx@Context{..} conf@Config{..} ftyp header@FrameHeader{streamId, payloadLength} | isControl streamId = do- pl <- recvN payloadLength- control ftyp header pl ctx+ pl <- confReadN payloadLength+ control ftyp header pl ctx conf | otherwise = do checkContinued mstrm <- getStream ctx ftyp streamId- pl <- recvN payloadLength+ pl <- confReadN payloadLength case mstrm of Just strm -> do state0 <- readStreamState strm@@ -194,7 +215,8 @@ tlr <- newIORef Nothing q <- newTQueueIO setStreamState ctx strm $ Open (Body q mcl bodyLength tlr)- bodySource <- mkSource (updateWindow controlQ streamId) q+ incref <- newIORef 0+ bodySource <- mkSource q $ updateWindow controlQ streamId incref let inpObj = InpObj tbl mcl (readSource bodySource) tlr if isServer ctx then do let si = toServerInfo roleInfo@@ -244,7 +266,12 @@ setPeerStreamID ctx streamId cnt <- readIORef concurrency -- Checking the limitation of concurrency- when (cnt >= maxConcurrency) $ E.throwIO $ StreamErrorIsSent RefusedStream streamId+ -- My SETTINGS_MAX_CONCURRENT_STREAMS+ mMaxConc <- maxConcurrentStreams <$> readIORef mySettings+ case mMaxConc of+ Nothing -> return ()+ Just maxConc -> when (cnt >= maxConc) $+ E.throwIO $ StreamErrorIsSent RefusedStream streamId Just <$> openStream ctx streamId ftyp | otherwise = undefined -- never reach @@ -252,31 +279,39 @@ type Payload = ByteString -control :: FrameType -> FrameHeader -> Payload -> Context -> IO ()-control FrameSettings header@FrameHeader{flags,streamId} bs Context{http2settings, controlQ, firstSettings, streamTable, settingsRate} = do- SettingsFrame alist <- guardIt $ decodeSettingsFrame header bs- traverse_ E.throwIO $ checkSettingsList alist- -- HTTP/2 Setting from a browser- unless (testAck flags) $ do+control :: FrameType -> FrameHeader -> Payload -> Context -> Config -> IO ()+control FrameSettings header@FrameHeader{flags,streamId} bs Context{myFirstSettings,myPendingAlist,mySettings,controlQ,settingsRate} conf = do+ SettingsFrame peerAlist <- guardIt $ decodeSettingsFrame header bs+ traverse_ E.throwIO $ checkSettingsList peerAlist+ if testAck flags then do+ when (peerAlist /= []) $+ E.throwIO $ ConnectionErrorIsSent FrameSizeError streamId "ack settings has a body"+ mAlist <- readIORef myPendingAlist+ case mAlist of+ Nothing -> return () -- fixme+ Just myAlist -> do+ modifyIORef' mySettings $ \old -> updateSettings old myAlist+ writeIORef myPendingAlist Nothing+ else do -- Settings Flood - CVE-2019-9515 rate <- getRate settingsRate- if rate > settingsRateLimit then+ when (rate > settingsRateLimit) $ E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "too many settings"+ let ack = settingsFrame setAck []+ sent <- readIORef myFirstSettings+ if sent then do+ let setframe = CFrames (Just peerAlist) [ack]+ enqueueControl controlQ setframe else do- oldws <- initialWindowSize <$> readIORef http2settings- modifyIORef' http2settings $ \old -> updateSettings old alist- newws <- initialWindowSize <$> readIORef http2settings- let diff = newws - oldws- when (diff /= 0) $ updateAllStreamWindow (+ diff) streamTable- let frame = settingsFrame setAck []- sent <- readIORef firstSettings- let setframe- | sent = CSettings frame alist- | otherwise = CSettings0 initialFrame frame alist- unless sent $ writeIORef firstSettings True+ -- Server side only+ writeIORef myFirstSettings True+ let myAlist = myInitialAlist conf+ writeIORef myPendingAlist $ Just myAlist+ let frames = initialFrames myAlist ++ [ack]+ setframe = CFrames (Just peerAlist) frames enqueueControl controlQ setframe -control FramePing FrameHeader{flags,streamId} bs Context{controlQ,pingRate} =+control FramePing FrameHeader{flags,streamId} bs Context{controlQ,pingRate} _ = unless (testAck flags) $ do -- Ping Flood - CVE-2019-9512 rate <- getRate pingRate@@ -284,26 +319,25 @@ E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "too many ping" else do let frame = pingFrame bs- enqueueControl controlQ $ CFrame frame+ enqueueControl controlQ $ CFrames Nothing [frame] -control FrameGoAway header bs Context{controlQ} = do- enqueueControl controlQ CFinish+control FrameGoAway header bs _ _ = do GoAwayFrame sid err msg <- guardIt $ decodeGoAwayFrame header bs if err == NoError then E.throwIO ConnectionIsClosed else E.throwIO $ ConnectionErrorIsReceived err sid $ Short.toShort msg -control FrameWindowUpdate header@FrameHeader{streamId} bs Context{connectionWindow} = do+control FrameWindowUpdate header@FrameHeader{streamId} bs Context{txConnectionWindow} _ = do WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs w <- atomically $ do- w0 <- readTVar connectionWindow+ w0 <- readTVar txConnectionWindow let w1 = w0 + n- writeTVar connectionWindow w1+ writeTVar txConnectionWindow w1 return w1 when (isWindowOverflow w) $ E.throwIO $ ConnectionErrorIsSent FlowControlError streamId "control window should be less than 2^31" -control _ _ _ _ =+control _ _ _ _ _ = -- must not reach here return () @@ -367,11 +401,9 @@ stream FrameData FrameHeader{flags,payloadLength} _bs- Context{controlQ} s@(HalfClosedLocal _)+ ctx s@(HalfClosedLocal _) _ = do- when (payloadLength /= 0) $ do- let frame = windowUpdateFrame 0 payloadLength- enqueueControl controlQ $ CFrame frame+ rxConnectionWindowIncrement ctx payloadLength let endOfStream = testEndStream flags if endOfStream then do return HalfClosedRemote@@ -381,8 +413,9 @@ stream FrameData header@FrameHeader{flags,payloadLength,streamId} bs- Context{emptyFrameRate} s@(Open (Body q mcl bodyLength _))+ ctx@Context{emptyFrameRate} s@(Open (Body q mcl bodyLength _)) _ = do+ rxConnectionWindowIncrement ctx payloadLength DataFrame body <- guardIt $ decodeDataFrame header bs len0 <- readIORef bodyLength let len = len0 + payloadLength@@ -443,9 +476,8 @@ when (isWindowOverflow w) $ E.throwIO $ StreamErrorIsSent FlowControlError streamId return s -stream FrameRSTStream header@FrameHeader{streamId} bs ctx@Context{..} _ strm = do- enqueueControl controlQ CFinish- RSTStreamFrame err <- guardIt $ decoderstStreamFrame header bs+stream FrameRSTStream header@FrameHeader{streamId} bs ctx _ strm = do+ RSTStreamFrame err <- guardIt $ decodeRSTStreamFrame header bs let cc = Reset err closed ctx strm cc E.throwIO $ StreamErrorIsReceived err streamId@@ -473,16 +505,20 @@ (IORef ByteString) (IORef Bool) -mkSource :: (Int -> IO ()) -> TQueue ByteString -> IO Source-mkSource update q = Source update q <$> newIORef "" <*> newIORef False+mkSource :: TQueue ByteString -> (Int -> IO ()) -> IO Source+mkSource q update = Source update q <$> newIORef "" <*> newIORef False -updateWindow :: TQueue Control -> StreamId -> Int -> IO ()-updateWindow _ _ 0 = return ()-updateWindow controlQ sid len = enqueueControl controlQ $ CFrame frame- where- frame1 = windowUpdateFrame 0 len- frame2 = windowUpdateFrame sid len- frame = frame1 `BS.append` frame2+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 readSource :: Source -> IO ByteString readSource (Source update q refBS refEOF) = do@@ -504,3 +540,16 @@ 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
@@ -1,6 +1,7 @@ {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-} module Network.HTTP2.Arch.Sender ( frameSender@@ -13,13 +14,14 @@ import qualified Data.ByteString as BS import Data.ByteString.Builder (Builder) import qualified Data.ByteString.Builder.Extra as B-import Foreign.Ptr (plusPtr)+import Data.IORef (readIORef, writeIORef, modifyIORef')+import Foreign.Ptr (plusPtr, minusPtr) import Network.ByteOrder import qualified UnliftIO.Exception as E import UnliftIO.STM import Imports-import Network.HPACK (setLimitForEncoding, toHeaderTable)+import Network.HPACK (setLimitForEncoding, toHeaderTable, TokenHeaderList) import Network.HTTP2.Arch.Config import Network.HTTP2.Arch.Context import Network.HTTP2.Arch.EncodeFrame@@ -59,15 +61,45 @@ | O (Output Stream) | Flush +wrapException :: E.SomeException -> IO ()+wrapException se+ | Just (e :: HTTP2Error) <- E.fromException se = E.throwIO e+ | otherwise = E.throwIO $ BadThingHappen se+ frameSender :: Context -> Config -> Manager -> IO ()-frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}+frameSender ctx@Context{outputQ,controlQ,txConnectionWindow,encodeDynamicTable,peerSettings,streamTable,outputBufferLimit} Config{..}- mgr = loop 0+ mgr = loop 0 `E.catch` wrapException where+ ----------------------------------------------------------------+ loop :: Offset -> IO ()+ loop off = do+ x <- atomically $ dequeue off+ case x of+ C ctl -> flushN off >> control ctl >> loop 0+ O out -> outputOrEnqueueAgain out off >>= flushIfNecessary >>= loop+ Flush -> flushN off >> loop 0++ -- Flush the connection buffer to the socket, where the first 'n' bytes of+ -- the buffer are filled.+ flushN :: Offset -> IO ()+ flushN 0 = return ()+ flushN n = bufferIO confWriteBuffer n confSendAll++ flushIfNecessary :: Offset -> IO Offset+ flushIfNecessary off = do+ buflim <- readIORef outputBufferLimit+ if off <= buflim - 512 then+ return off+ else do+ flushN off+ return 0++ dequeue :: Offset -> STM Switch dequeue off = do isEmpty <- isEmptyTQueue controlQ if isEmpty then do- w <- readTVar connectionWindow+ w <- readTVar txConnectionWindow checkSTM (w > 0) emp <- isEmptyTQueue outputQ if emp then@@ -77,51 +109,50 @@ else C <$> readTQueue controlQ - hardLimit = confBufferSize - 512-- loop off = do- x <- atomically $ dequeue off- case x of- C ctl -> do- when (off /= 0) $ flushN off- off' <- control ctl off- when (off' >= 0) $ loop off'- O out -> do- off' <- outputOrEnqueueAgain out off- case off' of- 0 -> loop 0- _ | off' > hardLimit -> flushN off' >> loop 0- | otherwise -> loop off'- Flush -> flushN off >> loop 0-- control CFinish _ = return (-1)- control (CGoaway frame) _ = confSendAll frame >> return (-1)- control (CFrame frame) _ = confSendAll frame >> return 0- control (CSettings frame alist) _ = do- confSendAll frame- setLimit alist- return 0- control (CSettings0 frame1 frame2 alist) off = do -- off == 0, just in case- let buf = confWriteBuffer `plusPtr` off- off' = off + BS.length frame1 + BS.length frame2- buf' <- copy buf frame1- void $ copy buf' frame2- setLimit alist- return off'+ ----------------------------------------------------------------+ copyAll [] buf = return buf+ copyAll (x:xs) buf = copy buf x >>= copyAll xs - {-# INLINE setLimit #-}- setLimit alist = case lookup SettingsHeaderTableSize alist of- Nothing -> return ()- Just siz -> setLimitForEncoding siz encodeDynamicTable+ -- called with off == 0+ control :: Control -> IO ()+ control (CFinish e) = E.throwIO e+ control (CFrames ms xs) = do+ buf <- copyAll xs confWriteBuffer+ let off = buf `minusPtr` confWriteBuffer+ flushN off+ case ms of+ 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+ -- Peer SETTINGS_MAX_FRAME_SIZE+ case lookup SettingsMaxFrameSize peerAlist of+ Nothing -> return ()+ Just payloadLen -> do+ let dlim = payloadLen + frameHeaderLength+ buflim | confBufferSize >= dlim = dlim+ | otherwise = confBufferSize+ writeIORef outputBufferLimit buflim+ -- Peer SETTINGS_HEADER_TABLE_SIZE+ case lookup SettingsHeaderTableSize peerAlist of+ Nothing -> return ()+ Just siz -> setLimitForEncoding siz encodeDynamicTable + ----------------------------------------------------------------+ output :: Output Stream -> Offset -> WindowSize -> IO Offset output out@(Output strm OutObj{} (ONext curr tlrmkr) _ sentinel) off0 lim = do -- Data frame payload+ buflim <- readIORef outputBufferLimit let payloadOff = off0 + frameHeaderLength datBuf = confWriteBuffer `plusPtr` payloadOff- datBufSiz = confBufferSize - payloadOff- Next datPayloadLen mnext <- curr datBuf datBufSiz lim -- checkme+ datBufSiz = buflim - payloadOff+ Next datPayloadLen reqflush mnext <- curr datBuf datBufSiz lim -- checkme NextTrailersMaker tlrmkr' <- runTrailersMaker tlrmkr datBuf datPayloadLen- fillDataHeaderEnqueueNext strm off0 datPayloadLen mnext tlrmkr' sentinel out+ fillDataHeaderEnqueueNext strm off0 datPayloadLen mnext tlrmkr' sentinel out reqflush output out@(Output strm (OutObj hdr body tlrmkr) OObj mtbq _) off0 lim = do -- Header frame and Continuation frame@@ -130,8 +161,8 @@ OutBodyNone -> True _ -> False (ths,_) <- toHeaderTable $ fixHeaders hdr- kvlen <- headerContinue sid ths endOfStream off0- off <- sendHeadersIfNecessary $ off0 + frameHeaderLength + kvlen+ off' <- headerContinue sid ths endOfStream off0+ off <- flushIfNecessary off' case body of OutBodyNone -> do -- halfClosedLocal calls closed which removes@@ -162,12 +193,13 @@ -- Frame id should be associated stream id from the client. let sid = streamNumber strm len <- pushPromise pid sid ths off0- off <- sendHeadersIfNecessary $ off0 + frameHeaderLength + len+ off <- flushIfNecessary $ off0 + frameHeaderLength + len output out{outputType=OObj} off lim output _ _ _ = undefined -- never reach - outputOrEnqueueAgain :: Output Stream -> Int -> IO Int+ ----------------------------------------------------------------+ outputOrEnqueueAgain :: Output Stream -> Offset -> IO Offset outputOrEnqueueAgain out@(Output strm _ otyp _ _) off = E.handle resetStream $ do state <- readStreamState strm if isHalfClosedLocal state then@@ -195,61 +227,64 @@ forkAndEnqueueWhenReady (waitStreamWindowSize strm) outputQ out mgr return off else do- cws <- readTVarIO connectionWindow -- not 0+ cws <- readTVarIO txConnectionWindow -- not 0 let lim = min cws sws output out off lim resetStream e = do closed ctx strm (ResetByMe e) let rst = resetFrame InternalError $ streamNumber strm- enqueueControl controlQ $ CFrame rst+ enqueueControl controlQ $ CFrames Nothing [rst] return off - {-# INLINE flushN #-}- -- Flush the connection buffer to the socket, where the first 'n' bytes of- -- the buffer are filled.- flushN :: Int -> IO ()- flushN n = bufferIO confWriteBuffer n confSendAll-- headerContinue sid ths endOfStream off = do- let offkv = off + frameHeaderLength- let bufkv = confWriteBuffer `plusPtr` offkv- limkv = confBufferSize - offkv- (hs,kvlen) <- hpackEncodeHeader ctx bufkv limkv ths- let flag0 = case hs of- [] -> setEndHeader defaultFlags- _ -> defaultFlags- flag = if endOfStream then setEndStream flag0 else flag0- let buf = confWriteBuffer `plusPtr` off- fillFrameHeader FrameHeaders kvlen sid flag buf- continue sid kvlen hs-- bufHeaderPayload = confWriteBuffer `plusPtr` frameHeaderLength- headerPayloadLim = confBufferSize - frameHeaderLength+ ----------------------------------------------------------------+ headerContinue :: StreamId -> TokenHeaderList -> Bool -> Offset -> IO Offset+ headerContinue sid ths0 endOfStream off0 = do+ buflim <- readIORef outputBufferLimit+ let offkv = off0 + frameHeaderLength+ bufkv = confWriteBuffer `plusPtr` offkv+ limkv = buflim - offkv+ (ths,kvlen) <- hpackEncodeHeader ctx bufkv limkv ths0+ if kvlen == 0 then+ continue off0 ths FrameHeaders+ else do+ let flag = getFlag ths+ buf = confWriteBuffer `plusPtr` off0+ off = offkv + kvlen+ fillFrameHeader FrameHeaders kvlen sid flag buf+ continue off ths FrameContinuation+ where+ eos = if endOfStream then setEndStream else id+ getFlag [] = eos $ setEndHeader defaultFlags+ getFlag _ = eos $ defaultFlags - continue _ kvlen [] = return kvlen- continue sid kvlen ths = do- flushN $ kvlen + frameHeaderLength- -- Now off is 0- (ths', kvlen') <- hpackEncodeHeaderLoop ctx bufHeaderPayload headerPayloadLim ths- when (ths == ths') $ E.throwIO $ ConnectionErrorIsSent CompressionError sid "cannot compress the header"- let flag = case ths' of- [] -> setEndHeader defaultFlags- _ -> defaultFlags- fillFrameHeader FrameContinuation kvlen' sid flag confWriteBuffer- continue sid kvlen' ths'+ continue :: Offset -> TokenHeaderList -> FrameType -> IO Offset+ continue off [] _ = return off+ continue off ths ft = do+ flushN off+ -- Now off is 0+ buflim <- readIORef outputBufferLimit+ let bufHeaderPayload = confWriteBuffer `plusPtr` frameHeaderLength - {-# INLINE sendHeadersIfNecessary #-}- -- Send headers if there is not room for a 1-byte data frame, and return- -- the offset of the next frame's first header byte.- sendHeadersIfNecessary off- -- True if the connection buffer has room for a 1-byte data frame.- | off + frameHeaderLength < confBufferSize = return off- | otherwise = do- flushN off- return 0+ headerPayloadLim = buflim - frameHeaderLength+ (ths', kvlen') <- hpackEncodeHeaderLoop ctx bufHeaderPayload headerPayloadLim ths+ when (ths == ths') $ E.throwIO $ ConnectionErrorIsSent CompressionError sid "cannot compress the header"+ let flag = getFlag ths'+ off' = frameHeaderLength + kvlen'+ fillFrameHeader ft kvlen' sid flag confWriteBuffer+ continue off' ths' FrameContinuation + ----------------------------------------------------------------+ fillDataHeaderEnqueueNext :: Stream+ -> Offset+ -> Int+ -> Maybe DynaNext+ -> (Maybe ByteString -> IO NextTrailersMaker)+ -> IO ()+ -> Output Stream+ -> Bool+ -> IO Offset fillDataHeaderEnqueueNext strm@Stream{streamWindow,streamNumber}- off datPayloadLen Nothing tlrmkr tell _ = do+ off datPayloadLen Nothing tlrmkr tell _ reqflush = do let buf = confWriteBuffer `plusPtr` off off' = off + frameHeaderLength + datPayloadLen (mtrailers, flag) <- do@@ -262,34 +297,47 @@ off'' <- handleTrailers mtrailers off' void tell when (isServer ctx) $ halfClosedLocal ctx strm Finished- atomically $ modifyTVar' connectionWindow (subtract datPayloadLen)+ atomically $ modifyTVar' txConnectionWindow (subtract datPayloadLen) atomically $ modifyTVar' streamWindow (subtract datPayloadLen)- return off''+ if reqflush then do+ flushN off''+ return 0+ else+ return off'' where handleTrailers Nothing off0 = return off0 handleTrailers (Just trailers) off0 = do (ths,_) <- toHeaderTable trailers- kvlen <- headerContinue streamNumber ths True off0- sendHeadersIfNecessary $ off0 + frameHeaderLength + kvlen+ headerContinue streamNumber ths True off0 fillDataHeaderEnqueueNext _- off 0 (Just next) tlrmkr _ out = do+ off 0 (Just next) tlrmkr _ out reqflush = do let out' = out { outputType = ONext next tlrmkr } enqueueOutput outputQ out'- return off+ if reqflush then do+ flushN off+ return 0+ else+ return off fillDataHeaderEnqueueNext Stream{streamWindow,streamNumber}- off datPayloadLen (Just next) tlrmkr _ out = do+ 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' connectionWindow (subtract datPayloadLen)+ atomically $ modifyTVar' txConnectionWindow (subtract datPayloadLen) atomically $ modifyTVar' streamWindow (subtract datPayloadLen) let out' = out { outputType = ONext next tlrmkr } enqueueOutput outputQ out'- return off'+ if reqflush then do+ flushN off'+ return 0+ else+ return off' + ----------------------------------------------------------------+ pushPromise :: StreamId -> StreamId -> TokenHeaderList -> Offset -> IO Int pushPromise pid sid ths off = do let offsid = off + frameHeaderLength -- checkme bufsid = confWriteBuffer `plusPtr` offsid@@ -304,10 +352,16 @@ fillFrameHeader FramePushPromise len pid flag buf return len + ---------------------------------------------------------------- {-# INLINE fillFrameHeader #-}+ fillFrameHeader :: FrameType -> Int -> StreamId -> FrameFlags -> Buffer -> IO () fillFrameHeader ftyp len sid flag buf = encodeFrameHeaderBuf ftyp hinfo buf where- hinfo = FrameHeader len flag sid+ hinfo = FrameHeader {+ payloadLength = len+ , flags = flag+ , streamId = sid+ } -- | Running trailers-maker. --@@ -333,8 +387,8 @@ fillStreamBodyGetNext :: IO (Maybe StreamingChunk) -> DynaNext fillStreamBodyGetNext takeQ buf siz lim = do let room = min siz lim- (leftover, cont, len) <- runStreamBuilder buf room takeQ- return $ nextForStream takeQ leftover cont len+ (cont, len, reqflush, leftover) <- runStreamBuilder buf room takeQ+ return $ nextForStream cont len reqflush leftover takeQ ---------------------------------------------------------------- @@ -361,39 +415,42 @@ nextForBuilder :: BytesFilled -> B.Next -> Next nextForBuilder len B.Done- = Next len Nothing+ = Next len True Nothing -- let's flush nextForBuilder len (B.More _ writer)- = Next len $ Just (fillBufBuilder (LOne writer))+ = Next len False $ Just (fillBufBuilder (LOne writer)) nextForBuilder len (B.Chunk bs writer)- = Next len $ Just (fillBufBuilder (LTwo bs writer))+ = Next len False $ Just (fillBufBuilder (LTwo bs writer)) ---------------------------------------------------------------- runStreamBuilder :: Buffer -> BufferSize -> IO (Maybe StreamingChunk)- -> IO (Leftover, Bool, BytesFilled)+ -> IO (Bool -- continue+ ,BytesFilled+ ,Bool -- require flusing+ ,Leftover) runStreamBuilder buf0 room0 takeQ = loop buf0 room0 0 where loop buf room total = do mbuilder <- takeQ case mbuilder of- Nothing -> return (LZero, True, total)+ Nothing -> return (True, total, False, LZero) Just (StreamingBuilder builder) -> do (len, signal) <- B.runBuilder builder buf room let total' = total + len case signal of B.Done -> loop (buf `plusPtr` len) (room - len) total'- B.More _ writer -> return (LOne writer, True, total')- B.Chunk bs writer -> return (LTwo bs writer, True, total')- Just StreamingFlush -> return (LZero, True, total)- Just StreamingFinished -> return (LZero, False, total)+ B.More _ writer -> return (True, total', False, LOne writer)+ B.Chunk bs writer -> return (True, total', False, LTwo bs writer)+ Just StreamingFlush -> return (True, total, True, LZero)+ Just StreamingFinished -> return (False, total, True, LZero) fillBufStream :: Leftover -> IO (Maybe StreamingChunk) -> DynaNext fillBufStream leftover0 takeQ buf0 siz0 lim0 = do let room0 = min siz0 lim0 case leftover0 of LZero -> do- (leftover, cont, len) <- runStreamBuilder buf0 room0 takeQ- getNext leftover cont len+ (cont, len, reqflush, leftover) <- runStreamBuilder buf0 room0 takeQ+ getNext cont len reqflush leftover LOne writer -> write writer buf0 room0 0 LTwo bs writer | BS.length bs <= room0 -> do@@ -403,29 +460,34 @@ | otherwise -> do let (bs1,bs2) = BS.splitAt room0 bs void $ copy buf0 bs1- getNext (LTwo bs2 writer) True room0+ getNext True room0 False $ LTwo bs2 writer where- getNext l b r = return $ nextForStream takeQ l b r+ getNext :: Bool -> BytesFilled -> Bool -> Leftover -> IO Next+ getNext cont len reqflush l = return $ nextForStream cont len reqflush l takeQ++ write :: (Buffer -> BufferSize -> IO (Int, B.Next))+ -> Buffer -> BufferSize -> Int+ -> IO Next write writer1 buf room sofar = do (len, signal) <- writer1 buf room case signal of B.Done -> do- (leftover, cont, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) takeQ+ (cont, extra, reqflush, leftover) <- runStreamBuilder (buf `plusPtr` len) (room - len) takeQ let total = sofar + len + extra- getNext leftover cont total+ getNext cont total reqflush leftover B.More _ writer -> do let total = sofar + len- getNext (LOne writer) True total+ getNext True total False $ LOne writer B.Chunk bs writer -> do let total = sofar + len- getNext (LTwo bs writer) True total+ getNext True total False $ LTwo bs writer -nextForStream :: IO (Maybe StreamingChunk)- -> Leftover -> Bool -> BytesFilled+nextForStream :: Bool -> BytesFilled -> Bool+ -> Leftover -> IO (Maybe StreamingChunk) -> Next-nextForStream _ _ False len = Next len Nothing-nextForStream takeQ leftOrZero True len =- Next len $ Just (fillBufStream leftOrZero takeQ)+nextForStream False len reqflush _ _ = Next len reqflush Nothing+nextForStream True len reqflush leftOrZero takeQ =+ Next len reqflush $ Just (fillBufStream leftOrZero takeQ) ---------------------------------------------------------------- @@ -438,10 +500,10 @@ return $ nextForFile len' pread (start + len) (bytes - len) refresh nextForFile :: BytesFilled -> PositionRead -> FileOffset -> ByteCount -> IO () -> Next-nextForFile 0 _ _ _ _ = Next 0 Nothing-nextForFile len _ _ 0 _ = Next len Nothing+nextForFile 0 _ _ _ _ = Next 0 True Nothing -- let's flush+nextForFile len _ _ 0 _ = Next len False Nothing nextForFile len pread start bytes refresh =- Next len $ Just (fillBufFile pread start bytes refresh)+ Next len False $ Just $ fillBufFile pread start bytes refresh {-# INLINE mini #-} mini :: Int -> Int64 -> Int64
Network/HTTP2/Arch/Types.hs view
@@ -182,15 +182,14 @@ type BytesFilled = Int -data Next = Next BytesFilled (Maybe DynaNext)+data Next = Next BytesFilled -- payload length+ Bool -- require flushing+ (Maybe DynaNext) ---------------------------------------------------------------- -data Control = CFinish- | CGoaway ByteString- | CFrame ByteString- | CSettings ByteString SettingsList- | CSettings0 ByteString ByteString SettingsList+data Control = CFinish HTTP2Error+ | CFrames (Maybe SettingsList) [ByteString] ---------------------------------------------------------------- @@ -234,7 +233,7 @@ checkSettingsValue (SettingsEnablePush,v) | v /= 0 && v /= 1 = Just $ ConnectionErrorIsSent ProtocolError 0 "enable push must be 0 or 1" checkSettingsValue (SettingsInitialWindowSize,v)- | v > 2147483647 = Just $ ConnectionErrorIsSent FlowControlError 0 "Window size must be less than or equal to 65535"+ | v > maxWindowSize = Just $ ConnectionErrorIsSent FlowControlError 0 "Window size must be less than or equal to 65535" checkSettingsValue (SettingsMaxFrameSize,v)- | v < 16384 || v > 16777215 = Just $ ConnectionErrorIsSent ProtocolError 0 "Max frame size must be in between 16384 and 16777215"+ | v < defaultPayloadLength || v > maxPayloadLength = Just $ ConnectionErrorIsSent ProtocolError 0 "Max frame size must be in between 16384 and 16777215" checkSettingsValue _ = Nothing
Network/HTTP2/Client/Run.hs view
@@ -3,10 +3,12 @@ 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+import UnliftIO.STM import Imports import Network.HTTP2.Arch@@ -24,31 +26,41 @@ run :: ClientConfig -> Config -> Client a -> IO a run ClientConfig{..} conf@Config{..} client = do clientInfo <- newClientInfo scheme authority cacheLimit- ctx <- newContext clientInfo+ ctx <- newContext clientInfo confBufferSize mgr <- start confTimeoutManager let runBackgroundThreads = do- let runReceiver = frameReceiver ctx confReadN- runSender = frameSender ctx conf mgr+ let runReceiver = frameReceiver ctx conf+ runSender = frameSender ctx conf mgr concurrently_ runReceiver runSender exchangeSettings conf ctx let runClient = do- x <- client $ sendRequest ctx scheme authority+ x <- client $ sendRequest ctx mgr scheme authority let frame = goawayFrame 0 NoError "graceful closing"- enqueueControl (controlQ ctx) $ CGoaway frame+ enqueueControl (controlQ ctx) $ CFrames Nothing [frame] return x ex <- race runBackgroundThreads runClient `E.finally` stop mgr case ex of Left () -> undefined -- never reach Right x -> return x -sendRequest :: Context -> Scheme -> Authority -> Request -> (Response -> IO a) -> IO a-sendRequest ctx@Context{..} scheme auth (Request req) processResponse = do+sendRequest :: Context -> Manager -> Scheme -> Authority -> Request -> (Response -> IO a) -> IO a+sendRequest ctx@Context{..} mgr scheme auth (Request req) processResponse = do+ -- Checking push promises let hdr0 = outObjHeaders req method = fromMaybe (error "sendRequest:method") $ lookup ":method" hdr0 path = fromMaybe (error "sendRequest:path") $ lookup ":path" hdr0 mstrm0 <- lookupCache method path roleInfo strm <- case mstrm0 of Nothing -> do+ -- Arch/Sender is originally implemented for servers where+ -- the ordering of responses can be out-of-order.+ -- But for clients, the ordering must be maintained.+ -- To implement this, 'outputQStreamID' is used.+ -- Also, for 'OutBodyStreaming', TBQ must not be empty+ -- when its 'Output' is enqueued into 'outputQ'.+ -- Otherwise, it would be re-enqueue because of empty+ -- resulting in out-of-order.+ -- To implement this, 'tbqNonMmpty' is used. let hdr1 | scheme /= "" = (":scheme", scheme) : hdr0 | otherwise = hdr0 hdr2 | auth /= "" = (":authority", auth) : hdr1@@ -56,16 +68,40 @@ req' = req { outObjHeaders = hdr2 } sid <- getMyNewStreamId ctx newstrm <- openStream ctx sid FrameHeaders- enqueueOutput outputQ $ Output newstrm req' OObj Nothing (return ())+ case outObjBody req of+ 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+ let push b = atomically $ do+ writeTBQueue tbq (StreamingBuilder b)+ writeTVar tbqNonMmpty True+ flush = atomically $ writeTBQueue tbq StreamingFlush+ strmbdy push flush+ atomically $ writeTBQueue tbq StreamingFinished+ atomically $ do+ sidOK <- readTVar outputQStreamID+ ready <- readTVar tbqNonMmpty+ check (sidOK == sid && ready)+ writeTVar outputQStreamID (sid + 2)+ writeTQueue outputQ $ Output newstrm req' OObj (Just tbq) (return ())+ _ -> atomically $ do+ sidOK <- readTVar outputQStreamID+ check (sidOK == sid)+ writeTVar outputQStreamID (sid + 2)+ writeTQueue outputQ $ Output newstrm req' OObj Nothing (return ()) return newstrm Just strm0 -> return strm0 rsp <- takeMVar $ streamInput strm processResponse $ Response rsp exchangeSettings :: Config -> Context -> IO ()-exchangeSettings Config{..} Context{..} = do- confSendAll connectionPreface- let setframe = CSettings initialFrame [] -- fixme alist- writeIORef firstSettings True+exchangeSettings conf Context{..} = do+ writeIORef myFirstSettings True+ let myAlist = myInitialAlist conf+ writeIORef myPendingAlist $ Just myAlist+ let frames = initialFrames myAlist+ setframe = CFrames Nothing (connectionPreface:frames) enqueueControl controlQ setframe-
Network/HTTP2/Frame.hs view
@@ -60,9 +60,12 @@ , Settings(..) , defaultSettings , updateSettings+ -- * Payload length+ , defaultPayloadLength+ , maxPayloadLength -- * Window , WindowSize- , defaultInitialWindowSize+ , defaultWindowSize , maxWindowSize , isWindowOverflow -- * Error code@@ -73,7 +76,6 @@ , connectionPreface , connectionPrefaceLength , frameHeaderLength- , maxPayloadLength , recommendedConcurrency -- * Deprecated , ErrorCodeId
Network/HTTP2/Frame/Decode.hs view
@@ -13,7 +13,7 @@ , decodeDataFrame , decodeHeadersFrame , decodePriorityFrame- , decoderstStreamFrame+ , decodeRSTStreamFrame , decodeSettingsFrame , decodePushPromiseFrame , decodePingFrame@@ -77,7 +77,7 @@ checkFrameHeader :: Settings -> (FrameType, FrameHeader) -> Either FrameDecodeError (FrameType, FrameHeader)-checkFrameHeader Settings {..} typfrm@(typ,FrameHeader{..})+checkFrameHeader Settings{..} typfrm@(typ,FrameHeader{..}) | payloadLength > maxFrameSize = Left $ FrameDecodeError FrameSizeError streamId "exceeds maximum frame size" | typ `elem` nonZeroFrameTypes && isControl streamId =@@ -143,7 +143,7 @@ [ decodeDataFrame , decodeHeadersFrame , decodePriorityFrame- , decoderstStreamFrame+ , decodeRSTStreamFrame , decodeSettingsFrame , decodePushPromiseFrame , decodePingFrame@@ -185,8 +185,8 @@ decodePriorityFrame _ bs = Right $ PriorityFrame $ priority bs -- | Frame payload decoder for RST_STREAM frame.-decoderstStreamFrame :: FramePayloadDecoder-decoderstStreamFrame _ bs = Right $ RSTStreamFrame $ toErrorCode (N.word32 bs)+decodeRSTStreamFrame :: FramePayloadDecoder+decodeRSTStreamFrame _ bs = Right $ RSTStreamFrame $ toErrorCode (N.word32 bs) -- | Frame payload decoder for SETTINGS frame. decodeSettingsFrame :: FramePayloadDecoder@@ -203,12 +203,9 @@ rawSetting <- N.peek16 p 0 let k = toSettingsKey rawSetting n' = n - 1- if k < minSettingsKey || k > maxSettingsKey then- settings n' (p +. 6) builder -- ignoring unknown one (Section 6.5.2)- else do- w32 <- N.peek32 p 2- let v = fromIntegral w32- settings n' (p +. 6) (builder. ((k,v):))+ w32 <- N.peek32 p 2+ let v = fromIntegral w32+ settings n' (p +. 6) (builder. ((k,v):)) -- | Frame payload decoder for PUSH_PROMISE frame. decodePushPromiseFrame :: FramePayloadDecoder
Network/HTTP2/Frame/Types.hs view
@@ -167,27 +167,27 @@ , maxConcurrentStreams :: Maybe Int , initialWindowSize :: WindowSize , maxFrameSize :: Int- , maxHeaderBlockSize :: Maybe Int+ , maxHeaderListSize :: Maybe Int } deriving (Show) -- | The default settings. -- -- >>> defaultSettings--- Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Nothing, initialWindowSize = 65535, maxFrameSize = 16384, maxHeaderBlockSize = Nothing}+-- Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Nothing, initialWindowSize = 65535, maxFrameSize = 16384, maxHeaderListSize = Nothing} defaultSettings :: Settings defaultSettings = Settings {- headerTableSize = 4096+ headerTableSize = 4096 -- defaultDynamicTableSize , enablePush = True , maxConcurrentStreams = Nothing- , initialWindowSize = defaultInitialWindowSize- , maxFrameSize = 16384- , maxHeaderBlockSize = Nothing+ , initialWindowSize = defaultWindowSize+ , maxFrameSize = defaultPayloadLength+ , maxHeaderListSize = Nothing } -- | Updating settings. -- -- >>> updateSettings defaultSettings [(SettingsEnablePush,0),(SettingsMaxHeaderBlockSize,200)]--- Settings {headerTableSize = 4096, enablePush = False, maxConcurrentStreams = Nothing, initialWindowSize = 65535, maxFrameSize = 16384, maxHeaderBlockSize = Just 200}+-- Settings {headerTableSize = 4096, enablePush = False, maxConcurrentStreams = Nothing, initialWindowSize = 65535, maxFrameSize = 16384, maxHeaderListSize = Just 200} updateSettings :: Settings -> SettingsList -> Settings updateSettings settings kvs = foldl' update settings kvs where@@ -197,7 +197,7 @@ update def (SettingsMaxConcurrentStreams,x) = def { maxConcurrentStreams = Just x } update def (SettingsInitialWindowSize,x) = def { initialWindowSize = x } update def (SettingsMaxFrameSize,x) = def { maxFrameSize = x }- update def (SettingsMaxHeaderBlockSize,x) = def { maxHeaderBlockSize = Just x }+ update def (SettingsMaxHeaderBlockSize,x) = def { maxHeaderListSize = Just x } update def _ = def -- | The type for window size.@@ -205,10 +205,10 @@ -- | The default initial window size. ----- >>> defaultInitialWindowSize+-- >>> defaultWindowSize -- 65535-defaultInitialWindowSize :: WindowSize-defaultInitialWindowSize = 65535+defaultWindowSize :: WindowSize+defaultWindowSize = 65535 -- | The maximum window size. --@@ -338,12 +338,19 @@ ---------------------------------------------------------------- --- | The maximum length of HTTP/2 payload.+-- | The maximum payload length of HTTP/2 payload. -- -- >>> maxPayloadLength--- 16384+-- 16777215 maxPayloadLength :: Int-maxPayloadLength = 2^(14::Int)+maxPayloadLength = 2^(24::Int) - 1++-- | The default payload length of HTTP/2 payload.+--+-- >>> defaultPayloadLength+-- 16384+defaultPayloadLength :: Int+defaultPayloadLength = 2^(14::Int) ---------------------------------------------------------------- -- Flags
Network/HTTP2/Server/Run.hs view
@@ -20,7 +20,7 @@ ok <- checkPreface when ok $ do serverInfo <- newServerInfo- ctx <- newContext serverInfo+ ctx <- newContext serverInfo confBufferSize -- Workers, worker manager and timer manager mgr <- start confTimeoutManager let wc = fromContext ctx@@ -31,8 +31,8 @@ -- If it is large, huge memory is consumed and many -- context switches happen. replicateM_ 3 $ spawnAction mgr- let runReceiver = frameReceiver ctx confReadN- runSender = frameSender ctx conf mgr+ let runReceiver = frameReceiver ctx conf+ runSender = frameSender ctx conf mgr race_ runReceiver runSender `E.finally` stop mgr where checkPreface = do
Network/HTTP2/Server/Types.hs view
@@ -29,7 +29,6 @@ promiseRequestPath :: ByteString -- | Accessor for response actually pushed from a server. , promiseResponse :: Response- -- | Accessor for response weight. } -- | Additional information.
Network/HTTP2/Server/Worker.hs view
@@ -41,12 +41,14 @@ , writeOutputQ = enqueueOutput outputQ , workerCleanup = \strm -> do closed ctx strm Killed- let frame = resetFrame InternalError (streamNumber strm)- enqueueControl controlQ $ CFrame frame- , isPushable = enablePush <$> readIORef http2settings+ let frame = resetFrame InternalError $ streamNumber strm+ enqueueControl controlQ $ CFrames Nothing [frame]+ -- Peer SETTINGS_ENABLE_PUSH+ , isPushable = enablePush <$> readIORef peerSettings , insertStream = insert streamTable+ -- Peer SETTINGS_INITIAL_WINDOW_SIZE , makePushStream = \pstrm _ -> do- ws <- initialWindowSize <$> readIORef http2settings+ ws <- initialWindowSize <$> readIORef peerSettings sid <- getMyNewStreamId ctx newstrm <- newPushStream sid ws let pid = streamNumber pstrm
http2.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: http2-version: 4.0.0+version: 4.1.0 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto <kazu@iij.ad.jp>@@ -51,10 +51,6 @@ description: Development commands default: False -flag doc- description: Doctest- default: False- library exposed-modules: Network.HPACK@@ -281,22 +277,6 @@ unordered-containers if flag(devel)-- else- buildable: False--test-suite doctest- type: exitcode-stdio-1.0- main-is: doctests.hs- hs-source-dirs: test- default-language: Haskell2010- default-extensions: Strict StrictData- ghc-options: -Wall- build-depends:- base >=4.9 && <5,- doctest >=0.9.3-- if flag(doc) else buildable: False
test/HTTP2/ClientSpec.hs view
@@ -26,7 +26,7 @@ spec :: Spec spec = do- describe "server" $ do+ describe "client" $ do it "receives an error if scheme is missing" $ E.bracket (forkIO runServer) killThread $ \_ -> do threadDelay 10000
test/HTTP2/ServerSpec.hs view
@@ -19,8 +19,10 @@ import Network.Socket import Network.Socket.ByteString import Test.Hspec+import System.IO import Network.HPACK+import Network.HPACK.Token import qualified Network.HTTP2.Client as C import Network.HTTP2.Server import Network.HTTP2.Frame@@ -53,7 +55,7 @@ runServer :: IO () runServer = runTCPServer (Just host) port runHTTP2Server where- runHTTP2Server s = E.bracket (allocSimpleConfig s 4096)+ runHTTP2Server s = E.bracket (allocSimpleConfig s 32768) freeSimpleConfig (`run` server) @@ -125,14 +127,15 @@ where h2rsp = responseStreaming ok200 header streamingBody header = [("Content-Type", "text/plain")]+ mhx = getHeaderValue (toToken "X-Tag") (snd (requestHeaders req)) streamingBody write _flush = do loop mt <- getRequestTrailers req- firstTrailerValue <$> mt `shouldBe` Just "b0870457df2b8cae06a88657a198d9b52f8e2b0a"+ firstTrailerValue <$> mt `shouldBe` mhx where loop = do bs <- getRequestBodyChunk req- unless (B.null bs) $ do+ when (bs /= "") $ do void $ write $ byteString bs loop maker = trailersMaker (CH.hashInit :: Context SHA1)@@ -156,7 +159,7 @@ freeSimpleConfig (\conf -> C.run cliconf conf client) client sendRequest = mapConcurrently_ ($ sendRequest) clients- clients = [client0,client1,client2,client3,client4,client5]+ clients = [client0,client1,client2,client3,client3',client3'',client4,client5] -- delay sending preface to be able to test if it is always sent first allocSlowPrefaceConfig :: Socket -> BufferSize -> IO Config@@ -191,15 +194,57 @@ client3 :: C.Client () client3 sendRequest = do- let req0 = C.requestFile methodPost "/echo" [] $ FileSpec "test/inputFile" 0 1012731+ let hx = "b0870457df2b8cae06a88657a198d9b52f8e2b0a"+ req0 = C.requestFile methodPost "/echo" [("X-Tag",hx)] $ FileSpec "test/inputFile" 0 1012731 req = C.setRequestTrailersMaker req0 maker sendRequest req $ \rsp -> do let comsumeBody = do bs <- C.getResponseBodyChunk rsp- unless (B.null bs) comsumeBody+ when (bs /= "") comsumeBody comsumeBody mt <- C.getResponseTrailers rsp- firstTrailerValue <$> mt `shouldBe` Just "b0870457df2b8cae06a88657a198d9b52f8e2b0a"+ firstTrailerValue <$> mt `shouldBe` Just hx+ where+ !maker = trailersMaker (CH.hashInit :: Context SHA1)++client3' :: C.Client ()+client3' sendRequest = do+ let hx = "b0870457df2b8cae06a88657a198d9b52f8e2b0a"+ req0 = C.requestStreaming methodPost "/echo" [("X-Tag",hx)] $ \write _flush -> do+ let sendFile h = do+ bs <- B.hGet h 1024+ when (bs /= "") $ do+ write $ byteString bs+ sendFile h+ withFile "test/inputFile" ReadMode sendFile+ req = C.setRequestTrailersMaker req0 maker+ sendRequest req $ \rsp -> do+ let comsumeBody = do+ bs <- C.getResponseBodyChunk rsp+ when (bs /= "") comsumeBody+ comsumeBody+ mt <- C.getResponseTrailers rsp+ firstTrailerValue <$> mt `shouldBe` Just hx+ where+ !maker = trailersMaker (CH.hashInit :: Context SHA1)++client3'' :: C.Client ()+client3'' sendRequest = do+ let hx = "59f82dfddc0adf5bdf7494b8704f203a67e25d4a"+ req0 = C.requestStreaming methodPost "/echo" [("X-Tag", hx)] $ \write _flush -> do+ let chunk = C8.replicate (16384 * 2) 'c'+ tag = C8.replicate 16 't'+ -- I don't think 9 is important here, this is just what I have, the client hangs on receiving the last one+ replicateM_ 9 $ write $ byteString chunk+ write $ byteString tag+ req = C.setRequestTrailersMaker req0 maker+ sendRequest req $ \rsp -> do+ let comsumeBody = do+ bs <- C.getResponseBodyChunk rsp+ when (bs /= "") comsumeBody+ comsumeBody+ mt <- C.getResponseTrailers rsp+ firstTrailerValue <$> mt `shouldBe` Just hx where !maker = trailersMaker (CH.hashInit :: Context SHA1)
− test/doctests.hs
@@ -1,14 +0,0 @@-module Main where--import Test.DocTest--main :: IO ()-main = do- doctest [- "-XOverloadedStrings"- , "Network/HPACK.hs"- ]- doctest [- "-XOverloadedStrings"- , "Network/HTTP2/Frame.hs"- ]