{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RecordWildCards #-}

module Network.HTTP2.H2.Context where

import Control.Concurrent.STM
import qualified Control.Exception as E
import Data.IORef
import qualified Data.IntMap.Strict as IntMap
import qualified Data.IntSet as IntSet
import Network.Control
import Network.Socket (SockAddr)
import qualified System.ThreadManager as T
import qualified System.TimeManager as TM
import System.Timeout (timeout)

import Imports hiding (insert)
import Network.HPACK
import Network.HTTP2.Frame
import Network.HTTP2.H2.Queue
import Network.HTTP2.H2.Settings
import Network.HTTP2.H2.Stream
import Network.HTTP2.H2.StreamTable
import Network.HTTP2.H2.Types

data Role = Client | Server deriving (Role -> Role -> Bool
(Role -> Role -> Bool) -> (Role -> Role -> Bool) -> Eq Role
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Role -> Role -> Bool
== :: Role -> Role -> Bool
$c/= :: Role -> Role -> Bool
/= :: Role -> Role -> Bool
Eq, Int -> Role -> ShowS
[Role] -> ShowS
Role -> String
(Int -> Role -> ShowS)
-> (Role -> String) -> ([Role] -> ShowS) -> Show Role
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Role -> ShowS
showsPrec :: Int -> Role -> ShowS
$cshow :: Role -> String
show :: Role -> String
$cshowList :: [Role] -> ShowS
showList :: [Role] -> ShowS
Show)

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

-- | Telling a connection's threads to stop, and seeing them out before
--   returning to a caller which tears the connection down under them.
--
-- The cleanup comes first: it closes the streams, which is how a thread
-- blocked on a request body is told the peer is gone.  Waiting for the
-- threads before it would be waiting for what the cleanup has to say.
--
-- The wait is capped because one of the threads waited for runs the
-- application's handler: an application which never stops would otherwise
-- keep this from ever returning.  It is a 'E.finally' so that the threads
-- are seen out whatever the cleanup does.
stopAfterAndWait
    :: T.ThreadManager -> IO a -> (Maybe E.SomeException -> IO ()) -> IO a
stopAfterAndWait :: forall a.
ThreadManager -> IO a -> (Maybe SomeException -> IO ()) -> IO a
stopAfterAndWait ThreadManager
mgr IO a
action Maybe SomeException -> IO ()
cleanup =
    ThreadManager -> IO a -> (Maybe SomeException -> IO ()) -> IO a
forall a.
ThreadManager -> IO a -> (Maybe SomeException -> IO ()) -> IO a
T.stopAfter ThreadManager
mgr IO a
action ((Maybe SomeException -> IO ()) -> IO a)
-> (Maybe SomeException -> IO ()) -> IO a
forall a b. (a -> b) -> a -> b
$ \Maybe SomeException
res ->
        Maybe SomeException -> IO ()
cleanup Maybe SomeException
res IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`E.finally` IO (Maybe ()) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Int -> IO () -> IO (Maybe ())
forall a. Int -> IO a -> IO (Maybe a)
timeout Int
stopTimeout (IO () -> IO (Maybe ())) -> IO () -> IO (Maybe ())
forall a b. (a -> b) -> a -> b
$ ThreadManager -> IO ()
T.waitUntilAllGone ThreadManager
mgr)

-- | How long the threads are given to end after they have been told to.
--   Unwinding takes milliseconds; this is for one that does not stop.
stopTimeout :: Int
stopTimeout :: Int
stopTimeout = Int
5000000

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

data RoleInfo = RIS ServerInfo | RIC ClientInfo

type Launch = Context -> Stream -> InpObj -> IO ()

newtype ServerInfo = ServerInfo
    { ServerInfo -> Launch
launch :: Launch
    }

data ClientInfo = ClientInfo
    { ClientInfo -> ByteString
scheme :: ByteString
    , ClientInfo -> String
authority :: Authority
    }

toServerInfo :: RoleInfo -> ServerInfo
toServerInfo :: RoleInfo -> ServerInfo
toServerInfo (RIS ServerInfo
x) = ServerInfo
x
toServerInfo RoleInfo
_ = String -> ServerInfo
forall a. HasCallStack => String -> a
error String
"toServerInfo"

toClientInfo :: RoleInfo -> ClientInfo
toClientInfo :: RoleInfo -> ClientInfo
toClientInfo (RIC ClientInfo
x) = ClientInfo
x
toClientInfo RoleInfo
_ = String -> ClientInfo
forall a. HasCallStack => String -> a
error String
"toClientInfo"

newServerInfo :: Launch -> RoleInfo
newServerInfo :: Launch -> RoleInfo
newServerInfo = ServerInfo -> RoleInfo
RIS (ServerInfo -> RoleInfo)
-> (Launch -> ServerInfo) -> Launch -> RoleInfo
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Launch -> ServerInfo
ServerInfo

newClientInfo :: ByteString -> Authority -> RoleInfo
newClientInfo :: ByteString -> String -> RoleInfo
newClientInfo ByteString
scm String
auth = ClientInfo -> RoleInfo
RIC (ClientInfo -> RoleInfo) -> ClientInfo -> RoleInfo
forall a b. (a -> b) -> a -> b
$ ByteString -> String -> ClientInfo
ClientInfo ByteString
scm String
auth

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

{- FOURMOLU_DISABLE -}
-- | The context for HTTP/2 connection.
data Context = Context
    { Context -> Role
role               :: Role
    , Context -> RoleInfo
roleInfo           :: RoleInfo
    , -- Settings
      Context -> Settings
mySettings         :: Settings
    , Context -> IORef Bool
myFirstSettings    :: IORef Bool
    , Context -> IORef Settings
peerSettings       :: IORef Settings
    , Context -> TVar OddStreamTable
oddStreamTable     :: TVar OddStreamTable
    , Context -> TVar EvenStreamTable
evenStreamTable    :: TVar EvenStreamTable
    , Context -> IORef (Maybe HeaderContinuation)
continued          :: IORef (Maybe HeaderContinuation)
    , Context -> TVar Int
myStreamId         :: TVar StreamId
    , Context -> IORef Int
peerStreamId       :: IORef StreamId
    , Context -> IORef Int
peerLastStreamId   :: IORef StreamId
    , Context -> IORef Int
outputBufferLimit  :: IORef Int
    , Context -> TQueue Output
outputQ            :: TQueue Output
    -- ^ Invariant: Each stream will only ever have at most one 'Output'
    -- object in this queue at any moment.
    , Context -> TVar Int
outputQStreamID    :: TVar StreamId
    , Context -> ControlQ
controlQ           :: ControlQ
    , Context -> DynamicTable
encodeDynamicTable :: DynamicTable
    , Context -> DynamicTable
decodeDynamicTable :: DynamicTable
    , -- the connection window for sending data
      Context -> TVar TxFlow
txFlow             :: TVar TxFlow
    , Context -> IORef RxFlow
rxFlow             :: IORef RxFlow
    , Context -> Rate
pingRate           :: Rate
    , Context -> Rate
settingsRate       :: Rate
    , Context -> Rate
emptyFrameRate     :: Rate
    , Context -> Rate
rstRate            :: Rate
    , Context -> SockAddr
mySockAddr         :: SockAddr
    , Context -> SockAddr
peerSockAddr       :: SockAddr
    , Context -> ThreadManager
threadManager      :: T.ThreadManager
    , Context -> Manager
timeoutManager     :: TM.Manager
    -- ^ For the threads that are timed out: owning a thread and
    --   timing it out are separate, so the manager carries no timer.
    , Context -> TVar (Maybe SomeException)
receiverDone       :: TVar (Maybe E.SomeException)
    , Context -> STM Bool
appDone            :: STM Bool
    -- ^ Server only: set by 'sioDone' once a 'runIO' application has
    --   nothing more to send.  Never for 'run' and clients.
    , Context -> Int -> TokenHeaderTable -> IO ()
informationalCallback :: StreamId -> TokenHeaderTable -> IO ()
    -- ^ Client only: called when a 1xx informational response (e.g. 103 Early
    --   Hints) is received, ahead of the final response. Copied from
    --   'confOnInformational'; no-op by default.
    , Context -> TVar (Maybe Int)
peerGoAway         :: TVar (Maybe StreamId)
    -- ^ The last stream identifier of the peer's GOAWAY(NO_ERROR), once
    --   one has come: see 'goingAway'.
    , Context -> TVar Int
activeRequests     :: TVar Int
    -- ^ Client only: requests whose 'processResponse' has not returned.
    --   A response can be complete, its stream gone from the table, and
    --   its body still being read.
    , Context -> IORef IntSet
resetByMe          :: IORef IntSet.IntSet
    -- ^ Server only: the latest streams of the peer's we have reset, gone
    --   from the stream table: see 'wasResetByMe'.
    }
{- FOURMOLU_ENABLE -}

