http2 5.4.12 → 5.4.13

raw patch · 13 files changed

+646/−77 lines, 13 filesdep ~networkdep ~stmPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: network, stm

API changes (from Hackage documentation)

Files

• ChangeLog.md +27/−0
• Network/HTTP2/Client.hs +2/−0
• Network/HTTP2/H2/Config.hs +9/−2
• Network/HTTP2/H2/Context.hs +90/−7
• Network/HTTP2/H2/Receiver.hs +51/−37
• Network/HTTP2/H2/Sender.hs +25/−4
• Network/HTTP2/H2/Types.hs +49/−0
• Network/HTTP2/Server.hs +4/−0
• Network/HTTP2/Server/Internal.hs +1/−0
• Network/HTTP2/Server/Run.hs +96/−21
• http2.cabal +6/−4
• test/HTTP2/GiveUpSpec.hs +149/−0
• test/HTTP2/ServerSpec.hs +137/−2

@@ -1,5 +1,32 @@ # ChangeLog for http2 +## 5.4.13++* `Config` gains `confSendAllWith` and `confReadNWith`, a send and a read+  which take an `STM` action to give up on.  `allocSimpleConfigWith` fills+  them with network's `sendAllWith` and `recvWith`.  With them, a write the+  peer does not read is given up on once it has taken the whole timeout,+  and the receiver is stopped after a GOAWAY without being interrupted.+* `runWith` and `runIOWith` take an `STM` action.  When it fires, the server+  sends GOAWAY(NO_ERROR) with the largest stream identifier and a PING, then,+  once the PING is answered or a second has passed, GOAWAY with the last+  stream taken.  Streams above it are refused with REFUSED_STREAM, and the+  connection is closed when those below it are done (RFC 9113, section 6.8).+* A stream is half-closed in the transaction which queues the end of its+  body.  A client which had read the whole response could find the stream+  still open and reset it; enough such resets made the peer close the+  connection with "too many rst_stream".+* Give up on a write which takes the whole timeout.+  [#233](https://github.com/kazu-yamamoto/http2/pull/233)+* Stop the receiver in STM when its reads can give up.+  [#234](https://github.com/kazu-yamamoto/http2/pull/234)+* Start no GOAWAY thread for run and runIO.+  [#235](https://github.com/kazu-yamamoto/http2/pull/235)+* Stop the receiver rather than cancel it when the sender fails first.+  [#238](https://github.com/kazu-yamamoto/http2/pull/238)+* Requires network 3.2.12.+* Allows iproute 1.8 and unix-time 0.6.+ ## 5.4.12  * Close a connection's streams before waiting for its threads.

@@ -76,7 +76,9 @@     confWriteBuffer,     confBufferSize,     confSendAll,+    confSendAllWith,     confReadN,+    confReadNWith,     confPositionReadMaker,     confTimeoutManager,     confMySockAddr,

@@ -3,11 +3,12 @@ module Network.HTTP2.H2.Config where  import Control.Concurrent.STM+import Control.Monad (void) import Data.IORef import Foreign.Marshal.Alloc (free, mallocBytes) import Network.HTTP.Semantics.Client import Network.Socket-import Network.Socket.ByteString (sendAll)+import Network.Socket.ByteString (sendAll, sendAllWith) import qualified System.TimeManager as T  import Network.HPACK@@ -31,13 +32,19 @@ --   reader ends as it does at end of input.  Nothing is closed and no --   thread is interrupted to make it return. --+--   It also fills 'confSendAllWith' and 'confReadNWith' (since 5.4.13), so+--   that the library can give up on a send or a read in the same way.+-- --   @since 5.4.11 allocSimpleConfigWith :: Socket -> BufferSize -> Int -> STM a -> IO Config allocSimpleConfigWith s bufsiz usec giveUp = do     confWriteBuffer <- mallocBytes bufsiz     let confBufferSize = bufsiz     let confSendAll = sendAll s-    confReadN <- (\ref -> defaultReadNWith s ref giveUp) <$> newIORef Nothing+        confSendAllWith = Just $ \giveUp' bs -> sendAllWith s bs giveUp'+    ref <- newIORef Nothing+    let confReadN = defaultReadNWith s ref giveUp+        confReadNWith = Just $ \stop -> defaultReadNWith s ref (void giveUp `orElse` stop)     let confPositionReadMaker = defaultPositionReadMaker     confTimeoutManager <- T.initialize usec     confMySockAddr <- getSocketName s

