http2 5.3.3 → 5.3.4
raw patch · 10 files changed
+231/−100 lines, 10 filesdep ~http-semanticsPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: http-semantics
API changes (from Hackage documentation)
Files
- ChangeLog.md +5/−0
- Network/HTTP2/Client/Run.hs +5/−21
- Network/HTTP2/H2/Context.hs +2/−0
- Network/HTTP2/H2/Receiver.hs +30/−10
- Network/HTTP2/H2/Sender.hs +48/−24
- Network/HTTP2/H2/Stream.hs +86/−6
- Network/HTTP2/H2/Sync.hs +28/−7
- Network/HTTP2/H2/Types.hs +6/−0
- Network/HTTP2/Server/Worker.hs +19/−30
- http2.cabal +2/−2
ChangeLog.md view
@@ -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.
Network/HTTP2/Client/Run.hs view
@@ -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 ()
Network/HTTP2/H2/Context.hs view
@@ -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
Network/HTTP2/H2/Receiver.hs view
@@ -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 -} ----------------------------------------------------------------
Network/HTTP2/H2/Sender.hs view
@@ -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
Network/HTTP2/H2/Stream.hs view
@@ -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
Network/HTTP2/H2/Sync.hs view
@@ -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 ()))
Network/HTTP2/H2/Types.hs view
@@ -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 =
Network/HTTP2/Server/Worker.hs view
@@ -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.
http2.cabal view
@@ -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 <kazu@iij.ad.jp>@@ -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,