-- | Header/trailer continuation
--
-- RFC 9113 says "Other frames (from any stream) MUST NOT occur between the
-- HEADERS frame and any CONTINUATION frames that might follow". This is used to
-- implement this requirement.
--
-- It also accumulates the fragments of the block. These are connection-level
-- state: the block must be decoded even if its stream is reset before the
-- block is complete, since it may modify the dynamic table.
data HeaderContinuation = HeaderContinuation
    { HeaderContinuation -> Int
hcStreamId :: StreamId
    , HeaderContinuation -> PartialHeaderBlock
hcBlock :: PartialHeaderBlock
    , HeaderContinuation -> Bool
hcEndOfStream :: Bool
    -- ^ END_STREAM, from the HEADERS frame that started the block
    }

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

{- FOURMOLU_DISABLE -}
newContext
    :: RoleInfo
    -> Config
    -> Int
    -> Int
    -> Settings
    -> TM.Manager
    -> Maybe (STM Bool)
    -> IO Context
newContext :: RoleInfo
-> Config
-> Int
-> Int
-> Settings
-> Manager
-> Maybe (STM Bool)
-> IO Context
newContext RoleInfo
roleInfo Config{Bool
Int
Buffer
SockAddr
Manager
Int -> IO ByteString
Int -> TokenHeaderTable -> IO ()
PositionReadMaker
ByteString -> IO ()
confOnInformational :: Config -> Int -> TokenHeaderTable -> 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 ()
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
..} Int
cacheSiz Int
connRxWS Settings
mySettings Manager
timmgr Maybe (STM Bool)
mdone = do
    -- My: Use this even if ack has not been received yet.
    IORef Bool
myFirstSettings <- Bool -> IO (IORef Bool)
forall a. a -> IO (IORef a)
newIORef Bool
False
    -- Peer: The spec defines max concurrency is infinite unless
    -- SETTINGS_MAX_CONCURRENT_STREAMS is exchanged.
    -- But it is vulnerable, so we set the limitations.
    IORef Settings
peerSettings <-
        Settings -> IO (IORef Settings)
forall a. a -> IO (IORef a)
newIORef Settings
baseSettings{maxConcurrentStreams = Just defaultMaxStreams}
    TVar OddStreamTable
oddStreamTable    <- OddStreamTable -> IO (TVar OddStreamTable)
forall a. a -> IO (TVar a)
newTVarIO OddStreamTable
emptyOddStreamTable
    TVar EvenStreamTable
evenStreamTable   <- EvenStreamTable -> IO (TVar EvenStreamTable)
forall a. a -> IO (TVar a)
newTVarIO (Int -> EvenStreamTable
emptyEvenStreamTable Int
cacheSiz)
    IORef (Maybe HeaderContinuation)
continued         <- Maybe HeaderContinuation -> IO (IORef (Maybe HeaderContinuation))
forall a. a -> IO (IORef a)
newIORef Maybe HeaderContinuation
forall a. Maybe a
Nothing
    TVar Int
myStreamId        <- Int -> IO (TVar Int)
forall a. a -> IO (TVar a)
newTVarIO Int
sid0
    IORef Int
peerStreamId      <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef Int
0
    IORef Int
peerLastStreamId  <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef Int
0
    IORef Int
outputBufferLimit <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef Int
buflim
    TQueue Output
outputQ           <- IO (TQueue Output)
forall a. IO (TQueue a)
newTQueueIO
    TVar Int
outputQStreamID   <- Int -> IO (TVar Int)
forall a. a -> IO (TVar a)
newTVarIO Int
sid0
    ControlQ
controlQ          <- IO ControlQ
newControlQ
    -- My SETTINGS_HEADER_TABLE_SIZE
    DynamicTable
encodeDynamicTable <- Int -> IO DynamicTable
newDynamicTableForEncoding Int
defaultDynamicTableSize
    DynamicTable
decodeDynamicTable <-
        Int -> Int -> IO DynamicTable
newDynamicTableForDecoding (Settings -> Int
headerTableSize Settings
mySettings) Int
4096
    -- Bound the decoded size of a field block: a near-empty field or a
    -- Cookie referenced thousands of times is cheap on the wire but large
    -- once decoded (CVE-2026-49975).  Our own advertised limit is enforced
    -- when one is set, and a default is used otherwise.
    Maybe Int -> DynamicTable -> IO ()
setHeaderListSizeLimit
        (Int -> Maybe Int
forall a. a -> Maybe a
Just (Int -> Maybe Int -> Int
forall a. a -> Maybe a -> a
fromMaybe Int
defaultHeaderListSizeLimit (Settings -> Maybe Int
maxHeaderListSize Settings
mySettings)))
        DynamicTable
decodeDynamicTable
    TVar TxFlow
txFlow          <- TxFlow -> IO (TVar TxFlow)
forall a. a -> IO (TVar a)
newTVarIO (Int -> TxFlow
newTxFlow Int
defaultWindowSize) -- 64K
    IORef RxFlow
rxFlow          <- RxFlow -> IO (IORef RxFlow)
forall a. a -> IO (IORef a)
newIORef (Int -> RxFlow
newRxFlow Int
connRxWS)
    Rate
pingRate        <- IO Rate
newRate
    Rate
settingsRate    <- IO Rate
newRate
    Rate
emptyFrameRate  <- IO Rate
newRate
    Rate
rstRate         <- IO Rate
newRate
    let mySockAddr :: SockAddr
mySockAddr   = SockAddr
confMySockAddr
    let peerSockAddr :: SockAddr
peerSockAddr = SockAddr
confPeerSockAddr
    ThreadManager
threadManager   <- IO ThreadManager
T.newThreadManager
    let timeoutManager :: Manager
