Method scan

src/matcher.go:157–240

(request MatchRequest)

}

func (m *Matcher) scan(request MatchRequest) MatchResult {
	startedAt := time.Now()

	numChunks := len(request.chunks)
	if numChunks == 0 {
		m := EmptyMerger(request.revision)
		return MatchResult{m, m, false}
	}
	pattern := request.pattern
	passMerger := PassMerger(&request.chunks, m.tac, request.revision, pattern.startIndex)
	if pattern.IsEmpty() {
		return MatchResult{passMerger, passMerger, false}
	}

	minIndex := request.chunks[0].items[0].Index()
	maxIndex := request.chunks[numChunks-1].lastIndex(minIndex)
	cancelled := util.NewAtomicBool(false)

	numWorkers := min(m.partitions, numChunks)
	var nextChunk atomic.Int32
	resultChan := make(chan partialResult, numWorkers)
	countChan := make(chan int, numChunks)
	waitGroup := sync.WaitGroup{}

	for idx := range numWorkers {
		waitGroup.Add(1)
		if m.slab[idx] == nil {
			m.slab[idx] = util.MakeSlab(slab16Size, slab32Size)
		}
		go func(idx int, slab *util.Slab) {
			defer waitGroup.Done()
			var matches []Result
			for {
				ci := int(nextChunk.Add(1)) - 1
				if ci >= numChunks {
					break
				}
				chunkMatches := request.pattern.Match(request.chunks[ci], slab)
				matches = append(matches, chunkMatches...)
				if cancelled.Get() {
					return
				}
				countChan <- len(chunkMatches)
			}
			if m.sort && request.pattern.sortable {
				m.sortBuf[idx] = radixSortResults(matches, m.tac, m.sortBuf[idx])
			}
			resultChan <- partialResult{idx, matches}
		}(idx, m.slab[idx])
	}

	wait := func() bool {
		cancelled.Set(true)
		waitGroup.Wait()
		return true
	}

	count := 0