TestChannelCacheHighLoadCache validates behaviour under high query load when the total number of channels is much higher than CompactHighWatermark. Validates that all changes requests return the expected response, even for queries issued while compaction is active.
(t *testing.T)
| 394 | // CompactHighWatermark. Validates that all changes requests return the expected response, even for queries issued while compaction is |
| 395 | // active. |
| 396 | func TestChannelCacheHighLoadCacheMiss(t *testing.T) { |
| 397 | |
| 398 | base.SetUpTestLogging(t, base.LevelWarn, base.KeyCache) |
| 399 | |
| 400 | // Define cache with max channels 100, watermarks 90/70 |
| 401 | options := DefaultCacheOptions().ChannelCacheOptions |
| 402 | options.MaxNumChannels = 100 |
| 403 | options.CompactHighWatermarkPercent = 90 |
| 404 | options.CompactLowWatermarkPercent = 70 |
| 405 | |
| 406 | stats, err := base.NewSyncGatewayStats() |
| 407 | require.NoError(t, err) |
| 408 | dbstats, err := stats.NewDBStats("", false, false, false, nil, nil) |
| 409 | require.NoError(t, err) |
| 410 | testStats := dbstats.Cache() |
| 411 | queryHandler := &testQueryHandler{} |
| 412 | activeChannelStat := &base.SgwIntStat{} |
| 413 | activeChannels := channels.NewActiveChannels(activeChannelStat) |
| 414 | ctx := base.TestCtx(t) |
| 415 | cache, err := newChannelCache(ctx, "testDb", options, queryHandler.asFactory, activeChannels, testStats) |
| 416 | require.NoError(t, err, "Background task error whilst creating channel cache") |
| 417 | defer cache.Stop(ctx) |
| 418 | |
| 419 | channelCount := 200 |
| 420 | // define channel set |
| 421 | channelNames := make([]string, 0) |
| 422 | for i := 1; i <= channelCount; i++ { |
| 423 | channelName := fmt.Sprintf("chan_%d", i) |
| 424 | channelNames = append(channelNames, channelName) |
| 425 | } |
| 426 | |
| 427 | // Seed the query handler with a single doc that's in all the channels |
| 428 | queryEntry := testLogEntryForChannels(1, channelNames) |
| 429 | queryHandler.seedEntries(LogEntries{queryEntry}) |
| 430 | |
| 431 | // Send entry to the cache. Don't reuse queryEntry here, as AddToCache strips out the channels property |
| 432 | logEntry := testLogEntryForChannels(1, channelNames) |
| 433 | cache.AddToCache(ctx, logEntry) |
| 434 | |
| 435 | workerCount := 25 |
| 436 | getChangesCount := 400 |
| 437 | // Start [workerCount] goroutines, each issuing [getChangesCount] changes queries against a random channel |
| 438 | |
| 439 | var workerWg sync.WaitGroup |
| 440 | for w := 0; w < workerCount; w++ { |
| 441 | workerWg.Add(1) |
| 442 | go func() { |
| 443 | changesSuccessCount := 0 |
| 444 | for i := 0; i < getChangesCount; i++ { |
| 445 | channelNumber := rand.Intn(channelCount) + 1 |
| 446 | channel := channels.NewID(fmt.Sprintf("chan_%d", channelNumber), base.DefaultCollectionID) |
| 447 | options := getChangesOptionsWithCtxOnly(t) |
| 448 | changes, err := cache.GetChanges(base.TestCtx(t), channel, options) |
| 449 | if len(changes) == 1 { |
| 450 | changesSuccessCount++ |
| 451 | } |
| 452 | assert.NoError(t, err, fmt.Sprintf("Error getting changes for channel %q", channel)) |
| 453 | assert.True(t, len(changes) == 1, "Expected one change per channel") |
nothing calls this directly
no test coverage detected