GoSuda

Improving Responsiveness with Redis Client-Side Caching

By snowmerak views ...

What is Redis?

I suspect that few people are unfamiliar with Redis. Still, if I briefly mention a few of its characteristics before moving on, they can be summarized as follows.

  • Operations are performed on a single thread, so all operations are atomic.
  • Data is stored and computed In-Memory, so all operations are fast.
  • Redis can store the WAL depending on the option, so it can quickly back up and restore the latest state.
  • It supports various types such as Set, Hash, Bit, and List, resulting in high productivity.
  • It has a large community, so you can share in various experiences, issues, and solutions.
  • It has been developed and operated for a long time, so it has reliable stability.

So, to the main topic

Imagine?

What if your service's cache met the following two conditions?

  1. Frequently queried data must be provided to users in its latest state, but since updates are irregular, the cache must be refreshed frequently.
  2. The data is not updated, but the same cached data is accessed and queried frequently.

The first case could be considered for a shopping mall's real-time popularity ranking. If the real-time popularity ranking of a shopping mall is stored as a sorted set, it is inefficient to read it from Redis every time a user accesses the main page.
The second case involves exchange rate data. Even though exchange rate data is published roughly every 10 minutes, actual queries occur very frequently. In particular, KRW-USD, KRW-JPY, and KRW-CNY caches are queried very frequently.
In such cases, it would be more efficient for the API server to hold a separate local cache and re-query Redis to refresh it when the data changes.

Then, how can this behavior be implemented in a Database - Redis - API server structure??

Can't we just use Redis PubSub?

When using a cache, let's subscribe to a channel that can receive notifications of updates!

  • Then, we need to create logic that sends a message at the time of an update.
  • The additional operations introduced by PubSub affect performance.

pubsub-write

pubsub-read

What if Redis detects the changes?

What if we use Keyspace Notification to receive command notifications for the relevant key?

  • There is the inconvenience of having to store and share in advance the keys and commands used for updates.
  • For example, for some keys a simple Set is the update command, while for others LPush, RPush, SAdd, or SRem are update commands, making things complicated.
  • This greatly increases the likelihood of communication mistakes during development and human errors in coding.

What if we use Keyevent Notification to receive notifications per command?

  • Subscription to all commands used for updates is required. Appropriate filtering of the incoming keys is also necessary.
  • For example, among all keys coming in via Del, the client is likely to have no local cache for some of them.
  • This may lead to unnecessary waste of resources.

That's why we need Invalidation Messages!

What are Invalidation Messages?

Invalidation Messages are a concept provided as part of Server Assisted Client-Side Cache, which was added in Redis 6.0. Invalidation Messages are delivered through the following flow.

  1. Assume that ClientB has already read a key once.
  2. ClientA sets the key to a new value.
  3. Redis detects the change and sends an Invalidation Message to ClientB, telling ClientB to clear its cache.
  4. ClientB receives the message and takes appropriate action.

invalidation-message

How do you use it?

Basic operating structure

A client connected to Redis executes CLIENT TRACKING ON REDIRECT <client-id> so that it receives invalidation messages. The client that needs to receive the messages then subscribes to SUBSCRIBE __redis__:invalidate to receive invalidation messages.

Default tracking

1# client 1
2> SET a 100

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2> CLIENT TRACKING ON REDIRECT 12
3> GET a # tracking

1# client 1
2> SET a 200

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "a"

Broadcasting tracking

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2CLIENT TRACKING ON BCAST PREFIX cache: REDIRECT 12

1# client 1
2> SET cache:name "Alice"
3> SET cache:age 26

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "cache:name"
51) "message"
62) "__redis__:invalidate"
73) 1) "cache:age"

Implementation! Implementation! Implementation!

Redigo + Ristretto

If I only explain it this way, it is unclear how to actually use it in code. So I will first set things up simply with redigo and ristretto.