@@ -4,6 +4,7 @@  module Network.HTTP2.H2.Context where +import Control.Concurrent (ThreadId) import Control.Concurrent.STM import qualified Control.Exception as E import Data.IORef@@ -95,7 +96,7 @@     , continued          :: IORef (Maybe HeaderContinuation)     , myStreamId         :: TVar StreamId     , peerStreamId       :: IORef StreamId-    , peerLastStreamId   :: IORef StreamId+    , peerLastStreamId   :: TVar StreamId     , outputBufferLimit  :: IORef Int     , outputQ            :: TQueue Output     -- ^ Invariant: Each stream will only ever have at most one 'Output'@@ -118,6 +119,10 @@     -- ^ For the threads that are timed out: owning a thread and     --   timing it out are separate, so the manager carries no timer.     , receiverDone       :: TVar (Maybe E.SomeException)+    , stopReceiving      :: TVar Bool+    -- ^ Set to end the receiver at the next frame: see 'stopReceiver'.+    , receiverStoppable  :: Bool+    -- ^ Whether the receiver's reads give up on 'stopReceiving'.     , appDone            :: STM Bool     -- ^ Server only: set by 'sioDone' once a 'runIO' application has     --   nothing more to send.  Never for 'run' and clients.@@ -128,6 +133,11 @@     , peerGoAway         :: TVar (Maybe StreamId)     -- ^ The last stream identifier of the peer's GOAWAY(NO_ERROR), once     --   one has come: see 'goingAway'.+    , myGoAway           :: TVar (Maybe StreamId)+    -- ^ Server only: the last stream identifier of our GOAWAY(NO_ERROR),+    --   once we have sent one: see 'admitPeerStream'.+    , pingAcked          :: TVar ByteString+    -- ^ The payload of the last PING the peer acknowledged.     , activeRequests     :: TVar Int     -- ^ Client only: requests whose 'processResponse' has not returned.     --   A response can be complete, its stream gone from the table, and@@ -179,7 +189,7 @@     continued         <- newIORef Nothing     myStreamId        <- newTVarIO sid0     peerStreamId      <- newIORef 0-    peerLastStreamId  <- newIORef 0+    peerLastStreamId  <- newTVarIO 0     outputBufferLimit <- newIORef buflim     outputQ           <- newTQueueIO     outputQStreamID   <- newTVarIO sid0@@ -206,9 +216,13 @@     threadManager   <- T.newThreadManager     let timeoutManager = timmgr     receiverDone    <- newTVarIO Nothing+    stopReceiving   <- newTVarIO False+    let receiverStoppable = isJust confReadNWith     let informationalCallback = confOnInformational     let appDone = fromMaybe (return False) mdone     peerGoAway      <- newTVarIO Nothing+    myGoAway        <- newTVarIO Nothing+    pingAcked       <- newTVarIO ""     activeRequests  <- newTVarIO 0     resetByMe       <- newIORef IntSet.empty     return Context{..}@@ -227,6 +241,33 @@  ---------------------------------------------------------------- +-- | Ending the receiver as at end of input, once the streams it reads for+--   are done: in 'STM' when its reads can give up, by an exception to it+--   otherwise.+stopReceiver :: Context -> ThreadId -> IO ()+stopReceiver Context{stopReceiving, receiverStoppable} receiver+    | receiverStoppable = atomically $ writeTVar stopReceiving True+    | otherwise = E.throwTo receiver ConnectionIsClosed++----------------------------------------------------------------++-- | Fires once the receiver has been told to stop.+receiverStopped :: Context -> STM ()+receiverStopped Context{stopReceiving} = readTVar stopReceiving >>= check++----------------------------------------------------------------++-- | Fires once the receiver has ended with 'ConnectionIsTimeout': a+--   write still blocked then is given up on.+timedOut :: Context -> STM ()+timedOut Context{receiverDone} = do+    done <- readTVar receiverDone+    case done >>= E.fromException of+        Just ConnectionIsTimeout -> return ()+        _ -> retry++----------------------------------------------------------------+ isClient :: Context -> Bool isClient ctx = role ctx == Client @@ -251,10 +292,19 @@ ----------------------------------------------------------------  getPeerLastStreamId :: Context -> IO StreamId-getPeerLastStreamId ctx = readIORef $ peerLastStreamId ctx+getPeerLastStreamId ctx = readTVarIO $ peerLastStreamId ctx -modifyPeerLastStreamId :: Context -> StreamId -> IO ()-modifyPeerLastStreamId ctx sid = atomicModifyIORef' (peerLastStreamId ctx) $ \n -> if sid > n then (sid, ()) else (n, ())+-- | Counting a stream of the peer's towards the last stream identifier,+--   unless our GOAWAY has gone out below it.  Answers whether it is+--   admitted; one that is not must be refused, not processed.+admitPeerStream :: Context -> StreamId -> IO Bool+admitPeerStream Context{peerLastStreamId, myGoAway} sid = atomically $ do+    mlast <- readTVar myGoAway+    case mlast of+        Just lastSid | sid > lastSid -> return False+        _ -> do+            modifyTVar' peerLastStreamId $ max sid+            return True  ---------------------------------------------------------------- @@ -323,9 +373,19 @@     open st = st  halfClosedRemote :: Context -> Stream -> IO ()-halfClosedRemote ctx stream@Stream{streamState} = do-    closingCode <- atomically $ stateTVar streamState closeHalf+halfClosedRemote ctx stream = halfClosedRemoteWith ctx stream $ return ()++-- | 'halfClosedRemote' in the transaction which hands the end of the body+--   to its reader, so that a reader which has seen the end never finds the+--   stream open, and resets it as one it did not read to the end.+halfClosedRemoteWith :: Context -> Stream -> STM a -> IO a+halfClosedRemoteWith ctx stream@Stream{streamState} deliver = do+    (a, closingCode) <- atomically $ do+        a <- deliver+        closingCode <- stateTVar streamState closeHalf+        return (a, closingCode)     traverse_ (closed ctx stream) closingCode+    return a   where     closeHalf :: StreamState -> (Maybe ClosedCode, StreamState)     closeHalf x@(Closed _) = (Nothing, x)@@ -588,3 +648,26 @@                     | otherwise = (evens, odds)                 noneOfMine = maybe True ((> lastSid) . fst) $ IntMap.lookupMin mine             check $ IntMap.null theirs && noneOfMine && active == 0++----------------------------------------------------------------+-- GOAWAY of ours++-- | Fixing the last stream identifier of our GOAWAY(NO_ERROR) at the+--   streams admitted so far.  Answers it the first time only.+goingAwayMine :: Context -> STM (Maybe StreamId)+goingAwayMine Context{peerLastStreamId, myGoAway} = do+    old <- readTVar myGoAway+    case old of+        Just _ -> return Nothing+        Nothing -> do+            lastSid <- readTVar peerLastStreamId+            writeTVar myGoAway $ Just lastSid+            return $ Just lastSid++-- | Waiting until the streams below our GOAWAY are done.  Those above it+--   are refused, so none of the peer's is left.+drainedMine :: Context -> STM ()+drainedMine Context{myGoAway, oddStreamTable} = do+    readTVar myGoAway >>= check . isJust+    odds <- oddTable <$> readTVar oddStreamTable+    check $ IntMap.null odds