timeoutManager = Manager
timmgr
    TVar (Maybe SomeException)
receiverDone    <- Maybe SomeException -> IO (TVar (Maybe SomeException))
forall a. a -> IO (TVar a)
newTVarIO Maybe SomeException
forall a. Maybe a
Nothing
    let informationalCallback :: Int -> TokenHeaderTable -> IO ()
informationalCallback = Int -> TokenHeaderTable -> IO ()
confOnInformational
    let appDone :: STM Bool
appDone = STM Bool -> Maybe (STM Bool) -> STM Bool
forall a. a -> Maybe a -> a
fromMaybe (Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
False) Maybe (STM Bool)
mdone
    TVar (Maybe Int)
peerGoAway      <- Maybe Int -> IO (TVar (Maybe Int))
forall a. a -> IO (TVar a)
newTVarIO Maybe Int
forall a. Maybe a
Nothing
    TVar Int
activeRequests  <- Int -> IO (TVar Int)
forall a. a -> IO (TVar a)
newTVarIO Int
0
    IORef IntSet
resetByMe       <- IntSet -> IO (IORef IntSet)
forall a. a -> IO (IORef a)
newIORef IntSet
IntSet.empty
    Context -> IO Context
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return 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
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)
informationalCallback :: Int -> TokenHeaderTable -> IO ()
appDone :: STM Bool
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
role :: Role
..}
  where
    role :: Role
role = case RoleInfo
roleInfo of
        RIC{} -> Role
Client
        RoleInfo
_ -> Role
Server
    sid0 :: Int
sid0
        | Role
role Role -> Role -> Bool
forall a. Eq a => a -> a -> Bool
== Role
Client = Int
1
        | Bool
otherwise = Int
2
    dlim :: Int
dlim = Int
defaultPayloadLength Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
frameHeaderLength
    buflim :: Int
buflim
        | Int
confBufferSize Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
>= Int
dlim = Int
dlim
        | Bool
otherwise = Int
confBufferSize
{- FOURMOLU_ENABLE -}

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

isClient :: Context -> Bool
isClient :: Context -> Bool
isClient Context
ctx = Context -> Role
role Context
ctx Role -> Role -> Bool
forall a. Eq a => a -> a -> Bool
== Role
Client

isServer :: Context -> Bool
isServer :: Context -> Bool
isServer Context
ctx = Context -> Role
role Context
ctx Role -> Role -> Bool
forall a. Eq a => a -> a -> Bool
== Role
Server

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

getMyNewStreamId :: Context -> STM StreamId
getMyNewStreamId :: Context -> STM Int
getMyNewStreamId 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 :: Context -> Role
roleInfo :: Context -> RoleInfo
mySettings :: Context -> Settings
myFirstSettings :: Context -> IORef Bool
peerSettings :: Context -> IORef Settings
oddStreamTable :: Context -> TVar OddStreamTable
evenStreamTable :: Context -> TVar EvenStreamTable
continued :: Context -> IORef (Maybe HeaderContinuation)
myStreamId :: Context -> TVar Int
peerStreamId :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
outputBufferLimit :: Context -> IORef Int
outputQ :: Context -> TQueue Output
outputQStreamID :: Context -> TVar Int
controlQ :: Context -> ControlQ
encodeDynamicTable :: Context -> DynamicTable
decodeDynamicTable :: Context -> DynamicTable
txFlow :: Context -> TVar TxFlow
rxFlow :: Context -> IORef RxFlow
pingRate :: Context -> Rate
settingsRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
rstRate :: Context -> Rate
mySockAddr :: Context -> SockAddr
peerSockAddr :: Context -> SockAddr
threadManager :: Context -> ThreadManager
timeoutManager :: Context -> Manager
receiverDone :: Context -> TVar (Maybe SomeException)
appDone :: Context -> STM Bool
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
peerGoAway :: Context -> TVar (Maybe Int)
activeRequests :: Context -> TVar Int
resetByMe :: Context -> IORef IntSet
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
..} = do
    Int
n <- TVar Int -> STM Int
forall a. TVar a -> STM a
readTVar TVar Int
myStreamId
    let n' :: Int
n' = Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
2
    TVar Int -> Int -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar Int
myStreamId Int
n'
    Int -> STM Int
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Int
n

getPeerStreamID :: Context -> IO StreamId
getPeerStreamID :: Context -> IO Int
getPeerStreamID Context
ctx = IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef (IORef Int -> IO Int) -> IORef Int -> IO Int
forall a b. (a -> b) -> a -> b
$ Context -> IORef Int
peerStreamId Context
ctx

setPeerStreamID :: Context -> StreamId -> IO ()
setPeerStreamID :: Context -> Int -> IO ()
setPeerStreamID Context
ctx Int
sid = IORef Int -> Int -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef (Context -> IORef Int
peerStreamId Context
ctx) Int
sid

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

getPeerLastStreamId :: Context -> IO StreamId
getPeerLastStreamId :: Context -> IO Int
getPeerLastStreamId Context
ctx = IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef (IORef Int -> IO Int) -> IORef Int -> IO Int
forall a b. (a -> b) -> a -> b
$ Context -> IORef Int
peerLastStreamId Context
ctx