First, install the two dependencies.

  • github.com/gomodule/redigo
  • github.com/dgraph-io/ristretto
 1package main
 2
 3import (
 4	"context"
 5	"fmt"
 6	"log/slog"
 7	"time"
 8
 9	"github.com/dgraph-io/ristretto"
10	"github.com/gomodule/redigo/redis"
11)
12
13type RedisClient struct {
14	conn  redis.Conn
15	cache *ristretto.Cache[string, any]
16	addr  string
17}
18
19func NewRedisClient(addr string) (*RedisClient, error) {
20	cache, err := ristretto.NewCache(&ristretto.Config[string, any]{
21		NumCounters: 1e7,     // number of keys to track frequency of (10M).
22		MaxCost:     1 << 30, // maximum cost of cache (1GB).
23		BufferItems: 64,      // number of keys per Get buffer.
24	})
25	if err != nil {
26		return nil, fmt.Errorf("failed to generate cache: %w", err)
27	}
28
29	conn, err := redis.Dial("tcp", addr)
30	if err != nil {
31		return nil, fmt.Errorf("failed to connect to redis: %w", err)
32	}
33
34	return &RedisClient{
35		conn:  conn,
36		cache: cache,
37		addr:  addr,
38	}, nil
39}
40
41func (r *RedisClient) Close() error {
42	err := r.conn.Close()
43	if err != nil {
44		return fmt.Errorf("failed to close redis connection: %w", err)
45	}
46
47	return nil
48}

First, we create a simple RedisClient that contains ristretto and redigo.

 1func (r *RedisClient) Tracking(ctx context.Context) error {
 2	psc, err := redis.Dial("tcp", r.addr)
 3	if err != nil {
 4		return fmt.Errorf("failed to connect to redis: %w", err)
 5	}
 6
 7	clientId, err := redis.Int64(psc.Do("CLIENT", "ID"))
 8	if err != nil {
 9		return fmt.Errorf("failed to get client id: %w", err)
10	}
11	slog.Info("client id", "id", clientId)
12
13	subscriptionResult, err := redis.String(r.conn.Do("CLIENT", "TRACKING", "ON", "REDIRECT", clientId))
14	if err != nil {
15		return fmt.Errorf("failed to enable tracking: %w", err)
16	}
17	slog.Info("subscription result", "result", subscriptionResult)
18
19	if err := psc.Send("SUBSCRIBE", "__redis__:invalidate"); err != nil {
20		return fmt.Errorf("failed to subscribe: %w", err)
21	}
22	psc.Flush()
23
24	for {
25		msg, err := psc.Receive()
26		if err != nil {
27			return fmt.Errorf("failed to receive message: %w", err)
28		}
29
30		switch msg := msg.(type) {
31		case redis.Message:
32			slog.Info("received message", "channel", msg.Channel, "data", msg.Data)
33			key := string(msg.Data)
34			r.cache.Del(key)
35		case redis.Subscription:
36			slog.Info("subscription", "kind", msg.Kind, "channel", msg.Channel, "count", msg.Count)
37		case error:
38			return fmt.Errorf("error: %w", msg)
39		case []interface{}:
40			if len(msg) != 3 || string(msg[0].([]byte)) != "message" || string(msg[1].([]byte)) != "__redis__:invalidate" {
41				slog.Warn("unexpected message", "message", msg)
42				continue
43			}
44
45			contents := msg[2].([]interface{})
46			keys := make([]string, len(contents))
47			for i, key := range contents {
48				keys[i] = string(key.([]byte))
49				r.cache.Del(keys[i])
50			}
51			slog.Info("received invalidation message", "keys", keys)
52		default:
53			slog.Warn("unexpected message", "type", fmt.Sprintf("%T", msg))
54		}
55	}
56}

The code is somewhat complicated.

  • To perform Tracking, we open one more connection. This is a measure taken in consideration of PubSub interfering with other operations.
  • We retrieve the ID of the added connection and redirect the tracking of the connection that queries data to that connection.
  • Then we subscribe to invalidation messages.
  • The code that handles subscriptions is somewhat complicated. Since redigo does not parse invalidation messages, we need to receive the response and handle it before parsing.
 1func (r *RedisClient) Get(key string) (any, error) {
 2	val, found := r.cache.Get(key)
 3	if found {
 4		switch v := val.(type) {
 5		case int64:
 6			slog.Info("cache hit", "key", key)
 7			return v, nil
 8		default:
 9			slog.Warn("unexpected type", "type", fmt.Sprintf("%T", v))
10		}
11	}
12	slog.Info("cache miss", "key", key)
13
14	val, err := redis.Int64(r.conn.Do("GET", key))
15	if err != nil {
16		return nil, fmt.Errorf("failed to get key: %w", err)
17	}
18
19	r.cache.SetWithTTL(key, val, 1, 10*time.Second)
20	return val, nil
21}

