http2 5.3.5 → 5.3.6
raw patch · 12 files changed
+337/−373 lines, 12 files
Files
- ChangeLog.md +8/−0
- Network/HTTP2/Client.hs +3/−3
- Network/HTTP2/Client/Run.hs +42/−27
- Network/HTTP2/H2/Manager.hs +44/−123
- Network/HTTP2/H2/Receiver.hs +14/−24
- Network/HTTP2/H2/Sender.hs +42/−52
- Network/HTTP2/H2/Sync.hs +94/−75
- Network/HTTP2/H2/Types.hs +13/−3
- Network/HTTP2/H2/Window.hs +2/−0
- Network/HTTP2/Server/Run.hs +4/−8
- Network/HTTP2/Server/Worker.hs +70/−57
- http2.cabal +1/−1
ChangeLog.md view
@@ -1,5 +1,13 @@ # ChangeLog for http2 +## 5.3.6++* Making `runIO` friendly with the new synchronism mechanism.+ [#152](https://github.com/kazu-yamamoto/http2/pull/152)+* Re-throwing asynchronous exceptions to prevent thread leak.+* Simplifying the synchronism mechanism between workers and the sender.+ [#148](https://github.com/kazu-yamamoto/http2/pull/148)+ ## 5.3.5 * Using `http-semantics` v0.3.
Network/HTTP2/Client.hs view
@@ -13,8 +13,8 @@ -- > import qualified Data.ByteString.Char8 as C8 -- > import Network.HTTP.Types -- > import Network.Run.TCP (runTCPClient) -- network-run--- > import Control.Concurrent.Async -- unliftio--- > import qualified Control.Exception as E -- unliftio+-- > import Control.Concurrent.Async+-- > import qualified Control.Exception as E -- > -- > import Network.HTTP2.Client -- >@@ -24,7 +24,7 @@ -- > main :: IO () -- > main = runTCPClient serverName "80" $ runHTTP2Client serverName -- > where--- > cliconf host = defaultClientConfig { authority = C8.pack host }+-- > cliconf host = defaultClientConfig { authority = host } -- > runHTTP2Client host s = E.bracket (allocSimpleConfig s 4096) -- > freeSimpleConfig -- > (\conf -> run (cliconf host) conf client)
Network/HTTP2/Client/Run.hs view
@@ -84,15 +84,18 @@ { auxPossibleClientStreams = possibleClientStream ctx } clientCore ctx req processResponse = do- strm <- sendRequest conf ctx scheme authority req+ (strm, moutobj) <- makeStream ctx scheme authority req+ case moutobj of+ Nothing -> return ()+ Just outobj -> sendRequest conf ctx strm outobj False rsp <- getResponse strm x <- processResponse rsp adjustRxWindow ctx strm return x- runClient ctx = wrapClinet ctx $ client (clientCore ctx) $ aux ctx+ runClient ctx = wrapClient ctx $ client (clientCore ctx) $ aux ctx -wrapClinet :: Context -> IO a -> IO a-wrapClinet ctx client = do+wrapClient :: Context -> IO a -> IO a+wrapClient ctx client = do x <- client waitCounter0 $ threadManager ctx let frame = goawayFrame 0 NoError "graceful closing"@@ -109,13 +112,16 @@ ctx@Context{..} <- setup cconf conf let putB bs = enqueueControl controlQ $ CFrames Nothing [bs] putR req = do- strm <- sendRequest conf ctx scheme authority req+ (strm, moutobj) <- makeStream ctx scheme authority req+ case moutobj of+ Nothing -> return ()+ Just outobj -> sendRequest conf ctx strm outobj True return (streamNumber strm, strm) get = getResponse create = openOddStreamWait ctx runClient <- do act <- action $ ClientIO confMySockAddr confPeerSockAddr putR get putB create- return $ wrapClinet ctx act+ return $ wrapClient ctx act runH2 conf ctx runClient getResponse :: Stream -> IO Response@@ -165,23 +171,22 @@ Left () -> do wait runningClient -sendRequest- :: Config- -> Context+makeStream+ :: Context -> Scheme -> Authority -> Request- -> IO Stream-sendRequest conf ctx@Context{..} scheme auth (Request req) = do+ -> IO (Stream, Maybe OutObj)+makeStream ctx@Context{..} scheme auth (Request req) = do -- Checking push promises let hdr0 = outObjHeaders req- method = fromMaybe (error "sendRequest:method") $ lookup ":method" hdr0- path = fromMaybe (error "sendRequest:path") $ lookup ":path" hdr0+ method = fromMaybe (error "makeStream:method") $ lookup ":method" hdr0+ path = fromMaybe (error "makeStream:path") $ lookup ":path" hdr0 mstrm0 <- lookupEvenCache evenStreamTable method path case mstrm0 of Just strm0 -> do deleteEvenCache evenStreamTable method path- return strm0+ return (strm0, Nothing) Nothing -> do -- Arch/Sender is originally implemented for servers where -- the ordering of responses can be out-of-order.@@ -200,12 +205,12 @@ | otherwise = hdr1 req' = req{outObjHeaders = hdr2} -- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: send: respecting peer's limit- (sid, newstrm) <- openOddStreamWait ctx- sendHeaderBody conf ctx sid newstrm req'- return newstrm+ (_sid, newstrm) <- openOddStreamWait ctx+ return (newstrm, Just req') -sendHeaderBody :: Config -> Context -> StreamId -> Stream -> OutObj -> IO ()-sendHeaderBody Config{..} ctx@Context{..} sid newstrm OutObj{..} = do+sendRequest :: Config -> Context -> Stream -> OutObj -> Bool -> IO ()+sendRequest Config{..} ctx@Context{..} strm OutObj{..} io = do+ let sid = streamNumber strm (mnext, mtbq) <- case outObjBody of OutBodyNone -> return (Nothing, Nothing) OutBodyFile (FileSpec path fileoff bytecount) -> do@@ -216,22 +221,31 @@ let next = fillBuilderBodyGetNext builder return (Just next, Nothing) OutBodyStreaming strmbdy -> do- q <- sendStreaming ctx newstrm $ \iface ->+ q <- sendStreaming ctx strm $ \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+ q <- sendStreaming ctx strm strmbdy let next = nextForStreaming q return (Just next, Just q)- ((var, sync), out) <-- prepareSync newstrm (OHeader outObjHeaders mnext outObjTrailers) mtbq- atomically $ do+ let ot = OHeader outObjHeaders mnext outObjTrailers+ if io+ then do+ let out = makeOutputIO ctx strm ot+ pushOutput sid out+ else do+ (pop, out) <- makeOutput strm ot+ pushOutput sid out+ lc <- newLoopCheck strm mtbq+ forkManaged threadManager label $ syncWithSender' ctx pop lc+ where+ label = "H2 request sender for stream " ++ show (streamNumber strm)+ pushOutput sid out = atomically $ do sidOK <- readTVar outputQStreamID check (sidOK == sid)- enqueueOutputSTM outputQ out writeTVar outputQStreamID (sid + 2)- forkManaged threadManager "H2 worker" $ syncWithSender ctx newstrm var sync+ enqueueOutputSTM outputQ out sendStreaming :: Context@@ -239,11 +253,12 @@ -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk) sendStreaming Context{..} strm strmbdy = do- let label = "H2 streaming supporter for stream " ++ show (streamNumber strm) tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 forkManagedUnmask threadManager label $ \unmask -> withOutBodyIface tbq unmask strmbdy return tbq+ where+ label = "H2 request streaming sender for stream " ++ show (streamNumber strm) exchangeSettings :: Context -> IO () exchangeSettings Context{..} = do
Network/HTTP2/H2/Manager.hs view
@@ -9,8 +9,7 @@ stopAfter, forkManaged, forkManagedUnmask,- timeoutKillThread,- timeoutClose,+ withTimeout, KilledByHttp2ThreadManager (..), waitCounter0, ) where@@ -28,56 +27,39 @@ ---------------------------------------------------------------- -data Command- = Stop (MVar ()) (Maybe SomeException)- | Add ThreadId- | RegisterTimeout ThreadId T.Handle- | Delete ThreadId- -- | Manager to manage the thread and the timer.-data Manager = Manager (TQueue Command) (TVar Int) T.Manager+data Manager = Manager T.Manager (TVar ManagedThreads) +type ManagedThreads = Map ThreadId TimeoutHandle++----------------------------------------------------------------+ data TimeoutHandle = ThreadWithTimeout T.Handle | ThreadWithoutTimeout cancelTimeout :: TimeoutHandle -> IO ()-cancelTimeout (ThreadWithTimeout h) = T.cancel h+cancelTimeout (ThreadWithTimeout th) = T.cancel th cancelTimeout ThreadWithoutTimeout = return () -type ManagedThreads = Map ThreadId TimeoutHandle+---------------------------------------------------------------- -- | Starting a thread manager. -- Its action is initially set to 'return ()' and should be set -- by 'setAction'. This allows that the action can include -- the manager itself. start :: T.Manager -> IO Manager-start timmgr = do- q <- newTQueueIO- cnt <- newTVarIO 0- void $ forkIO $ do- labelMe "H2 thread manager"- go q Map.empty- return $ Manager q cnt timmgr- where- -- This runs in a separate thread whose ThreadId is not known by anyone- -- else, so it cannot be killed by asynchronous exceptions.- go :: TQueue Command -> ManagedThreads -> IO ()- go q threadMap0 = do- x <- atomically $ readTQueue q- case x of- Stop signalTimeoutsDisabled err -> do- kill signalTimeoutsDisabled threadMap0 err- Add newtid -> do- let threadMap = add newtid threadMap0- go q threadMap- RegisterTimeout tid h -> do- let threadMap = registerTimeout tid h threadMap0- go q threadMap- Delete oldtid -> do- threadMap <- del oldtid threadMap0- go q threadMap+start timmgr = Manager timmgr <$> newTVarIO Map.empty +----------------------------------------------------------------++data KilledByHttp2ThreadManager = KilledByHttp2ThreadManager (Maybe SomeException)+ deriving (Show)++instance Exception KilledByHttp2ThreadManager where+ toException = asyncExceptionToException+ fromException = asyncExceptionFromException+ -- | Stopping the manager. -- -- The action is run in the scope of an exception handler that catches all@@ -85,16 +67,17 @@ -- to cleanup in all circumstances. If an exception is caught, it is rethrown -- after the cleanup is complete. stopAfter :: Manager -> IO a -> (Maybe SomeException -> IO ()) -> IO a-stopAfter (Manager q _ _) action cleanup = do+stopAfter (Manager _timmgr var) action cleanup = do mask $ \unmask -> do ma <- try $ unmask action- signalTimeoutsDisabled <- newEmptyMVar- atomically $- writeTQueue q $- Stop signalTimeoutsDisabled (either Just (const Nothing) ma)- -- This call to takeMVar /will/ eventually succeed, because the Manager- -- thread cannot be killed (see comment on 'go' in 'start').- takeMVar signalTimeoutsDisabled+ m <- atomically $ do+ m0 <- readTVar var+ writeTVar var Map.empty+ return m0+ forM_ (Map.elems m) cancelTimeout+ let er = either Just (const Nothing) ma+ forM_ (Map.keys m) $ \tid ->+ E.throwTo tid $ KilledByHttp2ThreadManager er case ma of Left err -> cleanup (Just err) >> throwIO err Right a -> cleanup Nothing >> return a@@ -113,96 +96,34 @@ -- | Like 'forkManaged', but run action with exceptions masked forkManagedUnmask :: Manager -> String -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()-forkManagedUnmask mgr label io =+forkManagedUnmask (Manager _timmgr var) label io = void $ mask_ $ forkIOWithUnmask $ \unmask -> E.handle handler $ do labelMe label- addMyId mgr- incCounter mgr+ tid <- myThreadId+ atomically $ modifyTVar var $ Map.insert tid ThreadWithoutTimeout -- 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+ atomically $ modifyTVar var $ Map.delete tid 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- tid <- myThreadId- atomically $ writeTQueue q $ Add tid---- | Deleting my thread id from the kill-thread list on stopping.------ This is /only/ necessary when you want to remove the thread's ID from--- the manager /before/ the thread terminates (thereby assuming responsibility--- for thread cleanup yourself).-deleteMyId :: Manager -> IO ()-deleteMyId (Manager q _ _) = do- tid <- myThreadId- atomically $ writeTQueue q $ Delete tid+waitCounter0 :: Manager -> IO ()+waitCounter0 (Manager _timmgr var) = atomically $ do+ m <- readTVar var+ check (Map.size m == 0) ---------------------------------------------------------------- -add :: ThreadId -> ManagedThreads -> ManagedThreads-add tid = Map.insert tid ThreadWithoutTimeout--registerTimeout :: ThreadId -> T.Handle -> ManagedThreads -> ManagedThreads-registerTimeout tid = Map.insert tid . ThreadWithTimeout--del :: ThreadId -> ManagedThreads -> IO ManagedThreads-del tid threadMap = do- forM_ (Map.lookup tid threadMap) cancelTimeout- return $ Map.delete tid threadMap---- | Kill all threads------ We first remove all threads from the timeout manager, then signal that that--- is complete, and finally kill all threads. This avoids a race between the--- timeout manager and our manager: we want to ensure that the exception that--- gets delivered is 'KilledByHttp2ThreadManager', not 'TimeoutThread'.-kill :: MVar () -> ManagedThreads -> Maybe SomeException -> IO ()-kill signalTimeoutsDisabled threadMap err = do- forM_ (Map.elems threadMap) cancelTimeout- putMVar signalTimeoutsDisabled ()- forM_ (Map.keys threadMap) $ \tid ->- E.throwTo tid $ KilledByHttp2ThreadManager err---- | Killing the IO action of the second argument on timeout.-timeoutKillThread :: Manager -> (T.Handle -> IO a) -> IO a-timeoutKillThread (Manager q _ tmgr) action = E.bracket register T.cancel action+withTimeout :: Manager -> (T.Handle -> IO a) -> IO a+withTimeout (Manager timmgr var) action = do+ E.bracket register unregister $ \h ->+ action h where register = do- h <- T.registerKillThread tmgr (return ()) tid <- myThreadId- atomically $ writeTQueue q (RegisterTimeout tid h)- return h---- | Registering closer for a resource and--- returning a timer refresher.-timeoutClose :: Manager -> IO () -> IO (IO ())-timeoutClose (Manager _ _ tmgr) closer = do- th <- T.register tmgr closer- return $ T.tickle th--data KilledByHttp2ThreadManager = KilledByHttp2ThreadManager (Maybe SomeException)- deriving (Show)--instance Exception KilledByHttp2ThreadManager where- toException = asyncExceptionToException- fromException = asyncExceptionFromException--------------------------------------------------------------------incCounter :: Manager -> IO ()-incCounter (Manager _ cnt _) = atomically $ modifyTVar' cnt (+ 1)--decCounter :: Manager -> IO ()-decCounter (Manager _ cnt _) = atomically $ modifyTVar' cnt (subtract 1)--waitCounter0 :: Manager -> IO ()-waitCounter0 (Manager _ cnt _) = atomically $ do- n <- readTVar cnt- check (n < 1)+ th <- T.registerKillThread timmgr $ return ()+ -- overriding ThreadWithoutTimeout+ atomically $ modifyTVar var $ Map.insert tid $ ThreadWithTimeout th+ return th+ unregister th = T.cancel th
Network/HTTP2/H2/Receiver.hs view
@@ -58,6 +58,7 @@ loop sendGoaway se+ | isAsyncException se = E.throwIO se | Just GoAwayIsSent <- E.fromException se = do waitCounter0 threadManager enqueueControl controlQ $ CFinish GoAwayIsSent@@ -628,35 +629,24 @@ ---------------------------------------------------------------- -- | Type for input streaming.-data Source = Source (Int -> IO ()) RxQ (IORef ByteString) (IORef Bool)+data Source = Source RxQ (Int -> IO ()) (IORef Bool) mkSource :: RxQ -> (Int -> IO ()) -> IO Source-mkSource q inform = Source inform q <$> newIORef "" <*> newIORef False+mkSource q inform = Source q inform <$> newIORef False readSource :: Source -> IO (ByteString, Bool)-readSource (Source inform q refBS refEOF) = do+readSource (Source q inform refEOF) = do eof <- readIORef refEOF if eof then return (mempty, True) else do- (bs, isEOF) <- readBS- let len = BS.length bs- inform len- return (bs, isEOF)- where- readBS :: IO (ByteString, Bool)- readBS = do- bs0 <- readIORef refBS- if bs0 == ""- then do- mBS <- atomically $ readTQueue q- case mBS of- Left err -> do- writeIORef refEOF True- E.throwIO err- Right (bs, isEOF) -> do- writeIORef refEOF isEOF- return (bs, isEOF)- else do- writeIORef refBS ""- return (bs0, False)+ mBS <- atomically $ readTQueue q+ case mBS of+ Left err -> do+ writeIORef refEOF True+ E.throwIO err+ Right (bs, isEOF) -> do+ writeIORef refEOF isEOF+ let len = BS.length bs+ inform len+ return (bs, isEOF)
Network/HTTP2/H2/Sender.hs view
@@ -38,6 +38,7 @@ 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@@ -77,7 +78,7 @@ x <- atomically $ dequeue off case x of C ctl -> flushN off >> control ctl >> loop 0- O out -> outputOrEnqueueAgain out off >>= flushIfNecessary >>= loop+ 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@@ -139,29 +140,27 @@ Just siz -> setLimitForEncoding siz encodeDynamicTable ----------------------------------------------------------------- outputOrEnqueueAgain :: Output -> Offset -> IO Offset- outputOrEnqueueAgain out@(Output strm otyp sync) off = E.handle (\e -> resetStream strm InternalError e >> return off) $ do+ -- INVARIANT+ --+ -- Both the stream window and the connection window are open.+ ----------------------------------------------------------------+ outputAndSync :: Output -> Offset -> IO Offset+ outputAndSync out@(Output strm otyp sync) off = E.handle (\e -> resetStream strm InternalError e >> return off) $ do state <- readStreamState strm if isHalfClosedLocal state then return off 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+ OHeader hdr mnext tlrmkr -> do+ (off', mout') <- outputHeader strm hdr mnext tlrmkr sync off+ sync mout'+ return off' _ -> do- -- The 'sync' function usage constraints hold here: We- -- just popped off the only 'Output' for this stream,- -- and we only enqueue a new output (in 'output') if- -- 'sync' returns 'True'- 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+ sws <- getStreamWindowSize strm+ cws <- getConnectionWindowSize ctx -- not 0+ let lim = min cws sws+ (off', mout') <- output out off lim+ sync mout'+ return off' resetStream :: Stream -> ErrorCode -> E.SomeException -> IO () resetStream strm err e = do@@ -175,9 +174,9 @@ -> [Header] -> Maybe DynaNext -> TrailersMaker- -> (Maybe OutputType -> IO Bool)+ -> (Maybe Output -> IO ()) -> Offset- -> IO Offset+ -> IO (Offset, Maybe Output) outputHeader strm hdr mnext tlrmkr sync off0 = do -- Header frame and Continuation frame let sid = streamNumber strm@@ -186,19 +185,19 @@ 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+ Nothing -> do+ -- endOfStream+ halfClosedLocal ctx strm Finished+ return (off, Nothing) Just next -> do let out' = Output strm (ONext next tlrmkr) sync- outputOrEnqueueAgain out' off+ return (off, Just out') ----------------------------------------------------------------- output :: Output -> Offset -> WindowSize -> IO Offset- output out@(Output strm (ONext curr tlrmkr) sync) off0 lim = do+ output :: Output -> Offset -> WindowSize -> IO (Offset, Maybe Output)+ output out@(Output strm (ONext curr tlrmkr) _) off0 lim = do -- Data frame payload buflim <- readIORef outputBufferLimit let payloadOff = off0 + frameHeaderLength@@ -208,13 +207,12 @@ case next of Next datPayloadLen reqflush mnext -> do NextTrailersMaker tlrmkr' <- runTrailersMaker tlrmkr datBuf datPayloadLen- fillDataHeaderEnqueueNext+ fillDataHeader strm off0 datPayloadLen mnext tlrmkr'- sync out reqflush CancelNext mErr -> do@@ -233,15 +231,14 @@ resetStream strm InternalError err Nothing -> resetStream strm Cancel (E.toException CancelledStream)- return off0- output (Output strm (OPush ths pid) sync) off0 _lim = do+ 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. let sid = streamNumber strm len <- pushPromise pid sid ths off0 off <- flushIfNecessary $ off0 + frameHeaderLength + len- _ <- sync Nothing- return off+ return (off, Nothing) output _ _ _ = undefined -- never reached ----------------------------------------------------------------@@ -285,23 +282,21 @@ continue off' ths' FrameContinuation ----------------------------------------------------------------- fillDataHeaderEnqueueNext+ fillDataHeader :: Stream -> Offset -> Int -> Maybe DynaNext -> (Maybe ByteString -> IO NextTrailersMaker)- -> (Maybe OutputType -> IO Bool) -> Output -> Bool- -> IO Offset- fillDataHeaderEnqueueNext+ -> IO (Offset, Maybe Output)+ fillDataHeader strm@Stream{streamNumber} off datPayloadLen Nothing tlrmkr- sync _ reqflush = do let buf = confWriteBuffer `plusPtr` off@@ -321,41 +316,37 @@ else return off off'' <- handleTrailers mtrailers off'- _ <- sync Nothing halfClosedLocal ctx strm Finished if reqflush then do flushN off''- return 0- else return off''+ return (0, Nothing)+ else return (off'', Nothing) where handleTrailers Nothing off0 = return off0 handleTrailers (Just trailers) off0 = do (ths, _) <- toTokenHeaderTable trailers headerContinue streamNumber ths True {- endOfStream -} off0- fillDataHeaderEnqueueNext+ fillDataHeader _ off 0 (Just next) tlrmkr- _ out reqflush = do let out' = out{outputType = ONext next tlrmkr}- enqueueOutput outputQ out' if reqflush then do flushN off- return 0- else return off- fillDataHeaderEnqueueNext+ return (0, Just out')+ else return (off, Just out')+ fillDataHeader strm@Stream{streamNumber} off datPayloadLen (Just next) tlrmkr- _ out reqflush = do let buf = confWriteBuffer `plusPtr` off@@ -364,12 +355,11 @@ fillFrameHeader FrameData datPayloadLen streamNumber flag buf decreaseWindowSize ctx strm datPayloadLen let out' = out{outputType = ONext next tlrmkr}- enqueueOutput outputQ out' if reqflush then do flushN off'- return 0- else return off'+ return (0, Just out')+ else return (off', Just out') ---------------------------------------------------------------- pushPromise :: StreamId -> StreamId -> TokenHeaderList -> Offset -> IO Int
Network/HTTP2/H2/Sync.hs view
@@ -1,99 +1,118 @@ {-# LANGUAGE RecordWildCards #-} -module Network.HTTP2.H2.Sync (prepareSync, syncWithSender) where+module Network.HTTP2.H2.Sync (+ LoopCheck (..),+ newLoopCheck,+ syncWithSender,+ syncWithSender',+ makeOutput,+ makeOutputIO,+ enqueueOutputSIO,+) where import Control.Concurrent import Control.Concurrent.STM+import Control.Monad+import Network.Control import Network.HTTP.Semantics.IO import Network.HTTP2.H2.Context import Network.HTTP2.H2.Queue import Network.HTTP2.H2.Types-import Network.HTTP2.H2.Window --- | Two assumptions about how this function is used:------ 1. A separate thread will be running 'syncWithSender' using the @var@ and--- @sync@ values constructed here.--- 2. The 'Output' will be enqueued in the 'outputQ' of some 'Context'------ The returned @sync@ function then has the following usage constraints:------ 1. It may only be called with a 'Just' 'OutputType' if there is no 'Output'--- already enqueued in the 'outputQ' for the given stream.--- 2. If the function returns 'False', no other 'Output' may be enqueued for--- this stream (until one has been dequeued).-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 var- out = Output strm otyp sync- return ((var, sync), out)- syncWithSender :: Context -> Stream- -> MVar Sync- -- ^ Precondition: When this is filled with an 'Output' for a particular- -- stream, the 'outputQ' in the 'Context' /must not/ already contain an- -- 'Output' for that stream.- -> (Maybe OutputType -> IO Bool)+ -> OutputType+ -> LoopCheck -> IO ()-syncWithSender Context{..} strm var sync = loop+syncWithSender ctx@Context{..} strm otyp lc = do+ (pop, out) <- makeOutput strm otyp+ enqueueOutput outputQ out+ syncWithSender' ctx pop lc++makeOutput :: Stream -> OutputType -> IO (IO Sync, Output)+makeOutput strm otyp = do+ var <- newEmptyMVar+ let push mout = case mout of+ Nothing -> putMVar var Done+ Just ot -> putMVar var $ Cont ot+ pop = takeMVar var+ out =+ Output+ { outputStream = strm+ , outputType = otyp+ , outputSync = push+ }+ return (pop, out)++makeOutputIO :: Context -> Stream -> OutputType -> Output+makeOutputIO Context{..} strm otyp = out where+ push mout = case mout of+ Nothing -> return ()+ -- Sender enqueues output again ignoring+ -- the stream TX window.+ Just ot -> enqueueOutput outputQ ot+ out =+ Output+ { outputStream = strm+ , outputType = otyp+ , outputSync = push+ }++enqueueOutputSIO :: Context -> Stream -> OutputType -> IO ()+enqueueOutputSIO ctx@Context{..} strm otyp = do+ let out = makeOutputIO ctx strm otyp+ enqueueOutput outputQ out++syncWithSender' :: Context -> IO Sync -> LoopCheck -> IO ()+syncWithSender' Context{..} pop lc = loop+ where loop = do- s <- takeMVar var+ s <- pop case s of Done -> return ()- Cont wait newotyp -> do- wait- -- This is justified by the precondition above- enqueueOutput outputQ $ Output strm newotyp sync- loop+ Cont newout -> do+ cont <- checkLoop lc+ when cont $ do+ -- This is justified by the precondition above+ enqueueOutput outputQ newout+ loop --- | Postcondition: This will only write to the 'MVar' if:------ 1. You pass 'Just' an 'OutputType'--- 2. The return value is 'False'-makeSync- :: Stream- -> Maybe (TBQueue StreamingChunk)- -> MVar Sync- -> Maybe OutputType- -> IO Bool-makeSync _ _ var Nothing = putMVar var Done >> return False-makeSync strm mtbq var (Just otyp) = do- mwait <- checkOpen strm mtbq- case mwait of- Nothing -> return True- Just wait -> do- putMVar var $ Cont wait otyp- return False+newLoopCheck :: Stream -> Maybe (TBQueue StreamingChunk) -> IO LoopCheck+newLoopCheck strm mtbq = do+ tovar <- newTVarIO False+ return $+ LoopCheck+ { lcTBQ = mtbq+ , lcTimeout = tovar+ , lcWindow = streamTxFlow strm+ } -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+data LoopCheck = LoopCheck+ { lcTBQ :: Maybe (TBQueue StreamingChunk)+ , lcTimeout :: TVar Bool+ , lcWindow :: TVar TxFlow+ } -{-# INLINE waitStreaming #-}-waitStreaming :: TBQueue a -> IO ()-waitStreaming tbq = atomically $ do+checkLoop :: LoopCheck -> IO Bool+checkLoop LoopCheck{..} = atomically $ do+ tout <- readTVar lcTimeout+ if tout+ then return False+ else do+ waitStreaming' lcTBQ+ waitStreamWindowSizeSTM lcWindow+ return True++waitStreaming' :: Maybe (TBQueue a) -> STM ()+waitStreaming' Nothing = return ()+waitStreaming' (Just tbq) = do isEmpty <- isEmptyTBQueue tbq check (not isEmpty)++waitStreamWindowSizeSTM :: TVar TxFlow -> STM ()+waitStreamWindowSizeSTM txf = do+ w <- txWindowSize <$> readTVar txf+ check (w > 0)
Network/HTTP2/H2/Types.hs view
@@ -7,7 +7,11 @@ import Control.Concurrent import Control.Concurrent.STM-import Control.Exception (SomeException)+import Control.Exception (+ Exception,+ SomeAsyncException (..),+ SomeException (..),+ ) import qualified Control.Exception as E import Data.IORef import Data.Typeable@@ -179,7 +183,7 @@ data Output = Output { outputStream :: Stream , outputType :: OutputType- , outputSync :: Maybe OutputType -> IO Bool+ , outputSync :: Maybe Output -> IO () } data OutputType@@ -187,7 +191,7 @@ | OPush TokenHeaderList StreamId -- associated stream id from client | ONext DynaNext TrailersMaker -data Sync = Done | Cont (IO ()) OutputType+data Sync = Done | Cont Output ---------------------------------------------------------------- @@ -270,3 +274,9 @@ , confPeerSockAddr :: SockAddr -- ^ This is copied into 'Aux', if exist, on server. }++isAsyncException :: Exception e => e -> Bool+isAsyncException e =+ case E.fromException (E.toException e) of+ Just (SomeAsyncException _) -> True+ Nothing -> False
Network/HTTP2/H2/Window.hs view
@@ -81,6 +81,8 @@ cframe = CFrames Nothing [frame] enqueueControl controlQ cframe +-- This must be called after an application is finished+-- to adjust RX window. adjustRxWindow :: Context -> Stream -> IO () adjustRxWindow ctx stream@Stream{streamRxQ} = do mq <- readIORef streamRxQ
Network/HTTP2/Server/Run.hs view
@@ -49,10 +49,7 @@ run sconf conf server = do ok <- checkPreface conf when ok $ do- 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+ let lnch = runServer conf server ctx <- setup sconf conf lnch runH2 conf ctx @@ -79,7 +76,7 @@ when ok $ do inpQ <- newTQueueIO let lnch _ strm inpObj = atomically $ writeTQueue inpQ (strm, inpObj)- ctx@Context{..} <- setup sconf conf lnch+ ctx <- setup sconf conf lnch let get = do (strm, inpObj) <- atomically $ readTQueue inpQ return (strm, Request inpObj)@@ -87,9 +84,8 @@ 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+ otyp = OHeader outObjHeaders (Just next) outObjTrailers+ enqueueOutputSIO ctx strm otyp _ -> error "Response other than OutBodyBuilder is not supported" serverIO = ServerIO
Network/HTTP2/Server/Worker.hs view
@@ -3,7 +3,7 @@ {-# LANGUAGE RecordWildCards #-} module Network.HTTP2.Server.Worker (- worker,+ runServer, ) where import Control.Concurrent.STM@@ -21,15 +21,54 @@ ---------------------------------------------------------------- -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)+runServer :: Config -> Server -> Launch+runServer conf server ctx@Context{..} strm req =+ forkManaged threadManager label $+ withTimeout threadManager $ \th -> do+ -- FIXME: exception+ let req' = pauseRequestBody th+ aux = Aux th mySockAddr peerSockAddr+ request = Request req'+ lc <- newLoopCheck strm Nothing+ server request aux $ sendResponse conf ctx lc th strm request+ adjustRxWindow ctx strm+ where+ label = "H2 response sender for stream " ++ show (streamNumber strm)+ pauseRequestBody th = req{inpObjBody = readBody'}+ where+ readBody = inpObjBody req+ readBody' = do+ T.pause th+ bs <- readBody+ T.resume th+ return bs ---------------------------------------------------------------- +-- | 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.+sendResponse+ :: Config+ -> Context+ -> LoopCheck+ -> T.Handle+ -> Stream+ -> Request+ -> Response+ -> [PushPromise]+ -> IO ()+sendResponse conf ctx lc 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 lc th strm rsp+ where+ (_, reqvt) = inpObjHeaders req++----------------------------------------------------------------+ pushStream :: Config -> Context@@ -60,7 +99,7 @@ push _ [] n = return (n :: Int) push tvar (pp : pps) n = do forkManaged threadManager "H2 server push" $ do- timeoutKillThread threadManager $ \th -> do+ withTimeout threadManager $ \th -> do (pid, newstrm) <- makePushStream ctx pstrm let scheme = fromJust $ getFieldValue tokenScheme reqvt -- fixme: this value can be Nothing@@ -78,36 +117,32 @@ ] ot = OPush promiseRequest pid Response rsp = promiseResponse pp- ((var, sync), out) <- prepareSync newstrm ot Nothing- enqueueOutput outputQ out- syncWithSender ctx newstrm var sync increment tvar- sendHeaderBody conf ctx th newstrm rsp+ lc <- newLoopCheck newstrm Nothing+ syncWithSender ctx newstrm ot lc+ sendHeaderBody conf ctx lc 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.-sendResponse+----------------------------------------------------------------++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)++----------------------------------------------------------------++sendHeaderBody :: Config -> Context+ -> LoopCheck -> T.Handle -> Stream- -> Request- -> Response- -> [PushPromise]+ -> OutObj -> IO ()-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+sendHeaderBody Config{..} ctx lc th strm OutObj{..} = do (mnext, mtbq) <- case outObjBody of OutBodyNone -> return (Nothing, Nothing) OutBodyFile (FileSpec path fileoff bytecount) -> do@@ -125,11 +160,11 @@ q <- sendStreaming ctx strm th strmbdy let next = nextForStreaming q return (Just next, Just q)- ((var, sync), out) <-- prepareSync strm (OHeader outObjHeaders mnext outObjTrailers) mtbq- enqueueOutput outputQ out- syncWithSender ctx strm var sync+ let lc' = lc{lcTBQ = mtbq}+ syncWithSender ctx strm (OHeader outObjHeaders mnext outObjTrailers) lc' +----------------------------------------------------------------+ sendStreaming :: Context -> Stream@@ -137,7 +172,6 @@ -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk) sendStreaming Context{..} strm th strmbdy = do- let label = "H2 streaming supporter for stream " ++ show (streamNumber strm) tbq <- newTBQueueIO 10 -- fixme: hard coding: 10 forkManaged threadManager label $ withOutBodyIface tbq id $ \iface -> do@@ -154,26 +188,5 @@ } strmbdy iface' return tbq---- | Worker for server applications.-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- pauseRequestBody th = req{inpObjBody = readBody'}- where- readBody = inpObjBody req- readBody' = do- T.pause th- bs <- readBody- T.resume th- return bs+ label = "H2 response streaming sender for " ++ show (streamNumber strm)
http2.cabal view
@@ -1,6 +1,6 @@ cabal-version: >=1.10 name: http2-version: 5.3.5+version: 5.3.6 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto <kazu@iij.ad.jp>