diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,33 @@ # ChangeLog for http2 +## 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. +* Server and client terminates their threads in the right order. +* Using `copy` in frame decoders to avoid potential fragmentation of + `ByteString`. +* Defining `confReadNTimeout` (default to `False`). If `confReadN` + implements timeout by itself, set it to `True`. +* TCP closing is now treaated as `ConnectionIsClosed` instead of + `ConnectionIsTimeout`. +* GOAWAY now contains a right last streamd ID. + +## 5.3.10 + +* Introducing closure. + [#157](https://github.com/kazu-yamamoto/http2/pull/157) + ## 5.3.9 * Using `ThreadManager` of `time-manager`. @@ -64,7 +92,7 @@ workers collaborate with the sender after queuing a response and finish after all response data are sent. * All threads are labeled with `labelThread`. You can see them by - `listThreas` if necessary. + `listThreads` if necessary. ## 5.2.6 @@ -178,7 +206,7 @@ [#80](https://github.com/kazu-yamamoto/http2/pull/80) * Introducing `KilledByHttp2ThreadManager` instead of `ThreadKilled`. [#79](https://github.com/kazu-yamamoto/http2/pull/79) - [#81](https://github.com/kazu-yamamoto/http2/pull/82) + [#81](https://github.com/kazu-yamamoto/http2/pull/81) [#82](https://github.com/kazu-yamamoto/http2/pull/82) * Handle RST_STREAM with NO_ERROR. [#78](https://github.com/kazu-yamamoto/http2/pull/78) 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 @@ -1,4 +1,3 @@ -{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} module Network.HPACK.HeaderBlock.Encode ( diff --git a/Network/HPACK/HeaderBlock/Integer.hs b/Network/HPACK/HeaderBlock/Integer.hs --- a/Network/HPACK/HeaderBlock/Integer.hs +++ b/Network/HPACK/HeaderBlock/Integer.hs @@ -1,5 +1,3 @@ -{-# LANGUAGE OverloadedStrings #-} - module Network.HPACK.HeaderBlock.Integer ( encodeI, encodeInteger, diff --git a/Network/HPACK/Huffman/Tree.hs b/Network/HPACK/Huffman/Tree.hs --- a/Network/HPACK/Huffman/Tree.hs +++ b/Network/HPACK/Huffman/Tree.hs @@ -72,9 +72,13 @@ (cnt2, r) = build cnt1 ts in (cnt2, Bin Nothing cnt0 l r) where - (fs', ts') = partition ((==) F . head . snd) xs - fs = map (second tail) fs' - ts = map (second tail) ts' + (fs', ts') = partition (isHeadF . snd) xs + fs = map (second (drop 1)) fs' + ts = map (second (drop 1)) ts' + +isHeadF :: Bits -> Bool +isHeadF [] = error "isHeadF" +isHeadF (b : _) = b == F -- | Marking the EOS path mark :: Int -> Bits -> HTree -> HTree 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 @@ -312,11 +312,12 @@ ---------------------------------------------------------------- -- | Inserting 'Entry' to 'DynamicTable'. --- New 'DynamicTable', the largest new 'Index' --- and a set of dropped OLD 'Index' --- are returned. insertEntry :: Entry -> DynamicTable -> IO () insertEntry e dyntbl@DynamicTable{..} = do + -- Theoretically speaking, dropping entries by adjustTableSize + -- should be first. However, non-used slots always exist since the + -- size of dynamic table calculated via the minimum entry size (32 + -- bytes). To simply adjustTableSize, insertFront is called first. insertFront e dyntbl es <- adjustTableSize dyntbl case codeInfo of @@ -359,6 +360,7 @@ ---------------------------------------------------------------- +-- Used in copyEntries. insertEnd :: Entry -> DynamicTable -> IO () insertEnd e DynamicTable{..} = do maxN <- readIORef maxNumOfEntries diff --git a/Network/HPACK/Types.hs b/Network/HPACK/Types.hs --- a/Network/HPACK/Types.hs +++ b/Network/HPACK/Types.hs @@ -1,5 +1,3 @@ -{-# LANGUAGE DeriveDataTypeable #-} - module Network.HPACK.Types ( -- * Header FieldValue, @@ -23,7 +21,6 @@ ) where import Control.Exception as E -import Data.Typeable import Network.ByteOrder (Buffer, BufferOverrun (..), BufferSize) import Imports @@ -88,6 +85,6 @@ | HeaderBlockTruncated | IllegalHeaderName | TooLargeHeader - deriving (Eq, Show, Typeable) + deriving (Eq, Show) instance Exception DecodeError diff --git a/Network/HTTP2/Client.hs b/Network/HTTP2/Client.hs --- a/Network/HTTP2/Client.hs +++ b/Network/HTTP2/Client.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} -- | HTTP\/2 client library. @@ -72,7 +71,17 @@ rstRateLimit, -- * Common configuration - Config (..), + Config, + defaultConfig, + confWriteBuffer, + confBufferSize, + confSendAll, + confReadN, + confPositionReadMaker, + confTimeoutManager, + confMySockAddr, + confPeerSockAddr, + confReadNTimeout, allocSimpleConfig, allocSimpleConfig', freeSimpleConfig, 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 @@ -1,4 +1,3 @@ -{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} @@ -81,8 +80,13 @@ n <- oddConc <$> readTVarIO (oddStreamTable ctx) return (x - n) aux ctx = - Aux + defaultAux { auxPossibleClientStreams = possibleClientStream ctx + , auxSendPing = + sendPing + ctx + False + "Haskell!" -- 8 bytes } clientCore ctx req processResponse = do (strm, moutobj) <- makeStream ctx scheme authority req @@ -93,19 +97,7 @@ x <- processResponse rsp adjustRxWindow ctx strm return x - runClient ctx = wrapClient ctx $ client (clientCore ctx) $ aux ctx - -wrapClient :: Context -> IO a -> IO a -wrapClient ctx client = do - x <- client - T.waitUntilAllGone $ threadManager ctx - let frame = goawayFrame 0 NoError "graceful closing" - enqueueControl (controlQ ctx) $ CFrames Nothing [frame] - enqueueControl (controlQ ctx) $ CFinish GoAwayIsSent - atomically $ do - done <- readTVar $ senderDone ctx - check done - return x + runClient ctx = client (clientCore ctx) $ aux ctx -- | Launching a receiver and a sender. runIO :: ClientConfig -> Config -> (ClientIO -> IO (IO a)) -> IO a @@ -120,9 +112,8 @@ return (streamNumber strm, strm) get = getResponse create = openOddStreamWait ctx - runClient <- do - act <- action $ ClientIO confMySockAddr confPeerSockAddr putR get putB create - return $ wrapClient ctx act + runClient <- + action $ ClientIO confMySockAddr confPeerSockAddr putR get putB create runH2 conf ctx runClient getResponse :: Stream -> IO Response @@ -143,34 +134,33 @@ connectionWindowSize settings confTimeoutManager + Nothing exchangeSettings ctx return ctx runH2 :: Config -> Context -> IO a -> IO a runH2 conf ctx runClient = do - T.stopAfter mgr runAll $ \res -> + T.stopAfter mgr (try runAll >>= closureClient conf ctx) $ \res -> closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) res where mgr = threadManager ctx runReceiver = frameReceiver ctx conf runSender = frameSender ctx conf - runBackgroundThreads = do - labelMe "H2 runBackgroundThreads" - concurrently_ runReceiver runSender + runClientReceiver = do + labelMe "H2 ClientReceiver" + er <- race runReceiver runClient + case er of + Right r -> return r + Left err -> throwIO err - -- Run the background threads and client concurrently. If the client - -- finishes first, cancel the background threads. If the background - -- threads finish first, wait for the client. - runAll = do - withAsync runBackgroundThreads $ \runningBackgroundThreads -> - withAsync runClient $ \runningClient -> do - result <- waitEither runningBackgroundThreads runningClient - case result of - Right clientResult -> do - cancel runningBackgroundThreads - return clientResult - Left () -> do - wait runningClient + -- 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 :: Context 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 @@ -1,4 +1,3 @@ -{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} @@ -215,11 +214,11 @@ -- | Frame payload decoder for DATA frame. decodeDataFrame :: FramePayloadDecoder -decodeDataFrame header bs = decodeWithPadding header bs DataFrame +decodeDataFrame header _bs = decodeWithPadding header _bs DataFrame -- | Frame payload decoder for HEADERS frame. decodeHeadersFrame :: FramePayloadDecoder -decodeHeadersFrame header bs = decodeWithPadding header bs $ \bs' -> +decodeHeadersFrame header _bs = decodeWithPadding header _bs $ \bs' -> if hasPriority then let (bs0, bs1) = BS.splitAt 5 bs' @@ -235,7 +234,7 @@ -- | Frame payload decoder for RST_STREAM frame. decodeRSTStreamFrame :: FramePayloadDecoder -decodeRSTStreamFrame _ bs = Right $ RSTStreamFrame $ toErrorCode (N.word32 bs) +decodeRSTStreamFrame _ bs = Right $ RSTStreamFrame $ toErrorCode $ N.word32 bs -- | Frame payload decoder for SETTINGS frame. decodeSettingsFrame :: FramePayloadDecoder @@ -259,19 +258,22 @@ -- | Frame payload decoder for PUSH_PROMISE frame. decodePushPromiseFrame :: FramePayloadDecoder -decodePushPromiseFrame header bs = decodeWithPadding header bs $ \bs' -> +decodePushPromiseFrame header _bs = decodeWithPadding header _bs $ \bs' -> let (bs0, bs1) = BS.splitAt 4 bs' sid = streamIdentifier (N.word32 bs0) in PushPromiseFrame sid bs1 -- | Frame payload decoder for PING frame. decodePingFrame :: FramePayloadDecoder -decodePingFrame _ bs = Right $ PingFrame bs +decodePingFrame _ _bs = Right $ PingFrame bs + where + bs = BS.copy _bs -- | Frame payload decoder for GOAWAY frame. decodeGoAwayFrame :: FramePayloadDecoder -decodeGoAwayFrame _ bs = Right $ GoAwayFrame sid ecid bs2 +decodeGoAwayFrame _ _bs = Right $ GoAwayFrame sid ecid bs2 where + bs = BS.copy _bs (bs0, bs1') = BS.splitAt 4 bs (bs1, bs2) = BS.splitAt 4 bs1' sid = streamIdentifier (N.word32 bs0) @@ -288,10 +290,14 @@ -- | Frame payload decoder for CONTINUATION frame. decodeContinuationFrame :: FramePayloadDecoder -decodeContinuationFrame _ bs = Right $ ContinuationFrame bs +decodeContinuationFrame _ _bs = Right $ ContinuationFrame bs + where + bs = BS.copy _bs decodeUnknownFrame :: FrameType -> FramePayloadDecoder -decodeUnknownFrame typ _ bs = Right $ UnknownFrame typ bs +decodeUnknownFrame typ _ _bs = Right $ UnknownFrame typ bs + where + bs = BS.copy _bs ---------------------------------------------------------------- @@ -312,14 +318,15 @@ -> Either FrameDecodeError FramePayload decodeWithPadding FrameHeader{..} bs body | padded = - let (w8, rest) = fromMaybe (error "decodeWithPadding") $ BS.uncons bs + let (w8, rest) = fromMaybe (error "decodeWithPadding") $ BS.uncons bs' padlen = intFromWord8 w8 bodylen = payloadLength - padlen - 1 in if bodylen < 0 then Left $ FrameDecodeError ProtocolError streamId "padding is not enough" else Right . body $ BS.take bodylen rest - | otherwise = Right $ body bs + | otherwise = Right $ body bs' where + bs' = BS.copy bs padded = testPadded flags streamIdentifier :: Word32 -> StreamId diff --git a/Network/HTTP2/Frame/Types.hs b/Network/HTTP2/Frame/Types.hs --- a/Network/HTTP2/Frame/Types.hs +++ b/Network/HTTP2/Frame/Types.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} @@ -541,6 +540,6 @@ type SettingsKeyId = SettingsKey type FrameTypeId = FrameType {- FOURMOLU_ENABLE -} -{- DEPRECATED ErrorCodeId "Use ErrorCode instead" -} -{- DEPRECATED SettingsKeyId "Use SettingsKey instead" -} -{- DEPRECATED FrameTypeId "Use FrameType instead" -} +{-# DEPRECATED ErrorCodeId "Use ErrorCode instead" #-} +{-# DEPRECATED SettingsKeyId "Use SettingsKey instead" #-} +{-# DEPRECATED FrameTypeId "Use FrameType instead" #-} 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 @@ -1,3 +1,5 @@ +{-# LANGUAGE RecordWildCards #-} + module Network.HTTP2.H2.Config where import Data.IORef @@ -22,23 +24,16 @@ -- timeout manager. allocSimpleConfig' :: Socket -> BufferSize -> Int -> IO Config allocSimpleConfig' s bufsiz usec = do - buf <- mallocBytes bufsiz - ref <- newIORef Nothing - timmgr <- T.initialize usec - mysa <- getSocketName s - peersa <- getPeerName s - let config = - Config - { confWriteBuffer = buf - , confBufferSize = bufsiz - , confSendAll = sendAll s - , confReadN = defaultReadN s ref - , confPositionReadMaker = defaultPositionReadMaker - , confTimeoutManager = timmgr - , confMySockAddr = mysa - , confPeerSockAddr = peersa - } - return config + confWriteBuffer <- mallocBytes bufsiz + let confBufferSize = bufsiz + let confSendAll = sendAll s + confReadN <- defaultReadN s <$> newIORef Nothing + let confPositionReadMaker = defaultPositionReadMaker + confTimeoutManager <- T.initialize usec + confMySockAddr <- getSocketName s + confPeerSockAddr <- getPeerName s + let confReadNTimeout = False + return Config{..} -- | Deallocating the resource of the simple configuration. freeSimpleConfig :: Config -> IO () diff --git a/Network/HTTP2/H2/Context.hs b/Network/HTTP2/H2/Context.hs --- a/Network/HTTP2/H2/Context.hs +++ b/Network/HTTP2/H2/Context.hs @@ -28,7 +28,7 @@ type Launch = Context -> Stream -> InpObj -> IO () -data ServerInfo = ServerInfo +newtype ServerInfo = ServerInfo { launch :: Launch } @@ -53,46 +53,51 @@ ---------------------------------------------------------------- +{- FOURMOLU_DISABLE -} -- | The context for HTTP/2 connection. data Context = Context - { role :: Role - , roleInfo :: RoleInfo + { role :: Role + , roleInfo :: RoleInfo , -- Settings - mySettings :: Settings - , myFirstSettings :: IORef Bool - , peerSettings :: IORef Settings - , oddStreamTable :: TVar OddStreamTable - , evenStreamTable :: TVar EvenStreamTable - , continued :: IORef (Maybe StreamId) + mySettings :: Settings + , myFirstSettings :: IORef Bool + , peerSettings :: IORef Settings + , oddStreamTable :: TVar OddStreamTable + , evenStreamTable :: TVar EvenStreamTable + , continued :: IORef (Maybe StreamId) -- ^ RFC 9113 says "Other frames (from any stream) MUST NOT -- occur between the HEADERS frame and any CONTINUATION -- frames that might follow". This field is used to implement -- this requirement. - , myStreamId :: TVar StreamId - , peerStreamId :: IORef StreamId - , outputBufferLimit :: IORef Int - , outputQ :: TQueue Output + , myStreamId :: TVar StreamId + , peerStreamId :: IORef StreamId + , peerLastStreamId :: IORef StreamId + , outputBufferLimit :: IORef Int + , outputQ :: TQueue Output -- ^ Invariant: Each stream will only ever have at most one 'Output' -- object in this queue at any moment. - , outputQStreamID :: TVar StreamId - , controlQ :: TQueue Control + , outputQStreamID :: TVar StreamId + , controlQ :: TQueue Control , encodeDynamicTable :: DynamicTable , decodeDynamicTable :: DynamicTable , -- the connection window for sending data - txFlow :: TVar TxFlow - , rxFlow :: IORef RxFlow - , pingRate :: Rate - , settingsRate :: Rate - , emptyFrameRate :: Rate - , rstRate :: Rate - , mySockAddr :: SockAddr - , peerSockAddr :: SockAddr - , threadManager :: T.ThreadManager - , senderDone :: TVar Bool + txFlow :: TVar TxFlow + , rxFlow :: IORef RxFlow + , pingRate :: Rate + , settingsRate :: Rate + , emptyFrameRate :: Rate + , rstRate :: Rate + , mySockAddr :: SockAddr + , peerSockAddr :: SockAddr + , threadManager :: T.ThreadManager + , receiverDone :: TVar (Maybe SomeException) + , workersDone :: STM Bool } +{- FOURMOLU_ENABLE -} ---------------------------------------------------------------- +{- FOURMOLU_DISABLE -} newContext :: RoleInfo -> Config @@ -100,48 +105,54 @@ -> Int -> Settings -> T.Manager + -> Maybe (STM Bool) -> IO Context -newContext rinfo Config{..} cacheSiz connRxWS settings timmgr = +newContext roleInfo Config{..} cacheSiz connRxWS mySettings timmgr mdone = do -- My: Use this even if ack has not been received yet. - Context rl rinfo settings - <$> newIORef False - -- Peer: The spec defines max concurrency is infinite unless - -- SETTINGS_MAX_CONCURRENT_STREAMS is exchanged. - -- But it is vulnerable, so we set the limitations. - <*> newIORef baseSettings{maxConcurrentStreams = Just defaultMaxStreams} - <*> newTVarIO emptyOddStreamTable - <*> newTVarIO (emptyEvenStreamTable cacheSiz) - <*> newIORef Nothing - <*> newTVarIO sid0 - <*> newIORef 0 - <*> newIORef buflim - <*> newTQueueIO - <*> newTVarIO sid0 - <*> newTQueueIO - -- My SETTINGS_HEADER_TABLE_SIZE - <*> newDynamicTableForEncoding defaultDynamicTableSize - <*> newDynamicTableForDecoding (headerTableSize settings) 4096 - <*> newTVarIO (newTxFlow defaultWindowSize) -- 64K - <*> newIORef (newRxFlow connRxWS) - <*> newRate - <*> newRate - <*> newRate - <*> newRate - <*> return confMySockAddr - <*> return confPeerSockAddr - <*> T.newThreadManager timmgr - <*> newTVarIO False + myFirstSettings <- newIORef False + -- Peer: The spec defines max concurrency is infinite unless + -- SETTINGS_MAX_CONCURRENT_STREAMS is exchanged. + -- But it is vulnerable, so we set the limitations. + peerSettings <- + newIORef baseSettings{maxConcurrentStreams = Just defaultMaxStreams} + oddStreamTable <- newTVarIO emptyOddStreamTable + evenStreamTable <- newTVarIO (emptyEvenStreamTable cacheSiz) + continued <- newIORef Nothing + myStreamId <- newTVarIO sid0 + peerStreamId <- newIORef 0 + peerLastStreamId <- newIORef 0 + outputBufferLimit <- newIORef buflim + outputQ <- newTQueueIO + outputQStreamID <- newTVarIO sid0 + controlQ <- newTQueueIO + -- My SETTINGS_HEADER_TABLE_SIZE + encodeDynamicTable <- newDynamicTableForEncoding defaultDynamicTableSize + decodeDynamicTable <- + newDynamicTableForDecoding (headerTableSize mySettings) 4096 + txFlow <- newTVarIO (newTxFlow defaultWindowSize) -- 64K + rxFlow <- newIORef (newRxFlow connRxWS) + pingRate <- newRate + settingsRate <- newRate + emptyFrameRate <- newRate + rstRate <- newRate + let mySockAddr = confMySockAddr + let peerSockAddr = confPeerSockAddr + threadManager <- T.newThreadManager timmgr + receiverDone <- newTVarIO Nothing + let workersDone = fromMaybe (T.isAllGone threadManager) mdone + return Context{..} where - rl = case rinfo of + role = case roleInfo of RIC{} -> Client _ -> Server sid0 - | rl == Client = 1 + | role == Client = 1 | otherwise = 2 dlim = defaultPayloadLength + frameHeaderLength buflim | confBufferSize >= dlim = dlim | otherwise = confBufferSize +{- FOURMOLU_ENABLE -} ---------------------------------------------------------------- @@ -165,6 +176,14 @@ setPeerStreamID :: Context -> StreamId -> IO () setPeerStreamID ctx sid = writeIORef (peerStreamId ctx) sid + +---------------------------------------------------------------- + +getPeerLastStreamId :: Context -> IO StreamId +getPeerLastStreamId ctx = readIORef $ peerLastStreamId ctx + +modifyPeerLastStreamId :: Context -> StreamId -> IO () +modifyPeerLastStreamId ctx sid = atomicModifyIORef' (peerLastStreamId ctx) $ \n -> if sid > n then (sid, ()) else (n, ()) ---------------------------------------------------------------- diff --git a/Network/HTTP2/H2/EncodeFrame.hs b/Network/HTTP2/H2/EncodeFrame.hs --- a/Network/HTTP2/H2/EncodeFrame.hs +++ b/Network/HTTP2/H2/EncodeFrame.hs @@ -23,10 +23,10 @@ where einfo = encodeInfo func 0 -pingFrame :: ByteString -> ByteString -pingFrame bs = encodeFrame einfo $ PingFrame bs +pingFrame :: Bool -> ByteString -> ByteString +pingFrame ack bs = encodeFrame einfo $ PingFrame bs where - einfo = encodeInfo setAck 0 + einfo = encodeInfo (if ack then setAck else id) 0 windowUpdateFrame :: StreamId -> WindowSize -> ByteString windowUpdateFrame sid winsiz = encodeFrame einfo $ WindowUpdateFrame winsiz diff --git a/Network/HTTP2/H2/Queue.hs b/Network/HTTP2/H2/Queue.hs --- a/Network/HTTP2/H2/Queue.hs +++ b/Network/HTTP2/H2/Queue.hs @@ -1,5 +1,3 @@ -{-# LANGUAGE RecordWildCards #-} - module Network.HTTP2.H2.Queue where import Control.Concurrent.STM 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 @@ -6,6 +6,9 @@ module Network.HTTP2.H2.Receiver ( frameReceiver, + closureClient, + closureServer, + sendPing, ) where import Control.Concurrent @@ -16,6 +19,7 @@ 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.ThreadManager as T @@ -42,50 +46,43 @@ ---------------------------------------------------------------- -frameReceiver :: Context -> Config -> IO () -frameReceiver ctx@Context{..} conf@Config{..} = do - labelMe "H2 receiver" - loop `E.catch` sendGoaway +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 + return err + Right x -> do + absurd x -- We only terminate due to exceptions where - loop = do - -- If 'confReadN' is timeouted, an exception is thrown + switch :: IO Void + switch = do + labelMe "H2 receiver" + tid <- myThreadId + if confReadNTimeout + then + loop1 + else + 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. hd <- confReadN frameHeaderLength - if BS.null hd - then enqueueControl controlQ $ CFinish ConnectionIsTimeout - else do - processFrame ctx conf $ decodeFrameHeader hd - loop - - sendGoaway se - | isAsyncException se = E.throwIO se - | Just GoAwayIsSent <- E.fromException se = do - T.waitUntilAllGone threadManager - enqueueControl controlQ $ CFinish GoAwayIsSent - | Just ConnectionIsClosed <- E.fromException se = do - T.waitUntilAllGone threadManager - enqueueControl controlQ $ CFinish ConnectionIsClosed - | Just e@(ConnectionErrorIsReceived _ _ _) <- E.fromException se = - enqueueControl controlQ $ CFinish e - | Just e@(ConnectionErrorIsSent err sid msg) <- E.fromException se = do - let frame = goawayFrame sid err $ Short.fromShort msg - enqueueControl controlQ $ CFrames Nothing [frame] - enqueueControl controlQ $ CFinish e - | Just e@(StreamErrorIsSent err sid msg) <- E.fromException se = do - let frame = resetFrame err sid - enqueueControl controlQ $ CFrames Nothing [frame] - let frame' = goawayFrame sid err $ Short.fromShort msg - enqueueControl controlQ $ CFrames Nothing [frame'] - enqueueControl controlQ $ CFinish e - | Just e@(StreamErrorIsReceived err sid) <- E.fromException se = do - let frame = goawayFrame sid err "treat a stream error as a connection error" - enqueueControl controlQ $ CFrames Nothing [frame] - enqueueControl controlQ $ CFinish e - -- this never happens - | Just e@(BadThingHappen _) <- E.fromException se = - enqueueControl controlQ $ CFinish e - | otherwise = - enqueueControl controlQ $ CFinish $ BadThingHappen se + T.tickle th + when (BS.null hd) $ E.throwIO ConnectionIsClosed + processFrame ctx conf $ decodeFrameHeader hd + loop2 th ---------------------------------------------------------------- @@ -278,7 +275,7 @@ let errmsg = Short.toShort ( "this frame is not allowed in an idle stream: " - `BS.append` (C8.pack (show ftyp)) + `BS.append` C8.pack (show ftyp) ) E.throwIO $ ConnectionErrorIsSent ProtocolError streamId errmsg when (ftyp == FrameHeaders) $ setPeerStreamID ctx streamId @@ -322,14 +319,12 @@ setframe = CFrames (Just peerAlist) (frames ++ [ack]) writeIORef myFirstSettings True enqueueControl controlQ setframe -control FramePing FrameHeader{flags, streamId} bs Context{mySettings, controlQ, pingRate} = +control FramePing FrameHeader{flags, streamId} bs ctx@Context{mySettings, pingRate} = unless (testAck flags) $ do rate <- getRate pingRate if rate > pingRateLimit mySettings then E.throwIO $ ConnectionErrorIsSent EnhanceYourCalm streamId "too many ping" - else do - let frame = pingFrame bs - enqueueControl controlQ $ CFrames Nothing [frame] + else sendPing ctx True bs control FrameGoAway header bs _ = do GoAwayFrame sid err msg <- guardIt $ decodeGoAwayFrame header bs if err == NoError @@ -650,3 +645,66 @@ let len = BS.length bs inform len return (bs, isEOF) + +---------------------------------------------------------------- + +closureClient :: Config -> Context -> Either E.SomeException a -> IO a +closureClient conf ctx (Right x) = do + frame <- goaway ctx NoError "no error" + sendGoaway conf frame + return x +closureClient conf ctx (Left se) = closureServer conf ctx se + +closureServer :: Config -> Context -> E.SomeException -> IO a +closureServer conf ctx se + | isAsyncException se = do + frame <- goaway ctx NoError "maybe timeout by manager" + sendGoaway conf frame + E.throwIO se + | Just ConnectionIsClosed <- E.fromException se = do + frame <- goaway ctx NoError "no error" + sendGoaway conf frame + E.throwIO ConnectionIsClosed + | Just ConnectionIsTimeout <- E.fromException se = do + frame <- goaway ctx NoError "timeout" + sendGoaway conf frame + E.throwIO ConnectionIsTimeout + | Just e@(ConnectionErrorIsReceived _err _sid msg) <- E.fromException se = do + frame <- goaway ctx NoError $ Short.fromShort msg + sendGoaway conf frame + E.throwIO e + | Just e@(ConnectionErrorIsSent err _sid msg) <- E.fromException se = do + frame <- goaway ctx err $ Short.fromShort msg + sendGoaway conf frame + E.throwIO e + | Just e@(StreamErrorIsSent err _sid msg) <- E.fromException se = do + let frame = resetFrame err _sid + frame' <- goaway ctx err $ Short.fromShort msg + sendGoaway conf (frame <> frame') + E.throwIO e + | Just e@(StreamErrorIsReceived err _sid) <- E.fromException se = do + frame <- goaway ctx err "treat a stream error as a connection error" + sendGoaway conf frame + E.throwIO e + | Just (_ :: HTTP2Error) <- E.fromException se = E.throwIO se + | otherwise = E.throwIO $ BadThingHappen se + +goaway :: Context -> ErrorCode -> ByteString -> IO ByteString +goaway ctx err msg = do + sid <- getPeerLastStreamId ctx + 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 () + +---------------------------------------------------------------- + +sendPing :: Context -> Bool -> ByteString -> IO () +sendPing Context{..} ack bs = enqueueControl controlQ $ CFrames Nothing [frame] + where + frame = pingFrame ack bs 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 @@ -1,3 +1,4 @@ +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} @@ -36,14 +37,6 @@ | O Output | Flush -wrapException :: E.SomeException -> IO () -wrapException se - | isAsyncException se = E.throwIO se - | Just GoAwayIsSent <- E.fromException se = return () - | Just ConnectionIsClosed <- E.fromException se = return () - | Just (e :: HTTP2Error) <- E.fromException se = E.throwIO e - | otherwise = E.throwIO $ BadThingHappen se - -- Peer SETTINGS_INITIAL_WINDOW_SIZE -- Adjusting initial window size for streams updatePeerSettings :: Context -> SettingsList -> IO () @@ -65,21 +58,42 @@ updateAllStreamTxFlow siz strms = forM_ strms $ \strm -> increaseStreamWindowSize strm siz -frameSender :: Context -> Config -> IO () +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 Nothing + else do + recv <- readTVar receiverDone + case recv of + Just done -> + return $ Just done + _otherwise -> + retry +checkDone _ _ = return Nothing + +frameSender :: Context -> Config -> IO E.SomeException frameSender - ctx@Context{outputQ, controlQ, encodeDynamicTable, outputBufferLimit, senderDone} + ctx@Context{outputQ, controlQ, encodeDynamicTable, outputBufferLimit} Config{..} = do labelMe "H2 sender" - (loop 0 `E.finally` setSenderDone) `E.catch` wrapException + loop 0 `E.catch` return where ---------------------------------------------------------------- - loop :: Offset -> IO () + loop :: Offset -> IO E.SomeException loop off = 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. @@ -115,7 +129,6 @@ -- called with off == 0 control :: Control -> IO () - control (CFinish e) = E.throwIO e control (CFrames ms xs) = do buf <- copyAll xs confWriteBuffer let off = buf `minusPtr` confWriteBuffer @@ -163,10 +176,12 @@ return off' resetStream :: Stream -> ErrorCode -> E.SomeException -> IO () - resetStream strm err e = do - closed ctx strm (ResetByMe e) - let rst = resetFrame err $ streamNumber strm - enqueueControl controlQ $ CFrames Nothing [rst] + resetStream strm err e + | isAsyncException e = E.throwIO e + | otherwise = do + closed ctx strm (ResetByMe e) + let rst = resetFrame err $ streamNumber strm + enqueueControl controlQ $ CFrames Nothing [rst] ---------------------------------------------------------------- outputHeader @@ -203,35 +218,37 @@ let payloadOff = off0 + frameHeaderLength datBuf = confWriteBuffer `plusPtr` payloadOff datBufSiz = buflim - payloadOff - curr datBuf (min datBufSiz lim) >>= \next -> - case next of - Next datPayloadLen reqflush mnext -> do - NextTrailersMaker tlrmkr' <- runTrailersMaker tlrmkr datBuf datPayloadLen - fillDataHeader - strm - off0 - datPayloadLen - mnext - tlrmkr' - out - reqflush - CancelNext mErr -> do - -- Stream cancelled - -- - -- At this point, the headers have already been sent. - -- Therefore, the stream cannot be in the 'Idle' state, so we - -- are justified in sending @RST_STREAM@. - -- - -- By the invariant on the 'outputQ', there are no other - -- outputs for this stream already enqueued. Therefore, we can - -- safely cancel it knowing that we won't try and send any - -- more data frames on this stream. - case mErr of - Just err -> - resetStream strm InternalError err - Nothing -> - resetStream strm Cancel (E.toException CancelledStream) - return (off0, Nothing) + curr datBuf (min datBufSiz lim) >>= \case + Next datPayloadLen reqflush mnext -> do + tm <- runTrailersMaker tlrmkr datBuf datPayloadLen + let tlrmkr' = case tm of + NextTrailersMaker t -> t + _ -> defaultTrailersMaker + fillDataHeader + strm + off0 + datPayloadLen + mnext + tlrmkr' + out + reqflush + CancelNext mErr -> do + -- Stream cancelled + -- + -- At this point, the headers have already been sent. + -- Therefore, the stream cannot be in the 'Idle' state, so we + -- are justified in sending @RST_STREAM@. + -- + -- By the invariant on the 'outputQ', there are no other + -- outputs for this stream already enqueued. Therefore, we can + -- safely cancel it knowing that we won't try and send any + -- more data frames on this stream. + case mErr of + Just err -> + resetStream strm InternalError err + Nothing -> + resetStream strm Cancel (E.toException CancelledStream) + return (off0, Nothing) output (Output strm (OPush ths pid) _) off0 _lim = do -- Creating a push promise header -- Frame id should be associated stream id from the client. @@ -260,7 +277,7 @@ where eos = if endOfStream then setEndStream else id getFlag [] = eos $ setEndHeader defaultFlags - getFlag _ = eos $ defaultFlags + getFlag _ = eos defaultFlags continue :: Offset -> TokenHeaderList -> FrameType -> IO Offset continue off [] _ = return off @@ -301,7 +318,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) @@ -388,5 +408,3 @@ , flags = flag , streamId = sid } - - setSenderDone = atomically $ writeTVar senderDone True diff --git a/Network/HTTP2/H2/Settings.hs b/Network/HTTP2/H2/Settings.hs --- a/Network/HTTP2/H2/Settings.hs +++ b/Network/HTTP2/H2/Settings.hs @@ -1,6 +1,3 @@ -{-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE RecordWildCards #-} - module Network.HTTP2.H2.Settings where import Network.Control 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 @@ -2,7 +2,6 @@ {-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE RankNTypes #-} -{-# LANGUAGE RecordWildCards #-} module Network.HTTP2.H2.Stream where @@ -78,12 +77,15 @@ closeAllStreams :: TVar OddStreamTable -> TVar EvenStreamTable -> Maybe SomeException -> IO () -closeAllStreams ovar evar mErr' = do +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 @@ -93,19 +95,8 @@ _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 = Left $ fromMaybe (toException ConnectionIsClosed) mErr ---------------------------------------------------------------- @@ -125,11 +116,7 @@ terminated <- newTVarIO Nothing let whenNotTerminated act = do mTerminated <- readTVar terminated - case mTerminated of - Just reason -> - throwSTM reason - Nothing -> - act + maybe act throwSTM mTerminated terminateWith reason act = do mTerminated <- readTVar terminated @@ -158,10 +145,11 @@ atomically $ whenNotTerminated $ writeTBQueue tbq StreamingFlush - , outBodyCancel = \mErr -> - atomically $ - terminateWith StreamCancelled $ - writeTBQueue tbq (StreamingCancelled mErr) + , outBodyCancel = + atomically + . terminateWith StreamCancelled + . writeTBQueue tbq + . StreamingCancelled } finished = atomically $ do terminateWith StreamOutOfScope $ 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 @@ -58,6 +58,7 @@ , -- Cache must contain Stream instead of StreamId because -- a Stream is deleted when end-of-stream is received. -- After that, cache is looked up. + -- LRUCache is not used as LRU but as fixed-size map. evenCache :: LRUCache (Method, ByteString) Stream } 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 @@ -14,7 +14,7 @@ ) import qualified Control.Exception as E import Data.IORef -import Data.Typeable +import Foreign.Ptr (nullPtr) import Network.Control import Network.HTTP.Semantics.Client import Network.HTTP.Semantics.IO @@ -195,9 +195,7 @@ ---------------------------------------------------------------- -data Control - = CFinish HTTP2Error - | CFrames (Maybe SettingsList) [ByteString] +data Control = CFrames (Maybe SettingsList) [ByteString] ---------------------------------------------------------------- @@ -217,8 +215,7 @@ | StreamErrorIsReceived ErrorCode StreamId | StreamErrorIsSent ErrorCode StreamId ReasonPhrase | BadThingHappen E.SomeException - | GoAwayIsSent - deriving (Show, Typeable) + deriving (Show) instance E.Exception HTTP2Error @@ -273,7 +270,24 @@ -- ^ This is copied into 'Aux', if exist, on server. , confPeerSockAddr :: SockAddr -- ^ This is copied into 'Aux', if exist, on server. + , confReadNTimeout :: 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 + } isAsyncException :: Exception e => e -> Bool isAsyncException e = diff --git a/Network/HTTP2/Server.hs b/Network/HTTP2/Server.hs --- a/Network/HTTP2/Server.hs +++ b/Network/HTTP2/Server.hs @@ -1,5 +1,3 @@ -{-# LANGUAGE OverloadedStrings #-} - -- | HTTP\/2 server library. -- -- Example: @@ -53,7 +51,17 @@ rstRateLimit, -- * Common configuration - Config (..), + Config, + defaultConfig, + confWriteBuffer, + confBufferSize, + confSendAll, + confReadN, + confPositionReadMaker, + confTimeoutManager, + confMySockAddr, + confPeerSockAddr, + confReadNTimeout, 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 @@ -1,10 +1,9 @@ -{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} module Network.HTTP2.Server.Run where -import Control.Concurrent.Async (concurrently_) +import Control.Concurrent.Async import Control.Concurrent.STM import Imports import Network.Control (defaultMaxData) @@ -51,7 +50,7 @@ ok <- checkPreface conf when ok $ do let lnch = runServer conf server - ctx <- setup sconf conf lnch + ctx <- setup sconf conf lnch Nothing runH2 conf ctx ---------------------------------------------------------------- @@ -63,6 +62,7 @@ , sioWriteResponse :: a -> Response -> IO () -- ^ 'Response' MUST be created with 'responseBuilder'. -- Others are not supported. + , sioDone :: IO () } -- | Launching a receiver and a sender without workers. @@ -77,7 +77,8 @@ when ok $ do inpQ <- newTQueueIO let lnch _ strm inpObj = atomically $ writeTQueue inpQ (strm, inpObj) - ctx <- setup sconf conf lnch + done <- newTVarIO False + ctx <- setup sconf conf lnch $ Just $ readTVar done let get = do (strm, inpObj) <- atomically $ readTQueue inpQ return (strm, Request inpObj) @@ -94,6 +95,7 @@ , sioPeerSockAddr = confPeerSockAddr , sioReadRequest = get , sioWriteResponse = putR + , sioDone = atomically $ writeTVar done True } io <- action serverIO concurrently_ io $ runH2 conf ctx @@ -107,8 +109,8 @@ return False else return True -setup :: ServerConfig -> Config -> Launch -> IO Context -setup ServerConfig{..} conf@Config{..} lnch = do +setup :: ServerConfig -> Config -> Launch -> Maybe (STM Bool) -> IO Context +setup ServerConfig{..} conf@Config{..} lnch mIsDone = do let serverInfo = newServerInfo lnch newContext serverInfo @@ -117,13 +119,16 @@ connectionWindowSize settings confTimeoutManager + mIsDone runH2 :: Config -> Context -> IO () runH2 conf ctx = do let mgr = threadManager ctx runReceiver = frameReceiver ctx conf runSender = frameSender ctx conf - runBackgroundThreads = concurrently_ runReceiver runSender + runBackgroundThreads = do + 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,5 +1,4 @@ {-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE PatternGuards #-} {-# LANGUAGE RecordWildCards #-} module Network.HTTP2.Server.Worker ( @@ -25,11 +24,17 @@ runServer conf server ctx@Context{..} strm req = T.forkManagedTimeout threadManager label $ \th -> do let req' = pauseRequestBody th - aux = Aux th mySockAddr peerSockAddr + aux = + defaultAux + { auxTimeHandle = th + , auxMySockAddr = mySockAddr + , auxPeerSockAddr = peerSockAddr + } request = Request req' lc <- newLoopCheck strm Nothing server request aux $ sendResponse conf ctx lc strm request adjustRxWindow ctx strm + modifyPeerLastStreamId ctx $ streamNumber strm where label = "H2 response sender for stream " ++ show (streamNumber strm) pauseRequestBody th = req{inpObjBody = readBody'} @@ -38,7 +43,7 @@ readBody' = do T.pause th bs <- readBody - T.resume th + T.resume th -- this is the same as 'tickle' return bs ---------------------------------------------------------------- @@ -174,11 +179,11 @@ { outBodyPush = \b -> do T.pause th outBodyPush iface b - T.resume th + T.resume th -- this is the same as 'tickle' , outBodyPushFinal = \b -> do T.pause th outBodyPushFinal iface b - T.resume th + T.resume th -- this is the same as 'tickle' } strmbdy iface' return tbq 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.9 +version: 5.4.1 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 @@ -114,15 +114,15 @@ bytestring >=0.10, case-insensitive >=1.2 && <1.3, containers >=0.6, - http-semantics >= 0.3 && <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.3, - unix-time >=0.4.11 && <0.5, + time-manager >=0.2.3 && <0.4, + unix-time >=0.4.11 && <0.6, utf8-string >=1.0 && <1.1 executable h2c-client @@ -138,7 +138,8 @@ bytestring, http-types, http2, - network-run >= 0.3 && <0.5, + network, + network-run >= 0.5 && <0.6, unix-time if flag(devel) @@ -307,7 +308,7 @@ http-types, http2, network, - network-run >=0.3.0, + network-run >= 0.5 && <0.6, random, typed-process @@ -326,7 +327,7 @@ hspec >=1.3, http-types, http2, - network-run >=0.3.0, + network-run >= 0.5 && <0.6, typed-process if flag(h2spec) diff --git a/test-frame/FrameSpec.hs b/test-frame/FrameSpec.hs --- a/test-frame/FrameSpec.hs +++ b/test-frame/FrameSpec.hs @@ -10,7 +10,7 @@ import qualified Data.ByteString.Base16 as B16 import qualified Data.ByteString.Lazy as BL import Network.HTTP2.Frame -import System.FilePath.Glob (compile, globDir) +import System.FilePath.Glob (compile, globDir1) import Test.Hspec import JSON @@ -19,7 +19,7 @@ testDir = "test-frame/http2-frame-test-case" getTestFiles :: FilePath -> IO [FilePath] -getTestFiles dir = head <$> globDir [compile "*/*.json"] dir +getTestFiles dir = globDir1 (compile "*/*.json") dir check :: FilePath -> IO () check file = do diff --git a/test-frame/frame-encode.hs b/test-frame/frame-encode.hs --- a/test-frame/frame-encode.hs +++ b/test-frame/frame-encode.hs @@ -1,5 +1,3 @@ -{-# LANGUAGE OverloadedStrings #-} - module Main where import Data.Aeson diff --git a/test-hpack/JSON.hs b/test-hpack/JSON.hs --- a/test-hpack/JSON.hs +++ b/test-hpack/JSON.hs @@ -100,7 +100,9 @@ toKey = toValue parseJSON (Object o) = pure (mk $ textToByteString $ Key.toText k, toValue v) -- new where - (k, v) = head $ H.toList o + (k, v) = case H.toList o of + [] -> error "parseJSON" + x : _ -> x parseJSON _ = mzero instance {-# OVERLAPPING #-} ToJSON Header where diff --git a/test/HTTP2/ClientSpec.hs b/test/HTTP2/ClientSpec.hs --- a/test/HTTP2/ClientSpec.hs +++ b/test/HTTP2/ClientSpec.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE BangPatterns #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} @@ -68,7 +67,7 @@ let maxConc = fromJust $ maxConcurrentStreams defaultSettings resultVars <- runClient "http" "localhost" $ \sendReq aux -> do - for [1 .. (maxConc + 1) :: Int] $ \_ -> do + replicateM ((maxConc + 1) :: Int) $ do resultVar <- newEmptyMVar concurrentClient resultVar sendReq aux pure resultVar @@ -111,7 +110,7 @@ body = byteString "Hello, world!\n" runClient :: Scheme -> Authority -> Client a -> IO a -runClient sc au client = runTCPClient host port $ runHTTP2Client +runClient sc au client = runTCPClient host port runHTTP2Client where cliconf = defaultClientConfig{scheme = sc, authority = au} runHTTP2Client s = diff --git a/test/HTTP2/ServerSpec.hs b/test/HTTP2/ServerSpec.hs --- a/test/HTTP2/ServerSpec.hs +++ b/test/HTTP2/ServerSpec.hs @@ -6,8 +6,6 @@ module HTTP2.ServerSpec (spec) where import Control.Concurrent --- cryptonite - import Control.Concurrent.Async import qualified Control.Exception as E import Control.Monad @@ -49,7 +47,7 @@ it "handles normal cases" $ E.bracket (forkIO runServer) killThread $ \_ -> do threadDelay 10000 - (runClient allocSimpleConfig) + runClient allocSimpleConfig it "should always send the connection preface first" $ do prefaceVar <- newEmptyMVar @@ -177,7 +175,7 @@ runClient :: (Socket -> BufferSize -> IO Config) -> IO () runClient allocConfig = - runTCPClient host port $ runHTTP2Client + runTCPClient host port runHTTP2Client where auth = host cliconf = C.defaultClientConfig{C.authority = auth} @@ -189,7 +187,8 @@ client :: C.Client () client sendRequest aux = - foldr1 concurrently_ $ + foldr1 + concurrently_ [ client0 sendRequest aux , client1 sendRequest aux , client2 sendRequest aux @@ -311,11 +310,13 @@ go (100 :: Int) firstTrailerValue :: TokenHeaderTable -> FieldValue -firstTrailerValue = snd . Prelude.head . fst +firstTrailerValue tbl = case fst tbl of + [] -> error "firstTrailerValue" + x : _ -> snd x runAttack :: (C.ClientIO -> IO ()) -> IO () runAttack attack = - runTCPClient host port $ runHTTP2Client + runTCPClient host port runHTTP2Client where auth = host cliconf = C.defaultClientConfig{C.authority = auth} diff --git a/test2/ServerSpec.hs b/test2/ServerSpec.hs --- a/test2/ServerSpec.hs +++ b/test2/ServerSpec.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE BangPatterns #-} {-# LANGUAGE OverloadedStrings #-} module ServerSpec (spec) where diff --git a/util/Client.hs b/util/Client.hs --- a/util/Client.hs +++ b/util/Client.hs @@ -4,13 +4,16 @@ module Client where +import Control.Concurrent import Control.Concurrent.Async import qualified Control.Exception as E import Control.Monad +import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as C8 import Data.UnixTime import Foreign.C.Types import Network.HTTP.Types +import System.IO import Text.Printf import Network.HTTP2.Client @@ -21,6 +24,7 @@ { optPerformance :: Int , optNumOfReqs :: Int , optMonitor :: Bool + , optInteractive :: Bool } deriving (Show) @@ -82,3 +86,32 @@ / fromIntegral millisecs / 1024 / 1024 + +console + :: Options -> [ByteString] -> IO () -> Aux -> IO () +console _opt paths cli aux = do + putStrLn "q -- quit" + putStrLn "g -- get" + putStrLn "p -- ping" + mvar <- newEmptyMVar + loop mvar `E.catch` \(E.SomeException _) -> return () + where + loop mvar = do + hSetBuffering stdout NoBuffering + putStr "> " + hSetBuffering stdout LineBuffering + l <- getLine + case l of + "q" -> putStrLn "bye" + "g" -> do + mapM_ (\p -> putStrLn $ "GET " ++ C8.unpack p) paths + _ <- forkIO $ cli >> putMVar mvar () + takeMVar mvar + loop mvar + "p" -> do + putStrLn "Ping" + auxSendPing aux + loop mvar + _ -> do + putStrLn "No such command" + loop mvar diff --git a/util/Monitor.hs b/util/Monitor.hs --- a/util/Monitor.hs +++ b/util/Monitor.hs @@ -16,7 +16,7 @@ showT (i, l, s) = i ++ " " ++ l ++ ": " ++ show s threadSummary :: IO [(String, String, ThreadStatus)] -threadSummary = (sort <$> listThreads) >>= mapM summary +threadSummary = listThreads >>= mapM summary . sort where summary t = do let idstr = drop 9 $ show t diff --git a/util/Server.hs b/util/Server.hs --- a/util/Server.hs +++ b/util/Server.hs @@ -3,7 +3,7 @@ module Server where import Control.Monad -import Crypto.Hash (Context, SHA1) -- cryptonite +import Crypto.Hash (Context, SHA1) -- crypton import qualified Crypto.Hash as CH import Data.ByteString (ByteString) import qualified Data.ByteString as B diff --git a/util/h2c-client.hs b/util/h2c-client.hs --- a/util/h2c-client.hs +++ b/util/h2c-client.hs @@ -1,5 +1,5 @@ {-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RankNTypes #-} module Main where @@ -9,6 +9,7 @@ import qualified Data.ByteString.Char8 as C8 import Network.HTTP2.Client import Network.Run.TCP (runTCPClient) +import Network.Socket import System.Console.GetOpt import System.Environment import System.Exit @@ -22,6 +23,7 @@ { optPerformance = 0 , optNumOfReqs = 1 , optMonitor = False + , optInteractive = False } usage :: String @@ -44,6 +46,11 @@ ["monitor"] (NoArg (\opts -> opts{optMonitor = True})) "run thread monitor" + , Option + ['i'] + ["interactive"] + (NoArg (\o -> o{optInteractive = True})) + "enter interactive mode" ] showUsageAndExit :: String -> IO a @@ -70,8 +77,19 @@ h : p : ps -> return (h, p, C8.pack <$> ps) when (optMonitor opts) $ void $ forkIO $ monitor $ threadDelay 1000000 let cliconf = defaultClientConfig{authority = host} - runTCPClient host port $ \s -> - E.bracket - (allocSimpleConfig' s 4096 5000000) - freeSimpleConfig - (\conf -> run cliconf conf $ client opts paths) + run' cliconf host port $ client' opts paths + +run' :: ClientConfig -> HostName -> ServiceName -> Client a -> IO a +run' cliconf host port f = runTCPClient host port $ \s -> + E.bracket + (allocSimpleConfig' s 4096 10000000) + freeSimpleConfig + (\conf -> run cliconf conf f) + +client' :: Options -> [Path] -> Client () +client' opts paths sendRequest _aux + | optInteractive opts = do + let action = client opts paths sendRequest _aux + console opts paths action _aux + return () + | otherwise = client opts paths sendRequest _aux diff --git a/util/h2c-server.hs b/util/h2c-server.hs --- a/util/h2c-server.hs +++ b/util/h2c-server.hs @@ -1,6 +1,4 @@ -{-# LANGUAGE BangPatterns #-} {-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE RecordWildCards #-} module Main (main) where @@ -37,7 +35,7 @@ (o, n, []) -> return (foldl (flip id) defaultOptions o, n) (_, _, errs) -> showUsageAndExit $ concat errs -data Options = Options +newtype Options = Options { optMonitor :: Bool } deriving (Show)