Documentation
¶
Overview ¶
Package backpressure controls Store dispatch using feedback from real work. All ready Store executions share one bounded FIFO. The controller owns no background goroutine, dependency probe or retry policy.
Index ¶
- Constants
- Variables
- func Observe(backend storage.Storage, controller *Controller) storage.Storage
- type Controller
- func (c *Controller) Acquire(ctx context.Context) (*Permit, error)
- func (c *Controller) Admit(ctx context.Context, requestBytes int) (context.Context, *Permit, error)
- func (c *Controller) Collect(ch chan<- prometheus.Metric)
- func (c *Controller) Describe(ch chan<- *prometheus.Desc)
- func (c *Controller) Enqueue(reservations []*Reservation) (*Ticket, <-chan struct{})
- func (c *Controller) Ready() (bool, <-chan struct{}, time.Duration)
- func (c *Controller) Reserve(requestBytes int) (*Reservation, error)
- func (c *Controller) TryAcquire() (*Permit, <-chan struct{}, time.Duration)
- func (c *Controller) Wait(ctx context.Context) error
- type Options
- type Permit
- type Reservation
- type Ticket
Constants ¶
const DefaultMaxConcurrent = 64
Variables ¶
var ErrQueueFull = status.Error(codes.ResourceExhausted, "Store waiting capacity is full")
ErrQueueFull proves that admission did not start backend work.
Functions ¶
Types ¶
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
func New ¶
func New(opts Options) (*Controller, error)
func (*Controller) Acquire ¶
func (c *Controller) Acquire(ctx context.Context) (*Permit, error)
Acquire waits only at the already bounded Kafka processing loop. Engine requests use Admit or enqueue a batch Ticket, never this unqueued wait path.
func (*Controller) Admit ¶
Admit reuses an execution's permit, or waits in the same FIFO as ready batches. Only the caller's original context ends admission waiting.
func (*Controller) Collect ¶
func (c *Controller) Collect(ch chan<- prometheus.Metric)
func (*Controller) Describe ¶
func (c *Controller) Describe(ch chan<- *prometheus.Desc)
func (*Controller) Enqueue ¶ added in v0.22.0
func (c *Controller) Enqueue(reservations []*Reservation) (*Ticket, <-chan struct{})
Enqueue transfers existing byte reservations into one FIFO task. A full task queue leaves ownership with the producer; changed wakes a batcher to retry. It does not count such retries as rejected business requests.
func (*Controller) Ready ¶
func (c *Controller) Ready() (bool, <-chan struct{}, time.Duration)
Ready lets startup and Worker prefetch observe capacity without reserving a slot or inventing demand. Engine execution tickets are polled separately.
func (*Controller) Reserve ¶ added in v0.22.0
func (c *Controller) Reserve(requestBytes int) (*Reservation, error)
func (*Controller) TryAcquire ¶
func (c *Controller) TryAcquire() (*Permit, <-chan struct{}, time.Duration)
TryAcquire is used by the bounded Worker loop; it cannot bypass queued tasks. Engine dispatchers enqueue a Ticket instead. On failure, changed and delay describe when to retry; delay == 0 means only a completion or another state change can free capacity.
type Permit ¶
type Permit struct {
// contains filtered or unexported fields
}
Permit reserves one sequential execution, including its existing conflict retries. Reducing the window never revokes already admitted executions.
type Reservation ¶ added in v0.22.0
type Reservation struct {
// contains filtered or unexported fields
}
Reservation charges retained request bytes across collection and admission. Moving a request into a batch never releases and reacquires its budget.
func (*Reservation) Release ¶ added in v0.22.0
func (r *Reservation) Release()
Source Files