http2 5.4.3 → 5.4.4
raw patch · 23 files changed
+262/−167 lines, 23 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ Network.HTTP2.Client: StreamCancelled :: StreamTerminated
+ Network.HTTP2.Client: StreamOutOfScope :: StreamTerminated
+ Network.HTTP2.Client: StreamPushedFinal :: StreamTerminated
+ Network.HTTP2.Client: StreamRemoteReset :: ClosedCode -> StreamTerminated
+ Network.HTTP2.Client: StreamResetIsReceived :: ErrorCode -> StreamId -> HTTP2Error
+ Network.HTTP2.Client: data StreamTerminated
Files
- ChangeLog.md +5/−0
- Network/HPACK/HeaderBlock/Decode.hs +12/−12
- Network/HPACK/HeaderBlock/Encode.hs +2/−3
- Network/HPACK/Huffman/Decode.hs +4/−4
- Network/HPACK/Huffman/Encode.hs +2/−2
- Network/HPACK/Table/Dynamic.hs +4/−4
- Network/HPACK/Types.hs +2/−2
- Network/HTTP2/Client.hs +2/−0
- Network/HTTP2/Client/Run.hs +7/−6
- Network/HTTP2/Frame/Decode.hs +2/−2
- Network/HTTP2/H2/Context.hs +24/−21
- Network/HTTP2/H2/OutBodyIface.hs +130/−0
- Network/HTTP2/H2/Receiver.hs +3/−6
- Network/HTTP2/H2/Sender.hs +16/−6
- Network/HTTP2/H2/Stream.hs +7/−68
- Network/HTTP2/H2/StreamTable.hs +3/−3
- Network/HTTP2/H2/Sync.hs +10/−6
- Network/HTTP2/H2/Types.hs +10/−13
- Network/HTTP2/Server/Worker.hs +3/−2
- bench-hpack/Main.hs +1/−1
- http2.cabal +2/−1
- test-hpack/HPACKDecode.hs +2/−2
- test/HTTP2/ServerSpec.hs +9/−3
ChangeLog.md view
@@ -1,5 +1,10 @@ # ChangeLog for http2 +## 5.4.4++* Improvements for dealing with RST_STREAM+ [#172](https://github.com/kazu-yamamoto/http2/pull/172)+ ## 5.4.3 * auxSendInformational: gate usage with CPP to http-semantics >= 0.4.1
Network/HPACK/HeaderBlock/Decode.hs view
@@ -14,7 +14,7 @@ decodeSimple, -- testing ) where -import Control.Exception (catch, throwIO)+import qualified Control.Exception as E import Data.Array.Base (unsafeRead, unsafeWrite) import qualified Data.Array.IO as IOA import qualified Data.Array.Unsafe as Unsafe@@ -65,7 +65,7 @@ -- ^ An HPACK format -> IO TokenHeaderTable decodeTokenHeader dyntbl inp =- decodeHPACK dyntbl inp (decodeSophisticated (toTokenHeader dyntbl)) `catch` \BufferOverrun -> throwIO HeaderBlockTruncated+ decodeHPACK dyntbl inp (decodeSophisticated (toTokenHeader dyntbl)) `E.catch` \BufferOverrun -> E.throwIO HeaderBlockTruncated decodeHPACK :: DynamicTable@@ -150,34 +150,34 @@ then do mx <- unsafeRead arr tokenIx -- duplicated- when (isJust mx) $ throwIO IllegalHeaderName+ when (isJust mx) $ E.throwIO IllegalHeaderName -- unknown- when (isMaxTokenIx tokenIx) $ throwIO IllegalHeaderName+ when (isMaxTokenIx tokenIx) $ E.throwIO IllegalHeaderName unsafeWrite arr tokenIx (Just v) pseudo else do -- 0-Length Headers Leak - CVE-2019-9516- when (tokenKey == "") $ throwIO IllegalHeaderName+ when (tokenKey == "") $ E.throwIO IllegalHeaderName when (isMaxTokenIx tokenIx && B8.any isUpper (original tokenKey)) $- throwIO IllegalHeaderName+ E.throwIO IllegalHeaderName unsafeWrite arr tokenIx (Just v) if isCookieTokenIx tokenIx then normal 0 empty (empty << v) else normal 0 (empty << tv) empty else return [] normal n builder cookie- | n > headerLimit = throwIO TooLargeHeader+ | n > headerLimit = E.throwIO TooLargeHeader | otherwise = do leftover <- remainingSize rbuf if leftover >= 1 then do w <- read8 rbuf tv@(Token{..}, v) <- decTokenHeader w rbuf- when isPseudo $ throwIO IllegalHeaderName+ when isPseudo $ E.throwIO IllegalHeaderName -- 0-Length Headers Leak - CVE-2019-9516- when (tokenKey == "") $ throwIO IllegalHeaderName+ when (tokenKey == "") $ E.throwIO IllegalHeaderName when (isMaxTokenIx tokenIx && B8.any isUpper (original tokenKey)) $- throwIO IllegalHeaderName+ E.throwIO IllegalHeaderName unsafeWrite arr tokenIx (Just v) if isCookieTokenIx tokenIx then normal (n + 1) builder (cookie << v)@@ -197,7 +197,7 @@ toTokenHeader dyntbl w rbuf | w `testBit` 7 = indexed dyntbl w rbuf | w `testBit` 6 = incrementalIndexing dyntbl w rbuf- | w `testBit` 5 = throwIO IllegalTableSizeUpdate+ | w `testBit` 5 = E.throwIO IllegalTableSizeUpdate | w `testBit` 4 = neverIndexing dyntbl w rbuf | otherwise = withoutIndexing dyntbl w rbuf @@ -206,7 +206,7 @@ let w' = mask5 w siz <- decodeI 5 w' rbuf suitable <- isSuitableSize siz dyntbl- unless suitable $ throwIO TooLargeTableSize+ unless suitable $ E.throwIO TooLargeTableSize renewDynamicTable siz dyntbl ----------------------------------------------------------------
Network/HPACK/HeaderBlock/Encode.hs view
@@ -7,7 +7,6 @@ encodeS, ) where -import Control.Exception (bracket, throwIO) import qualified Control.Exception as E import qualified Data.ByteString as BS import Data.ByteString.Internal (create)@@ -68,13 +67,13 @@ -> TokenHeaderList -> IO ByteString -- ^ An HPACK format-encodeHeader' stgy siz dyntbl hs = bracket (mallocBytes siz) free enc+encodeHeader' stgy siz dyntbl hs = E.bracket (mallocBytes siz) free enc where enc buf = do (hs', len) <- encodeTokenHeader buf siz stgy True dyntbl hs case hs' of [] -> create len $ \p -> copyBytes p buf len- _ -> throwIO BufferOverrun+ _ -> E.throwIO BufferOverrun ----------------------------------------------------------------
Network/HPACK/Huffman/Decode.hs view
@@ -9,7 +9,7 @@ GCBuffer, ) where -import Control.Exception (throwIO)+import qualified Control.Exception as E import Data.Array (Array, listArray) import Data.Array.Base (unsafeAt) import qualified Data.ByteString as BS@@ -69,16 +69,16 @@ decH wbuf rbuf len = go len (way256 `unsafeAt` 0) where go 0 way0 = case way0 of- WayStep Nothing _ -> throwIO IllegalEos+ WayStep Nothing _ -> E.throwIO IllegalEos WayStep (Just i) _ | i <= 8 -> return ()- | otherwise -> throwIO TooLongEos+ | otherwise -> E.throwIO TooLongEos go n way0 = do w <- read8 rbuf way <- doit way0 w go (n - 1) way doit way w = case next way w of- EndOfString -> throwIO EosInTheMiddle+ EndOfString -> E.throwIO EosInTheMiddle Forward n -> return $ way256 `unsafeAt` fromIntegral n GoBack n v -> do write8 wbuf v
Network/HPACK/Huffman/Encode.hs view
@@ -6,7 +6,7 @@ encodeHuffman, ) where -import Control.Exception (throwIO)+import qualified Control.Exception as E import Data.Array.Base (unsafeAt) import Data.Array.IArray (listArray) import Data.Array.Unboxed (UArray)@@ -75,7 +75,7 @@ off' = off - len {-# INLINE write #-} write p w = do- when (p >= limit) $ throwIO BufferOverrun+ when (p >= limit) $ E.throwIO BufferOverrun let w8 = fromIntegral (w `shiftR` shiftForWrite) :: Word8 poke p w8 let p' = p `plusPtr` 1
Network/HPACK/Table/Dynamic.hs view
@@ -24,7 +24,7 @@ getRevIndex, ) where -import Control.Exception (throwIO)+import qualified Control.Exception as E import Data.Array.Base (unsafeRead, unsafeWrite) import Data.Array.IO (IOArray, newArray) import qualified Data.ByteString.Char8 as BS@@ -43,7 +43,7 @@ {-# INLINE toIndexedEntry #-} toIndexedEntry :: DynamicTable -> Index -> IO Entry toIndexedEntry dyntbl idx- | idx <= 0 = throwIO $ IndexOverrun idx+ | idx <= 0 = E.throwIO $ IndexOverrun idx | idx <= staticTableSize = return $ toStaticEntry idx | otherwise = toDynamicEntry dyntbl idx @@ -121,7 +121,7 @@ {-# INLINE adj #-} adj :: Int -> Int -> IO Int adj maxN x- | maxN == 0 = throwIO TooSmallTableSize+ | maxN == 0 = E.throwIO TooSmallTableSize | otherwise = let ret = (x + maxN) `mod` maxN in return ret@@ -402,7 +402,7 @@ maxN <- readIORef maxNumOfEntries off <- readIORef offset n <- readIORef numOfEntries- when (idx > n + staticTableSize) $ throwIO $ IndexOverrun idx+ when (idx > n + staticTableSize) $ E.throwIO $ IndexOverrun idx didx <- adj maxN (idx + off - staticTableSize) table <- readIORef circularTable unsafeRead table didx
Network/HPACK/Types.hs view
@@ -20,7 +20,7 @@ BufferOverrun (..), ) where -import Control.Exception as E+import qualified Control.Exception as E import Network.ByteOrder (Buffer, BufferOverrun (..), BufferSize) import Imports@@ -87,4 +87,4 @@ | TooLargeHeader deriving (Eq, Show) -instance Exception DecodeError+instance E.Exception DecodeError
Network/HTTP2/Client.hs view
@@ -90,6 +90,7 @@ -- * Error HTTP2Error (..),+ StreamTerminated (..), ReasonPhrase, ErrorCode ( ErrorCode,@@ -115,3 +116,4 @@ import Network.HTTP2.Client.Run import Network.HTTP2.Frame import Network.HTTP2.H2 hiding (authority, scheme)+import Network.HTTP2.H2.OutBodyIface
Network/HTTP2/Client/Run.hs view
@@ -7,7 +7,7 @@ import Control.Concurrent import Control.Concurrent.Async import Control.Concurrent.STM-import Control.Exception+import qualified Control.Exception as E import qualified Data.ByteString.UTF8 as UTF8 import Data.IORef import Data.IP (IPv6)@@ -22,6 +22,7 @@ import Imports import Network.HTTP2.Frame import Network.HTTP2.H2+import Network.HTTP2.H2.OutBodyIface -- | Client configuration data ClientConfig = ClientConfig@@ -120,7 +121,7 @@ getResponse strm = do mRsp <- takeMVar $ streamInput strm case mRsp of- Left err -> throwIO err+ Left err -> E.throwIO err Right rsp -> return $ Response rsp setup :: ClientConfig -> Config -> IO Context@@ -140,7 +141,7 @@ runH2 :: Config -> Context -> IO a -> IO a runH2 conf ctx runClient = do- T.stopAfter mgr (try runAll >>= closureClient conf ctx) $ \res ->+ T.stopAfter mgr (E.try runAll >>= closureClient conf ctx) $ \res -> closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) res where mgr = threadManager ctx@@ -151,7 +152,7 @@ er <- race runReceiver runClient case er of Right r -> return r- Left err -> throwIO err+ Left err -> E.throwIO err -- When 'runClientReceiver' terminates, it is important we give the sender -- a chance to terminate cleanly also (it's possible the client terminated@@ -243,10 +244,10 @@ -> Stream -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)-sendStreaming Context{..} strm strmbdy = do+sendStreaming ctx@Context{..} strm strmbdy = do tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 T.forkManagedUnmask threadManager label $ \unmask ->- withOutBodyIface tbq unmask strmbdy+ withOutBodyIface ctx strm tbq unmask strmbdy return tbq where label = "H2 request streaming sender for stream " ++ show (streamNumber strm)
Network/HTTP2/Frame/Decode.hs view
@@ -23,7 +23,7 @@ decodeContinuationFrame, ) where -import Control.Exception (Exception)+import qualified Control.Exception as E import Data.Array (Array, listArray, (!)) import qualified Data.ByteString as BS import Foreign.Ptr (Ptr, plusPtr)@@ -38,7 +38,7 @@ data FrameDecodeError = FrameDecodeError ErrorCode StreamId ShortByteString deriving (Eq, Show) -instance Exception FrameDecodeError+instance E.Exception FrameDecodeError ----------------------------------------------------------------
Network/HTTP2/H2/Context.hs view
@@ -5,7 +5,6 @@ module Network.HTTP2.H2.Context where import Control.Concurrent.STM-import Control.Exception import qualified Control.Exception as E import Data.IORef import Network.Control@@ -90,7 +89,7 @@ , mySockAddr :: SockAddr , peerSockAddr :: SockAddr , threadManager :: T.ThreadManager- , receiverDone :: TVar (Maybe SomeException)+ , receiverDone :: TVar (Maybe E.SomeException) , workersDone :: STM Bool , informationalCallback :: StreamId -> TokenHeaderTable -> IO () -- ^ Client only: called when a 1xx informational response (e.g. 103 Early@@ -194,46 +193,50 @@ {-# INLINE setStreamState #-} setStreamState :: Context -> Stream -> StreamState -> IO ()-setStreamState _ Stream{streamState} newState = do- oldState <- readIORef streamState+setStreamState _ Stream{streamNumber, streamState} newState = atomically $ do+ oldState <- readTVar streamState++ -- Inform consumers of any streams that we close case (oldState, newState) of (Open _ (Body q _ _ _), Open _ (Body q' _ _ _)) | q == q' -> -- The stream stays open with the same body; nothing to do return ()+ (Open _ (Body q _ _ _), Closed cc) ->+ writeTQueue q $ Left $ E.toException $ closedCodeToError streamNumber cc (Open _ (Body q _ _ _), _) ->- -- The stream is either closed, or is open with a /new/ body- -- We need to close the old queue so that any reads from it won't block- atomically $ writeTQueue q $ Left $ toException ConnectionIsClosed+ -- The stream is opened with a /new/ body+ writeTQueue q $ Left $ E.toException ConnectionIsClosed _otherwise -> -- The stream wasn't open to start with; nothing to do return ()- writeIORef streamState newState + writeTVar streamState newState+ opened :: Context -> Stream -> IO () opened ctx strm = setStreamState ctx strm (Open Nothing JustOpened) halfClosedRemote :: Context -> Stream -> IO () halfClosedRemote ctx stream@Stream{streamState} = do- closingCode <- atomicModifyIORef streamState closeHalf+ closingCode <- atomically $ stateTVar streamState closeHalf traverse_ (closed ctx stream) closingCode where- closeHalf :: StreamState -> (StreamState, Maybe ClosedCode)- closeHalf x@(Closed _) = (x, Nothing)- closeHalf (Open (Just cc) _) = (Closed cc, Just cc)- closeHalf _ = (HalfClosedRemote, Nothing)+ closeHalf :: StreamState -> (Maybe ClosedCode, StreamState)+ closeHalf x@(Closed _) = (Nothing, x)+ closeHalf (Open (Just cc) _) = (Just cc, Closed cc)+ closeHalf _ = (Nothing, HalfClosedRemote) halfClosedLocal :: Context -> Stream -> ClosedCode -> IO () halfClosedLocal ctx stream@Stream{streamState} cc = do- shouldFinalize <- atomicModifyIORef streamState closeHalf+ shouldFinalize <- atomically $ stateTVar streamState closeHalf when shouldFinalize $ closed ctx stream cc where- closeHalf :: StreamState -> (StreamState, Bool)- closeHalf x@(Closed _) = (x, False)- closeHalf HalfClosedRemote = (Closed cc, True)- closeHalf (Open Nothing o) = (Open (Just cc) o, False)- closeHalf _ = (Open (Just cc) JustOpened, False)+ closeHalf :: StreamState -> (Bool, StreamState)+ closeHalf x@(Closed _) = (False, x)+ closeHalf HalfClosedRemote = (True, Closed cc)+ closeHalf (Open Nothing o) = (False, Open (Just cc) o)+ closeHalf _ = (False, Open (Just cc) JustOpened) closed :: Context -> Stream -> ClosedCode -> IO () closed ctx@Context{oddStreamTable, evenStreamTable} strm@Stream{streamNumber} cc = do@@ -242,8 +245,8 @@ else deleteOdd oddStreamTable streamNumber err setStreamState ctx strm (Closed cc) -- anyway where- err :: SomeException- err = toException (closedCodeToError streamNumber cc)+ err :: E.SomeException+ err = E.toException (closedCodeToError streamNumber cc) ---------------------------------------------------------------- -- From peer
+ Network/HTTP2/H2/OutBodyIface.hs view
@@ -0,0 +1,130 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RankNTypes #-}++module Network.HTTP2.H2.OutBodyIface (+ StreamTerminated (..),+ withOutBodyIface,+) where++import Control.Concurrent.STM+import Control.Exception+import Network.HTTP.Semantics+import Network.HTTP.Semantics.IO+import Network.HTTP2.H2.Context+import Network.HTTP2.H2.Sync+import Network.HTTP2.H2.Types++----------------------------------------------------------------++data StreamTerminated+ = StreamPushedFinal+ | StreamCancelled+ | StreamOutOfScope+ | StreamRemoteReset ClosedCode+ deriving (Show)+ deriving anyclass (Exception)++----------------------------------------------------------------++withOutBodyIface+ :: Context+ -> Stream+ -> TBQueue StreamingChunk+ -> (forall a. IO a -> IO a)+ -> (OutBodyIface -> IO r)+ -> IO r+withOutBodyIface ctx@Context{outputQ} strm tbq unmask k = do+ terminated <- newTVarIO Nothing+ let checkNotTerminated :: STM ()+ checkNotTerminated = do+ mTerminated <- readTVar terminated+ maybe (return ()) throwSTM mTerminated++ -- Check if the peer is still listening for messages+ --+ -- It is important to call 'checkNotClosed' prior to enqueuing stream+ -- chunks to ensure that 'writeTBQueue' will not block indefinitely+ -- (because nothing is consuming elements from the queue anymore).+ --+ -- Assumes 'checkNotTerminated'.+ checkNotClosed :: STM ()+ checkNotClosed = do+ mClosed <- getIsClosed+ case mClosed of+ Just code ->+ -- When the stream is closed, but /we/ did not close it (or+ -- 'checkNotTerminated' would have thrown an exception), it+ -- must mean that our peer send us a RST_STREAM, indicating+ -- that they do not want to receive any further messages.+ throwSTM $ StreamRemoteReset code+ _otherwise ->+ return ()++ getIsClosed :: STM (Maybe ClosedCode)+ getIsClosed = do+ st <- readTVar (streamState strm)+ case st of+ Closed code -> return $ Just code+ _otherwise -> return Nothing++ cancelAfterFinish :: Maybe SomeException -> STM ()+ cancelAfterFinish mErr =+ writeTQueue outputQ $ makeOutputIO ctx strm (OReset mErr)++ iface :: OutBodyIface+ iface =+ OutBodyIface+ { outBodyUnmask = unmask+ , outBodyPush = \b -> atomically $ do+ checkNotTerminated+ checkNotClosed+ writeTBQueue tbq $ StreamingBuilder b NotEndOfStream+ , outBodyPushFinal = \b -> atomically $ do+ checkNotTerminated+ checkNotClosed+ writeTVar terminated (Just StreamPushedFinal)+ writeTBQueue tbq $ StreamingBuilder b (EndOfStream Nothing)+ writeTBQueue tbq $ StreamingFinished Nothing+ , outBodyFlush = atomically $ do+ checkNotTerminated+ checkNotClosed+ writeTBQueue tbq StreamingFlush+ , outBodyCancel = \mErr -> atomically $ do+ mTerminated <- readTVar terminated+ mClosed <- getIsClosed+ case (mClosed, mTerminated) of+ (Nothing, Nothing) -> do+ writeTVar terminated (Just StreamCancelled)+ writeTBQueue tbq $ StreamingCancelled mErr+ (Nothing, Just StreamCancelled) ->+ -- Already cancelled+ return ()+ (Nothing, Just _) -> do+ -- We finished streaming (that is, sending messages to the peer),+ -- but we must still be able to cancel the stream entirely+ -- (that is, tell the peer that we no longer want to /receive/ messages: RST_STREAM)+ writeTVar terminated (Just StreamCancelled)+ cancelAfterFinish mErr+ (Just _code, _) ->+ -- Peer already closed+ return ()+ }++ finished :: IO ()+ finished = atomically $ do+ mTerminated <- readTVar terminated+ mClosed <- getIsClosed+ case (mClosed, mTerminated) of+ (Nothing, Nothing) -> do+ writeTVar terminated (Just StreamOutOfScope)+ writeTBQueue tbq $ StreamingFinished Nothing+ (Nothing, Just _) ->+ -- We already terminated+ return ()+ (Just _code, _) ->+ -- Peer already closed+ return ()++ k iface `finally` finished
Network/HTTP2/H2/Receiver.hs view
@@ -22,6 +22,7 @@ import Data.Void import Network.Control import Network.HTTP.Semantics+import qualified System.IO.Error as E import qualified System.ThreadManager as T import Imports hiding (delete, insert)@@ -53,6 +54,7 @@ case mErr of Left err -> do atomically $ writeTVar receiverDone $ Just err+ -- err is re-thrown by "runH2" return err Right x -> do absurd x -- We only terminate due to exceptions@@ -708,12 +710,7 @@ return $ goawayFrame sid err msg sendGoaway :: Config -> ByteString -> IO ()-sendGoaway Config{..} frame = confSendAll frame `E.catch` ignore--ignore :: E.SomeException -> IO ()-ignore (E.SomeException e)- | isAsyncException e = E.throwIO e- | otherwise = return ()+sendGoaway Config{..} frame = confSendAll frame `E.catchIOError` \_ -> return () ----------------------------------------------------------------
Network/HTTP2/H2/Sender.hs view
@@ -79,6 +79,7 @@ ctx@Context{outputQ, controlQ, encodeDynamicTable, outputBufferLimit} Config{..} = do labelMe "H2 sender"+ -- err is re-thrown by "runH2" loop 0 `E.catch` return where ----------------------------------------------------------------@@ -161,7 +162,13 @@ outputAndSync 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+ then case otyp of+ OReset mErr | not (isClosed state) -> do+ -- RST_STREAM is the only frame we can still send after half-closing+ resetStreamWith strm mErr+ return off+ _otherwise ->+ return off else case otyp of OHeader hdr mnext tlrmkr -> do (off', mout') <- outputHeader strm hdr mnext tlrmkr sync off@@ -179,6 +186,7 @@ sync mout' return off' + ---------------------------------------------------------------- resetStream :: Stream -> ErrorCode -> E.SomeException -> IO () resetStream strm err e | isAsyncException e = E.throwIO e@@ -187,6 +195,12 @@ let rst = resetFrame err $ streamNumber strm enqueueControl controlQ $ CFrames Nothing [rst] + resetStreamWith :: Stream -> Maybe E.SomeException -> IO ()+ resetStreamWith strm (Just err) =+ resetStream strm InternalError err+ resetStreamWith strm Nothing =+ resetStream strm Cancel (E.toException CancelledStream)+ ---------------------------------------------------------------- outputHeader :: Stream@@ -262,11 +276,7 @@ -- 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)+ resetStreamWith strm mErr return (off0, Nothing) output (Output strm (OPush ths pid) _) off0 _lim = do -- Creating a push promise header
Network/HTTP2/H2/Stream.hs view
@@ -1,18 +1,14 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE RankNTypes #-} module Network.HTTP2.H2.Stream where import Control.Concurrent import Control.Concurrent.STM-import Control.Exception+import qualified Control.Exception as E import Control.Monad import Data.IORef import Data.Maybe (fromMaybe) import Network.Control-import Network.HTTP.Semantics import Network.HTTP.Semantics.IO import Network.HTTP2.Frame@@ -52,7 +48,7 @@ newOddStream :: StreamId -> WindowSize -> WindowSize -> IO Stream newOddStream sid txwin rxwin = Stream sid- <$> newIORef Idle+ <$> newTVarIO Idle <*> newEmptyMVar <*> newTVarIO (newTxFlow txwin) <*> newIORef (newRxFlow rxwin)@@ -61,7 +57,7 @@ newEvenStream :: StreamId -> WindowSize -> WindowSize -> IO Stream newEvenStream sid txwin rxwin = Stream sid- <$> newIORef Reserved+ <$> newTVarIO Reserved <*> newEmptyMVar <*> newTVarIO (newTxFlow txwin) <*> newIORef (newRxFlow rxwin)@@ -71,12 +67,12 @@ {-# INLINE readStreamState #-} readStreamState :: Stream -> IO StreamState-readStreamState Stream{streamState} = readIORef streamState+readStreamState Stream{streamState} = readTVarIO streamState ---------------------------------------------------------------- closeAllStreams- :: TVar OddStreamTable -> TVar EvenStreamTable -> Maybe SomeException -> IO ()+ :: TVar OddStreamTable -> TVar EvenStreamTable -> Maybe E.SomeException -> IO () closeAllStreams ovar evar mErr = do ostrms <- clearOddStreamTable ovar mapM_ finalize ostrms@@ -95,67 +91,10 @@ _otherwise -> return () - err :: Either SomeException a- err = Left $ fromMaybe (toException ConnectionIsClosed) mErr+ err :: Either E.SomeException a+ err = Left $ fromMaybe (E.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- maybe act throwSTM mTerminated-- 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 =- atomically- . terminateWith StreamCancelled- . writeTBQueue tbq- . StreamingCancelled- }- finished = atomically $ do- terminateWith StreamOutOfScope $- writeTBQueue tbq $- StreamingFinished Nothing- k iface `finally` finished nextForStreaming :: TBQueue StreamingChunk
Network/HTTP2/H2/StreamTable.hs view
@@ -33,7 +33,7 @@ import Control.Concurrent import Control.Concurrent.STM-import Control.Exception+import qualified Control.Exception as E import Data.IntMap.Strict (IntMap) import qualified Data.IntMap.Strict as IntMap import Network.Control (LRUCache)@@ -78,7 +78,7 @@ let oddTable' = IntMap.insert k v oddTable in OddStreamTable oddConc oddTable' -deleteOdd :: TVar OddStreamTable -> IntMap.Key -> SomeException -> IO ()+deleteOdd :: TVar OddStreamTable -> IntMap.Key -> E.SomeException -> IO () deleteOdd var k err = do mv <- atomically deleteStream case mv of@@ -128,7 +128,7 @@ let evenTable' = IntMap.insert k v evenTable in EvenStreamTable evenConc evenTable' evenCache -deleteEven :: TVar EvenStreamTable -> IntMap.Key -> SomeException -> IO ()+deleteEven :: TVar EvenStreamTable -> IntMap.Key -> E.SomeException -> IO () deleteEven var k err = do mv <- atomically deleteStream case mv of
Network/HTTP2/H2/Sync.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE MultiWayIf #-} {-# LANGUAGE RecordWildCards #-} module Network.HTTP2.H2.Sync (@@ -76,7 +77,6 @@ Cont newout -> do cont <- checkLoop lc when cont $ do- -- This is justified by the precondition above enqueueOutput outputQ newout loop @@ -85,13 +85,15 @@ tovar <- newTVarIO False return $ LoopCheck- { lcTBQ = mtbq+ { lcState = streamState strm+ , lcTBQ = mtbq , lcTimeout = tovar , lcWindow = streamTxFlow strm } data LoopCheck = LoopCheck- { lcTBQ :: Maybe (TBQueue StreamingChunk)+ { lcState :: TVar StreamState+ , lcTBQ :: Maybe (TBQueue StreamingChunk) , lcTimeout :: TVar Bool , lcWindow :: TVar TxFlow }@@ -99,9 +101,11 @@ checkLoop :: LoopCheck -> IO Bool checkLoop LoopCheck{..} = atomically $ do tout <- readTVar lcTimeout- if tout- then return False- else do+ state <- readTVar lcState+ if+ | tout -> return False+ | Closed{} <- state -> return False+ | otherwise -> do waitStreaming' lcTBQ waitStreamWindowSizeSTM lcWindow return True
Network/HTTP2/H2/Types.hs view
@@ -7,11 +7,6 @@ import Control.Concurrent import Control.Concurrent.STM-import Control.Exception (- Exception,- SomeAsyncException (..),- SomeException (..),- ) import qualified Control.Exception as E import Data.IORef import Foreign.Ptr (nullPtr)@@ -114,7 +109,7 @@ | NoBody TokenHeaderTable | HasBody TokenHeaderTable | Body- (TQueue (Either SomeException (ByteString, Bool)))+ (TQueue (Either E.SomeException (ByteString, Bool))) (Maybe Int) -- received Content-Length -- compared the body length for error checking (IORef Int) -- actual body length@@ -124,7 +119,7 @@ = Finished | Killed | Reset ErrorCode- | ResetByMe SomeException+ | ResetByMe E.SomeException deriving (Show) -- | Used for streams which are cancelled by calling@@ -137,7 +132,7 @@ case cc of Finished -> ConnectionIsClosed Killed -> ConnectionIsTimeout- Reset err -> ConnectionErrorIsReceived err sid "Connection was reset"+ Reset err -> StreamResetIsReceived err sid ResetByMe err -> BadThingHappen err ----------------------------------------------------------------@@ -163,8 +158,8 @@ data Stream = Stream { streamNumber :: StreamId- , streamState :: IORef StreamState- , streamInput :: MVar (Either SomeException InpObj) -- Client only+ , streamState :: TVar StreamState+ , streamInput :: MVar (Either E.SomeException InpObj) -- Client only , streamTxFlow :: TVar TxFlow , streamRxFlow :: IORef RxFlow , streamRxQ :: IORef (Maybe RxQ)@@ -175,7 +170,7 @@ "Stream{id=" ++ show streamNumber ++ ",state="- ++ show (unsafePerformIO (readIORef streamState))+ ++ show (unsafePerformIO (readTVarIO streamState)) ++ "}" ----------------------------------------------------------------@@ -191,6 +186,7 @@ | OPush TokenHeaderList StreamId -- associated stream id from client | ONext DynaNext TrailersMaker | OInformational [Header]+ | OReset (Maybe E.SomeException) data Sync = Done | Cont Output @@ -214,6 +210,7 @@ | ConnectionErrorIsReceived ErrorCode StreamId ReasonPhrase | ConnectionErrorIsSent ErrorCode StreamId ReasonPhrase | StreamErrorIsReceived ErrorCode StreamId+ | StreamResetIsReceived ErrorCode StreamId | StreamErrorIsSent ErrorCode StreamId ReasonPhrase | BadThingHappen E.SomeException deriving (Show)@@ -297,8 +294,8 @@ , confOnInformational = \_ _ -> return () } -isAsyncException :: Exception e => e -> Bool+isAsyncException :: E.Exception e => e -> Bool isAsyncException e = case E.fromException (E.toException e) of- Just (SomeAsyncException _) -> True+ Just (E.SomeAsyncException _) -> True Nothing -> False
Network/HTTP2/Server/Worker.hs view
@@ -18,6 +18,7 @@ import Imports hiding (insert) import Network.HTTP2.Frame import Network.HTTP2.H2+import Network.HTTP2.H2.OutBodyIface #if MIN_VERSION_http_semantics(0,4,1) import qualified Data.ByteString.Char8 as C8@@ -193,10 +194,10 @@ -> Stream -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)-sendStreaming Context{..} strm strmbdy = do+sendStreaming ctx@Context{..} strm strmbdy = do tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 T.forkManagedTimeout threadManager label $ \th ->- withOutBodyIface tbq id $ \iface -> do+ withOutBodyIface ctx strm tbq id $ \iface -> do let iface' = iface { outBodyPush = \b -> do
bench-hpack/Main.hs view
@@ -2,7 +2,7 @@ module Main where -import Control.Exception+import qualified Control.Exception as E import Criterion.Main import Data.ByteString (ByteString) import Network.HPACK
http2.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: http2-version: 5.4.3+version: 5.4.4 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto <kazu@iij.ad.jp>@@ -89,6 +89,7 @@ Network.HTTP2.H2.Context Network.HTTP2.H2.EncodeFrame Network.HTTP2.H2.HPACK+ Network.HTTP2.H2.OutBodyIface Network.HTTP2.H2.Queue Network.HTTP2.H2.Receiver Network.HTTP2.H2.Sender
test-hpack/HPACKDecode.hs view
@@ -12,7 +12,7 @@ #if __GLASGOW_HASKELL__ < 709 import Control.Applicative ((<$>)) #endif-import Control.Exception+import qualified Control.Exception as E import Control.Monad (when) import qualified Data.ByteString.Base16 as B16 import qualified Data.ByteString.Char8 as B8@@ -66,7 +66,7 @@ case size c of Nothing -> return () Just siz -> renewDynamicTable siz dyntbl- x <- try $ decodeHeader dyntbl inp+ x <- E.try $ decodeHeader dyntbl inp case x of Left e -> return $ Just $ show (e :: DecodeError) Right hs' -> do
test/HTTP2/ServerSpec.hs view
@@ -118,8 +118,14 @@ Just "GET" -> case requestPath req of Just "/" -> sendResponse responseHello [] Just "/early" -> do- auxSendInformational aux earlyHints103 [("link", "</style.css>; rel=preload; as=style")]- auxSendInformational aux earlyHints103 [("link", "</app.js>; rel=preload; as=script")]+ auxSendInformational+ aux+ earlyHints103+ [("link", "</style.css>; rel=preload; as=style")]+ auxSendInformational+ aux+ earlyHints103+ [("link", "</app.js>; rel=preload; as=script")] sendResponse responseHello [] Just "/stream" -> sendResponse responseInfinite [] Just "/push" -> do@@ -196,7 +202,7 @@ runClientEarly :: IORef [TokenHeaderTable] -> IO (Maybe Status) runClientEarly hintsRef = runTCPClient host port $ \s -> E.bracket (allocSimpleConfig s 4096) freeSimpleConfig $ \conf0 ->- C.run cliconf (conf0{confOnInformational = onInformational }) $ \sendRequest _aux ->+ C.run cliconf (conf0{confOnInformational = onInformational}) $ \sendRequest _aux -> sendRequest (C.requestNoBody methodGet "/early" []) (return . C.responseStatus) where cliconf = C.defaultClientConfig{C.authority = host}