The Get method first queries ristretto as shown below, and if the value is not found, it fetches it from Redis.

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9)
10
11func main() {
12	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
13	defer cancel()
14
15	client, err := NewRedisClient("localhost:6379")
16	if err != nil {
17		panic(err)
18	}
19	defer client.Close()
20
21	go func() {
22		if err := client.Tracking(ctx); err != nil {
23			slog.Error("failed to track invalidation message", "error", err)
24		}
25	}()
26
27	ticker := time.NewTicker(1 * time.Second)
28	defer ticker.Stop()
29	done := ctx.Done()
30
31	for {
32		select {
33		case <-done:
34			slog.Info("shutting down")
35			return
36		case <-ticker.C:
37			v, err := client.Get("key")
38			if err != nil {
39				slog.Error("failed to get key", "error", err)
40				return
41			}
42			slog.Info("got key", "value", v)
43		}
44	}
45}

The code above is for testing. If you try it, you will be able to see the value being refreshed each time the data is updated in Redis.

However, this is too complicated. Above all, to scale across a cluster, it is inevitably necessary to enable Tracking on all masters or replicas.

Rueidis

Since we write Go, we have rueidis, the most modern and advanced option. Let's write code that uses server assisted client side cache in a Redis cluster environment with rueidis.

First, install the dependency.

  • github.com/redis/rueidis

Then we write code that queries data from Redis.

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9
10	"github.com/redis/rueidis"
11)
12
13func main() {
14	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
15	defer cancel()
16
17	client, err := rueidis.NewClient(rueidis.ClientOption{
18		InitAddress: []string{"localhost:6379"},
19	})
20	if err != nil {
21		panic(err)
22	}
23
24	ticker := time.NewTicker(1 * time.Second)
25	defer ticker.Stop()
26	done := ctx.Done()
27
28	for {
29		select {
30		case <-done:
31			slog.Info("shutting down")
32			return
33		case <-ticker.C:
34			const key = "key"
35			resp := client.DoCache(ctx, client.B().Get().Key(key).Cache(), 10*time.Second)
36			if resp.Error() != nil {
37				slog.Error("failed to get key", "error", resp.Error())
38				continue
39			}
40			i, err := resp.AsInt64()
41			if err != nil {
42				slog.Error("failed to convert response to int64", "error", err)
43				continue
44			}
45			switch resp.IsCacheHit() {
46			case true:
47				slog.Info("cache hit", "key", key)
48			case false:
49				slog.Info("missed key", "key", key)
50			}
51			slog.Info("got key", "value", i)
52		}
53	}
54}

To use the client side cache with rueidis, you simply call DoCache. Then, by adding the entry to the local cache, such as specifying how long to keep it locally, and calling DoCache in the same way, the data is retrieved from the local cache. Naturally, invalidation messages are also handled properly.

Why not redis-go?

Unfortunately, redis-go does not support server assisted client side cache through its official API. Moreover, when creating a PubSub, it creates a new connection without providing an API to access that connection directly, so the client id cannot be known either. Therefore, we judged that building it with redis-go is impossible and passed on it.

Sexy

Through the client side cache structure

  • If the data can be prepared in advance, this structure will minimize queries and traffic to Redis while always providing the latest data.
  • Through this, you can create a kind of CQRS structure and dramatically improve read performance.

cqrs

How much sexier did it become?

We are actually using this structure in the field, so we looked into the simple latency of two APIs. Please forgive me for being able to describe this only in very abstract terms.

  1. First API
    1. First query: average 14.63ms
    2. Subsequent queries: average 2.82ms
    3. Average gap: 10.98ms
  2. Second API
    1. First query: average 14.05ms
    2. Subsequent queries: average 1.60ms
    3. Average gap: 11.57ms

At best, there was an additional latency improvement of about 82%!

We expect that the following improvements were likely achieved.

  • Skipping the network communication process between the client and Redis and saving traffic
  • Reduction in the number of read commands Redis itself must perform
    • This also has the effect of improving write performance.
  • Minimized parsing of the Redis protocol
    • Parsing the Redis protocol is not zero cost either. Being able to reduce this is a great opportunity.

However, everything is a trade-off. For this, we sacrificed at least the following two things.

  • The need to implement, operate, and maintain the client side cache management component
  • The resulting increase in the client's CPU and memory usage

Conclusion

Personally, it was a satisfying architectural component, and the latency and stress on the API server were also very low. Going forward, I think it would be good to structure our architecture this way whenever possible.