work

Documentation

Index

• func TestMapState(modName string, rng string) storage.ModuleStorageState
• func TestModStateMap(modStates ...storage.ModuleStorageState) (out storage.ModuleStorageStateMap)
• func TestStoreState(modName string, rng string) storage.ModuleStorageState
• type Job
  • func NewJob(storeName string, requestRange *block.Range, requiredModules []string, ...) *Job
  • func TestJob(modName string, rng string, prio int) *Job
  • func TestJobDeps(modName string, rng string, prio int, deps string) *Job
  • func (j *Job) CreateRequest(originalModules *pbsubstreams.Modules) *pbssinternal.ProcessRangeRequest
  • func (j *Job) MarshalLogObject(enc zapcore.ObjectEncoder) error
  • func (j *Job) Matches(moduleName string, blockNum uint64) bool
  • func (j *Job) String() string
• type Plan
  • func BuildNewPlan(ctx context.Context, modulesStateMap storage.ModuleStorageStateMap, ...) (*Plan, error)
  • func TestPlanReadyJobs(jobs ...*Job) *Plan
  • func (p *Plan) MarkDependencyComplete(modName string, upToBlock uint64)
  • func (p *Plan) NextJob() (job *Job, more bool)
  • func (p *Plan) SendInitialProgressMessages(respFunc substreams.ResponseFunc) error
  • func (p *Plan) String() string
• type Prioritizer
• type RemoteWorker
  • func NewRemoteWorker(clientFactory client.InternalClientFactory, logger *zap.Logger) *RemoteWorker
  • func (w *RemoteWorker) ID() string
  • func (w *RemoteWorker) Work(ctx context.Context, request *pbssinternal.ProcessRangeRequest, ...) *Result
• type Result
• type RetryableErr
  • func NewRetryableErr(cause error) *RetryableErr
  • func (r *RetryableErr) Error() string
• type SimpleWorkerFactory
  • func NewWorkerFactoryFromFunc(...) *SimpleWorkerFactory
  • func (f SimpleWorkerFactory) ID() string
  • func (f SimpleWorkerFactory) Work(ctx context.Context, request *pbssinternal.ProcessRangeRequest, ...) *Result
• type Worker
• type WorkerFactory
• type WorkerPool
  • func NewWorkerPool(ctx context.Context, workerCount uint64, workerFactory WorkerFactory) WorkerPool

Functions

func TestMapState

func TestMapState(modName string, rng string) storage.ModuleStorageState

func TestModStateMap

func TestModStateMap(modStates ...storage.ModuleStorageState) (out storage.ModuleStorageStateMap)

func TestStoreState

func TestStoreState(modName string, rng string) storage.ModuleStorageState

Types

type Job

type Job struct {
	ModuleName   string // target
	RequestRange *block.Range
	// contains filtered or unexported fields
}

Job is a single unit of scheduling, meaning it is a request that goes to a remote gRPC service for execution.

func NewJob

func NewJob(storeName string, requestRange *block.Range, requiredModules []string, priority int) *Job

func TestJob

func TestJob(modName string, rng string, prio int) *Job

func TestJobDeps

func TestJobDeps(modName string, rng string, prio int, deps string) *Job

func (*Job) CreateRequest

func (j *Job) CreateRequest(originalModules *pbsubstreams.Modules) *pbssinternal.ProcessRangeRequest

func (*Job) MarshalLogObject

func (j *Job) MarshalLogObject(enc zapcore.ObjectEncoder) error

func (*Job) Matches

func (j *Job) Matches(moduleName string, blockNum uint64) bool

func (*Job) String

func (j *Job) String() string

type Plan

type Plan struct {
	ModulesStateMap storage.ModuleStorageStateMap
	// contains filtered or unexported fields
}

func BuildNewPlan

func BuildNewPlan(ctx context.Context, modulesStateMap storage.ModuleStorageStateMap, subrequestSplitSize, upToBlock uint64, maxJobsAhead uint64, outputGraph *outputmodules.Graph) (*Plan, error)

func TestPlanReadyJobs

func TestPlanReadyJobs(jobs ...*Job) *Plan

func (*Plan) MarkDependencyComplete

func (p *Plan) MarkDependencyComplete(modName string, upToBlock uint64)

func (*Plan) NextJob

func (p *Plan) NextJob() (job *Job, more bool)

func (*Plan) SendInitialProgressMessages

func (p *Plan) SendInitialProgressMessages(respFunc substreams.ResponseFunc) error

func (*Plan) String

func (p *Plan) String() string

type Prioritizer

type Prioritizer interface {
	Sort(jobs []*Job) []*Job
}

type RemoteWorker

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

func NewRemoteWorker

func NewRemoteWorker(clientFactory client.InternalClientFactory, logger *zap.Logger) *RemoteWorker

func (*RemoteWorker) ID

func (w *RemoteWorker) ID() string

func (*RemoteWorker) Work

func (w *RemoteWorker) Work(ctx context.Context, request *pbssinternal.ProcessRangeRequest, respFunc substreams.ResponseFunc) *Result

type Result

type Result struct {
	PartialsWritten []*block.Range
	Error           error
}

type RetryableErr

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

func NewRetryableErr

func NewRetryableErr(cause error) *RetryableErr

func (*RetryableErr) Error

func (r *RetryableErr) Error() string

type SimpleWorkerFactory

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

func NewWorkerFactoryFromFunc

func NewWorkerFactoryFromFunc(f func(ctx context.Context, request *pbssinternal.ProcessRangeRequest, respFunc substreams.ResponseFunc) *Result) *SimpleWorkerFactory

func (SimpleWorkerFactory) ID

func (f SimpleWorkerFactory) ID() string

func (SimpleWorkerFactory) Work

func (f SimpleWorkerFactory) Work(ctx context.Context, request *pbssinternal.ProcessRangeRequest, respFunc substreams.ResponseFunc) *Result

type Worker

type Worker interface {
	ID() string
	Work(ctx context.Context, request *pbssinternal.ProcessRangeRequest, respFunc substreams.ResponseFunc) *Result
}

type WorkerFactory

type WorkerFactory = func(logger *zap.Logger) Worker

The tracer will be provided by the worker pool, on worker creation

type WorkerPool

type WorkerPool interface {
	Borrow(context.Context) Worker
	Return(Worker)
}

func NewWorkerPool

func NewWorkerPool(ctx context.Context, workerCount uint64, workerFactory WorkerFactory) WorkerPool