batch

Documentation

Index

• type Request
• type Scheduler
• type SchedulerOption
  • func LazyAdd() SchedulerOption
• type TimeScheduler
  • func NewTimeScheduler(db kvdb.Backend, locker sync.Locker, duration time.Duration) *TimeScheduler
  • func (s *TimeScheduler) Execute(r *Request) error

Types

type Request

type Request struct {
	// Reset is called before each invocation of Update and is used to clear
	// any possible modifications to local state as a result of previous
	// calls to Update that were not committed due to a concurrent batch
	// failure.
	//
	// NOTE: This field is optional.
	Reset func()

	// Update is applied alongside other operations in the batch.
	//
	// NOTE: This method MUST NOT acquire any mutexes.
	Update func(tx kvdb.RwTx) error

	// OnCommit is called if the batch or a subset of the batch including
	// this request all succeeded without failure. The passed error should
	// contain the result of the transaction commit, as that can still fail
	// even if none of the closures returned an error.
	//
	// NOTE: This field is optional.
	OnCommit func(commitErr error) error
	// contains filtered or unexported fields
}

Request defines an operation that can be batched into a single bbolt transaction.

type Scheduler

type Scheduler interface {
	// Execute schedules a Request for execution with the next available
	// batch. This method blocks until the the underlying closure has been
	// run against the database. The resulting error is returned to the
	// caller.
	Execute(req *Request) error
}

Scheduler abstracts a generic batching engine that accumulates an incoming set of Requests, executes them, and returns the error from the operation.

type SchedulerOption

type SchedulerOption func(r *Request)

SchedulerOption is a type that can be used to supply options to a scheduled request.

func LazyAdd

func LazyAdd() SchedulerOption

LazyAdd will make the request be executed lazily, added to the next batch to reduce db contention.

type TimeScheduler

type TimeScheduler struct {
	// contains filtered or unexported fields
}

TimeScheduler is a batching engine that executes requests within a fixed horizon. When the first request is received, a TimeScheduler waits a configurable duration for other concurrent requests to join the batch. Once this time has elapsed, the batch is closed and executed. Subsequent requests are then added to a new batch which undergoes the same process.

func NewTimeScheduler

func NewTimeScheduler(db kvdb.Backend, locker sync.Locker,
	duration time.Duration) *TimeScheduler

NewTimeScheduler initializes a new TimeScheduler with a fixed duration at which to schedule batches. If the operation needs to modify a higher-level cache, the cache's lock should be provided to so that external consistency can be maintained, as successful db operations will cause a request's OnCommit method to be executed while holding this lock.

func (*TimeScheduler) Execute

func (s *TimeScheduler) Execute(r *Request) error

Execute schedules the provided request for batch execution along with other concurrent requests. The request will be executed within a fixed horizon, parameterizeed by the duration of the scheduler. The error from the underlying operation is returned to the caller.

NOTE: Part of the Scheduler interface.