warp 3.1.8 → 3.1.9
raw patch · 6 files changed
+71/−63 lines, 6 filesdep ~http2
Dependency ranges changed: http2
Files
- ChangeLog.md +4/−0
- Network/Wai/Handler/Warp/HTTP2/Receiver.hs +25/−21
- Network/Wai/Handler/Warp/HTTP2/Sender.hs +18/−16
- Network/Wai/Handler/Warp/HTTP2/Types.hs +15/−17
- Network/Wai/Handler/Warp/HTTP2/Worker.hs +6/−6
- warp.cabal +3/−3
ChangeLog.md view
@@ -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.
Network/Wai/Handler/Warp/HTTP2/Receiver.hs view
@@ -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
Network/Wai/Handler/Warp/HTTP2/Sender.hs view
@@ -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
Network/Wai/Handler/Warp/HTTP2/Types.hs view
@@ -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
Network/Wai/Handler/Warp/HTTP2/Worker.hs view
@@ -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
warp.cabal view
@@ -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