{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}

module Network.HTTP2.H2.Window where

import Control.Concurrent.STM
import qualified Control.Exception as E
import qualified Data.ByteString as BS
import Data.IORef
import Network.Control

import Imports
import Network.HTTP2.Frame
import Network.HTTP2.H2.Context
import Network.HTTP2.H2.EncodeFrame
import Network.HTTP2.H2.Queue
import Network.HTTP2.H2.Types

getStreamWindowSize :: Stream -> IO WindowSize
getStreamWindowSize :: Stream -> IO Int
getStreamWindowSize Stream{TVar TxFlow
streamTxFlow :: TVar TxFlow
streamTxFlow :: Stream -> TVar TxFlow
streamTxFlow} =
    TxFlow -> Int
txWindowSize (TxFlow -> Int) -> IO TxFlow -> IO Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar TxFlow -> IO TxFlow
forall a. TVar a -> IO a
readTVarIO TVar TxFlow
streamTxFlow

getConnectionWindowSize :: Context -> IO WindowSize
getConnectionWindowSize :: Context -> IO Int
getConnectionWindowSize Context{TVar TxFlow
txFlow :: TVar TxFlow
txFlow :: Context -> TVar TxFlow
txFlow} =
    TxFlow -> Int
txWindowSize (TxFlow -> Int) -> IO TxFlow -> IO Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar TxFlow -> IO TxFlow
forall a. TVar a -> IO a
readTVarIO TVar TxFlow
txFlow

waitStreamWindowSize :: Stream -> IO ()
waitStreamWindowSize :: Stream -> IO ()
waitStreamWindowSize Stream{TVar TxFlow
streamTxFlow :: Stream -> TVar TxFlow
streamTxFlow :: TVar TxFlow
streamTxFlow} = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    Int
w <- TxFlow -> Int
txWindowSize (TxFlow -> Int) -> STM TxFlow -> STM Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar TxFlow -> STM TxFlow
forall a. TVar a -> STM a
readTVar TVar TxFlow
streamTxFlow
    Bool -> STM ()
check (Int
w Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0)

connectionWindowSizeSTM :: Context -> STM WindowSize
connectionWindowSizeSTM :: Context -> STM Int
connectionWindowSizeSTM Context{TVar TxFlow
txFlow :: Context -> TVar TxFlow
txFlow :: TVar TxFlow
txFlow} = TxFlow -> Int
txWindowSize (TxFlow -> Int) -> STM TxFlow -> STM Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar TxFlow -> STM TxFlow
forall a. TVar a -> STM a
readTVar TVar TxFlow
txFlow

waitConnectionWindowSize :: Context -> STM ()
waitConnectionWindowSize :: Context -> STM ()
waitConnectionWindowSize Context{TVar TxFlow
txFlow :: Context -> TVar TxFlow
txFlow :: TVar TxFlow
txFlow} = do
    Int
w <- TxFlow -> Int
txWindowSize (TxFlow -> Int) -> STM TxFlow -> STM Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar TxFlow -> STM TxFlow
forall a. TVar a -> STM a
readTVar TVar TxFlow
txFlow
    Bool -> STM ()
check (Int
w Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0)

----------------------------------------------------------------
-- Receiving window update

increaseWindowSize :: StreamId -> TVar TxFlow -> WindowSize -> IO ()
increaseWindowSize :: Int -> TVar TxFlow -> Int -> IO ()
increaseWindowSize Int
sid TVar TxFlow
tvar Int
n = do
    STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar TxFlow -> (TxFlow -> TxFlow) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' TVar TxFlow
tvar ((TxFlow -> TxFlow) -> STM ()) -> (TxFlow -> TxFlow) -> STM ()
forall a b. (a -> b) -> a -> b
$ \TxFlow
flow -> TxFlow
flow{txfLimit = txfLimit flow + n}
    Int
