Feature: ring buffer stats

[proost]

Summary

Adding ring buffer monitoring stats method. ring buffer monitoring helps to find debugging increasing latency and resource sweet spot.

Problem Definition

Background

rueidis is queuing redis commands to use connections effectively and efficiently with high throughput. But sometime commands queue can be bottleneck of latency. Although rueidis supports RingScaleEachConn and PipelineMultiplex options to tune commands queue, it is not easy to find sweet spot. Because it is hard to monitor commands queue.

Goal

• Monitoring average waiting for write commands in a client.
• Monitoring average waiting for result commands in a client.

High Level Plan

Queue

We should add returning queue stats function. commandQueueStats has state of queue. Using commandQueueStats , we can monitor queue

type queue interface {
	PutOne(m Completed) chan RedisResult
	PutMulti(m []Completed, resps []RedisResult) chan RedisResult
	NextWriteCmd() (Completed, []Completed, chan RedisResult)
	WaitForWrite() (Completed, []Completed, chan RedisResult)
	NextResultCh() (Completed, []Completed, chan RedisResult, []RedisResult, *sync.Cond)
	Stats() commandQueueStats
}

type commandQueueStats struct {
	waitingForWriteCount   uint32 // Counter for node waiting for writing
  waitingForResultCount  uint32 // Counter for node waiting for results
}

Wire

We should add returning queue stats function. wire returns own queue stats to call queue.Stats

type wire interface {
	Do(ctx context.Context, cmd Completed) RedisResult
	DoCache(ctx context.Context, cmd Cacheable, ttl time.Duration) RedisResult
	DoMulti(ctx context.Context, multi ...Completed) *redisresults
	DoMultiCache(ctx context.Context, multi ...CacheableTTL) *redisresults
	Receive(ctx context.Context, subscribe Completed, fn func(message PubSubMessage)) error
	DoStream(ctx context.Context, pool *pool, cmd Completed) RedisResultStream
	DoMultiStream(ctx context.Context, pool *pool, multi ...Completed) MultiRedisResultStream
	Info() map[string]RedisMessage
	Version() int
	Error() error
	Close()
	CommandQueueStats() commandQueueStats

	CleanSubscriptions()
	SetPubSubHooks(hooks PubSubHooks) <-chan error
	SetOnCloseHook(fn func(error))
}

conn

conn can have multiple wire s. So we have to supports how many wire has and sum of waitingForWriteCommandsCount and sum of waitingForResultCommandsCount . Using 3 fields, we can deduce average of waitingForWriteCommandsCount and average of waitingForResultCommandsCount .

type conn interface {
	Do(ctx context.Context, cmd Completed) RedisResult
	DoCache(ctx context.Context, cmd Cacheable, ttl time.Duration) RedisResult
	DoMulti(ctx context.Context, multi ...Completed) *redisresults
	DoMultiCache(ctx context.Context, multi ...CacheableTTL) *redisresults
	Receive(ctx context.Context, subscribe Completed, fn func(message PubSubMessage)) error
	DoStream(ctx context.Context, cmd Completed) RedisResultStream
	DoMultiStream(ctx context.Context, multi ...Completed) MultiRedisResultStream
	Info() map[string]RedisMessage
	Version() int
	Error() error
	Close()
	Dial() error
	Override(conn)
	Acquire() wire
	Store(w wire)
	Addr() string
	SetOnCloseHook(func(error))
	AggregatedCommandQueueStats() aggregatedCommandQueueStats
}

type aggregatedCommandQueueStats struct {
	totalConns uint64 // total wire counts underlying conn.
	totalWaitingForWriteCommands uint64 // sum of WaitingForWriteCommandsCount.
	totalWaitingForResultCommadns uint64 // sum of WaitingForResultCommandsCount.
}

Client

We define new interface ClientStatsProvider . ClientStatsProvider returns Stats of client.

We can’t add Stats() method to Client directly. Because adding new method can be breaking change to some implementations of Client interface .

type ClientStatsProvider interface {
	Stats() Stats
}

type Stats struct {
	TotalConns uint64
	TotalWaitingForWriteCommands uint64
	TotalWaitingForResultCommands uint64
}

Detailed Plan

ring

ring struct has two fields waitingForWriteCount , waitingForResultCount

