Support caching for map primitive

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/gogo/protobuf v1.3.1
 	github.com/golang/protobuf v1.3.2
 	github.com/google/uuid v1.1.1
+	github.com/hashicorp/golang-lru v0.5.4
 	github.com/stretchr/testify v1.4.0
 	google.golang.org/grpc v1.27.0
 )

go.sum

@@ -174,6 +174,8 @@ github.com/grpc-ecosystem/go-grpc-middleware v1.1.0/go.mod h1:f5nM7jw/oeRSadq3xC
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.3/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
+github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=

pkg/client/client.go

@@ -380,8 +380,8 @@ func (d *Database) GetLog(ctx context.Context, name string) (log.Log, error) {
 }
 
 // GetMap gets or creates a Map with the given name
-func (d *Database) GetMap(ctx context.Context, name string) (_map.Map, error) {
-	return _map.New(ctx, primitive.NewName(d.Namespace, d.Name, d.scope, name), d.sessions)
+func (d *Database) GetMap(ctx context.Context, name string, opts ..._map.Option) (_map.Map, error) {
+	return _map.New(ctx, primitive.NewName(d.Namespace, d.Name, d.scope, name), d.sessions, opts...)
 }
 
 // GetSet gets or creates a Set with the given name
@@ -442,8 +442,8 @@ func (g *PartitionGroup) GetLog(ctx context.Context, name string) (log.Log, erro
 }
 
 // GetMap gets or creates a Map with the given name
-func (g *PartitionGroup) GetMap(ctx context.Context, name string) (_map.Map, error) {
-	return _map.New(ctx, primitive.NewName(g.Namespace, g.Name, g.scope, name), g.sessions)
+func (g *PartitionGroup) GetMap(ctx context.Context, name string, opts ..._map.Option) (_map.Map, error) {
+	return _map.New(ctx, primitive.NewName(g.Namespace, g.Name, g.scope, name), g.sessions, opts...)
 }
 
 // GetSet gets or creates a Set with the given name

pkg/client/map/cache.go

@@ -0,0 +1,205 @@
+// Copyright 2019-present Open Networking Foundation.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package _map //nolint:golint
+
+import (
+	"context"
+	"github.com/hashicorp/golang-lru"
+	"sync"
+)
+
+// newCachingMap returns a decorated map that caches updates to the given map
+func newCachingMap(_map Map, size int) (Map, error) {
+	cache, err := lru.New(size)
+	if err != nil {
+		return nil, err
+	}
+	cachingMap := &cachingMap{
+		delegatingMap: newDelegatingMap(_map),
+		pending:       make(map[string]*cachedEntry),
+		cache:         cache,
+	}
+	if err := cachingMap.open(); err != nil {
+		return nil, err
+	}
+	return cachingMap, nil
+}
+
+// cachingMap is an implementation of the Map interface that caches entries
+type cachingMap struct {
+	*delegatingMap
+	cancel       context.CancelFunc
+	pending      map[string]*cachedEntry
+	cache        *lru.Cache
+	cacheVersion int64
+	mu           sync.RWMutex
+}
+
+// open opens the map listeners
+func (m *cachingMap) open() error {
+	ch := make(chan *Event)
+	ctx, cancel := context.WithCancel(context.Background())
+	m.mu.Lock()
+	m.cancel = cancel
+	m.mu.Unlock()
+	if err := m.delegatingMap.Watch(ctx, ch, WithReplay()); err != nil {
+		return err
+	}
+	go func() {
+		for event := range ch {
+			m.cacheUpdate(event.Entry, event.Type == EventRemoved)
+		}
+	}()
+	return nil
+}
+
+// cacheUpdate caches the given updated entry
+func (m *cachingMap) cacheUpdate(update *Entry, tombstone bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	// If the update version is less than the cache version, the cache contains
+	// more recent updates. Ignore the update.
+	if update.Version <= m.cacheVersion {
+		return
+	}
+
+	// If the pending entry is newer than the update entry, the update can be ignored.
+	// Otherwise, remove the entry from the pending cache if present.
+	if pending, ok := m.pending[update.Key]; ok {
+		if pending.Version > update.Version {
+			return
+		}
+		delete(m.pending, update.Key)
+	}
+
+	// If the entry is a tombstone, remove it from the cache, otherwise insert it.
+	if tombstone {
+		m.cache.Remove(update.Key)
+	} else {
+		m.cache.Add(update.Key, update)
+	}
+
+	// Update the cache version.
+	m.cacheVersion = update.Version
+}
+
+// cacheRead caches the given read entry
+func (m *cachingMap) cacheRead(read *Entry, tombstone bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	// If the entry version is less than the cache version, ignore the update. The entry will
+	// have been cached as an update.
+	if read.Version <= m.cacheVersion {
+		return
+	}
+
+	// The pending cache contains the most recent known state for the entry.
+	// If the read entry is newer than the pending entry for the key, update
+	// the pending cache.
+	if pending, ok := m.pending[read.Key]; !ok || read.Version > pending.Version {
+		m.pending[read.Key] = &cachedEntry{
+			Entry:     read,
+			tombstone: tombstone,
+		}
+	}
+}
+
+// getCache gets a cached entry
+func (m *cachingMap) getCache(key string) (*Entry, bool) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	// The pending cache contains the most recent known states. If the entry is present
+	// in the pending cache, return it rather than using the LRU cache.
+	if entry, ok := m.pending[key]; ok {
+		if entry.tombstone {
+			return nil, true
+		}
+		return entry.Entry, true
+	}
+
+	// If the entry is present in the LRU cache, return it.
+	if entry, ok := m.cache.Get(key); ok {
+		return entry.(*Entry), true
+	}
+	return nil, false
+}
+
+func (m *cachingMap) Get(ctx context.Context, key string, opts ...GetOption) (*Entry, error) {
+	// If the entry is already in the cache, return it
+	if entry, ok := m.getCache(key); ok {
+		return entry, nil
+	}
+
+	// Otherwise, fetch the entry from the underlying map
+	entry, err := m.delegatingMap.Get(ctx, key, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	// Update the cache if necessary
+	if err != nil {
+		return nil, err
+	}
+	m.cacheRead(entry, entry.Value == nil)
+	return entry, nil
+}
+
+func (m *cachingMap) Put(ctx context.Context, key string, value []byte, opts ...PutOption) (*Entry, error) {
+	// Put the entry in the map using the underlying map delegate
+	entry, err := m.delegatingMap.Put(ctx, key, value, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	// Update the cache if necessary
+	if err != nil {
+		return nil, err
+	}
+	m.cacheRead(entry, false)
+	return entry, nil
+}
+
+func (m *cachingMap) Remove(ctx context.Context, key string, opts ...RemoveOption) (*Entry, error) {
+	// Remove the entry from the map using the underlying map delegate
+	entry, err := m.delegatingMap.Remove(ctx, key, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	// Update the cache if necessary
+	if err != nil {
+		return nil, err
+	}
+	m.cacheRead(entry, true)
+	return entry, nil
+}
+
+func (m *cachingMap) Close(ctx context.Context) error {
+	m.mu.Lock()
+	if m.cancel != nil {
+		m.cancel()
+	}
+	m.mu.Unlock()
+	return m.delegatingMap.Close(ctx)
+}
+
+// cachedEntry is a cached entry
+type cachedEntry struct {
+	*Entry
+	tombstone bool
+}