From 7be7a47fd3b6612465d4c1ee713bebb76d33e625 Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Thu, 19 Sep 2024 09:38:38 +0530 Subject: [PATCH] MM-59933: Increment/Decrement the member count directly (#28196) While using Redis, there is no need to invalidate and then update the new count from the DB when we know the exact number it is going to be incremented or decremented by. In that case, we can directly use Redis primitives to update the cache and prevent yet another DB query. To achieve this, we modify the LRU cache get/set paths slightly to not marshal into byte slices for *int64 values. This is needed for Redis to operate the INCR/DECR commands. https://mattermost.atlassian.net/browse/MM-59933 ```release-note NONE ``` Co-authored-by: Mattermost Build --- .../store/localcachelayer/channel_layer.go | 34 ++- .../channels/store/localcachelayer/layer.go | 14 + server/platform/services/cache/cache.go | 14 + .../services/cache/mocks/ExternalCache.go | 249 ++++++++++++++++++ server/platform/services/cache/redis.go | 119 +++++++-- 5 files changed, 402 insertions(+), 28 deletions(-) create mode 100644 server/platform/services/cache/mocks/ExternalCache.go diff --git a/server/channels/store/localcachelayer/channel_layer.go b/server/channels/store/localcachelayer/channel_layer.go index 2504c91a93..8fb2106caf 100644 --- a/server/channels/store/localcachelayer/channel_layer.go +++ b/server/channels/store/localcachelayer/channel_layer.go @@ -441,8 +441,13 @@ func (s LocalCacheChannelStore) SaveMember(rctx request.CTX, member *model.Chann if err != nil { return nil, err } + // For redis, directly increment member count. - s.InvalidateMemberCount(member.ChannelId) + if externalCache, ok := s.rootStore.channelMemberCountsCache.(cache.ExternalCache); ok { + s.rootStore.doIncrementCache(externalCache, member.ChannelId, 1) + } else { + s.InvalidateMemberCount(member.ChannelId) + } return member, nil } @@ -452,7 +457,15 @@ func (s LocalCacheChannelStore) SaveMultipleMembers(members []*model.ChannelMemb return nil, err } for _, member := range members { - s.InvalidateMemberCount(member.ChannelId) + // For redis, directly increment member count. + // It should be possible to group the members from the slice + // by channelID and increment it once per channel. But it depends + // on whether all members are part of the same channel or not. + if externalCache, ok := s.rootStore.channelMemberCountsCache.(cache.ExternalCache); ok { + s.rootStore.doIncrementCache(externalCache, member.ChannelId, 1) + } else { + s.InvalidateMemberCount(member.ChannelId) + } } return members, nil } @@ -522,8 +535,13 @@ func (s LocalCacheChannelStore) RemoveMember(rctx request.CTX, channelId, userId if err != nil { return err } - // For redis, directly decrement member count - s.InvalidateMemberCount(channelId) + + // For redis, directly decrement member count. + if externalCache, ok := s.rootStore.channelMemberCountsCache.(cache.ExternalCache); ok { + s.rootStore.doDecrementCache(externalCache, channelId, 1) + } else { + s.InvalidateMemberCount(channelId) + } return nil } @@ -532,7 +550,11 @@ func (s LocalCacheChannelStore) RemoveMembers(rctx request.CTX, channelId string if err != nil { return err } - // For redis, directly decrement member count - s.InvalidateMemberCount(channelId) + // For redis, directly decrement member count. + if externalCache, ok := s.rootStore.channelMemberCountsCache.(cache.ExternalCache); ok { + s.rootStore.doDecrementCache(externalCache, channelId, len(userIds)) + } else { + s.InvalidateMemberCount(channelId) + } return nil } diff --git a/server/channels/store/localcachelayer/layer.go b/server/channels/store/localcachelayer/layer.go index bd41dbc029..76858bdf8c 100644 --- a/server/channels/store/localcachelayer/layer.go +++ b/server/channels/store/localcachelayer/layer.go @@ -526,6 +526,20 @@ func (s *LocalCacheStore) doMultiReadCache(cache cache.Cache, keys []string, val return errs } +func (s *LocalCacheStore) doIncrementCache(cache cache.ExternalCache, key string, val int) { + err := cache.Increment(key, val) + if err != nil { + s.logger.Warn("Error while incrementing cache entry", mlog.Err(err), mlog.String("cache_name", cache.Name())) + } +} + +func (s *LocalCacheStore) doDecrementCache(cache cache.ExternalCache, key string, val int) { + err := cache.Decrement(key, val) + if err != nil { + s.logger.Warn("Error while decrementing cache entry", mlog.Err(err), mlog.String("cache_name", cache.Name())) + } +} + func (s *LocalCacheStore) doClearCacheCluster(cache cache.Cache) { err := cache.Purge() if err != nil { diff --git a/server/platform/services/cache/cache.go b/server/platform/services/cache/cache.go index 63b94a8a8b..801cfc448e 100644 --- a/server/platform/services/cache/cache.go +++ b/server/platform/services/cache/cache.go @@ -52,3 +52,17 @@ type Cache interface { // Name returns the name of the cache Name() string } + +// ExternalCache is a super-set of the Cache interface with +// a couple of more methods that allows for more efficient cache updates. +// This can be achieved because the cache is external and an update +// is visible to all nodes. +type ExternalCache interface { + Cache + // Increment will increment the + // number stored at that key by the value. + Increment(key string, val int) error + // Decrement will decrement the + // number stored at that key by the value. + Decrement(key string, val int) error +} diff --git a/server/platform/services/cache/mocks/ExternalCache.go b/server/platform/services/cache/mocks/ExternalCache.go new file mode 100644 index 0000000000..70592f26c8 --- /dev/null +++ b/server/platform/services/cache/mocks/ExternalCache.go @@ -0,0 +1,249 @@ +// Code generated by mockery v2.42.2. DO NOT EDIT. + +// Regenerate this file using `make cache-mocks`. + +package mocks + +import ( + time "time" + + model "github.com/mattermost/mattermost/server/public/model" + mock "github.com/stretchr/testify/mock" +) + +// ExternalCache is an autogenerated mock type for the ExternalCache type +type ExternalCache struct { + mock.Mock +} + +// Decrement provides a mock function with given fields: key, val +func (_m *ExternalCache) Decrement(key string, val int) error { + ret := _m.Called(key, val) + + if len(ret) == 0 { + panic("no return value specified for Decrement") + } + + var r0 error + if rf, ok := ret.Get(0).(func(string, int) error); ok { + r0 = rf(key, val) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// Get provides a mock function with given fields: key, value +func (_m *ExternalCache) Get(key string, value interface{}) error { + ret := _m.Called(key, value) + + if len(ret) == 0 { + panic("no return value specified for Get") + } + + var r0 error + if rf, ok := ret.Get(0).(func(string, interface{}) error); ok { + r0 = rf(key, value) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// GetInvalidateClusterEvent provides a mock function with given fields: +func (_m *ExternalCache) GetInvalidateClusterEvent() model.ClusterEvent { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for GetInvalidateClusterEvent") + } + + var r0 model.ClusterEvent + if rf, ok := ret.Get(0).(func() model.ClusterEvent); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(model.ClusterEvent) + } + + return r0 +} + +// GetMulti provides a mock function with given fields: keys, values +func (_m *ExternalCache) GetMulti(keys []string, values []interface{}) []error { + ret := _m.Called(keys, values) + + if len(ret) == 0 { + panic("no return value specified for GetMulti") + } + + var r0 []error + if rf, ok := ret.Get(0).(func([]string, []interface{}) []error); ok { + r0 = rf(keys, values) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]error) + } + } + + return r0 +} + +// Increment provides a mock function with given fields: key, val +func (_m *ExternalCache) Increment(key string, val int) error { + ret := _m.Called(key, val) + + if len(ret) == 0 { + panic("no return value specified for Increment") + } + + var r0 error + if rf, ok := ret.Get(0).(func(string, int) error); ok { + r0 = rf(key, val) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// Name provides a mock function with given fields: +func (_m *ExternalCache) Name() string { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for Name") + } + + var r0 string + if rf, ok := ret.Get(0).(func() string); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(string) + } + + return r0 +} + +// Purge provides a mock function with given fields: +func (_m *ExternalCache) Purge() error { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for Purge") + } + + var r0 error + if rf, ok := ret.Get(0).(func() error); ok { + r0 = rf() + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// Remove provides a mock function with given fields: key +func (_m *ExternalCache) Remove(key string) error { + ret := _m.Called(key) + + if len(ret) == 0 { + panic("no return value specified for Remove") + } + + var r0 error + if rf, ok := ret.Get(0).(func(string) error); ok { + r0 = rf(key) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// RemoveMulti provides a mock function with given fields: keys +func (_m *ExternalCache) RemoveMulti(keys []string) error { + ret := _m.Called(keys) + + if len(ret) == 0 { + panic("no return value specified for RemoveMulti") + } + + var r0 error + if rf, ok := ret.Get(0).(func([]string) error); ok { + r0 = rf(keys) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// Scan provides a mock function with given fields: f +func (_m *ExternalCache) Scan(f func([]string) error) error { + ret := _m.Called(f) + + if len(ret) == 0 { + panic("no return value specified for Scan") + } + + var r0 error + if rf, ok := ret.Get(0).(func(func([]string) error) error); ok { + r0 = rf(f) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// SetWithDefaultExpiry provides a mock function with given fields: key, value +func (_m *ExternalCache) SetWithDefaultExpiry(key string, value interface{}) error { + ret := _m.Called(key, value) + + if len(ret) == 0 { + panic("no return value specified for SetWithDefaultExpiry") + } + + var r0 error + if rf, ok := ret.Get(0).(func(string, interface{}) error); ok { + r0 = rf(key, value) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// SetWithExpiry provides a mock function with given fields: key, value, ttl +func (_m *ExternalCache) SetWithExpiry(key string, value interface{}, ttl time.Duration) error { + ret := _m.Called(key, value, ttl) + + if len(ret) == 0 { + panic("no return value specified for SetWithExpiry") + } + + var r0 error + if rf, ok := ret.Get(0).(func(string, interface{}, time.Duration) error); ok { + r0 = rf(key, value, ttl) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// NewExternalCache creates a new instance of ExternalCache. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +// The first argument is typically a *testing.T value. +func NewExternalCache(t interface { + mock.TestingT + Cleanup(func()) +}) *ExternalCache { + mock := &ExternalCache{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +} diff --git a/server/platform/services/cache/redis.go b/server/platform/services/cache/redis.go index 9bf5bbf68a..b8d32acbc7 100644 --- a/server/platform/services/cache/redis.go +++ b/server/platform/services/cache/redis.go @@ -7,6 +7,7 @@ import ( "context" "errors" "fmt" + "strconv" "strings" "time" @@ -58,28 +59,71 @@ func (r *Redis) SetWithExpiry(key string, value any, ttl time.Duration) error { r.metrics.ObserveRedisEndpointDuration(r.name, "Set", elapsed) } }() - var buf []byte - var err error - // We use a fast path for hot structs. - if msgpVal, ok := value.(msgp.Marshaler); ok { - buf, err = msgpVal.MarshalMsg(nil) + + var valueString string + if intVal, ok := value.(int64); ok { + valueString = strconv.Itoa(int(intVal)) } else { - // Slow path for other structs. - buf, err = msgpack.Marshal(value) - } - if err != nil { - return err + var buf []byte + var err error + // We use a fast path for hot structs. + if msgpVal, ok := value.(msgp.Marshaler); ok { + buf, err = msgpVal.MarshalMsg(nil) + } else { + // Slow path for other structs. + buf, err = msgpack.Marshal(value) + } + if err != nil { + return err + } + valueString = rueidis.BinaryString(buf) } return r.client.Do(context.Background(), r.client.B().Set(). Key(r.name+":"+key). - Value(rueidis.BinaryString(buf)). + Value(valueString). Ex(ttl). Build(), ).Error() } +// Increment increments the value of the key by the value. +func (r *Redis) Increment(key string, val int) error { + now := time.Now() + defer func() { + if r.metrics != nil { + elapsed := time.Since(now).Seconds() + r.metrics.ObserveRedisEndpointDuration(r.name, "Incr", elapsed) + } + }() + + return r.client.Do(context.Background(), + r.client.B().Incrby(). + Key(r.name+":"+key). + Increment(int64(val)). + Build(), + ).Error() +} + +// Decrement decrements the value of the key by the value. +func (r *Redis) Decrement(key string, val int) error { + now := time.Now() + defer func() { + if r.metrics != nil { + elapsed := time.Since(now).Seconds() + r.metrics.ObserveRedisEndpointDuration(r.name, "Decr", elapsed) + } + }() + + return r.client.Do(context.Background(), + r.client.B().Decrby(). + Key(r.name+":"+key). + Decrement(int64(val)). + Build(), + ).Error() +} + // Get the content stored in the cache for the given key, and decode it into the value interface. // Return ErrKeyNotFound if the key is missing from the cache func (r *Redis) Get(key string, value any) error { @@ -90,12 +134,23 @@ func (r *Redis) Get(key string, value any) error { r.metrics.ObserveRedisEndpointDuration(r.name, "Get", elapsed) } }() - val, err := r.client.DoCache(context.Background(), + + resp := r.client.DoCache(context.Background(), r.client.B().Get(). Key(r.name+":"+key). Cache(), clientSideTTL, - ).AsBytes() + ) + + var intVal int64 + var bytesVal []byte + var err error + vPtr, ok := value.(*int64) + if ok { + intVal, err = resp.AsInt64() + } else { + bytesVal, err = resp.AsBytes() + } if err != nil { if rueidis.IsRedisNil(err) { return ErrKeyNotFound @@ -103,9 +158,14 @@ func (r *Redis) Get(key string, value any) error { return err } + if ok { + *vPtr = intVal + return nil + } + // We use a fast path for hot structs. if msgpVal, ok := value.(msgp.Unmarshaler); ok { - _, err := msgpVal.UnmarshalMsg(val) + _, err := msgpVal.UnmarshalMsg(bytesVal) return err } @@ -120,15 +180,16 @@ func (r *Redis) Get(key string, value any) error { switch v := value.(type) { case **model.User: var u model.User - _, err := u.UnmarshalMsg(val) + _, err := u.UnmarshalMsg(bytesVal) *v = &u return err } // Slow path for other structs. - return msgpack.Unmarshal(val, value) + return msgpack.Unmarshal(bytesVal, value) } +// GetMulti uses the MGET primitive to fetch multiple keys in a single operation. func (r *Redis) GetMulti(keys []string, values []any) []error { now := time.Now() defer func() { @@ -162,21 +223,35 @@ func (r *Redis) GetMulti(keys []string, values []any) []error { return errs } - for i, val := range vals { - if val.IsNil() { + for i, resp := range vals { + if resp.IsNil() { errs[i] = ErrKeyNotFound continue } - buf, err := val.AsBytes() + var intVal int64 + var bytesVal []byte + var err error + vPtr, ok := values[i].(*int64) + if ok { + intVal, err = resp.AsInt64() + } else { + bytesVal, err = resp.AsBytes() + } if err != nil { errs[i] = err continue } + if ok { + *vPtr = intVal + errs[i] = nil + continue + } + // We use a fast path for hot structs. if msgpVal, ok := values[i].(msgp.Unmarshaler); ok { - _, err := msgpVal.UnmarshalMsg(buf) + _, err := msgpVal.UnmarshalMsg(bytesVal) errs[i] = err continue } @@ -184,14 +259,14 @@ func (r *Redis) GetMulti(keys []string, values []any) []error { switch v := values[i].(type) { case **model.User: var u model.User - _, err := u.UnmarshalMsg(buf) + _, err := u.UnmarshalMsg(bytesVal) *v = &u errs[i] = err continue } // Slow path for other structs. - errs[i] = msgpack.Unmarshal(buf, values[i]) + errs[i] = msgpack.Unmarshal(bytesVal, values[i]) } return errs