modifyPeerLastStreamId :: Context -> StreamId -> IO ()
modifyPeerLastStreamId :: Context -> Int -> IO ()
modifyPeerLastStreamId Context
ctx Int
sid = IORef Int -> (Int -> (Int, ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' (Context -> IORef Int
peerLastStreamId Context
ctx) ((Int -> (Int, ())) -> IO ()) -> (Int -> (Int, ())) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Int
n -> if Int
sid Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
n then (Int
sid, ()) else (Int
n, ())

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

{-# INLINE setStreamState #-}
setStreamState :: Context -> Stream -> StreamState -> IO ()
setStreamState :: Context -> Stream -> StreamState -> IO ()
setStreamState Context
_ Stream{Int
streamNumber :: Int
streamNumber :: Stream -> Int
streamNumber, TVar StreamState
streamState :: TVar StreamState
streamState :: Stream -> TVar StreamState
streamState} StreamState
newState = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    StreamState
oldState <- TVar StreamState -> STM StreamState
forall a. TVar a -> STM a
readTVar TVar StreamState
streamState
    Int -> StreamState -> StreamState -> STM ()
informReplaced Int
streamNumber StreamState
oldState StreamState
newState
    TVar StreamState -> StreamState -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar StreamState
streamState StreamState
newState

-- | Replacing the open state of a stream as the receiver moves it on, from
-- headers to body.
--
-- The receiver reads a stream's state, works out the next one from the
-- frame, and writes it back -- in a transaction of its own.  In between, the
-- sender may have half-closed the stream on our side ('halfClosedLocal',
-- which records it as @Open (Just cc) _@) or closed it.  Writing the whole
-- state back undid that: the half-close was lost, the peer's END_STREAM then
-- took the stream to half-closed (remote) rather than closed, and it stayed
-- in the stream table, holding its concurrency slot for good.  With both ends
-- streaming at once -- gRPC-style -- a client ran out of streams and a
-- server refused every new one.
--
-- So only the open state is replaced, keeping whatever the sender recorded
-- about our side, and a stream that is no longer open is left alone.
setOpenState :: Context -> Stream -> OpenState -> IO ()
setOpenState :: Context -> Stream -> OpenState -> IO ()
setOpenState Context
_ Stream{Int
streamNumber :: Stream -> Int
streamNumber :: Int
streamNumber, TVar StreamState
streamState :: Stream -> TVar StreamState
streamState :: TVar StreamState
streamState} OpenState
o = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    StreamState
oldState <- TVar StreamState -> STM StreamState
forall a. TVar a -> STM a
readTVar TVar StreamState
streamState
    case StreamState
oldState of
        Open Maybe ClosedCode
hcl OpenState
_ -> do
            let newState :: StreamState
newState = Maybe ClosedCode -> OpenState -> StreamState
Open Maybe ClosedCode
hcl OpenState
o
            Int -> StreamState -> StreamState -> STM ()
informReplaced Int
streamNumber StreamState
oldState StreamState
newState
            TVar StreamState -> StreamState -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar StreamState
streamState StreamState
newState
        StreamState
_otherwise -> () -> STM ()
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return ()

-- | Inform consumers of any streams that we close
informReplaced :: StreamId -> StreamState -> StreamState -> STM ()
informReplaced :: Int -> StreamState -> StreamState -> STM ()
informReplaced Int
streamNumber StreamState
oldState StreamState
newState =
    case (StreamState
oldState, StreamState
newState) of
        (Open Maybe ClosedCode
_ (Body TQueue (Either SomeException (ByteString, Bool))
q Maybe Int
_ IORef Int
_ IORef (Maybe TokenHeaderTable)
_), Open Maybe ClosedCode
_ (Body TQueue (Either SomeException (ByteString, Bool))
q' Maybe Int
_ IORef Int
_ IORef (Maybe TokenHeaderTable)
_))
            | TQueue (Either SomeException (ByteString, Bool))
q TQueue (Either SomeException (ByteString, Bool))
-> TQueue (Either SomeException (ByteString, Bool)) -> Bool
forall a. Eq a => a -> a -> Bool
== TQueue (Either SomeException (ByteString, Bool))
q' ->
                -- The stream stays open with the same body; nothing to do
                () -> STM ()
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
        (Open Maybe ClosedCode
_ (Body TQueue (Either SomeException (ByteString, Bool))
q Maybe Int
_ IORef Int
_ IORef (Maybe TokenHeaderTable)
_), Closed ClosedCode
cc) ->
            TQueue (Either SomeException (ByteString, Bool))
-> Either SomeException (ByteString, Bool) -> STM ()
forall a. TQueue a -> a -> STM ()
writeTQueue TQueue (Either SomeException (ByteString, Bool))
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
$ HTTP2Error -> SomeException
forall e. Exception e => e -> SomeException
E.toException (HTTP2Error -> SomeException) -> HTTP2Error -> SomeException
forall a b. (a -> b) -> a -> b
$ Int -> ClosedCode -> HTTP2Error
closedCodeToError Int
streamNumber ClosedCode
cc
        (Open Maybe ClosedCode
_ (Body TQueue (Either SomeException (ByteString, Bool))
q Maybe Int
_ IORef Int
_ IORef (Maybe TokenHeaderTable)
_), StreamState
_) ->
            -- The stream is opened with a /new/ body
            TQueue (Either SomeException (ByteString, Bool))
-> Either SomeException (ByteString, Bool) -> STM ()
forall a. TQueue a -> a -> STM ()
writeTQueue TQueue (Either SomeException (ByteString, Bool))
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
$ HTTP2Error -> SomeException
forall e. Exception e => e -> SomeException
E.toException HTTP2Error
ConnectionIsClosed
        (StreamState, StreamState)
_otherwise ->
            -- The stream wasn't open to start with; nothing to do
            () -> STM ()
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return ()

-- | Opening an idle stream.
--
-- Only an idle one: the receiver checks that the stream is idle and then
-- opens it, and a client's request stream stays idle while the request is
-- being sent -- so by the time it opens the stream for the response's
-- HEADERS, the sender may already have half-closed it ('halfClosedLocal'
-- turns an idle stream into @Open (Just cc) JustOpened@).  Opening it over
-- that lost the half-close, with the same result as described at
-- 'setOpenState'.
opened :: Context -> Stream -> IO ()
opened :: Context -> Stream -> IO ()
opened Context
_ Stream{TVar StreamState
streamState :: Stream -> TVar StreamState
streamState :: TVar StreamState
streamState} = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar StreamState -> (StreamState -> StreamState) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' TVar StreamState
streamState StreamState -> StreamState
open
  where
    open :: StreamState -> StreamState
open StreamState
Idle = Maybe ClosedCode -> OpenState -> StreamState
Open Maybe ClosedCode
forall a. Maybe a
Nothing OpenState
JustOpened
    open StreamState
st = StreamState
st

halfClosedRemote :: Context -> Stream -> IO ()
halfClosedRemote :: Context -> Stream -> IO ()
halfClosedRemote Context
ctx stream :: Stream
stream@Stream{TVar StreamState
streamState :: Stream -> TVar StreamState
streamState :: TVar StreamState
streamState} = do
    Maybe ClosedCode
closingCode <- STM (Maybe ClosedCode) -> IO (Maybe ClosedCode)
forall a. STM a -> IO a
atomically (STM (Maybe ClosedCode) -> IO (Maybe ClosedCode))
-> STM (Maybe ClosedCode) -> IO (Maybe ClosedCode)
forall a b. (a -> b) -> a -> b
$ TVar StreamState
-> (StreamState -> (Maybe ClosedCode, StreamState))
-> STM (Maybe ClosedCode)
forall s a. TVar s -> (s -> (a, s)) -> STM a
stateTVar TVar StreamState
streamState StreamState -> (Maybe ClosedCode, StreamState)
closeHalf
    (ClosedCode -> IO ()) -> Maybe ClosedCode -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Context -> Stream -> ClosedCode -> IO ()
closed Context
ctx Stream
stream) Maybe ClosedCode
closingCode
  where
    closeHalf :: StreamState -> (Maybe ClosedCode, StreamState)
    closeHalf :: StreamState -> (Maybe ClosedCode, StreamState)
closeHalf x :: StreamState
x@(Closed ClosedCode
_) = (Maybe ClosedCode
forall a. Maybe a
Nothing, StreamState
x)
    closeHalf (Open (Just ClosedCode
cc) OpenState
_) = (ClosedCode -> Maybe ClosedCode
forall a. a -> Maybe a
Just ClosedCode
cc, ClosedCode -> StreamState
Closed ClosedCode
cc)
    closeHalf StreamState
_ = (Maybe ClosedCode
forall a. Maybe a
Nothing, StreamState
HalfClosedRemote)

halfClosedLocal :: Context -> Stream -> ClosedCode -> IO ()
halfClosedLocal :: Context -> Stream -> ClosedCode -> IO ()
halfClosedLocal Context
ctx stream :: Stream
stream@Stream{TVar StreamState
streamState :: Stream -> TVar StreamState
streamState :: TVar StreamState
streamState} ClosedCode
cc = do
    Bool
shouldFinalize <- 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
$ TVar StreamState
-> (StreamState -> (Bool, StreamState)) -> STM Bool
forall s a. TVar s -> (s -> (a, s)) -> STM a
stateTVar TVar StreamState
streamState StreamState -> (Bool, StreamState)
closeHalf
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
shouldFinalize (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        Context -> Stream -> ClosedCode -> IO ()
closed Context
ctx Stream
stream ClosedCode
cc
  where
    closeHalf :: StreamState -> (Bool, StreamState)
    closeHalf :: StreamState -> (Bool, StreamState)
closeHalf x :: StreamState
x@(Closed ClosedCode
_) = (Bool
False, StreamState
x)
    closeHalf StreamState
HalfClosedRemote = (Bool
True, ClosedCode -> StreamState
Closed ClosedCode
cc)
    closeHalf (Open Maybe ClosedCode
Nothing OpenState
o) = (Bool
False, Maybe ClosedCode -> OpenState -> StreamState
Open (ClosedCode -> Maybe ClosedCode
forall a. a -> Maybe a
Just ClosedCode
cc) OpenState
o)
    closeHalf StreamState
_ = (Bool
False, Maybe ClosedCode -> OpenState -> StreamState
Open (ClosedCode -> Maybe ClosedCode
forall a. a -> Maybe a
Just ClosedCode
cc) OpenState
JustOpened)

-- | Closing a stream whose response is still coming in, and saying
-- whether it was.
--
-- Decided and done in one transaction, so that a stream the peer finishes
-- meanwhile is not taken for one still open.  A response whose END_STREAM
-- the receiver has already queued, and not yet recorded in the state, can
-- still be taken for one coming in; the reset that follows is one RFC 9113
-- section 5.1 has the peer ignore ("for a short period after a DATA or
-- HEADERS frame containing an END_STREAM flag is sent").
closeIfReceiving :: Context -> Stream -> ClosedCode -> IO Bool
closeIfReceiving :: Context -> Stream -> ClosedCode -> IO Bool
closeIfReceiving Context
ctx strm :: Stream
strm@Stream{Int
streamNumber :: Stream -> Int
streamNumber :: Int
streamNumber, TVar StreamState
streamState :: Stream -> TVar StreamState
streamState :: TVar StreamState
streamState} ClosedCode
cc = do
    Bool
receiving <- 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
        case StreamState
st of
            -- END_STREAM came with the headers.
            Open Maybe ClosedCode
_ (NoBody TokenHeaderTable
_) -> Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
False
            Open{} -> 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
            StreamState
_otherwise -> Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
False
    -- Out of the stream table, giving its concurrency slot back.
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
receiving (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Context -> Stream -> ClosedCode -> IO ()
closed Context
ctx Stream
strm ClosedCode
cc
    Bool -> IO Bool
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
receiving

closed :: Context -> Stream -> ClosedCode -> IO ()
closed :: Context -> Stream -> ClosedCode -> IO ()
closed ctx :: Context
ctx@Context{TVar OddStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
oddStreamTable :: TVar OddStreamTable
oddStreamTable, TVar EvenStreamTable
evenStreamTable :: Context -> TVar EvenStreamTable
evenStreamTable :: TVar EvenStreamTable
evenStreamTable} strm :: Stream
strm@Stream{Int
streamNumber :: Stream -> Int
streamNumber :: Int
streamNumber} ClosedCode
cc = do
    if Int -> Bool
isServerInitiated Int
streamNumber
        then TVar EvenStreamTable -> Int -> SomeException -> IO ()
deleteEven TVar EvenStreamTable
evenStreamTable Int
streamNumber SomeException
err
        else TVar OddStreamTable -> Int -> SomeException -> IO ()
deleteOdd TVar OddStreamTable
oddStreamTable Int
streamNumber SomeException
err
    Context -> Stream -> StreamState -> IO ()
setStreamState Context
ctx Stream
strm (ClosedCode -> StreamState
Closed ClosedCode
cc) -- anyway
    case ClosedCode
cc of
        ResetByMe SomeException
_
            | Context -> Bool
isServer Context
ctx Bool -> Bool -> Bool
&& Bool -> Bool
not (Int -> Bool
isServerInitiated Int
streamNumber) ->
                Context -> Int -> IO ()
rememberReset Context
ctx Int
streamNumber
        ClosedCode
_otherwise -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
  where
    err :: E.SomeException
    err :: SomeException
err = HTTP2Error -> SomeException
forall e. Exception e => e -> SomeException
E.toException (Int -> ClosedCode -> HTTP2Error
closedCodeToError Int
streamNumber ClosedCode
cc)

-- | Remember that we reset a stream of the peer's.
--
-- RFC 9113, section 5.1: "An endpoint MUST ignore frames that it receives
-- on closed streams after it has sent a RST_STREAM frame."  Trailers among
-- them -- the peer sent them before our reset reached it -- but once the
-- stream is gone from the table, HEADERS on it look like the reuse of a
-- stream identifier, a connection error.  So the latest streams we reset
-- are remembered, 'resetByMeLimit' of them, the oldest forgotten first.
rememberReset :: Context -> StreamId -> IO ()
rememberReset :: Context -> Int -> IO ()
rememberReset Context{IORef IntSet
resetByMe :: Context -> IORef IntSet
resetByMe :: IORef IntSet
resetByMe} Int
sid =
    IORef IntSet -> (IntSet -> (IntSet, ())) -> IO ()
forall a b. IORef a -> (a -> (a, b)) -> IO b
atomicModifyIORef' IORef IntSet
resetByMe ((IntSet -> (IntSet, ())) -> IO ())
-> (IntSet -> (IntSet, ())) -> IO ()
forall a b. (a -> b) -> a -> b
$ \IntSet
s ->
        let s' :: IntSet
s' = Int -> IntSet -> IntSet
IntSet.insert Int
sid IntSet
s
         in (if IntSet -> Int
IntSet.size IntSet
s' Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
resetByMeLimit then IntSet -> IntSet
IntSet.deleteMin IntSet
s' else IntSet
s', ())

-- | Whether we reset a stream of the peer's lately: see 'rememberReset'.
wasResetByMe :: Context -> StreamId -> IO Bool
wasResetByMe :: Context -> Int -> IO Bool
wasResetByMe Context{IORef IntSet
resetByMe :: Context -> IORef IntSet
resetByMe :: IORef IntSet
resetByMe} Int
sid = Int -> IntSet -> Bool
IntSet.member Int
sid (IntSet -> Bool) -> IO IntSet -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef IntSet -> IO IntSet
forall a. IORef a -> IO a
readIORef IORef IntSet
resetByMe

resetByMeLimit :: Int
resetByMeLimit :: Int
resetByMeLimit = Int
1000

----------------------------------------------------------------
-- From peer

-- Server
--
-- Note that this does not apply SETTINGS_MAX_CONCURRENT_STREAMS.  A stream
-- over the limit still has to be admitted this far, because its field block
-- has to be decoded before it can be refused; 'checkOddConcurrency' does the
-- refusing once that has happened.
openOddStreamCheck :: Context -> StreamId -> FrameType -> IO Stream
openOddStreamCheck :: Context -> Int -> FrameType -> IO Stream
openOddStreamCheck ctx :: Context
ctx@Context{TVar OddStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
oddStreamTable :: TVar OddStreamTable
oddStreamTable, IORef Settings
peerSettings :: Context -> IORef Settings
peerSettings :: IORef Settings
peerSettings, Settings
mySettings :: Context -> Settings
mySettings :: Settings
mySettings} Int
sid FrameType
ftyp = do
    Int
txws <- Settings -> Int
initialWindowSize (Settings -> Int) -> IO Settings -> IO Int
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
    let rxws :: Int
rxws = Settings -> Int
initialWindowSize Settings
mySettings
    Stream
newstrm <- Int -> Int -> Int -> IO Stream
newOddStream Int
sid Int
txws Int
rxws
    Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (FrameType
ftyp FrameType -> FrameType -> Bool
forall a. Eq a => a -> a -> Bool
== FrameType
FrameHeaders Bool -> Bool -> Bool
|| FrameType
ftyp FrameType -> FrameType -> Bool
forall a. Eq a => a -> a -> Bool
== FrameType
FramePushPromise) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Context -> Stream -> IO ()
opened Context
ctx Stream
newstrm
    TVar OddStreamTable -> Int -> Stream -> IO ()
insertOdd TVar OddStreamTable
oddStreamTable Int
sid Stream
newstrm
    Stream -> IO Stream
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Stream
newstrm

-- Client
openEvenStreamCacheCheck :: Context -> StreamId -> Method -> ByteString -> IO ()
openEvenStreamCacheCheck :: Context -> Int -> ByteString -> ByteString -> IO ()
openEvenStreamCacheCheck Context{TVar EvenStreamTable
evenStreamTable :: Context -> TVar EvenStreamTable
evenStreamTable :: TVar EvenStreamTable
evenStreamTable, IORef Settings
peerSettings :: Context -> IORef Settings
peerSettings :: IORef Settings
peerSettings, Settings
mySettings :: Context -> Settings
mySettings :: Settings
mySettings} Int
sid ByteString
method ByteString
path = do
    -- My SETTINGS_MAX_CONCURRENT_STREAMS
    Int
conc <- TVar EvenStreamTable -> IO Int
getEvenConcurrency TVar EvenStreamTable
evenStreamTable
    Int -> Settings -> Int -> IO ()
checkMyConcurrency Int
sid Settings
mySettings (Int
conc Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)
    Int
txws <- Settings -> Int
initialWindowSize (Settings -> Int) -> IO Settings -> IO Int
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
    let rxws :: Int
rxws = Settings -> Int
initialWindowSize Settings
mySettings
    Stream
newstrm <- Int -> Int -> Int -> IO Stream
newEvenStream Int
sid Int
txws Int
rxws
    TVar EvenStreamTable -> ByteString -> ByteString -> Stream -> IO ()
insertEvenCache TVar EvenStreamTable
evenStreamTable ByteString
method ByteString
path Stream
newstrm

-- | Refuse a peer-initiated stream that puts us over the limit we advertised
-- in SETTINGS_MAX_CONCURRENT_STREAMS.
--
-- Checked once the stream's field block has been decoded, rather than when
-- its HEADERS frame arrived.  A block has to be decoded whatever becomes of
-- its stream -- RFC 9113 section 10.5.1, "The field block MUST be processed
-- to ensure a consistent connection state" -- and refusing at arrival meant
-- throwing before the frame's payload had even been read, which left nothing
-- to do but drop the connection.  From here the throw lands inside the
-- receiver's per-frame reset handler, so the answer is
-- RST_STREAM(REFUSED_STREAM) and the connection carries on, which is what
-- section 5.1.2 asks for and what section 8.7 lets the peer retry against.
--
-- The stream is in the table by the time we get here, so it counts itself.
checkOddConcurrency :: Context -> StreamId -> IO ()
checkOddConcurrency :: Context -> Int -> IO ()
checkOddConcurrency Context{TVar OddStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
oddStreamTable :: TVar OddStreamTable
oddStreamTable, Settings
mySettings :: Context -> Settings
mySettings :: Settings
mySettings} Int
sid = do
    Int
conc <- TVar OddStreamTable -> IO Int
getOddConcurrency TVar OddStreamTable
oddStreamTable
    Int -> Settings -> Int -> IO ()
checkMyConcurrency Int
sid Settings
mySettings Int
conc

checkMyConcurrency
    :: StreamId -> Settings -> Int -> IO ()
checkMyConcurrency :: Int -> Settings -> Int -> IO ()
checkMyConcurrency Int
sid Settings
settings Int
conc = do
    let mMaxConc :: Maybe Int
mMaxConc = Settings -> Maybe Int
maxConcurrentStreams Settings
settings
    case Maybe Int
mMaxConc of
        Maybe Int
Nothing -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
        Just Int
maxConc ->
            Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
conc Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
maxConc) (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
StreamErrorIsSent ErrorCode
RefusedStream Int
sid ReasonPhrase
"exceeds max concurrent"

----------------------------------------------------------------
-- From me

-- Clinet
openOddStreamWait :: Context -> IO (StreamId, Stream)
openOddStreamWait :: Context -> IO (Int, Stream)
openOddStreamWait ctx :: Context
ctx@Context{TVar OddStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
oddStreamTable :: TVar OddStreamTable
oddStreamTable, Settings
mySettings :: Context -> Settings
mySettings :: Settings
mySettings, IORef Settings
peerSettings :: Context -> IORef Settings
peerSettings :: IORef Settings
peerSettings} = do
    -- Peer SETTINGS_MAX_CONCURRENT_STREAMS
    Maybe Int
mMaxConc <- Settings -> Maybe Int
maxConcurrentStreams (Settings -> Maybe Int) -> IO Settings -> IO (Maybe Int)
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
    let rxws :: Int
rxws = Settings -> Int
initialWindowSize Settings
mySettings
    case Maybe Int
mMaxConc of
        Maybe Int
Nothing -> do
            Int
sid <- 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
$ do
                Context -> STM ()
refuseIfGoingAway Context
ctx
                Context -> STM Int
getMyNewStreamId Context
ctx
            Int
txws <- Settings -> Int
initialWindowSize (Settings -> Int) -> IO Settings -> IO Int
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
            Stream
newstrm <- Int -> Int -> Int -> IO Stream
newOddStream Int
sid Int
txws Int
rxws
            TVar OddStreamTable -> Int -> Stream -> IO ()
insertOdd TVar OddStreamTable
oddStreamTable Int
sid Stream
newstrm
            (Int, Stream) -> IO (Int, Stream)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Int
sid, Stream
newstrm)
        Just Int
maxConc -> do
            Int
sid <- 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
$ do
                Context -> STM ()
refuseIfGoingAway Context
ctx
                TVar OddStreamTable -> Int -> STM ()
waitIncOdd TVar OddStreamTable
oddStreamTable Int
maxConc
                Context -> STM Int
getMyNewStreamId Context
ctx
            Int
txws <- Settings -> Int
initialWindowSize (Settings -> Int) -> IO Settings -> IO Int
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
            Stream
newstrm <- Int -> Int -> Int -> IO Stream
newOddStream Int
sid Int
txws Int
rxws
            TVar OddStreamTable -> Int -> Stream -> IO ()
insertOdd' TVar OddStreamTable
oddStreamTable Int
sid Stream
newstrm
            (Int, Stream) -> IO (Int, Stream)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Int
sid, Stream
newstrm)

-- Server

-- | Opening a stream for a push, if the peer's
-- SETTINGS_MAX_CONCURRENT_STREAMS leaves room for one.
--
-- Not waiting for room: the response the push belongs to waits for its
-- PUSH_PROMISE to go out, and with no room -- a peer can announce 0 to
-- refuse pushes (RFC 9113, section 8.4) -- it waited for ever.  A push is
-- only ever an offer, so one there is no room for is not made.
openEvenStreamTry :: Context -> IO (Maybe (StreamId, Stream))
openEvenStreamTry :: Context -> IO (Maybe (Int, Stream))
openEvenStreamTry 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 :: Context -> Role
roleInfo :: Context -> RoleInfo
mySettings :: Context -> Settings
myFirstSettings :: Context -> IORef Bool
peerSettings :: Context -> IORef Settings
oddStreamTable :: Context -> TVar OddStreamTable
evenStreamTable :: Context -> TVar EvenStreamTable
continued :: Context -> IORef (Maybe HeaderContinuation)
myStreamId :: Context -> TVar Int
peerStreamId :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
outputBufferLimit :: Context -> IORef Int
outputQ :: Context -> TQueue Output
outputQStreamID :: Context -> TVar Int
controlQ :: Context -> ControlQ
encodeDynamicTable :: Context -> DynamicTable
decodeDynamicTable :: Context -> DynamicTable
txFlow :: Context -> TVar TxFlow
rxFlow :: Context -> IORef RxFlow
pingRate :: Context -> Rate
settingsRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
rstRate :: Context -> Rate
mySockAddr :: Context -> SockAddr
peerSockAddr :: Context -> SockAddr
threadManager :: Context -> ThreadManager
timeoutManager :: Context -> Manager
receiverDone :: Context -> TVar (Maybe SomeException)
appDone :: Context -> STM Bool
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
peerGoAway :: Context -> TVar (Maybe Int)
activeRequests :: Context -> TVar Int
resetByMe :: Context -> IORef IntSet
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
..} = do
    Bool
goingaway <- Maybe Int -> Bool
forall a. Maybe a -> Bool
isJust (Maybe Int -> Bool) -> IO (Maybe Int) -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar (Maybe Int) -> IO (Maybe Int)
forall a. TVar a -> IO a
readTVarIO TVar (Maybe Int)
peerGoAway
    if Bool
goingaway then Maybe (Int, Stream) -> IO (Maybe (Int, Stream))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe (Int, Stream)
forall a. Maybe a
Nothing else Context -> IO (Maybe (Int, Stream))
openEvenStreamTry' Context
ctx

openEvenStreamTry' :: Context -> IO (Maybe (StreamId, Stream))
openEvenStreamTry' :: Context -> IO (Maybe (Int, Stream))
openEvenStreamTry' 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 :: Context -> Role
roleInfo :: Context -> RoleInfo
mySettings :: Context -> Settings
myFirstSettings :: Context -> IORef Bool
peerSettings :: Context -> IORef Settings
oddStreamTable :: Context -> TVar OddStreamTable
evenStreamTable :: Context -> TVar EvenStreamTable
continued :: Context -> IORef (Maybe HeaderContinuation)
myStreamId :: Context -> TVar Int
peerStreamId :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
outputBufferLimit :: Context -> IORef Int
outputQ :: Context -> TQueue Output
outputQStreamID :: Context -> TVar Int
controlQ :: Context -> ControlQ
encodeDynamicTable :: Context -> DynamicTable
decodeDynamicTable :: Context -> DynamicTable
txFlow :: Context -> TVar TxFlow
rxFlow :: Context -> IORef RxFlow
pingRate :: Context -> Rate
settingsRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
rstRate :: Context -> Rate
mySockAddr :: Context -> SockAddr
peerSockAddr :: Context -> SockAddr
threadManager :: Context -> ThreadManager
timeoutManager :: Context -> Manager
receiverDone :: Context -> TVar (Maybe SomeException)
appDone :: Context -> STM Bool
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
peerGoAway :: Context -> TVar (Maybe Int)
activeRequests :: Context -> TVar Int
resetByMe :: Context -> IORef IntSet
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
..} = do
    -- Peer SETTINGS_MAX_CONCURRENT_STREAMS
    Maybe Int
mMaxConc <- Settings -> Maybe Int
maxConcurrentStreams (Settings -> Maybe Int) -> IO Settings -> IO (Maybe Int)
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
    let rxws :: Int
rxws = Settings -> Int
initialWindowSize Settings
mySettings
    case Maybe Int
mMaxConc of
        Maybe Int
Nothing -> do
            Int
sid <- 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
$ Context -> STM Int
getMyNewStreamId Context
ctx
            Int
txws <- Settings -> Int
initialWindowSize (Settings -> Int) -> IO Settings -> IO Int
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
            Stream
newstrm <- Int -> Int -> Int -> IO Stream
newEvenStream Int
sid Int
txws Int
rxws
            TVar EvenStreamTable -> Int -> Stream -> IO ()
insertEven TVar EvenStreamTable
evenStreamTable Int
sid Stream
newstrm
            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) -> Maybe (Int, Stream)
