diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,12 @@ +## 5.2.6 + +* Recover rxflow on closing. + [#126](https://github.com/kazu-yamamoto/http2/pull/126) +* Fixing ClientSpec for stream errors. +* Allowing negative window. (h2spec http2/6.9.2) +* Update for latest http-semantics + [#122](https://github.com/kazu-yamamoto/http2/pull/124) + ## 5.2.5 * Setting peer initial window size properly. 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 @@ -85,7 +85,9 @@ clientCore ctx mgr req processResponse = do strm <- sendRequest ctx mgr scheme authority req rsp <- getResponse strm - processResponse rsp + x <- processResponse rsp + adjustRxWindow ctx strm + return x runClient ctx mgr = do x <- client (clientCore ctx mgr) $ aux ctx waitCounter0 mgr @@ -205,14 +207,15 @@ forkManagedUnmask mgr $ \unmask -> do decrementedCounter <- newIORef False let decCounterOnce = do - alreadyDecremented <- atomicModifyIORef decrementedCounter $ \b -> (True, b) - unless alreadyDecremented $ decCounter mgr - let iface = OutBodyIface { - outBodyUnmask = unmask - , outBodyPush = \b -> atomically $ writeTBQueue tbq (StreamingBuilder b Nothing) - , outBodyPushFinal = \b -> atomically $ writeTBQueue tbq (StreamingBuilder b (Just decCounterOnce)) - , outBodyFlush = atomically $ writeTBQueue tbq StreamingFlush - } + alreadyDecremented <- atomicModifyIORef decrementedCounter $ \b -> (True, b) + unless alreadyDecremented $ decCounter mgr + let iface = + OutBodyIface + { outBodyUnmask = unmask + , outBodyPush = \b -> atomically $ writeTBQueue tbq (StreamingBuilder b Nothing) + , outBodyPushFinal = \b -> atomically $ writeTBQueue tbq (StreamingBuilder b (Just decCounterOnce)) + , outBodyFlush = atomically $ writeTBQueue tbq StreamingFlush + } finished = atomically $ writeTBQueue tbq $ StreamingFinished decCounterOnce incCounter mgr strmbdy iface `finally` finished 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 @@ -190,7 +190,7 @@ -- >>> isWindowOverflow (maxWindowSize + 1) -- True isWindowOverflow :: WindowSize -> Bool -isWindowOverflow w = testBit w 31 +isWindowOverflow w = w > maxWindowSize -- | Default concurrency. -- diff --git a/Network/HTTP2/H2/Receiver.hs b/Network/HTTP2/H2/Receiver.hs --- a/Network/HTTP2/H2/Receiver.hs +++ b/Network/HTTP2/H2/Receiver.hs @@ -190,11 +190,12 @@ return False -- Transition (process2) -processState (Open hcl (HasBody tbl@(_, reqvt))) ctx@Context{..} strm@Stream{streamInput} _streamId = do +processState (Open hcl (HasBody tbl@(_, reqvt))) ctx@Context{..} strm@Stream{streamInput, streamRxQ} _streamId = do let mcl = fst <$> (getFieldValue tokenContentLength reqvt >>= C8.readInt) bodyLength <- newIORef 0 tlr <- newIORef Nothing q <- newTQueueIO + writeIORef streamRxQ $ Just q setStreamState ctx strm $ Open hcl (Body q mcl bodyLength tlr) -- FLOW CONTROL: WINDOW_UPDATE 0: recv: announcing my limit properly -- FLOW CONTROL: WINDOW_UPDATE: recv: announcing my limit properly @@ -612,17 +613,9 @@ ---------------------------------------------------------------- -- | Type for input streaming. -data Source - = Source - (Int -> IO ()) - (TQueue (Either E.SomeException (ByteString, Bool))) - (IORef ByteString) - (IORef Bool) +data Source = Source (Int -> IO ()) RxQ (IORef ByteString) (IORef Bool) -mkSource - :: TQueue (Either E.SomeException (ByteString, Bool)) - -> (Int -> IO ()) - -> IO Source +mkSource :: RxQ -> (Int -> IO ()) -> IO Source mkSource q inform = Source inform q <$> newIORef "" <*> newIORef False readSource :: Source -> IO (ByteString, Bool) diff --git a/Network/HTTP2/H2/Stream.hs b/Network/HTTP2/H2/Stream.hs --- a/Network/HTTP2/H2/Stream.hs +++ b/Network/HTTP2/H2/Stream.hs @@ -52,6 +52,7 @@ <*> newEmptyMVar <*> newTVarIO (newTxFlow txwin) <*> newIORef (newRxFlow rxwin) + <*> newIORef Nothing newEvenStream :: StreamId -> WindowSize -> WindowSize -> IO Stream newEvenStream sid txwin rxwin = @@ -60,6 +61,7 @@ <*> newEmptyMVar <*> newTVarIO (newTxFlow txwin) <*> newIORef (newRxFlow rxwin) + <*> newIORef Nothing ---------------------------------------------------------------- diff --git a/Network/HTTP2/H2/Types.hs b/Network/HTTP2/H2/Types.hs --- a/Network/HTTP2/H2/Types.hs +++ b/Network/HTTP2/H2/Types.hs @@ -149,12 +149,15 @@ ---------------------------------------------------------------- +type RxQ = TQueue (Either E.SomeException (ByteString, Bool)) + data Stream = Stream { streamNumber :: StreamId , streamState :: IORef StreamState , streamInput :: MVar (Either SomeException InpObj) -- Client only , streamTxFlow :: TVar TxFlow , streamRxFlow :: IORef RxFlow + , streamRxQ :: IORef (Maybe RxQ) } instance Show Stream where diff --git a/Network/HTTP2/H2/Window.hs b/Network/HTTP2/H2/Window.hs --- a/Network/HTTP2/H2/Window.hs +++ b/Network/HTTP2/H2/Window.hs @@ -1,8 +1,9 @@ {-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE BangPatterns #-} module Network.HTTP2.H2.Window where +import qualified Data.ByteString as BS import Data.IORef import Network.Control import qualified UnliftIO.Exception as E @@ -79,3 +80,24 @@ let frame = windowUpdateFrame streamNumber ws cframe = CFrames Nothing [frame] enqueueControl controlQ cframe + +adjustRxWindow :: Context -> Stream -> IO () +adjustRxWindow ctx stream@Stream{streamRxQ} = do + mq <- readIORef streamRxQ + case mq of + Nothing -> return () + Just q -> do + len <- readQ q + informWindowUpdate ctx stream len + where + readQ q = atomically $ loop 0 + where + loop !total = do + meb <- tryReadTQueue q + case meb of + Just (Right (bs, _)) -> loop (total + BS.length bs) + Just le@(Left _) -> do + -- reserving HTTP2Error + writeTQueue q le + return total + _ -> return total 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 @@ -49,7 +49,7 @@ when ok $ do (ctx, mgr) <- setup sconf conf let wc = fromContext ctx - setAction mgr $ worker wc mgr server + setAction mgr $ worker ctx wc mgr server replicateM_ numberOfWorkers $ spawnAction mgr runH2 conf ctx mgr 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 @@ -1,4 +1,3 @@ -{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternGuards #-} {-# LANGUAGE RecordWildCards #-} @@ -15,7 +14,6 @@ import Network.HTTP.Semantics.Server import Network.HTTP.Semantics.Server.Internal import Network.HTTP.Types -import Network.Socket (SockAddr) import qualified System.TimeManager as T import UnliftIO.Exception (SomeException (..)) import qualified UnliftIO.Exception as E @@ -33,8 +31,6 @@ , workerCleanup :: a -> IO () , isPushable :: IO Bool , makePushStream :: a -> PushPromise -> IO (StreamId, a) - , mySockAddr :: SockAddr - , peerSockAddr :: SockAddr } fromContext :: Context -> WorkerConf Stream @@ -54,8 +50,6 @@ (_, newstrm) <- openEvenStreamWait ctx let pid = streamNumber pstrm return (pid, newstrm) - , mySockAddr = mySockAddr - , peerSockAddr = peerSockAddr } ---------------------------------------------------------------- @@ -162,8 +156,8 @@ (_, reqvt) = inpObjHeaders req -- | Worker for server applications. -worker :: WorkerConf a -> Manager -> Server -> Action -worker wc@WorkerConf{..} mgr server = do +worker :: Context -> WorkerConf Stream -> Manager -> Server -> Action +worker ctx@Context{..} wc@WorkerConf{..} mgr server = do sinfo <- newStreamInfo tcont <- newThreadContinue timeoutKillThread mgr $ go sinfo tcont @@ -178,7 +172,9 @@ T.resume th T.tickle th let aux = Aux th mySockAddr peerSockAddr - server (Request req') aux $ response wc mgr th tcont strm (Request req') + r <- server (Request req') aux $ response wc mgr th tcont strm (Request req') + adjustRxWindow ctx strm + return r cont1 <- case ex of Right () -> return True Left e@(SomeException _) 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: 5.2.5 +version: 5.2.6 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto @@ -115,7 +115,7 @@ bytestring >=0.10, case-insensitive >=1.2 && <1.3, containers >=0.6, - http-semantics >= 0.1.1 && <0.2, + http-semantics >= 0.1.2 && <0.2, http-types >=0.12 && <0.13, network >=3.1, network-byte-order >=0.1.7 && <0.2, diff --git a/test/HTTP2/ClientSpec.hs b/test/HTTP2/ClientSpec.hs --- a/test/HTTP2/ClientSpec.hs +++ b/test/HTTP2/ClientSpec.hs @@ -3,7 +3,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} -module HTTP2.ClientSpec where +module HTTP2.ClientSpec (spec) where import Control.Concurrent import Control.Monad @@ -37,18 +37,18 @@ it "receives an error if scheme is missing" $ E.bracket (forkIO $ runServer defaultServer) killThread $ \_ -> do threadDelay 10000 - runClient "" host (defaultClient []) `shouldThrow` connectionError + runClient "" host (defaultClient []) `shouldThrow` streamError it "receives an error if authority is missing" $ E.bracket (forkIO $ runServer defaultServer) killThread $ \_ -> do threadDelay 10000 - runClient "http" "" (defaultClient []) `shouldThrow` connectionError + runClient "http" "" (defaultClient []) `shouldThrow` streamError it "receives an error if authority and host are different" $ E.bracket (forkIO $ runServer defaultServer) killThread $ \_ -> do threadDelay 10000 runClient "http" host (defaultClient [("Host", "foo")]) - `shouldThrow` connectionError + `shouldThrow` streamError it "does not deadlock (in concurrent setting)" $ E.bracket (forkIO $ runServer irresponsiveServer) killThread $ \_ -> do @@ -137,6 +137,7 @@ putMVar resultVar result threadDelay 10000 -connectionError :: Selector HTTP2Error -connectionError ConnectionErrorIsReceived{} = True -connectionError _ = False +streamError :: Selector HTTP2Error +streamError StreamErrorIsReceived{} = True +streamError ConnectionErrorIsReceived{} = True +streamError _ = False diff --git a/test/HTTP2/ServerSpec.hs b/test/HTTP2/ServerSpec.hs --- a/test/HTTP2/ServerSpec.hs +++ b/test/HTTP2/ServerSpec.hs @@ -3,7 +3,7 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} -module HTTP2.ServerSpec where +module HTTP2.ServerSpec (spec) where import Control.Concurrent import Control.Monad @@ -237,10 +237,10 @@ FileSpec "test/inputFile" 0 1012731 req = C.setRequestTrailersMaker req0 maker sendRequest req $ \rsp -> do - let comsumeBody = do + let consumeBody = do bs <- C.getResponseBodyChunk rsp - when (bs /= "") comsumeBody - comsumeBody + when (bs /= "") consumeBody + consumeBody mt <- C.getResponseTrailers rsp firstTrailerValue <$> mt `shouldBe` Just hx where @@ -258,10 +258,10 @@ withFile "test/inputFile" ReadMode sendFile req = C.setRequestTrailersMaker req0 maker sendRequest req $ \rsp -> do - let comsumeBody = do + let consumeBody = do bs <- C.getResponseBodyChunk rsp - when (bs /= "") comsumeBody - comsumeBody + when (bs /= "") consumeBody + consumeBody mt <- C.getResponseTrailers rsp firstTrailerValue <$> mt `shouldBe` Just hx where @@ -278,10 +278,10 @@ write $ byteString tag req = C.setRequestTrailersMaker req0 maker sendRequest req $ \rsp -> do - let comsumeBody = do + let consumeBody = do bs <- C.getResponseBodyChunk rsp - when (bs /= "") comsumeBody - comsumeBody + when (bs /= "") consumeBody + consumeBody mt <- C.getResponseTrailers rsp firstTrailerValue <$> mt `shouldBe` Just hx where