@@ -23,7 +23,6 @@ import Data.Void import Network.Control import Network.HTTP.Semantics-import qualified System.IO.Error as E import qualified System.ThreadManager as T import qualified System.TimeManager as TM @@ -72,10 +71,19 @@             else                 TM.withHandle (timeoutManager ctx) (E.throwTo tid ConnectionIsTimeout) loop2 +    -- Told to stop, the receiver ends between frames as at end of input.+    stopped = receiverStopped ctx+    waitOrStop = do+        stop <-+            atomically $+                (True <$ stopped) `orElse` (False <$ waitControlQ (controlQ ctx))+        when stop $ E.throwIO ConnectionIsClosed+    readHeader = readNGivingUp conf stopped frameHeaderLength+     loop1 :: IO Void     loop1 = do-        atomically $ waitControlQ $ controlQ ctx-        hd <- confReadN frameHeaderLength -- throwing an exception on timeout+        waitOrStop+        hd <- readHeader -- throwing an exception on timeout         when (BS.null hd) $ E.throwIO ConnectionIsClosed         processFrame ctx conf $ decodeFrameHeader hd         loop1@@ -85,8 +93,8 @@         -- If 'confReadN' is timeouted, 'ConnectionIsTimeout' is thrown         -- to destroy the thread trees.  So it is if the peer does not         -- read what we send, and we wait here for ever.-        atomically $ waitControlQ $ controlQ ctx-        hd <- confReadN frameHeaderLength+        waitOrStop+        hd <- readHeader         TM.tickle th         when (BS.null hd) $ E.throwIO ConnectionIsClosed         processFrame ctx conf $ decodeFrameHeader hd@@ -105,13 +113,13 @@     | isServer ctx =         E.throwIO $             ConnectionErrorIsSent ProtocolError streamId "push promise is not allowed"-processFrame Context{..} conf (ftyp, FrameHeader{payloadLength, streamId})+processFrame ctx@Context{..} conf (ftyp, FrameHeader{payloadLength, streamId})     | ftyp > maxFrameType = do         mx <- readIORef continued         case mx of             Nothing -> do                 -- ignoring unknown frame-                void $ readPayload conf payloadLength+                void $ readPayload ctx conf payloadLength             Just _ -> E.throwIO $ ConnectionErrorIsSent ProtocolError streamId "unknown frame" processFrame ctx@Context{..} conf typhdr@(ftyp, header) = do     -- My SETTINGS_MAX_FRAME_SIZE@@ -136,20 +144,20 @@ -- 'confReadN' answers with an empty string at end of input, so a payload that -- comes back short means the peer hung up in the middle of the frame.  Saying -- so here keeps every decoder below from being handed fewer bytes than the--- frame header promised it.-readPayload :: Config -> Int -> IO ByteString-readPayload Config{..} len = do-    bs <- confReadN len+-- frame header promised it.  A receiver told to stop gives it up the same.+readPayload :: Context -> Config -> Int -> IO ByteString+readPayload ctx conf len = do+    bs <- readNGivingUp conf (receiverStopped ctx) len     when (BS.length bs /= len) $ E.throwIO ConnectionIsClosed     return bs  controlOrStream :: Context -> Config -> FrameType -> FrameHeader -> IO () controlOrStream ctx@Context{..} conf ftyp header@FrameHeader{flags, streamId, payloadLength}     | isControl streamId = do-        bs <- readPayload conf payloadLength+        bs <- readPayload ctx conf payloadLength         control ftyp header bs ctx     | ftyp == FramePushPromise = do-        bs <- readPayload conf payloadLength+        bs <- readPayload ctx conf payloadLength         -- A promised stream can be refused over concurrency too, and by the         -- time 'push' gets that far it has decoded the field block, so the         -- same reasoning as 'resettable' applies: reset the promised stream@@ -159,7 +167,7 @@     | otherwise = do         mcont <- checkContinued         mstrm <- getStream ctx ftyp streamId-        bs <- readPayload conf payloadLength+        bs <- readPayload ctx conf payloadLength         case mcont of             Just hc -> continuation hc bs mstrm             Nothing ->@@ -262,7 +270,7 @@                         state <- onResponseHeaders ctx streamId hcl (hcEndOfStream hc) tbl                         processState state ctx strm streamId                     Open _ (Body q mcl bodyLength tlr) -> do-                        state <- onTrailers ctx streamId blk q mcl bodyLength tlr+                        state <- onTrailers ctx strm streamId blk q mcl bodyLength tlr                         processState state ctx strm streamId                     _otherwise ->                         -- The block began on a stream that was open for@@ -370,9 +378,13 @@ -- that had been acted on. launchHandler :: Context -> Stream -> InpObj -> IO () launchHandler ctx@Context{roleInfo} strm inpObj = do-    modifyPeerLastStreamId ctx $ streamNumber strm+    -- One above our GOAWAY is refused: the peer retries it elsewhere.+    admitted <- admitPeerStream ctx sid+    unless admitted $ E.throwIO $ StreamErrorIsSent RefusedStream sid "going away"     let ServerInfo{..} = toServerInfo roleInfo     launch ctx strm inpObj+  where+    sid = streamNumber strm  ---------------------------------------------------------------- @@ -496,8 +508,9 @@                         setframe = CFrames (Just peerAlist) (frames ++ [ack])                     writeIORef myFirstSettings True                     enqueueControl controlQ setframe-control FramePing FrameHeader{flags, streamId} bs ctx@Context{mySettings, pingRate} =-    unless (testAck flags) $ do+control FramePing FrameHeader{flags, streamId} bs ctx@Context{mySettings, pingRate, pingAcked}+    | testAck flags = atomically $ writeTVar pingAcked bs+    | otherwise = do         rate <- getRate pingRate         if rate > pingRateLimit mySettings             then E.throwIO $ ConnectionErrorIsSent EnhanceYourCalm streamId "too many ping"@@ -514,7 +527,7 @@                 receiver <- myThreadId                 T.forkManaged (threadManager ctx) "H2 draining after GOAWAY" $ do                     atomically $ drained ctx-                    E.throwTo receiver ConnectionIsClosed+                    stopReceiver ctx receiver         else E.throwIO $ ConnectionErrorIsReceived err sid $ Short.toShort msg control FrameWindowUpdate header bs ctx = do     WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs@@ -635,14 +648,14 @@                     return s  -- Transition (stream2)-stream FrameHeaders header@FrameHeader{flags, streamId} bs ctx s@(Open _ (Body q mcl bodyLength tlr)) _ = do+stream FrameHeaders header@FrameHeader{flags, streamId} bs ctx s@(Open _ (Body q mcl bodyLength tlr)) strm = do     HeadersFrame _ frag <- guardIt $ decodeHeadersFrame header bs     let endOfStream = testEndStream flags     -- checking frag == "" is not necessary     if endOfStream         then do             if testEndHeader flags-                then onTrailers ctx streamId frag q mcl bodyLength tlr+                then onTrailers ctx strm streamId frag q mcl bodyLength tlr                 else do                     startHeaderBlock ctx streamId endOfStream frag                     return s@@ -714,12 +727,20 @@                     ProtocolError                     streamId                     "actual body length is not the same as content-length"+        -- The end goes in with the half-close: see 'halfClosedRemoteWith'.+        let deliver+                | endOfStream =+                    halfClosedRemoteWith ctx strm . (<* writeTQueue q (Right (mempty, True)))+                | otherwise = atomically         -- Empty Frame Flooding - CVE-2019-9518         if body == ""-            then unless endOfStream $ do-                rate <- getRate emptyFrameRate-                when (rate > emptyFrameRateLimit mySettings) $ do-                    E.throwIO $ ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty data"+            then+                if endOfStream+                    then deliver $ return ()+                    else do+                        rate <- getRate emptyFrameRate+                        when (rate > emptyFrameRateLimit mySettings) $ do+                            E.throwIO $ ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty data"             else do                 writeIORef bodyLength len                 -- Not for a stream closed since its state was read, by@@ -732,7 +753,7 @@                 -- the stream.  Checked in the same transaction, so that                 -- a closer that has emptied the queue finds nothing put                 -- in after it.-                outcome <- atomically $ do+                outcome <- deliver $ do                     st <- readTVar streamState                     discard <- readTVar streamDiscardBody                     if@@ -748,12 +769,7 @@                         unless (connectionCreditedOnArrival streamId) $                             giveBackConnectionWindow ctx $                                 BS.length body-        if endOfStream-            then do-                -- no trailers-                atomically $ writeTQueue q $ Right (mempty, True)-                return HalfClosedRemote-            else return s+        return $ if endOfStream then HalfClosedRemote else s  -- (No state transition) stream FrameWindowUpdate header bs _ s strm = do@@ -938,9 +954,6 @@     sid <- getPeerLastStreamId ctx     return $ goawayFrame sid err msg -sendGoaway :: Config -> ByteString -> IO ()-sendGoaway Config{..} frame = confSendAll frame `E.catchIOError` \_ -> return ()- ----------------------------------------------------------------  sendPing :: Context -> Bool -> ByteString -> IO ()@@ -958,6 +971,7 @@ -- the block is decoded, for the dynamic table. onTrailers     :: Context+    -> Stream     -> StreamId     -> HeaderBlockFragment     -> TQueue (Either E.SomeException (ByteString, Bool))@@ -967,7 +981,7 @@     -- ^ The length of the body received     -> IORef (Maybe TokenHeaderTable)     -> IO StreamState-onTrailers ctx streamId blk q mcl bodyLength tlr = do+onTrailers ctx strm streamId blk q mcl bodyLength tlr = do     tbl <- hpackDecodeTrailer blk streamId ctx     len <- readIORef bodyLength     when (just mcl (/= len)) $@@ -977,7 +991,7 @@                 streamId                 "actual body length is not the same as content-length"     writeIORef tlr (Just tbl)-    atomically $ writeTQueue q $ Right (mempty, True)+    halfClosedRemoteWith ctx strm $ writeTQueue q $ Right (mempty, True)     return HalfClosedRemote  -- | Start accumulating a header block that does not fit in a single frame

