Documentation
¶
Overview ¶
Package bus is the process's execution bus: every piece of work that reaches a database runs on one of its lanes, and every background task is started through it, so what the process runs is bounded, visible and stopped together.
A lane bounds concurrent work and the work waiting for it. All lanes draw on one IO semaphore; the lanes other than Serve share an auxiliary half of it, so administration, scans and background work never take the whole IO pool from tool calls. Work admitted on a lane carries the admission in its context, and Ensure runs nested work under it instead of taking a second slot.
Go and AfterFunc start background tasks: each is named, counted, recovers from panics, sees its context canceled on Close, and Close waits for it. The bus uses only the standard library.
Index ¶
- Constants
- Variables
- type Bus
- func (b *Bus) AfterFunc(name string, d time.Duration, fn func()) (stop func() bool)
- func (b *Bus) Close(ctx context.Context) error
- func (b *Bus) Fork(wg *sync.WaitGroup, name string, fn func(context.Context))
- func (b *Bus) Go(name string, fn func(context.Context)) bool
- func (b *Bus) IO() *Semaphore
- func (b *Bus) Lane(name LaneName) *Lane
- func (b *Bus) Stats() Stats
- func (b *Bus) Update(io int, limits map[LaneName]Limits)
- type Lane
- func (l *Lane) Acquire(ctx context.Context) (admitted context.Context, release func(), err error)
- func (l *Lane) Ensure(ctx context.Context, fn func(context.Context) error) error
- func (l *Lane) Name() LaneName
- func (l *Lane) Queue() (*Ticket, error)
- func (l *Lane) Run(ctx context.Context, fn func(context.Context) error) error
- func (l *Lane) RunWaiting(ctx context.Context, fn func(context.Context) error) error
- func (l *Lane) Stats() LaneStats
- func (l *Lane) TryEnter() (release func(), err error)
- type LaneName
- type LaneStats
- type Limits
- type Semaphore
- type SlotStats
- type Stats
- type TaskStats
- type Ticket
Constants ¶
const DefaultIO = 16
DefaultIO is the IO slots of a bus given none.
Variables ¶
var ( // ErrOverloaded is returned when a lane's concurrency and queue are full. ErrOverloaded = errors.New("bus: overloaded; retry later") // ErrClosed is returned once the bus is closed. ErrClosed = errors.New("bus: closed") )
Sentinel errors.
var DefaultLimits = map[LaneName]Limits{
Serve: {MaxConcurrent: 256},
Admin: {MaxConcurrent: 4, MaxQueue: 16},
Scan: {MaxConcurrent: 2, MaxQueue: 8},
Background: {MaxConcurrent: 2, MaxQueue: 64},
}
DefaultLimits are the lane limits of a bus given none.
var Lanes = []LaneName{Serve, Admin, Scan, Background}
Lanes lists the lanes in the order Stats reports them.
Functions ¶
This section is empty.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus admits the process's database work and runs its background tasks. A nil *Bus has nil lanes, which run work without bounds, and starts tasks that still recover from panics but are neither counted nor waited for.
func New ¶
New returns a bus of io IO slots (DefaultIO when not positive) and lanes limited as limits says, DefaultLimits for the ones it leaves out.
func (*Bus) AfterFunc ¶
AfterFunc runs fn as the background task name once d has passed, unless stop is called first or the bus is closed by then.
func (*Bus) Close ¶
Close rejects new work and tasks, cancels the tasks' context and waits for them until ctx ends.
func (*Bus) Fork ¶
Fork runs fn as the task name and marks it done on wg, for work a caller waits for on wg. Once the bus is closed fn still runs, uncounted, so the caller never waits for work that does not come.
func (*Bus) Go ¶
Go runs fn as the background task name; fn's context ends on Close. It returns false, not running fn, once the bus is closed.
type Lane ¶
type Lane struct {
// contains filtered or unexported fields
}
Lane bounds the concurrency of one kind of work. A nil *Lane runs work without bounds.
func (*Lane) Acquire ¶
Acquire admits work, waiting in the lane's queue until ctx ends, and takes an IO slot for it; the work runs on the returned context, which carries the admission, and release ends it. A full queue fails at once.
func (*Lane) Queue ¶
Queue takes a place in the lane's queue at once, for work admitted later in the order its caller chooses: it fails when the lane's running and waiting work already fill its concurrency and queue. The work counts as waiting until the ticket's Acquire admits it or Cancel gives the place up.
func (*Lane) Run ¶
Run runs fn once admitted; fn's context says it was, so Ensure runs nested work under this admission.
func (*Lane) RunWaiting ¶
RunWaiting is Run that waits however full the queue is, for work that must not fail as busy and whose number the caller bounds, such as assembling a configuration (one at a time, a task per datasource).
type LaneName ¶
type LaneName string
LaneName names a lane.
const ( // Serve runs tool calls; the engine admits them and takes their IO slots. Serve LaneName = "serve" // Admin runs reads that answer an administrator's request. Admin LaneName = "admin" // Scan runs schema scans. Scan LaneName = "scan" // Background runs work nobody waits for, such as capability assessment. Background LaneName = "background" )
The lanes of a bus.
type LaneStats ¶
type LaneStats struct {
Name LaneName
MaxConcurrent, MaxQueue int
Inflight, Queued int
Rejected, Completed int64
}
LaneStats describe a lane.
type Semaphore ¶
type Semaphore struct {
// contains filtered or unexported fields
}
Semaphore is a counting semaphore whose size can change. Waiters are granted in arrival order.
func NewSemaphore ¶
NewSemaphore returns a semaphore of n slots.
func (*Semaphore) Release ¶
func (s *Semaphore) Release()
Release frees a slot taken by Acquire or TryAcquire.
func (*Semaphore) Resize ¶
Resize sets the number of slots. Holders keep their slots; when the slots shrink, new work waits until the running work fits.
func (*Semaphore) TryAcquire ¶
TryAcquire takes a slot when one is free and nobody waits for it.
Source Files