diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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 diff --git a/Network/HTTP2/Arch/Config.hs b/Network/HTTP2/Arch/Config.hs --- a/Network/HTTP2/Arch/Config.hs +++ b/Network/HTTP2/Arch/Config.hs @@ -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 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 @@ -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 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 @@ -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 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 @@ -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 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 @@ -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 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 @@ -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 - diff --git a/Network/HTTP2/Frame.hs b/Network/HTTP2/Frame.hs --- a/Network/HTTP2/Frame.hs +++ b/Network/HTTP2/Frame.hs @@ -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 diff --git a/Network/HTTP2/Frame/Decode.hs b/Network/HTTP2/Frame/Decode.hs --- a/Network/HTTP2/Frame/Decode.hs +++ b/Network/HTTP2/Frame/Decode.hs @@ -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 diff --git a/Network/HTTP2/Frame/Types.hs b/Network/HTTP2/Frame/Types.hs --- a/Network/HTTP2/Frame/Types.hs +++ b/Network/HTTP2/Frame/Types.hs @@ -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 diff --git a/Network/HTTP2/Server/Run.hs b/Network/HTTP2/Server/Run.hs --- a/Network/HTTP2/Server/Run.hs +++ b/Network/HTTP2/Server/Run.hs @@ -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 diff --git a/Network/HTTP2/Server/Types.hs b/Network/HTTP2/Server/Types.hs --- a/Network/HTTP2/Server/Types.hs +++ b/Network/HTTP2/Server/Types.hs @@ -29,7 +29,6 @@ promiseRequestPath :: ByteString -- | Accessor for response actually pushed from a server. , promiseResponse :: Response - -- | Accessor for response weight. } -- | Additional information. 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 @@ -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 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.0.0 +version: 4.1.0 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto @@ -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 diff --git a/test/HTTP2/ClientSpec.hs b/test/HTTP2/ClientSpec.hs --- a/test/HTTP2/ClientSpec.hs +++ b/test/HTTP2/ClientSpec.hs @@ -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 diff --git a/test/HTTP2/ServerSpec.hs b/test/HTTP2/ServerSpec.hs --- a/test/HTTP2/ServerSpec.hs +++ b/test/HTTP2/ServerSpec.hs @@ -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) diff --git a/test/doctests.hs b/test/doctests.hs deleted file mode 100644 --- a/test/doctests.hs +++ /dev/null @@ -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" - ]