test: add a benchmarking code

Signed-off-by: Manan Gupta <manan@planetscale.com>

Author: GuptaManan100

go/vt/concurrency/message_queue.go

@@ -50,6 +50,13 @@ func (mq *MessageQueue[T]) Send(value T) {
 	mq.cond.Signal() // Notify a waiting receiver.
 }
 
+// Length returns the length.
+func (mq *MessageQueue[T]) Length() int {
+	mq.mu.Lock()
+	defer mq.mu.Unlock()
+	return len(mq.queue)
+}
+
 // Receive fetches a message of type T from the queue.
 // Blocks if no messages are available or until the queue is closed.
 func (mq *MessageQueue[T]) Receive() (T, bool) {

go/vt/discovery/healthcheck_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/test/utils"
+	"vitess.io/vitess/go/vt/concurrency"
 	"vitess.io/vitess/go/vt/grpcclient"
 	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/topo/memorytopo"
@@ -1513,6 +1514,66 @@ func TestConcurrentUpdates(t *testing.T) {
 	}, 5*time.Second, 100*time.Millisecond, "expected all updates to be processed")
 }
 
+// BenchmarkAccess_FastConsumer benchmarks the access time of the healthcheck for a fast consumer.
+func BenchmarkAccess_FastConsumer(b *testing.B) {
+	ctx := context.Background()
+	// reset error counters
+	hcErrorCounters.ResetAll()
+	ts := memorytopo.NewServer(ctx, "cell")
+	defer ts.Close()
+	hc := createTestHc(ctx, ts)
+	// close healthcheck
+	defer hc.Close()
+
+	// Subscribe to the healthcheck with a fast consumer.
+	ch := hc.Subscribe()
+	go func() {
+		for range ch {
+		}
+	}()
+
+	for i := 0; i < b.N; i++ {
+		hc.broadcast(&TabletHealth{})
+	}
+	hc.Unsubscribe(ch)
+	waitForEmptyMessageQueue(hc.subscribers[ch])
+}
+
+// BenchmarkAccess_SlowConsumer benchmarks the access time of the healthcheck for a slow consumer.
+func BenchmarkAccess_SlowConsumer(b *testing.B) {
+	ctx := context.Background()
+	// reset error counters
+	hcErrorCounters.ResetAll()
+	ts := memorytopo.NewServer(ctx, "cell")
+	defer ts.Close()
+	hc := createTestHc(ctx, ts)
+	// close healthcheck
+	defer hc.Close()
+
+	// Subscribe to the healthcheck with a fast consumer.
+	ch := hc.Subscribe()
+	go func() {
+		for range ch {
+			time.Sleep(100 * time.Second)
+		}
+	}()
+
+	for i := 0; i < b.N; i++ {
+		hc.broadcast(&TabletHealth{})
+	}
+	hc.Unsubscribe(ch)
+	waitForEmptyMessageQueue(hc.subscribers[ch])
+}
+
+func waitForEmptyMessageQueue(queue *concurrency.MessageQueue[*TabletHealth]) {
+	for {
+		if queue.Length() == 0 {
+			return
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+}
+
 func tabletDialer(ctx context.Context, tablet *topodatapb.Tablet, _ grpcclient.FailFast) (queryservice.QueryService, error) {
 	connMapMu.Lock()
 	defer connMapMu.Unlock()