queue

Documentation

Index

• Constants
• type Job
• type Jobs
  • func New(db database.Database, metricsNamespace string, ...) (*Jobs, error)
  • func (j *Jobs) Clear() error
  • func (j *Jobs) Commit() error
  • func (j *Jobs) ExecuteAll(ctx *snow.ConsensusContext, halter common.Haltable, restarted bool, ...) (int, error)
  • func (j *Jobs) Has(jobID ids.ID) (bool, error)
  • func (j *Jobs) PendingJobs() uint64
  • func (j *Jobs) Push(job Job) (bool, error)
  • func (j *Jobs) SetParser(parser Parser) error
• type JobsWithMissing
  • func NewWithMissing(db database.Database, metricsNamespace string, ...) (*JobsWithMissing, error)
  • func (jm *JobsWithMissing) AddMissingID(jobIDs ...ids.ID)
  • func (jm *JobsWithMissing) Clear() error
  • func (jm *JobsWithMissing) Commit() error
  • func (jm *JobsWithMissing) Has(jobID ids.ID) (bool, error)
  • func (jm *JobsWithMissing) MissingIDs() []ids.ID
  • func (jm *JobsWithMissing) NumMissingIDs() int
  • func (jm *JobsWithMissing) Push(job Job) (bool, error)
  • func (jm *JobsWithMissing) RemoveMissingID(jobIDs ...ids.ID)
  • func (jm *JobsWithMissing) SetParser(parser Parser) error
• type Parser
• type TestJob
  • func (j *TestJob) Bytes() []byte
  • func (j *TestJob) Default(cant bool)
  • func (j *TestJob) Execute() error
  • func (j *TestJob) HasMissingDependencies() (bool, error)
  • func (j *TestJob) ID() ids.ID
  • func (j *TestJob) MissingDependencies() (set.Set[ids.ID], error)
• type TestParser
  • func (p *TestParser) Default(cant bool)
  • func (p *TestParser) Parse(b []byte) (Job, error)

Constants

const (
	// StatusUpdateFrequency is how many containers should be processed between
	// logs
	StatusUpdateFrequency = 2500
)

Types

type Job

type Job interface {
	ID() ids.ID
	MissingDependencies() (set.Set[ids.ID], error)
	// Returns true if this job has at least 1 missing dependency
	HasMissingDependencies() (bool, error)
	Execute() error
	Bytes() []byte
}

Job defines the interface required to be placed on the job queue.

type Jobs

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

Jobs tracks a series of jobs that form a DAG of dependencies.

func New

func New(
	db database.Database,
	metricsNamespace string,
	metricsRegisterer prometheus.Registerer,
) (*Jobs, error)

New attempts to create a new job queue from the provided database.

func (*Jobs) Clear

func (j *Jobs) Clear() error

func (*Jobs) Commit

func (j *Jobs) Commit() error

Commit the versionDB to the underlying database.

func (*Jobs) ExecuteAll

func (j *Jobs) ExecuteAll(ctx *snow.ConsensusContext, halter common.Haltable, restarted bool, acceptors ...snow.Acceptor) (int, error)

func (*Jobs) Has

func (j *Jobs) Has(jobID ids.ID) (bool, error)

func (*Jobs) PendingJobs

func (j *Jobs) PendingJobs() uint64

Returns how many pending jobs are waiting in the queue.

func (*Jobs) Push

func (j *Jobs) Push(job Job) (bool, error)

Push adds a new job to the queue. Returns true if [job] was added to the queue and false if [job] was already in the queue.

func (*Jobs) SetParser

func (j *Jobs) SetParser(parser Parser) error

SetParser tells this job queue how to parse jobs from the database.

type JobsWithMissing

type JobsWithMissing struct {
	*Jobs
	// contains filtered or unexported fields
}

func NewWithMissing

func NewWithMissing(
	db database.Database,
	metricsNamespace string,
	metricsRegisterer prometheus.Registerer,
) (*JobsWithMissing, error)

func (*JobsWithMissing) AddMissingID

func (jm *JobsWithMissing) AddMissingID(jobIDs ...ids.ID)

AddMissingID adds [jobID] to missingIDs

func (*JobsWithMissing) Clear

func (jm *JobsWithMissing) Clear() error

func (*JobsWithMissing) Commit

func (jm *JobsWithMissing) Commit() error

Commit the versionDB to the underlying database.

func (*JobsWithMissing) Has

func (jm *JobsWithMissing) Has(jobID ids.ID) (bool, error)

func (*JobsWithMissing) MissingIDs

func (jm *JobsWithMissing) MissingIDs() []ids.ID

func (*JobsWithMissing) NumMissingIDs

func (jm *JobsWithMissing) NumMissingIDs() int

func (*JobsWithMissing) Push

func (jm *JobsWithMissing) Push(job Job) (bool, error)

Push adds a new job to the queue. Returns true if [job] was added to the queue and false if [job] was already in the queue.

func (*JobsWithMissing) RemoveMissingID

func (jm *JobsWithMissing) RemoveMissingID(jobIDs ...ids.ID)

RemoveMissingID removes [jobID] from missingIDs

func (*JobsWithMissing) SetParser

func (jm *JobsWithMissing) SetParser(parser Parser) error

SetParser tells this job queue how to parse jobs from the database.

type Parser

type Parser interface {
	Parse([]byte) (Job, error)
}

Parser allows parsing a job from bytes.

type TestJob

type TestJob struct {
	T *testing.T

	CantID,
	CantMissingDependencies,
	CantExecute,
	CantBytes,
	CantHasMissingDependencies bool

	IDF                     func() ids.ID
	MissingDependenciesF    func() (set.Set[ids.ID], error)
	ExecuteF                func() error
	BytesF                  func() []byte
	HasMissingDependenciesF func() (bool, error)
}

TestJob is a test Job

func (*TestJob) Bytes

func (j *TestJob) Bytes() []byte

func (*TestJob) Default

func (j *TestJob) Default(cant bool)

func (*TestJob) Execute

func (j *TestJob) Execute() error

func (*TestJob) HasMissingDependencies

func (j *TestJob) HasMissingDependencies() (bool, error)

func (*TestJob) ID

func (j *TestJob) ID() ids.ID

func (*TestJob) MissingDependencies

func (j *TestJob) MissingDependencies() (set.Set[ids.ID], error)

type TestParser

type TestParser struct {
	T *testing.T

	CantParse bool

	ParseF func([]byte) (Job, error)
}

TestParser is a test Parser

func (*TestParser) Default

func (p *TestParser) Default(cant bool)

func (*TestParser) Parse

func (p *TestParser) Parse(b []byte) (Job, error)