http2-5.4.9: Network/HTTP2/Server/Worker.hs
{-# LANGUAGE CPP #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RecordWildCards #-}
module Network.HTTP2.Server.Worker (
runServer,
) where
import Control.Concurrent.STM
import qualified Control.Exception as E
import Data.IORef
import Network.HTTP.Semantics
import Network.HTTP.Semantics.IO
import Network.HTTP.Semantics.Server
import Network.HTTP.Semantics.Server.Internal
import Network.HTTP.Types
import qualified System.ThreadManager as T
import Imports hiding (insert)
import Network.HTTP2.Frame
import Network.HTTP2.H2
import Network.HTTP2.H2.OutBodyIface
#if MIN_VERSION_http_semantics(0,4,1)
import qualified Data.ByteString.Char8 as C8
#endif
----------------------------------------------------------------
runServer :: Config -> Server -> Launch
runServer conf server ctx@Context{..} strm req =
T.forkManagedTimeout threadManager label $ \th -> do
let req' = pauseRequestBody th
aux =
defaultAux
{ auxTimeHandle = th
, auxMySockAddr = mySockAddr
, auxPeerSockAddr = peerSockAddr
#if MIN_VERSION_http_semantics(0,4,1)
, auxSendInformational = sendInformational ctx strm
#endif
}
request = Request req'
lc <- newLoopCheck strm Nothing (Just th)
server request aux (sendResponse conf ctx lc strm request)
`E.catch` \e -> do
resetFailedStream ctx strm e
E.throwIO (e :: E.SomeException)
doneWithRequestBody 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 -- this is the same as 'tickle'
return bs
-- | Called once the handler has returned: nobody reads the request body
-- from here on.
--
-- What is left of it used to be queued all the same, for a reader that was
-- gone, one entry per DATA frame. Flow control bounds the octets, not the
-- frames: a peer sending one octet a frame had us hold some ten times what
-- it sent, for as long as it kept the stream open. Nor was any of it given
-- back to the windows, so a peer still sending the rest of a body the
-- handler had not waited for -- a 404 to a POST, say -- stalled once the
-- stream's window was used up, and took the connection window with it.
--
-- So the rest is thrown away as it comes ('streamDiscardBody') and given
-- back to both windows, and the peer can send it to the end and close the
-- stream. What was already queued is given back too, and a reader still
-- there -- a thread the handler left behind -- is told.
doneWithRequestBody :: Context -> Stream -> IO ()
doneWithRequestBody ctx strm@Stream{..} = do
atomically $ writeTVar streamDiscardBody True
adjustRxWindow ctx strm
mq <- readIORef streamRxQ
forM_ mq $ \q ->
atomically $ writeTQueue q $ Left $ E.toException CancelledStream
-- | Reset the stream of a handler that has failed: it threw, or the timeout
-- manager killed it.
--
-- Nothing else would. The response, or the rest of it, never comes, so
-- the stream stayed open: the client waited for ever, the stream held its
-- slot of our SETTINGS_MAX_CONCURRENT_STREAMS -- after that many failures
-- every new request was refused -- and, after a GOAWAY, the connection
-- never drained. A stream already closed, by a response that went out in
-- full or by a reset, is left alone.
--
-- Closed in the same transaction as it is checked, so that a reader of the
-- request body is told ('informReplaced'), and so is the sender, which
-- sends nothing more on a closed stream. What was left unread goes back
-- to the connection window.
resetFailedStream :: Context -> Stream -> E.SomeException -> IO ()
resetFailedStream ctx strm@Stream{..} e = do
open <- atomically $ do
st <- readTVar streamState
if isClosed st
then return False
else do
informReplaced streamNumber st (Closed cc)
writeTVar streamState (Closed cc)
return True
when open $ do
closed ctx strm cc
enqueueControl (controlQ ctx) $
CFrames Nothing [resetFrame InternalError streamNumber]
giveBackUnread ctx strm
where
cc = ResetByMe e
----------------------------------------------------------------
#if MIN_VERSION_http_semantics(0,4,1)
-- | Send an informational (1xx) response, e.g. 103 Early Hints, on the given
-- stream ahead of the final response. This is wired into 'auxSendInformational'
-- so that a server (or WAI handler via Warp) can emit early hints. It blocks
-- until the informational HEADERS have been handed to the sender, preserving
-- ordering with respect to the final response.
sendInformational :: Context -> Stream -> Status -> ResponseHeaders -> IO ()
sendInformational ctx strm st hdrs = do
lc <- newLoopCheck strm Nothing Nothing
let hdr = (":status", C8.pack (show (statusCode st))) : hdrs
syncWithSender ctx strm (OInformational hdr) lc
#endif
----------------------------------------------------------------
-- | This function is passed to workers.
-- They also pass 'Response's from a server to this function.
-- This function enqueues commands for the HTTP/2 sender.
sendResponse
:: Config
-> Context
-> LoopCheck
-> Stream
-> Request
-> Response
-> [PushPromise]
-> IO ()
sendResponse conf ctx lc 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 strm rsp
where
(_, reqvt) = inpObjHeaders req
----------------------------------------------------------------
pushStream
:: Config
-> Context
-> Stream -- parent stream
-> ValueTable -- request
-> [PushPromise]
-> IO (Maybe (IO ()))
pushStream _ _ _ _ [] = return Nothing
pushStream conf ctx@Context{..} pstrm reqvt pps0
| len == 0 = return Nothing
| otherwise = do
pushable <- enablePush <$> readIORef peerSettings
if pushable
then do
tvar <- newTVarIO 0
lim <- push tvar pps0 0
if lim == 0
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
check (n >= lim)
push _ [] n = return (n :: Int)
push tvar (pp : pps) n = do
T.forkManaged threadManager "H2 server push" $ do
mpushed <- promise pp `E.finally` increment tvar
forM_ mpushed $ \(newstrm, lc) -> do
let Response rsp = promiseResponse pp
sendHeaderBody conf ctx lc newstrm rsp
push tvar pps (n + 1)
-- Sending the PUSH_PROMISE, and only then counting the push as done:
-- 'waiter' holds the parent's response back until every push is
-- counted. The PUSH_PROMISE has to go out before the parent's frames
-- (RFC 9113, section 8.4.1) -- before its END_STREAM above all, after
-- which a PUSH_PROMISE on it is a connection error. Counted before it
-- was queued, the parent's response could overtake it, and a client
-- asked for the pushed resource itself before hearing of the promise.
-- 'syncWithSender' returns once the sender has written the frame.
-- Counted however it ends, or the parent would wait for ever. A push
-- the peer has no room for is not made ('openEvenStreamTry').
promise pp = do
mstrm <- makePushStream ctx pstrm
forM mstrm $ \(pid, newstrm) -> promiseOn pp pid newstrm
promiseOn pp pid newstrm = do
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
lc <- newLoopCheck newstrm Nothing Nothing
syncWithSender ctx newstrm ot lc
-- Reserved (local) until now. The peer sends nothing on a pushed
-- stream, so its side is closed from here (RFC 9113, section 5.1:
-- "half-closed (remote)" once the HEADERS go out), and the END_STREAM
-- of the pushed response closes the stream. Left reserved, that
-- END_STREAM only half-closed it: the stream stayed in the table
-- holding a slot of the peer's SETTINGS_MAX_CONCURRENT_STREAMS, and
-- once that many pushes had been made, the next waited for a slot
-- for ever, and so did the response it belonged to.
halfClosedRemote ctx newstrm
return (newstrm, lc)
----------------------------------------------------------------
makePushStream :: Context -> Stream -> IO (Maybe (StreamId, Stream))
makePushStream ctx pstrm = do
-- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: send: respecting peer's limit
mstrm <- openEvenStreamTry ctx
let pid = streamNumber pstrm
return $ (\(_, newstrm) -> (pid, newstrm)) <$> mstrm
----------------------------------------------------------------
sendHeaderBody
:: Config
-> Context
-> LoopCheck
-> Stream
-> OutObj
-> IO ()
sendHeaderBody Config{..} ctx lc strm OutObj{..} = do
(mnext, mtbq) <- case outObjBody of
OutBodyNone -> return (Nothing, Nothing)
OutBodyFile (FileSpec path fileoff bytecount) -> do
(pread, sentinel) <- confPositionReadMaker path
let next = fillFileBodyGetNext pread fileoff bytecount sentinel
return (Just next, Nothing)
OutBodyBuilder builder -> do
let next = fillBuilderBodyGetNext builder
return (Just next, Nothing)
OutBodyStreaming strmbdy -> do
q <- sendStreaming ctx strm $ \OutBodyIface{..} -> strmbdy outBodyPush outBodyFlush
let next = nextForStreaming q
return (Just next, Just q)
OutBodyStreamingIface strmbdy -> do
q <- sendStreaming ctx strm strmbdy
let next = nextForStreaming q
return (Just next, Just q)
let lc' = lc{lcTBQ = mtbq}
syncWithSender ctx strm (OHeader outObjHeaders mnext outObjTrailers) lc'
----------------------------------------------------------------
sendStreaming
:: Context
-> Stream
-> (OutBodyIface -> IO ())
-> IO (TBQueue StreamingChunk)
sendStreaming ctx@Context{..} strm strmbdy = do
tbq <- newTBQueueIO 10 -- fixme: hard coding: 10
T.forkManagedTimeout threadManager label $ \th ->
withOutBodyIface ctx strm tbq id $ \iface -> do
let iface' =
iface
{ outBodyPush = \b -> do
T.pause th
outBodyPush iface b
T.resume th -- this is the same as 'tickle'
, outBodyPushFinal = \b -> do
T.pause th
outBodyPushFinal iface b
T.resume th -- this is the same as 'tickle'
}
strmbdy iface'
return tbq
where
label = "H2 response streaming sender for " ++ show (streamNumber strm)