forall a. a -> Maybe a
Just (Int
sid, Stream
newstrm)
        Just Int
maxConc -> do
            Maybe Int
msid <- STM (Maybe Int) -> IO (Maybe Int)
forall a. STM a -> IO a
atomically (STM (Maybe Int) -> IO (Maybe Int))
-> STM (Maybe Int) -> IO (Maybe Int)
forall a b. (a -> b) -> a -> b
$ do
                let open :: STM (Maybe Int)
open = do
                        TVar EvenStreamTable -> Int -> STM ()
waitIncEven TVar EvenStreamTable
evenStreamTable Int
maxConc
                        Int -> Maybe Int
forall a. a -> Maybe a
Just (Int -> Maybe Int) -> STM Int -> STM (Maybe Int)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Context -> STM Int
getMyNewStreamId Context
ctx
                STM (Maybe Int)
open STM (Maybe Int) -> STM (Maybe Int) -> STM (Maybe Int)
forall a. STM a -> STM a -> STM a
`orElse` Maybe Int -> STM (Maybe Int)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe Int
forall a. Maybe a
Nothing
            Maybe Int -> (Int -> IO (Int, Stream)) -> IO (Maybe (Int, Stream))
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM Maybe Int
msid ((Int -> IO (Int, Stream)) -> IO (Maybe (Int, Stream)))
-> (Int -> IO (Int, Stream)) -> IO (Maybe (Int, Stream))
forall a b. (a -> b) -> a -> b
$ \Int
sid -> do
                Int
