{-# 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 qualified System.TimeManager as TM

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 :: Config -> Server -> Launch
runServer Config
conf Server
server ctx :: Context
ctx@Context{TVar Int
TVar (Maybe Int)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef Int
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
Int -> TokenHeaderTable -> IO ()
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar Int
peerStreamId :: IORef Int
peerLastStreamId :: IORef Int
outputBufferLimit :: IORef Int
outputQ :: TQueue Output
outputQStreamID :: TVar Int
controlQ :: ControlQ
encodeDynamicTable :: DynamicTable
decodeDynamicTable :: DynamicTable
txFlow :: TVar TxFlow
rxFlow :: IORef RxFlow
pingRate :: Rate
settingsRate :: Rate
emptyFrameRate :: Rate
rstRate :: Rate
mySockAddr :: SockAddr
peerSockAddr :: SockAddr
threadManager :: ThreadManager
timeoutManager :: Manager
receiverDone :: TVar (Maybe SomeException)
appDone :: STM Bool
informationalCallback :: Int -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
resetByMe :: Context -> IORef IntSet
activeRequests :: Context -> TVar Int
peerGoAway :: Context -> TVar (Maybe Int)
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
appDone :: Context -> STM Bool
receiverDone :: Context -> TVar (Maybe SomeException)
timeoutManager :: Context -> Manager
threadManager :: Context -> ThreadManager
peerSockAddr :: Context -> SockAddr
mySockAddr :: Context -> SockAddr
rstRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
settingsRate :: Context -> Rate
pingRate :: Context -> Rate
rxFlow :: Context -> IORef RxFlow
txFlow :: Context -> TVar TxFlow
decodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable :: Context -> DynamicTable
controlQ :: Context -> ControlQ
outputQStreamID :: Context -> TVar Int
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
peerStreamId :: Context -> IORef Int
myStreamId :: Context -> TVar Int
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
peerSettings :: Context -> IORef Settings
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
..} Stream
strm InpObj
req =
    ThreadManager -> Manager -> String -> (Handle -> IO ()) -> IO ()
T.forkManagedTimeout ThreadManager
threadManager Manager
timeoutManager String
label ((Handle -> IO ()) -> IO ()) -> (Handle -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Handle
th -> do
        let req' :: InpObj
req' = Handle -> InpObj
pauseRequestBody Handle
th
            aux :: Aux
aux =
                Aux
defaultAux
                    { auxTimeHandle = th
                    , auxMySockAddr = mySockAddr
                    , auxPeerSockAddr = peerSockAddr
#if MIN_VERSION_http_semantics(0,4,1)
                    , auxSendInformational = sendInformational ctx strm
#endif
                    }
            request :: Request
request = InpObj -> Request
Request InpObj
req'
        LoopCheck
lc <- Stream
-> Maybe (TBQueue StreamingChunk) -> Maybe Handle -> IO LoopCheck
newLoopCheck Stream
strm Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing (Handle -> Maybe Handle
forall a. a -> Maybe a
Just Handle
th)
        Server
server Request
request Aux
aux (Config
-> Context
-> LoopCheck
-> Stream
-> Request
-> Response
-> [PushPromise]
-> IO ()
sendResponse Config
conf Context
ctx LoopCheck
lc Stream
strm Request
request)
            IO () -> (SomeException -> IO ()) -> IO ()
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`E.catch` \SomeException
e -> do
                Context -> Stream -> SomeException -> IO ()
resetFailedStream Context
ctx Stream
strm SomeException
e
                SomeException -> IO ()
forall e a. Exception e => e -> IO a
E.throwIO (SomeException
e :: E.SomeException)
        Context -> Stream -> IO ()
doneWithRequestBody Context
ctx Stream
strm
  where
    label :: String
label = String
"H2 response sender for stream " String -> String -> String
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show (Stream -> Int
streamNumber Stream
strm)
    pauseRequestBody :: Handle -> InpObj
pauseRequestBody Handle
th = InpObj
req{inpObjBody = readBody'}
      where
        readBody :: IO (ByteString, Bool)
readBody = InpObj -> IO (ByteString, Bool)
inpObjBody InpObj
req
        readBody' :: IO (ByteString, Bool)
readBody' = do
            Handle -> IO ()
TM.pause Handle
th
            (ByteString, Bool)
bs <- IO (ByteString, Bool)
readBody
            Handle -> IO ()
TM.resume Handle
th -- this is the same as 'tickle'
            (ByteString, Bool) -> IO (ByteString, Bool)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (ByteString, Bool)
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 :: Context -> Stream -> IO ()
doneWithRequestBody Context
ctx strm :: Stream
strm@Stream{Int
MVar (Either SomeException InpObj)
TVar Bool
TVar TxFlow
TVar StreamState
IORef (Maybe ByteString)
IORef (Maybe RxQ)
IORef RxFlow
streamNumber :: Stream -> Int
streamDiscardBody :: Stream -> TVar Bool
streamNumber :: Int
streamState :: TVar StreamState
streamInput :: MVar (Either SomeException InpObj)
streamTxFlow :: TVar TxFlow
streamRxFlow :: IORef RxFlow
streamRxQ :: IORef (Maybe RxQ)
streamRequestMethod :: IORef (Maybe ByteString)
streamDiscardBody :: TVar Bool
streamRequestMethod :: Stream -> IORef (Maybe ByteString)
streamRxQ :: Stream -> IORef (Maybe RxQ)
streamRxFlow :: Stream -> IORef RxFlow
streamTxFlow :: Stream -> TVar TxFlow
streamInput :: Stream -> MVar (Either SomeException InpObj)
streamState :: Stream -> TVar StreamState
..} = do
    STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar Bool -> Bool -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar Bool
streamDiscardBody Bool
True
    Context -> Stream -> IO ()
adjustRxWindow Context
ctx Stream
strm
    Maybe RxQ
mq <- IORef (Maybe RxQ) -> IO (Maybe RxQ)
forall a. IORef a -> IO a
readIORef IORef (Maybe RxQ)
streamRxQ
    Maybe RxQ -> (RxQ -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe RxQ
mq ((RxQ -> IO ()) -> IO ()) -> (RxQ -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \RxQ
q ->
        STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ RxQ -> Either SomeException (ByteString, Bool) -> STM ()
forall a. TQueue a -> a -> STM ()
writeTQueue RxQ
q (Either SomeException (ByteString, Bool) -> STM ())
-> Either SomeException (ByteString, Bool) -> STM ()
forall a b. (a -> b) -> a -> b
$ SomeException -> Either SomeException (ByteString, Bool)
forall a b. a -> Either a b
Left (SomeException -> Either SomeException (ByteString, Bool))
-> SomeException -> Either SomeException (ByteString, Bool)
forall a b. (a -> b) -> a -> b
$ CancelledStream -> SomeException
forall e. Exception e => e -> SomeException
E.toException CancelledStream
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 :: Context -> Stream -> SomeException -> IO ()
resetFailedStream Context
ctx strm :: Stream
strm@Stream{Int
MVar (Either SomeException InpObj)
TVar Bool
TVar TxFlow
TVar StreamState
IORef (Maybe ByteString)
IORef (Maybe RxQ)
IORef RxFlow
streamNumber :: Stream -> Int
streamDiscardBody :: Stream -> TVar Bool
streamRequestMethod :: Stream -> IORef (Maybe ByteString)
streamRxQ :: Stream -> IORef (Maybe RxQ)
streamRxFlow :: Stream -> IORef RxFlow
streamTxFlow :: Stream -> TVar TxFlow
streamInput :: Stream -> MVar (Either SomeException InpObj)
streamState :: Stream -> TVar StreamState
streamNumber :: Int
streamState :: TVar StreamState
streamInput :: MVar (Either SomeException InpObj)
streamTxFlow :: TVar TxFlow
streamRxFlow :: IORef RxFlow
streamRxQ :: IORef (Maybe RxQ)
streamRequestMethod :: IORef (Maybe ByteString)
streamDiscardBody :: TVar Bool
..} SomeException
e = do
    Bool
open <- STM Bool -> IO Bool
forall a. STM a -> IO a
atomically (STM Bool -> IO Bool) -> STM Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ do
        StreamState
st <- TVar StreamState -> STM StreamState
forall a. TVar a -> STM a
readTVar TVar StreamState
streamState
        if StreamState -> Bool
isClosed StreamState
st
            then Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
False
            else do
                Int -> StreamState -> StreamState -> STM ()
informReplaced Int
streamNumber StreamState
st (ClosedCode -> StreamState
Closed ClosedCode
cc)
                TVar StreamState -> StreamState -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar StreamState
streamState (ClosedCode -> StreamState
Closed ClosedCode
cc)
                Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
True
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
open (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        Context -> Stream -> ClosedCode -> IO ()
closed Context
ctx Stream
strm ClosedCode
cc
        ControlQ -> Control -> IO ()
enqueueControl (Context -> ControlQ
controlQ Context
ctx) (Control -> IO ()) -> Control -> IO ()
forall a b. (a -> b) -> a -> b
$
            Maybe SettingsList -> [ByteString] -> Control
CFrames Maybe SettingsList
forall a. Maybe a
Nothing [ErrorCode -> Int -> ByteString
resetFrame ErrorCode
InternalError Int
streamNumber]
        Context -> Stream -> IO ()
giveBackUnread Context
ctx Stream
strm
  where
    cc :: ClosedCode
cc = SomeException -> ClosedCode
ResetByMe SomeException
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 :: Context -> Stream -> Status -> ResponseHeaders -> IO ()
sendInformational Context
ctx Stream
strm Status
st ResponseHeaders
hdrs = do
    LoopCheck
lc <- Stream
-> Maybe (TBQueue StreamingChunk) -> Maybe Handle -> IO LoopCheck
newLoopCheck Stream
strm Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing Maybe Handle
forall a. Maybe a
Nothing
    let hdr :: ResponseHeaders
hdr = (HeaderName
":status", String -> ByteString
C8.pack (Int -> String
forall a. Show a => a -> String
show (Status -> Int
statusCode Status
st))) (HeaderName, ByteString) -> ResponseHeaders -> ResponseHeaders
forall a. a -> [a] -> [a]
: ResponseHeaders
hdrs
    Context -> Stream -> OutputType -> LoopCheck -> IO ()
syncWithSender Context
ctx Stream
strm (ResponseHeaders -> OutputType
OInformational ResponseHeaders
hdr) LoopCheck
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 :: Config
-> Context
-> LoopCheck
-> Stream
-> Request
-> Response
-> [PushPromise]
-> IO ()
sendResponse Config
conf Context
ctx LoopCheck
lc Stream
strm (Request InpObj
req) (Response OutObj
rsp) [PushPromise]
pps = do
    Maybe (IO ())
mwait <- Config
-> Context
-> Stream
-> ValueTable
-> [PushPromise]
-> IO (Maybe (IO ()))
pushStream Config
conf Context
ctx Stream
strm ValueTable
reqvt [PushPromise]
pps
    case Maybe (IO ())
mwait of
        Maybe (IO ())
Nothing -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
        Just IO ()
wait -> IO ()
wait -- all pushes are sent
    Config -> Context -> LoopCheck -> Stream -> OutObj -> IO ()
sendHeaderBody Config
conf Context
ctx LoopCheck
lc Stream
strm OutObj
rsp
  where
    (TokenHeaderList
_, ValueTable
reqvt) = InpObj -> TokenHeaderTable
inpObjHeaders InpObj
req

----------------------------------------------------------------

pushStream
    :: Config
    -> Context
    -> Stream -- parent stream
    -> ValueTable -- request
    -> [PushPromise]
    -> IO (Maybe (IO ()))
pushStream :: Config
-> Context
-> Stream
-> ValueTable
-> [PushPromise]
-> IO (Maybe (IO ()))
pushStream Config
_ Context
_ Stream
_ ValueTable
_ [] = Maybe (IO ()) -> IO (Maybe (IO ()))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe (IO ())
forall a. Maybe a
Nothing
pushStream Config
conf ctx :: Context
ctx@Context{TVar Int
TVar (Maybe Int)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef Int
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
Int -> TokenHeaderTable -> IO ()
resetByMe :: Context -> IORef IntSet
activeRequests :: Context -> TVar Int
peerGoAway :: Context -> TVar (Maybe Int)
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
appDone :: Context -> STM Bool
receiverDone :: Context -> TVar (Maybe SomeException)
timeoutManager :: Context -> Manager
threadManager :: Context -> ThreadManager
peerSockAddr :: Context -> SockAddr
mySockAddr :: Context -> SockAddr
rstRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
settingsRate :: Context -> Rate
pingRate :: Context -> Rate
rxFlow :: Context -> IORef RxFlow
txFlow :: Context -> TVar TxFlow
decodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable :: Context -> DynamicTable
controlQ :: Context -> ControlQ
outputQStreamID :: Context -> TVar Int
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
peerStreamId :: Context -> IORef Int
myStreamId :: Context -> TVar Int
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
peerSettings :: Context -> IORef Settings
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar Int
peerStreamId :: IORef Int
peerLastStreamId :: IORef Int
outputBufferLimit :: IORef Int
outputQ :: TQueue Output
outputQStreamID :: TVar Int
controlQ :: ControlQ
encodeDynamicTable :: DynamicTable
decodeDynamicTable :: DynamicTable
txFlow :: TVar TxFlow
rxFlow :: IORef RxFlow
pingRate :: Rate
settingsRate :: Rate
emptyFrameRate :: Rate
rstRate :: Rate
mySockAddr :: SockAddr
peerSockAddr :: SockAddr
threadManager :: ThreadManager
timeoutManager :: Manager
receiverDone :: TVar (Maybe SomeException)
appDone :: STM Bool
informationalCallback :: Int -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
..} Stream
pstrm ValueTable
reqvt [PushPromise]
pps0
    | Int
len Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0 = Maybe (IO ()) -> IO (Maybe (IO ()))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe (IO ())
forall a. Maybe a
Nothing
    | Bool
otherwise = do
        Bool
pushable <- Settings -> Bool
enablePush (Settings -> Bool) -> IO Settings -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef Settings -> IO Settings
forall a. IORef a -> IO a
readIORef IORef Settings
peerSettings
        if Bool
pushable
            then do
                TVar Int
tvar <- Int -> IO (TVar Int)
forall a. a -> IO (TVar a)
newTVarIO Int
0
                Int
lim <- TVar Int -> [PushPromise] -> Int -> IO Int
forall {a}. Num a => TVar a -> [PushPromise] -> Int -> IO Int
push TVar Int
tvar [PushPromise]
pps0 Int
0
                if Int
lim Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0
                    then Maybe (IO ()) -> IO (Maybe (IO ()))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe (IO ())
forall a. Maybe a
Nothing
                    else Maybe (IO ()) -> IO (Maybe (IO ()))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe (IO ()) -> IO (Maybe (IO ())))
-> Maybe (IO ()) -> IO (Maybe (IO ()))
forall a b. (a -> b) -> a -> b
$ IO () -> Maybe (IO ())
forall a. a -> Maybe a
Just (IO () -> Maybe (IO ())) -> IO () -> Maybe (IO ())
forall a b. (a -> b) -> a -> b
$ Int -> TVar Int -> IO ()
forall {a}. Ord a => a -> TVar a -> IO ()
waiter Int
lim TVar Int
tvar
            else Maybe (IO ()) -> IO (Maybe (IO ()))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe (IO ())
forall a. Maybe a
Nothing
  where
    len :: Int
len = [PushPromise] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [PushPromise]
pps0
    increment :: TVar a -> IO ()
increment TVar a
tvar = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar a -> (a -> a) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' TVar a
tvar (a -> a -> a
forall a. Num a => a -> a -> a
+ a
1)
    -- Checking if all push are done.
    waiter :: a -> TVar a -> IO ()
waiter a
lim TVar a
tvar = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        a
n <- TVar a -> STM a
forall a. TVar a -> STM a
readTVar TVar a
tvar
        Bool -> STM ()
check (a
n a -> a -> Bool
forall a. Ord a => a -> a -> Bool
>= a
lim)
    push :: TVar a -> [PushPromise] -> Int -> IO Int
push TVar a
_ [] Int
n = Int -> IO Int
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Int
n :: Int)
    push TVar a
tvar (PushPromise
pp : [PushPromise]
pps) Int
n = do
        ThreadManager -> String -> IO () -> IO ()
T.forkManaged ThreadManager
threadManager String
"H2 server push" (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
            Maybe (Stream, LoopCheck)
mpushed <- PushPromise -> IO (Maybe (Stream, LoopCheck))
promise PushPromise
pp IO (Maybe (Stream, LoopCheck))
-> IO () -> IO (Maybe (Stream, LoopCheck))
forall a b. IO a -> IO b -> IO a
`E.finally` TVar a -> IO ()
forall {a}. Num a => TVar a -> IO ()
increment TVar a
tvar
            Maybe (Stream, LoopCheck)
-> ((Stream, LoopCheck) -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe (Stream, LoopCheck)
mpushed (((Stream, LoopCheck) -> IO ()) -> IO ())
-> ((Stream, LoopCheck) -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \(Stream
newstrm, LoopCheck
lc) -> do
                let Response OutObj
rsp = PushPromise -> Response
promiseResponse PushPromise
pp
                Config -> Context -> LoopCheck -> Stream -> OutObj -> IO ()
sendHeaderBody Config
conf Context
ctx LoopCheck
lc Stream
newstrm OutObj
rsp
        TVar a -> [PushPromise] -> Int -> IO Int
push TVar a
tvar [PushPromise]
pps (Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
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 :: PushPromise -> IO (Maybe (Stream, LoopCheck))
promise PushPromise
pp = do
        Maybe (Int, Stream)
mstrm <- Context -> Stream -> IO (Maybe (Int, Stream))
makePushStream Context
ctx Stream
pstrm
        Maybe (Int, Stream)
-> ((Int, Stream) -> IO (Stream, LoopCheck))
-> IO (Maybe (Stream, LoopCheck))
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM Maybe (Int, Stream)
mstrm (((Int, Stream) -> IO (Stream, LoopCheck))
 -> IO (Maybe (Stream, LoopCheck)))
-> ((Int, Stream) -> IO (Stream, LoopCheck))
-> IO (Maybe (Stream, LoopCheck))
forall a b. (a -> b) -> a -> b
$ \(Int
pid, Stream
newstrm) -> PushPromise -> Int -> Stream -> IO (Stream, LoopCheck)
promiseOn PushPromise
pp Int
pid Stream
newstrm
    promiseOn :: PushPromise -> Int -> Stream -> IO (Stream, LoopCheck)
promiseOn PushPromise
pp Int
pid Stream
newstrm = do
        let scheme :: ByteString
scheme = Maybe ByteString -> ByteString
forall a. HasCallStack => Maybe a -> a
fromJust (Maybe ByteString -> ByteString) -> Maybe ByteString -> ByteString
forall a b. (a -> b) -> a -> b
$ Token -> ValueTable -> Maybe ByteString
getFieldValue Token
tokenScheme ValueTable
reqvt
            -- fixme: this value can be Nothing
            auth :: ByteString
auth =
                Maybe ByteString -> ByteString
forall a. HasCallStack => Maybe a -> a
fromJust
                    ( Token -> ValueTable -> Maybe ByteString
getFieldValue Token
tokenAuthority ValueTable
reqvt
                        Maybe ByteString -> Maybe ByteString -> Maybe ByteString
forall a. Maybe a -> Maybe a -> Maybe a
forall (f :: * -> *) a. Alternative f => f a -> f a -> f a
<|> Token -> ValueTable -> Maybe ByteString
getFieldValue Token
tokenHost ValueTable
reqvt
                    )
            path :: ByteString
path = PushPromise -> ByteString
promiseRequestPath PushPromise
pp
            promiseRequest :: TokenHeaderList
promiseRequest =
                [ (Token
tokenMethod, ByteString
methodGet)
                , (Token
tokenScheme, ByteString
scheme)
                , (Token
tokenAuthority, ByteString
auth)
                , (Token
tokenPath, ByteString
path)
                ]
            ot :: OutputType
ot = TokenHeaderList -> Int -> OutputType
OPush TokenHeaderList
promiseRequest Int
pid
        LoopCheck
lc <- Stream
-> Maybe (TBQueue StreamingChunk) -> Maybe Handle -> IO LoopCheck
newLoopCheck Stream
newstrm Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing Maybe Handle
forall a. Maybe a
Nothing
        Context -> Stream -> OutputType -> LoopCheck -> IO ()
syncWithSender Context
ctx Stream
newstrm OutputType
ot LoopCheck
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.
        Context -> Stream -> IO ()
halfClosedRemote Context
ctx Stream
newstrm
        (Stream, LoopCheck) -> IO (Stream, LoopCheck)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Stream
newstrm, LoopCheck
lc)

----------------------------------------------------------------

makePushStream :: Context -> Stream -> IO (Maybe (StreamId, Stream))
makePushStream :: Context -> Stream -> IO (Maybe (Int, Stream))
makePushStream Context
ctx Stream
pstrm = do
    -- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: send: respecting peer's limit
    Maybe (Int, Stream)
mstrm <- Context -> IO (Maybe (Int, Stream))
openEvenStreamTry Context
ctx
    let pid :: Int
pid = Stream -> Int
streamNumber Stream
pstrm
    Maybe (Int, Stream) -> IO (Maybe (Int, Stream))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe (Int, Stream) -> IO (Maybe (Int, Stream)))
-> Maybe (Int, Stream) -> IO (Maybe (Int, Stream))
forall a b. (a -> b) -> a -> b
$ (\(Int
_, Stream
newstrm) -> (Int
pid, Stream
newstrm)) ((Int, Stream) -> (Int, Stream))
-> Maybe (Int, Stream) -> Maybe (Int, Stream)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe (Int, Stream)
mstrm

----------------------------------------------------------------

sendHeaderBody
    :: Config
    -> Context
    -> LoopCheck
    -> Stream
    -> OutObj
    -> IO ()
sendHeaderBody :: Config -> Context -> LoopCheck -> Stream -> OutObj -> IO ()
sendHeaderBody Config{Bool
Int
Buffer
SockAddr
Manager
Int -> IO ByteString
Int -> TokenHeaderTable -> IO ()
PositionReadMaker
ByteString -> IO ()
confWriteBuffer :: Buffer
confBufferSize :: Int
confSendAll :: ByteString -> IO ()
confReadN :: Int -> IO ByteString
confPositionReadMaker :: PositionReadMaker
confTimeoutManager :: Manager
confMySockAddr :: SockAddr
confPeerSockAddr :: SockAddr
confReadNTimeout :: Bool
confOnInformational :: Int -> TokenHeaderTable -> IO ()
confOnInformational :: Config -> Int -> TokenHeaderTable -> IO ()
confReadNTimeout :: Config -> Bool
confPeerSockAddr :: Config -> SockAddr
confMySockAddr :: Config -> SockAddr
confTimeoutManager :: Config -> Manager
confPositionReadMaker :: Config -> PositionReadMaker
confReadN :: Config -> Int -> IO ByteString
confSendAll :: Config -> ByteString -> IO ()
confBufferSize :: Config -> Int
confWriteBuffer :: Config -> Buffer
..} Context
ctx LoopCheck
lc Stream
strm OutObj{ResponseHeaders
OutBody
TrailersMaker
outObjHeaders :: ResponseHeaders
outObjBody :: OutBody
outObjTrailers :: TrailersMaker
outObjTrailers :: OutObj -> TrailersMaker
outObjBody :: OutObj -> OutBody
outObjHeaders :: OutObj -> ResponseHeaders
..} = do
    (Maybe DynaNext
mnext, Maybe (TBQueue StreamingChunk)
mtbq) <- case OutBody
outObjBody of
        OutBody
OutBodyNone -> (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe DynaNext
forall a. Maybe a
Nothing, Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing)
        OutBodyFile (FileSpec String
path FileOffset
fileoff FileOffset
bytecount) -> do
            (PositionRead
pread, Sentinel
sentinel) <- PositionReadMaker
confPositionReadMaker String
path
            let next :: DynaNext
next = PositionRead -> FileOffset -> FileOffset -> Sentinel -> DynaNext
fillFileBodyGetNext PositionRead
pread FileOffset
fileoff FileOffset
bytecount Sentinel
sentinel
            (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (DynaNext -> Maybe DynaNext
forall a. a -> Maybe a
Just DynaNext
next, Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing)
        OutBodyBuilder Builder
builder -> do
            let next :: DynaNext
next = Builder -> DynaNext
fillBuilderBodyGetNext Builder
builder
            (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (DynaNext -> Maybe DynaNext
forall a. a -> Maybe a
Just DynaNext
next, Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing)
        OutBodyStreaming (Builder -> IO ()) -> IO () -> IO ()
strmbdy -> do
            TBQueue StreamingChunk
q <- Context
-> Stream -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)
sendStreaming Context
ctx Stream
strm ((OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk))
-> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)
forall a b. (a -> b) -> a -> b
$ \OutBodyIface{IO ()
Maybe SomeException -> IO ()
Builder -> IO ()
forall x. IO x -> IO x
outBodyUnmask :: forall x. IO x -> IO x
outBodyPush :: Builder -> IO ()
outBodyPushFinal :: Builder -> IO ()
outBodyCancel :: Maybe SomeException -> IO ()
outBodyFlush :: IO ()
outBodyFlush :: OutBodyIface -> IO ()
outBodyCancel :: OutBodyIface -> Maybe SomeException -> IO ()
outBodyPushFinal :: OutBodyIface -> Builder -> IO ()
outBodyPush :: OutBodyIface -> Builder -> IO ()
outBodyUnmask :: OutBodyIface -> forall x. IO x -> IO x
..} -> (Builder -> IO ()) -> IO () -> IO ()
strmbdy Builder -> IO ()
outBodyPush IO ()
outBodyFlush
            let next :: DynaNext
next = TBQueue StreamingChunk -> DynaNext
nextForStreaming TBQueue StreamingChunk
q
            (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (DynaNext -> Maybe DynaNext
forall a. a -> Maybe a
Just DynaNext
next, TBQueue StreamingChunk -> Maybe (TBQueue StreamingChunk)
forall a. a -> Maybe a
Just TBQueue StreamingChunk
q)
        OutBodyStreamingIface OutBodyIface -> IO ()
strmbdy -> do
            TBQueue StreamingChunk
q <- Context
-> Stream -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)
sendStreaming Context
ctx Stream
strm OutBodyIface -> IO ()
strmbdy
            let next :: DynaNext
next = TBQueue StreamingChunk -> DynaNext
nextForStreaming TBQueue StreamingChunk
q
            (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (DynaNext -> Maybe DynaNext
forall a. a -> Maybe a
Just DynaNext
next, TBQueue StreamingChunk -> Maybe (TBQueue StreamingChunk)
forall a. a -> Maybe a
Just TBQueue StreamingChunk
q)
    let lc' :: LoopCheck
lc' = LoopCheck
lc{lcTBQ = mtbq}
    Context -> Stream -> OutputType -> LoopCheck -> IO ()
syncWithSender Context
ctx Stream
strm (ResponseHeaders -> Maybe DynaNext -> TrailersMaker -> OutputType
OHeader ResponseHeaders
outObjHeaders Maybe DynaNext
mnext TrailersMaker
outObjTrailers) LoopCheck
lc'

----------------------------------------------------------------

sendStreaming
    :: Context
    -> Stream
    -> (OutBodyIface -> IO ())
    -> IO (TBQueue StreamingChunk)
sendStreaming :: Context
-> Stream -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)
sendStreaming ctx :: Context
ctx@Context{TVar Int
TVar (Maybe Int)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef Int
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
Int -> TokenHeaderTable -> IO ()
resetByMe :: Context -> IORef IntSet
activeRequests :: Context -> TVar Int
peerGoAway :: Context -> TVar (Maybe Int)
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
appDone :: Context -> STM Bool
receiverDone :: Context -> TVar (Maybe SomeException)
timeoutManager :: Context -> Manager
threadManager :: Context -> ThreadManager
peerSockAddr :: Context -> SockAddr
mySockAddr :: Context -> SockAddr
rstRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
settingsRate :: Context -> Rate
pingRate :: Context -> Rate
rxFlow :: Context -> IORef RxFlow
txFlow :: Context -> TVar TxFlow
decodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable :: Context -> DynamicTable
controlQ :: Context -> ControlQ
outputQStreamID :: Context -> TVar Int
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
peerStreamId :: Context -> IORef Int
myStreamId :: Context -> TVar Int
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
peerSettings :: Context -> IORef Settings
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar Int
peerStreamId :: IORef Int
peerLastStreamId :: IORef Int
outputBufferLimit :: IORef Int
outputQ :: TQueue Output
outputQStreamID :: TVar Int
controlQ :: ControlQ
encodeDynamicTable :: DynamicTable
decodeDynamicTable :: DynamicTable
txFlow :: TVar TxFlow
rxFlow :: IORef RxFlow
pingRate :: Rate
settingsRate :: Rate
emptyFrameRate :: Rate
rstRate :: Rate
mySockAddr :: SockAddr
peerSockAddr :: SockAddr
threadManager :: ThreadManager
timeoutManager :: Manager
receiverDone :: TVar (Maybe SomeException)
appDone :: STM Bool
informationalCallback :: Int -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
..} Stream
strm OutBodyIface -> IO ()
strmbdy = do
    TBQueue StreamingChunk
tbq <- Natural -> IO (TBQueue StreamingChunk)
forall a. Natural -> IO (TBQueue a)
newTBQueueIO Natural
10 -- fixme: hard coding: 10
    ThreadManager -> Manager -> String -> (Handle -> IO ()) -> IO ()
T.forkManagedTimeout ThreadManager
threadManager Manager
timeoutManager String
label ((Handle -> IO ()) -> IO ()) -> (Handle -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Handle
th ->
        Context
-> Stream
-> TBQueue StreamingChunk
-> (forall x. IO x -> IO x)
-> (OutBodyIface -> IO ())
-> IO ()
forall r.
Context
-> Stream
-> TBQueue StreamingChunk
-> (forall x. IO x -> IO x)
-> (OutBodyIface -> IO r)
-> IO r
withOutBodyIface Context
ctx Stream
strm TBQueue StreamingChunk
tbq IO a -> IO a
forall a. a -> a
forall x. IO x -> IO x
id ((OutBodyIface -> IO ()) -> IO ())
-> (OutBodyIface -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \OutBodyIface
iface -> do
            let iface' :: OutBodyIface
iface' =
                    OutBodyIface
iface
                        { outBodyPush = \Builder
b -> do
                            Handle -> IO ()
TM.pause Handle
th
                            OutBodyIface -> Builder -> IO ()
outBodyPush OutBodyIface
iface Builder
b
                            Handle -> IO ()
TM.resume Handle
th -- this is the same as 'tickle'
                        , outBodyPushFinal = \Builder
b -> do
                            Handle -> IO ()
TM.pause Handle
th
                            OutBodyIface -> Builder -> IO ()
outBodyPushFinal OutBodyIface
iface Builder
b
                            Handle -> IO ()
TM.resume Handle
th -- this is the same as 'tickle'
                        }
            OutBodyIface -> IO ()
strmbdy OutBodyIface
iface'
    TBQueue StreamingChunk -> IO (TBQueue StreamingChunk)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return TBQueue StreamingChunk
tbq
  where
    label :: String
label = String
"H2 response streaming sender for " String -> String -> String
forall a. [a] -> [a] -> [a]
++ Int -> String
forall a. Show a => a -> String
show (Stream -> Int
streamNumber Stream
strm)