@@ -19,6 +19,7 @@ import Network.HTTP.Semantics.Client import Network.HTTP.Semantics.IO import qualified System.ThreadManager as T+import qualified System.TimeManager as TM  import Imports import Network.HPACK (setLimitForEncoding, toTokenHeaderTable)@@ -107,10 +108,27 @@ checkDone _ _ _ = return Nothing  frameSender :: Context -> Config -> IO E.SomeException-frameSender+frameSender ctx conf@Config{confSendAllWith} = do+    labelMe "H2 sender"+    case confSendAllWith of+        Nothing -> frameSenderWith ctx conf $ sendAllGivingUp conf retry+        -- A write is given up on once it has taken the whole timeout: the+        -- receiver's timeout no longer covers it once the receiver has ended.+        Just _ -> do+            stalled <- newTVarIO False+            TM.withHandle (timeoutManager ctx) (atomically $ writeTVar stalled True) $ \th -> do+                TM.pause th+                frameSenderWith ctx conf $ \bs -> do+                    TM.resume th+                    let giveUp = timedOut ctx `orElse` (readTVar stalled >>= check)+                    sendAllGivingUp conf giveUp bs <* TM.pause th++frameSenderWith+    :: Context -> Config -> (ByteString -> IO (Either () ())) -> IO E.SomeException+frameSenderWith     ctx@Context{outputQ, controlQ, encodeDynamicTable, outputBufferLimit}-    Config{..} = do-        labelMe "H2 sender"+    Config{..}+    send = do         -- DATA that has to wait for the connection window, in the order it         -- came: see 'dequeue'.         parked <- newTVarIO []@@ -139,7 +157,10 @@         -- the buffer are filled.         flushN :: Offset -> IO ()         flushN 0 = return ()-        flushN n = bufferIO confWriteBuffer n confSendAll+        flushN n = bufferIO confWriteBuffer n $ \bs ->+            -- Given up on once the connection has timed out: closed with+            -- GOAWAY like any other timeout.+            send bs >>= either (\_ -> E.throwIO ConnectionIsTimeout) return          flushIfNecessary :: Offset -> IO Offset         flushIfNecessary off = do