txws <- Settings -> Int
initialWindowSize (Settings -> Int) -> IO Settings -> IO Int
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
                Stream
newstrm <- Int -> Int -> Int -> IO Stream
newEvenStream Int
sid Int
txws Int
rxws
                TVar EvenStreamTable -> Int -> Stream -> IO ()
insertEven' TVar EvenStreamTable
evenStreamTable Int
sid Stream
newstrm
                (Int, Stream) -> IO (Int, Stream)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Int
sid, Stream
newstrm)

----------------------------------------------------------------
-- GOAWAY from the peer

-- | No new stream once the peer has sent GOAWAY: "Receivers of a GOAWAY
-- frame MUST NOT open additional streams on the connection" (RFC 9113,
-- section 6.8).  A request that has not got a stream yet is answered as
-- one on a stream above the last stream identifier would be.
refuseIfGoingAway :: Context -> STM ()
refuseIfGoingAway :: Context -> STM ()
refuseIfGoingAway Context{TVar (Maybe Int)
peerGoAway :: Context -> TVar (Maybe Int)
peerGoAway :: TVar (Maybe Int)
peerGoAway} = do
    Bool
goingaway <- Maybe Int -> Bool
forall a. Maybe a -> Bool
isJust (Maybe Int -> Bool) -> STM (Maybe Int) -> STM Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar (Maybe Int) -> STM (Maybe Int)
forall a. TVar a -> STM a
readTVar TVar (Maybe Int)
peerGoAway
    Bool -> STM () -> STM ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
