diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,7 @@ +## 3.1.9 + +* Using the new priority queue based on PSQ provided by http2 lib again. + ## 3.1.8 * Using the new priority queue based on PSQ provided by http2 lib. diff --git a/Network/Wai/Handler/Warp/HTTP2/Receiver.hs b/Network/Wai/Handler/Warp/HTTP2/Receiver.hs --- a/Network/Wai/Handler/Warp/HTTP2/Receiver.hs +++ b/Network/Wai/Handler/Warp/HTTP2/Receiver.hs @@ -41,17 +41,17 @@ | Just (ConnectionError err msg) <- E.fromException e = do csid <- readIORef currentStreamId let frame = goawayFrame csid err msg - enqueue outputQ 0 controlPriority $ OGoaway frame + enqueueControl outputQ 0 $ OGoaway frame | otherwise = return () sendReset err sid = do let frame = resetFrame err sid - enqueue outputQ 0 controlPriority $ OFrame frame + enqueueControl outputQ 0 $ OFrame frame loop = do hd <- recvN frameHeaderLength if BS.null hd then - enqueue outputQ 0 controlPriority OFinish + enqueueControl outputQ 0 OFinish else do cont <- processStreamGuardingError $ decodeFrameHeader hd when cont loop @@ -92,7 +92,7 @@ control ftyp header pl ctx | otherwise = do checkContinued - strm@Stream{streamState,streamContentLength,streamPriority} <- getStream + strm@Stream{streamState,streamContentLength,streamPrecedence} <- getStream pl <- recvN payloadLength state <- readIORef streamState state' <- stream ftyp header pl ctx state strm @@ -103,7 +103,7 @@ Just vh -> do when (isJust (vhCL vh) && vhCL vh /= Just 0) $ E.throwIO $ StreamError ProtocolError streamId - writeIORef streamPriority pri + writeIORef streamPrecedence $ toPrecedence pri writeIORef streamState HalfClosed let req = mkreq vh (return "") atomically $ writeTQueue inputQ $ Input strm req @@ -113,7 +113,7 @@ case validateHeaders hdr of Just vh -> do q <- newTQueueIO - writeIORef streamPriority pri + writeIORef streamPrecedence $ toPrecedence pri writeIORef streamState (Open (Body q)) writeIORef streamContentLength $ vhCL vh readQ <- newReadBody q @@ -186,7 +186,7 @@ unless (testAck flags) $ do modifyIORef http2settings $ \old -> updateSettings old alist let frame = settingsFrame setAck [] - enqueue outputQ 0 controlPriority $ OSettings frame alist + enqueueControl outputQ 0 $ OSettings frame alist return True control FramePing FrameHeader{flags} bs Context{outputQ} = @@ -194,11 +194,11 @@ E.throwIO $ ConnectionError ProtocolError "the ack flag of this ping frame must not be set" else do let frame = pingFrame bs - enqueue outputQ 0 controlPriority $ OFrame frame + enqueueControl outputQ 0 $ OFrame frame return True control FrameGoAway _ _ Context{outputQ} = do - enqueue outputQ 0 controlPriority OFinish + enqueueControl outputQ 0 OFinish return False control FrameWindowUpdate header bs Context{connectionWindow} = do @@ -271,7 +271,7 @@ let frame1 = windowUpdateFrame 0 payloadLength frame2 = windowUpdateFrame streamNumber payloadLength frame = frame1 `BS.append` frame2 - enqueue outputQ 0 controlPriority $ OFrame frame + enqueueControl outputQ 0 $ OFrame frame atomically $ writeTQueue q body if endOfStream then do mcl <- readIORef streamContentLength @@ -310,25 +310,29 @@ atomically $ writeTVar streamWindow w return s -stream FrameRSTStream header bs ctx _ strm = do +stream FrameRSTStream header bs _ _ strm = do RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs let cc = Reset e - closed strm cc ctx + closed strm cc return $ Closed cc -- will be written to streamState again -stream FramePriority header bs Context{outputQ} s Stream{streamNumber,streamPriority} = do - PriorityFrame newp <- guardIt $ decodePriorityFrame header bs - checkPriority newp streamNumber +stream FramePriority header bs Context{outputQ,priorityTreeSize} s Stream{streamNumber,streamPrecedence} = do + PriorityFrame newpri <- guardIt $ decodePriorityFrame header bs + checkPriority newpri streamNumber -- checkme: this should be tested - oldp <- readIORef streamPriority - writeIORef streamPriority newp - if isIdle s then - prepare outputQ streamNumber newp + oldpre <- readIORef streamPrecedence + let !newpre = toPrecedence newpri + writeIORef streamPrecedence newpre + if isIdle s then do + n <- atomicModifyIORef' priorityTreeSize (\x -> (x+1,x+1)) + -- fixme hard coding + when (n >= 20) $ E.throwIO $ ConnectionError EnhanceYourCalm "too many idle priority frames" + prepare outputQ streamNumber newpri else do - mx <- delete outputQ streamNumber oldp + mx <- delete outputQ streamNumber oldpre case mx of Nothing -> return () - Just x -> enqueue outputQ streamNumber newp x + Just x -> enqueue outputQ streamNumber newpre x return s -- this ordering is important diff --git a/Network/Wai/Handler/Warp/HTTP2/Sender.hs b/Network/Wai/Handler/Warp/HTTP2/Sender.hs --- a/Network/Wai/Handler/Warp/HTTP2/Sender.hs +++ b/Network/Wai/Handler/Warp/HTTP2/Sender.hs @@ -60,16 +60,15 @@ -- | Run the given action if the stream is not closed; handle any exceptions by -- resetting the stream. -unlessClosed :: Context -> Connection -> Stream -> IO () -> IO Bool -unlessClosed ctx - Connection{connSendAll} +unlessClosed :: Connection -> Stream -> IO () -> IO Bool +unlessClosed Connection{connSendAll} strm@Stream{streamState,streamNumber} body = E.handle resetStream $ do state <- readIORef streamState if (isClosed state) then return False else body >> return True where resetStream e = do - closed strm (ResetByMe e) ctx + closed strm (ResetByMe e) let rst = resetFrame InternalError streamNumber connSendAll rst return False @@ -100,14 +99,14 @@ loop -- ignoring the old priority because the value might be changed. - loop = dequeue outputQ >>= \(_sid,out) -> switch out + loop = dequeue outputQ >>= \(_sid,pre,out) -> switch out pre ignore :: E.SomeException -> IO () ignore _ = return () - switch OFinish = return () - switch (OGoaway frame) = connSendAll frame - switch (OSettings frame alist) = do + switch OFinish _ = return () + switch (OGoaway frame) _ = connSendAll frame + switch (OSettings frame alist) _ = do connSendAll frame case lookup SettingsHeaderTableSize alist of Nothing -> return () @@ -115,24 +114,27 @@ dyntbl <- readIORef encodeDynamicTable setLimitForEncoding siz dyntbl loop - switch (OFrame frame) = do + switch (OFrame frame) _ = do connSendAll frame loop - switch (OResponse strm s h aux) = do - _ <- unlessClosed ctx conn strm $ + switch (OResponse strm s h aux) pre = do + writeIORef (streamPrecedence strm) pre -- fixme + _ <- unlessClosed conn strm $ getWindowSize connectionWindow (streamWindow strm) >>= sendResponse strm s h aux loop - switch (ONext strm curr) = do - _ <- unlessClosed ctx conn strm $ do + switch (ONext strm curr) pre = do + writeIORef (streamPrecedence strm) pre + _ <- unlessClosed conn strm $ do lim <- getWindowSize connectionWindow (streamWindow strm) -- Data frame payload Next datPayloadLen mnext <- curr lim fillDataHeaderSend strm 0 datPayloadLen dispatchNext strm mnext loop - switch (OPush oldStrm push mvar strm s h aux) = do - pushed <- unlessClosed ctx conn oldStrm $ do + switch (OPush oldStrm push mvar strm s h aux) pre = do + writeIORef (streamPrecedence strm) pre -- fixme + pushed <- unlessClosed conn oldStrm $ do lim <- getWindowSize connectionWindow (streamWindow strm) -- Write and send the promise. builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push @@ -184,7 +186,7 @@ -- 'closed' must be before 'flushN'. If not, the context would be -- switched to the receiver, resulting in the inconsistency of -- concurrency. - closed strm Finished ctx + closed strm Finished flushN toFlush -- Flush the connection buffer to the socket, where the first 'n' bytes of diff --git a/Network/Wai/Handler/Warp/HTTP2/Types.hs b/Network/Wai/Handler/Warp/HTTP2/Types.hs --- a/Network/Wai/Handler/Warp/HTTP2/Types.hs +++ b/Network/Wai/Handler/Warp/HTTP2/Types.hs @@ -1,5 +1,5 @@ {-# LANGUAGE OverloadedStrings, CPP #-} -{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE NamedFieldPuns, RecordWildCards #-} module Network.Wai.Handler.Warp.HTTP2.Types where @@ -130,6 +130,7 @@ -- | Number of active streams initiated by the server; for respecting the -- client's max concurrency setting. , pushConcurrency :: IORef Int + , priorityTreeSize :: IORef Int -- | RFC 7540 says "Other frames (from any stream) MUST NOT -- occur between the HEADERS frame and any CONTINUATION -- frames that might follow". This field is used to implement @@ -153,6 +154,7 @@ <*> initialize 10 -- fixme: hard coding: 10 <*> newIORef 0 <*> newIORef 0 + <*> newIORef 0 <*> newIORef Nothing <*> newIORef 0 <*> newIORef 2 -- first server push stream; 0 is reserved @@ -221,7 +223,7 @@ , streamContentLength :: IORef (Maybe Int) , streamBodyLength :: IORef Int , streamWindow :: TVar WindowSize - , streamPriority :: IORef Priority + , streamPrecedence :: IORef Precedence -- | The concurrency IORef in which this stream has been counted. The client -- and server each have separate concurrency values to respect, so pushed -- streams need to decrement a different count when they're closed. This @@ -238,7 +240,7 @@ <*> newIORef Nothing <*> newIORef 0 <*> newTVarIO win - <*> newIORef defaultPriority + <*> newIORef defaultPrecedence <*> pure ref ---------------------------------------------------------------- @@ -248,14 +250,10 @@ atomicModifyIORef' concurrencyRef (\x -> (x+1,())) writeIORef streamState (Open JustOpened) -closed :: Stream -> ClosedCode -> Context -> IO () -closed Stream{concurrencyRef,streamState,streamNumber,streamPriority} - cc - Context{outputQ} = do +closed :: Stream -> ClosedCode -> IO () +closed Stream{concurrencyRef,streamState} cc = do atomicModifyIORef' concurrencyRef (\x -> (x-1,())) writeIORef streamState (Closed cc) - pri <- readIORef streamPriority - clear outputQ streamNumber pri ---------------------------------------------------------------- @@ -296,19 +294,19 @@ -- INVARIANT: streams in the output queue have non-zero window size. enqueueWhenWindowIsOpen :: PriorityTree Output -> Output -> IO () enqueueWhenWindowIsOpen outQ out = do - let strm = outputStream out + let Stream{..} = outputStream out atomically $ do - x <- readTVar $ streamWindow strm + x <- readTVar streamWindow check (x > 0) - pri <- readIORef $ streamPriority strm - enqueue outQ (streamNumber strm) pri out + pre <- readIORef streamPrecedence + enqueue outQ streamNumber pre out enqueueOrSpawnTemporaryWaiter :: Stream -> PriorityTree Output -> Output -> IO () -enqueueOrSpawnTemporaryWaiter strm outQ out = do - sw <- atomically $ readTVar $ streamWindow strm +enqueueOrSpawnTemporaryWaiter Stream{..} outQ out = do + sw <- atomically $ readTVar streamWindow if sw == 0 then -- This waiter waits only for the stream window. void $ forkIO $ enqueueWhenWindowIsOpen outQ out else do - pri <- readIORef $ streamPriority strm - enqueue outQ (streamNumber strm) pri out + pre <- readIORef streamPrecedence + enqueue outQ streamNumber pre out diff --git a/Network/Wai/Handler/Warp/HTTP2/Worker.hs b/Network/Wai/Handler/Warp/HTTP2/Worker.hs --- a/Network/Wai/Handler/Warp/HTTP2/Worker.hs +++ b/Network/Wai/Handler/Warp/HTTP2/Worker.hs @@ -83,7 +83,7 @@ -- queue, we spawn a thread to ensure that the designated -- queue is not empty. void $ forkIO $ waiter tvar sq strm outputQ - atomically $ writeTVar tvar (SyncNext out) + atomically $ writeTVar tvar $ SyncNext out let write chunk = do atomically $ writeTBQueue sq $ case chunk of BuilderChunk b -> SBuilder b @@ -96,11 +96,11 @@ -- | Handle abnormal termination of a stream: mark it as closed, send a reset -- frame, and call the user's 'settingsOnException' handler if applicable. cleanupStream :: Context -> S.Settings -> Stream -> Maybe Request -> Maybe SomeException -> IO () -cleanupStream ctx@Context{outputQ} set strm req me = do - closed strm Killed ctx +cleanupStream Context{outputQ} set strm req me = do + closed strm Killed let sid = streamNumber strm frame = resetFrame InternalError sid - enqueue outputQ sid controlPriority $ OFrame frame + enqueueControl outputQ sid $ OFrame frame case me of Nothing -> return () Just e -> S.settingsOnException set req e @@ -144,8 +144,8 @@ -- Section 5.3.5 of RFC 7540 defines the weight of push promise is 16. -- But we need not to follow the spec. So, this value would change -- if necessary. - writeIORef (streamPriority newStrm) $ - defaultPriority { streamDependency = streamNumber strm } + writeIORef (streamPrecedence newStrm) $ + toPrecedence $ defaultPriority { streamDependency = streamNumber strm } opened newStrm insert streamTable newSid newStrm diff --git a/warp.cabal b/warp.cabal --- a/warp.cabal +++ b/warp.cabal @@ -1,5 +1,5 @@ Name: warp -Version: 3.1.8 +Version: 3.1.9 Synopsis: A fast, light-weight web server for WAI applications. License: MIT License-file: LICENSE @@ -42,7 +42,7 @@ , ghc-prim , http-types >= 0.8.5 , iproute >= 1.3.1 - , http2 >= 1.2 + , http2 >= 1.3 , simple-sendfile >= 0.2.7 && < 0.3 , unix-compat >= 0.2 , wai >= 3.0.4 && < 3.1 @@ -160,7 +160,7 @@ , directory , process , containers - , http2 >= 1.2 + , http2 >= 1.3 , word8 , hashable