kv: make disk I/O asynchronous with respect to Raft state machine

[nvb]

⚠️ EDIT: The original optimization proposed here was implemented in #38954. See #17500 (comment) for the remainder of this issue.

Raft, along with most consensus protocols in the Paxos family, distinguishes committed entries from executed entries. An entry is committed when the Raft protocol has completed and the entry has been sufficiently replicated and persisted in the Raft log. The entry is then executed when it is applied to a given replica's state machine. This distinction is important because in cases where the execution of an entry onto the state machine does not produce a result, it is not necessary to wait for execution before sending a commit notification to clients.

Currently in Cockroach, the proposing replica waits until command execution before responding to clients. However, changes made for PropEval KV assured that all MVCC related logic is made upstream of Raft and that by the time we reach entry execution, we're simply applying a WriteBatch to RockDB. While some of these execution steps can create ReplicaCorruptionErrors, I don't think it's necessary or possibly even correct that we attach these errors to the proposal response itself. This is because the entry has already been replicated through Raft, so a local ReplicaCorruptionError doesn't mean that all replicas are corrupted or that the command failed. In fact, after looking at the code I don't think that proposalResult needs anything populated at execution time for correctness. Because of this, I have a suspicion that all of the time spent in applyRaftCommand, including writing to RocksDB and performing stats computations, is unnecessary latency.

Preliminary results from a (very) rough draft of the change show a 2-3% improvement on average latency for the KV benchmark with a --batch size of 4:

Without change (batch=4):

_elapsed___errors_____ops(total)___ops/sec(cum)__avg(ms)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)
  600.0s        0         597164          995.3     32.2     32.5     41.9     65.0    939.5

BenchmarkBlocks	  597164	   1004753.2 ns/op

With change (batch=4):

_elapsed___errors_____ops(total)___ops/sec(cum)__avg(ms)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)
  600.0s        0         612652         1021.1     31.4     30.4     41.9     58.7    604.0

BenchmarkBlocks	  612652	    979351.4 ns/op

With a --batch size of 40, the results were even more pronounced:

Without change (batch=40):

_elapsed___errors_____ops(total)___ops/sec(cum)__avg(ms)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)
  300.0s        0         616040         2053.4    134.0     54.5    151.0   2415.9  10200.5

BenchmarkBlocks	  616040	    486990.5 ns/op

With change (batch=40):

_elapsed___errors_____ops(total)___ops/sec(cum)__avg(ms)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)
  300.0s        0         708040         2360.1    114.7     54.5    130.0   1140.9  10200.5

BenchmarkBlocks	  708040	    423713.8 ns/op

Note that while I think we can respond to clients sooner, I don't think we can pop the command out of the CommandQueue until after it has actually been executed because of how proposer evaluated kv works upstream of Raft.

@tschottdorf @bdarnell

Jira issue: CRDB-6037

Epic CRDB-22644

[andreimatei]

But there's a lease check in processRaftCommand (which is "downstream" of Raft). The result depends on that check...

[nvb]

The lease check happens before applyRaftCommand is called though, which is what I'm trying to avoid waiting on. We can still check the lease before replying like we do now while avoiding the extra latency created by the majority of entry execution. Somewhat accidentally, this is exactly what my PoC branch does.

[petermattis]

I can't think of anything off-hand that would break with this change, but @tschottdorf and @bdarnell know this area of code the best.

[bdarnell]

I think it would be fine to respond to the client before applying the write batch. I'm not sure it's OK to do so before we update the local HLC with the command's timestamp, so I'd move it down a little further.

How does this interact with @irfansharif's #16624? Applying the write batch should be faster when we've decoupled it from the raft log and its synchronous writes, so the benefit of this change may be smaller.

[petermattis]

Mentioned this to @nvanbenschoten in person: a further optimization here would be to respond to all of the committed commands before applying any of the associated write batches. I have an abandoned PR that combined the Raft batches. Probably worth taking another look at it: #15648

[nvb]

@petermattis the only thing to be careful with there is that none of the batches' corresponding commands can be removed from the CommandQueue until we apply the merged WriteBatch. I like the idea though and will investigate because there's some serious overlap here.

I'm not sure how this interacts with #16624, but I would also expect it to reduce the benefit we can gain from this change. Still, right now we're just leaving extra latency on the table, which could be especially detrimental to large Raft batches.

[petermattis]

Note that #16624 isn't making it into 1.1 due to the modest improvement combined with the stability concerns. We should keep #16624 in mind while making changes, but certainly not use it to block nearer term wins.

[bdarnell]

OK, if we've decided to punt #16624 into 1.2, this seems like a good idea.

[petermattis]

Another area for investigation is handling applying Raft commands differently on followers vs the leader. Followers need to keep track of the commit Raft log index, but they don't actually need to apply the commands until they become the leader. At the very least this suggests there are opportunities for batching of the command application.

[bdarnell]

they don't actually need to apply the commands until they become the leader

They need to be caught up (at least with respect to all ChangeReplicas commits) to become a candidate, not just the leader. And I think even as a follower, falling too far behind on ChangeReplicas can be a bad thing. But yes, queueing up changes and batching them on followers can be helpful.

[bdarnell]

Another possibility: Currently, in Replica.handleRaftReady, we write all new log entries and the HardState to disk before sending any messages. This is conservative; some messages can be sent concurrently with the write to disk and this would improve latency by allowing the followers to start their disk syncs before the leader has completed its.

Specifically, a MsgApp can be sent before the Entries it contains have been synced to disk. However, MsgApp also contains the Commit index (stored in the HardState), and it cannot be sent until the corresponding HardState has been persisted (I think). In theory, most MsgApps either contain new Entries or an updated Commit index, but not both, so many MsgApp messages should be able to take advantage of this optimization.