goingaway (STM () -> STM ()) -> STM () -> STM ()
forall a b. (a -> b) -> a -> b
$ HTTP2Error -> STM ()
forall e a. Exception e => e -> STM a
throwSTM HTTP2Error
ConnectionIsClosed

-- | Taking the peer's GOAWAY(NO_ERROR) on board.
--
-- The peer has said it will go no further than the last stream identifier,
-- not that it is going at once.  Streams of ours above it will not be
-- processed, and are closed with 'ConnectionIsClosed', which a client may
-- retry elsewhere; no stream is opened from now on.  The others go on
-- until they are done, and then so is the connection: 'drained' tells
-- when.  It used to be closed right away, cutting off every stream in
-- flight, even those the peer had promised to finish.
--
-- A later GOAWAY can lower the last stream identifier, not raise it.
-- Answers whether this is the first.
goingAway :: Context -> StreamId -> IO Bool
goingAway :: Context -> Int -> IO Bool
goingAway ctx :: Context
ctx@Context{TVar (Maybe Int)
peerGoAway :: Context -> TVar (Maybe Int)
peerGoAway :: TVar (Maybe Int)
peerGoAway, TVar OddStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
oddStreamTable :: TVar OddStreamTable
oddStreamTable, TVar EvenStreamTable
evenStreamTable :: Context -> TVar EvenStreamTable
evenStreamTable :: TVar EvenStreamTable
evenStreamTable} Int
lastSid = do
    Bool
