execout

package
v1.25.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 8, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Index

Constants

View Source 
const MaxPrefetchDepth = 4

MaxPrefetchDepth is the most segments a walker ever holds or downloads ahead, whatever PrefetchConfig.Depth says.

View Source 
const MetadataDataSize = "datasize"

MetadataDataSize is the object-store metadata key under which the uncompressed size of an execution output file's payloads is recorded. Same key as the one used for store snapshots.

View Source 
const MetadataItemCount = "itemcount"

MetadataItemCount is the object-store metadata key under which the number of items an execution output file holds is recorded. One item is one block the module produced output for, and one `BlockScopedData` message on the wire — which is not the same as one block of the range: a module gated by a block index runs, and is sent, only on matching blocks.

Variables

View Source 
var ErrNotFound = errors.New("inputs module value not found")

Functions

func OutputStats added in v1.22.0 

func OutputStats(ctx context.Context, objStore dstore.Store, logger *zap.Logger, targetRange *block.Range, moduleName string) (size, items uint64, fromMetadata bool, err error)

OutputStats returns the total uncompressed size of the module payloads held in the output file for the given range, and how many items it holds. The item count is what a consumer of that segment receives as `BlockScopedData` messages, so it is what the per-message overhead of a real request has to be multiplied by.

It prefers the figures recorded as object metadata when the file was written, which cost a single attributes lookup; `fromMetadata` then reports true. When the attributes carry neither — a file written before that metadata existed, a backend that does not support metadata at all, or one where recording it is too expensive (see metadataRewriteSchemes) — the file is read and its items added up, which is much more expensive.

A failing attributes lookup is not one of those fallbacks: it is reported as dstore.ErrNotFound, because backends disagree on how they signal a missing object (only GCS maps it to that error) and there is nothing to measure either way, so the caller gets a single condition to handle.

Types

type Buffer 

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

Buffer holds the values produced by modules and exchanged between them as a sort of buffer. Here are the types of exec outputs per module type: 

               values         valuesForFileOutput
---------------------------------------------------
store:         deltas               kvops
mapper:         data              same data
index:          keys                 --

func NewBuffer 

func NewBuffer(blockType string, block *pbbstream.Block, clock *pbsubstreams.Clock) (*Buffer, error)

func (*Buffer) Clock 

func (i *Buffer) Clock() *pbsubstreams.Clock

func (*Buffer) Clone added in v1.17.8 

func (i *Buffer) Clone() ExecutionOutput

func (*Buffer) Get 

func (i *Buffer) Get(moduleName string) (value []byte, cached bool, err error)

func (*Buffer) Len added in v1.4.0 

func (i *Buffer) Len() (out int)

func (*Buffer) Set 

func (i *Buffer) Set(moduleName string, value []byte) (err error)

func (*Buffer) SetFileOutput added in v1.6.0 

func (i *Buffer) SetFileOutput(moduleName string, value []byte) (err error)

type Config 

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

func NewConfig 

func NewConfig(name string, moduleInitialBlock uint64, modKind pbsubstreams.ModuleKind, moduleHash string, extendedModuleHash string, baseStore dstore.Store, logger *zap.Logger) (*Config, error)

func (*Config) ListSnapshotFiles 

func (c *Config) ListSnapshotFiles(ctx context.Context, from uint64, to uint64) (files FileInfos, err error)

func (*Config) ModuleInitialBlock 

func (c *Config) ModuleInitialBlock() uint64

func (*Config) ModuleKind 

func (c *Config) ModuleKind() pbsubstreams.ModuleKind

func (*Config) Name 

func (c *Config) Name() string

func (*Config) NewFile 

func (c *Config) NewFile(targetRange *block.Range) *File

func (*Config) NewFileWriter added in v1.15.8 

func (c *Config) NewFileWriter(ctx context.Context, targetRange *block.Range) FileWriter

func (*Config) OpenFileReader added in v1.15.8 

func (c *Config) OpenFileReader(ctx context.Context, targetRange *block.Range) (FileReader, error)

func (*Config) OutputStats added in v1.22.0 

func (c *Config) OutputStats(ctx context.Context, targetRange *block.Range) (size, items uint64, fromMetadata bool, err error)

OutputStats returns the total uncompressed size of the module payloads held in the output file for the given range, and how many items it holds.

func (*Config) WriteDeterministicError added in v1.15.0 

func (c *Config) WriteDeterministicError(ctx context.Context, atBlock uint64, err error) error

type Configs 

type Configs struct {
	ConfigMap map[string]*Config
	// contains filtered or unexported fields
}

func NewConfigs 

func NewConfigs(baseObjectStore dstore.Store, allRequestedModules []*pbsubstreams.Module, moduleHashes map[string]string, execOutputSaveInterval uint64, firstStreamableBlock uint64, logger *zap.Logger) (*Configs, error)

func WrapConfigs added in v1.13.0 

func WrapConfigs(execOutputSaveInterval uint64, logger *zap.Logger, confs ...*Config) *Configs

func (*Configs) NewFile 

