orchestrator

Documentation

Index

• type Job
  • func NewJob(storeName string, requestRange *block.Range, ...) *Job
  • func (j *Job) MarshalLogObject(enc zapcore.ObjectEncoder) error
  • func (j *Job) String() string
• type JobsPlanner
  • func NewJobsPlanner(ctx context.Context, workPlan WorkPlan, subrequestSplitSize uint64, ...) (*JobsPlanner, error)
  • func (p *JobsPlanner) JobCount() int
  • func (p *JobsPlanner) MarshalLogObject(enc zapcore.ObjectEncoder) error
  • func (p *JobsPlanner) SignalCompletionUpUntil(storeName string, blockNum uint64)
• type Scheduler
  • func NewScheduler(ctx context.Context, availableJobs chan *Job, squasher *Squasher, ...) (*Scheduler, error)
  • func (s *Scheduler) Launch(ctx context.Context, result chan error)
• type Snapshot
• type Snapshots
  • func (s *Snapshots) ContainsPartial(r *block.Range) bool
  • func (s *Snapshots) LastCompleteSnapshotBefore(blockNum uint64) *block.Range
  • func (s *Snapshots) LastCompletedBlock() uint64
  • func (s *Snapshots) Sort()
  • func (s *Snapshots) String() string
• type Squasher
  • func NewSquasher(ctx context.Context, workPlan WorkPlan, stores map[string]*state.Store, ...) (*Squasher, error)
  • func (s *Squasher) Squash(moduleName string, partialsRanges block.Ranges) error
  • func (s *Squasher) ValidateStoresReady() (out map[string]*state.Store, err error)
• type StorageState
  • func FetchStorageState(ctx context.Context, stores map[string]*state.Store) (out *StorageState, err error)
  • func NewStorageState() *StorageState
  • func (s *StorageState) String() string
• type StoreSquasher
  • func NewStoreSquasher(initialStore *state.Store, targetExclusiveBlock, nextExpectedStartBlock uint64, ...) *StoreSquasher
  • func (s *StoreSquasher) IsEmpty() bool
  • func (s *StoreSquasher) String() string
• type WorkPlan
  • func (p WorkPlan) ProgressMessages(hostname string) (out []*pbsubstreams.ModuleProgress)
  • func (p WorkPlan) SquashPartialsPresent(squasher *Squasher) error
  • func (p WorkPlan) String() string
• type WorkUnit
  • func SplitWork(modName string, storeSaveInterval, modInitBlock, incomingReqStartBlock uint64, ...) *WorkUnit
• type Worker
  • func (w *Worker) Run(ctx context.Context, job *Job, respFunc substreams.ResponseFunc) ([]*block.Range, error)
• type WorkerPool
  • func NewWorkerPool(workerCount int, originalRequestModules *pbsubstreams.Modules, ...) *WorkerPool
  • func (p *WorkerPool) Borrow() *Worker
  • func (p *WorkerPool) ReturnWorker(worker *Worker)

Types

type Job

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

func NewJob

func NewJob(storeName string, requestRange *block.Range, ancestorStoreModules []*pbsubstreams.Module, totalJobs, myJobIndex int) *Job

func (*Job) MarshalLogObject

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

func (*Job) String

func (j *Job) String() string

type JobsPlanner

type JobsPlanner struct {
	sync.Mutex

	AvailableJobs chan *Job
	// contains filtered or unexported fields
}

func NewJobsPlanner

func NewJobsPlanner(
	ctx context.Context,
	workPlan WorkPlan,
	subrequestSplitSize uint64,
	stores map[string]*state.Store,
	graph *manifest.ModuleGraph,
) (*JobsPlanner, error)

func (*JobsPlanner) JobCount

func (p *JobsPlanner) JobCount() int

func (*JobsPlanner) MarshalLogObject

func (p *JobsPlanner) MarshalLogObject(enc zapcore.ObjectEncoder) error

func (*JobsPlanner) SignalCompletionUpUntil

func (p *JobsPlanner) SignalCompletionUpUntil(storeName string, blockNum uint64)

type Scheduler

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

func NewScheduler

