http2-5.2.0: Network/HTTP2/Server/Worker.hs
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternGuards #-}
{-# LANGUAGE RecordWildCards #-}
module Network.HTTP2.Server.Worker (
worker,
WorkerConf (..),
fromContext,
) where
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 Network.Socket (SockAddr)
import qualified System.TimeManager as T
import UnliftIO.Exception (SomeException (..))
import qualified UnliftIO.Exception as E
import UnliftIO.STM
import Imports hiding (insert)
import Network.HTTP2.Frame
import Network.HTTP2.H2
----------------------------------------------------------------
data WorkerConf a = WorkerConf
{ readInputQ :: IO (Input a)
, writeOutputQ :: Output a -> IO ()
, workerCleanup :: a -> IO ()
, isPushable :: IO Bool
, makePushStream :: a -> PushPromise -> IO (StreamId, a)
, mySockAddr :: SockAddr
, peerSockAddr :: SockAddr
}
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)
, mySockAddr = mySockAddr
, peerSockAddr = peerSockAddr
}
----------------------------------------------------------------
pushStream
:: WorkerConf a
-> a -- parent stream
-> ValueTable -- request
-> [PushPromise]
-> IO OutputType
pushStream _ _ _ [] = return OObj
pushStream WorkerConf{..} pstrm reqvt pps0
| len == 0 = return OObj
| otherwise = do
pushable <- isPushable
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
where
len = length pps0
increment tvar = atomically $ modifyTVar' tvar (+ 1)
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
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
-> T.Handle
-> ThreadContinue
-> a
-> 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)
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"
where
(_, reqvt) = inpObjHeaders req
-- | Worker for server applications.
worker :: WorkerConf a -> Manager -> Server -> Action
worker wc@WorkerConf{..} mgr server = do
sinfo <- newStreamInfo
tcont <- newThreadContinue
timeoutKillThread mgr $ go sinfo tcont
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
server (Request req') aux $ response wc mgr th tcont strm (Request req')
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'}
where
readBody = inpObjBody req
readBody' = do
T.pause th
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