blob: 265fdaf85511af20533e3ccc61e0dd4557481274 [file] [log] [blame]
// Copyright 2023 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
//go:build go1.21
package quic
import (
"sync/atomic"
"time"
)
// connInflow tracks connection-level flow control for data sent by the peer to us.
//
// There are four byte offsets of significance in the stream of data received from the peer,
// each >= to the previous:
//
// - bytes read by the user
// - bytes received from the peer
// - limit sent to the peer in a MAX_DATA frame
// - potential new limit to sent to the peer
//
// We maintain a flow control window, so as bytes are read by the user
// the potential limit is extended correspondingly.
//
// We keep an atomic counter of bytes read by the user and not yet applied to the
// potential limit (credit). When this count grows large enough, we update the
// new limit to send and mark that we need to send a new MAX_DATA frame.
type connInflow struct {
sent sentVal // set when we need to send a MAX_DATA update to the peer
usedLimit int64 // total bytes sent by the peer, must be less than sentLimit
sentLimit int64 // last MAX_DATA sent to the peer
newLimit int64 // new MAX_DATA to send
credit atomic.Int64 // bytes read but not yet applied to extending the flow-control window
}
func (c *Conn) inflowInit() {
// The initial MAX_DATA limit is sent as a transport parameter.
c.streams.inflow.sentLimit = c.config.maxConnReadBufferSize()
c.streams.inflow.newLimit = c.streams.inflow.sentLimit
}
// handleStreamBytesReadOffLoop records that the user has consumed bytes from a stream.
// We may extend the peer's flow control window.
//
// This is called indirectly by the user, via Read or CloseRead.
func (c *Conn) handleStreamBytesReadOffLoop(n int64) {
if c.shouldUpdateFlowControl(c.streams.inflow.credit.Add(n)) {
// We should send a MAX_DATA update to the peer.
// Record this on the Conn's main loop.
c.sendMsg(func(now time.Time, c *Conn) {
c.sendMaxDataUpdate()
})
}
}
// handleStreamBytesReadOnLoop extends the peer's flow control window after
// data has been discarded due to a RESET_STREAM frame.
//
// This is called on the conn's loop.
func (c *Conn) handleStreamBytesReadOnLoop(n int64) {
if c.shouldUpdateFlowControl(c.streams.inflow.credit.Add(n)) {
c.sendMaxDataUpdate()
}
}
func (c *Conn) sendMaxDataUpdate() {
c.streams.inflow.sent.setUnsent()
// Apply current credit to the limit.
// We don't strictly need to do this here
// since appendMaxDataFrame will do so as well,
// but this avoids redundant trips down this path
// if the MAX_DATA frame doesn't go out right away.
c.streams.inflow.newLimit += c.streams.inflow.credit.Swap(0)
}
func (c *Conn) shouldUpdateFlowControl(credit int64) bool {
return shouldUpdateFlowControl(c.config.maxConnReadBufferSize(), credit)
}
// handleStreamBytesReceived records that the peer has sent us stream data.
func (c *Conn) handleStreamBytesReceived(n int64) error {
c.streams.inflow.usedLimit += n
if c.streams.inflow.usedLimit > c.streams.inflow.sentLimit {
return localTransportError(errFlowControl)
}
return nil
}
// appendMaxDataFrame appends a MAX_DATA frame to the current packet.
//
// It returns true if no more frames need appending,
// false if it could not fit a frame in the current packet.
func (c *Conn) appendMaxDataFrame(w *packetWriter, pnum packetNumber, pto bool) bool {
if c.streams.inflow.sent.shouldSendPTO(pto) {
// Add any unapplied credit to the new limit now.
c.streams.inflow.newLimit += c.streams.inflow.credit.Swap(0)
if !w.appendMaxDataFrame(c.streams.inflow.newLimit) {
return false
}
c.streams.inflow.sent.setSent(pnum)
}
return true
}
// ackOrLossMaxData records the fate of a MAX_DATA frame.
func (c *Conn) ackOrLossMaxData(pnum packetNumber, fate packetFate) {
c.streams.inflow.sent.ackLatestOrLoss(pnum, fate)
}
// connOutflow tracks connection-level flow control for data sent by us to the peer.
type connOutflow struct {
max int64 // largest MAX_DATA received from peer
used int64 // total bytes of STREAM data sent to peer
}
// setMaxData updates the connection-level flow control limit
// with the initial limit conveyed in transport parameters
// or an update from a MAX_DATA frame.
func (f *connOutflow) setMaxData(maxData int64) {
f.max = max(f.max, maxData)
}
// avail returns the number of connection-level flow control bytes available.
func (f *connOutflow) avail() int64 {
return f.max - f.used
}
// consume records consumption of n bytes of flow.
func (f *connOutflow) consume(n int64) {
f.used += n
}