pnf

Documentation

Index

• type Option
  • func WithBatchSize(batchSize int) Option
  • func WithFlushDuration(flushDuration time.Duration) Option
  • func WithGCEventsTracker(gcTracker unaligned.GCEventsWAL) Option
  • func WithWindowType(windowType window.Type) Option
• type ProcessAndForward
  • func NewProcessAndForward(ctx context.Context, vertexInstance *dfv1.VertexInstance, ...) *ProcessAndForward
  • func (pf *ProcessAndForward) AsyncSchedulePnF(ctx context.Context, partitionID *partition.ID, pbq pbq.Reader)
  • func (pf *ProcessAndForward) Shutdown()

Types

type Option

type Option func(options *options) error

func WithBatchSize

func WithBatchSize(batchSize int) Option

WithBatchSize sets the batch size for forwarding messages to ISB.

func WithFlushDuration

func WithFlushDuration(flushDuration time.Duration) Option

WithFlushDuration sets the flush duration for forwarding messages to ISB.

func WithGCEventsTracker

func WithGCEventsTracker(gcTracker unaligned.GCEventsWAL) Option

WithGCEventsTracker sets the GCEventsWAL option

func WithWindowType

func WithWindowType(windowType window.Type) Option

WithWindowType sets the window type option

type ProcessAndForward

type ProcessAndForward struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

ProcessAndForward invokes the UDF for each partition separately in a go routine and writes the response to the ISBs. It also publishes the watermark and invokes GC on PBQ.

func NewProcessAndForward

func NewProcessAndForward(ctx context.Context,
	vertexInstance *dfv1.VertexInstance,
	udf applier.ReduceApplier,
	toBuffers map[string][]isb.BufferWriter,
	pbqManager *pbq.Manager,
	whereToDecider forwarder.ToWhichStepDecider,
	watermarkPublishers map[string]publish.Publisher,
	idleManager wmb.IdleManager,
	windower window.TimedWindower,
	opts ...Option) *ProcessAndForward

NewProcessAndForward returns a new ProcessAndForward.

func (*ProcessAndForward) AsyncSchedulePnF

func (pf *ProcessAndForward) AsyncSchedulePnF(ctx context.Context, partitionID *partition.ID, pbq pbq.Reader)

AsyncSchedulePnF creates a go routine for each partition to invoke the UDF. does not maintain the order of execution between partitions.

func (*ProcessAndForward) Shutdown

func (pf *ProcessAndForward) Shutdown()

Shutdown closes all the partitions of the buffer.