--- id: 1c72ae6efee059c2c727ea53b56bf6c5 author: snowmerak title: Improving Responsiveness with Redis Client-Side Caching description: Improving API Responsiveness Using Redis Server-Assisted Client-Side Caching language: en date: 2024-11-29T00:51:27.525566863Z path: /blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502 category: database --- ## 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](/assets/images/redis-client-side-cache/01-pubsub-write.png) ![pubsub-read](/assets/images/redis-client-side-cache/01-pubsub-read.png) ### 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](/assets/images/redis-client-side-cache/02-invalidation-message.png) ### How do you use it? #### Basic operating structure A client connected to Redis executes `CLIENT TRACKING ON REDIRECT ` 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 ```shell # client 1 > SET a 100 ``` ```shell # client 3 > CLIENT ID 12 > SUBSCRIBE __redis__:invalidate 1) "subscribe" 2) "__redis__:invalidate" 3) (integer) 1 ``` ```shell # client 2 > CLIENT TRACKING ON REDIRECT 12 > GET a # tracking ``` ```shell # client 1 > SET a 200 ``` ```shell # client 3 1) "message" 2) "__redis__:invalidate" 3) 1) "a" ``` #### Broadcasting tracking ```shell # client 3 > CLIENT ID 12 > SUBSCRIBE __redis__:invalidate 1) "subscribe" 2) "__redis__:invalidate" 3) (integer) 1 ``` ```shell # client 2 CLIENT TRACKING ON BCAST PREFIX cache: REDIRECT 12 ``` ```shell # client 1 > SET cache:name "Alice" > SET cache:age 26 ``` ```shell # client 3 1) "message" 2) "__redis__:invalidate" 3) 1) "cache:name" 1) "message" 2) "__redis__:invalidate" 3) 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` ```go package main import ( "context" "fmt" "log/slog" "time" "github.com/dgraph-io/ristretto" "github.com/gomodule/redigo/redis" ) type RedisClient struct { conn redis.Conn cache *ristretto.Cache[string, any] addr string } func NewRedisClient(addr string) (*RedisClient, error) { cache, err := ristretto.NewCache(&ristretto.Config[string, any]{ NumCounters: 1e7, // number of keys to track frequency of (10M). MaxCost: 1 << 30, // maximum cost of cache (1GB). BufferItems: 64, // number of keys per Get buffer. }) if err != nil { return nil, fmt.Errorf("failed to generate cache: %w", err) } conn, err := redis.Dial("tcp", addr) if err != nil { return nil, fmt.Errorf("failed to connect to redis: %w", err) } return &RedisClient{ conn: conn, cache: cache, addr: addr, }, nil } func (r *RedisClient) Close() error { err := r.conn.Close() if err != nil { return fmt.Errorf("failed to close redis connection: %w", err) } return nil } ``` First, we create a simple `RedisClient` that contains ristretto and redigo. ```go func (r *RedisClient) Tracking(ctx context.Context) error { psc, err := redis.Dial("tcp", r.addr) if err != nil { return fmt.Errorf("failed to connect to redis: %w", err) } clientId, err := redis.Int64(psc.Do("CLIENT", "ID")) if err != nil { return fmt.Errorf("failed to get client id: %w", err) } slog.Info("client id", "id", clientId) subscriptionResult, err := redis.String(r.conn.Do("CLIENT", "TRACKING", "ON", "REDIRECT", clientId)) if err != nil { return fmt.Errorf("failed to enable tracking: %w", err) } slog.Info("subscription result", "result", subscriptionResult) if err := psc.Send("SUBSCRIBE", "__redis__:invalidate"); err != nil { return fmt.Errorf("failed to subscribe: %w", err) } psc.Flush() for { msg, err := psc.Receive() if err != nil { return fmt.Errorf("failed to receive message: %w", err) } switch msg := msg.(type) { case redis.Message: slog.Info("received message", "channel", msg.Channel, "data", msg.Data) key := string(msg.Data) r.cache.Del(key) case redis.Subscription: slog.Info("subscription", "kind", msg.Kind, "channel", msg.Channel, "count", msg.Count) case error: return fmt.Errorf("error: %w", msg) case []interface{}: if len(msg) != 3 || string(msg[0].([]byte)) != "message" || string(msg[1].([]byte)) != "__redis__:invalidate" { slog.Warn("unexpected message", "message", msg) continue } contents := msg[2].([]interface{}) keys := make([]string, len(contents)) for i, key := range contents { keys[i] = string(key.([]byte)) r.cache.Del(keys[i]) } slog.Info("received invalidation message", "keys", keys) default: slog.Warn("unexpected message", "type", fmt.Sprintf("%T", msg)) } } } ``` 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. ```go func (r *RedisClient) Get(key string) (any, error) { val, found := r.cache.Get(key) if found { switch v := val.(type) { case int64: slog.Info("cache hit", "key", key) return v, nil default: slog.Warn("unexpected type", "type", fmt.Sprintf("%T", v)) } } slog.Info("cache miss", "key", key) val, err := redis.Int64(r.conn.Do("GET", key)) if err != nil { return nil, fmt.Errorf("failed to get key: %w", err) } r.cache.SetWithTTL(key, val, 1, 10*time.Second) return val, nil } ``` The `Get` method first queries ristretto as shown below, and if the value is not found, it fetches it from Redis. ```go package main import ( "context" "log/slog" "os" "os/signal" "time" ) func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) defer cancel() client, err := NewRedisClient("localhost:6379") if err != nil { panic(err) } defer client.Close() go func() { if err := client.Tracking(ctx); err != nil { slog.Error("failed to track invalidation message", "error", err) } }() ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() done := ctx.Done() for { select { case <-done: slog.Info("shutting down") return case <-ticker.C: v, err := client.Get("key") if err != nil { slog.Error("failed to get key", "error", err) return } slog.Info("got key", "value", v) } } } ``` 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. ```go package main import ( "context" "log/slog" "os" "os/signal" "time" "github.com/redis/rueidis" ) func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) defer cancel() client, err := rueidis.NewClient(rueidis.ClientOption{ InitAddress: []string{"localhost:6379"}, }) if err != nil { panic(err) } ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() done := ctx.Done() for { select { case <-done: slog.Info("shutting down") return case <-ticker.C: const key = "key" resp := client.DoCache(ctx, client.B().Get().Key(key).Cache(), 10*time.Second) if resp.Error() != nil { slog.Error("failed to get key", "error", resp.Error()) continue } i, err := resp.AsInt64() if err != nil { slog.Error("failed to convert response to int64", "error", err) continue } switch resp.IsCacheHit() { case true: slog.Info("cache hit", "key", key) case false: slog.Info("missed key", "key", key) } slog.Info("got key", "value", i) } } } ``` 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](/assets/images/redis-client-side-cache/003-cqrs.jpg) ### 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.