http2 5.2.6 → 5.3.0

raw patch · 24 files changed

+602/−528 lines, 24 filesdep ~http-semanticsdep ~network-rundep ~unliftioPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: http-semantics, network-run, unliftio

API changes (from Hackage documentation)

Files

• ChangeLog.md +12/−0
• Imports.hs +7/−0
• Network/HTTP2/Client.hs +6/−0
• Network/HTTP2/Client/Run.hs +86/−58
• Network/HTTP2/H2.hs +2/−0
• Network/HTTP2/H2/Context.hs +14/−5
• Network/HTTP2/H2/Manager.hs +31/−54
• Network/HTTP2/H2/Queue.hs +5/−10
• Network/HTTP2/H2/Receiver.hs +26/−34
• Network/HTTP2/H2/Sender.hs +75/−127
• Network/HTTP2/H2/Settings.hs +12/−3
• Network/HTTP2/H2/Sync.hs +78/−0
• Network/HTTP2/H2/Types.hs +6/−10
• Network/HTTP2/H2/Window.hs +1/−1
• Network/HTTP2/Server.hs +6/−0
• Network/HTTP2/Server/Run.hs +50/−34
• Network/HTTP2/Server/Worker.hs +127/−180
• http2.cabal +10/−8
• test/HTTP2/ClientSpec.hs +1/−1
• test2/ServerSpec.hs +1/−1
• util/Client.hs +7/−1
• util/Monitor.hs +30/−0
• util/h2c-client.hs +4/−0
• util/h2c-server.hs +5/−1

@@ -1,3 +1,15 @@+# ChangeLog for http2++## 5.3.0++* New server architecture: spawning worker on demand instead of the+  worker pool. This reduce huge numbers of threads for streaming into+  only 2. No API changes but workers do not terminate quicly. Rather+  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.+ ## 5.2.6  * Recover rxflow on closing.

@@ -20,6 +20,7 @@     GCBuffer,     withForeignPtr,     mallocPlainForeignPtrBytes,+    labelMe, ) where  import Control.Applicative@@ -38,9 +39,15 @@ import Data.String import Data.Word import Foreign.ForeignPtr+import GHC.Conc.Sync import GHC.ForeignPtr (mallocPlainForeignPtrBytes) import Network.HTTP.Semantics import Network.HTTP.Types import Numeric  type GCBuffer = ForeignPtr Word8++labelMe :: String -> IO ()+labelMe l = do+    tid <- myThreadId+    labelThread tid l

@@ -64,7 +64,13 @@     initialWindowSize,     maxFrameSize,     maxHeaderListSize,++    -- ** Rate limits     pingRateLimit,+    settingsRateLimit,+    emptyFrameRateLimit,+    rstRateLimit,+      -- * Common configuration     Config (..),

@@ -52,7 +52,7 @@ -- @userinfo\@@ as part of the authority. -- -- >>> defaultClientConfig--- ClientConfig {scheme = "http", authority = "localhost", cacheLimit = 64, connectionWindowSize = 1048576, settings = Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Nothing, pingRateLimit = 10}}+-- ClientConfig {scheme = "http", authority = "localhost", cacheLimit = 64, connectionWindowSize = 16777216, settings = Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Nothing, pingRateLimit = 10, emptyFrameRateLimit = 4, settingsRateLimit = 4, rstRateLimit = 4}} defaultClientConfig :: ClientConfig defaultClientConfig =     ClientConfig@@ -66,8 +66,8 @@ -- | Running HTTP/2 client. run :: ClientConfig -> Config -> Client a -> IO a run cconf@ClientConfig{..} conf client = do-    (ctx, mgr) <- setup cconf conf-    runH2 conf ctx mgr $ runClient ctx mgr+    ctx <- setup cconf conf+    runH2 conf ctx $ runClient ctx   where     serverMaxStreams ctx = do         mx <- maxConcurrentStreams <$> readIORef (peerSettings ctx)@@ -82,34 +82,40 @@         Aux             { auxPossibleClientStreams = possibleClientStream ctx             }-    clientCore ctx mgr req processResponse = do-        strm <- sendRequest ctx mgr scheme authority req+    clientCore ctx req processResponse = do+        strm <- sendRequest conf ctx scheme authority req         rsp <- getResponse strm         x <- processResponse rsp         adjustRxWindow ctx strm         return x-    runClient ctx mgr = do-        x <- client (clientCore ctx mgr) $ aux ctx-        waitCounter0 mgr-        let frame = goawayFrame 0 NoError "graceful closing"-        mvar <- newMVar ()-        enqueueControl (controlQ ctx) $ CGoaway frame mvar-        takeMVar mvar-        return x+    runClient ctx = wrapClinet ctx $ client (clientCore ctx) $ aux ctx +wrapClinet :: Context -> IO a -> IO a+wrapClinet ctx client = do+    x <- client+    waitCounter0 $ 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+ -- | Launching a receiver and a sender. runIO :: ClientConfig -> Config -> (ClientIO -> IO (IO a)) -> IO a runIO cconf@ClientConfig{..} conf@Config{..} action = do-    (ctx@Context{..}, mgr) <- setup cconf conf+    ctx@Context{..} <- setup cconf conf     let putB bs = enqueueControl controlQ $ CFrames Nothing [bs]         putR req = do-            strm <- sendRequest ctx mgr scheme authority req+            strm <- sendRequest conf ctx scheme authority req             return (streamNumber strm, strm)         get = getResponse         create = openOddStreamWait ctx-    runClient <--        action $ ClientIO confMySockAddr confPeerSockAddr putR get putB create-    runH2 conf ctx mgr runClient+    runClient <- do+        act <- action $ ClientIO confMySockAddr confPeerSockAddr putR get putB create+        return $ wrapClinet ctx act+    runH2 conf ctx runClient  getResponse :: Stream -> IO Response getResponse strm = do@@ -118,7 +124,7 @@         Left err -> throwIO err         Right rsp -> return $ Response rsp -setup :: ClientConfig -> Config -> IO (Context, Manager)+setup :: ClientConfig -> Config -> IO Context setup ClientConfig{..} conf@Config{..} = do     let clientInfo = newClientInfo scheme authority     ctx <-@@ -128,12 +134,12 @@             cacheLimit             connectionWindowSize             settings-    mgr <- start confTimeoutManager+            confTimeoutManager     exchangeSettings ctx-    return (ctx, mgr)+    return ctx -runH2 :: Config -> Context -> Manager -> IO a -> IO a-runH2 conf ctx mgr runClient =+runH2 :: Config -> Context -> IO a -> IO a+runH2 conf ctx runClient = do     stopAfter mgr (race runBackgroundThreads runClient) $ \res -> do         closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) $             either Just (const Nothing) res@@ -145,18 +151,21 @@             Right (Right x) ->                 return x   where+    mgr = threadManager ctx     runReceiver = frameReceiver ctx conf-    runSender = frameSender ctx conf mgr-    runBackgroundThreads = concurrently_ runReceiver runSender+    runSender = frameSender ctx conf+    runBackgroundThreads = do+        labelMe "H2 runBackgroundThreads"+        concurrently_ runReceiver runSender  sendRequest-    :: Context-    -> Manager+    :: Config+    -> Context     -> Scheme     -> Authority     -> Request     -> IO Stream-sendRequest ctx@Context{..} mgr scheme auth (Request req) = do+sendRequest conf ctx@Context{..} scheme auth (Request req) = do     -- Checking push promises     let hdr0 = outObjHeaders req         method = fromMaybe (error "sendRequest:method") $ lookup ":method" hdr0@@ -181,49 +190,68 @@                 req' = req{outObjHeaders = hdr2}             -- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: send: respecting peer's limit             (sid, newstrm) <- openOddStreamWait ctx-            case outObjBody req of-                OutBodyStreaming strmbdy ->-                    sendStreaming ctx mgr req' sid newstrm $ \iface ->-                        outBodyUnmask iface $ strmbdy (outBodyPush iface) (outBodyFlush iface)-                OutBodyStreamingUnmask strmbdy ->-                    sendStreaming ctx mgr req' sid newstrm strmbdy-                _ -> atomically $ do-                    sidOK <- readTVar outputQStreamID-                    check (sidOK == sid)-                    writeTVar outputQStreamID (sid + 2)-                    writeTQueue outputQ $ Output newstrm req' OObj Nothing (return ())+            sendHeaderBody conf ctx sid newstrm req'             return newstrm +sendHeaderBody :: Config -> Context -> StreamId -> Stream -> OutObj -> IO ()+sendHeaderBody Config{..} ctx@Context{..} sid newstrm OutObj{..} = do+    (mnext, mtbq) <- case outObjBody of+        OutBodyNone -> return (Nothing, Nothing)+        OutBodyFile (FileSpec path fileoff bytecount) -> do+            (pread, closerOrRefresher) <- confPositionReadMaker path+            refresh <- case closerOrRefresher of+                Closer closer -> timeoutClose threadManager closer+                Refresher refresher -> return refresher+            let next = fillFileBodyGetNext pread fileoff bytecount refresh+            return (Just next, Nothing)+        OutBodyBuilder builder -> do+            let next = fillBuilderBodyGetNext builder+            return (Just next, Nothing)+        OutBodyStreaming strmbdy -> do+            q <- sendStreaming ctx newstrm $ \iface ->+                outBodyUnmask iface $ strmbdy (outBodyPush iface) (outBodyFlush iface)+            let next = nextForStreaming q+            return (Just next, Just q)+        OutBodyStreamingIface strmbdy -> do+            q <- sendStreaming ctx newstrm strmbdy+            let next = nextForStreaming q+            return (Just next, Just q)+    (vs, out) <-+        prepareSync newstrm (OHeader outObjHeaders mnext outObjTrailers) mtbq+    atomically $ do+        sidOK <- readTVar outputQStreamID+        check (sidOK == sid)+        enqueueOutputSTM outputQ out+        writeTVar outputQStreamID (sid + 2)+    forkManaged threadManager "H2 worker" $ syncWithSender ctx newstrm vs+  where+    nextForStreaming+        :: TBQueue StreamingChunk+        -> DynaNext+    nextForStreaming tbq =+        let takeQ = atomically $ tryReadTBQueue tbq+            next = fillStreamBodyGetNext takeQ+         in next+ sendStreaming     :: Context-    -> Manager-    -> OutObj-    -> StreamId     -> Stream     -> (OutBodyIface -> IO ())-    -> IO ()-sendStreaming Context{..} mgr req sid newstrm strmbdy = do+    -> IO (TBQueue StreamingChunk)+sendStreaming Context{..} strm strmbdy = do     tbq <- newTBQueueIO 10 -- fixme: hard coding: 10-    forkManagedUnmask mgr $ \unmask -> do-        decrementedCounter <- newIORef False-        let decCounterOnce = do-                alreadyDecremented <- atomicModifyIORef decrementedCounter $ \b -> (True, b)-                unless alreadyDecremented $ decCounter mgr+    let label = "H2 streaming supporter for stream " ++ show (streamNumber strm)+    forkManagedUnmask threadManager label $ \unmask -> do         let iface =                 OutBodyIface                     { outBodyUnmask = unmask-                    , outBodyPush = \b -> atomically $ writeTBQueue tbq (StreamingBuilder b Nothing)-                    , outBodyPushFinal = \b -> atomically $ writeTBQueue tbq (StreamingBuilder b (Just decCounterOnce))+                    , outBodyPush = \b -> atomically $ writeTBQueue tbq $ StreamingBuilder b NotEndOfStream+                    , outBodyPushFinal = \b -> atomically $ writeTBQueue tbq $ StreamingBuilder b (EndOfStream Nothing)                     , outBodyFlush = atomically $ writeTBQueue tbq StreamingFlush                     }-            finished = atomically $ writeTBQueue tbq $ StreamingFinished decCounterOnce-        incCounter mgr+            finished = atomically $ writeTBQueue tbq $ StreamingFinished Nothing         strmbdy iface `finally` finished-    atomically $ do-        sidOK <- readTVar outputQStreamID-        check (sidOK == sid)-        writeTVar outputQStreamID (sid + 2)-        writeTQueue outputQ $ Output newstrm req OObj (Just tbq) (return ())+    return tbq  exchangeSettings :: Context -> IO () exchangeSettings Context{..} = do

