worker

Documentation

Index

• type BaseHandler
• type BucketLeakingLimiter
  • func NewBucketLeakingLimiter(rps uint64) *BucketLeakingLimiter
  • func (limiter *BucketLeakingLimiter) Take()
• type BulkWriteHandler
  • func (h *BulkWriteHandler) Execute() error
• type DropCollection
  • func (h *DropCollection) Execute() error
• type JobHandler
  • func NewJobHandler(job *config.Job, client database.Client, dataPool schema.DataPool, ...) JobHandler
• type JobPool
  • func NewDeductionJobPool(requestsNumber uint64) JobPool
  • func NewJobPool(cfg *config.Job) JobPool
  • func NewNoLimitTimerJobPool() JobPool
  • func NewTimerJobPool(duration time.Duration) JobPool
• type Limiter
  • func NewLimiter(rate uint64) Limiter
• type Metrics
  • func NewMetrics(job *config.Job) *Metrics
  • func (m *Metrics) DurationSeconds() uint64
  • func (m *Metrics) ErrorRate() float32
  • func (m *Metrics) Init()
  • func (m *Metrics) Meter(handler func() error)
  • func (m *Metrics) Requests() uint64
  • func (m *Metrics) Rps() uint64
• type MutableBucketLeakingLimiter
  • func NewMutableBucketLeakingLimiter(rate uint64) *MutableBucketLeakingLimiter
  • func (l *MutableBucketLeakingLimiter) SetRate(rate uint64)
  • func (l *MutableBucketLeakingLimiter) Take()
• type NoLimitLimiter
  • func NewNoLimitLimiter() *NoLimitLimiter
  • func (*NoLimitLimiter) SetRate(uint64)
  • func (*NoLimitLimiter) Take()
• type ReadHandler
  • func (h *ReadHandler) Execute() error
• type SleepHandler
  • func (h *SleepHandler) Execute() error
• type UpdateHandler
  • func (h *UpdateHandler) Execute() error
• type Worker
  • func NewWorker(ctx context.Context, cfg *config.Config, job *config.Job, ...) (*Worker, error)
  • func (w *Worker) Cancel()
  • func (w *Worker) Close()
  • func (w *Worker) ExtendCopySavedFieldsToDataPool()
  • func (w *Worker) InitMetrics()
  • func (w *Worker) IsDone() bool
  • func (w *Worker) JobName() string
  • func (w *Worker) RequestedDurationSeconds() uint64
  • func (w *Worker) RequestedOperations() uint64
  • func (w *Worker) Summary()
  • func (w *Worker) Work(agents chan uint64)
• type WriteHandler
  • func (h *WriteHandler) Execute() error

Types

type BaseHandler

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

type BucketLeakingLimiter

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

func NewBucketLeakingLimiter

func NewBucketLeakingLimiter(rps uint64) *BucketLeakingLimiter

func (*BucketLeakingLimiter) Take

func (limiter *BucketLeakingLimiter) Take()

type BulkWriteHandler

type BulkWriteHandler struct {
	*BaseHandler
}

func (*BulkWriteHandler) Execute

func (h *BulkWriteHandler) Execute() error

type DropCollection

type DropCollection struct {
	*BaseHandler
}

func (*DropCollection) Execute

func (h *DropCollection) Execute() error

type JobHandler

type JobHandler interface {
	Execute() error
}

func NewJobHandler

func NewJobHandler(job *config.Job, client database.Client, dataPool schema.DataPool, s *config.Schema) JobHandler

type JobPool

type JobPool interface {
	SpawnJob() bool
	MarkJobDone()
	Cancel()
	GetRequestsStarted() uint64
	GetRequestsDone() uint64
}

func NewDeductionJobPool

func NewDeductionJobPool(requestsNumber uint64) JobPool

func NewJobPool

func NewJobPool(cfg *config.Job) JobPool

func NewNoLimitTimerJobPool

func NewNoLimitTimerJobPool() JobPool

func NewTimerJobPool

func NewTimerJobPool(duration time.Duration) JobPool

type Limiter

type Limiter interface {
	Take()
	SetRate(uint64)
}

func NewLimiter

func NewLimiter(rate uint64) Limiter

type Metrics

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

func NewMetrics

func NewMetrics(job *config.Job) *Metrics

func (*Metrics) DurationSeconds

func (m *Metrics) DurationSeconds() uint64

func (*Metrics) ErrorRate

func (m *Metrics) ErrorRate() float32

func (*Metrics) Init

func (m *Metrics) Init()

func (*Metrics) Meter

func (m *Metrics) Meter(handler func() error)

func (*Metrics) Requests

func (m *Metrics) Requests() uint64

func (*Metrics) Rps

func (m *Metrics) Rps() uint64

type MutableBucketLeakingLimiter

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

func NewMutableBucketLeakingLimiter

func NewMutableBucketLeakingLimiter(rate uint64) *MutableBucketLeakingLimiter

newAtomicBased returns a new atomic based limiter.

func (*MutableBucketLeakingLimiter) SetRate

func (l *MutableBucketLeakingLimiter) SetRate(rate uint64)

func (*MutableBucketLeakingLimiter) Take

func (l *MutableBucketLeakingLimiter) Take()

Take blocks to ensure that the time spent between multiple Take calls is on average time.Second/rate.

type NoLimitLimiter

type NoLimitLimiter struct{}

func NewNoLimitLimiter

func NewNoLimitLimiter() *NoLimitLimiter

func (*NoLimitLimiter) SetRate

func (*NoLimitLimiter) SetRate(uint64)

func (*NoLimitLimiter) Take

func (*NoLimitLimiter) Take()

type ReadHandler

type ReadHandler struct {
	*BaseHandler
}

func (*ReadHandler) Execute

func (h *ReadHandler) Execute() error

type SleepHandler

type SleepHandler struct {
	Duration time.Duration
}

func (*SleepHandler) Execute

func (h *SleepHandler) Execute() error

type UpdateHandler

type UpdateHandler struct {
	*BaseHandler
}

func (*UpdateHandler) Execute

func (h *UpdateHandler) Execute() error

type Worker

type Worker struct {
	Metrics *Metrics
	// contains filtered or unexported fields
}

todo: split this function to setup and to starting workers

func NewWorker

func NewWorker(ctx context.Context, cfg *config.Config, job *config.Job, dataPool schema.DataPool, runningAgents uint64) (*Worker, error)

func (*Worker) Cancel

func (w *Worker) Cancel()

func (*Worker) Close

func (w *Worker) Close()

func (*Worker) ExtendCopySavedFieldsToDataPool

func (w *Worker) ExtendCopySavedFieldsToDataPool()

todo: fix wrong place invalid

func (*Worker) InitMetrics

func (w *Worker) InitMetrics()

func (*Worker) IsDone

func (w *Worker) IsDone() bool

func (*Worker) JobName

func (w *Worker) JobName() string

func (*Worker) RequestedDurationSeconds

func (w *Worker) RequestedDurationSeconds() uint64

func (*Worker) RequestedOperations

func (w *Worker) RequestedOperations() uint64

func (*Worker) Summary

func (w *Worker) Summary()

func (*Worker) Work

func (w *Worker) Work(agents chan uint64)

type WriteHandler

type WriteHandler struct {
	*BaseHandler
}

func (*WriteHandler) Execute

func (h *WriteHandler) Execute() error