Function startChanges

tools/cache_perf_tool/main.go:246–291

(ctx context.Context, t *testing.T, dbContext *db.DatabaseContext, clientChans int, numClients int, totalSystemChannels int)

}

func startChanges(ctx context.Context, t *testing.T, dbContext *db.DatabaseContext, clientChans int, numClients int, totalSystemChannels int) {
	mutationListener := dbContext.GetMutationListener(t)
	chanIDList := make([]channels.ID, 0, clientChans)
	chanCount := 0
	var chanID channels.ID
	for i := 0; i < numClients; i++ {
		for j := 0; j < clientChans; j++ { // create clientChans number of channels for each change waiter
			if chanCount == totalSystemChannels {
				chanCount = 0 // reset channel count so we don't go over system channels count
			}
			chanID = channels.NewID("test-"+strconv.Itoa(chanCount), base.DefaultCollectionID)
			chanIDList = append(chanIDList, chanID)
			chanCount++
		}
		chans, err := channels.SetOf(chanIDList...)
		if err != nil {
			log.Printf("Error creating channel set: %v", err)
			return
		}
		chanIDList = make([]channels.ID, 0, clientChans) // overwrite the list for next client
		waiter := mutationListener.NewWaiterWithChannels(chans, nil, true)

		go func(ctx context.Context, wait *db.ChangeWaiter, chanMap channels.Set) {
			numGoroutines.Add(1)
			defer numGoroutines.Add(-1)
			for {
				if ctx.Err() != nil {
					return
				}
				num := wait.Wait(ctx)
				if num == db.WaiterClosed {
					return
				} else if num == db.WaiterHasChanges {
					// get cached changes for map, simulating changes feeds actually using the channel cache
					for id := range chanMap {
						_, err := dbContext.GetCachedChanges(t, ctx, id)
						if err != nil {
							log.Printf("Error getting cached changes: %v", err)
							return
						}
					}
				}
			}
		}(ctx, waiter, chans)
	}
}

func printEndofTestStatsFile(ctx context.Context, dbContext *db.DatabaseContext) {
	// Print the csv file to stdout