@@ -10,6 +10,7 @@     module Network.HTTP2.H2.Settings,     module Network.HTTP2.H2.Stream,     module Network.HTTP2.H2.StreamTable,+    module Network.HTTP2.H2.Sync,     module Network.HTTP2.H2.Types,     module Network.HTTP2.H2.Window, ) where@@ -25,5 +26,6 @@ import Network.HTTP2.H2.Settings import Network.HTTP2.H2.Stream import Network.HTTP2.H2.StreamTable+import Network.HTTP2.H2.Sync import Network.HTTP2.H2.Types import Network.HTTP2.H2.Window

@@ -7,6 +7,7 @@ import Data.IORef import Network.Control import Network.Socket (SockAddr)+import qualified System.TimeManager as T import UnliftIO.Exception import qualified UnliftIO.Exception as E import UnliftIO.STM@@ -14,6 +15,7 @@ import Imports hiding (insert) import Network.HPACK import Network.HTTP2.Frame+import Network.HTTP2.H2.Manager import Network.HTTP2.H2.Settings import Network.HTTP2.H2.Stream import Network.HTTP2.H2.StreamTable@@ -25,8 +27,10 @@  data RoleInfo = RIS ServerInfo | RIC ClientInfo +type Launch = Context -> Stream -> InpObj -> IO ()+ data ServerInfo = ServerInfo-    { inputQ :: TQueue (Input Stream)+    { launch :: Launch     }  data ClientInfo = ClientInfo@@ -42,8 +46,8 @@ toClientInfo (RIC x) = x toClientInfo _ = error "toClientInfo" -newServerInfo :: IO RoleInfo-newServerInfo = RIS . ServerInfo <$> newTQueueIO+newServerInfo :: Launch -> RoleInfo+newServerInfo = RIS . ServerInfo  newClientInfo :: ByteString -> Authority -> RoleInfo newClientInfo scm auth = RIC $ ClientInfo scm auth@@ -68,7 +72,7 @@     , myStreamId :: TVar StreamId     , peerStreamId :: IORef StreamId     , outputBufferLimit :: IORef Int-    , outputQ :: TQueue (Output Stream)+    , outputQ :: TQueue Output     , outputQStreamID :: TVar StreamId     , controlQ :: TQueue Control     , encodeDynamicTable :: DynamicTable@@ -82,6 +86,8 @@     , rstRate :: Rate     , mySockAddr :: SockAddr     , peerSockAddr :: SockAddr+    , threadManager :: Manager+    , senderDone :: TVar Bool     }  ----------------------------------------------------------------@@ -92,8 +98,9 @@     -> Int     -> Int     -> Settings+    -> T.Manager     -> IO Context-newContext rinfo Config{..} cacheSiz connRxWS settings =+newContext rinfo Config{..} cacheSiz connRxWS settings timmgr =     -- My: Use this even if ack has not been received yet.     Context rl rinfo settings         <$> newIORef False@@ -121,6 +128,8 @@         <*> newRate         <*> return confMySockAddr         <*> return confPeerSockAddr+        <*> start timmgr+        <*> newTVarIO False   where     rl = case rinfo of         RIC{} -> Client

