raft: introduce term cache

[pav-kv]

Loading an entry term is expensive. Worst case it boils down to loading a full entry (which can be large), only to read its term.

Term checks are ubiquitous:

• Every commit on the leader does one.
• Every log append does a bunch.
• Every StateProbe matching between the leader and follower does a bunch.

We are hopefully saved by the raft log cache, but it doesn’t save us when some followers get behind. Also we’re interfering with other access patterns to this cache, and it would be nice to not use it for term checks since the latter can be done more efficiently.

The terms in the log are ordered, and typically have long runs like: t1 t1 t1 t1 t1 t2 t2 t2 t2. We could maintain a short list / “index” next to the log: a sorted slice of term change positions. Then getting a term of a log entry boils down to a search in this list. This list can be in memory only, or backed by storage so that we don’t start cold. It will typically be short (<= 2 term flips) because a) the log is compacted quickly, b) leader changes are rare. In memory, we could also limit it to like <= 4 latest term flips.

Once this is done, another thing to do with it is reducing StateProbe latency, e.g. post-election. Right now, StateProbe handshake can be rejected with a hint that only communicates a term of a single raft log position. So, potentially, StateProbe needs to do a few roundtrips before it converges to a fork point of the follower’s log and starts actually replicating. We can extend the hint: send a suffix of the term cache, so that this matching has a better chance to succeed in one go.

Thirdly, we can reduce post-election tail latency. Right now, when a leader is elected, it assumes all followers in StateProbe. It appends a dummy entry to the log and broadcasts a single append to each follower. Only when a follower replies to this initial append, the leader exits StateProbe and starts replicating actively. However, we can pipeline this probing into election: when voting, a voter can attach a short suffix of its term cache; the elected leader will likely immediately know the fork point of the follower's log and can enter StateReplicate immediately (at least for the quorum whose votes it received).

Finally, this allows us to make StateProbe messages empty. Today, we include a bunch of log entries into probe messages (optimistically, to reduce replication latency by 1 RTT). The problem is that StateProbe messages are not subject to admission control, so this can lead to overload/OOM (when there are many ranges probing simultaneously). With faster/pipelined probing, it no longer needs to be done.

Jira issue: CRDB-44961

[hakuuww]

for eventually persisting the term cache to hardstate, I think I can take some inspiration from section 3.1.2 of Eng Design Doc: Leader Leases which talks about how Lead and LeadEpoch are persisted

[hakuuww]

just for brain storming:
When entries are sent to storage for persisting, there is a metric, TotalDuration time.Duration being recorded to log the time it takes for a batch to be persisted on storage.

	// TotalDuration is the time spent in DB.{Apply,ApplyNoSyncWait} or
	// Batch.Commit, plus the time waiting in Batch.SyncWait.

If its too slow, it gets written to logs.

func (r *replicaSyncCallback) OnLogSync)

	if commitStats.TotalDuration > defaultReplicaRaftMuWarnThreshold {
		log.Infof(repl.raftCtx, "slow non-blocking raft commit: %s", commitStats)
	}

---

I think currently it is hard to profile/monitor the latency of loadTerm through stuff like pprof/jaeger. Since it is too far down the stack.

Maybe it is possible to add this type of metric logging/profiling for knowing

• how often a term access operation takes too long, due to a cache miss
• the total latency of loadTerm if there is a cache miss

(like a new metric to log the latency of loadTerm, we can add something similar to logstore.
We can print stuff like the rangeId, the index that caused the latency, etc)

This will make it easier to understand the scenarios that lead to a raftentry.Cache miss in loadTerm.

---

I think cache miss occurs when there are slow followers. But exactly how slow are those followers?

It will typically be short (<= 2 term flips) because a) the log is compacted quickly, b) leader changes are rare. In memory, we could also limit it to like <= 4 latest term flips.

Maybe we can make better decisions in how many terms to keep in the cache if there are more metrics that reflect the issue we are solving.

[hakuuww]

---

(anthony): so the term cache could look like:

[(11, t1), (14, t2), (18, t4), (22, t6)]

A list sorted by entry.index. each entry.index has a term associated with it, signifying "changed to this term at this index"

The most important property to keep in mind is entry.indices are always increasing with no gaps in between

properties: the first entry is not applied to state machine?

Questions:

t will typically be short (<= 2 term flips) because
a) the log is compacted quickly,
b) leader changes are rare.

By compact, do you mean the entries are applied to state machine, not just persisted to storage?

Don't understand why we only need 4 latest term flips, why not more?

I don't see how this is related to leader changes, when there are no leader changes, we can have slow followers, which will be benefitted by the term cache.

• Every log append does a bunch.

For this part, when the leader is trying to append a Entries log slice to the follower. To find a fork point it would potentially need to check for every existing entry in the log.

The important property we are using is:

In a given raftlog at a node at any given time, entry indices are increasing in order, with no gaps in between.

However, terms can have gaps in between, so the cache may be more complicated than we originally thought.
we may need to keep track of
(term->flip index, term->last index before next term)

pseudo code for how it may look like in matchTerm when we are trying to do log append on a follower
this is not taking into account for missing terms in the logs.:

wip
termCache[term] flipIndex

