_addPendingLogs Add the first change(s) from pendingLogs if they're the next sequence. If not, and we've been waiting too long for nextSequence, move nextSequence to skipped queue. Returns the channels that changed. This may return the same channel more than once, channels should be deduplicated be
(ctx context.Context)
| 876 | // Returns the channels that changed. This may return the same channel more than once, channels should be deduplicated |
| 877 | // before notifying the changes. |
| 878 | func (c *changeCache) _addPendingLogs(ctx context.Context) []channels.ID { |
| 879 | // pre allocate slice for changed channels, give size 5 to allow for some headroom so we |
| 880 | // aren't constantly growing the slice |
| 881 | changedChannels := make([]channels.ID, 0, 5) |
| 882 | var isNext bool |
| 883 | |
| 884 | for len(c.pendingLogs) > 0 { |
| 885 | oldestPending := c.pendingLogs[0] |
| 886 | isNext = oldestPending.Sequence == c.nextSequence |
| 887 | |
| 888 | if isNext { |
| 889 | oldestPending = c._popPendingLog(ctx) |
| 890 | changedChannels = append(changedChannels, c._addToCache(ctx, oldestPending)...) |
| 891 | } else if oldestPending.Sequence < c.nextSequence { |
| 892 | // oldest pending is lower than next sequence, should be ignored |
| 893 | base.InfofCtx(ctx, base.KeyCache, "Oldest entry in pending logs %v (%d, %d) is earlier than cache next sequence (%d), ignoring as sequence has already been cached", base.UD(oldestPending.DocID), oldestPending.Sequence, oldestPending.EndSequence, c.nextSequence) |
| 894 | oldestPending = c._popPendingLog(ctx) |
| 895 | |
| 896 | // If the oldestPending was a range that extended past nextSequence, update nextSequence |
| 897 | if oldestPending.IsUnusedRange() && oldestPending.EndSequence >= c.nextSequence { |
| 898 | c.nextSequence = oldestPending.EndSequence + 1 |
| 899 | } |
| 900 | } else if len(c.pendingLogs) > c.options.CachePendingSeqMaxNum || c.pendingLogs[0].TimeReceived.OlderOrEqual(c.options.CachePendingSeqMaxWait) { |
| 901 | // Skip all sequences up to the oldest Pending |
| 902 | c.PushSkipped(ctx, c.nextSequence, oldestPending.Sequence-1) |
| 903 | c.nextSequence = oldestPending.Sequence |
| 904 | } else { |
| 905 | // nextSequence is not in pending logs, and pending logs size/age doesn't trigger skipped sequences |
| 906 | break |
| 907 | } |
| 908 | } |
| 909 | |
| 910 | c.internalStats.pendingSeqLen = len(c.pendingLogs) |
| 911 | |
| 912 | atomic.StoreInt64(&c.lastAddPendingTime, time.Now().UnixNano()) |
| 913 | return changedChannels |
| 914 | } |
| 915 | |
| 916 | // _popPendingLog pops the next pending LogEntry from the c.pendingLogs heap. When the popped entry is an unused range, |
| 917 | // performs a defensive check for duplicates with the next entry in pending. If unused range overlaps with next entry, |