<commit_before>// Package indexer continuously creates an index of the test results
// as the tiles, expectations and ignores change.
package indexer

import (
	"net/url"
	"sync"
	"time"

	"go.skia.org/infra/go/metrics2"
	"go.skia.org/infra/go/paramtools"
	"go.skia.org/infra/go/skerr"
	"go.skia.org/infra/go/sklog"
	"go.skia.org/infra/go/tiling"
	"go.skia.org/infra/go/vcsinfo"
	"go.skia.org/infra/go/vcsinfo/bt_vcs"
	"go.skia.org/infra/golden/go/blame"
	"go.skia.org/infra/golden/go/diff"
	"go.skia.org/infra/golden/go/digest_counter"
	"go.skia.org/infra/golden/go/expstorage"
	"go.skia.org/infra/golden/go/paramsets"
	"go.skia.org/infra/golden/go/pdag"
	"go.skia.org/infra/golden/go/shared"
	"go.skia.org/infra/golden/go/storage"
	"go.skia.org/infra/golden/go/summary"
	"go.skia.org/infra/golden/go/types"
	"go.skia.org/infra/golden/go/warmer"
)

const (
	// Event emitted when the indexer updates the search index.
	// Callback argument: *SearchIndex
	EV_INDEX_UPDATED = "indexer:index-updated"

	// Metric to track the number of digests that do not have be uploaded by bots.
	METRIC_KNOWN_HASHES = "known_digests"
)

// SearchIndex contains everything that is necessary to search
// our current knowledge about test results. It should be
// considered as immutable. Whenever the underlying data changes,
// a new index is calculated via a pdag.
type SearchIndex struct {
	cpxTile           types.ComplexTile
	dCounters         map[types.IgnoreState]digest_counter.DigestCounter
	summaries         map[types.IgnoreState]*summary.Summaries
	paramsetSummaries map[types.IgnoreState]paramsets.ParamSummary
	blamer            blame.Blamer
	warmer            *warmer.Warmer

	// This is set by the indexing pipeline when we just want to update
	// individual tests that have changed.
	testNames []types.TestName
	storages  *storage.Storage

	// Prevents map[types.IgnoreState]foo from being written to at the
	// same time during the main indexer loop. We don'
	mapMutex sync.Mutex
}

// newSearchIndex creates a new instance of SearchIndex. It is not intended to
// be used outside of this package. SearchIndex instances are created by the
// Indexer and retrieved via GetIndex().
func newSearchIndex(storages *storage.Storage, cpxTile types.ComplexTile) *SearchIndex {
	return &SearchIndex{
		cpxTile:   cpxTile,
		dCounters: map[types.IgnoreState]digest_counter.DigestCounter{},
		summaries: map[types.IgnoreState]*summary.Summaries{
			types.ExcludeIgnoredTraces: summary.New(storages),
			types.IncludeIgnoredTraces: summary.New(storages),
		},
		paramsetSummaries: map[types.IgnoreState]paramsets.ParamSummary{},
		blamer:            nil,
		warmer:            warmer.New(storages),
		storages:          storages,
	}
}

// CpxTile returns the current complex tile from which simpler tiles, like one without ignored
// traces can be retrieved
func (idx *SearchIndex) CpxTile() types.ComplexTile {
	return idx.cpxTile
}

// GetIgnoreMatcher returns a matcher for the ignore rules that were used to
// build the tile with ignores.
func (idx *SearchIndex) GetIgnoreMatcher() paramtools.ParamMatcher {
	return idx.cpxTile.IgnoreRules()
}

// Proxy to digest_counter.DigestCounter.ByTest
func (idx *SearchIndex) DigestCountsByTest(is types.IgnoreState) map[types.TestName]digest_counter.DigestCount {
	return idx.dCounters[is].ByTest()
}

// Proxy to digest_counter.DigestCounter.MaxDigestsByTest
func (idx *SearchIndex) MaxDigestsByTest(is types.IgnoreState) map[types.TestName]types.DigestSet {
	return idx.dCounters[is].MaxDigestsByTest()
}

// Proxy to digest_counter.DigestCounter.ByTrace
func (idx *SearchIndex) DigestCountsByTrace(is types.IgnoreState) map[tiling.TraceId]digest_counter.DigestCount {
	return idx.dCounters[is].ByTrace()
}

// ByQuery returns a DigestCount of all the digests that match the given query.
func (idx *SearchIndex) DigestCountsByQuery(query url.Values, is types.IgnoreState) digest_counter.DigestCount {
	return idx.dCounters[is].ByQuery(idx.cpxTile.GetTile(is), query)
}

// Proxy to summary.Summary.Get.
func (idx *SearchIndex) GetSummaries(is types.IgnoreState) map[types.TestName]*summary.Summary {
	return idx.summaries[is].Get()
}