w <- TxFlow -> Int
txWindowSize (TxFlow -> Int) -> IO TxFlow -> IO Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar TxFlow -> IO TxFlow
forall a. TVar a -> IO a
readTVarIO TVar TxFlow
tvar
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int -> Bool
isWindowOverflow Int
w) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        let msg :: ReasonPhrase
msg = [Char] -> ReasonPhrase
forall a. IsString a => [Char] -> a
fromString ([Char]
"window update for stream " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ Int -> [Char]
forall a. Show a => a -> [Char]
show Int
sid [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
" is overflow")
            err :: ErrorCode -> Int -> ReasonPhrase -> HTTP2Error
err =
                if Int -> Bool
isControl Int
sid
                    then ErrorCode -> Int -> ReasonPhrase -> HTTP2Error
ConnectionErrorIsSent
                    else ErrorCode -> Int -> ReasonPhrase -> HTTP2Error
StreamErrorIsSent
        HTTP2Error -> IO ()
forall e a. Exception e => e -> IO a
E.throwIO (HTTP2Error -> IO ()) -> HTTP2Error -> IO ()
forall a b. (a -> b) -> a -> b
$ ErrorCode -> Int -> ReasonPhrase -> HTTP2Error
err ErrorCode
FlowControlError Int
sid ReasonPhrase
msg

increaseStreamWindowSize :: Stream -> WindowSize -> IO ()
increaseStreamWindowSize :: Stream -> Int -> IO ()
increaseStreamWindowSize Stream{Int
streamNumber :: Int
streamNumber :: Stream -> Int
streamNumber, TVar TxFlow
streamTxFlow :: Stream -> TVar TxFlow
streamTxFlow :: TVar TxFlow
streamTxFlow} Int
n =
    Int -> TVar TxFlow -> Int -> IO ()
increaseWindowSize Int
streamNumber TVar TxFlow
streamTxFlow Int
n

increaseConnectionWindowSize :: Context -> Int -> IO ()
increaseConnectionWindowSize :: Context -> Int -> IO ()
increaseConnectionWindowSize Context{TVar TxFlow
txFlow :: Context -> TVar TxFlow
txFlow :: TVar TxFlow
txFlow} Int
n =
    Int -> TVar TxFlow -> Int -> IO ()
increaseWindowSize Int
0 TVar TxFlow
txFlow Int
n

decreaseWindowSize :: Context -> Stream -> WindowSize -> IO ()
decreaseWindowSize :: Context -> Stream -> Int -> IO ()
decreaseWindowSize Context{TVar TxFlow
txFlow :: Context -> TVar TxFlow
txFlow :: TVar TxFlow
txFlow} Stream{TVar TxFlow
streamTxFlow :: Stream -> TVar TxFlow
streamTxFlow :: TVar TxFlow
streamTxFlow} Int
siz = do
    TVar TxFlow -> IO ()
dec TVar TxFlow
txFlow
    TVar TxFlow -> IO ()
dec TVar TxFlow
streamTxFlow
  where
    dec :: TVar TxFlow -> IO ()
dec TVar TxFlow
tvar = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar TxFlow -> (TxFlow -> TxFlow) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' TVar TxFlow
tvar ((TxFlow -> TxFlow) -> STM ()) -> (TxFlow -> TxFlow) -> STM ()
forall a b. (a -> b) -> a -> b
$ \TxFlow
flow -> TxFlow
flow{txfSent = txfSent flow + siz}

----------------------------------------------------------------
-- Sending window update

-- | Whether a stream's DATA is given back to the connection window as it
-- arrives, rather than as it is read.
--
-- So it is for pushes, the only streams of the server's we receive on.  A
-- push is read only if a request for it comes along, and maybe never: held
-- against the connection window until then, pushes that nobody asked for
-- used it up, and the connection stalled, responses to requests and all.
-- Their stream windows still hold back what each can send unread.
connectionCreditedOnArrival :: StreamId -> Bool
connectionCreditedOnArrival :: Int -> Bool
connectionCreditedOnArrival = Int -> Bool
isServerInitiated

informWindowUpdate :: Context -> Stream -> Int -> IO ()
informWindowUpdate :: Context -> Stream -> Int -> IO ()
informWindowUpdate Context
_ Stream
_ Int
0 = () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
informWindowUpdate ctx :: Context
ctx@Context{ControlQ
controlQ :: ControlQ
controlQ :: Context -> ControlQ
controlQ} Stream{Int
streamNumber :: Stream -> Int
streamNumber :: Int
streamNumber, IORef RxFlow
streamRxFlow :: IORef RxFlow
streamRxFlow :: Stream -> IORef RxFlow
streamRxFlow} Int
len = do
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Int -> Bool
connectionCreditedOnArrival Int
streamNumber) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        Context -> Int -> IO ()
giveBackConnectionWindow Context
ctx Int
len
    Maybe Int
