http2-5.4.6: 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
server request aux $ sendResponse conf ctx lc strm request
adjustRxWindow ctx strm
modifyPeerLastStreamId ctx $ streamNumber strm
where
label = "H2 response sender for stream " ++ show (streamNumber strm)
pauseRequestBody th = req{inpObjBody = readBody'}
where
readBody = inpObjBody req
readBody' = do
T.pause th
bs <- readBody
T.resume th -- this is the same as 'tickle'
return bs
----------------------------------------------------------------
#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
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
(newstrm, lc) <- promise pp `E.finally` increment tvar
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.
promise pp = 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
lc <- newLoopCheck newstrm 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 (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
-> 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)