@@ -15,6 +15,7 @@ import Network.HTTP.Semantics.Client import Network.HTTP.Semantics.IO import Network.Socket hiding (Stream)+import System.IO.Error (catchIOError) import System.IO.Unsafe import qualified System.TimeManager as T @@ -266,7 +267,28 @@     --   So, this value is announced via SETTINGS_MAX_FRAME_SIZE     --   to the peer.     , confSendAll :: ByteString -> IO ()+    -- ^ Not used when 'confSendAllWith' is 'Just'.  When replacing it in+    --   a config which has one, e.g. from 'allocSimpleConfig', set+    --   'confSendAllWith' to 'Nothing' or to a send of the same kind.+    , confSendAllWith :: Maybe (STM () -> ByteString -> IO (Either () ()))+    -- ^ 'confSendAll' which the library can give up on: 'Left' when the+    --   'STM' action fires first.  With it, a write which the peer does+    --   not read is ended by the timeout rather than by killing a thread.+    --   Part of the data may have gone, so the connection is closed after.+    --+    --   @since 5.4.13     , confReadN :: Int -> IO ByteString+    -- ^ When 'confReadNWith' is 'Just', frames are read with it and only+    --   the connection preface with this.  When replacing it in a config+    --   which has one, e.g. from 'allocSimpleConfig', set 'confReadNWith'+    --   to 'Nothing' or to a read of the same kind sharing its buffer.+    , confReadNWith :: Maybe (STM () -> Int -> IO ByteString)+    -- ^ 'confReadN' which the library can give up on: it answers with an+    --   empty string, as at end of input, when the 'STM' action fires+    --   first.  With it, the reader is stopped without being interrupted.+    --   It must share its buffer with 'confReadN'.+    --+    --   @since 5.4.13     , confPositionReadMaker :: PositionReadMaker     , confTimeoutManager :: TM.Manager     , confMySockAddr :: SockAddr@@ -290,7 +312,9 @@         { confWriteBuffer = nullPtr         , confBufferSize = 0         , confSendAll = \_ -> return ()+        , confSendAllWith = Nothing         , confReadN = \_ -> return ""+        , confReadNWith = Nothing         , confPositionReadMaker = defaultPositionReadMaker         , confTimeoutManager = T.defaultManager         , confMySockAddr = SockAddrInet 0 0@@ -298,6 +322,31 @@         , confReadNTimeout = False         , confOnInformational = \_ _ -> return ()         }++-- | Sending with 'confSendAllWith' if there is one, giving up when the+--   'STM' action fires, and with 'confSendAll' otherwise.+sendAllGivingUp :: Config -> STM () -> ByteString -> IO (Either () ())+sendAllGivingUp Config{..} giveUp bs = case confSendAllWith of+    Just sendWith -> sendWith giveUp bs+    Nothing -> Right <$> confSendAll bs++-- | Reading with 'confReadNWith' if there is one, giving up when the+--   'STM' action fires, and with 'confReadN' otherwise.+readNGivingUp :: Config -> STM () -> Int -> IO ByteString+readNGivingUp Config{..} giveUp n = case confReadNWith of+    Just readWith -> readWith giveUp n+    Nothing -> confReadN n++-- | Sending GOAWAY, as a last word: it waits one second at most, since a+--   peer which does not read would hold the connection open for good.+sendGoaway :: Config -> ByteString -> IO ()+sendGoaway conf frame = send `catchIOError` \_ -> return ()+  where+    send = case confSendAllWith conf of+        Nothing -> confSendAll conf frame+        Just _ -> do+            late <- registerDelay 1000000+            void $ sendAllGivingUp conf (readTVar late >>= check) frame  isAsyncException :: E.Exception e => e -> Bool isAsyncException e =

@@ -26,6 +26,7 @@ module Network.HTTP2.Server (     -- * Runner     run,+    runWith,      -- * Server configuration     ServerConfig,@@ -56,7 +57,9 @@     confWriteBuffer,     confBufferSize,     confSendAll,+    confSendAllWith,     confReadN,+    confReadNWith,     confPositionReadMaker,     confTimeoutManager,     confMySockAddr,@@ -77,4 +80,5 @@     ServerConfig (..),     defaultServerConfig,     run,+    runWith,  )

@@ -9,6 +9,7 @@     Stream,     ServerIO (..),     runIO,+    runIOWith, ) where  import Network.HTTP.Semantics.Server

@@ -45,12 +45,24 @@  -- | Running HTTP/2 server. run :: ServerConfig -> Config -> Server -> IO ()-run sconf conf server = do+run = runMaybe Nothing++-- | Like 'run', but the connection goes away once the 'STM' action fires:+--   GOAWAY(NO_ERROR) at once, streams above it refused, and the+--   connection closed when those below it are done.+--+--   @since 5.4.13+runWith :: STM () -> ServerConfig -> Config -> Server -> IO ()+runWith = runMaybe . Just++-- Nothing for 'run': no thread waits on an action which never fires.+runMaybe :: Maybe (STM ()) -> ServerConfig -> Config -> Server -> IO ()+runMaybe mGoAway sconf conf server = do     ok <- checkPreface conf     when ok $ do         let lnch = runServer conf server         ctx <- setup sconf conf lnch Nothing-        runH2 conf ctx+        runH2 conf ctx mGoAway  ---------------------------------------------------------------- @@ -71,7 +83,27 @@     -> Config     -> (ServerIO Stream -> IO (IO ()))     -> IO ()-runIO sconf conf@Config{..} action = do+runIO = runIOMaybe Nothing++-- | Like 'runIO', but the connection goes away once the 'STM' action+--   fires, as with 'runWith'.+--+--   @since 5.4.13+runIOWith+    :: STM ()+    -> ServerConfig+    -> Config+    -> (ServerIO Stream -> IO (IO ()))+    -> IO ()+runIOWith = runIOMaybe . Just++runIOMaybe+    :: Maybe (STM ())+    -> ServerConfig+    -> Config+    -> (ServerIO Stream -> IO (IO ()))+    -> IO ()+runIOMaybe mGoAway sconf conf@Config{..} action = do     ok <- checkPreface conf     when ok $ do         inpQ <- newTQueueIO@@ -97,7 +129,7 @@                     , sioDone = atomically $ writeTVar done True                     }         io <- action serverIO-        concurrently_ io $ runH2 conf ctx+        concurrently_ io $ runH2 conf ctx mGoAway  checkPreface :: Config -> IO Bool checkPreface conf@Config{..} = do@@ -120,32 +152,75 @@         confTimeoutManager         mIsDone -runH2 :: Config -> Context -> IO ()-runH2 conf ctx = do+runH2 :: Config -> Context -> Maybe (STM ()) -> IO ()+runH2 conf ctx mGoAway = do     let mgr = threadManager ctx         runReceiver = frameReceiver ctx conf         runSender = frameSender ctx conf+        -- Two GOAWAYs, as RFC 9113 section 6.8 has it: the first, with+        -- the largest identifier, stops the client opening streams; the+        -- second fixes the last one taken once a PING has gone round, so+        -- that the streams it opened meanwhile are not refused.  The+        -- receiver is stopped as it is after the peer's GOAWAY, once the+        -- streams below ours are done; it reads until then.+        sendGoAway :: STM () -> Async a -> IO ()+        sendGoAway goAway ar = do+            atomically goAway+            enqueueControl (controlQ ctx) $+                CFrames+                    Nothing+                    [ goawayFrame maxStreamId NoError "going away"+                    , pingFrame False goAwayPing+                    ]+            roundTrip <- registerDelay 1000000+            atomically $+                (readTVar (pingAcked ctx) >>= check . (== goAwayPing))+                    `orElse` (readTVar roundTrip >>= check)+            mlast <- atomically $ goingAwayMine ctx+            forM_ mlast $ \lastSid ->+                enqueueControl (controlQ ctx) $+                    CFrames Nothing [goawayFrame lastSid NoError "going away"]+            atomically $ drainedMine ctx+            stopReceiver ctx $ asyncThreadId ar+        withGoAway ar body = case mGoAway of+            Nothing -> body+            Just goAway -> withAsync (sendGoAway goAway ar) $ \_ -> body         runBackgroundThreads =             withAsync runReceiver $ \ar ->-                withAsync runSender $ \as -> do-                    r <- waitEither ar as-                    e <- case r of-                        -- The receiver is done; the sender finishes once it-                        -- has flushed what is queued.-                        Left _ -> wait as-                        -- The sender finished first.  Either the receiver is-                        -- done too and not yet seen to be, and this is its-                        -- error, or the sender failed: nothing more would go-                        -- out, and leaving the receiver to run on left the-                        -- connection open and silent, with no GOAWAY.  Both-                        -- are closed with it.-                        Right e -> return e-                    closureServer conf ctx e+                withGoAway ar $+                    withAsync runSender $ \as -> do+                        r <- waitEither ar as+                        e <- case r of+                            -- The receiver is done; the sender finishes once it+                            -- has flushed what is queued.+                            Left _ -> wait as+                            -- The sender finished first.  Either the receiver is+                            -- done too and not yet seen to be, and this is its+                            -- error, or the sender failed: nothing more would go+                            -- out, and leaving the receiver to run on left the+                            -- connection open and silent, with no GOAWAY.  Both+                            -- are closed with it: the receiver is stopped+                            -- rather than cancelled when its reads can give up.+                            Right e -> do+                                stopReceiver ctx $ asyncThreadId ar+                                late <- registerDelay 1000000+                                atomically $+                                    void (waitCatchSTM ar) `orElse` (readTVar late >>= check)+                                return e+                        closureServer conf ctx e     stopAfterAndWait mgr runBackgroundThreads $ \res ->         closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) res +-- | The payload of the PING sent with the first GOAWAY.+goAwayPing :: ByteString+goAwayPing = "goaway\0\0"++-- | The largest stream identifier.+maxStreamId :: StreamId+maxStreamId = 2147483647+ -- connClose must not be called here since Run:fork calls it goaway :: Config -> ErrorCode -> ByteString -> IO ()-goaway Config{..} etype debugmsg = confSendAll bytestream+goaway conf etype debugmsg = sendGoaway conf bytestream   where     bytestream = goawayFrame 0 etype debugmsg