// Proxy to summary.CalcSummaries.
func (idx *SearchIndex) CalcSummaries(testNames []types.TestName, query url.Values, is types.IgnoreState, head bool) (map[types.TestName]*summary.Summary, error) {
	return idx.summaries[is].CalcSummaries(idx.cpxTile.GetTile(is), testNames, query, head)
}

// Proxy to paramsets.Get
func (idx *SearchIndex) GetParamsetSummary(test types.TestName, digest types.Digest, is types.IgnoreState) paramtools.ParamSet {
	return idx.paramsetSummaries[is].Get(test, digest)
}

// Proxy to paramsets.GetByTest
func (idx *SearchIndex) GetParamsetSummaryByTest(is types.IgnoreState) map[types.TestName]map[types.Digest]paramtools.ParamSet {
	return idx.paramsetSummaries[is].GetByTest()
}

// Proxy to blame.Blamer.GetBlame.
func (idx *SearchIndex) GetBlame(test types.TestName, digest types.Digest, commits []*tiling.Commit) *blame.BlameDistribution {
	if idx.blamer == nil {
		// should never happen - indexer should have this initialized
		// before the web server starts serving requests.
		return nil
	}
	return idx.blamer.GetBlame(test, digest, commits)
}

// Indexer is the type that drive continously indexing as the underlying
// data change. It uses a DAG that encodes the dependencies of the
// different components of an index and creates a processing pipeline on top
// of it.
type Indexer struct {
	storages          *storage.Storage
	pipeline          *pdag.Node
	indexTestsNode    *pdag.Node
	writeBaselineNode *pdag.Node
	lastIndex         *SearchIndex
	testNames         []types.TestName
	mutex             sync.RWMutex
}

// New returns a new Indexer instance. It synchronously indexes the initially
// available tile. If the indexing fails an error is returned.
// The provided interval defines how often the index should be refreshed.
func New(storages *storage.Storage, interval time.Duration) (*Indexer, error) {
	ret := &Indexer{
		storages: storages,
	}

	// Set up the processing pipeline.
	root := pdag.NewNodeWithParents(pdag.NoOp)

	// At the top level, Add the DigestCounters...
	countsNodeInclude := root.Child(calcDigestCountsInclude)
	// These are run in parallel because they can take tens of seconds
	// in large repos like Skia.
	countsNodeExclude := root.Child(calcDigestCountsExclude)

	// Node that triggers blame and writing baselines.
	// This is used to trigger when expectations change.
	// We don't need to re-calculate DigestCounts if the
	// expectations change because the DigestCounts don't care about
	// the expectations, only on the tile.
	indexTestsNode := root.Child(pdag.NoOp)

	// ... and invoke the Blamer to calculate the blames.
	blamerNode := indexTestsNode.Child(calcBlame)

	// Write baselines whenever a new tile is processed, new commits become available, or
	// when the expectations change.
	writeBaselineNode := indexTestsNode.Child(writeMasterBaseline)

	// Parameters depend on DigestCounter.
	paramsNodeInclude := pdag.NewNodeWithParents(calcParamsetsInclude, countsNodeInclude)
	// These are run in parallel because they can take tens of seconds
	// in large repos like Skia.
	paramsNodeExclude := pdag.NewNodeWithParents(calcParamsetsExclude, countsNodeExclude)

	// Write known hashes after ignores are computed. DigestCount is a
	// convenient way to get all the hashes, so that's what this node uses.
	writeHashes := countsNodeInclude.Child(writeKnownHashesList)

	// Summaries depend on DigestCounter and Blamer.
	summariesNode := pdag.NewNodeWithParents(calcSummaries, countsNodeInclude, countsNodeExclude, blamerNode)

	// The Warmer depends on summaries.
	pdag.NewNodeWithParents(runWarmer, summariesNode)

	// Set the result on the Indexer instance, once summaries, parameters and writing
	// the hash files is done.
	pdag.NewNodeWithParents(ret.setIndex, summariesNode, paramsNodeInclude, paramsNodeExclude, writeHashes)

	ret.pipeline = root
	ret.indexTestsNode = indexTestsNode
	ret.writeBaselineNode = writeBaselineNode

	// Process the first tile and start the indexing process.
	return ret, ret.start(interval)
}

// GetIndex returns the current index, which is updated continuously in the
// background. The returned instances of SearchIndex can be considered immutable
// and is not going to change. It should be used to handle an entire request
// to provide consistent information.
func (ixr *Indexer) GetIndex() *SearchIndex {
	ixr.mutex.RLock()
	defer ixr.mutex.RUnlock()
	return ixr.lastIndex
}