@@ -2,27 +2,20 @@ {-# LANGUAGE ScopedTypeVariables #-}  -- | A thread manager.---   The manager has responsibility to spawn and kill---   worker threads.+--   The manager has responsibility to kill worker threads. module Network.HTTP2.H2.Manager (     Manager,-    Action,     start,-    setAction,     stopAfter,-    spawnAction,     forkManaged,     forkManagedUnmask,     timeoutKillThread,     timeoutClose,     KilledByHttp2ThreadManager (..),-    incCounter,-    decCounter,     waitCounter0, ) where  import Data.Foldable-import Data.IORef import Data.Set (Set) import qualified Data.Set as Set import qualified System.TimeManager as T@@ -35,16 +28,10 @@  ---------------------------------------------------------------- --- | Action to be spawned by the manager.-type Action = IO ()--noAction :: Action-noAction = return ()--data Command = Stop (Maybe SomeException) | Spawn | Add ThreadId | Delete ThreadId+data Command = Stop (Maybe SomeException) | Add ThreadId | Delete ThreadId  -- | Manager to manage the thread and the timer.-data Manager = Manager (TQueue Command) (IORef Action) (TVar Int) T.Manager+data Manager = Manager (TQueue Command) (TVar Int) T.Manager  -- | Starting a thread manager. --   Its action is initially set to 'return ()' and should be set@@ -53,47 +40,31 @@ start :: T.Manager -> IO Manager start timmgr = do     q <- newTQueueIO-    ref <- newIORef noAction     cnt <- newTVarIO 0-    void $ forkIO $ go q Set.empty ref-    return $ Manager q ref cnt timmgr+    void $ forkIO $ do+        labelMe "H2 thread manager"+        go q Set.empty+    return $ Manager q cnt timmgr   where-    go q tset0 ref = do+    go q tset0 = do         x <- atomically $ readTQueue q         case x of             Stop err -> kill tset0 err-            Spawn -> next tset0             Add newtid ->                 let tset = add newtid tset0-                 in go q tset ref+                 in go q tset             Delete oldtid ->                 let tset = del oldtid tset0-                 in go q tset ref-      where-        next tset = do-            action <- readIORef ref-            newtid <- forkFinally action $ \_ -> do-                mytid <- myThreadId-                atomically $ writeTQueue q $ Delete mytid-            let tset' = add newtid tset-            go q tset' ref---- | Setting the action to be spawned.-setAction :: Manager -> Action -> IO ()-setAction (Manager _ ref _ _) action = writeIORef ref action+                 in go q tset  -- | Stopping the manager. stopAfter :: Manager -> IO a -> (Either SomeException a -> IO b) -> IO b-stopAfter (Manager q _ _ _) action cleanup = do+stopAfter (Manager q _ _) action cleanup = do     mask $ \unmask -> do         ma <- try $ unmask action         atomically $ writeTQueue q $ Stop (either Just (const Nothing) ma)         cleanup ma --- | Spawning the action.-spawnAction :: Manager -> IO ()-spawnAction (Manager q _ _ _) = atomically $ writeTQueue q Spawn- ----------------------------------------------------------------  -- | Fork managed thread@@ -101,25 +72,31 @@ -- This guarantees that the thread ID is added to the manager's queue before -- the thread starts, and is removed again when the thread terminates -- (normally or abnormally).-forkManaged :: Manager -> IO () -> IO ()-forkManaged mgr io =-    forkManagedUnmask mgr $ \unmask -> unmask io+forkManaged :: Manager -> String -> IO () -> IO ()+forkManaged mgr label io =+    forkManagedUnmask mgr label $ \unmask -> unmask io  -- | Like 'forkManaged', but run action with exceptions masked-forkManagedUnmask :: Manager -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()-forkManagedUnmask mgr io =-    void $ mask_ $ forkIOWithUnmask $ \unmask -> do+forkManagedUnmask+    :: Manager -> String -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()+forkManagedUnmask mgr label io =+    void $ mask_ $ forkIOWithUnmask $ \unmask -> E.handleSyncOrAsync handler $ do+        labelMe label         addMyId mgr+        incCounter mgr         -- We catch the exception and do not rethrow it: we don't want the         -- exception printed to stderr.         io unmask `catch` \(_e :: SomeException) -> return ()         deleteMyId mgr+        decCounter mgr+  where+    handler (E.SomeException _) = return ()  -- | Adding my thread id to the kill-thread list on stopping. -- -- This is not part of the public API; see 'forkManaged' instead. addMyId :: Manager -> IO ()-addMyId (Manager q _ _ _) = do+addMyId (Manager q _ _) = do     tid <- myThreadId     atomically $ writeTQueue q $ Add tid @@ -129,7 +106,7 @@ -- the manager /before/ the thread terminates (thereby assuming responsibility -- for thread cleanup yourself). deleteMyId :: Manager -> IO ()-deleteMyId (Manager q _ _ _) = do+deleteMyId (Manager q _ _) = do     tid <- myThreadId     atomically $ writeTQueue q $ Delete tid @@ -150,14 +127,14 @@  -- | Killing the IO action of the second argument on timeout. timeoutKillThread :: Manager -> (T.Handle -> IO a) -> IO a-timeoutKillThread (Manager _ _ _ tmgr) action = E.bracket register T.cancel action+timeoutKillThread (Manager _ _ tmgr) action = E.bracket register T.cancel action   where-    register = T.registerKillThread tmgr noAction+    register = T.registerKillThread tmgr (return ())  -- | Registering closer for a resource and --   returning a timer refresher. timeoutClose :: Manager -> IO () -> IO (IO ())-timeoutClose (Manager _ _ _ tmgr) closer = do+timeoutClose (Manager _ _ tmgr) closer = do     th <- T.register tmgr closer     return $ T.tickle th @@ -171,12 +148,12 @@ ----------------------------------------------------------------  incCounter :: Manager -> IO ()-incCounter (Manager _ _ cnt _) = atomically $ modifyTVar' cnt (+ 1)+incCounter (Manager _ cnt _) = atomically $ modifyTVar' cnt (+ 1)  decCounter :: Manager -> IO ()-decCounter (Manager _ _ cnt _) = atomically $ modifyTVar' cnt (subtract 1)+decCounter (Manager _ cnt _) = atomically $ modifyTVar' cnt (subtract 1)  waitCounter0 :: Manager -> IO ()-waitCounter0 (Manager _ _ cnt _) = atomically $ do+waitCounter0 (Manager _ cnt _) = atomically $ do     n <- readTVar cnt     checkSTM (n < 1)

@@ -4,20 +4,15 @@  import UnliftIO.STM -import Network.HTTP2.H2.Manager import Network.HTTP2.H2.Types -{-# INLINE forkAndEnqueueWhenReady #-}-forkAndEnqueueWhenReady-    :: IO () -> TQueue (Output Stream) -> Output Stream -> Manager -> IO ()-forkAndEnqueueWhenReady wait outQ out mgr =-    forkManaged mgr $ do-        wait-        enqueueOutput outQ out- {-# INLINE enqueueOutput #-}-enqueueOutput :: TQueue (Output Stream) -> Output Stream -> IO ()+enqueueOutput :: TQueue Output -> Output -> IO () enqueueOutput outQ out = atomically $ writeTQueue outQ out++{-# INLINE enqueueOutputSTM #-}+enqueueOutputSTM :: TQueue Output -> Output -> STM ()+enqueueOutputSTM outQ out = writeTQueue outQ out  {-# INLINE enqueueControl #-} enqueueControl :: TQueue Control -> Control -> IO ()

@@ -23,6 +23,7 @@ import Network.HTTP2.H2.Context import Network.HTTP2.H2.EncodeFrame import Network.HTTP2.H2.HPACK+import Network.HTTP2.H2.Manager import Network.HTTP2.H2.Queue import Network.HTTP2.H2.Settings import Network.HTTP2.H2.Stream@@ -38,36 +39,27 @@ headerFragmentLimit :: Int headerFragmentLimit = 51200 -- 50K -settingsRateLimit :: Int-settingsRateLimit = 4--emptyFrameRateLimit :: Int-emptyFrameRateLimit = 4--rstRateLimit :: Int-rstRateLimit = 4- ----------------------------------------------------------------  frameReceiver :: Context -> Config -> IO ()-frameReceiver ctx@Context{..} conf@Config{..} = loop 0 `E.catch` sendGoaway+frameReceiver ctx@Context{..} conf@Config{..} = do+    labelMe "H2 receiver"+    loop `E.catch` sendGoaway   where-    loop :: Int -> IO ()-    loop n-        | n == 6 = do-            yield-            loop 0-        | otherwise = do-            hd <- confReadN frameHeaderLength-            if BS.null hd-                then enqueueControl controlQ $ CFinish ConnectionIsClosed-                else do-                    processFrame ctx conf $ decodeFrameHeader hd-                    loop (n + 1)+    loop = do+        -- If 'confReadN' is timeouted, an exception 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-        | Just e@ConnectionIsClosed <- E.fromException se =-            enqueueControl controlQ $ CFinish e+        | Just ConnectionIsClosed <- E.fromException se = do+            waitCounter0 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@@ -183,8 +175,8 @@     let inpObj = InpObj tbl (Just 0) (return (mempty, True)) tlr     if isServer ctx         then do-            let si = toServerInfo roleInfo-            atomically $ writeTQueue (inputQ si) $ Input strm inpObj+            let ServerInfo{..} = toServerInfo roleInfo+            launch ctx strm inpObj         else putMVar streamInput $ Right inpObj     halfClosedRemote ctx strm     return False@@ -203,8 +195,8 @@     let inpObj = InpObj tbl mcl (readSource bodySource) tlr     if isServer ctx         then do-            let si = toServerInfo roleInfo-            atomically $ writeTQueue (inputQ si) $ Input strm inpObj+            let ServerInfo{..} = toServerInfo roleInfo+            launch ctx strm inpObj         else putMVar streamInput $ Right inpObj     return False @@ -309,7 +301,7 @@         else do             -- Settings Flood - CVE-2019-9515             rate <- getRate settingsRate-            when (rate > settingsRateLimit) $+            when (rate > settingsRateLimit mySettings) $                 E.throwIO $                     ConnectionErrorIsSent EnhanceYourCalm streamId "too many settings"             let ack = settingsFrame setAck []@@ -412,7 +404,7 @@         then do             -- Empty Frame Flooding - CVE-2019-9518             rate <- getRate $ emptyFrameRate ctx-            if rate > emptyFrameRateLimit+            if rate > emptyFrameRateLimit (mySettings ctx)                 then                     E.throwIO $                         ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty headers"@@ -456,7 +448,7 @@     FrameData     header@FrameHeader{flags, payloadLength, streamId}     bs-    Context{emptyFrameRate, rxFlow}+    Context{emptyFrameRate, rxFlow, mySettings}     s@(Open _ (Body q mcl bodyLength _))     Stream{..} = do         DataFrame body <- guardIt $ decodeDataFrame header bs@@ -483,7 +475,7 @@         if body == ""             then unless endOfStream $ do                 rate <- getRate emptyFrameRate-                when (rate > emptyFrameRateLimit) $ do+                when (rate > emptyFrameRateLimit mySettings) $ do                     E.throwIO $ ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty data"             else do                 writeIORef bodyLength len@@ -511,7 +503,7 @@         then do             -- Empty Frame Flooding - CVE-2019-9518             rate <- getRate $ emptyFrameRate ctx-            if rate > emptyFrameRateLimit+            if rate > emptyFrameRateLimit (mySettings ctx)                 then                     E.throwIO $                         ConnectionErrorIsSent EnhanceYourCalm streamId "too many empty continuation"@@ -547,7 +539,7 @@ stream FrameRSTStream header@FrameHeader{streamId} bs ctx s strm = do     -- Rapid Rest: CVE-2023-44487     rate <- getRate $ rstRate ctx-    when (rate > rstRateLimit) $+    when (rate > rstRateLimit (mySettings ctx)) $         E.throwIO $             ConnectionErrorIsSent EnhanceYourCalm streamId "too many rst_stream"     RSTStreamFrame err <- guardIt $ decodeRSTStreamFrame header bs

@@ -7,7 +7,6 @@     frameSender, ) where -import Control.Concurrent.MVar (putMVar) import Data.IORef (modifyIORef', readIORef, writeIORef) import Data.IntMap.Strict (IntMap) import Foreign.Ptr (minusPtr, plusPtr)@@ -23,7 +22,6 @@ import Network.HTTP2.H2.Context import Network.HTTP2.H2.EncodeFrame import Network.HTTP2.H2.HPACK-import Network.HTTP2.H2.Manager hiding (start) import Network.HTTP2.H2.Queue import Network.HTTP2.H2.Settings import Network.HTTP2.H2.Stream@@ -33,19 +31,15 @@  ---------------------------------------------------------------- -{-# INLINE waitStreaming #-}-waitStreaming :: TBQueue a -> IO ()-waitStreaming tbq = atomically $ do-    isEmpty <- isEmptyTBQueue tbq-    checkSTM (not isEmpty)- data Switch     = C Control-    | O (Output Stream)+    | O Output     | Flush  wrapException :: E.SomeException -> IO () wrapException 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 @@ -70,11 +64,12 @@     updateAllStreamTxFlow siz strms =         forM_ strms $ \strm -> increaseStreamWindowSize strm siz -frameSender :: Context -> Config -> Manager -> IO ()+frameSender :: Context -> Config -> IO () frameSender-    ctx@Context{outputQ, controlQ, encodeDynamicTable, outputBufferLimit}-    Config{..}-    mgr = loop 0 `E.catch` wrapException+    ctx@Context{outputQ, controlQ, encodeDynamicTable, outputBufferLimit, senderDone}+    Config{..} = do+        labelMe "H2 sender"+        (loop 0 `E.finally` setSenderDone) `E.catch` wrapException       where         ----------------------------------------------------------------         loop :: Offset -> IO ()@@ -120,12 +115,6 @@         -- called with off == 0         control :: Control -> IO ()         control (CFinish e) = E.throwIO e-        control (CGoaway bs mvar) = do-            buf <- copyAll [bs] confWriteBuffer-            let off = buf `minusPtr` confWriteBuffer-            flushN off-            putMVar mvar ()-            E.throwIO GoAwayIsSent         control (CFrames ms xs) = do             buf <- copyAll xs confWriteBuffer             let off = buf `minusPtr` confWriteBuffer@@ -150,8 +139,62 @@                         Just siz -> setLimitForEncoding siz encodeDynamicTable          -----------------------------------------------------------------        output :: Output Stream -> Offset -> WindowSize -> IO Offset-        output out@(Output strm OutObj{} (ONext curr tlrmkr) _ sentinel) off0 lim = do+        outputOrEnqueueAgain :: Output -> Offset -> IO Offset+        outputOrEnqueueAgain out@(Output strm otyp sync) off = E.handle resetStream $ do+            state <- readStreamState strm+            if isHalfClosedLocal state+                then return off+                else case otyp of+                    OHeader hdr mnext tlrmkr ->+                        -- Send headers immediately, without waiting for data+                        -- No need to check the streaming window (applies to DATA frames only)+                        outputHeader strm hdr mnext tlrmkr sync off+                    _ -> do+                        ok <- sync $ Just otyp+                        if ok+                            then do+                                sws <- getStreamWindowSize strm+                                cws <- getConnectionWindowSize ctx -- not 0+                                let lim = min cws sws+                                output out off lim+                            else return off+          where+            resetStream e = do+                closed ctx strm (ResetByMe e)+                let rst = resetFrame InternalError $ streamNumber strm+                enqueueControl controlQ $ CFrames Nothing [rst]+                return off++        ----------------------------------------------------------------+        outputHeader+            :: Stream+            -> [Header]+            -> Maybe DynaNext+            -> TrailersMaker+            -> (Maybe OutputType -> IO Bool)+            -> Offset+            -> IO Offset+        outputHeader strm hdr mnext tlrmkr sync off0 = do+            -- Header frame and Continuation frame+            let sid = streamNumber strm+                endOfStream = isNothing mnext+            (ths, _) <- toTokenHeaderTable $ fixHeaders hdr+            off' <- headerContinue sid ths endOfStream off0+            -- halfClosedLocal calls closed which removes+            -- the stream from stream table.+            when endOfStream $ do+                halfClosedLocal ctx strm Finished+                void $ sync Nothing+            off <- flushIfNecessary off'+            case mnext of+                Nothing -> return off+                Just next -> do+                    let out' = Output strm (ONext next tlrmkr) sync+                    outputOrEnqueueAgain out' off++        ----------------------------------------------------------------+        output :: Output -> Offset -> WindowSize -> IO Offset+        output out@(Output strm (ONext curr tlrmkr) sync) off0 lim = do             -- Data frame payload             buflim <- readIORef outputBufferLimit             let payloadOff = off0 + frameHeaderLength@@ -165,115 +208,18 @@                 datPayloadLen                 mnext                 tlrmkr'-                sentinel+                sync                 out                 reqflush-        output (Output strm obj OObj mtbq sentinel) off0 _lim = do-            outputObj strm obj mtbq sentinel off0-        output out@(Output strm _ (OPush ths pid) _ _) off0 lim = do+        output (Output strm (OPush ths pid) sync) off0 _lim = do             -- Creating a push promise header             -- Frame id should be associated stream id from the client.             let sid = streamNumber strm             len <- pushPromise pid sid ths off0             off <- flushIfNecessary $ off0 + frameHeaderLength + len-            output out{outputType = OObj} off lim-        output _ _ _ = undefined -- never reach--        -----------------------------------------------------------------        outputObj-            :: Stream-            -> OutObj-            -> Maybe (TBQueue StreamingChunk)-            -> IO ()-            -> Offset-            -> IO Offset-        outputObj strm obj@(OutObj hdr body tlrmkr) mtbq sentinel off0 = do-            -- Header frame and Continuation frame-            let sid = streamNumber strm-                endOfStream = case body of-                    OutBodyNone -> True-                    _ -> False-            (ths, _) <- toTokenHeaderTable $ fixHeaders hdr-            off' <- headerContinue sid ths endOfStream off0-            -- halfClosedLocal calls closed which removes-            -- the stream from stream table.-            when endOfStream $ halfClosedLocal ctx strm Finished-            off <- flushIfNecessary off'-            let setOutputType otyp = Output strm obj otyp mtbq sentinel-            case body of-                OutBodyNone -> return off-                OutBodyFile (FileSpec path fileoff bytecount) -> do-                    (pread, sentinel') <- confPositionReadMaker path-                    refresh <- case sentinel' of-                        Closer closer -> timeoutClose mgr closer-                        Refresher refresher -> return refresher-                    let next = fillFileBodyGetNext pread fileoff bytecount refresh-                        out' = setOutputType $ ONext next tlrmkr-                    outputOrEnqueueAgain out' off-                OutBodyBuilder builder -> do-                    let next = fillBuilderBodyGetNext builder-                        out' = setOutputType $ ONext next tlrmkr-                    outputOrEnqueueAgain out' off-                OutBodyStreaming _ -> do-                    let out' = setOutputType $ nextForStreaming mtbq tlrmkr-                    outputOrEnqueueAgain out' off-                OutBodyStreamingUnmask _ -> do-                    let out' = setOutputType $ nextForStreaming mtbq tlrmkr-                    outputOrEnqueueAgain out' off--        -----------------------------------------------------------------        nextForStreaming-            :: Maybe (TBQueue StreamingChunk)-            -> TrailersMaker-            -> OutputType-        nextForStreaming mtbq tlrmkr =-            let tbq = fromJust mtbq-                takeQ = atomically $ tryReadTBQueue tbq-                next = fillStreamBodyGetNext takeQ-             in ONext next tlrmkr--        -----------------------------------------------------------------        outputOrEnqueueAgain :: Output Stream -> Offset -> IO Offset-        outputOrEnqueueAgain out@(Output strm obj otyp mtbq sentinel) off = E.handle resetStream $ do-            state <- readStreamState strm-            if isHalfClosedLocal state-                then return off-                else case otyp of-                    OWait wait -> do-                        -- Checking if all push are done.-                        forkAndEnqueueWhenReady wait outputQ out{outputType = OObj} mgr-                        return off-                    OObj ->-                        -- Send headers immediately, without waiting for data-                        -- No need to check the streaming window (applies to DATA frames only)-                        outputObj strm obj mtbq sentinel off-                    _ -> case mtbq of-                        Just tbq -> checkStreaming tbq-                        _ -> checkStreamWindowSize-          where-            checkStreaming tbq = do-                isEmpty <- atomically $ isEmptyTBQueue tbq-                if isEmpty-                    then do-                        forkAndEnqueueWhenReady (waitStreaming tbq) outputQ out mgr-                        return off-                    else checkStreamWindowSize-            -- FLOW CONTROL: WINDOW_UPDATE: send: respecting peer's limit-            checkStreamWindowSize = do-                sws <- getStreamWindowSize strm-                if sws == 0-                    then do-                        forkAndEnqueueWhenReady (waitStreamWindowSize strm) outputQ out mgr-                        return off-                    else do-                        cws <- getConnectionWindowSize ctx -- not 0-                        let lim = min cws sws-                        output out off lim-            resetStream e = do-                closed ctx strm (ResetByMe e)-                let rst = resetFrame InternalError $ streamNumber strm-                enqueueControl controlQ $ CFrames Nothing [rst]-                return off+            _ <- sync Nothing+            return off+        output _ _ _ = undefined -- never reached          ----------------------------------------------------------------         headerContinue :: StreamId -> TokenHeaderList -> Bool -> Offset -> IO Offset@@ -322,8 +268,8 @@             -> Int             -> Maybe DynaNext             -> (Maybe ByteString -> IO NextTrailersMaker)-            -> IO ()-            -> Output Stream+            -> (Maybe OutputType -> IO Bool)+            -> Output             -> Bool             -> IO Offset         fillDataHeaderEnqueueNext@@ -332,7 +278,7 @@             datPayloadLen             Nothing             tlrmkr-            tell+            sync             _             reqflush = do                 let buf = confWriteBuffer `plusPtr` off@@ -344,7 +290,7 @@                         else return (Just trailers, defaultFlags)                 fillFrameHeader FrameData datPayloadLen streamNumber flag buf                 off'' <- handleTrailers mtrailers off'-                void tell+                _ <- sync Nothing                 halfClosedLocal ctx strm Finished                 decreaseWindowSize ctx strm datPayloadLen                 if reqflush@@ -422,3 +368,5 @@                     , flags = flag                     , streamId = sid                     }++        setSenderDone = atomically $ writeTVar senderDone True

@@ -27,13 +27,19 @@     -- ^ SETTINGS_MAX_HEADER_LIST_SIZE     , pingRateLimit :: Int     -- ^ Maximum number of pings allowed per second (CVE-2019-9512)+    , emptyFrameRateLimit :: Int+    -- ^ Maximum number of empty data frames allowed per second (CVE-2019-9518)+    , settingsRateLimit :: Int+    -- ^ Maximum number of settings frames allowed per second (CVE-2019-9515)+    , rstRateLimit :: Int+    -- ^ Maximum number of reset frames allowed per second (CVE-2023-44487)     }     deriving (Eq, Show)  -- | The default settings. -- -- >>> baseSettings--- Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Nothing, initialWindowSize = 65535, maxFrameSize = 16384, maxHeaderListSize = Nothing, pingRateLimit = 10}+-- Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Nothing, initialWindowSize = 65535, maxFrameSize = 16384, maxHeaderListSize = Nothing, pingRateLimit = 10, emptyFrameRateLimit = 4, settingsRateLimit = 4, rstRateLimit = 4} baseSettings :: Settings baseSettings =     Settings@@ -44,12 +50,15 @@         , maxFrameSize = defaultPayloadLength -- 2^14 (16,384)         , maxHeaderListSize = Nothing         , pingRateLimit = 10+        , emptyFrameRateLimit = 4+        , settingsRateLimit = 4+        , rstRateLimit = 4         }  -- | The default settings. -- -- >>> defaultSettings--- Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Nothing, pingRateLimit = 10}+-- Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Nothing, pingRateLimit = 10, emptyFrameRateLimit = 4, settingsRateLimit = 4, rstRateLimit = 4} defaultSettings :: Settings defaultSettings =     baseSettings@@ -62,7 +71,7 @@ -- | Updating settings. -- -- >>> fromSettingsList defaultSettings [(SettingsEnablePush,0),(SettingsMaxHeaderListSize,200)]--- Settings {headerTableSize = 4096, enablePush = False, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Just 200, pingRateLimit = 10}+-- Settings {headerTableSize = 4096, enablePush = False, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Just 200, pingRateLimit = 10, emptyFrameRateLimit = 4, settingsRateLimit = 4, rstRateLimit = 4} {- FOURMOLU_DISABLE -} fromSettingsList :: Settings -> SettingsList -> Settings fromSettingsList settings kvs = foldl' update settings kvs

