pnf

Documentation

Overview

Package pnf processes and then forwards messages belonging to a window. It reads the data from PBQ (which is populated by the `data forwarder`), calls the UDF reduce function, and then forwards to the next ISB. After a successful forward, it invokes `GC` to clean up the PBQ. Since pnf is a reducer, it mutates the watermark. The watermark after the pnf will be the end time of the window.

Index

• type ForwardTask
• type OrderedProcessor
  • func NewOrderedProcessor(ctx context.Context, vertexInstance *dfv1.VertexInstance, ...) *OrderedProcessor
  • func (op *OrderedProcessor) InsertTask(t *ForwardTask)
  • func (op *OrderedProcessor) SchedulePnF(ctx context.Context, partitionID partition.ID) *ForwardTask
  • func (op *OrderedProcessor) Shutdown()

Types

type ForwardTask

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

ForwardTask wraps the `processAndForward`.

type OrderedProcessor

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

OrderedProcessor orders the forwarding of the writeMessages of the execution of the tasks, even though the tasks itself are run concurrently in an out of ordered fashion.

func NewOrderedProcessor

func NewOrderedProcessor(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) *OrderedProcessor

NewOrderedProcessor returns an OrderedProcessor.

func (*OrderedProcessor) InsertTask

func (op *OrderedProcessor) InsertTask(t *ForwardTask)

func (*OrderedProcessor) SchedulePnF

func (op *OrderedProcessor) SchedulePnF(
	ctx context.Context,
	partitionID partition.ID) *ForwardTask

SchedulePnF creates and schedules the PnF routine.

func (*OrderedProcessor) Shutdown

func (op *OrderedProcessor) Shutdown()

Shutdown closes all the partitions of the buffer.