func (c *Configs) NewFile(moduleName string, targetRange *block.Range) *File

func (*Configs) NewFileWalker added in v1.1.9 

func (c *Configs) NewFileWalker(moduleName string, segmenter *block.Segmenter) *FileWalker

type ExecutionOutput 

type ExecutionOutput interface {
	ExecutionOutputGetter
	ExecutionOutputSetter
	ExecutionOutputCloner
}

ExecutionOutput gets/sets execution output for a given graph at a given block

type ExecutionOutputCloner added in v1.17.8 

type ExecutionOutputCloner interface {
	Clone() ExecutionOutput
}

type ExecutionOutputGetter 

type ExecutionOutputGetter interface {
	Len() int
	Clock() *pbsubstreams.Clock
	Get(name string) (value []byte, cached bool, err error)
}

type ExecutionOutputSetter 

type ExecutionOutputSetter interface {
	Set(name string, value []byte) (err error)
	SetFileOutput(name string, value []byte) (err error)
}

type File 

type File struct {
	*block.Range
	// contains filtered or unexported fields
}

A File in `execout` stores, for a given module (with a given hash), the outputs of module execution for _multiple blocks_, based on their block ID.

func (*File) Filename 

func (c *File) Filename() string

func (*File) FullFilename added in v1.9.0 

func (c *File) FullFilename() string

func (*File) MarshalLogObject 

func (c *File) MarshalLogObject(enc zapcore.ObjectEncoder) error

func (*File) ModuleName 

func (c *File) ModuleName() string

func (*File) String 

func (c *File) String() string

type FileInfo 

type FileInfo struct {
	Filename   string
	BlockRange *block.Range
}

type FileInfos 

type FileInfos = []*FileInfo

type FileReader added in v1.15.8 

type FileReader interface {
	ReadNext() (*pboutput.Item, error)
	Iter() iter.Seq2[*pboutput.Item, error]
	Get(ctx context.Context, blockNumber uint64) (payload []byte, found bool, err error)
	ModuleName() string
	Filename() string
	Close() error
}

func OpenFileReader added in v1.15.8 

func OpenFileReader(ctx context.Context, store dstore.Store, logger *zap.Logger, rng *block.Range, moduleName string) (FileReader, error)

type FileWalker added in v1.1.9 

type FileWalker struct {
	IsLocal bool
	// contains filtered or unexported fields
}

FileWalker allows you to jump from file to file, from segment to segment

func NewFileWalker added in v1.6.2 

func NewFileWalker(c *Config, segmenter *block.Segmenter, logger *zap.Logger) *FileWalker

func (*FileWalker) FileReader added in v1.15.8 

func (fw *FileWalker) FileReader(ctx context.Context) (FileReader, error)

If the current segment is out of ranges, returns nil.

func (*FileWalker) FileWriter added in v1.15.8 

func (fw *FileWalker) FileWriter(ctx context.Context) FileWriter

If the current segment is out of ranges, returns nil.

func (*FileWalker) IsDone added in v1.1.9 

func (fw *FileWalker) IsDone() bool

func (*FileWalker) Next added in v1.1.9 

func (fw *FileWalker) Next()

func (*FileWalker) Progress added in v1.3.2 

func (fw *FileWalker) Progress() (first, current, last int)

func (*FileWalker) WithPrefetch added in v1.23.0 

func (fw *FileWalker) WithPrefetch(cfg PrefetchConfig) *FileWalker

WithPrefetch makes FileReader download the segments following the current one in the background, within the bounds of cfg. Prefetching starts on the first FileReader call and lives as long as that call's context.

type FileWriter added in v1.15.8 

type FileWriter interface {
	SetItem(clock *pbsubstreams.Clock, data []byte) error
	Close() error
	ModuleName() string
	Filename() string
	Range() *block.Range
}

func NewFileWriter added in v1.15.8 

func NewFileWriter(ctx context.Context, store dstore.Store, logger *zap.Logger, rng *block.Range, moduleName string) FileWriter

type PrefetchConfig added in v1.23.0 

type PrefetchConfig struct {
	// Depth is the number of segments that may be held or in flight at once,
	// capped at MaxPrefetchDepth.
	Depth int
	// BudgetBytes is the total decompressed size of the segments held in memory.
	BudgetBytes uint64
}

PrefetchConfig bounds how far ahead of the segment being streamed a FileWalker downloads execution output files, and how much decompressed data it may hold in memory while doing so. A zero Depth or BudgetBytes disables prefetching.

type Writer 

type Writer struct {
	CurrentFile FileWriter
	// contains filtered or unexported fields
}

The Writer writes a single file with executionOutputs that will be read by the LinearExecOutReader. `initialBlockBoundary` is expected to be on a boundary, or to be the module's initial block.

func NewWriter 

func NewWriter(ctx context.Context, initialBlockBoundary, exclusiveEndBlock uint64, outputModule string, configs *Configs) *Writer

func (*Writer) Close 

func (w *Writer) Close() error

func (*Writer) Write 

func (w *Writer) Write(clock *pbsubstreams.Clock, buffer *Buffer) error

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.