Move LL cache
MODIFY
internal/caching/cache_lazy_load_members.go
internal/caching/cache_lazy_load_members.go
@@ -15,33 +15,14 @@ const (
LazyLoadCacheMaxAge = time.Minute * 30
)
-type LazyLoadCache struct {
- // InMemoryLRUCachePartition containing other InMemoryLRUCachePartitions
- // with the actual cached members
- userCaches *InMemoryLRUCachePartition
+type LazyLoadCache interface {
+ StoreLazyLoadedUser(device *userapi.Device, roomID, userID, eventID string)
+ IsLazyLoadedUserCached(device *userapi.Device, roomID, userID string) (string, bool)
}
-// NewLazyLoadCache creates a new LazyLoadCache.
-func NewLazyLoadCache() (*LazyLoadCache, error) {
- cache, err := NewInMemoryLRUCachePartition(
- LazyLoadCacheName,
- LazyLoadCacheMutable,
- LazyLoadCacheMaxEntries,
- LazyLoadCacheMaxAge,
- true,
- )
- if err != nil {
- return nil, err
- }
- go cacheCleaner(cache)
- return &LazyLoadCache{
- userCaches: cache,
- }, nil
-}
-
-func (c *LazyLoadCache) lazyLoadCacheForUser(device *userapi.Device) (*InMemoryLRUCachePartition, error) {
+func (c Caches) lazyLoadCacheForUser(device *userapi.Device) (*InMemoryLRUCachePartition, error) {
cacheName := fmt.Sprintf("%s/%s", device.UserID, device.ID)
- userCache, ok := c.userCaches.Get(cacheName)
+ userCache, ok := c.LazyLoading.Get(cacheName)
if ok && userCache != nil {
if cache, ok := userCache.(*InMemoryLRUCachePartition); ok {
return cache, nil
@@ -57,12 +38,12 @@ func (c *LazyLoadCache) lazyLoadCacheForUser(device *userapi.Device) (*InMemoryL
if err != nil {
return nil, err
}
- c.userCaches.Set(cacheName, cache)
+ c.LazyLoading.Set(cacheName, cache)
go cacheCleaner(cache)
return cache, nil
}
-func (c *LazyLoadCache) StoreLazyLoadedUser(device *userapi.Device, roomID, userID, eventID string) {
+func (c Caches) StoreLazyLoadedUser(device *userapi.Device, roomID, userID, eventID string) {
cache, err := c.lazyLoadCacheForUser(device)
if err != nil {
return
@@ -71,7 +52,7 @@ func (c *LazyLoadCache) StoreLazyLoadedUser(device *userapi.Device, roomID, user
cache.Set(cacheKey, eventID)
}
-func (c *LazyLoadCache) IsLazyLoadedUserCached(device *userapi.Device, roomID, userID string) (string, bool) {
+func (c Caches) IsLazyLoadedUserCached(device *userapi.Device, roomID, userID string) (string, bool) {
cache, err := c.lazyLoadCacheForUser(device)
if err != nil {
return "", false

MODIFY
internal/caching/caches.go
internal/caching/caches.go
package caching
-import "time"
+import (
+ "time"
+)
// Caches contains a set of references to caches. They may be
// different implementations as long as they satisfy the Cache
@@ -13,6 +15,7 @@ type Caches struct {
RoomInfos Cache // RoomInfoCache
FederationEvents Cache // FederationEventsCache
SpaceSummaryRooms Cache // SpaceSummaryRoomsCache
+ LazyLoading Cache // LazyLoadCache
}
// Cache is the interface that an implementation must satisfy.

MODIFY
internal/caching/impl_inmemorylru.go
internal/caching/impl_inmemorylru.go
@@ -70,9 +70,21 @@ func NewInMemoryLRUCache(enablePrometheus bool) (*Caches, error) {
if err != nil {
return nil, err
}
+
+ lazyLoadCache, err := NewInMemoryLRUCachePartition(
+ LazyLoadCacheName,
+ LazyLoadCacheMutable,
+ LazyLoadCacheMaxEntries,
+ LazyLoadCacheMaxAge,
+ enablePrometheus,
+ )
+ if err != nil {
+ return nil, err
+ }
+
go cacheCleaner(
roomVersions, serverKeys, roomServerRoomIDs,
- roomInfos, federationEvents, spaceRooms,
+ roomInfos, federationEvents, spaceRooms, lazyLoadCache,
)
return &Caches{
RoomVersions: roomVersions,
@@ -81,6 +93,7 @@ func NewInMemoryLRUCache(enablePrometheus bool) (*Caches, error) {
RoomInfos: roomInfos,
FederationEvents: federationEvents,
SpaceSummaryRooms: spaceRooms,
+ LazyLoading: lazyLoadCache,
}, nil
}

MODIFY
syncapi/routing/context.go
syncapi/routing/context.go
@@ -45,7 +45,7 @@ func Context(
rsAPI roomserver.SyncRoomserverAPI,
syncDB storage.Database,
roomID, eventID string,
- lazyLoadCache *caching.LazyLoadCache,
+ lazyLoadCache caching.LazyLoadCache,
) util.JSONResponse {
filter, err := parseRoomEventFilter(req)
if err != nil {
@@ -155,7 +155,7 @@ func applyLazyLoadMembers(
filter *gomatrixserverlib.RoomEventFilter,
eventsAfter, eventsBefore []gomatrixserverlib.ClientEvent,
state []*gomatrixserverlib.HeaderedEvent,
- lazyLoadCache *caching.LazyLoadCache,
+ lazyLoadCache caching.LazyLoadCache,
) []*gomatrixserverlib.HeaderedEvent {
if filter == nil || !filter.LazyLoadMembers {
return state

MODIFY
syncapi/routing/messages.go
syncapi/routing/messages.go
@@ -63,7 +63,7 @@ func OnIncomingMessagesRequest(
rsAPI api.SyncRoomserverAPI,
cfg *config.SyncAPI,
srp *sync.RequestPool,
- lazyLoadCache *caching.LazyLoadCache,
+ lazyLoadCache caching.LazyLoadCache,
) util.JSONResponse {
var err error

MODIFY
syncapi/routing/routing.go
syncapi/routing/routing.go
@@ -39,7 +39,7 @@ func Setup(
userAPI userapi.SyncUserAPI,
rsAPI api.SyncRoomserverAPI,
cfg *config.SyncAPI,
- lazyLoadCache *caching.LazyLoadCache,
+ lazyLoadCache caching.LazyLoadCache,
) {
v3mux := csMux.PathPrefix("/{apiversion:(?:r0|v3)}/").Subrouter()

MODIFY
syncapi/streams/stream_pdu.go
syncapi/streams/stream_pdu.go
@@ -32,7 +32,7 @@ type PDUStreamProvider struct {
tasks chan func()
workers atomic.Int32
// userID+deviceID -> lazy loading cache
- lazyLoadCache *caching.LazyLoadCache
+ lazyLoadCache caching.LazyLoadCache
rsAPI roomserverAPI.SyncRoomserverAPI
}

MODIFY
syncapi/streams/streams.go
syncapi/streams/streams.go
@@ -27,7 +27,7 @@ type Streams struct {
func NewSyncStreamProviders(
d storage.Database, userAPI userapi.SyncUserAPI,
rsAPI rsapi.SyncRoomserverAPI, keyAPI keyapi.SyncKeyAPI,
- eduCache *caching.EDUCache, lazyLoadCache *caching.LazyLoadCache, notifier *notifier.Notifier,
+ eduCache *caching.EDUCache, lazyLoadCache caching.LazyLoadCache, notifier *notifier.Notifier,
) *Streams {
streams := &Streams{
PDUStreamProvider: &PDUStreamProvider{

MODIFY
syncapi/syncapi.go
syncapi/syncapi.go
@@ -53,12 +53,8 @@ func AddPublicRoutes(
}
eduCache := caching.NewTypingCache()
- lazyLoadCache, err := caching.NewLazyLoadCache()
- if err != nil {
- logrus.WithError(err).Panicf("failed to create lazy loading cache")
- }
notifier := notifier.NewNotifier()
- streams := streams.NewSyncStreamProviders(syncDB, userAPI, rsAPI, keyAPI, eduCache, lazyLoadCache, notifier)
+ streams := streams.NewSyncStreamProviders(syncDB, userAPI, rsAPI, keyAPI, eduCache, base.Caches, notifier)
notifier.SetCurrentPosition(streams.Latest(context.Background()))
if err = notifier.Load(context.Background(), syncDB); err != nil {
logrus.WithError(err).Panicf("failed to load notifier ")
@@ -146,6 +142,6 @@ func AddPublicRoutes(
routing.Setup(
base.PublicClientAPIMux, requestPool, syncDB, userAPI,
- rsAPI, cfg, lazyLoadCache,
+ rsAPI, cfg, base.Caches,
)
}

Move LL cache (#2429)