first <- 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
        Maybe Int
old <- TVar (Maybe Int) -> STM (Maybe Int)
forall a. TVar a -> STM a
readTVar TVar (Maybe Int)
peerGoAway
        TVar (Maybe Int) -> Maybe Int -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar (Maybe Int)
peerGoAway (Maybe Int -> STM ()) -> Maybe Int -> STM ()
forall a b. (a -> b) -> a -> b
$ Int -> Maybe Int
forall a. a -> Maybe a
Just (Int -> Maybe Int) -> Int -> Maybe Int
forall a b. (a -> b) -> a -> b
$ Int -> (Int -> Int) -> Maybe Int -> Int
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Int
lastSid (Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
lastSid) Maybe Int
old
        Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return (Bool -> STM Bool) -> Bool -> STM Bool
forall a b. (a -> b) -> a -> b
$ Maybe Int -> Bool
forall a. Maybe a -> Bool
isNothing Maybe Int
old
    IntMap Stream
mine <-
        if Context -> Bool
isClient Context
ctx
            then TVar OddStreamTable -> IO (IntMap Stream)
getOddStreams TVar OddStreamTable
oddStreamTable
            else TVar EvenStreamTable -> IO (IntMap Stream)
getEvenStreams TVar EvenStreamTable
evenStreamTable
    let (IntMap Stream
_, IntMap Stream
above) = Int -> IntMap Stream -> (IntMap Stream, IntMap Stream)
forall a. Int -> IntMap a -> (IntMap a, IntMap a)
IntMap.split Int
lastSid IntMap Stream
mine
    IntMap Stream -> (Stream -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ IntMap Stream
above ((Stream -> IO ()) -> IO ()) -> (Stream -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Stream
strm -> Context -> Stream -> ClosedCode -> IO ()
closed Context
ctx Stream
strm ClosedCode
Finished
    Bool -> IO Bool
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
first

-- | Waiting until nothing is left in flight after the peer's GOAWAY:
-- neither a stream of the peer's, nor one of ours the peer is to process,
-- nor -- on a client -- a response still being read.  Streams of ours above
-- the last stream identifier are not waited for; those that got their
-- identifier too late to be closed by 'goingAway' are closed with the
-- connection.
drained :: Context -> STM ()
drained :: Context -> STM ()
drained ctx :: Context
ctx@Context{TVar (Maybe Int)
peerGoAway :: Context -> TVar (Maybe Int)
peerGoAway :: TVar (Maybe Int)
peerGoAway, TVar OddStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
oddStreamTable :: TVar OddStreamTable
oddStreamTable, TVar EvenStreamTable
evenStreamTable :: Context -> TVar EvenStreamTable
evenStreamTable :: TVar EvenStreamTable
evenStreamTable, TVar Int
activeRequests :: Context -> TVar Int
activeRequests :: TVar Int
activeRequests} = do
    Maybe Int
mlast <- TVar (Maybe Int) -> STM (Maybe Int)
forall a. TVar a -> STM a
readTVar TVar (Maybe Int)
peerGoAway
    case Maybe Int
mlast of
        Maybe Int
Nothing -> STM ()
forall a. STM a
retry
        Just Int
lastSid -> do
            IntMap Stream
odds <- OddStreamTable -> IntMap Stream
oddTable (OddStreamTable -> IntMap Stream)
-> STM OddStreamTable -> STM (IntMap Stream)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar OddStreamTable -> STM OddStreamTable
forall a. TVar a -> STM a
readTVar TVar OddStreamTable
oddStreamTable
            IntMap Stream
evens <- EvenStreamTable -> IntMap Stream
evenTable (EvenStreamTable -> IntMap Stream)
-> STM EvenStreamTable -> STM (IntMap Stream)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar EvenStreamTable -> STM EvenStreamTable
forall a. TVar a -> STM a
readTVar TVar EvenStreamTable
evenStreamTable
            Int
active <- TVar Int -> STM Int
forall a. TVar a -> STM a
readTVar TVar Int
activeRequests
            let (IntMap Stream
mine, IntMap Stream
theirs)
                    | Context -> Bool
isClient Context
ctx = (IntMap Stream
odds, IntMap Stream
evens)
                    | Bool
otherwise = (IntMap Stream
evens, IntMap Stream
odds)
                noneOfMine :: Bool
noneOfMine = Bool -> ((Int, Stream) -> Bool) -> Maybe (Int, Stream) -> Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Bool
True ((Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
lastSid) (Int -> Bool) -> ((Int, Stream) -> Int) -> (Int, Stream) -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int, Stream) -> Int
forall a b. (a, b) -> a
fst) (Maybe (Int, Stream) -> Bool) -> Maybe (Int, Stream) -> Bool
forall a b. (a -> b) -> a -> b
$ IntMap Stream -> Maybe (Int, Stream)
forall a. IntMap a -> Maybe (Int, a)
IntMap.lookupMin IntMap Stream
mine
            Bool -> STM ()
check (Bool -> STM ()) -> Bool -> STM ()
forall a b. (a -> b) -> a -> b
$ IntMap Stream -> Bool
forall a. IntMap a -> Bool
IntMap.null IntMap Stream
theirs Bool -> Bool -> Bool
&& Bool
noneOfMine Bool -> Bool -> Bool
&& Int
active Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0