@@ -0,0 +1,78 @@+{-# LANGUAGE RecordWildCards #-}++module Network.HTTP2.H2.Sync (prepareSync, syncWithSender) where++import Control.Concurrent+import Network.HTTP.Semantics.IO+import UnliftIO.STM++import Network.HTTP2.H2.Context+import Network.HTTP2.H2.Queue+import Network.HTTP2.H2.Types+import Network.HTTP2.H2.Window++prepareSync+    :: Stream+    -> OutputType+    -> Maybe (TBQueue StreamingChunk)+    -> IO ((MVar Sync, Maybe OutputType -> IO Bool), Output)+prepareSync strm otyp mtbq = do+    var <- newEmptyMVar+    let sync = makeSync strm mtbq (putMVar var)+        out = Output strm otyp sync+    return ((var, sync), out)++syncWithSender+    :: Context+    -> Stream+    -> (MVar Sync, Maybe OutputType -> IO Bool)+    -> IO ()+syncWithSender Context{..} strm (var, sync) = loop+  where+    loop = do+        s <- takeMVar var+        case s of+            Done -> return ()+            Cont wait newotyp -> do+                wait+                enqueueOutput outputQ $ Output strm newotyp sync+                loop++makeSync+    :: Stream+    -> Maybe (TBQueue StreamingChunk)+    -> (Sync -> IO ())+    -> Maybe OutputType+    -> IO Bool+makeSync _ _ sync Nothing = sync Done >> return False+makeSync strm mtbq sync (Just otyp) = do+    mwait <- checkOpen strm mtbq+    case mwait of+        Nothing -> return True+        Just wait -> do+            sync $ Cont wait otyp+            return False++checkOpen :: Stream -> Maybe (TBQueue StreamingChunk) -> IO (Maybe (IO ()))+checkOpen strm mtbq = case mtbq of+    Nothing -> checkStreamWindowSize+    Just tbq -> checkStreaming tbq+  where+    checkStreaming tbq = do+        isEmpty <- atomically $ isEmptyTBQueue tbq+        if isEmpty+            then do+                return $ Just (waitStreaming tbq)+            else checkStreamWindowSize+    -- FLOW CONTROL: WINDOW_UPDATE: send: respecting peer's limit+    checkStreamWindowSize = do+        sws <- getStreamWindowSize strm+        if sws <= 0+            then return $ Just (waitStreamWindowSize strm)+            else return Nothing++{-# INLINE waitStreaming #-}+waitStreaming :: TBQueue a -> IO ()+waitStreaming tbq = atomically $ do+    isEmpty <- isEmptyTBQueue tbq+    checkSTM (not isEmpty)

@@ -170,28 +170,24 @@  ---------------------------------------------------------------- -data Input a = Input a InpObj--data Output a = Output-    { outputStream :: a-    , outputObject :: OutObj+data Output = Output+    { outputStream :: Stream     , outputType :: OutputType-    , outputStrmQ :: Maybe (TBQueue StreamingChunk)-    , outputSentinel :: IO ()+    , outputSync :: Maybe OutputType -> IO Bool     }  data OutputType-    = OObj-    | OWait (IO ())+    = OHeader [Header] (Maybe DynaNext) TrailersMaker     | OPush TokenHeaderList StreamId -- associated stream id from client     | ONext DynaNext TrailersMaker +data Sync = Done | Cont (IO ()) OutputType+ ----------------------------------------------------------------  data Control     = CFinish HTTP2Error     | CFrames (Maybe SettingsList) [ByteString]-    | CGoaway ByteString (MVar ())  ---------------------------------------------------------------- 

@@ -1,5 +1,5 @@-{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}  module Network.HTTP2.H2.Window where 

@@ -46,6 +46,12 @@     maxFrameSize,     maxHeaderListSize, +    -- ** Rate limits+    pingRateLimit,+    settingsRateLimit,+    emptyFrameRateLimit,+    rstRateLimit,+     -- * Common configuration     Config (..),     allocSimpleConfig,

@@ -7,6 +7,7 @@ import Control.Concurrent.STM import Imports import Network.Control (defaultMaxData)+import Network.HTTP.Semantics.IO import Network.HTTP.Semantics.Server import Network.HTTP.Semantics.Server.Internal import Network.Socket (SockAddr)@@ -31,7 +32,7 @@ -- | The default server config. -- -- >>> defaultServerConfig--- ServerConfig {numberOfWorkers = 8, connectionWindowSize = 1048576, settings = Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Nothing, pingRateLimit = 10}}+-- ServerConfig {numberOfWorkers = 8, connectionWindowSize = 16777216, settings = Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Nothing, pingRateLimit = 10, emptyFrameRateLimit = 4, settingsRateLimit = 4, rstRateLimit = 4}} defaultServerConfig :: ServerConfig defaultServerConfig =     ServerConfig@@ -44,14 +45,15 @@  -- | Running HTTP/2 server. run :: ServerConfig -> Config -> Server -> IO ()-run sconf@ServerConfig{numberOfWorkers} conf server = do+run sconf conf server = do     ok <- checkPreface conf     when ok $ do-        (ctx, mgr) <- setup sconf conf-        let wc = fromContext ctx-        setAction mgr $ worker ctx wc mgr server-        replicateM_ numberOfWorkers $ spawnAction mgr-        runH2 conf ctx mgr+        let lnch ctx strm inpObj = do+                let label = "H2 worker for stream " ++ show (streamNumber strm)+                forkManaged (threadManager ctx) label $+                    worker conf server ctx strm inpObj+        ctx <- setup sconf conf lnch+        runH2 conf ctx  ---------------------------------------------------------------- @@ -60,6 +62,8 @@     , sioPeerSockAddr :: SockAddr     , sioReadRequest :: IO (StreamId, Stream, Request)     , sioWriteResponse :: Stream -> Response -> IO ()+    -- ^ 'Response' MUST be created with 'responseBuilder'.+    -- Others are not supported.     , sioWriteBytes :: ByteString -> IO ()     } @@ -73,17 +77,31 @@ runIO sconf conf@Config{..} action = do     ok <- checkPreface conf     when ok $ do-        (ctx@Context{..}, mgr) <- setup sconf conf-        let ServerInfo{..} = toServerInfo roleInfo-            get = do-                Input strm inObj <- atomically $ readTQueue inputQ-                return (streamNumber strm, strm, Request inObj)-            putR strm (Response outObj) = do-                let out = Output strm outObj OObj Nothing (return ())-                enqueueOutput outputQ out+        inpQ <- newTQueueIO+        let lnch _ strm inpObj = atomically $ writeTQueue inpQ (strm, inpObj)+        ctx@Context{..} <- setup sconf conf lnch+        let get = do+                (strm, inpObj) <- atomically $ readTQueue inpQ+                return (streamNumber strm, strm, Request inpObj)+            putR strm (Response OutObj{..}) = do+                case outObjBody of+                    OutBodyBuilder builder -> do+                        let next = fillBuilderBodyGetNext builder+                            sync _ = return True+                            out = OHeader outObjHeaders (Just next) outObjTrailers+                        enqueueOutput outputQ $ Output strm out sync+                    _ -> error "Response other than OutBodyBuilder is not supported"             putB bs = enqueueControl controlQ $ CFrames Nothing [bs]-        io <- action $ ServerIO confMySockAddr confPeerSockAddr get putR putB-        concurrently_ io $ runH2 conf ctx mgr+            serverIO =+                ServerIO+                    { sioMySockAddr = confMySockAddr+                    , sioPeerSockAddr = confPeerSockAddr+                    , sioReadRequest = get+                    , sioWriteResponse = putR+                    , sioWriteBytes = putB+                    }+        io <- action serverIO+        concurrently_ io $ runH2 conf ctx  checkPreface :: Config -> IO Bool checkPreface conf@Config{..} = do@@ -94,24 +112,22 @@             return False         else return True -setup :: ServerConfig -> Config -> IO (Context, Manager)-setup ServerConfig{..} conf@Config{..} = do-    serverInfo <- newServerInfo-    ctx <--        newContext-            serverInfo-            conf-            0-            connectionWindowSize-            settings-    -- Workers, worker manager and timer manager-    mgr <- start confTimeoutManager-    return (ctx, mgr)+setup :: ServerConfig -> Config -> Launch -> IO Context+setup ServerConfig{..} conf@Config{..} lnch = do+    let serverInfo = newServerInfo lnch+    newContext+        serverInfo+        conf+        0+        connectionWindowSize+        settings+        confTimeoutManager -runH2 :: Config -> Context -> Manager -> IO ()-runH2 conf ctx mgr = do-    let runReceiver = frameReceiver ctx conf-        runSender = frameSender ctx conf mgr+runH2 :: Config -> Context -> IO ()+runH2 conf ctx = do+    let mgr = threadManager ctx+        runReceiver = frameReceiver ctx conf+        runSender = frameSender ctx conf         runBackgroundThreads = concurrently_ runReceiver runSender     stopAfter mgr runBackgroundThreads $ \res -> do         closeAllStreams (oddStreamTable ctx) (evenStreamTable ctx) $

