Method beginProcessing

timeline/processing.go:49–160

(ctx context.Context, po ProcessingOptions, countOnly bool, done <-chan struct{})

const defaultBatchSize = 10

func (p *processor) beginProcessing(ctx context.Context, po ProcessingOptions, countOnly bool, done <-chan struct{}) (*sync.WaitGroup, chan<- *Graph) {
	wg := new(sync.WaitGroup)
	ch := make(chan *Graph)

	if po.BatchSize <= 0 {
		po.BatchSize = defaultBatchSize
	}

	wg.Add(1)
	go func() {
		defer wg.Done()

		// addToBatch adds g to the batch, and if the batch is full, it
		// sends it for processing and resets the batch. If g is nil,
		// the batch is processed regardless of its size.
		addToBatch := func(g *Graph) {
			var batch []*Graph

			// add the new graph to the batch, keeping track of its actual
			// nested size; and if the batch is now large enough to process,
			// copy it (just the slice header) then reset the batch
			if g != nil {
				p.batch = append(p.batch, g)
				p.batchSize += g.Size()
			}
			if p.batchSize >= po.BatchSize || (g == nil && len(p.batch) > 0) {
				batch = p.batch
				p.batch = make([]*Graph, 0, po.BatchSize)
				p.batchSize = 0
			}

			if len(batch) > 0 {
				err := p.pipeline(ctx, batch)
				if err != nil {
					p.log.Error("batch pipeline", zap.Error(err))
				}

				lastChkptIdx := len(batch) - 1
				for i := len(batch) - 1; i >= 0; i-- {
					if batch[i].Checkpoint != nil {
						lastChkptIdx = i
						break
					}
				}

				var batchSizeToCheckpoint int
				for i := 0; i <= lastChkptIdx; i++ {
					batchSizeToCheckpoint += batch[i].Size()
				}

				p.ij.job.Progress(batchSizeToCheckpoint)

				// persist the last (most recent) checkpoint in the batch
				if batch[lastChkptIdx].Checkpoint != nil {
					if err := p.ij.checkpoint(p.estimatedCount, p.outerLoopIdx, p.innerLoopIdx, batch[lastChkptIdx].Checkpoint); err != nil {
						p.log.Error("checkpointing", zap.Error(err))
					}
				}