kvserver: introduce a Raft-based transport for closedts

pkg/clusterversion/cockroach_versions.go

@@ -252,6 +252,9 @@ const (
  // database, such as adding REGIONS to a DATABASE or setting the LOCALITY
  // on a TABLE.
  MultiRegionFeatures
+ // ClosedTimestampsRaftTransport enables the Raft transport for closed
+ // timestamps and disables the previous per-node transport.
+ ClosedTimestampsRaftTransport
 
  // Step (1): Add new versions here.
 )
@@ -426,6 +429,10 @@ var versionsSingleton = keyedVersions([]keyedVersion{
  Key: MultiRegionFeatures,
  Version: roachpb.Version{Major: 20, Minor: 2, Internal: 34},
  },
+ {
+ Key: ClosedTimestampsRaftTransport,
+ Version: roachpb.Version{Major: 20, Minor: 2, Internal: 36},
+ },
  // Step (2): Add new versions here.
 })
 

pkg/kv/kvserver/BUILD.bazel

@@ -117,6 +117,7 @@ go_library(
  "//pkg/kv/kvserver/closedts/container",
  "//pkg/kv/kvserver/closedts/ctpb",
  "//pkg/kv/kvserver/closedts/storage",
+ "//pkg/kv/kvserver/closedts/tracker",
  "//pkg/kv/kvserver/concurrency",
  "//pkg/kv/kvserver/constraint",
  "//pkg/kv/kvserver/gc",
@@ -299,6 +300,7 @@ go_test(
  "//pkg/kv/kvserver/batcheval/result",
  "//pkg/kv/kvserver/closedts",
  "//pkg/kv/kvserver/closedts/ctpb",
+ "//pkg/kv/kvserver/closedts/tracker",
  "//pkg/kv/kvserver/concurrency",
  "//pkg/kv/kvserver/concurrency/lock",
  "//pkg/kv/kvserver/constraint",

pkg/kv/kvserver/batcheval/cmd_subsume.go

@@ -151,6 +151,15 @@ func Subsume(
  reply.MVCCStats = cArgs.EvalCtx.GetMVCCStats()
  reply.LeaseAppliedIndex = lai
  reply.FreezeStart = cArgs.EvalCtx.Clock().NowAsClockTimestamp()
+ // FrozenClosedTimestamp might return an empty timestamp if the Raft-based
+ // closed timestamp transport hasn't been enabled yet. That's OK because, if
+ // the new transport is not enabled, then ranges with leading closed
+ // timestamps can't exist yet, and so the closed timestamp must be below the
+ // FreezeStart. The FreezeStart is used by Store.MergeRange to bump the RHS'
+ // ts cache if LHS/RHS leases are not collocated. The case when the leases are
+ // collocated also works out because then the closed timestamp (according to
+ // the old mechanism) is the same for both ranges being merged.
+ reply.ClosedTimestamp = cArgs.EvalCtx.FrozenClosedTimestamp(ctx)
 
  return result.Result{
  Local: result.LocalResult{FreezeStart: reply.FreezeStart.ToTimestamp()},

pkg/kv/kvserver/batcheval/eval_context.go

@@ -68,6 +68,7 @@ type EvalContext interface {
  GetTerm(uint64) (uint64, error)
  GetLeaseAppliedIndex() uint64
  GetTracker() closedts.TrackerI
+ FrozenClosedTimestamp(ctx context.Context) hlc.Timestamp
 
  Desc() *roachpb.RangeDescriptor
  ContainsKey(key roachpb.Key) bool
@@ -184,6 +185,9 @@ func (m *mockEvalCtxImpl) GetLeaseAppliedIndex() uint64 {
 func (m *mockEvalCtxImpl) GetTracker() closedts.TrackerI {
  panic("unimplemented")
 }
+func (m *mockEvalCtxImpl) FrozenClosedTimestamp(ctx context.Context) hlc.Timestamp {
+ panic("unimplemented")
+}
 func (m *mockEvalCtxImpl) Desc() *roachpb.RangeDescriptor {
  return m.MockEvalCtx.Desc
 }

pkg/kv/kvserver/below_raft_protos_test.go

@@ -77,13 +77,8 @@ var belowRaftGoldenProtos = map[reflect.Type]fixture{
  populatedConstructor: func(r *rand.Rand) protoutil.Message {
  return enginepb.NewPopulatedRangeAppliedState(r, false)
  },
- // The populatedSum has changed from 10390885694280604642 to
- // 7958815789228166749, as of 21.1, due to the addition of the
- // SeparatedIntentCount field in MVCCStats. This field will not actually
- // be populated until all nodes are on 21.1, so there isn't a risk of
- // divergence.
  emptySum: 615555020845646359,
- populatedSum: 7958815789228166749,
+ populatedSum: 3253881774919630461,
  },
  reflect.TypeOf(&raftpb.HardState{}): {
  populatedConstructor: func(r *rand.Rand) protoutil.Message {

pkg/kv/kvserver/client_replica_test.go

@@ -3332,7 +3332,7 @@ func TestProposalOverhead(t *testing.T) {
  // overhead is that users ranges do not have rangefeeds on by default whereas
  // system ranges do.
  const (
- expectedUserOverhead uint32 = 42
+ expectedUserOverhead uint32 = 45
  )
  t.Run("user-key overhead", func(t *testing.T) {
  userKey := tc.ScratchRange(t)

pkg/kv/kvserver/closedts/tracker/heap_tracker.go

@@ -124,3 +124,10 @@ func (h *heapTracker) LowerBound(ctx context.Context) hlc.Timestamp {
  }
  return h.mu.rs[0].ts
 }
+
+// Count is part of the Tracker interface.
+func (h *heapTracker) Count() int {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return h.mu.rs.Len()
+}

pkg/kv/kvserver/closedts/tracker/lockfree_tracker.go

@@ -16,6 +16,7 @@ import (
  "sync/atomic"
 
  "github.com/cockroachdb/cockroach/pkg/util/hlc"
+ "github.com/cockroachdb/cockroach/pkg/util/log"
  "github.com/cockroachdb/cockroach/pkg/util/timeutil"
 )
 
@@ -169,6 +170,9 @@ func (t *lockfreeTracker) Untrack(ctx context.Context, tok RemovalToken) {
  b := tok.(lockfreeToken).b
  // Note that atomic ops are not required here, as we hold the exclusive lock.
  b.refcnt--
+ if b.refcnt < 0 {
+ log.Fatalf(ctx, "negative bucket refcount: %d", b.refcnt)
+ }
  if b.refcnt == 0 {
  // Reset the bucket, so that future Track() calls can create a new one.
  b.ts = 0
@@ -198,6 +202,11 @@ func (t *lockfreeTracker) LowerBound(ctx context.Context) hlc.Timestamp {
  }
 }
 
+// Count is part of the Tracker interface.
+func (t *lockfreeTracker) Count() int {
+ return int(t.b1.refcnt) + int(t.b2.refcnt)
+}
+
 // bucket represent a Tracker bucket: a data structure that coalesces a number
 // of timestamps, keeping track only of their count and minimum.
 //

pkg/kv/kvserver/closedts/tracker/tracker.go

@@ -85,6 +85,11 @@ type Tracker interface {
  // make is that, if no synthethic timestamp is inserted into the tracked set
  // for a while, eventually the LowerBound value will not be synthetic.
  LowerBound(context.Context) hlc.Timestamp
+
+ // Count returns the current size of the tracked set.
+ //
+ // Count cannot be called concurrently with other methods.
+ Count() int
 }
 
 // RemovalToken represents the result of Track: a token to be later used with

pkg/kv/kvserver/kvserverpb/proposer_kv.go

@@ -10,18 +10,36 @@
 
 package kvserverpb
 
-import "math"
+import (
+ "math"
+
+ "github.com/cockroachdb/cockroach/pkg/util/hlc"
+)
 
 var maxRaftCommandFooterSize = (&RaftCommandFooter{
  MaxLeaseIndex: math.MaxUint64,
 }).Size()
 
+var maxClosedTimestampFooterSize = (&ClosedTimestampFooter{
+ ClosedTimestamp: hlc.Timestamp{
+ WallTime: math.MaxInt64,
+ Logical: math.MaxInt32,
+ Synthetic: true,
+ },
+}).Size()
+
 // MaxRaftCommandFooterSize returns the maximum possible size of an
 // encoded RaftCommandFooter proto.
 func MaxRaftCommandFooterSize() int {
  return maxRaftCommandFooterSize
 }
 
+// MaxClosedTimestampFooterSize returns the maximmum possible size of an encoded
+// ClosedTimestampFooter.
+func MaxClosedTimestampFooterSize() int {
+ return maxClosedTimestampFooterSize
+}
+
 // IsZero returns whether all fields are set to their zero value.
 func (r ReplicatedEvalResult) IsZero() bool {
  return r == ReplicatedEvalResult{}