// start builds the initial index and starts the background
// process to continuously build indices.
func (ixr *Indexer) start(interval time.Duration) error {
	defer shared.NewMetricsTimer("initial_synchronous_index").Stop()
	// Build the first index synchronously.
	tileStream := ixr.storages.GetTileStreamNow(interval, "gold-indexer")
	if err := ixr.executePipeline(<-tileStream); err != nil {
		return err
	}

	// When the master expectations change, update the blamer and its dependents. We choose size
	// 100 because that is plenty capture an unlikely torrent of changes (they are usually triggered
	// by a user).
	expCh := make(chan types.Expectations, 100)
	ixr.storages.EventBus.SubscribeAsync(expstorage.EV_EXPSTORAGE_CHANGED, func(e interface{}) {
		// Schedule the list of test names to be recalculated.
		expCh <- e.(*expstorage.EventExpectationChange).TestChanges
	})

	// When the expectations of a Gerrit issue change then trigger pushing the
	// new expectations to GCS.
	ixr.storages.EventBus.SubscribeAsync(expstorage.EV_TRYJOB_EXP_CHANGED, ixr.writeIssueBaseline)

	// When new commits have become available trigger writing the baselines. We choose size 100
	// because that is large enough to handle an unlikely torrent of commits being added to the repo.
	commitCh := make(chan []*vcsinfo.IndexCommit, 100)
	ixr.storages.EventBus.SubscribeAsync(bt_vcs.EV_NEW_GIT_COMMIT, func(e interface{}) {
		commitCh <- e.([]*vcsinfo.IndexCommit)
	})

	// Keep building indices for different types of events. This is the central
	// event loop of the indexer.
	go func() {
		var cpxTile types.ComplexTile
		for {
			testChanges := []types.Expectations{}

			// See if there is a tile or changed tests.
			cpxTile = nil
			select {
			// Catch a new tile.
			case cpxTile = <-tileStream:
				sklog.Infof("Indexer saw a new tile")

				// Catch any test changes.
			case tn := <-expCh:
				testChanges = append(testChanges, tn)
				sklog.Infof("Indexer saw %d tests change", len(tn))

				// Catch changes in the commits.
			case <-commitCh:
				sklog.Infof("Indexer saw commits change")
			}

			// Drain all input channels, effectively bunching signals together that arrive in short
			// succession.
		DrainLoop:
			for {
				select {
				case tn := <-expCh:
					testChanges = append(testChanges, tn)
				case <-commitCh:
				default:
					break DrainLoop
				}
			}

			// If there is a tile, re-index everything and forget the
			// individual tests that changed.
			if cpxTile != nil {
				if err := ixr.executePipeline(cpxTile); err != nil {
					sklog.Errorf("Unable to index tile: %s", err)
				}
			} else if len(testChanges) > 0 {
				// Only index the tests that have changed.
				ixr.indexTests(testChanges)
			} else {
				// At this point new commits have discovered and we just want to write the baselines.
				ixr.writeBaselines()
			}
		}
	}()

	return nil
}

// executePipeline runs the given tile through the the indexing pipeline.
// pipeline.Trigger blocks until everything is done, so this function will as well.
func (ixr *Indexer) executePipeline(cpxTile types.ComplexTile) error {
	defer shared.NewMetricsTimer("indexer_execute_pipeline").Stop()
	// Create a new index from the given tile.
	return ixr.pipeline.Trigger(newSearchIndex(ixr.storages, cpxTile))
}

// writeBaselines triggers the node that causes baselines to be written to GCS.
func (ixr *Indexer) writeBaselines() {
	idx := ixr.cloneLastIndex()
	if err := ixr.writeBaselineNode.Trigger(idx); err != nil {
		sklog.Errorf("Error writing baselines: %s", err)
	}
}

// indexTest creates an updated index by indexing the given list of expectation changes.
func (ixr *Indexer) indexTests(testChanges []types.Expectations) {
	// Get all the test names that had expectations changed.
	testNames := types.TestNameSet{}
	for _, testChange := range testChanges {
		for testName := range testChange {
			testNames[testName] = true
		}
	}
	tests := testNames.Keys()
	if len(tests) == 0 {
		return
	}

	sklog.Infof("Going to re-index %d tests, starting with %s", len(tests), tests[0])

	defer shared.NewMetricsTimer("index_tests").Stop()
	newIdx := ixr.cloneLastIndex()
	// Set the testNames such that we only recompute those tests.
	newIdx.testNames = tests
	if err := ixr.indexTestsNode.Trigger(newIdx); err != nil {
		sklog.Errorf("Error indexing tests: %v \n\n Got error: %s", tests, err)
	}
}