@@ -1,6 +1,6 @@ cabal-version:      2.0 name:               http2-version:            5.4.12+version:            5.4.13 license:            BSD3 license-file:       LICENSE maintainer:         Kazu Yamamoto <kazu@iij.ad.jp>@@ -117,13 +117,13 @@         containers >=0.6,         http-semantics >= 0.4.2 && <0.5,         http-types >=0.12 && <0.13,-        iproute >= 1.7 && < 1.8,-        network >=3.1,+        iproute >= 1.7 && < 1.9,+        network >=3.2.12,         network-byte-order >=0.1.7 && <0.2,         network-control >=0.1 && <0.2,         stm >=2.5 && <2.6,         time-manager >=0.5 && <0.6,-        unix-time >=0.4.11 && <0.6,+        unix-time >=0.4.11 && <0.7,         utf8-string >=1.0 && <1.1  executable h2c-client@@ -293,6 +293,7 @@         HPACK.IntegerSpec         HTTP2.ClientSpec         HTTP2.FrameSpec+        HTTP2.GiveUpSpec         HTTP2.ServerSpec      default-language:   Haskell2010@@ -312,6 +313,7 @@         network-byte-order,         network-run >= 0.7 && <0.8,         random,+        stm,         time-manager >=0.5 && <0.6,         typed-process 

