diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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. diff --git a/Network/HTTP2/Client.hs b/Network/HTTP2/Client.hs --- a/Network/HTTP2/Client.hs +++ b/Network/HTTP2/Client.hs @@ -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) diff --git a/Network/HTTP2/Client/Run.hs b/Network/HTTP2/Client/Run.hs --- a/Network/HTTP2/Client/Run.hs +++ b/Network/HTTP2/Client/Run.hs @@ -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 diff --git a/Network/HTTP2/H2/Manager.hs b/Network/HTTP2/H2/Manager.hs --- a/Network/HTTP2/H2/Manager.hs +++ b/Network/HTTP2/H2/Manager.hs @@ -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 diff --git a/Network/HTTP2/H2/Receiver.hs b/Network/HTTP2/H2/Receiver.hs --- a/Network/HTTP2/H2/Receiver.hs +++ b/Network/HTTP2/H2/Receiver.hs @@ -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) diff --git a/Network/HTTP2/H2/Sender.hs b/Network/HTTP2/H2/Sender.hs --- a/Network/HTTP2/H2/Sender.hs +++ b/Network/HTTP2/H2/Sender.hs @@ -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 diff --git a/Network/HTTP2/H2/Sync.hs b/Network/HTTP2/H2/Sync.hs --- a/Network/HTTP2/H2/Sync.hs +++ b/Network/HTTP2/H2/Sync.hs @@ -1,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) diff --git a/Network/HTTP2/H2/Types.hs b/Network/HTTP2/H2/Types.hs --- a/Network/HTTP2/H2/Types.hs +++ b/Network/HTTP2/H2/Types.hs @@ -7,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 diff --git a/Network/HTTP2/H2/Window.hs b/Network/HTTP2/H2/Window.hs --- a/Network/HTTP2/H2/Window.hs +++ b/Network/HTTP2/H2/Window.hs @@ -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 diff --git a/Network/HTTP2/Server/Run.hs b/Network/HTTP2/Server/Run.hs --- a/Network/HTTP2/Server/Run.hs +++ b/Network/HTTP2/Server/Run.hs @@ -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 diff --git a/Network/HTTP2/Server/Worker.hs b/Network/HTTP2/Server/Worker.hs --- a/Network/HTTP2/Server/Worker.hs +++ b/Network/HTTP2/Server/Worker.hs @@ -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) diff --git a/http2.cabal b/http2.cabal --- a/http2.cabal +++ b/http2.cabal @@ -1,6 +1,6 @@ cabal-version: >=1.10 name: http2 -version: 5.3.5 +version: 5.3.6 license: BSD3 license-file: LICENSE maintainer: Kazu Yamamoto