diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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 diff --git a/Network/HPACK/HeaderBlock/Decode.hs b/Network/HPACK/HeaderBlock/Decode.hs --- a/Network/HPACK/HeaderBlock/Decode.hs +++ b/Network/HPACK/HeaderBlock/Decode.hs @@ -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 ---------------------------------------------------------------- diff --git a/Network/HPACK/HeaderBlock/Encode.hs b/Network/HPACK/HeaderBlock/Encode.hs --- a/Network/HPACK/HeaderBlock/Encode.hs +++ b/Network/HPACK/HeaderBlock/Encode.hs @@ -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 ---------------------------------------------------------------- diff --git a/Network/HPACK/Huffman/Decode.hs b/Network/HPACK/Huffman/Decode.hs --- a/Network/HPACK/Huffman/Decode.hs +++ b/Network/HPACK/Huffman/Decode.hs @@ -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 diff --git a/Network/HPACK/Huffman/Encode.hs b/Network/HPACK/Huffman/Encode.hs --- a/Network/HPACK/Huffman/Encode.hs +++ b/Network/HPACK/Huffman/Encode.hs @@ -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 diff --git a/Network/HPACK/Table/Dynamic.hs b/Network/HPACK/Table/Dynamic.hs --- a/Network/HPACK/Table/Dynamic.hs +++ b/Network/HPACK/Table/Dynamic.hs @@ -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 diff --git a/Network/HPACK/Types.hs b/Network/HPACK/Types.hs --- a/Network/HPACK/Types.hs +++ b/Network/HPACK/Types.hs @@ -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 diff --git a/Network/HTTP2/Client.hs b/Network/HTTP2/Client.hs --- a/Network/HTTP2/Client.hs +++ b/Network/HTTP2/Client.hs @@ -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 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 @@ -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) diff --git a/Network/HTTP2/Frame/Decode.hs b/Network/HTTP2/Frame/Decode.hs --- a/Network/HTTP2/Frame/Decode.hs +++ b/Network/HTTP2/Frame/Decode.hs @@ -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 ---------------------------------------------------------------- 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 @@ -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 diff --git a/Network/HTTP2/H2/OutBodyIface.hs b/Network/HTTP2/H2/OutBodyIface.hs new file mode 100644 --- /dev/null +++ b/Network/HTTP2/H2/OutBodyIface.hs @@ -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 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 @@ -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 () ---------------------------------------------------------------- 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 @@ -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 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,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 diff --git a/Network/HTTP2/H2/StreamTable.hs b/Network/HTTP2/H2/StreamTable.hs --- a/Network/HTTP2/H2/StreamTable.hs +++ b/Network/HTTP2/H2/StreamTable.hs @@ -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 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 @@ -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 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 @@ -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 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 @@ -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 diff --git a/bench-hpack/Main.hs b/bench-hpack/Main.hs --- a/bench-hpack/Main.hs +++ b/bench-hpack/Main.hs @@ -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 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.4.3 +version: 5.4.4 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto @@ -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 diff --git a/test-hpack/HPACKDecode.hs b/test-hpack/HPACKDecode.hs --- a/test-hpack/HPACKDecode.hs +++ b/test-hpack/HPACKDecode.hs @@ -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 diff --git a/test/HTTP2/ServerSpec.hs b/test/HTTP2/ServerSpec.hs --- a/test/HTTP2/ServerSpec.hs +++ b/test/HTTP2/ServerSpec.hs @@ -118,8 +118,14 @@ Just "GET" -> case requestPath req of Just "/" -> sendResponse responseHello [] Just "/early" -> do - auxSendInformational aux earlyHints103 [("link", "; rel=preload; as=style")] - auxSendInformational aux earlyHints103 [("link", "; rel=preload; as=script")] + auxSendInformational + aux + earlyHints103 + [("link", "; rel=preload; as=style")] + auxSendInformational + aux + earlyHints103 + [("link", "; 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}