mxs <- IORef RxFlow -> (RxFlow -> (RxFlow, Maybe Int)) -> IO (Maybe Int)
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef IORef RxFlow
streamRxFlow ((RxFlow -> (RxFlow, Maybe Int)) -> IO (Maybe Int))
-> (RxFlow -> (RxFlow, Maybe Int)) -> IO (Maybe Int)
forall a b. (a -> b) -> a -> b
$ Int -> FlowControlType -> RxFlow -> (RxFlow, Maybe Int)
maybeOpenRxWindow Int
len FlowControlType
FCTWindowUpdate
    Maybe Int -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe Int
mxs ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
ws -> do
        let frame :: ByteString
frame = Int -> Int -> ByteString
windowUpdateFrame Int
streamNumber Int
ws
            cframe :: Control
cframe = Maybe SettingsList -> [ByteString] -> Control
CFrames Maybe SettingsList
forall a. Maybe a
Nothing [ByteString
frame]
        ControlQ -> Control -> IO ()
enqueueControl ControlQ
controlQ Control
cframe

-- | Account for a DATA frame that is being dropped.
--
-- Its stream is gone -- reset, or closed and forgotten -- so there is no
-- stream window to adjust.  The peer charged these octets against the
-- connection window before sending them, though, and if we say nothing its
-- view of that window shrinks for good; enough dropped frames and the
-- connection stalls with both sides believing the other is at fault.  So
-- charge them and give them straight back.
informIgnoredData :: Context -> StreamId -> Int -> IO ()
informIgnoredData :: Context -> Int -> Int -> IO ()
informIgnoredData Context
_ Int
_ Int
0 = () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
informIgnoredData ctx :: Context
ctx@Context{IORef RxFlow
rxFlow :: IORef RxFlow
rxFlow :: Context -> IORef RxFlow
rxFlow} Int
sid Int
len = do
    Bool