// cloneLastIndex returns a copy of the most recent index.
func (ixr *Indexer) cloneLastIndex() *SearchIndex {
	lastIdx := ixr.GetIndex()
	return &SearchIndex{
		cpxTile:   lastIdx.cpxTile,
		dCounters: lastIdx.dCounters, // stay the same even if tests change.
		summaries: map[types.IgnoreState]*summary.Summaries{
			types.ExcludeIgnoredTraces: lastIdx.summaries[types.ExcludeIgnoredTraces].Clone(),
			types.IncludeIgnoredTraces: lastIdx.summaries[types.IncludeIgnoredTraces].Clone(),
		},
		paramsetSummaries: lastIdx.paramsetSummaries, //paramsetSummaries are immutable
		blamer:            lastIdx.blamer,            // blamer is immutable and thus, thread-safe.
		warmer:            warmer.New(ixr.storages),
		storages:          lastIdx.storages,

		// Force testNames to be empty, just to be sure we re-compute everything by default
		testNames: nil,
	}
}

// setIndex sets the lastIndex value at the very end of the pipeline.
func (ixr *Indexer) setIndex(state interface{}) error {
	newIndex := state.(*SearchIndex)
	ixr.mutex.Lock()
	defer ixr.mutex.Unlock()
	ixr.lastIndex = newIndex
	if ixr.storages.EventBus != nil {
		ixr.storages.EventBus.Publish(EV_INDEX_UPDATED, state, false)
	}
	return nil
}

// writeIssueBaseline handles changes to baselines for Gerrit issues and dumps
// the updated baseline to disk.
func (ixr *Indexer) writeIssueBaseline(evData interface{}) {
	if !ixr.storages.Baseliner.CanWriteBaseline() {
		return
	}

	issueID := evData.(*expstorage.EventExpectationChange).IssueID
	if issueID <= 0 {
		sklog.Errorf("Invalid issue id received for issue exp change: %d", issueID)
		return
	}

	idx := ixr.GetIndex()
	if err := ixr.storages.Baseliner.PushIssueBaseline(issueID, idx.cpxTile, idx.dCounters[types.ExcludeIgnoredTraces]); err != nil {
		sklog.Errorf("Unable to push baseline for issue %d to GCS: %s", issueID, err)
		return
	}
}

// calcDigestCountsInclude is the pipeline function to calculate DigestCounts from
// the full tile (not applying ignore rules)
func calcDigestCountsInclude(state interface{}) error {
	idx := state.(*SearchIndex)
	idx.mapMutex.Lock()
	defer idx.mapMutex.Unlock()
	is := types.IncludeIgnoredTraces
	idx.dCounters[is] = digest_counter.New(idx.cpxTile.GetTile(is))
	return nil
}

// calcDigestCountsExclude is the pipeline function to calculate DigestCounts from
// the partial tile (applying ignore rules).
func calcDigestCountsExclude(state interface{}) error {
	idx := state.(*SearchIndex)
	idx.mapMutex.Lock()
	defer idx.mapMutex.Unlock()
	is := types.ExcludeIgnoredTraces
	idx.dCounters[is] = digest_counter.New(idx.cpxTile.GetTile(is))
	return nil
}

// calcSummaries is the pipeline function to calculate the summaries.
func calcSummaries(state interface{}) error {
	idx := state.(*SearchIndex)
	for _, is := range types.IgnoreStates {
		err := idx.summaries[is].Calculate(idx.cpxTile.GetTile(is), idx.testNames, idx.dCounters[is], idx.blamer)
		if err != nil {
			return skerr.Fmt("Could not calculate summaries with ignore state %d: %s", is, err)
		}
	}

	return nil
}

// calcParamsetsInclude is the pipeline function to calculate the parameters from
// the full tile (not applying ignore rules)
func calcParamsetsInclude(state interface{}) error {
	idx := state.(*SearchIndex)
	idx.mapMutex.Lock()
	defer idx.mapMutex.Unlock()
	is := types.IncludeIgnoredTraces
	idx.paramsetSummaries[is] = paramsets.NewParamSummary(idx.cpxTile.GetTile(is), idx.dCounters[is])

	return nil
}

// calcParamsetsExclude is the pipeline function to calculate the parameters from
// the partial tile (applying ignore rules)
func calcParamsetsExclude(state interface{}) error {
	idx := state.(*SearchIndex)
	idx.mapMutex.Lock()
	defer idx.mapMutex.Unlock()
	is := types.ExcludeIgnoredTraces
	idx.paramsetSummaries[is] = paramsets.NewParamSummary(idx.cpxTile.GetTile(is), idx.dCounters[is])
	return nil
}

