chunkconsumer

Documentation

Index

• Constants
• func JobToChunkLocator(job module.Job) (*chunks.Locator, error)
• type ChunkConsumer
  • func NewChunkConsumer(log zerolog.Logger, metrics module.VerificationMetrics, ...) *ChunkConsumer
  • func (c ChunkConsumer) Check()
  • func (c *ChunkConsumer) Done() <-chan struct{}
  • func (c *ChunkConsumer) NotifyJobIsDone(jobID module.JobID)
  • func (c *ChunkConsumer) Ready() <-chan struct{}
  • func (c *ChunkConsumer) Size() uint
• type ChunkJob
  • func ChunkLocatorToJob(locator *chunks.Locator) *ChunkJob
  • func (j ChunkJob) ID() module.JobID
• type ChunkJobs
  • func (j *ChunkJobs) AtIndex(index uint64) (module.Job, error)
  • func (j *ChunkJobs) Head() (uint64, error)
• type Worker
  • func NewWorker(engine fetcher.AssignedChunkProcessor) *Worker
  • func (w *Worker) Notify(chunkLocatorID flow.Identifier)
  • func (w *Worker) Run(job module.Job) error

Constants

const (
	DefaultJobIndex     = uint64(0)
	DefaultChunkWorkers = uint64(5)
)

Functions

func JobToChunkLocator

func JobToChunkLocator(job module.Job) (*chunks.Locator, error)

Types

type ChunkConsumer

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

ChunkConsumer consumes the jobs from the job queue, and pass it to the Worker for processing. It wraps the generic job consumer in order to be used as a ReadyDoneAware on startup

func NewChunkConsumer

func NewChunkConsumer(
	log zerolog.Logger,
	metrics module.VerificationMetrics,
	processedIndex storage.ConsumerProgress,
	chunksQueue storage.ChunksQueue,
	chunkProcessor fetcher.AssignedChunkProcessor,
	maxProcessing uint64,
) *ChunkConsumer

func (ChunkConsumer) Check

func (c ChunkConsumer) Check()

func (*ChunkConsumer) Done

func (c *ChunkConsumer) Done() <-chan struct{}

func (*ChunkConsumer) NotifyJobIsDone

func (c *ChunkConsumer) NotifyJobIsDone(jobID module.JobID)

func (*ChunkConsumer) Ready

func (c *ChunkConsumer) Ready() <-chan struct{}

func (*ChunkConsumer) Size

func (c *ChunkConsumer) Size() uint

Size returns number of in-memory chunk jobs that chunk consumer is processing.

type ChunkJob

type ChunkJob struct {
	ChunkLocator *chunks.Locator
}

ChunkJob converts a chunk locator into a Job to be used by job queue.

func ChunkLocatorToJob

func ChunkLocatorToJob(locator *chunks.Locator) *ChunkJob

func (ChunkJob) ID

func (j ChunkJob) ID() module.JobID

ID converts chunk locator identifier into job id, which guarantees uniqueness.

type ChunkJobs

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

ChunkJobs wraps the storage layer to provide an abstraction for consumers to read jobs.

func (*ChunkJobs) AtIndex

func (j *ChunkJobs) AtIndex(index uint64) (module.Job, error)

func (*ChunkJobs) Head

func (j *ChunkJobs) Head() (uint64, error)

type Worker

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

Worker receives job from job consumer and converts it back to Chunk for engine to process

func NewWorker

func NewWorker(engine fetcher.AssignedChunkProcessor) *Worker

func (*Worker) Notify

func (w *Worker) Notify(chunkLocatorID flow.Identifier)

func (*Worker) Run

func (w *Worker) Run(job module.Job) error

Run converts the job to Chunk, it's guaranteed to work, because ChunkJobs converted chunk into job symmetrically