func (l *raftLog) matchTerm(id entryID) bool {

        //check cache first
       use id.term to find the term flip position
       
       // if the term is in the term cache
       if id.term >= termCache.firstTerm && id.term  <= termCache.lastTerm {
                 // (the entry index is greater or equal than the term flip index) AND
                 // ( (the entry index is less than the next term flip index) OR (is less than or equal to the last entry.index in the entire log) )
                 if  id.index >= termCache[id.term] && 
                      ( (id.term < l.lastEntry.term && id.index < termCache[id.term+1]) ||
                      (id.term == l.lastEntryId.term && id.index <= l.lastIndex()))
                 {
                     return true
                 }
        }
        

	t, err := l.term(id.index)
	if err != nil {
		return false
	}
	return t == id.term
}

Also we’re interfering with other access patterns to this cache, and it would be nice to not use it for term checks since the latter can be done more efficiently.

I think this is due to the LRU nature of the eCache, if we access the entry for term, then it updates LRU, and the entry gets re-queued last for eviction. Which may interfere with other usages of the eCache.

---

[pav-kv]

properties: the first entry is not applied to state machine?

Not necessarily. The applied index can be higher than the first log index, potentially equal to the log's lastIndex. We don't have to worry much about this property.

For searching: i think a map can work? termCache[term] flipIndex

This depends on what kinds of queries we want to perform on this data structure. We won't be searching index by term, it's the other way around: search term by index; find a lower-bound/higher-bound index with term <= some value, etc. So a list sorted by index (and, therefore, by term as well) should do well.

By compact, do you mean the entries are applied to state machine, not just persisted to storage?

Log compaction means we're removing a prefix of the log. Yes, it happens some time after these entries have been committed and applied to the state machine.

Don't understand why we only need 4 latest term flips, why not more?

This is arbitrary. We need a balance between the happy and worst case. We would need this list to be big enough to cover the vast majority of cases, so between 2-4 terms sounds reasonable. Not limiting it at all is a bit risky because the list can grow large in the absolute worst case when the leadership is spinning, and term changes are more frequent than log compactions. In the worst case, we could still fall back to reading the term from storage.

I don't see how this is related to leader changes

Leader changes lead to starting a new term, which introduces entries at this new term into the log. So, the more leader changes the more term flips in logs.

For this part, when the leader is trying to append a Entries log slice to the follower. To find a fork point it would potentially need to check for every existing entry in the log.

The point of this issue is to avoid the log scan. Instead of scanning, we use the information in the term cache. The information in the term cache is sufficient to know terms of all entries in a very long suffix of the log.

However, terms can have gaps in between, so the cache may be more complicated than we originally thought.
we may need to keep track of
(term->flip index, term->last index before next term)

Not following this bit. There will be no gaps in the term cache, it will cover a contiguous suffix of the log. The "last index before next term" is equal to the "flip index" of the next term, minus 1.

I think this is due to the LRU nature of the eCache, if we access the entry for term, then it updates LRU, and the entry gets re-queued last for eviction. Which may interfere with other usages of the eCache.

Pretty much, yes. Another detail of the raft log cache is that it keeps a contiguous slice of the log. So, by loading a randomly located entry, we potentially cause a drop of an entire log slice.

[hakuuww]

The point of this issue is to avoid the log scan. Instead of scanning, we use the information in the term cache. The information in the term cache is sufficient to know terms of all entries in a very long suffix of the log.

However, terms can have gaps in between, so the cache may be more complicated than we originally thought.
we may need to keep track of (term->flip index, term->last index before next term)

Not following this bit. There will be no gaps in the term cache, it will cover a contiguous suffix of the log. The "last index before next term" is equal to the "flip index" of the next term, minus 1.

I was thinking about something like in the below scenario. where terms are not continuous.
But now I understand.

For follower logs like (e), the term cache could be [(index1, t1), (index4, t4)].
For use cases like: "log matching when accepting a log append"
if we are given an entry (index2,t2) that we want to check matchTerm with our own log

I see, if we are using a list sorted by index
for the term cache, we can know this entry doesn't exist in the raftLog.

We look up the index in the sorted list, locate index2 is between t1 and t4, so we know it doesn't exist in log, since there are no entries that have term = 2 in the log.

Since each term change must update the term cache.

[hakuuww]

cockroach/pkg/kv/kvserver/replica_application_result.go

Line 500 in e6d48ec

r.shMu.raftTruncState = *t

we can truncate the termCache here, we have access to rawnode.raft.raftlog , since we are holding Replica.mu

when we are updating the raftTruncState. (this indicates we have truncated raftLog in storage

explains Replica.mu and Replica.raftMu:

cockroach/pkg/kv/kvserver/replica_raftstorage.go

Lines 57 to 60 in 2ccaa29

	// replicaRaftStorage implements the raft.Storage interface.
	//
	// All mutating calls to raft.RawNode require that r.mu is held. All read-only
	// calls to raft.RawNode require that r.mu is held at least for reads.

[hakuuww]

size of 4 slice
with truncation
append
snapshot path - clears the log and initializes it to the truncated state

keep it in raftLog

update termCache in func (u *unstable) truncateAndAppend(a LogSlice) bool { where there is an entry with new term, different from last of termCache or raftLog.last

implement the term check in all necessary places

truncate term cache, happens when we update truncate state