// calcBlame is the pipeline function to calculate the blame.
func calcBlame(state interface{}) error {
	idx := state.(*SearchIndex)
	exp, err := idx.storages.ExpectationsStore.Get()
	if err != nil {
		return skerr.Fmt("Could not fetch expectaions needed to calculate blame: %s", err)
	}<commit_msg>[gold] fix mutex in indexer<commit_after>// Package indexer continuously creates an index of the test results
// as the tiles, expectations and ignores change.
package indexer

import (
	"net/url"
	"sync"
	"time"

	"go.skia.org/infra/go/metrics2"
	"go.skia.org/infra/go/paramtools"
	"go.skia.org/infra/go/skerr"
	"go.skia.org/infra/go/sklog"
	"go.skia.org/infra/go/tiling"
	"go.skia.org/infra/go/vcsinfo"
	"go.skia.org/infra/go/vcsinfo/bt_vcs"
	"go.skia.org/infra/golden/go/blame"
	"go.skia.org/infra/golden/go/diff"
	"go.skia.org/infra/golden/go/digest_counter"
	"go.skia.org/infra/golden/go/expstorage"
	"go.skia.org/infra/golden/go/paramsets"
	"go.skia.org/infra/golden/go/pdag"
	"go.skia.org/infra/golden/go/shared"
	"go.skia.org/infra/golden/go/storage"
	"go.skia.org/infra/golden/go/summary"
	"go.skia.org/infra/golden/go/types"
	"go.skia.org/infra/golden/go/warmer"
)

const (
	// Event emitted when the indexer updates the search index.
	// Callback argument: *SearchIndex
	EV_INDEX_UPDATED = "indexer:index-updated"

	// Metric to track the number of digests that do not have be uploaded by bots.
	METRIC_KNOWN_HASHES = "known_digests"
)

// SearchIndex contains everything that is necessary to search
// our current knowledge about test results. It should be
// considered as immutable. Whenever the underlying data changes,
// a new index is calculated via a pdag.
type SearchIndex struct {
	// The indices of these slices are the int values of types.IgnoreState
	dCounters         []digest_counter.DigestCounter
	summaries         []*summary.Summaries
	paramsetSummaries []paramsets.ParamSummary

	cpxTile types.ComplexTile
	blamer  blame.Blamer
	warmer  *warmer.Warmer

	// This is set by the indexing pipeline when we just want to update
	// individual tests that have changed.
	testNames []types.TestName
	storages  *storage.Storage
}

// newSearchIndex creates a new instance of SearchIndex. It is not intended to
// be used outside of this package. SearchIndex instances are created by the
// Indexer and retrieved via GetIndex().
func newSearchIndex(storages *storage.Storage, cpxTile types.ComplexTile) *SearchIndex {
	return &SearchIndex{
		// The indices of these slices are the int values of types.IgnoreState
		dCounters: make([]digest_counter.DigestCounter, 2),
		summaries: []*summary.Summaries{
			summary.New(storages),
			summary.New(storages),
		},
		paramsetSummaries: make([]paramsets.ParamSummary, 2),
		cpxTile:           cpxTile,
		blamer:            nil,
		warmer:            warmer.New(storages),
		storages:          storages,
	}
}

// CpxTile returns the current complex tile from which simpler tiles, like one without ignored
// traces can be retrieved
func (idx *SearchIndex) CpxTile() types.ComplexTile {
	return idx.cpxTile
}

// GetIgnoreMatcher returns a matcher for the ignore rules that were used to
// build the tile with ignores.
func (idx *SearchIndex) GetIgnoreMatcher() paramtools.ParamMatcher {
	return idx.cpxTile.IgnoreRules()
}

// Proxy to digest_counter.DigestCounter.ByTest
func (idx *SearchIndex) DigestCountsByTest(is types.IgnoreState) map[types.TestName]digest_counter.DigestCount {
	return idx.dCounters[is].ByTest()
}

// Proxy to digest_counter.DigestCounter.MaxDigestsByTest
func (idx *SearchIndex) MaxDigestsByTest(is types.IgnoreState) map[types.TestName]types.DigestSet {
	return idx.dCounters[is].MaxDigestsByTest()
}

// Proxy to digest_counter.DigestCounter.ByTrace
func (idx *SearchIndex) DigestCountsByTrace(is types.IgnoreState) map[tiling.TraceId]digest_counter.DigestCount {
	return idx.dCounters[is].ByTrace()
}

// ByQuery returns a DigestCount of all the digests that match the given query.
func (idx *SearchIndex) DigestCountsByQuery(query url.Values, is types.IgnoreState) digest_counter.DigestCount {
	return idx.dCounters[is].ByQuery(idx.cpxTile.GetTile(is), query)
}

// Proxy to summary.Summary.Get.
func (idx *SearchIndex) GetSummaries(is types.IgnoreState) map[types.TestName]*summary.Summary {
	return idx.summaries[is].Get()
}

// Proxy to summary.CalcSummaries.
func (idx *SearchIndex) CalcSummaries(testNames []types.TestName, query url.Values, is types.IgnoreState, head bool) (map[types.TestName]*summary.Summary, error) {
	return idx.summaries[is].CalcSummaries(idx.cpxTile.GetTile(is), testNames, query, head)
}

// Proxy to paramsets.Get
func (idx *SearchIndex) GetParamsetSummary(test types.TestName, digest types.Digest, is types.IgnoreState) paramtools.ParamSet {
	return idx.paramsetSummaries[is].Get(test, digest)
}

// Proxy to paramsets.GetByTest
func (idx *SearchIndex) GetParamsetSummaryByTest(is types.IgnoreState) map[types.TestName]map[types.Digest]paramtools.ParamSet {
	return idx.paramsetSummaries[is].GetByTest()
}

// Proxy to blame.Blamer.GetBlame.
func (idx *SearchIndex) GetBlame(test types.TestName, digest types.Digest, commits []*tiling.Commit) *blame.BlameDistribution {
	if idx.blamer == nil {
		// should never happen - indexer should have this initialized
		// before the web server starts serving requests.
		return nil
	}
	return idx.blamer.GetBlame(test, digest, commits)
}

// Indexer is the type that drive continously indexing as the underlying
// data change. It uses a DAG that encodes the dependencies of the
// different components of an index and creates a processing pipeline on top
// of it.
type Indexer struct {
	storages          *storage.Storage
	pipeline          *pdag.Node
	indexTestsNode    *pdag.Node
	writeBaselineNode *pdag.Node
	lastIndex         *SearchIndex
	testNames         []types.TestName
	mutex             sync.RWMutex
}

// New returns a new Indexer instance. It synchronously indexes the initially
// available tile. If the indexing fails an error is returned.
// The provided interval defines how often the index should be refreshed.
func New(storages *storage.Storage, interval time.Duration) (*Indexer, error) {
	ret := &Indexer{
		storages: storages,
	}

	// Set up the processing pipeline.
	root := pdag.NewNodeWithParents(pdag.NoOp)

	// At the top level, Add the DigestCounters...
	countsNodeInclude := root.Child(calcDigestCountsInclude)
	// These are run in parallel because they can take tens of seconds
	// in large repos like Skia.
	countsNodeExclude := root.Child(calcDigestCountsExclude)

	// Node that triggers blame and writing baselines.
	// This is used to trigger when expectations change.
	// We don't need to re-calculate DigestCounts if the
	// expectations change because the DigestCounts don't care about
	// the expectations, only on the tile.
	indexTestsNode := root.Child(pdag.NoOp)

	// ... and invoke the Blamer to calculate the blames.
	blamerNode := indexTestsNode.Child(calcBlame)

	// Write baselines whenever a new tile is processed, new commits become available, or
	// when the expectations change.
	writeBaselineNode := indexTestsNode.Child(writeMasterBaseline)

	// Parameters depend on DigestCounter.
	paramsNodeInclude := pdag.NewNodeWithParents(calcParamsetsInclude, countsNodeInclude)
	// These are run in parallel because they can take tens of seconds
	// in large repos like Skia.
	paramsNodeExclude := pdag.NewNodeWithParents(calcParamsetsExclude, countsNodeExclude)

	// Write known hashes after ignores are computed. DigestCount is a
	// convenient way to get all the hashes, so that's what this node uses.
	writeHashes := countsNodeInclude.Child(writeKnownHashesList)

	// Summaries depend on DigestCounter and Blamer.
	summariesNode := pdag.NewNodeWithParents(calcSummaries, countsNodeInclude, countsNodeExclude, blamerNode)

	// The Warmer depends on summaries.
	pdag.NewNodeWithParents(runWarmer, summariesNode)

	// Set the result on the Indexer instance, once summaries, parameters and writing
	// the hash files is done.
	pdag.NewNodeWithParents(ret.setIndex, summariesNode, paramsNodeInclude, paramsNodeExclude, writeHashes)

	ret.pipeline = root
	ret.indexTestsNode = indexTestsNode
	ret.writeBaselineNode = writeBaselineNode

	// Process the first tile and start the indexing process.
	return ret, ret.start(interval)
}

// GetIndex returns the current index, which is updated continuously in the
// background. The returned instances of SearchIndex can be considered immutable
// and is not going to change. It should be used to handle an entire request
// to provide consistent information.
func (ixr *Indexer) GetIndex() *SearchIndex {
	ixr.mutex.RLock()
	defer ixr.mutex.RUnlock()
	return ixr.lastIndex
}

// start builds the initial index and starts the background
// process to continuously build indices.
func (ixr *Indexer) start(interval time.Duration) error {
	defer shared.NewMetricsTimer("initial_synchronous_index").Stop()
	// Build the first index synchronously.
	tileStream := ixr.storages.GetTileStreamNow(interval, "gold-indexer")
	if err := ixr.executePipeline(<-tileStream); err != nil {
		return err
	}

	// When the master expectations change, update the blamer and its dependents. We choose size
	// 100 because that is plenty capture an unlikely torrent of changes (they are usually triggered
	// by a user).
	expCh := make(chan types.Expectations, 100)
	ixr.storages.EventBus.SubscribeAsync(expstorage.EV_EXPSTORAGE_CHANGED, func(e interface{}) {
		// Schedule the list of test names to be recalculated.
		expCh <- e.(*expstorage.EventExpectationChange).TestChanges
	})

	// When the expectations of a Gerrit issue change then trigger pushing the
	// new expectations to GCS.
	ixr.storages.EventBus.SubscribeAsync(expstorage.EV_TRYJOB_EXP_CHANGED, ixr.writeIssueBaseline)

	// When new commits have become available trigger writing the baselines. We choose size 100
	// because that is large enough to handle an unlikely torrent of commits being added to the repo.
	commitCh := make(chan []*vcsinfo.IndexCommit, 100)
	ixr.storages.EventBus.SubscribeAsync(bt_vcs.EV_NEW_GIT_COMMIT, func(e interface{}) {
		commitCh <- e.([]*vcsinfo.IndexCommit)
	})

	// Keep building indices for different types of events. This is the central
	// event loop of the indexer.
	go func() {
		var cpxTile types.ComplexTile
		for {
			testChanges := []types.Expectations{}

			// See if there is a tile or changed tests.
			cpxTile = nil
			select {
			// Catch a new tile.
			case cpxTile = <-tileStream:
				sklog.Infof("Indexer saw a new tile")

				// Catch any test changes.
			case tn := <-expCh:
				testChanges = append(testChanges, tn)
				sklog.Infof("Indexer saw %d tests change", len(tn))

				// Catch changes in the commits.
			case <-commitCh:
				sklog.Infof("Indexer saw commits change")
			}

			// Drain all input channels, effectively bunching signals together that arrive in short
			// succession.
		DrainLoop:
			for {
				select {
				case tn := <-expCh:
					testChanges = append(testChanges, tn)
				case <-commitCh:
				default:
					break DrainLoop
				}
			}

			// If there is a tile, re-index everything and forget the
			// individual tests that changed.
			if cpxTile != nil {
				if err := ixr.executePipeline(cpxTile); err != nil {
					sklog.Errorf("Unable to index tile: %s", err)
				}
			} else if len(testChanges) > 0 {
				// Only index the tests that have changed.
				ixr.indexTests(testChanges)
			} else {
				// At this point new commits have discovered and we just want to write the baselines.
				ixr.writeBaselines()
			}
		}
	}()

	return nil
}

// executePipeline runs the given tile through the the indexing pipeline.
// pipeline.Trigger blocks until everything is done, so this function will as well.
func (ixr *Indexer) executePipeline(cpxTile types.ComplexTile) error {
	defer shared.NewMetricsTimer("indexer_execute_pipeline").Stop()
	// Create a new index from the given tile.
	return ixr.pipeline.Trigger(newSearchIndex(ixr.storages, cpxTile))
}

// writeBaselines triggers the node that causes baselines to be written to GCS.
func (ixr *Indexer) writeBaselines() {
	idx := ixr.cloneLastIndex()
	if err := ixr.writeBaselineNode.Trigger(idx); err != nil {
		sklog.Errorf("Error writing baselines: %s", err)
	}
}

// indexTest creates an updated index by indexing the given list of expectation changes.
func (ixr *Indexer) indexTests(testChanges []types.Expectations) {
	// Get all the test names that had expectations changed.
	testNames := types.TestNameSet{}
	for _, testChange := range testChanges {
		for testName := range testChange {
			testNames[testName] = true
		}
	}
	tests := testNames.Keys()
	if len(tests) == 0 {
		return
	}

	sklog.Infof("Going to re-index %d tests, starting with %s", len(tests), tests[0])

	defer shared.NewMetricsTimer("index_tests").Stop()
	newIdx := ixr.cloneLastIndex()
	// Set the testNames such that we only recompute those tests.
	newIdx.testNames = tests
	if err := ixr.indexTestsNode.Trigger(newIdx); err != nil {
		sklog.Errorf("Error indexing tests: %v \n\n Got error: %s", tests, err)
	}
}

// cloneLastIndex returns a copy of the most recent index.
func (ixr *Indexer) cloneLastIndex() *SearchIndex {
	lastIdx := ixr.GetIndex()
	return &SearchIndex{
		cpxTile:   lastIdx.cpxTile,
		dCounters: lastIdx.dCounters, // stay the same even if tests change.
		summaries: []*summary.Summaries{
			lastIdx.summaries[types.ExcludeIgnoredTraces].Clone(),
			lastIdx.summaries[types.IncludeIgnoredTraces].Clone(),
		},
		paramsetSummaries: lastIdx.paramsetSummaries, //paramsetSummaries are immutable
		blamer:            lastIdx.blamer,            // blamer is immutable and thus, thread-safe.
		warmer:            warmer.New(ixr.storages),
		storages:          lastIdx.storages,

		// Force testNames to be empty, just to be sure we re-compute everything by default
		testNames: nil,
	}
}

// setIndex sets the lastIndex value at the very end of the pipeline.
func (ixr *Indexer) setIndex(state interface{}) error {
	newIndex := state.(*SearchIndex)
	ixr.mutex.Lock()
	defer ixr.mutex.Unlock()
	ixr.lastIndex = newIndex
	if ixr.storages.EventBus != nil {
		ixr.storages.EventBus.Publish(EV_INDEX_UPDATED, state, false)
	}
	return nil
}

// writeIssueBaseline handles changes to baselines for Gerrit issues and dumps
// the updated baseline to disk.
func (ixr *Indexer) writeIssueBaseline(evData interface{}) {
	if !ixr.storages.Baseliner.CanWriteBaseline() {
		return
	}

	issueID := evData.(*expstorage.EventExpectationChange).IssueID
	if issueID <= 0 {
		sklog.Errorf("Invalid issue id received for issue exp change: %d", issueID)
		return
	}

	idx := ixr.GetIndex()
	if err := ixr.storages.Baseliner.PushIssueBaseline(issueID, idx.cpxTile, idx.dCounters[types.ExcludeIgnoredTraces]); err != nil {
		sklog.Errorf("Unable to push baseline for issue %d to GCS: %s", issueID, err)
		return
	}
}

// calcDigestCountsInclude is the pipeline function to calculate DigestCounts from
// the full tile (not applying ignore rules)
func calcDigestCountsInclude(state interface{}) error {
	idx := state.(*SearchIndex)

	is := types.IncludeIgnoredTraces
	idx.dCounters[is] = digest_counter.New(idx.cpxTile.GetTile(is))
	return nil
}

// calcDigestCountsExclude is the pipeline function to calculate DigestCounts from
// the partial tile (applying ignore rules).
func calcDigestCountsExclude(state interface{}) error {
	idx := state.(*SearchIndex)

	is := types.ExcludeIgnoredTraces
	idx.dCounters[is] = digest_counter.New(idx.cpxTile.GetTile(is))
	return nil
}

// calcSummaries is the pipeline function to calculate the summaries.
func calcSummaries(state interface{}) error {
	idx := state.(*SearchIndex)
	for _, is := range types.IgnoreStates {
		err := idx.summaries[is].Calculate(idx.cpxTile.GetTile(is), idx.testNames, idx.dCounters[is], idx.blamer)
		if err != nil {
			return skerr.Fmt("Could not calculate summaries with ignore state %d: %s", is, err)
		}
	}

	return nil
}

// calcParamsetsInclude is the pipeline function to calculate the parameters from
// the full tile (not applying ignore rules)
func calcParamsetsInclude(state interface{}) error {
	idx := state.(*SearchIndex)

	is := types.IncludeIgnoredTraces
	idx.paramsetSummaries[is] = paramsets.NewParamSummary(idx.cpxTile.GetTile(is), idx.dCounters[is])
	return nil
}

// calcParamsetsExclude is the pipeline function to calculate the parameters from
// the partial tile (applying ignore rules)
func calcParamsetsExclude(state interface{}) error {
	idx := state.(*SearchIndex)

	is := types.ExcludeIgnoredTraces
	idx.paramsetSummaries[is] = paramsets.NewParamSummary(idx.cpxTile.GetTile(is), idx.dCounters[is])
	return nil
}

// calcBlame is the pipeline function to calculate the blame.
func calcBlame(state interface{}) error {
	idx := state.(*SearchIndex)
	exp, err := idx.storages.ExpectationsStore.Get()
	if err != nil {
		return skerr.Fmt("Could not fetch expectaions needed to calculate blame: %s", err)
	}
{'fraction_non_alphanumeric': 0.10028687191316948, 'fraction_numerical': 0.001064679265371307, 'mean_word_length': 4.160079352968106, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7121987', 'n_tokens_mistral': 11203, 'n_tokens_neox': 10599, 'n_words': 3199}