func NewScheduler(ctx context.Context, availableJobs chan *Job, squasher *Squasher, workerPool *WorkerPool, respFunc substreams.ResponseFunc) (*Scheduler, error)

func (*Scheduler) Launch

func (s *Scheduler) Launch(ctx context.Context, result chan error)

type Snapshot

type Snapshot struct {
	block.Range
	Path string
}

type Snapshots

type Snapshots struct {
	Completes block.Ranges // Shortest completes first, largest last.
	Partials  block.Ranges // First partials first, last last
}

func (*Snapshots) ContainsPartial

func (s *Snapshots) ContainsPartial(r *block.Range) bool

func (*Snapshots) LastCompleteSnapshotBefore

func (s *Snapshots) LastCompleteSnapshotBefore(blockNum uint64) *block.Range

func (*Snapshots) LastCompletedBlock

func (s *Snapshots) LastCompletedBlock() uint64

func (*Snapshots) Sort

func (s *Snapshots) Sort()

func (*Snapshots) String

func (s *Snapshots) String() string

type Squasher

type Squasher struct {
	*shutter.Shutter
	// contains filtered or unexported fields
}

Squasher produces _complete_ stores, by merging backing partial stores.

func NewSquasher

func NewSquasher(ctx context.Context, workPlan WorkPlan, stores map[string]*state.Store, reqStartBlock uint64, jobsPlanner *JobsPlanner) (*Squasher, error)

NewSquasher receives stores, initializes them and fetches them from the existing storage. It prepares itself to receive Squash() requests that should correspond to what is missing for those stores to reach `targetExclusiveEndBlock`. This is managed externally by the Scheduler/Strategy. Eventually, ideally, all components are synchronizes around the actual data: the state of storages present, the requests needed to fill in those stores up to the target block, etc..

func (*Squasher) Squash

func (s *Squasher) Squash(moduleName string, partialsRanges block.Ranges) error

func (*Squasher) ValidateStoresReady

func (s *Squasher) ValidateStoresReady() (out map[string]*state.Store, err error)

type StorageState

type StorageState struct {
	sync.Mutex
	Snapshots map[string]*Snapshots
}

func FetchStorageState

func FetchStorageState(ctx context.Context, stores map[string]*state.Store) (out *StorageState, err error)

func NewStorageState

func NewStorageState() *StorageState

func (*StorageState) String

func (s *StorageState) String() string

type StoreSquasher

type StoreSquasher struct {
	*shutter.Shutter
	// contains filtered or unexported fields
}

func NewStoreSquasher

func NewStoreSquasher(initialStore *state.Store, targetExclusiveBlock, nextExpectedStartBlock uint64, jobsPlanner *JobsPlanner) *StoreSquasher

func (*StoreSquasher) IsEmpty

func (s *StoreSquasher) IsEmpty() bool

func (*StoreSquasher) String

func (s *StoreSquasher) String() string

type WorkPlan

type WorkPlan map[string]*WorkUnit

func (WorkPlan) ProgressMessages

func (p WorkPlan) ProgressMessages(hostname string) (out []*pbsubstreams.ModuleProgress)

func (WorkPlan) SquashPartialsPresent

func (p WorkPlan) SquashPartialsPresent(squasher *Squasher) error

func (WorkPlan) String

func (p WorkPlan) String() string

type WorkUnit

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

func SplitWork

func SplitWork(modName string, storeSaveInterval, modInitBlock, incomingReqStartBlock uint64, snapshots *Snapshots) *WorkUnit

type Worker

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

func (*Worker) Run

func (w *Worker) Run(ctx context.Context, job *Job, respFunc substreams.ResponseFunc) ([]*block.Range, error)

type WorkerPool

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

func NewWorkerPool

func NewWorkerPool(workerCount int, originalRequestModules *pbsubstreams.Modules, grpcClientFactory substreams.GrpcClientFactory) *WorkerPool

func (*WorkerPool) Borrow

func (p *WorkerPool) Borrow() *Worker

func (*WorkerPool) ReturnWorker

func (p *WorkerPool) ReturnWorker(worker *Worker)