orchestrator

Documentation

Index

• Variables
• type MultiSquasher
  • func NewMultiSquasher(ctx context.Context, runtimeConfig config.RuntimeConfig, ...) (*MultiSquasher, error)
  • func (s *MultiSquasher) Launch(ctx context.Context)
  • func (s *MultiSquasher) Squash(ctx context.Context, moduleName string, partialsFiles store.FileInfos) error
  • func (s *MultiSquasher) Wait(ctx context.Context) (out store.Map, err error)
• type NoopMapSquasher
• type ParallelProcessor
  • func BuildParallelProcessor(ctx context.Context, reqDetails *reqctx.RequestDetails, ...) (*ParallelProcessor, error)
  • func (b *ParallelProcessor) Run(ctx context.Context) (storeMap store.Map, err error)
• type Scheduler
  • func NewScheduler(workPlan *work.Plan, respFunc substreams.ResponseFunc, ...) *Scheduler
  • func (s *Scheduler) OnStoreCompletedUntilBlock(storeName string, blockNum uint64)
  • func (s *Scheduler) Schedule(ctx context.Context, pool work.WorkerPool) (err error)
• type StoreSquasher
  • func NewStoreSquasher(initialStore *store.FullKV, ...) *StoreSquasher
  • func (s *StoreSquasher) IsEmpty() bool
  • func (s *StoreSquasher) String() string

Variables

var PartialsChannelClosed = errors.New("partial chunks done")

var SkipFile = errors.New("skip file")

Types

type MultiSquasher

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

MultiSquasher produces _complete_ stores, by merging backing partial stores.

func NewMultiSquasher

func NewMultiSquasher(
	ctx context.Context,
	runtimeConfig config.RuntimeConfig,
	modulesStorageStateMap storage.ModuleStorageStateMap,
	storeConfigs store.ConfigMap,
	upToBlock uint64,
	onStoreCompletedUntilBlock func(storeName string, blockNum uint64),
) (*MultiSquasher, error)

NewMultiSquasher 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 (*MultiSquasher) Launch

func (s *MultiSquasher) Launch(ctx context.Context)

func (*MultiSquasher) Squash

func (s *MultiSquasher) Squash(ctx context.Context, moduleName string, partialsFiles store.FileInfos) error

func (*MultiSquasher) Wait

func (s *MultiSquasher) Wait(ctx context.Context) (out store.Map, err error)

type NoopMapSquasher

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

type ParallelProcessor

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

func BuildParallelProcessor

func BuildParallelProcessor(
	ctx context.Context,
	reqDetails *reqctx.RequestDetails,
	runtimeConfig config.RuntimeConfig,
	outputGraph *outputmodules.Graph,
	execoutStorage *execout.Configs,
	respFunc func(resp substreams.ResponseFromAnyTier) error,
	storeConfigs store.ConfigMap,
	pendingUndoMessage *pbsubstreamsrpc.Response,
) (*ParallelProcessor, error)

BuildParallelProcessor is only called on tier1

func (*ParallelProcessor) Run

func (b *ParallelProcessor) Run(ctx context.Context) (storeMap store.Map, err error)

type Scheduler

type Scheduler struct {
	OnStoreJobTerminated func(ctx context.Context, moduleName string, partialFilesWritten store.FileInfos) error
	// contains filtered or unexported fields
}

func NewScheduler

func NewScheduler(workPlan *work.Plan, respFunc substreams.ResponseFunc, upstreamRequestModules *pbsubstreams.Modules) *Scheduler

func (*Scheduler) OnStoreCompletedUntilBlock

func (s *Scheduler) OnStoreCompletedUntilBlock(storeName string, blockNum uint64)

OnStoreCompletedUntilBlock is called to indicate that the given storeName has snapshots at the `storeSaveIntervals` up to `blockNum` here.

This should unlock all jobs that were dependent

func (*Scheduler) Schedule

func (s *Scheduler) Schedule(ctx context.Context, pool work.WorkerPool) (err error)

type StoreSquasher

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

func NewStoreSquasher

func NewStoreSquasher(
	initialStore *store.FullKV,
	targetExclusiveBlock,
	nextExpectedStartBlock uint64,
	storeSaveInterval uint64,
	onStoreCompletedUntilBlock func(storeName string, blockNum uint64),
) *StoreSquasher

func (*StoreSquasher) IsEmpty

func (s *StoreSquasher) IsEmpty() bool

func (*StoreSquasher) String

func (s *StoreSquasher) String() string