ok <- IORef RxFlow -> (RxFlow -> (RxFlow, Bool)) -> IO Bool
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef RxFlow
rxFlow ((RxFlow -> (RxFlow, Bool)) -> IO Bool)
-> (RxFlow -> (RxFlow, Bool)) -> IO Bool
forall a b. (a -> b) -> a -> b
$ Int -> RxFlow -> (RxFlow, Bool)
checkRxLimit Int
len
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless Bool
ok (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        HTTP2Error -> IO ()
forall e a. Exception e => e -> IO a
E.throwIO (HTTP2Error -> IO ()) -> HTTP2Error -> IO ()
forall a b. (a -> b) -> a -> b
$
            ErrorCode -> Int -> ReasonPhrase -> HTTP2Error
ConnectionErrorIsSent
                ErrorCode
EnhanceYourCalm
                Int
sid
                ReasonPhrase
"exceeds connection flow-control limit"
    Context -> Int -> IO ()
giveBackConnectionWindow Context
ctx Int
len

-- | Give octets already charged to the connection window back to it, and to
-- it alone: for a stream that is closed, whose own window no longer
-- matters.
giveBackConnectionWindow :: Context -> Int -> IO ()
giveBackConnectionWindow :: Context -> Int -> IO ()
giveBackConnectionWindow Context
_ Int
0 = () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
giveBackConnectionWindow Context{ControlQ
controlQ :: Context -> ControlQ
controlQ :: ControlQ
controlQ, IORef RxFlow
rxFlow :: Context -> IORef RxFlow
rxFlow :: IORef RxFlow
rxFlow} Int
len = do
    Maybe Int
mxc <- IORef RxFlow -> (RxFlow -> (RxFlow, Maybe Int)) -> IO (Maybe Int)
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef IORef RxFlow
rxFlow ((RxFlow -> (RxFlow, Maybe Int)) -> IO (Maybe Int))
-> (RxFlow -> (RxFlow, Maybe Int)) -> IO (Maybe Int)
forall a b. (a -> b) -> a -> b
$ Int -> FlowControlType -> RxFlow -> (RxFlow, Maybe Int)
maybeOpenRxWindow Int
len FlowControlType
FCTWindowUpdate
    Maybe Int -> (Int -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe Int
mxc ((Int -> IO ()) -> IO ()) -> (Int -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
ws ->
        ControlQ -> Control -> IO ()
enqueueControl ControlQ
controlQ (Control -> IO ()) -> Control -> IO ()
forall a b. (a -> b) -> a -> b
$ Maybe SettingsList -> [ByteString] -> Control
CFrames Maybe SettingsList
forall a. Maybe a
Nothing [Int -> Int -> ByteString
windowUpdateFrame Int
0 Int
ws]

-- This must be called after an application is finished
-- to adjust RX window.
adjustRxWindow :: Context -> Stream -> IO ()
adjustRxWindow :: Context -> Stream -> IO ()
adjustRxWindow Context
ctx Stream
stream = do
    Int
len <- Stream -> IO Int
takeUnread Stream
stream
    Context -> Stream -> Int -> IO ()
informWindowUpdate Context
ctx Stream
stream Int
len

-- | Like 'adjustRxWindow', for a stream that has been closed: what was
-- left unread goes back to the connection window only.
--
-- Closed first, so that nothing is queued after this has looked: the
-- receiver does not queue DATA for a closed stream ('stream'), but gives
-- it back to the connection window itself.
giveBackUnread :: Context -> Stream -> IO ()
giveBackUnread :: Context -> Stream -> IO ()
giveBackUnread Context
ctx Stream
stream = do
    Int
len <- Stream -> IO Int
takeUnread Stream
stream
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Int -> Bool
connectionCreditedOnArrival (Int -> Bool) -> Int -> Bool
forall a b. (a -> b) -> a -> b
$ Stream -> Int
streamNumber Stream
stream) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        Context -> Int -> IO ()
giveBackConnectionWindow Context
ctx Int
len

-- | Take what is left unread in a stream's queue, and say how many octets
-- of body it was.
takeUnread :: Stream -> IO Int
takeUnread :: Stream -> IO Int
takeUnread Stream{IORef (Maybe RxQ)
streamRxQ :: IORef (Maybe RxQ)
streamRxQ :: Stream -> IORef (Maybe RxQ)
streamRxQ} = do
    Maybe RxQ
mq <- IORef (Maybe RxQ) -> IO (Maybe RxQ)
forall a. IORef a -> IO a
readIORef IORef (Maybe RxQ)
streamRxQ
    case Maybe RxQ
mq of
        Maybe RxQ
Nothing -> Int -> IO Int
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Int
0
        Just RxQ
q -> STM Int -> IO Int
forall a. STM a -> IO a
atomically (STM Int -> IO Int) -> STM Int -> IO Int
forall a b. (a -> b) -> a -> b
$ RxQ -> Int -> STM Int
forall {a} {b}. TQueue (Either a (ByteString, b)) -> Int -> STM Int
loop RxQ
q Int
0
  where
    loop :: TQueue (Either a (ByteString, b)) -> Int -> STM Int
loop TQueue (Either a (ByteString, b))
q !Int
total = do
        Maybe (Either a (ByteString, b))
meb <- TQueue (Either a (ByteString, b))
-> STM (Maybe (Either a (ByteString, b)))
forall a. TQueue a -> STM (Maybe a)
tryReadTQueue TQueue (Either a (ByteString, b))
q
        case Maybe (Either a (ByteString, b))
meb of
            Just (Right (ByteString
bs, b
_)) -> TQueue (Either a (ByteString, b)) -> Int -> STM Int
loop TQueue (Either a (ByteString, b))
q (Int
total Int -> Int -> Int
forall a. Num a => a -> a -> a
+ ByteString -> Int
BS.length ByteString
bs)
            Just le :: Either a (ByteString, b)
le@(Left a
_) -> do
                -- reserving HTTP2Error
                TQueue (Either a (ByteString, b))
-> Either a (ByteString, b) -> STM ()
forall a. TQueue a -> a -> STM ()
writeTQueue TQueue (Either a (ByteString, b))
q Either a (ByteString, b)
le
                Int -> STM Int
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Int
total
            Maybe (Either a (ByteString, b))
_ -> Int -> STM Int
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Int
total