diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,7 @@ +## 3.1.8 + +* Using the new priority queue based on PSQ provided by http2 lib. + ## 3.1.7 * A concatenated Cookie header is prepended to the headers to ensure that it flows pseudo headers. [#454](https://github.com/yesodweb/wai/pull/454) diff --git a/Network/Wai/Handler/Warp/HTTP2/HPACK.hs b/Network/Wai/Handler/Warp/HTTP2/HPACK.hs --- a/Network/Wai/Handler/Warp/HTTP2/HPACK.hs +++ b/Network/Wai/Handler/Warp/HTTP2/HPACK.hs @@ -22,7 +22,8 @@ -- $setup -- >>> :set -XOverloadedStrings --- fixme: should we split "Set-Cookie:" headers? +-- Set-Cookie: contains only one cookie value. +-- So, we don't need to split it. hpackEncodeHeader :: Context -> InternalInfo -> S.Settings -> H.Status -> H.ResponseHeaders -> IO Builder hpackEncodeHeader ctx ii settings s h = do 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 (OGoaway frame) highestPriority + enqueue outputQ 0 controlPriority $ OGoaway frame | otherwise = return () sendReset err sid = do let frame = resetFrame err sid - enqueue outputQ (OFrame frame) highestPriority + enqueue outputQ 0 controlPriority $ OFrame frame loop = do hd <- recvN frameHeaderLength if BS.null hd then - enqueue outputQ OFinish highestPriority + enqueue outputQ 0 controlPriority OFinish else do cont <- processStreamGuardingError $ decodeFrameHeader hd when cont loop @@ -186,7 +186,7 @@ unless (testAck flags) $ do modifyIORef http2settings $ \old -> updateSettings old alist let frame = settingsFrame setAck [] - enqueue outputQ (OSettings frame alist) highestPriority + enqueue outputQ 0 controlPriority $ 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 (OFrame frame) defaultPriority + enqueue outputQ 0 controlPriority $ OFrame frame return True control FrameGoAway _ _ Context{outputQ} = do - enqueue outputQ OFinish highestPriority + enqueue outputQ 0 controlPriority 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 (OFrame frame) highestPriority + enqueue outputQ 0 controlPriority $ OFrame frame atomically $ writeTQueue q body if endOfStream then do mcl <- readIORef streamContentLength @@ -310,24 +310,25 @@ atomically $ writeTVar streamWindow w return s -stream FrameRSTStream header bs _ _ strm = do +stream FrameRSTStream header bs ctx _ strm = do RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs let cc = Reset e - closed strm cc + closed strm cc ctx return $ Closed cc -- will be written to streamState again stream FramePriority header bs Context{outputQ} s Stream{streamNumber,streamPriority} = do - PriorityFrame p <- guardIt $ decodePriorityFrame header bs - checkPriority p streamNumber - -- checkme: this should be tested - -- fixme: This works well when the priority gets lower because - -- the old higher priority value comes out from the queue quickly - -- and the new lower priority is used when enqueuing again. - -- But when the priority get higher, it takes time to use the new - -- priority. - writeIORef streamPriority p + PriorityFrame newp <- guardIt $ decodePriorityFrame header bs + checkPriority newp streamNumber -- checkme: this should be tested - when (isIdle s) $ prepare outputQ streamNumber p + oldp <- readIORef streamPriority + writeIORef streamPriority newp + if isIdle s then + prepare outputQ streamNumber newp + else do + mx <- delete outputQ streamNumber oldp + case mx of + Nothing -> return () + Just x -> enqueue outputQ streamNumber newp 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,15 +60,16 @@ -- | Run the given action if the stream is not closed; handle any exceptions by -- resetting the stream. -unlessClosed :: Connection -> Stream -> IO () -> IO Bool -unlessClosed Connection{connSendAll} +unlessClosed :: Context -> Connection -> Stream -> IO () -> IO Bool +unlessClosed ctx + 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) + closed strm (ResetByMe e) ctx let rst = resetFrame InternalError streamNumber connSendAll rst return False @@ -99,7 +100,7 @@ loop -- ignoring the old priority because the value might be changed. - loop = dequeue outputQ >>= \(out, _) -> switch out + loop = dequeue outputQ >>= \(_sid,out) -> switch out ignore :: E.SomeException -> IO () ignore _ = return () @@ -118,12 +119,12 @@ connSendAll frame loop switch (OResponse strm s h aux) = do - _ <- unlessClosed conn strm $ + _ <- unlessClosed ctx conn strm $ getWindowSize connectionWindow (streamWindow strm) >>= sendResponse strm s h aux loop switch (ONext strm curr) = do - _ <- unlessClosed conn strm $ do + _ <- unlessClosed ctx conn strm $ do lim <- getWindowSize connectionWindow (streamWindow strm) -- Data frame payload Next datPayloadLen mnext <- curr lim @@ -131,7 +132,7 @@ dispatchNext strm mnext loop switch (OPush oldStrm push mvar strm s h aux) = do - pushed <- unlessClosed conn oldStrm $ do + pushed <- unlessClosed ctx conn oldStrm $ do lim <- getWindowSize connectionWindow (streamWindow strm) -- Write and send the promise. builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push @@ -183,7 +184,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 + closed strm Finished ctx 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 @@ -248,10 +248,14 @@ atomicModifyIORef' concurrencyRef (\x -> (x+1,())) writeIORef streamState (Open JustOpened) -closed :: Stream -> ClosedCode -> IO () -closed Stream{concurrencyRef,streamState} cc = do +closed :: Stream -> ClosedCode -> Context -> IO () +closed Stream{concurrencyRef,streamState,streamNumber,streamPriority} + cc + Context{outputQ} = do atomicModifyIORef' concurrencyRef (\x -> (x-1,())) writeIORef streamState (Closed cc) + pri <- readIORef streamPriority + clear outputQ streamNumber pri ---------------------------------------------------------------- @@ -297,7 +301,7 @@ x <- readTVar $ streamWindow strm check (x > 0) pri <- readIORef $ streamPriority strm - enqueue outQ out pri + enqueue outQ (streamNumber strm) pri out enqueueOrSpawnTemporaryWaiter :: Stream -> PriorityTree Output -> Output -> IO () enqueueOrSpawnTemporaryWaiter strm outQ out = do @@ -307,4 +311,4 @@ void $ forkIO $ enqueueWhenWindowIsOpen outQ out else do pri <- readIORef $ streamPriority strm - enqueue outQ out pri + enqueue outQ (streamNumber strm) pri 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 @@ -96,10 +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 Context{outputQ} set strm req me = do - closed strm Killed - let frame = resetFrame InternalError (streamNumber strm) - enqueue outputQ (OFrame frame) highestPriority +cleanupStream ctx@Context{outputQ} set strm req me = do + closed strm Killed ctx + let sid = streamNumber strm + frame = resetFrame InternalError sid + enqueue outputQ sid controlPriority $ OFrame frame case me of Nothing -> return () Just e -> S.settingsOnException set req e diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,4 +1,3 @@ -## warp +# Warp -The premier WAI handler. For more information, see [Warp: A Haskell Web -Server](http://steve.vinoski.net/blog/2011/05/01/warp-a-haskell-web-server/). +Warp is a server library for HTTP/1.x and HTTP/2 based WAI(Web Application Interface in Haskell). For more information, see [Warp](http://www.aosabook.org/en/posa/warp.html). diff --git a/warp.cabal b/warp.cabal --- a/warp.cabal +++ b/warp.cabal @@ -1,5 +1,5 @@ Name: warp -Version: 3.1.7 +Version: 3.1.8 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.1 + , http2 >= 1.2 , 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.1 + , http2 >= 1.2 , word8 , hashable