diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,33 @@ # 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 + [#170](https://github.com/kazu-yamamoto/http2/pull/170) + +## 5.4.2 + +* Support informational (1xx) responses, e.g. 103 Early Hints. Servers can send + them via `auxSendInformational`; clients can observe them via the new + `confOnInformational` callback in `Config`. + [#168](https://github.com/kazu-yamamoto/http2/pull/168) + +## 5.4.1 + +* Ensure sender notices when receiver has terminated. + [#167](https://github.com/kazu-yamamoto/http2/pull/167) + +## 5.4.0 + +* Providing `defaultConfig`. +* Except the item above, this version is identical to v5.3.11 which + includes breaking changes and is thus deprecated. + ## 5.3.11 * Implementing `auxSendPing` for client. 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 @@ -71,7 +71,18 @@ rstRateLimit, -- * Common configuration - Config (..), + Config, + defaultConfig, + confWriteBuffer, + confBufferSize, + confSendAll, + confReadN, + confPositionReadMaker, + confTimeoutManager, + confMySockAddr, + confPeerSockAddr, + confReadNTimeout, + confOnInformational, allocSimpleConfig, allocSimpleConfig', freeSimpleConfig, @@ -79,6 +90,7 @@ -- * Error HTTP2Error (..), + StreamTerminated (..), ReasonPhrase, ErrorCode ( ErrorCode, @@ -104,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/Internal.hs b/Network/HTTP2/Client/Internal.hs --- a/Network/HTTP2/Client/Internal.hs +++ b/Network/HTTP2/Client/Internal.hs @@ -1,6 +1,7 @@ module Network.HTTP2.Client.Internal ( Request (..), Response (..), + Config (..), ClientConfig (..), Settings (..), Aux (..), 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,8 +152,15 @@ er <- race runReceiver runClient case er of Right r -> return r - -- never reached because runReceiver throws an exception to exit. - Left () -> throwIO ConnectionIsClosed + 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 + -- but there are still some messages in the queue to be sent). + -- + -- If the client terminated successfully, we ignore any other errors in the + -- sender (indeed, any exception here might simply be that the background + -- threads were cancelled /because/ the client terminated). runAll = snd <$> concurrently runSender runClientReceiver makeStream @@ -236,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/Config.hs b/Network/HTTP2/H2/Config.hs --- a/Network/HTTP2/H2/Config.hs +++ b/Network/HTTP2/H2/Config.hs @@ -33,6 +33,7 @@ confMySockAddr <- getSocketName s confPeerSockAddr <- getPeerName s let confReadNTimeout = False + let confOnInformational = \_ _ -> return () return Config{..} -- | Deallocating the resource of the simple configuration. 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,8 +89,12 @@ , mySockAddr :: SockAddr , peerSockAddr :: SockAddr , threadManager :: T.ThreadManager - , receiverDone :: TVar Bool + , receiverDone :: TVar (Maybe E.SomeException) , workersDone :: STM Bool + , informationalCallback :: StreamId -> TokenHeaderTable -> IO () + -- ^ Client only: called when a 1xx informational response (e.g. 103 Early + -- Hints) is received, ahead of the final response. Copied from + -- 'confOnInformational'; no-op by default. } {- FOURMOLU_ENABLE -} @@ -138,7 +141,8 @@ let mySockAddr = confMySockAddr let peerSockAddr = confPeerSockAddr threadManager <- T.newThreadManager timmgr - receiverDone <- newTVarIO False + receiverDone <- newTVarIO Nothing + let informationalCallback = confOnInformational let workersDone = fromMaybe (T.isAllGone threadManager) mdone return Context{..} where @@ -189,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 @@ -237,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 @@ -19,8 +19,10 @@ import qualified Data.ByteString.Short as Short import qualified Data.ByteString.UTF8 as UTF8 import Data.IORef +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) @@ -45,14 +47,19 @@ ---------------------------------------------------------------- -frameReceiver :: Context -> Config -> IO () -frameReceiver ctx conf@Config{..} = - (switch `E.catch` handler) - `E.finally` atomically - (writeTVar (receiverDone ctx) True) +frameReceiver :: Context -> Config -> IO E.SomeException +frameReceiver ctx@Context{receiverDone} conf@Config{..} = + E.mask $ \unmask -> do + mErr <- E.try $ unmask switch + 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 where - handler ConnectionIsClosed = return () - handler e = E.throwIO e + switch :: IO Void switch = do labelMe "H2 receiver" tid <- myThreadId @@ -60,13 +67,16 @@ then loop1 else - void $ - T.withHandle (threadManager ctx) (E.throwTo tid ConnectionIsTimeout) loop2 + T.withHandle (threadManager ctx) (E.throwTo tid ConnectionIsTimeout) loop2 + + loop1 :: IO Void loop1 = do hd <- confReadN frameHeaderLength -- throwing an exception on timeout when (BS.null hd) $ E.throwIO ConnectionIsClosed processFrame ctx conf $ decodeFrameHeader hd loop1 + + loop2 :: T.Handle -> IO Void loop2 th = do -- If 'confReadN' is timeouted, 'ConnectionIsTimeout' is thrown -- to destroy the thread trees. @@ -379,6 +389,27 @@ where dep = streamDependency p +-- | Handle a decoded response HEADERS section. On the client, a 1xx +-- informational response (e.g. 103 Early Hints) is delivered to the +-- informational callback and the stream keeps waiting for the final response; +-- otherwise the headers become the (final) response. +onResponseHeaders + :: Context + -> StreamId + -> Maybe ClosedCode + -> Bool + -> TokenHeaderTable + -> IO StreamState +onResponseHeaders ctx streamId hcl endOfStream tbl + | endOfStream = return $ Open hcl (NoBody tbl) + | role ctx == Client && isInformational = do + informationalCallback ctx streamId tbl + return $ Open hcl JustOpened + | otherwise = return $ Open hcl (HasBody tbl) + where + isInformational = + maybe False ("1" `BS.isPrefixOf`) $ getFieldValue tokenStatus (snd tbl) + stream :: FrameType -> FrameHeader @@ -408,11 +439,7 @@ if endOfHeader then do tbl <- hpackDecodeHeader frag streamId ctx - return $ - if endOfStream - then -- turned into HalfClosedRemote in processState - Open hcl (NoBody tbl) - else Open hcl (HasBody tbl) + onResponseHeaders ctx streamId hcl endOfStream tbl else do let siz = BS.length frag return $ Open hcl $ Continued [frag] siz 1 endOfStream @@ -514,11 +541,7 @@ then do let hdrblk = BS.concat $ reverse rfrags' tbl <- hpackDecodeHeader hdrblk streamId ctx - return $ - if endOfStream - then -- turned into HalfClosedRemote in processState - Open hcl (NoBody tbl) - else Open hcl (HasBody tbl) + onResponseHeaders ctx streamId hcl endOfStream tbl else return $ Open hcl $ Continued rfrags' siz' n' endOfStream -- (No state transition) @@ -687,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 @@ -16,7 +16,6 @@ import Network.ByteOrder import Network.HTTP.Semantics.Client import Network.HTTP.Semantics.IO -import System.ThreadManager import Imports import Network.HPACK (setLimitForEncoding, toTokenHeaderTable) @@ -59,38 +58,43 @@ updateAllStreamTxFlow siz strms = forM_ strms $ \strm -> increaseStreamWindowSize strm siz -checkDone :: Context -> Int -> IO Bool +checkDone :: Context -> Int -> IO (Maybe E.SomeException) checkDone Context{..} 0 = atomically $ do isEmptyC <- isEmptyTQueue controlQ isEmptyO <- isEmptyTQueue outputQ if not isEmptyC || not isEmptyO then - return False + return Nothing else do - gone <- isAllGone threadManager - unless gone retry - done <- readTVar receiverDone - unless done retry - return True -checkDone _ _ = return False + recv <- readTVar receiverDone + case recv of + Just done -> + return $ Just done + _otherwise -> + retry +checkDone _ _ = return Nothing -frameSender :: Context -> Config -> IO () +frameSender :: Context -> Config -> IO E.SomeException frameSender ctx@Context{outputQ, controlQ, encodeDynamicTable, outputBufferLimit} Config{..} = do labelMe "H2 sender" - loop 0 + -- err is re-thrown by "runH2" + loop 0 `E.catch` return where ---------------------------------------------------------------- - loop :: Offset -> IO () + loop :: Offset -> IO E.SomeException loop off = do - done <- checkDone ctx off - unless done $ do - x <- atomically $ dequeue off - case x of - C ctl -> flushN off >> control ctl >> loop 0 - O out -> outputAndSync out off >>= flushIfNecessary >>= loop - Flush -> flushN off >> loop 0 + mDone <- checkDone ctx off + case mDone of + Just done -> + return done + Nothing -> do + x <- atomically $ dequeue off + case x of + C ctl -> flushN off >> control ctl >> loop 0 + O out -> outputAndSync out off >>= flushIfNecessary >>= loop + Flush -> flushN off >> loop 0 -- Flush the connection buffer to the socket, where the first 'n' bytes of -- the buffer are filled. @@ -158,12 +162,22 @@ 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 sync mout' return off' + OInformational hdr -> do + off' <- outputInformational strm hdr off + sync Nothing + return off' _ -> do sws <- getStreamWindowSize strm cws <- getConnectionWindowSize ctx -- not 0 @@ -172,6 +186,7 @@ sync mout' return off' + ---------------------------------------------------------------- resetStream :: Stream -> ErrorCode -> E.SomeException -> IO () resetStream strm err e | isAsyncException e = E.throwIO e @@ -180,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 @@ -208,6 +229,21 @@ return (off, Just out') ---------------------------------------------------------------- + -- Emit an informational (1xx) HEADERS section. Unlike 'outputHeader', + -- this never sets END_STREAM and never half-closes the stream, so the + -- final response can still be sent afterwards. + outputInformational + :: Stream + -> [Header] + -> Offset + -> IO Offset + outputInformational strm hdr off0 = do + let sid = streamNumber strm + (ths, _) <- toTokenHeaderTable $ fixHeaders hdr + off' <- headerContinue sid ths False {- not endOfStream -} off0 + flushIfNecessary off' + + ---------------------------------------------------------------- output :: Output -> Offset -> WindowSize -> IO (Offset, Maybe Output) output out@(Output strm (ONext curr tlrmkr) _) off0 lim = do -- Data frame payload @@ -217,7 +253,10 @@ datBufSiz = buflim - payloadOff curr datBuf (min datBufSiz lim) >>= \case Next datPayloadLen reqflush mnext -> do - NextTrailersMaker tlrmkr' <- runTrailersMaker tlrmkr datBuf datPayloadLen + tm <- runTrailersMaker tlrmkr datBuf datPayloadLen + let tlrmkr' = case tm of + NextTrailersMaker t -> t + _ -> defaultTrailersMaker fillDataHeader strm off0 @@ -237,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 @@ -312,7 +347,10 @@ reqflush = do let buf = confWriteBuffer `plusPtr` off (mtrailers, flag) <- do - Trailers trailers <- tlrmkr Nothing + tm <- tlrmkr Nothing + let trailers = case tm of + Trailers t -> t + _ -> [] if null trailers then return (Nothing, setEndStream defaultFlags) else return (Just trailers, defaultFlags) 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,18 +67,21 @@ {-# INLINE readStreamState #-} readStreamState :: Stream -> IO StreamState -readStreamState Stream{streamState} = readIORef streamState +readStreamState Stream{streamState} = readTVarIO streamState ---------------------------------------------------------------- closeAllStreams - :: TVar OddStreamTable -> TVar EvenStreamTable -> Maybe SomeException -> IO () -closeAllStreams ovar evar mErr' = do + :: TVar OddStreamTable -> TVar EvenStreamTable -> Maybe E.SomeException -> IO () +closeAllStreams ovar evar mErr = do ostrms <- clearOddStreamTable ovar mapM_ finalize ostrms estrms <- clearEvenStreamTable evar mapM_ finalize estrms where + -- We treat /every/ exception, including 'ConectionIsClosed', as abnormal + -- termination: we should only report a clean termination when we receive an + -- explicit @END_STREAM@ frame. finalize strm = do st <- readStreamState strm void $ tryPutMVar (streamInput strm) err @@ -92,75 +91,10 @@ _otherwise -> return () - mErr :: Maybe SomeException - mErr = case mErr' of - Just e - | Just ConnectionIsClosed <- fromException e -> - Nothing - _otherwise -> - mErr' - - 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,13 +7,9 @@ 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) import Network.Control import Network.HTTP.Semantics.Client import Network.HTTP.Semantics.IO @@ -113,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 @@ -123,7 +119,7 @@ = Finished | Killed | Reset ErrorCode - | ResetByMe SomeException + | ResetByMe E.SomeException deriving (Show) -- | Used for streams which are cancelled by calling @@ -136,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 ---------------------------------------------------------------- @@ -162,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) @@ -174,7 +170,7 @@ "Stream{id=" ++ show streamNumber ++ ",state=" - ++ show (unsafePerformIO (readIORef streamState)) + ++ show (unsafePerformIO (readTVarIO streamState)) ++ "}" ---------------------------------------------------------------- @@ -189,6 +185,8 @@ = OHeader [Header] (Maybe DynaNext) TrailersMaker | OPush TokenHeaderList StreamId -- associated stream id from client | ONext DynaNext TrailersMaker + | OInformational [Header] + | OReset (Maybe E.SomeException) data Sync = Done | Cont Output @@ -212,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) @@ -270,10 +269,33 @@ , confPeerSockAddr :: SockAddr -- ^ This is copied into 'Aux', if exist, on server. , confReadNTimeout :: Bool + , confOnInformational :: StreamId -> TokenHeaderTable -> IO () + -- ^ Client only: called when a 1xx informational response (e.g. 103 Early + -- Hints) is received on the given stream, ahead of the final response. + -- No-op by default. + -- + -- @since 5.4.2 } -isAsyncException :: Exception e => e -> Bool +-- | Default config. This is just a template to modify via +-- field names. Don't use this without modifications. +defaultConfig :: Config +defaultConfig = + Config + { confWriteBuffer = nullPtr + , confBufferSize = 0 + , confSendAll = \_ -> return () + , confReadN = \_ -> return "" + , confPositionReadMaker = defaultPositionReadMaker + , confTimeoutManager = T.defaultManager + , confMySockAddr = SockAddrInet 0 0 + , confPeerSockAddr = SockAddrInet 0 0 + , confReadNTimeout = False + , confOnInformational = \_ _ -> return () + } + +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.hs b/Network/HTTP2/Server.hs --- a/Network/HTTP2/Server.hs +++ b/Network/HTTP2/Server.hs @@ -51,7 +51,18 @@ rstRateLimit, -- * Common configuration - Config (..), + Config, + defaultConfig, + confWriteBuffer, + confBufferSize, + confSendAll, + confReadN, + confPositionReadMaker, + confTimeoutManager, + confMySockAddr, + confPeerSockAddr, + confReadNTimeout, + confOnInformational, allocSimpleConfig, allocSimpleConfig', freeSimpleConfig, diff --git a/Network/HTTP2/Server/Internal.hs b/Network/HTTP2/Server/Internal.hs --- a/Network/HTTP2/Server/Internal.hs +++ b/Network/HTTP2/Server/Internal.hs @@ -1,6 +1,8 @@ module Network.HTTP2.Server.Internal ( Request (..), Response (..), + Config (..), + ServerConfig (..), Aux (..), -- * Low level diff --git a/Network/HTTP2/Server/Run.hs b/Network/HTTP2/Server/Run.hs --- a/Network/HTTP2/Server/Run.hs +++ b/Network/HTTP2/Server/Run.hs @@ -3,9 +3,8 @@ module Network.HTTP2.Server.Run where -import Control.Concurrent.Async (concurrently_) +import Control.Concurrent.Async import Control.Concurrent.STM -import qualified Control.Exception as E import Imports import Network.Control (defaultMaxData) import Network.HTTP.Semantics.IO @@ -128,10 +127,8 @@ runReceiver = frameReceiver ctx conf runSender = frameSender ctx conf runBackgroundThreads = do - er <- E.try $ concurrently_ runReceiver runSender - case er of - Right () -> return () - Left e -> closureServer conf ctx e + e <- snd <$> concurrently runReceiver runSender + closureServer conf ctx e T.stopAfter mgr runBackgroundThreads $ \res -> closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) res 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 @@ -1,3 +1,4 @@ +{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} @@ -17,7 +18,12 @@ 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 +#endif + ---------------------------------------------------------------- runServer :: Config -> Server -> Launch @@ -29,6 +35,9 @@ { auxTimeHandle = th , auxMySockAddr = mySockAddr , auxPeerSockAddr = peerSockAddr +#if MIN_VERSION_http_semantics(0,4,1) + , auxSendInformational = sendInformational ctx strm +#endif } request = Request req' lc <- newLoopCheck strm Nothing @@ -48,6 +57,21 @@ ---------------------------------------------------------------- +#if MIN_VERSION_http_semantics(0,4,1) +-- | Send an informational (1xx) response, e.g. 103 Early Hints, on the given +-- stream ahead of the final response. This is wired into 'auxSendInformational' +-- so that a server (or WAI handler via Warp) can emit early hints. It blocks +-- until the informational HEADERS have been handed to the sender, preserving +-- ordering with respect to the final response. +sendInformational :: Context -> Stream -> Status -> ResponseHeaders -> IO () +sendInformational ctx strm st hdrs = do + lc <- newLoopCheck strm Nothing + let hdr = (":status", C8.pack (show (statusCode st))) : hdrs + syncWithSender ctx strm (OInformational hdr) lc +#endif + +---------------------------------------------------------------- + -- | This function is passed to workers. -- They also pass 'Response's from a server to this function. -- This function enqueues commands for the HTTP/2 sender. @@ -170,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.3.11 +version: 5.4.4 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto @@ -8,7 +8,7 @@ homepage: https://github.com/kazu-yamamoto/http2 synopsis: HTTP/2 library description: - HTTP/2 library including frames, priority queues, HPACK, client and server. + HTTP/2 library including frames, HPACK, client and server. category: Network build-type: Simple @@ -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 @@ -114,15 +115,15 @@ bytestring >=0.10, case-insensitive >=1.2 && <1.3, containers >=0.6, - http-semantics >= 0.3.1 && <0.4, + http-semantics >= 0.4 && <0.5, http-types >=0.12 && <0.13, iproute >= 1.7 && < 1.8, network >=3.1, network-byte-order >=0.1.7 && <0.2, network-control >=0.1 && <0.2, stm >=2.5 && <2.6, - time-manager >=0.2 && <0.4, - unix-time >=0.4.11 && <0.5, + time-manager >=0.3.0 && <0.4, + unix-time >=0.4.11 && <0.6, utf8-string >=1.0 && <1.1 executable h2c-client 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 @@ -49,6 +49,17 @@ threadDelay 10000 runClient allocSimpleConfig + it "delivers 103 Early Hints to the client's informational handler" $ + E.bracket (forkIO runServer) killThread $ \_ -> do + threadDelay 10000 + hintsRef <- newIORef [] + runClientEarly hintsRef >>= (`shouldBe` Just ok200) + hints <- readIORef hintsRef + map (getFieldValue (toToken "link") . snd) hints + `shouldBe` [ Just "; rel=preload; as=style" + , Just "; rel=preload; as=script" + ] + it "should always send the connection preface first" $ do prefaceVar <- newEmptyMVar E.bracket (forkIO (runFakeServer prefaceVar)) killThread $ \_ -> do @@ -103,9 +114,19 @@ threadDelay 10000 server :: Server -server req _aux sendResponse = case requestMethod req of +server req aux sendResponse = case requestMethod req of 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")] + sendResponse responseHello [] Just "/stream" -> sendResponse responseInfinite [] Just "/push" -> do let pp = pushPromise "/push-pp" responsePP 0 @@ -122,6 +143,9 @@ header = [("Content-Type", "text/plain")] body = byteString "Hello, world!\n" +earlyHints103 :: Status +earlyHints103 = mkStatus 103 "Early Hints" + responsePP :: Response responsePP = responseBuilder ok200 header body where @@ -172,6 +196,17 @@ trailersMaker ctx (Just bs) = return $ NextTrailersMaker $ trailersMaker ctx' where !ctx' = CH.hashUpdate ctx bs + +-- | Request @/early@ with an informational handler installed, recording each +-- 103 Early Hints section and returning the final response status. +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 -> + sendRequest (C.requestNoBody methodGet "/early" []) (return . C.responseStatus) + where + cliconf = C.defaultClientConfig{C.authority = host} + onInformational _sid tbl = modifyIORef' hintsRef (++ [tbl]) runClient :: (Socket -> BufferSize -> IO Config) -> IO () runClient allocConfig =