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) *pbsubstreams.Request
  • func (j *Job) MarshalLogObject(enc zapcore.ObjectEncoder) error
  • 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.Factory, logger *zap.Logger) *RemoteWorker
  • func (w *RemoteWorker) Work(ctx context.Context, request *pbsubstreams.Request, ...) *Result
• type Result
• type RetryableErr
  • func (r *RetryableErr) Error() string
• type Worker
• type WorkerFactory
• type WorkerFunc
  • func (f WorkerFunc) Work(ctx context.Context, request *pbsubstreams.Request, ...) *Result
• type WorkerPool
  • func NewWorkerPool(ctx context.Context, workerCount uint64, newWorkerFunc 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) *pbsubstreams.Request

func (*Job) MarshalLogObject

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

func (*Job) String

func (j *Job) String() string

type Plan

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

func BuildNewPlan

func BuildNewPlan(ctx context.Context, modulesStateMap storage.ModuleStorageStateMap, subrequestSplitSize, upToBlock 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.Factory, logger *zap.Logger) *RemoteWorker

func (*RemoteWorker) Work

func (w *RemoteWorker) Work(ctx context.Context, request *pbsubstreams.Request, 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 (*RetryableErr) Error

func (r *RetryableErr) Error() string

type Worker

type Worker interface {
	Work(ctx context.Context, request *pbsubstreams.Request, 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 WorkerFunc

type WorkerFunc func(ctx context.Context, request *pbsubstreams.Request, respFunc substreams.ResponseFunc) *Result

func (WorkerFunc) Work

func (f WorkerFunc) Work(ctx context.Context, request *pbsubstreams.Request, respFunc substreams.ResponseFunc) *Result

type WorkerPool

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

func NewWorkerPool

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