Method compactChannelCache

db/channel_cache.go:455–555

Compact runs until the number of channels in the cache is lower than compactLowWatermark

(ctx context.Context)

// Compact runs until the number of channels in the cache is lower than compactLowWatermark
func (c *channelCacheImpl) compactChannelCache(ctx context.Context) {
	defer c.compactRunning.Set(false)

	// Increment compact count on start, as timing is updated per loop iteration
	c.cacheStats.ChannelCacheCompactCount.Add(1)

	cacheSize := c.channelCaches.Length()
	base.InfofCtx(ctx, base.KeyCache, "Starting channel cache compaction, size %d", cacheSize)
	for {
		// channelCache close handling
		compactIterationStart := time.Now()

		select {
		case <-c.terminator:
			base.DebugfCtx(ctx, base.KeyCache, "Channel cache compaction stopped due to cache close.")
			return
		default:
			// continue
		}

		// Maintain a target number of items to compact per iteration.  Break the list iteration when the target is reached
		targetEvictCount := cacheSize - c.compactLowWatermark
		if targetEvictCount <= 0 {
			base.InfofCtx(ctx, base.KeyCache, "Stopping channel cache compaction, size %d", cacheSize)
			return
		}
		base.TracefCtx(ctx, base.KeyCache, "Target eviction count: %d (lwm:%d)", targetEvictCount, c.compactLowWatermark)

		// Iterates through cache entries based on cache size at start of compaction iteration loop.  Intentionally
		// ignores channels added during compaction iteration
		inactiveEvictionCandidates := make([]*channels.AppendOnlyListElement, 0)
		nruEvictionCandidates := make([]*channels.AppendOnlyListElement, 0)

		// channelCacheList is an append only list.  Iterator iterates over the current list at the time the iterator was created.
		// Ensures that there are no data races with goroutines appending to the list.
		var elementCount int
		compactCallback := func(elem *channels.AppendOnlyListElement) bool {
			elementCount++
			singleChannelCache, ok := elem.Value.(*singleChannelCacheImpl)
			if !ok {
				base.WarnfCtx(ctx, "Non-cache entry (%T) found in channel cache during compaction - ignoring", elem.Value)
				return true
			}

			// If channel marked as recently used, update recently used flag and continue
			if singleChannelCache.recentlyUsed.IsTrue() {
				singleChannelCache.recentlyUsed.Set(false)
				return true
			}

			// Determine whether NRU channel is active, to establish eviction priority
			isActive := c.activeChannels.IsActive(singleChannelCache.channelID)
			if !isActive {
				base.TracefCtx(ctx, base.KeyCache, "Marking inactive cache entry %q for eviction ", base.UD(singleChannelCache.channelID))
				inactiveEvictionCandidates = append(inactiveEvictionCandidates, elem)
			} else {
				base.TracefCtx(ctx, base.KeyCache, "Marking NRU cache entry %q for eviction", base.UD(singleChannelCache.channelID))
				nruEvictionCandidates = append(nruEvictionCandidates, elem)