diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,10 @@ # ChangeLog for http2 +## 5.3.4 + +* Support stream cancellation + [#142](https://github.com/kazu-yamamoto/http2/pull/142) + ## 5.3.3 * Enclosing IPv6 literal authority with square brackets. 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 @@ -218,22 +218,14 @@ q <- sendStreaming ctx newstrm strmbdy let next = nextForStreaming q return (Just next, Just q) - (vs, out) <- + ((var, sync), out) <- prepareSync newstrm (OHeader outObjHeaders mnext outObjTrailers) mtbq atomically $ do sidOK <- readTVar outputQStreamID check (sidOK == sid) enqueueOutputSTM outputQ out writeTVar outputQStreamID (sid + 2) - forkManaged threadManager "H2 worker" $ syncWithSender ctx newstrm vs - where - nextForStreaming - :: TBQueue StreamingChunk - -> DynaNext - nextForStreaming tbq = - let takeQ = atomically $ tryReadTBQueue tbq - next = fillStreamBodyGetNext takeQ - in next + forkManaged threadManager "H2 worker" $ syncWithSender ctx newstrm var sync sendStreaming :: Context @@ -241,18 +233,10 @@ -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk) sendStreaming Context{..} strm strmbdy = do - tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 let label = "H2 streaming supporter for stream " ++ show (streamNumber strm) - forkManagedUnmask threadManager label $ \unmask -> do - let iface = - OutBodyIface - { outBodyUnmask = unmask - , outBodyPush = \b -> atomically $ writeTBQueue tbq $ StreamingBuilder b NotEndOfStream - , outBodyPushFinal = \b -> atomically $ writeTBQueue tbq $ StreamingBuilder b (EndOfStream Nothing) - , outBodyFlush = atomically $ writeTBQueue tbq StreamingFlush - } - finished = atomically $ writeTBQueue tbq $ StreamingFinished Nothing - strmbdy iface `finally` finished + tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 + forkManagedUnmask threadManager label $ \unmask -> + withOutBodyIface tbq unmask strmbdy return tbq exchangeSettings :: Context -> IO () diff --git a/Network/HTTP2/H2/Context.hs b/Network/HTTP2/H2/Context.hs --- a/Network/HTTP2/H2/Context.hs +++ b/Network/HTTP2/H2/Context.hs @@ -73,6 +73,8 @@ , peerStreamId :: IORef StreamId , outputBufferLimit :: IORef Int , outputQ :: TQueue Output + -- ^ Invariant: Each stream will only ever have at most one 'Output' + -- object in this queue at any moment. , outputQStreamID :: TVar StreamId , controlQ :: TQueue Control , encodeDynamicTable :: DynamicTable 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 @@ -2,6 +2,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternGuards #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} module Network.HTTP2.H2.Receiver ( frameReceiver, @@ -561,18 +562,26 @@ -- > /Either endpoint/ can send a RST_STREAM frame from this state, causing -- > it to transition immediately to "closed". -- - -- (emphasis not in original). This justifies the two non-error cases, - -- below. (Section 8.1 of the spec is also relevant, but it is less explicit - -- about the /either endpoint/ part.) - case (s, err) of - (Open (Just _) _, NoError) -> - -- HalfClosedLocal - return (Closed cc) - (HalfClosedRemote, NoError) -> - return (Closed cc) + -- (emphasis not in original). + -- + -- In addition, the spec states (about the open state): + -- + -- > Either endpoint can send a RST_STREAM frame from this state, causing it + -- > to transition immediately to "closed". + -- + -- This justifies the two non-error cases, below. (Section 8.1 of the spec + -- is also relevant, but it is less explicit about the /either endpoint/ + -- part.) + case s of + Open _ _ + | isNonCritical err -> + -- Open /or/ half-closed (local) + return (Closed cc) + HalfClosedRemote + | isNonCritical err -> + return (Closed cc) _otherwise -> do E.throwIO $ StreamErrorIsReceived err streamId - -- (No state transition) stream FramePriority header bs _ s Stream{streamNumber} = do -- ignore @@ -601,6 +610,17 @@ E.throwIO $ StreamErrorIsSent ProtocolError streamId $ fromString ("illegal frame " ++ show x ++ " for " ++ show streamId) + +{- FOURMOLU_DISABLE -} +-- Although some stream errors indicate misbehaving peers, such as +-- FLOW_CONTROL_ERROR, not all errors do. We will close the connection only +-- for critical errors. +isNonCritical :: ErrorCode -> Bool +isNonCritical NoError = True +isNonCritical Cancel = True +isNonCritical InternalError = True +isNonCritical _ = False +{- FOURMOLU_ENABLE -} ---------------------------------------------------------------- diff --git a/Network/HTTP2/H2/Sender.hs b/Network/HTTP2/H2/Sender.hs --- a/Network/HTTP2/H2/Sender.hs +++ b/Network/HTTP2/H2/Sender.hs @@ -140,7 +140,7 @@ ---------------------------------------------------------------- outputOrEnqueueAgain :: Output -> Offset -> IO Offset - outputOrEnqueueAgain out@(Output strm otyp sync) off = E.handle resetStream $ do + outputOrEnqueueAgain out@(Output strm otyp sync) off = E.handle (\e -> resetStream strm InternalError e >> return off) $ do state <- readStreamState strm if isHalfClosedLocal state then return off @@ -150,6 +150,10 @@ -- No need to check the streaming window (applies to DATA frames only) outputHeader strm hdr mnext tlrmkr sync off _ -> do + -- The 'sync' function usage constraints hold here: We + -- just popped off the only 'Output' for this stream, + -- and we only enqueue a new output (in 'output') if + -- 'sync' returns 'True' ok <- sync $ Just otyp if ok then do @@ -158,13 +162,13 @@ let lim = min cws sws output out off lim else return off - where - resetStream e = do - closed ctx strm (ResetByMe e) - let rst = resetFrame InternalError $ streamNumber strm - enqueueControl controlQ $ CFrames Nothing [rst] - return off + resetStream :: Stream -> ErrorCode -> E.SomeException -> IO () + resetStream strm err e = do + closed ctx strm (ResetByMe e) + let rst = resetFrame err $ streamNumber strm + enqueueControl controlQ $ CFrames Nothing [rst] + ---------------------------------------------------------------- outputHeader :: Stream @@ -200,17 +204,36 @@ let payloadOff = off0 + frameHeaderLength datBuf = confWriteBuffer `plusPtr` payloadOff datBufSiz = buflim - payloadOff - Next datPayloadLen reqflush mnext <- curr datBuf (min datBufSiz lim) - NextTrailersMaker tlrmkr' <- runTrailersMaker tlrmkr datBuf datPayloadLen - fillDataHeaderEnqueueNext - strm - off0 - datPayloadLen - mnext - tlrmkr' - sync - out - reqflush + curr datBuf (min datBufSiz lim) >>= \next -> + case next of + Next datPayloadLen reqflush mnext -> do + NextTrailersMaker tlrmkr' <- runTrailersMaker tlrmkr datBuf datPayloadLen + fillDataHeaderEnqueueNext + strm + off0 + datPayloadLen + mnext + tlrmkr' + sync + out + reqflush + CancelNext mErr -> do + -- Stream cancelled + -- + -- At this point, the headers have already been sent. + -- Therefore, the stream cannot be in the 'Idle' state, so we + -- are justified in sending @RST_STREAM@. + -- + -- By the invariant on the 'outputQ', there are no other + -- outputs for this stream already enqueued. Therefore, we can + -- safely cancel it knowing that we won't try and send any + -- more data frames on this stream. + case mErr of + Just err -> + resetStream strm InternalError err + Nothing -> + resetStream strm Cancel (E.toException CancelledStream) + return off0 output (Output strm (OPush ths pid) sync) off0 _lim = do -- Creating a push promise header -- Frame id should be associated stream id from the client. @@ -290,12 +313,13 @@ -- Avoid sending an empty data frame before trailers at the end -- of a stream off' <- - if datPayloadLen /= 0 || isNothing mtrailers then do - decreaseWindowSize ctx strm datPayloadLen - fillFrameHeader FrameData datPayloadLen streamNumber flag buf - return $ off + frameHeaderLength + datPayloadLen - else - return off + if datPayloadLen /= 0 || isNothing mtrailers + then do + decreaseWindowSize ctx strm datPayloadLen + fillFrameHeader FrameData datPayloadLen streamNumber flag buf + return $ off + frameHeaderLength + datPayloadLen + else + return off off'' <- handleTrailers mtrailers off' _ <- sync Nothing halfClosedLocal ctx strm Finished 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 @@ -1,12 +1,18 @@ +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} module Network.HTTP2.H2.Stream where import Control.Monad +import Control.Monad.STM (throwSTM) import Data.IORef import Data.Maybe (fromMaybe) import Network.Control +import Network.HTTP.Semantics +import Network.HTTP.Semantics.IO import UnliftIO.Concurrent import UnliftIO.Exception import UnliftIO.STM @@ -81,19 +87,93 @@ where finalize strm = do st <- readStreamState strm - void . tryPutMVar (streamInput strm) $ - Left $ - fromMaybe (toException ConnectionIsClosed) $ - mErr + void $ tryPutMVar (streamInput strm) err case st of Open _ (Body q _ _ _) -> atomically $ writeTQueue q $ maybe (Right (mempty, True)) Left mErr _otherwise -> return () + mErr :: Maybe SomeException mErr = case mErr' of - Just err - | Just ConnectionIsClosed <- fromException err -> + Just e + | Just ConnectionIsClosed <- fromException e -> Nothing _otherwise -> mErr' + + err :: Either SomeException a + err = + Left $ + fromMaybe (toException ConnectionIsClosed) $ + mErr + +---------------------------------------------------------------- + +data StreamTerminated + = StreamPushedFinal + | StreamCancelled + | StreamOutOfScope + deriving (Show) + deriving anyclass (Exception) + +withOutBodyIface + :: TBQueue StreamingChunk + -> (forall a. IO a -> IO a) + -> (OutBodyIface -> IO r) + -> IO r +withOutBodyIface tbq unmask k = do + terminated <- newTVarIO Nothing + let whenNotTerminated act = do + mTerminated <- readTVar terminated + case mTerminated of + Just reason -> + throwSTM reason + Nothing -> + act + + terminateWith reason act = do + mTerminated <- readTVar terminated + case mTerminated of + Just _ -> + -- Already terminated + return () + Nothing -> do + writeTVar terminated (Just reason) + act + + iface = + OutBodyIface + { outBodyUnmask = unmask + , outBodyPush = \b -> + atomically $ + whenNotTerminated $ + writeTBQueue tbq $ + StreamingBuilder b NotEndOfStream + , outBodyPushFinal = \b -> + atomically $ whenNotTerminated $ do + writeTVar terminated (Just StreamPushedFinal) + writeTBQueue tbq $ StreamingBuilder b (EndOfStream Nothing) + writeTBQueue tbq $ StreamingFinished Nothing + , outBodyFlush = + atomically $ + whenNotTerminated $ + writeTBQueue tbq StreamingFlush + , outBodyCancel = \mErr -> + atomically $ + terminateWith StreamCancelled $ + writeTBQueue tbq (StreamingCancelled mErr) + } + finished = atomically $ do + terminateWith StreamOutOfScope $ + writeTBQueue tbq $ + StreamingFinished Nothing + k iface `finally` finished + +nextForStreaming + :: TBQueue StreamingChunk + -> DynaNext +nextForStreaming tbq = + let takeQ = atomically $ tryReadTBQueue tbq + next = fillStreamBodyGetNext takeQ + in next diff --git a/Network/HTTP2/H2/Sync.hs b/Network/HTTP2/H2/Sync.hs --- a/Network/HTTP2/H2/Sync.hs +++ b/Network/HTTP2/H2/Sync.hs @@ -11,6 +11,18 @@ import Network.HTTP2.H2.Types import Network.HTTP2.H2.Window +-- | Two assumptions about how this function is used: +-- +-- 1. A separate thread will be running 'syncWithSender' using the @var@ and +-- @sync@ values constructed here. +-- 2. The 'Output' will be enqueued in the 'outputQ' of some 'Context' +-- +-- The returned @sync@ function then has the following usage constraints: +-- +-- 1. It may only be called with a 'Just' 'OutputType' if there is no 'Output' +-- already enqueued in the 'outputQ' for the given stream. +-- 2. If the function returns 'False', no other 'Output' may be enqueued for +-- this stream (until one has been dequeued). prepareSync :: Stream -> OutputType @@ -18,16 +30,20 @@ -> IO ((MVar Sync, Maybe OutputType -> IO Bool), Output) prepareSync strm otyp mtbq = do var <- newEmptyMVar - let sync = makeSync strm mtbq (putMVar var) + let sync = makeSync strm mtbq var out = Output strm otyp sync return ((var, sync), out) syncWithSender :: Context -> Stream - -> (MVar Sync, Maybe OutputType -> IO Bool) + -> MVar Sync + -- ^ Precondition: When this is filled with an 'Output' for a particular + -- stream, the 'outputQ' in the 'Context' /must not/ already contain an + -- 'Output' for that stream. + -> (Maybe OutputType -> IO Bool) -> IO () -syncWithSender Context{..} strm (var, sync) = loop +syncWithSender Context{..} strm var sync = loop where loop = do s <- takeMVar var @@ -35,22 +51,27 @@ Done -> return () Cont wait newotyp -> do wait + -- This is justified by the precondition above enqueueOutput outputQ $ Output strm newotyp sync loop +-- | Postcondition: This will only write to the 'MVar' if: +-- +-- 1. You pass 'Just' an 'OutputType' +-- 2. The return value is 'False' makeSync :: Stream -> Maybe (TBQueue StreamingChunk) - -> (Sync -> IO ()) + -> MVar Sync -> Maybe OutputType -> IO Bool -makeSync _ _ sync Nothing = sync Done >> return False -makeSync strm mtbq sync (Just otyp) = do +makeSync _ _ var Nothing = putMVar var Done >> return False +makeSync strm mtbq var (Just otyp) = do mwait <- checkOpen strm mtbq case mwait of Nothing -> return True Just wait -> do - sync $ Cont wait otyp + putMVar var $ Cont wait otyp return False checkOpen :: Stream -> Maybe (TBQueue StreamingChunk) -> IO (Maybe (IO ())) 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 @@ -1,3 +1,4 @@ +{-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} @@ -121,6 +122,11 @@ | Reset ErrorCode | ResetByMe SomeException deriving (Show) + +-- | Used for streams which are cancelled by calling +-- 'Network.HTTP.Semantics.outBodyCancel'. +data CancelledStream = CancelledStream + deriving (Show, E.Exception) closedCodeToError :: StreamId -> ClosedCode -> HTTP2Error closedCodeToError sid cc = 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 @@ -13,7 +13,6 @@ import Network.HTTP.Semantics.Server.Internal import Network.HTTP.Types import qualified System.TimeManager as T -import qualified UnliftIO.Exception as E import UnliftIO.STM import Imports hiding (insert) @@ -79,9 +78,9 @@ ] ot = OPush promiseRequest pid Response rsp = promiseResponse pp - (vc, out) <- prepareSync newstrm ot Nothing + ((var, sync), out) <- prepareSync newstrm ot Nothing enqueueOutput outputQ out - syncWithSender ctx newstrm vc + syncWithSender ctx newstrm var sync increment tvar sendHeaderBody conf ctx th newstrm rsp push tvar pps (n + 1) @@ -129,18 +128,10 @@ q <- sendStreaming ctx strm th strmbdy let next = nextForStreaming q return (Just next, Just q) - (vc, out) <- + ((var, sync), out) <- prepareSync strm (OHeader outObjHeaders mnext outObjTrailers) mtbq enqueueOutput outputQ out - syncWithSender ctx strm vc - where - nextForStreaming - :: TBQueue StreamingChunk - -> DynaNext - nextForStreaming tbq = - let takeQ = atomically $ tryReadTBQueue tbq - next = fillStreamBodyGetNext takeQ - in next + syncWithSender ctx strm var sync sendStreaming :: Context @@ -149,24 +140,22 @@ -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk) sendStreaming Context{..} strm th strmbdy = do - tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 let label = "H2 streaming supporter for stream " ++ show (streamNumber strm) - forkManaged threadManager label $ do - let iface = - OutBodyIface - { outBodyUnmask = id - , outBodyPush = \b -> do - T.pause th - atomically $ writeTBQueue tbq (StreamingBuilder b NotEndOfStream) - T.resume th - , outBodyPushFinal = \b -> do - T.pause th - atomically $ writeTBQueue tbq (StreamingBuilder b (EndOfStream Nothing)) - T.resume th - , outBodyFlush = atomically $ writeTBQueue tbq StreamingFlush - } - finished = atomically $ writeTBQueue tbq $ StreamingFinished Nothing - strmbdy iface `E.finally` finished + tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 + forkManaged threadManager label $ + withOutBodyIface tbq id $ \iface -> do + let iface' = + iface + { outBodyPush = \b -> do + T.pause th + outBodyPush iface b + T.resume th + , outBodyPushFinal = \b -> do + T.pause th + outBodyPushFinal iface b + T.resume th + } + strmbdy iface' return tbq -- | Worker for server applications. 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.3.3 +version: 5.3.4 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto @@ -116,7 +116,7 @@ bytestring >=0.10, case-insensitive >=1.2 && <1.3, containers >=0.6, - http-semantics >= 0.2 && <0.3, + http-semantics >= 0.2.1 && <0.3, http-types >=0.12 && <0.13, iproute >= 1.7 && < 1.8, network >=3.1,