@@ -0,0 +1,149 @@+{-# LANGUAGE OverloadedStrings #-}++module HTTP2.GiveUpSpec (spec) where++import Control.Concurrent+import Control.Concurrent.STM+import qualified Control.Exception as E+import Control.Monad (forever)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import Data.ByteString.Builder (byteString)+import Data.Word (Word8)+import Network.HTTP.Types+import Network.Run.TCP+import Network.Socket (+    ShutdownCmd (..),+    SocketOption (..),+    setSocketOption,+    shutdown,+ )+import Network.Socket.ByteString+import System.IO.Unsafe+import System.Random+import System.Timeout (timeout)+import Test.Hspec++import Network.HTTP2.Frame (connectionPreface)+import Network.HTTP2.Server++port :: String+port = show $ unsafePerformIO (randomPort <$> getStdGen)+  where+    randomPort = fst . randomR (45321 :: Int, 46320)++host :: String+host = "127.0.0.1"++-- | The timeout of the server, in microseconds.+serverTimeout :: Int+serverTimeout = 1000000++spec :: Spec+spec = describe "give-up" $ do+    it "gives up on a write the peer does not read" $ do+        ended <- newEmptyMVar+        let server _req _aux sendResponse =+                sendResponse (responseStreaming ok200 [] endless) []+            endless write flush = forever $ write (byteString chunk) >> flush+            chunk = B.replicate 65536 0x41+            runServer = runTCPServer (Just host) port $ \s ->+                E.bracket (allocSimpleConfig' s 32768 serverTimeout) freeSimpleConfig $ \conf ->+                    (run defaultServerConfig conf server `E.catch` ignore)+                        `E.finally` putMVar ended ()+        E.bracket (forkIO runServer) killThread $ \_ -> do+            threadDelay 10000+            -- The client opens the windows wide, asks, and never reads,+            -- so the server's writes stop in the socket, not on a window.+            r <- runTCPClient host port $ \s -> do+                -- A fixed small buffer: one that grows would take the data in.+                setSocketOption s RecvBuffer 4096+                sendAll s connectionPreface+                sendAll s $ frame 4 0 0 $ B.pack [0, 4, 0x7f, 0xff, 0xff, 0xff]+                sendAll s $ frame 8 0 0 $ B.pack [0x7f, 0xff, 0, 0]+                sendAll s $ frame 1 5 1 getHeaderBlock+                timeout 6000000 $ takeMVar ended+            r `shouldBe` Just ()+    it "gives up on a write the peer does not read after the receiver ends" $ do+        ended <- newEmptyMVar+        let server _req _aux sendResponse =+                sendResponse (responseStreaming ok200 [] endless) []+            endless write flush = forever $ write (byteString chunk) >> flush+            chunk = B.replicate 65536 0x41+            runServer = runTCPServer (Just host) port $ \s ->+                E.bracket (allocSimpleConfig' s 32768 serverTimeout) freeSimpleConfig $ \conf ->+                    (run defaultServerConfig conf server `E.catch` ignore)+                        `E.finally` putMVar ended ()+        E.bracket (forkIO runServer) killThread $ \_ -> do+            threadDelay 10000+            -- As above, but the client ends its side once it has asked, so+            -- the receiver ends on EOF rather than by its timeout.+            r <- runTCPClient host port $ \s -> do+                setSocketOption s RecvBuffer 4096+                sendAll s connectionPreface+                sendAll s $ frame 4 0 0 $ B.pack [0, 4, 0x7f, 0xff, 0xff, 0xff]+                sendAll s $ frame 8 0 0 $ B.pack [0x7f, 0xff, 0, 0]+                sendAll s $ frame 1 5 1 getHeaderBlock+                threadDelay 100000+                shutdown s ShutdownSend+                timeout 6000000 $ takeMVar ended+            r `shouldBe` Just ()+    it "stops a receiver which cannot be interrupted once it has gone away" $ do+        ended <- newEmptyMVar+        goVar <- newTVarIO False+        let server _req _aux sendResponse = sendResponse (responseNoBody ok200 []) []+            -- A read no exception reaches, as on a platform where a blocked+            -- recv cannot be interrupted; the one which gives up still can.+            uninterruptible conf = conf{confReadN = E.uninterruptibleMask_ . confReadN conf}+            runServer = runTCPServer (Just host) port $ \s ->+                E.bracket (allocSimpleConfig' s 32768 30000000) freeSimpleConfig $ \conf ->+                    (go (uninterruptible conf) `E.catch` ignore) `E.finally` putMVar ended ()+            go conf = runWith (readTVar goVar >>= check) defaultServerConfig conf server+        E.bracket (forkIO runServer) killThread $ \_ -> do+            threadDelay 10000+            -- The client says nothing after its SETTINGS and does not answer+            -- the PING, so the receiver waits in a read when it is stopped.+            r <- runTCPClient host port $ \s -> do+                sendAll s connectionPreface+                sendAll s $ frame 4 0 0 ""+                threadDelay 100000+                atomically $ writeTVar goVar True+                timeout 5000000 $ takeMVar ended+            r `shouldBe` Just ()+    it "stops a receiver which cannot be interrupted once the sender has failed" $ do+        ended <- newEmptyMVar+        let server _req _aux sendResponse = sendResponse (responseNoBody ok200 []) []+            -- No exception reaches the reads, and every write fails: the+            -- receiver used to be cancelled, which waited for good.+            broken conf =+                conf+                    { confReadN = E.uninterruptibleMask_ . confReadN conf+                    , confReadNWith = fmap (\r stop -> E.uninterruptibleMask_ . r stop) (confReadNWith conf)+                    , confSendAllWith = Just $ \_ _ -> E.throwIO $ userError "broken"+                    }+            runServer = runTCPServer (Just host) port $ \s ->+                E.bracket (allocSimpleConfig' s 32768 30000000) freeSimpleConfig $ \conf ->+                    (run defaultServerConfig (broken conf) server `E.catch` ignore)+                        `E.finally` putMVar ended ()+        E.bracket (forkIO runServer) killThread $ \_ -> do+            threadDelay 10000+            r <- runTCPClient host port $ \s -> do+                sendAll s connectionPreface+                sendAll s $ frame 4 0 0 ""+                timeout 5000000 $ takeMVar ended+            r `shouldBe` Just ()+  where+    ignore :: E.SomeException -> IO ()+    ignore _ = return ()++-- | A frame of this type, flags and stream with this payload.+frame :: Word8 -> Word8 -> Word8 -> ByteString -> ByteString+frame typ flags sid payload =+    B.pack [0, fromIntegral (len `div` 256), fromIntegral (len `mod` 256), typ, flags, 0, 0, 0, sid]+        <> payload+  where+    len = B.length payload++-- | GET / for 127.0.0.1, from the static table and one literal.+getHeaderBlock :: ByteString+getHeaderBlock = B.pack [0x82, 0x86, 0x84, 0x01, 0x09] <> "127.0.0.1"

@@ -18,6 +18,7 @@  import Control.Concurrent import Control.Concurrent.Async+import Control.Concurrent.STM (TVar, atomically, check, newTVarIO, readTVar, writeTVar) import qualified Control.Exception as E import Control.Monad import Crypto.Hash (Context, SHA1)@@ -27,6 +28,7 @@ import Data.ByteString.Builder (Builder, byteString) import qualified Data.ByteString.Char8 as C8 import Data.IORef+import Data.List (sort) import Data.Maybe (isJust, isNothing) import Network.HTTP.Semantics import Network.HTTP.Types@@ -445,6 +447,56 @@                     Just (Left e) -> expectationFailure $ show (e :: C.HTTP2Error)                     Nothing -> expectationFailure "timed out" +        it "answers the streams it has taken and refuses the rest once it goes away" $ do+            -- RFC 9113, section 6.8: GOAWAY with the largest identifier, a+            -- PING, and once the PING is answered GOAWAY(NO_ERROR) with the+            -- last stream taken; one opened after it is refused, and the+            -- connection is closed once stream 1 is answered.+            goVar <- newTVarIO False+            withServer (runServerGoingAway id goVar) $ do+                threadDelay 10000+                timeout 5000000 (goAwayFromServer False goVar)+                    `shouldReturn` Just+                        [ "GOAWAY 2147483647 NoError"+                        , "GOAWAY 1 NoError"+                        , "RST_STREAM 3 RefusedStream"+                        , "DATA 1 END_STREAM"+                        , "GOAWAY 1 NoError"+                        ]++        it "goes away the same with reads which cannot give up" $ do+            -- Without 'confReadNWith', as for warp and http2-tls, the+            -- receiver is stopped by an exception instead.+            goVar <- newTVarIO False+            withServer (runServerGoingAway noReadNWith goVar) $ do+                threadDelay 10000+                timeout 5000000 (goAwayFromServer False goVar)+                    `shouldReturn` Just+                        [ "GOAWAY 2147483647 NoError"+                        , "GOAWAY 1 NoError"+                        , "RST_STREAM 3 RefusedStream"+                        , "DATA 1 END_STREAM"+                        , "GOAWAY 1 NoError"+                        ]++        it "answers a stream opened between its two GOAWAYs" $ do+            -- The second GOAWAY waits for the PING, so a stream already on+            -- its way when the first came is taken, not refused.+            goVar <- newTVarIO False+            withServer (runServerGoingAway id goVar) $ do+                threadDelay 10000+                r <- timeout 5000000 (goAwayFromServer True goVar)+                sort <$> r+                    `shouldBe` Just+                        ( sort+                            [ "GOAWAY 2147483647 NoError"+                            , "GOAWAY 3 NoError"+                            , "DATA 3 END_STREAM"+                            , "DATA 1 END_STREAM"+                            , "GOAWAY 3 NoError"+                            ]+                        )+         it "closes the connection once a runIO application is done" $             -- 'sioDone' set a flag that nothing read, so the connection             -- stayed open until the client closed it.@@ -478,6 +530,12 @@                 timeout 5000000 goAwayFromClient                     `shouldReturn` Just ["HEADERS 1", "DATA 1 END_STREAM", "GOAWAY NoError"] +        it "answers them the same with reads which cannot give up" $+            withServer (runServerWithConfig noReadNWith) $ do+                threadDelay 10000+                timeout 5000000 goAwayFromClient+                    `shouldReturn` Just ["HEADERS 1", "DATA 1 END_STREAM", "GOAWAY NoError"]+         it "resets a malformed request and goes on serving the connection" $             -- An upper-case field name makes the request malformed: a             -- stream error (RFC 9113, section 8.1.1).  It used to close the@@ -670,14 +728,22 @@ ignoreHTTP2Error _ = pure ()  runServer :: ServerSettings -> IO ()-runServer set = runTCPServerWithSettings set (Just host) port runHTTP2Server+runServer = runServerWithConfig id++-- | Like 'runServer', with the config changed.+runServerWithConfig :: (Config -> Config) -> ServerSettings -> IO ()+runServerWithConfig modify set = runTCPServerWithSettings set (Just host) port runHTTP2Server   where     runHTTP2Server s =         E.bracket             (allocSimpleConfig s 32768)             freeSimpleConfig-            (\conf -> run defaultServerConfig conf server)+            (\conf -> run defaultServerConfig (modify conf) server) +-- | A config whose reads the library cannot give up on.+noReadNWith :: Config -> Config+noReadNWith conf = conf{confReadNWith = Nothing}+ -- | Like 'runServer', but announcing room for a single concurrent stream. -- | Uploading 100000 octets of a file through 'C.runIO', and what the server -- says it received.@@ -929,6 +995,75 @@                     ]     _ <- E.try (action s) :: IO (Either E.IOException ())     collectFrames s++-- | Like 'runServerWithConfig', going away once the variable is set.+runServerGoingAway :: (Config -> Config) -> TVar Bool -> ServerSettings -> IO ()+runServerGoingAway modify goVar set = runTCPServerWithSettings set (Just host) port runHTTP2Server+  where+    runHTTP2Server s =+        E.bracket+            (allocSimpleConfig s 32768)+            freeSimpleConfig+            (\conf -> runWith (readTVar goVar >>= check) defaultServerConfig (modify conf) server)++-- | A request for /slow, then the server told to go away, and a request for+-- /: before the PING that came with the first GOAWAY is answered when+-- early, after the second GOAWAY otherwise.  What the server sends from its+-- first GOAWAY on.+goAwayFromServer :: Bool -> TVar Bool -> IO [String]+goAwayFromServer early goVar = runTCPClient host port $ \s -> do+    sendAll s connectionPreface+    sendAll s $ encodeFrame (EncodeInfo defaultFlags 0 Nothing) $ SettingsFrame []+    request s 1 "/slow"+    threadDelay 50000+    atomically $ writeTVar goVar True+    first <- readUntil s isGoAway+    when early $ request s 3 "/"+    ping <- readUntil s isPing+    second <-+        if early+            then return []+            else readUntil s isGoAway <* request s 3 "/"+    rest <- collect s+    return $ first ++ ping ++ second ++ rest+  where+    request s sid path =+        sendAll s $+            encodeFrame (EncodeInfo (setEndStream $ setEndHeader defaultFlags) sid Nothing) $+                HeadersFrame Nothing $+                    hpackEncode+                        [ (":scheme", "http")+                        , (":authority", "127.0.0.1")+                        , (":path", path)+                        , (":method", "GET")+                        ]+    isGoAway (ftyp, _, _) = ftyp == FrameGoAway+    isPing (ftyp, fh, _) = ftyp == FramePing && not (testAck (flags fh))+    -- Up to and including the first frame which satisfies the predicate;+    -- a PING is answered as it comes.+    readUntil s done = do+        mf <- recvFrame s+        case mf of+            Nothing -> return []+            Just f -> do+                answerPing s f+                if done f+                    then return $ maybe [] pure $ frameText f+                    else maybe id (:) (frameText f) <$> readUntil s done+    collect s = readUntil s (const False)+    answerPing s (FramePing, fh, p)+        | not (testAck (flags fh)) =+            sendAll s $ encodeFrame (EncodeInfo (setAck defaultFlags) 0 Nothing) $ PingFrame p+    answerPing _ _ = return ()+    frameText (FrameGoAway, fh, p)+        | Right (GoAwayFrame sid err _) <- decodeGoAwayFrame fh p =+            Just $ "GOAWAY " ++ show sid ++ " " ++ show err+    frameText (FrameRSTStream, fh, p)+        | Right (RSTStreamFrame err) <- decodeRSTStreamFrame fh p =+            Just $ "RST_STREAM " ++ show (streamId fh) ++ " " ++ show err+    frameText (FrameData, fh, _)+        | testEndStream (flags fh) = Just $ "DATA " ++ show (streamId fh) ++ " END_STREAM"+    frameText _ = Nothing  -- | A request for /slow, then GOAWAY(NO_ERROR) at once.  What the server -- sends from then on until it closes the connection.