balancer

Documentation

Overview

Package balancer is an implementation of the controller.Balancer interface.

Index

• type Balancer
  • func New(ns controller.NodeService, fjs FreeJobService, wjs WorkerJobService, ...) *Balancer
  • func (b *Balancer) AddJobs(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID, ...) ([]dax.WorkerDiff, error)
  • func (b *Balancer) AddWorker(tx dax.Transaction, node *dax.Node) ([]dax.WorkerDiff, error)
  • func (b *Balancer) Balance(tx dax.Transaction) ([]dax.WorkerDiff, error)
  • func (b *Balancer) BalanceDatabase(tx dax.Transaction, qdbid dax.QualifiedDatabaseID) ([]dax.WorkerDiff, error)
  • func (b *Balancer) CurrentState(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) ([]dax.WorkerInfo, error)
  • func (b *Balancer) FreeWorkers(tx dax.Transaction, addrs ...dax.Address) error
  • func (b *Balancer) Nodes(tx dax.Transaction) ([]*dax.Node, error)
  • func (b *Balancer) ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node, error)
  • func (b *Balancer) RemoveJobs(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID, ...) ([]dax.WorkerDiff, error)
  • func (b *Balancer) RemoveWorker(tx dax.Transaction, addr dax.Address) ([]dax.WorkerDiff, error)
  • func (b *Balancer) WorkerState(tx dax.Transaction, roleType dax.RoleType, addr dax.Address) (dax.WorkerInfo, error)
  • func (b *Balancer) WorkersForJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, ...) ([]dax.WorkerInfo, error)
  • func (b *Balancer) WorkersForTable(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID) ([]dax.WorkerInfo, error)
• type FreeJobService
• type FreeWorkerService
• type InternalDiffs
  • func NewInternalDiffs() InternalDiffs
  • func (d InternalDiffs) Added(address dax.Address, job dax.Job)
  • func (d InternalDiffs) Merge(d2 InternalDiffs)
  • func (d InternalDiffs) Output() []dax.WorkerDiff
  • func (d InternalDiffs) Removed(address dax.Address, job dax.Job)
• type WorkerJobService

Types

type Balancer

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

Balancer is an implementation of the balancer.Balancer interface which isolates workers and jobs by database. It helps manage the relationships between workers and jobs. The logic it uses to balance jobs across workers is very simple; it bases everything off the number of workers and number of jobs. It does not take anything else (such as job size, worker capabilities, etc) into consideration.

func New

func New(ns controller.NodeService, fjs FreeJobService, wjs WorkerJobService, fws FreeWorkerService, schemar schemar.Schemar, logger logger.Logger) *Balancer

New returns a new instance of Balancer.

func (*Balancer) AddJobs

func (b *Balancer) AddJobs(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID, jobs ...dax.Job) ([]dax.WorkerDiff, error)

func (*Balancer) AddWorker

func (b *Balancer) AddWorker(tx dax.Transaction, node *dax.Node) ([]dax.WorkerDiff, error)

AddWorker adds the given Node to the Balancer's available worker pool. TODO(tlt): this method takes a Node (as opposed to a Worker) because in the future we may want to maintain separate worker pools based on RoleType (compute, translate, etc.).

func (*Balancer) Balance

func (b *Balancer) Balance(tx dax.Transaction) ([]dax.WorkerDiff, error)

Balance calls balanceDatabase on every database in the schemar.

func (*Balancer) BalanceDatabase

func (b *Balancer) BalanceDatabase(tx dax.Transaction, qdbid dax.QualifiedDatabaseID) ([]dax.WorkerDiff, error)

func (*Balancer) CurrentState

func (b *Balancer) CurrentState(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) ([]dax.WorkerInfo, error)

func (*Balancer) FreeWorkers

func (b *Balancer) FreeWorkers(tx dax.Transaction, addrs ...dax.Address) error

func (*Balancer) Nodes

func (b *Balancer) Nodes(tx dax.Transaction) ([]*dax.Node, error)

func (*Balancer) ReadNode

func (b *Balancer) ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node, error)

func (*Balancer) RemoveJobs

func (b *Balancer) RemoveJobs(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID, jobs ...dax.Job) ([]dax.WorkerDiff, error)

func (*Balancer) RemoveWorker

func (b *Balancer) RemoveWorker(tx dax.Transaction, addr dax.Address) ([]dax.WorkerDiff, error)

func (*Balancer) WorkerState

func (b *Balancer) WorkerState(tx dax.Transaction, roleType dax.RoleType, addr dax.Address) (dax.WorkerInfo, error)

func (*Balancer) WorkersForJobs

func (b *Balancer) WorkersForJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, jobs ...dax.Job) ([]dax.WorkerInfo, error)

func (*Balancer) WorkersForTable

func (b *Balancer) WorkersForTable(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID) ([]dax.WorkerInfo, error)

type FreeJobService

type FreeJobService interface {
	CreateJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, job ...dax.Job) error
	DeleteJob(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, job dax.Job) error
	DeleteJobsForTable(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID) error
	ListJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) (dax.Jobs, error)
	MergeJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, jobs dax.Jobs) error
}

type FreeWorkerService

type FreeWorkerService interface {
	AddWorkers(tx dax.Transaction, roleType dax.RoleType, addrs ...dax.Address) error
	RemoveWorker(tx dax.Transaction, roleType dax.RoleType, addr dax.Address) error
	PopWorkers(tx dax.Transaction, roleType dax.RoleType, num int) ([]dax.Address, error)
	ListWorkers(tx dax.Transaction, roleType dax.RoleType) (dax.Addresses, error)
}

type InternalDiffs

type InternalDiffs map[dax.Address]jobSetDiffs

func NewInternalDiffs

func NewInternalDiffs() InternalDiffs

func (InternalDiffs) Added

func (d InternalDiffs) Added(address dax.Address, job dax.Job)

func (InternalDiffs) Merge

func (d InternalDiffs) Merge(d2 InternalDiffs)

func (InternalDiffs) Output

func (d InternalDiffs) Output() []dax.WorkerDiff

Output converts internalDiff to []controller.WorkerDiff for external consumption.

func (InternalDiffs) Removed

func (d InternalDiffs) Removed(address dax.Address, job dax.Job)

type WorkerJobService

type WorkerJobService interface {
	WorkersJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) ([]dax.WorkerInfo, error)

	WorkerCount(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) (int, error)
	ListWorkers(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) (dax.Addresses, error)

	CreateWorker(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, addr dax.Address) error
	DeleteWorker(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, addr dax.Address) error
	FreeWorkers(tx dax.Transaction, addrs ...dax.Address) error

	CreateJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, addr dax.Address, job ...dax.Job) error
	DeleteJob(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, addr dax.Address, job dax.Job) error
	DeleteJobsForTable(tx dax.Transaction, roleType dax.RoleType, qtid dax.QualifiedTableID) (InternalDiffs, error)
	JobCounts(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, addr ...dax.Address) (map[dax.Address]int, error)
	ListJobs(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, addr dax.Address) (dax.Jobs, error)

	DatabaseForWorker(tx dax.Transaction, addr dax.Address) dax.DatabaseKey
}