type ring struct {
		...
    waitingForWriteCount   uint32 // Counter for nodes waiting for writing
    waitingForResultCount  uint32 // Counter for nodes waiting for results
}

waitingForWriteCount increase by one when before returning PutOne , PutMulti . waitingForWriteCount decrease by one when n.mark equals 1 in NextWriteCmd , before returning WaitForWrite .

waitingForResultCount increase by one when n.mark equals 1 in NextWriteCmd , before returning WaitForWrite .

waitingForResultCount decrease by one when n.mark equals 2 in NextResultCh .

above operations must be atomic.

Also, we should support Stats method.

func (r *ring) Stats() commandQueueStats {
	return commandQueueStats{
		waitingForWriteCount: atomic.LoadUint32(&r.waitingForWriteCount),
		waitingForResultCount: atomic.LoadUint32(&r.waitingForResultCount),
	}
}

pipe

func (p *pipe) CommandQueueStats() commandQueueStats {
	return p.queue.Stats()
}

mux

func (m *mux) AggregatedCommandQueueStats() aggregatedCommandQueueStats {
	n := len(m.wire)
	
	s := aggregatedCommandQueueStats{
		totalConns: uint64(n)	
	}
	for i := 0; i < n; i++ {
		w := m2.wire[i].Load().(wire)
		
		queueStats := w.CommandQueueStats()
		s.totalWaitingForWriteCommands += uint64(queueStats.waitingForWriteCount)
		s.totalWaitingForResultCommands += uint64(queueStats.waitingForResultCount)
	}
	return s
}

singleClient

func (c *singleClient) Stats() Stats {
	queueStats := c.conn.AggregatedCommandQueueStats()
	return Stats{
		TotalConns: queueStats.totalConns
		TotalWaitingForWriteCommands: queueStats.totalWaitingForWriteCommands
		TotalWaitingForResultCommands: queueSTats.totalWaitingForResultCommands
	}
}

sentinelClient

func (c *sentinelClient) Stats() Stats {
	queueStats := c.mconn.Load().(conn).AggregatedCommandQueueStats()
	return Stats{
		TotalConns: queueStats.totalConns
		TotalWaitingForWriteCommands: queueStats.totalWaitingForWriteCommands
		TotalWaitingForResultCommands: queueSTats.totalWaitingForResultCommands
	}
}

clusterClient

func (c *clusterClient) Stats() Stats {
	stats := Stats{}
	for _, role := range c.conns {
		s := role.conn.AggregatedCommandQueueStats()
		
		stats.TotalConns += s.totalConns
		stats.TotalWaitingForWriteCommands += s.totalWaitingForWriteCommands
		stats.TotalWaitingForResultCommands += s.totalWaitingForResultCommands
	}
	return stats
}

How to use stats?

I expect that Stats() is called periodically in loop which is made by user.

[rueian]

Hi @proost,

Thank you for the proposal. The two waiting counters and their implementation look good to me.

However, I am afraid that these two metrics can't help us improve latencies. Latencies depend on how fast we flush the output buffer and how fast we deliver responses back. In other words, I think we should probably monitor _backgroundWrite() and _backgroundRead(), not the ring.

[rueian]

Also, I think we may need to find a way to improve latencies first, then figure out what metrics it needs.

[proost]

Latencies depend on how fast we flush the output buffer and how fast we deliver responses back.

Yes. you are right. In that point, ReadBufferEachConn and WriteBufferEachConn helps a lot because those options can reduce system calls.

rueidis works well, does many things internally and supports many options. Fine tuning with options is not easy because working rueidis is quite black box to user. In a high level view, rueidisotel is helpful but it is not enough. If stats of ring buffer is not proper, could you propose any ideas to raise up observability?

[rueian]

I guess you have also tried to tune these parameters. My finding was that these parameters actually have little impact on latencies.

What really adds to latencies could be the cost of goroutine scheduling on these lines:

rueidis/pipe.go

Line 419 in 80122c0

ones[0], multi, ch = p.queue.WaitForWrite()

rueidis/pipe.go

Lines 586 to 588 in 80122c0

	ch <- resp
	cond.L.Unlock()
	cond.Signal()

I also don't have a good idea yet. I wonder if this https://go.dev/blog/execution-traces-2024 could help.

[proost]

@rueian I have to dig out that points. Thanks a lot!

[proost]

Close due to lack of usage ring buffer stats