@@ -4,8 +4,6 @@  module Network.HTTP2.Server.Worker (     worker,-    WorkerConf (..),-    fromContext, ) where  import Data.IORef@@ -15,7 +13,6 @@ import Network.HTTP.Semantics.Server.Internal import Network.HTTP.Types import qualified System.TimeManager as T-import UnliftIO.Exception (SomeException (..)) import qualified UnliftIO.Exception as E import UnliftIO.STM @@ -25,172 +22,168 @@  ---------------------------------------------------------------- -data WorkerConf a = WorkerConf-    { readInputQ :: IO (Input a)-    , writeOutputQ :: Output a -> IO ()-    , workerCleanup :: a -> IO ()-    , isPushable :: IO Bool-    , makePushStream :: a -> PushPromise -> IO (StreamId, a)-    }--fromContext :: Context -> WorkerConf Stream-fromContext ctx@Context{..} =-    WorkerConf-        { readInputQ = atomically $ readTQueue $ inputQ $ toServerInfo roleInfo-        , writeOutputQ = enqueueOutput outputQ-        , workerCleanup = \strm -> do-            closed ctx strm Killed-            let frame = resetFrame InternalError $ streamNumber strm-            enqueueControl controlQ $ CFrames Nothing [frame]-        , -- Peer SETTINGS_ENABLE_PUSH-          isPushable = enablePush <$> readIORef peerSettings-        , -- Peer SETTINGS_INITIAL_WINDOW_SIZE-          makePushStream = \pstrm _ -> do-            -- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: send: respecting peer's limit-            (_, newstrm) <- openEvenStreamWait ctx-            let pid = streamNumber pstrm-            return (pid, newstrm)-        }+makePushStream :: Context -> Stream -> IO (StreamId, Stream)+makePushStream ctx pstrm = do+    -- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: send: respecting peer's limit+    (_, newstrm) <- openEvenStreamWait ctx+    let pid = streamNumber pstrm+    return (pid, newstrm)  ----------------------------------------------------------------  pushStream-    :: WorkerConf a-    -> a -- parent stream+    :: Config+    -> Context+    -> Stream -- parent stream     -> ValueTable -- request     -> [PushPromise]-    -> IO OutputType-pushStream _ _ _ [] = return OObj-pushStream WorkerConf{..} pstrm reqvt pps0-    | len == 0 = return OObj+    -> IO (Maybe (IO ()))+pushStream _ _ _ _ [] = return Nothing+pushStream conf ctx@Context{..} pstrm reqvt pps0+    | len == 0 = return Nothing     | otherwise = do-        pushable <- isPushable+        pushable <- enablePush <$> readIORef peerSettings         if pushable             then do                 tvar <- newTVarIO 0                 lim <- push tvar pps0 0                 if lim == 0-                    then return OObj-                    else return $ OWait (waiter lim tvar)-            else return OObj+                    then return Nothing+                    else return $ Just $ waiter lim tvar+            else return Nothing   where     len = length pps0     increment tvar = atomically $ modifyTVar' tvar (+ 1)+    -- Checking if all push are done.     waiter lim tvar = atomically $ do         n <- readTVar tvar         checkSTM (n >= lim)     push _ [] n = return (n :: Int)     push tvar (pp : pps) n = do-        (pid, newstrm) <- makePushStream pstrm pp-        let scheme = fromJust $ getFieldValue tokenScheme reqvt-            -- fixme: this value can be Nothing-            auth =-                fromJust-                    ( getFieldValue tokenAuthority reqvt-                        <|> getFieldValue tokenHost reqvt-                    )-            path = promiseRequestPath pp-            promiseRequest =-                [ (tokenMethod, methodGet)-                , (tokenScheme, scheme)-                , (tokenAuthority, auth)-                , (tokenPath, path)-                ]-            ot = OPush promiseRequest pid-            Response rsp = promiseResponse pp-            out = Output newstrm rsp ot Nothing $ increment tvar-        writeOutputQ out+        forkManaged threadManager "H2 server push" $ do+            timeoutKillThread threadManager $ \th -> do+                (pid, newstrm) <- makePushStream ctx pstrm+                let scheme = fromJust $ getFieldValue tokenScheme reqvt+                    -- fixme: this value can be Nothing+                    auth =+                        fromJust+                            ( getFieldValue tokenAuthority reqvt+                                <|> getFieldValue tokenHost reqvt+                            )+                    path = promiseRequestPath pp+                    promiseRequest =+                        [ (tokenMethod, methodGet)+                        , (tokenScheme, scheme)+                        , (tokenAuthority, auth)+                        , (tokenPath, path)+                        ]+                    ot = OPush promiseRequest pid+                    Response rsp = promiseResponse pp+                (vc, out) <- prepareSync newstrm ot Nothing+                enqueueOutput outputQ out+                syncWithSender ctx newstrm vc+                increment tvar+                sendHeaderBody conf ctx th newstrm rsp         push tvar pps (n + 1)  -- | 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.-response-    :: WorkerConf a-    -> Manager+sendResponse+    :: Config+    -> Context     -> T.Handle-    -> ThreadContinue-    -> a+    -> Stream     -> Request     -> Response     -> [PushPromise]     -> IO ()-response wc@WorkerConf{..} mgr th tconf strm (Request req) (Response rsp) pps = case outObjBody rsp of-    OutBodyNone -> do-        setThreadContinue tconf True-        writeOutputQ $ Output strm rsp OObj Nothing (return ())-    OutBodyBuilder _ -> do-        otyp <- pushStream wc strm reqvt pps-        setThreadContinue tconf True-        writeOutputQ $ Output strm rsp otyp Nothing (return ())-    OutBodyFile _ -> do-        otyp <- pushStream wc strm reqvt pps-        setThreadContinue tconf True-        writeOutputQ $ Output strm rsp otyp Nothing (return ())-    OutBodyStreaming strmbdy -> do-        otyp <- pushStream wc strm reqvt pps-        -- We must not exit this server application.-        -- If the application exits, streaming would be also closed.-        -- So, this work occupies this thread.-        ---        -- We need to increase the number of workers.-        spawnAction mgr-        -- After this work, this thread stops to decease-        -- the number of workers.-        setThreadContinue tconf False-        -- Since streaming body is loop, we cannot control it.-        -- So, let's serialize 'Builder' with a designated queue.-        tbq <- newTBQueueIO 10 -- fixme: hard coding: 10-        writeOutputQ $ Output strm rsp otyp (Just tbq) (return ())-        let push b = do-                T.pause th-                atomically $ writeTBQueue tbq (StreamingBuilder b Nothing)-                T.resume th-            flush = atomically $ writeTBQueue tbq StreamingFlush-            finished = atomically $ writeTBQueue tbq $ StreamingFinished (decCounter mgr)-        incCounter mgr-        strmbdy push flush `E.finally` finished-    OutBodyStreamingUnmask _ ->-        error "response: server does not support OutBodyStreamingUnmask"+sendResponse conf ctx th strm (Request req) (Response rsp) pps = do+    mwait <- pushStream conf ctx strm reqvt pps+    case mwait of+        Nothing -> return ()+        Just wait -> wait -- all pushes are sent+    sendHeaderBody conf ctx th strm rsp   where     (_, reqvt) = inpObjHeaders req +sendHeaderBody :: Config -> Context -> T.Handle -> Stream -> OutObj -> IO ()+sendHeaderBody Config{..} ctx@Context{..} th strm OutObj{..} = do+    (mnext, mtbq) <- case outObjBody of+        OutBodyNone -> return (Nothing, Nothing)+        OutBodyFile (FileSpec path fileoff bytecount) -> do+            (pread, closerOrRefresher) <- confPositionReadMaker path+            refresh <- case closerOrRefresher of+                Closer closer -> timeoutClose threadManager closer+                Refresher refresher -> return refresher+            let next = fillFileBodyGetNext pread fileoff bytecount refresh+            return (Just next, Nothing)+        OutBodyBuilder builder -> do+            let next = fillBuilderBodyGetNext builder+            return (Just next, Nothing)+        OutBodyStreaming strmbdy -> do+            q <- sendStreaming ctx strm th $ \OutBodyIface{..} -> strmbdy outBodyPush outBodyFlush+            let next = nextForStreaming q+            return (Just next, Just q)+        OutBodyStreamingIface strmbdy -> do+            q <- sendStreaming ctx strm th strmbdy+            let next = nextForStreaming q+            return (Just next, Just q)+    (vc, out) <-+        prepareSync strm (OHeader outObjHeaders mnext outObjTrailers) mtbq+    enqueueOutput outputQ out+    syncWithSender ctx strm vc+  where+    nextForStreaming+        :: TBQueue StreamingChunk+        -> DynaNext+    nextForStreaming tbq =+        let takeQ = atomically $ tryReadTBQueue tbq+            next = fillStreamBodyGetNext takeQ+         in next++sendStreaming+    :: Context+    -> Stream+    -> T.Handle+    -> (OutBodyIface -> IO ())+    -> IO (TBQueue StreamingChunk)+sendStreaming Context{..} strm th strmbdy = do+    tbq <- newTBQueueIO 10 -- fixme: hard coding: 10+    let label = "H2 streaming supporter for stream " ++ show (streamNumber strm)+    forkManaged threadManager label $ do+        let iface =+                OutBodyIface+                    { outBodyUnmask = id+                    , outBodyPush = \b -> do+                        T.pause th+                        atomically $ writeTBQueue tbq (StreamingBuilder b NotEndOfStream)+                        T.resume th+                    , outBodyPushFinal = \b -> do+                        T.pause th+                        atomically $ writeTBQueue tbq (StreamingBuilder b (EndOfStream Nothing))+                        T.resume th+                    , outBodyFlush = atomically $ writeTBQueue tbq StreamingFlush+                    }+            finished = atomically $ writeTBQueue tbq $ StreamingFinished Nothing+        strmbdy iface `E.finally` finished+    return tbq+ -- | Worker for server applications.-worker :: Context -> WorkerConf Stream -> Manager -> Server -> Action-worker ctx@Context{..} wc@WorkerConf{..} mgr server = do-    sinfo <- newStreamInfo-    tcont <- newThreadContinue-    timeoutKillThread mgr $ go sinfo tcont+worker :: Config -> Server -> Context -> Stream -> InpObj -> IO ()+worker conf server ctx@Context{..} strm req =+    timeoutKillThread threadManager $ \th -> do+        -- FIXME: exception+        T.pause th+        let req' = pauseRequestBody th+        T.resume th+        T.tickle th+        let aux = Aux th mySockAddr peerSockAddr+            request = Request req'+        server request aux $ sendResponse conf ctx th strm request+        adjustRxWindow ctx strm   where-    go sinfo tcont th = do-        setThreadContinue tcont True-        ex <- E.trySyncOrAsync $ do-            T.pause th-            Input strm req <- readInputQ-            let req' = pauseRequestBody req th-            setStreamInfo sinfo strm-            T.resume th-            T.tickle th-            let aux = Aux th mySockAddr peerSockAddr-            r <- server (Request req') aux $ response wc mgr th tcont strm (Request req')-            adjustRxWindow ctx strm-            return r-        cont1 <- case ex of-            Right () -> return True-            Left e@(SomeException _)-                -- killed by the local worker manager-                | Just KilledByHttp2ThreadManager{} <- E.fromException e -> return False-                -- killed by the local timeout manager-                | Just T.TimeoutThread <- E.fromException e -> do-                    cleanup sinfo-                    return True-                | otherwise -> do-                    cleanup sinfo-                    return True-        cont2 <- getThreadContinue tcont-        clearStreamInfo sinfo-        when (cont1 && cont2) $ go sinfo tcont th-    pauseRequestBody req th = req{inpObjBody = readBody'}+    pauseRequestBody th = req{inpObjBody = readBody'}       where         readBody = inpObjBody req         readBody' = do@@ -198,49 +191,3 @@             bs <- readBody             T.resume th             return bs-    cleanup sinfo = do-        minp <- getStreamInfo sinfo-        case minp of-            Nothing -> return ()-            Just strm -> workerCleanup strm----------------------------------------------------------------------   A reference is shared by a responder and its worker.---   The reference refers a value of this type as a return value.---   If 'True', the worker continue to serve requests.---   Otherwise, the worker get finished.-newtype ThreadContinue = ThreadContinue (IORef Bool)--{-# INLINE newThreadContinue #-}-newThreadContinue :: IO ThreadContinue-newThreadContinue = ThreadContinue <$> newIORef True--{-# INLINE setThreadContinue #-}-setThreadContinue :: ThreadContinue -> Bool -> IO ()-setThreadContinue (ThreadContinue ref) x = writeIORef ref x--{-# INLINE getThreadContinue #-}-getThreadContinue :: ThreadContinue -> IO Bool-getThreadContinue (ThreadContinue ref) = readIORef ref---------------------------------------------------------------------- | The type for cleaning up.-newtype StreamInfo a = StreamInfo (IORef (Maybe a))--{-# INLINE newStreamInfo #-}-newStreamInfo :: IO (StreamInfo a)-newStreamInfo = StreamInfo <$> newIORef Nothing--{-# INLINE clearStreamInfo #-}-clearStreamInfo :: StreamInfo a -> IO ()-clearStreamInfo (StreamInfo ref) = writeIORef ref Nothing--{-# INLINE setStreamInfo #-}-setStreamInfo :: StreamInfo a -> a -> IO ()-setStreamInfo (StreamInfo ref) inp = writeIORef ref $ Just inp--{-# INLINE getStreamInfo #-}-getStreamInfo :: StreamInfo a -> IO (Maybe a)-getStreamInfo (StreamInfo ref) = readIORef ref

@@ -1,6 +1,6 @@ cabal-version:      >=1.10 name:               http2-version:            5.2.6+version:            5.3.0 license:            BSD3 license-file:       LICENSE maintainer:         Kazu Yamamoto <kazu@iij.ad.jp>@@ -97,6 +97,7 @@         Network.HTTP2.H2.Settings         Network.HTTP2.H2.Stream         Network.HTTP2.H2.StreamTable+        Network.HTTP2.H2.Sync         Network.HTTP2.H2.Types         Network.HTTP2.H2.Window         Network.HTTP2.Frame.Decode@@ -115,7 +116,7 @@         bytestring >=0.10,         case-insensitive >=1.2 && <1.3,         containers >=0.6,-        http-semantics >= 0.1.2 && <0.2,+        http-semantics >= 0.2 && <0.3,         http-types >=0.12 && <0.13,         network >=3.1,         network-byte-order >=0.1.7 && <0.2,@@ -130,7 +131,7 @@     main-is:            h2c-client.hs     hs-source-dirs:     util     default-language:   Haskell2010-    other-modules:      Client+    other-modules:      Client Monitor     default-extensions: Strict StrictData     ghc-options:        -Wall -threaded -rtsopts     build-depends:@@ -139,7 +140,7 @@         bytestring,         http-types,         http2,-        network-run >= 0.3 && <0.4,+        network-run >= 0.3 && <0.5,         unix-time,         unliftio @@ -151,7 +152,7 @@ executable h2c-server     main-is:            h2c-server.hs     hs-source-dirs:     util-    other-modules:      Server+    other-modules:      Server Monitor     default-language:   Haskell2010     default-extensions: Strict StrictData     ghc-options:        -Wall -threaded@@ -311,7 +312,7 @@         http-types,         http2,         network,-        network-run >=0.1.0,+        network-run >=0.3.0,         random,         unliftio,         typed-process@@ -331,8 +332,9 @@         hspec >=1.3,         http-types,         http2,-        network-run >=0.1.0,-        typed-process+        network-run >=0.3.0,+        typed-process,+        unliftio      if flag(h2spec) 

@@ -13,7 +13,7 @@ import Data.Traversable (for) import Network.HTTP.Semantics import Network.HTTP.Types-import Network.Run.TCP+import Network.Run.TCP hiding (defaultSettings) import System.IO.Unsafe (unsafePerformIO) import System.Random import System.Timeout (timeout)

@@ -34,7 +34,7 @@         E.bracket             (allocSimpleConfig s 4096)             freeSimpleConfig-            (`run` server)+            (\conf -> run defaultServerConfig conf server)  server :: Server server req _aux sendResponse = case requestMethod req of

@@ -15,6 +15,8 @@  import Network.HTTP2.Client +import Monitor+ data Options = Options     { optPerformance :: Int     , optNumOfReqs :: Int@@ -23,6 +25,7 @@  client :: Options -> [Path] -> Client () client Options{..} paths sendRequest _aux = do+    labelMe "h2c client"     let cli             | optPerformance /= 0 = clientPF optPerformance sendRequest             | otherwise = clientNReqs optNumOfReqs sendRequest@@ -32,7 +35,9 @@         Left e -> print (e :: HTTP2Error)  clientNReqs :: Int -> SendRequest -> Path -> IO ()-clientNReqs n0 sendRequest path = loop n0+clientNReqs n0 sendRequest path = do+    labelMe "h2c clinet N requests"+    loop n0   where     req = requestNoBody methodGet path []     loop 0 = return ()@@ -45,6 +50,7 @@ -- Path is dummy clientPF :: Int -> SendRequest -> Path -> IO () clientPF n sendRequest _ = do+    labelMe "h2c clinet performance"     t1 <- getUnixTime     sendRequest req loop     t2 <- getUnixTime

@@ -0,0 +1,30 @@+module Monitor (monitor, labelMe) where++import Control.Monad+import Data.List+import Data.Maybe+import GHC.Conc.Sync++monitor :: IO () -> IO ()+monitor action = do+    labelMe "monitor"+    forever $ do+        action+        threadSummary >>= mapM_ (putStrLn . showT)+        putStr "\n"+  where+    showT (i, l, s) = i ++ " " ++ l ++ ": " ++ show s++threadSummary :: IO [(String, String, ThreadStatus)]+threadSummary = (sort <$> listThreads) >>= mapM summary+  where+    summary t = do+        let idstr = drop 9 $ show t+        l <- fromMaybe "(no name)" <$> threadLabel t+        s <- threadStatus t+        return (idstr, l, s)++labelMe :: String -> IO ()+labelMe lbl = do+    tid <- myThreadId+    labelThread tid lbl

@@ -3,6 +3,7 @@  module Main where +import Control.Concurrent import qualified Data.ByteString.Char8 as C8 import Network.HTTP2.Client import Network.Run.TCP (runTCPClient)@@ -12,6 +13,7 @@ import qualified UnliftIO.Exception as E  import Client+import Monitor  defaultOptions :: Options defaultOptions =@@ -51,6 +53,7 @@  main :: IO () main = do+    labelMe "h2c-client main"     args <- getArgs     (opts, ips) <- clientOpts args     (host, port, paths) <- case ips of@@ -59,6 +62,7 @@         h : p : [] -> return (h, p, ["/"])         h : p : ps -> return (h, p, C8.pack <$> ps)     let cliconf = defaultClientConfig{authority = host}+    _ <- forkIO $ monitor $ threadDelay 1000000     runTCPClient host port $ \s ->         E.bracket             (allocSimpleConfig s 4096)

@@ -4,6 +4,7 @@  module Main (main) where +import Control.Concurrent import Network.HTTP2.Server import Network.Run.TCP import System.Console.GetOpt@@ -11,6 +12,7 @@ import System.Exit import qualified UnliftIO.Exception as E +import Monitor import Server  options :: [OptDescr (Options -> Options)]@@ -38,12 +40,14 @@  main :: IO () main = do+    labelMe "h2c-server main"     args <- getArgs     (Options, ips) <- serverOpts args     (host, port) <- case ips of         [h, p] -> return (h, p)         _ -> showUsageAndExit usage-    runTCPServer (Just host) port $ \s ->+    _ <- forkIO $ monitor $ threadDelay 1000000+    runTCPServer (Just host) port $ \s -> do         E.bracket             (allocSimpleConfig s 4096)             freeSimpleConfig