Merge #144187 #145555

144187: gossip: batch both client and server gossip messages r=iskettaneh a=iskettaneh

This commit adds an optimization in gossip where before sending a gossip
info update message, we wait for a small amount of time before sending
the updates. This gives some time for updates to be batched, reducing
the number of messages in the system.

Fixes: #119420

Release note: None

145555: sql: disable buffered writes when changing txn isolation r=yuzefovich a=yuzefovich

We currently only support write buffering under serializable isolation, and we explicitly disable that on the txn creation. However, we forgot to do so in one spot where we might be changing the isolation of already created txn, and this commit fixes that oversight.

Epic: None
Release note: None

Co-authored-by: Ibrahim Kettaneh <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: 3 people

docs/generated/metrics/metrics.yaml

@@ -10965,6 +10965,22 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: gossip.messages.received
+      exported_name: gossip_messages_received
+      description: Number of received gossip messages
+      y_axis_label: Messages
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: gossip.messages.sent
+      exported_name: gossip_messages_sent
+      description: Number of sent gossip messages
+      y_axis_label: Messages
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: intentage
       exported_name: intentage
       description: Cumulative age of locks

pkg/gossip/client.go

@@ -219,8 +219,10 @@ func (c *client) sendGossip(g *Gossip, stream Gossip_GossipClient, firstReq bool
 		infosSent := int64(len(delta))
 		c.clientMetrics.BytesSent.Inc(bytesSent)
 		c.clientMetrics.InfosSent.Inc(infosSent)
+		c.clientMetrics.MessagesSent.Inc(1)
 		c.nodeMetrics.BytesSent.Inc(bytesSent)
 		c.nodeMetrics.InfosSent.Inc(infosSent)
+		c.nodeMetrics.MessagesSent.Inc(1)
 
 		if log.V(1) {
 			ctx := c.AnnotateCtx(stream.Context())
@@ -247,8 +249,10 @@ func (c *client) handleResponse(ctx context.Context, g *Gossip, reply *Response)
 	infosReceived := int64(len(reply.Delta))
 	c.clientMetrics.BytesReceived.Inc(bytesReceived)
 	c.clientMetrics.InfosReceived.Inc(infosReceived)
+	c.clientMetrics.MessagesReceived.Inc(1)
 	c.nodeMetrics.BytesReceived.Inc(bytesReceived)
 	c.nodeMetrics.InfosReceived.Inc(infosReceived)
+	c.nodeMetrics.MessagesReceived.Inc(1)
 
 	// Combine remote node's infostore delta with ours.
 	if reply.Delta != nil {
@@ -326,6 +330,7 @@ func (c *client) gossip(
 
 	errCh := make(chan error, 1)
 	initCh := make(chan struct{}, 1)
+
 	// This wait group is used to allow the caller to wait until gossip
 	// processing is terminated.
 	wg.Add(1)
@@ -390,6 +395,9 @@ func (c *client) gossip(
 		case <-initTimer.C:
 			maybeRegister()
 		case <-sendGossipChan:
+			// We need to send the gossip delta to the remote server. Wait a bit to
+			// batch the updates in one message.
+			batchAndConsume(sendGossipChan, infosBatchDelay)
 			if err := c.sendGossip(g, stream, count == 0); err != nil {
 				return err
 			}

pkg/gossip/gossip.go

@@ -118,6 +118,10 @@ const (
 	// we didn't need to tighten the last time we checked.
 	gossipTightenInterval = time.Second
 
+	// infosBatchDelay controls how much time do we wait to batch infos before
+	// sending them.
+	infosBatchDelay = 10 * time.Millisecond
+
 	unknownNodeID roachpb.NodeID = 0
 )
 
@@ -141,6 +145,18 @@ var (
 		Measurement: "Connections",
 		Unit:        metric.Unit_COUNT,
 	}
+	MetaMessagesSent = metric.Metadata{
+		Name:        "gossip.messages.sent",
+		Help:        "Number of sent gossip messages",
+		Measurement: "Messages",
+		Unit:        metric.Unit_COUNT,
+	}
+	MetaMessagesReceived = metric.Metadata{
+		Name:        "gossip.messages.received",
+		Help:        "Number of received gossip messages",
+		Measurement: "Messages",
+		Unit:        metric.Unit_COUNT,
+	}
 	MetaInfosSent = metric.Metadata{
 		Name:        "gossip.infos.sent",
 		Help:        "Number of sent gossip Info objects",

pkg/gossip/gossip.proto

@@ -76,6 +76,8 @@ message MetricSnap {
   int64 bytes_sent = 3;
   int64 infos_received = 4;
   int64 infos_sent = 5;
+  int64 messages_received = 7;
+  int64 messages_sent = 8;
   int64 conns_refused = 6;
 }
 

pkg/gossip/gossip_test.go

@@ -1058,3 +1058,85 @@ func TestServerSendsHighStampsDiff(t *testing.T) {
 		return nil
 	})
 }
+
+// TestGossipBatching verifies that both server and client gossip updates are
+// batched.
+func TestGossipBatching(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	skip.UnderDeadlock(t, "might be flaky since it relies on some upper-bound timing")
+	skip.UnderRace(t, "might be flaky since it relies on some upper-bound timing")
+
+	stopper := stop.NewStopper()
+	defer stopper.Stop(context.Background())
+
+	// Shared cluster ID by all gossipers
+	clusterID := uuid.MakeV4()
+
+	local, localCtx := startGossip(clusterID, 1, stopper, t, metric.NewRegistry())
+	remote, remoteCtx := startGossip(clusterID, 2, stopper, t, metric.NewRegistry())
+	remote.mu.Lock()
+	rAddr := remote.mu.is.NodeAddr
+	remote.mu.Unlock()
+	local.manage(localCtx)
+	remote.manage(remoteCtx)
+
+	// Start a client connection to the remote node
+	local.mu.Lock()
+	local.startClientLocked(rAddr, roachpb.Locality{}, localCtx)
+	local.mu.Unlock()
+
+	// Wait for connection to be established
+	var c *client
+	testutils.SucceedsSoon(t, func() error {
+		c = local.findClient(func(c *client) bool { return c.addr.String() == rAddr.String() })
+		if c == nil {
+			return fmt.Errorf("client not found")
+		}
+		return nil
+	})
+
+	// Prepare 10,000 keys to gossip. This is a large enough number to allow
+	// batching to kick in.
+	numKeys := 10_000
+	localKeys := make([]string, numKeys)
+	remoteKeys := make([]string, numKeys)
+	for i := 0; i < numKeys; i++ {
+		localKeys[i] = fmt.Sprintf("local-key-%d", i)
+		remoteKeys[i] = fmt.Sprintf("remote-key-%d", i)
+	}
+
+	// Gossip the keys to both local and remote nodes.
+	for i := range numKeys {
+		require.NoError(t, local.AddInfo(localKeys[i], []byte("value"), time.Hour))
+		require.NoError(t, remote.AddInfo(remoteKeys[i], []byte("value"), time.Hour))
+	}
+
+	// Wait for updates to propagate
+	testutils.SucceedsSoon(t, func() error {
+		for i := range numKeys {
+			if _, err := local.GetInfo(remoteKeys[i]); err != nil {
+				return err
+			}
+			if _, err := remote.GetInfo(localKeys[i]); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+
+	// Record the number of messages both the client and the server sent, and
+	// assert that it's within the expected bounds.
+	serverMessagesSentCount := remote.serverMetrics.MessagesSent.Count()
+	clientMessagesSentCount := local.serverMetrics.MessagesSent.Count()
+
+	fmt.Printf("client msgs sent: %+v\n", clientMessagesSentCount)
+	fmt.Printf("server msgs sent: %+v\n", serverMessagesSentCount)
+
+	// upperBoundMessages is the maximum number of sent messages we expect to see.
+	// Note that in reality with batching, we see 3-10 messages sent in this test,
+	// However, in order to avoid flakiness, we set a very high number here. The
+	// test would fail even with this high number if we don't have batching.
+	upperBoundMessages := int64(500)
+	require.LessOrEqual(t, serverMessagesSentCount, upperBoundMessages)
+	require.LessOrEqual(t, clientMessagesSentCount, upperBoundMessages)
+}

pkg/gossip/server.go

@@ -125,8 +125,10 @@ func (s *server) Gossip(stream Gossip_GossipServer) error {
 			infoCount := int64(len(reply.Delta))
 			s.nodeMetrics.BytesSent.Inc(bytesSent)
 			s.nodeMetrics.InfosSent.Inc(infoCount)
+			s.nodeMetrics.MessagesSent.Inc(1)
 			s.serverMetrics.BytesSent.Inc(bytesSent)
 			s.serverMetrics.InfosSent.Inc(infoCount)
+			s.serverMetrics.MessagesSent.Inc(1)
 
 			return stream.Send(reply)
 		}
@@ -199,6 +201,9 @@ func (s *server) Gossip(stream Gossip_GossipServer) error {
 		case err := <-errCh:
 			return err
 		case <-ready:
+			// We just sleep here instead of calling batchAndConsume() because the
+			// channel is closed, and sleeping won't block the sender of the channel.
+			time.Sleep(infosBatchDelay)
 		}
 	}
 }
@@ -307,8 +312,10 @@ func (s *server) gossipReceiver(
 		infosReceived := int64(len(args.Delta))
 		s.nodeMetrics.BytesReceived.Inc(bytesReceived)
 		s.nodeMetrics.InfosReceived.Inc(infosReceived)
+		s.nodeMetrics.MessagesReceived.Inc(1)
 		s.serverMetrics.BytesReceived.Inc(bytesReceived)
 		s.serverMetrics.InfosReceived.Inc(infosReceived)
+		s.serverMetrics.MessagesReceived.Inc(1)
 
 		freshCount, err := s.mu.is.combine(args.Delta, args.NodeID)
 		if err != nil {

pkg/gossip/status.go

@@ -26,6 +26,8 @@ type Metrics struct {
 	BytesSent                   *metric.Counter
 	InfosReceived               *metric.Counter
 	InfosSent                   *metric.Counter
+	MessagesSent                *metric.Counter
+	MessagesReceived            *metric.Counter
 	CallbacksProcessed          *metric.Counter
 	CallbacksPending            *metric.Gauge
 	CallbacksProcessingDuration metric.IHistogram
@@ -39,6 +41,8 @@ func makeMetrics() Metrics {
 		BytesSent:          metric.NewCounter(MetaBytesSent),
 		InfosReceived:      metric.NewCounter(MetaInfosReceived),
 		InfosSent:          metric.NewCounter(MetaInfosSent),
+		MessagesReceived:   metric.NewCounter(MetaMessagesReceived),
+		MessagesSent:       metric.NewCounter(MetaMessagesSent),
 		CallbacksProcessed: metric.NewCounter(MetaCallbacksProcessed),
 		CallbacksPending:   metric.NewGauge(MetaCallbacksPending),
 		CallbacksProcessingDuration: metric.NewHistogram(metric.HistogramOptions{
@@ -63,11 +67,13 @@ func (m Metrics) String() string {
 // Snapshot returns a snapshot of the metrics.
 func (m Metrics) Snapshot() MetricSnap {
 	return MetricSnap{
-		ConnsRefused:  m.ConnectionsRefused.Count(),
-		BytesReceived: m.BytesReceived.Count(),
-		BytesSent:     m.BytesSent.Count(),
-		InfosReceived: m.InfosReceived.Count(),
-		InfosSent:     m.InfosSent.Count(),
+		ConnsRefused:     m.ConnectionsRefused.Count(),
+		BytesReceived:    m.BytesReceived.Count(),
+		BytesSent:        m.BytesSent.Count(),
+		InfosReceived:    m.InfosReceived.Count(),
+		InfosSent:        m.InfosSent.Count(),
+		MessagesReceived: m.MessagesReceived.Count(),
+		MessagesSent:     m.MessagesSent.Count(),
 	}
 }
 
@@ -77,7 +83,8 @@ func (m MetricSnap) String() string {
 
 // SafeFormat implements the redact.SafeFormatter interface.
 func (m MetricSnap) SafeFormat(w redact.SafePrinter, _ rune) {
-	w.Printf("infos %d/%d sent/received, bytes %dB/%dB sent/received",
+	w.Printf("messages %d/%d sent/received, infos %d/%d sent/received, bytes %dB/%dB sent/received",
+		m.MessagesSent, m.MessagesReceived,
 		m.InfosSent, m.InfosReceived,
 		m.BytesSent, m.BytesReceived)
 	if m.ConnsRefused > 0 {

pkg/gossip/status_test.go

@@ -21,14 +21,16 @@ func TestGossipStatus(t *testing.T) {
 		},
 		MaxConns: 3,
 		MetricSnap: MetricSnap{
-			BytesReceived: 1000,
-			BytesSent:     2000,
-			InfosReceived: 10,
-			InfosSent:     20,
-			ConnsRefused:  17,
+			BytesReceived:    1000,
+			BytesSent:        2000,
+			InfosReceived:    10,
+			InfosSent:        20,
+			MessagesSent:     2,
+			MessagesReceived: 1,
+			ConnsRefused:     17,
 		},
 	}
-	if exp, act := `gossip server (2/3 cur/max conns, infos 20/10 sent/received, bytes 2000B/1000B sent/received, refused 17 conns)
+	if exp, act := `gossip server (2/3 cur/max conns, messages 2/1 sent/received, infos 20/10 sent/received, bytes 2000B/1000B sent/received, refused 17 conns)
   1: localhost:1234 (17s)
   4: localhost:4567 (18s)
 `, ss.String(); exp != act {
@@ -39,13 +41,13 @@ func TestGossipStatus(t *testing.T) {
 		ConnStatus: []OutgoingConnStatus{
 			{
 				ConnStatus: ss.ConnStatus[0],
-				MetricSnap: MetricSnap{BytesReceived: 77, BytesSent: 88, InfosReceived: 11, InfosSent: 22},
+				MetricSnap: MetricSnap{BytesReceived: 77, BytesSent: 88, InfosReceived: 11, InfosSent: 22, MessagesReceived: 3, MessagesSent: 4},
 			},
 		},
 		MaxConns: 3,
 	}
 	if exp, act := `gossip client (1/3 cur/max conns)
-  1: localhost:1234 (17s: infos 22/11 sent/received, bytes 88B/77B sent/received)
+  1: localhost:1234 (17s: messages 4/3 sent/received, infos 22/11 sent/received, bytes 88B/77B sent/received)
 `, cs.String(); exp != act {
 		t.Errorf("expected:\n%q\ngot:\n%q", exp, act)
 	}

pkg/gossip/util.go

@@ -8,9 +8,11 @@ package gossip
 import (
 	"bytes"
 	"sort"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/config"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 )
 
 // SystemConfigDeltaFilter keeps track of SystemConfig values so that unmodified
@@ -92,3 +94,18 @@ func (df *SystemConfigDeltaFilter) ForModified(
 		}
 	}
 }
+
+// batchAndConsume waits on a channel to allow batching more events. It keeps
+// while consuming events as they come to avoid blocking the channel producer.
+func batchAndConsume(ch <-chan struct{}, batchDuration time.Duration) {
+	var batchTimer timeutil.Timer
+	defer batchTimer.Stop()
+	batchTimer.Reset(batchDuration)
+	for !batchTimer.Read {
+		select {
+		case <-ch: // event happened while we are waiting for our batchTimer to end.
+		case <-batchTimer.C:
+			batchTimer.Read = true
+		}
+	}
+}

pkg/sql/conn_executor.go

@@ -3536,6 +3536,11 @@ func (ex *connExecutor) setTransactionModes(
 		if err := ex.state.setIsolationLevel(level); err != nil {
 			return pgerror.WithCandidateCode(err, pgcode.ActiveSQLTransaction)
 		}
+		if level != isolation.Serializable {
+			// TODO(#143497): we currently only support buffered writes under
+			// serializable isolation.
+			ex.state.mu.txn.SetBufferedWritesEnabled(false)
+		}
 	}
 	rwMode := modes.ReadWriteMode
 	if modes.AsOf.Expr != nil && asOfTs.IsEmpty() {