forked from TrueCloudLab/frostfs-node
[#5] frost-node: Used generic cache
Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
parent
9936b112b8
commit
fdb0affc31
3 changed files with 72 additions and 100 deletions
|
@ -12,35 +12,35 @@ import (
|
|||
cid "github.com/TrueCloudLab/frostfs-sdk-go/container/id"
|
||||
netmapSDK "github.com/TrueCloudLab/frostfs-sdk-go/netmap"
|
||||
"github.com/TrueCloudLab/frostfs-sdk-go/user"
|
||||
lru "github.com/hashicorp/golang-lru"
|
||||
lru "github.com/hashicorp/golang-lru/v2"
|
||||
)
|
||||
|
||||
type netValueReader func(interface{}) (interface{}, error)
|
||||
type netValueReader[K any, V any] func(K) (V, error)
|
||||
|
||||
type valueWithTime struct {
|
||||
v interface{}
|
||||
type valueWithTime[V any] struct {
|
||||
v V
|
||||
t time.Time
|
||||
// cached error in order to not repeat failed request for some time
|
||||
e error
|
||||
}
|
||||
|
||||
// entity that provides TTL cache interface.
|
||||
type ttlNetCache struct {
|
||||
type ttlNetCache[K comparable, V any] struct {
|
||||
ttl time.Duration
|
||||
|
||||
sz int
|
||||
|
||||
cache *lru.Cache
|
||||
cache *lru.Cache[K, *valueWithTime[V]]
|
||||
|
||||
netRdr netValueReader
|
||||
netRdr netValueReader[K, V]
|
||||
}
|
||||
|
||||
// complicates netValueReader with TTL caching mechanism.
|
||||
func newNetworkTTLCache(sz int, ttl time.Duration, netRdr netValueReader) *ttlNetCache {
|
||||
cache, err := lru.New(sz)
|
||||
func newNetworkTTLCache[K comparable, V any](sz int, ttl time.Duration, netRdr netValueReader[K, V]) *ttlNetCache[K, V] {
|
||||
cache, err := lru.New[K, *valueWithTime[V]](sz)
|
||||
fatalOnErr(err)
|
||||
|
||||
return &ttlNetCache{
|
||||
return &ttlNetCache[K, V]{
|
||||
ttl: ttl,
|
||||
sz: sz,
|
||||
cache: cache,
|
||||
|
@ -53,47 +53,45 @@ func newNetworkTTLCache(sz int, ttl time.Duration, netRdr netValueReader) *ttlNe
|
|||
// updates the value from the network on cache miss or by TTL.
|
||||
//
|
||||
// returned value should not be modified.
|
||||
func (c *ttlNetCache) get(key interface{}) (interface{}, error) {
|
||||
func (c *ttlNetCache[K, V]) get(key K) (V, error) {
|
||||
val, ok := c.cache.Peek(key)
|
||||
if ok {
|
||||
valWithTime := val.(*valueWithTime)
|
||||
|
||||
if time.Since(valWithTime.t) < c.ttl {
|
||||
return valWithTime.v, valWithTime.e
|
||||
if time.Since(val.t) < c.ttl {
|
||||
return val.v, val.e
|
||||
}
|
||||
|
||||
c.cache.Remove(key)
|
||||
}
|
||||
|
||||
val, err := c.netRdr(key)
|
||||
v, err := c.netRdr(key)
|
||||
|
||||
c.set(key, val, err)
|
||||
c.set(key, v, err)
|
||||
|
||||
return val, err
|
||||
return v, err
|
||||
}
|
||||
|
||||
func (c *ttlNetCache) set(k, v interface{}, e error) {
|
||||
c.cache.Add(k, &valueWithTime{
|
||||
func (c *ttlNetCache[K, V]) set(k K, v V, e error) {
|
||||
c.cache.Add(k, &valueWithTime[V]{
|
||||
v: v,
|
||||
t: time.Now(),
|
||||
e: e,
|
||||
})
|
||||
}
|
||||
|
||||
func (c *ttlNetCache) remove(key interface{}) {
|
||||
func (c *ttlNetCache[K, V]) remove(key K) {
|
||||
c.cache.Remove(key)
|
||||
}
|
||||
|
||||
// entity that provides LRU cache interface.
|
||||
type lruNetCache struct {
|
||||
cache *lru.Cache
|
||||
cache *lru.Cache[uint64, *netmapSDK.NetMap]
|
||||
|
||||
netRdr netValueReader
|
||||
netRdr netValueReader[uint64, *netmapSDK.NetMap]
|
||||
}
|
||||
|
||||
// newNetworkLRUCache returns wrapper over netValueReader with LRU cache.
|
||||
func newNetworkLRUCache(sz int, netRdr netValueReader) *lruNetCache {
|
||||
cache, err := lru.New(sz)
|
||||
func newNetworkLRUCache(sz int, netRdr netValueReader[uint64, *netmapSDK.NetMap]) *lruNetCache {
|
||||
cache, err := lru.New[uint64, *netmapSDK.NetMap](sz)
|
||||
fatalOnErr(err)
|
||||
|
||||
return &lruNetCache{
|
||||
|
@ -107,7 +105,7 @@ func newNetworkLRUCache(sz int, netRdr netValueReader) *lruNetCache {
|
|||
// updates the value from the network on cache miss.
|
||||
//
|
||||
// returned value should not be modified.
|
||||
func (c *lruNetCache) get(key interface{}) (interface{}, error) {
|
||||
func (c *lruNetCache) get(key uint64) (*netmapSDK.NetMap, error) {
|
||||
val, ok := c.cache.Get(key)
|
||||
if ok {
|
||||
return val, nil
|
||||
|
@ -125,73 +123,53 @@ func (c *lruNetCache) get(key interface{}) (interface{}, error) {
|
|||
|
||||
// wrapper over TTL cache of values read from the network
|
||||
// that implements container storage.
|
||||
type ttlContainerStorage ttlNetCache
|
||||
type ttlContainerStorage struct {
|
||||
*ttlNetCache[cid.ID, *container.Container]
|
||||
}
|
||||
|
||||
func newCachedContainerStorage(v container.Source, ttl time.Duration) *ttlContainerStorage {
|
||||
func newCachedContainerStorage(v container.Source, ttl time.Duration) ttlContainerStorage {
|
||||
const containerCacheSize = 100
|
||||
|
||||
lruCnrCache := newNetworkTTLCache(containerCacheSize, ttl, func(key interface{}) (interface{}, error) {
|
||||
var id cid.ID
|
||||
|
||||
err := id.DecodeString(key.(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
lruCnrCache := newNetworkTTLCache[cid.ID, *container.Container](containerCacheSize, ttl, func(id cid.ID) (*container.Container, error) {
|
||||
return v.Get(id)
|
||||
})
|
||||
|
||||
return (*ttlContainerStorage)(lruCnrCache)
|
||||
return ttlContainerStorage{lruCnrCache}
|
||||
}
|
||||
|
||||
func (s *ttlContainerStorage) handleRemoval(cnr cid.ID) {
|
||||
(*ttlNetCache)(s).set(cnr.EncodeToString(), nil, apistatus.ContainerNotFound{})
|
||||
func (s ttlContainerStorage) handleRemoval(cnr cid.ID) {
|
||||
s.set(cnr, nil, apistatus.ContainerNotFound{})
|
||||
}
|
||||
|
||||
// Get returns container value from the cache. If value is missing in the cache
|
||||
// or expired, then it returns value from side chain and updates the cache.
|
||||
func (s *ttlContainerStorage) Get(cnr cid.ID) (*container.Container, error) {
|
||||
val, err := (*ttlNetCache)(s).get(cnr.EncodeToString())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return val.(*container.Container), nil
|
||||
func (s ttlContainerStorage) Get(cnr cid.ID) (*container.Container, error) {
|
||||
return s.get(cnr)
|
||||
}
|
||||
|
||||
type ttlEACLStorage ttlNetCache
|
||||
type ttlEACLStorage struct {
|
||||
*ttlNetCache[cid.ID, *container.EACL]
|
||||
}
|
||||
|
||||
func newCachedEACLStorage(v container.EACLSource, ttl time.Duration) *ttlEACLStorage {
|
||||
func newCachedEACLStorage(v container.EACLSource, ttl time.Duration) ttlEACLStorage {
|
||||
const eaclCacheSize = 100
|
||||
|
||||
lruCnrCache := newNetworkTTLCache(eaclCacheSize, ttl, func(key interface{}) (interface{}, error) {
|
||||
var id cid.ID
|
||||
|
||||
err := id.DecodeString(key.(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
lruCnrCache := newNetworkTTLCache(eaclCacheSize, ttl, func(id cid.ID) (*container.EACL, error) {
|
||||
return v.GetEACL(id)
|
||||
})
|
||||
|
||||
return (*ttlEACLStorage)(lruCnrCache)
|
||||
return ttlEACLStorage{lruCnrCache}
|
||||
}
|
||||
|
||||
// GetEACL returns eACL value from the cache. If value is missing in the cache
|
||||
// or expired, then it returns value from side chain and updates cache.
|
||||
func (s *ttlEACLStorage) GetEACL(cnr cid.ID) (*container.EACL, error) {
|
||||
val, err := (*ttlNetCache)(s).get(cnr.EncodeToString())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return val.(*container.EACL), nil
|
||||
func (s ttlEACLStorage) GetEACL(cnr cid.ID) (*container.EACL, error) {
|
||||
return s.get(cnr)
|
||||
}
|
||||
|
||||
// InvalidateEACL removes cached eACL value.
|
||||
func (s *ttlEACLStorage) InvalidateEACL(cnr cid.ID) {
|
||||
(*ttlNetCache)(s).remove(cnr.EncodeToString())
|
||||
func (s ttlEACLStorage) InvalidateEACL(cnr cid.ID) {
|
||||
s.remove(cnr)
|
||||
}
|
||||
|
||||
type lruNetmapSource struct {
|
||||
|
@ -203,8 +181,8 @@ type lruNetmapSource struct {
|
|||
func newCachedNetmapStorage(s netmap.State, v netmap.Source) netmap.Source {
|
||||
const netmapCacheSize = 10
|
||||
|
||||
lruNetmapCache := newNetworkLRUCache(netmapCacheSize, func(key interface{}) (interface{}, error) {
|
||||
return v.GetNetMapByEpoch(key.(uint64))
|
||||
lruNetmapCache := newNetworkLRUCache(netmapCacheSize, func(key uint64) (*netmapSDK.NetMap, error) {
|
||||
return v.GetNetMapByEpoch(key)
|
||||
})
|
||||
|
||||
return &lruNetmapSource{
|
||||
|
@ -227,7 +205,7 @@ func (s *lruNetmapSource) getNetMapByEpoch(epoch uint64) (*netmapSDK.NetMap, err
|
|||
return nil, err
|
||||
}
|
||||
|
||||
return val.(*netmapSDK.NetMap), nil
|
||||
return val, nil
|
||||
}
|
||||
|
||||
func (s *lruNetmapSource) Epoch() (uint64, error) {
|
||||
|
@ -236,7 +214,9 @@ func (s *lruNetmapSource) Epoch() (uint64, error) {
|
|||
|
||||
// wrapper over TTL cache of values read from the network
|
||||
// that implements container lister.
|
||||
type ttlContainerLister ttlNetCache
|
||||
type ttlContainerLister struct {
|
||||
*ttlNetCache[string, *cacheItemContainerList]
|
||||
}
|
||||
|
||||
// value type for ttlNetCache used by ttlContainerLister.
|
||||
type cacheItemContainerList struct {
|
||||
|
@ -246,14 +226,11 @@ type cacheItemContainerList struct {
|
|||
list []cid.ID
|
||||
}
|
||||
|
||||
func newCachedContainerLister(c *cntClient.Client, ttl time.Duration) *ttlContainerLister {
|
||||
func newCachedContainerLister(c *cntClient.Client, ttl time.Duration) ttlContainerLister {
|
||||
const containerListerCacheSize = 100
|
||||
|
||||
lruCnrListerCache := newNetworkTTLCache(containerListerCacheSize, ttl, func(key interface{}) (interface{}, error) {
|
||||
var (
|
||||
id *user.ID
|
||||
strID = key.(string)
|
||||
)
|
||||
lruCnrListerCache := newNetworkTTLCache(containerListerCacheSize, ttl, func(strID string) (*cacheItemContainerList, error) {
|
||||
var id *user.ID
|
||||
|
||||
if strID != "" {
|
||||
id = new(user.ID)
|
||||
|
@ -274,28 +251,24 @@ func newCachedContainerLister(c *cntClient.Client, ttl time.Duration) *ttlContai
|
|||
}, nil
|
||||
})
|
||||
|
||||
return (*ttlContainerLister)(lruCnrListerCache)
|
||||
return ttlContainerLister{lruCnrListerCache}
|
||||
}
|
||||
|
||||
// List returns list of container IDs from the cache. If list is missing in the
|
||||
// cache or expired, then it returns container IDs from side chain and updates
|
||||
// the cache.
|
||||
func (s *ttlContainerLister) List(id *user.ID) ([]cid.ID, error) {
|
||||
func (s ttlContainerLister) List(id *user.ID) ([]cid.ID, error) {
|
||||
var str string
|
||||
|
||||
if id != nil {
|
||||
str = id.EncodeToString()
|
||||
}
|
||||
|
||||
val, err := (*ttlNetCache)(s).get(str)
|
||||
item, err := s.get(str)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// panic on typecast below is OK since developer must be careful,
|
||||
// runtime can do nothing with wrong type occurrence
|
||||
item := val.(*cacheItemContainerList)
|
||||
|
||||
item.mtx.RLock()
|
||||
res := make([]cid.ID, len(item.list))
|
||||
copy(res, item.list)
|
||||
|
@ -314,16 +287,14 @@ func (s *ttlContainerLister) List(id *user.ID) ([]cid.ID, error) {
|
|||
func (s *ttlContainerLister) update(owner user.ID, cnr cid.ID, add bool) {
|
||||
strOwner := owner.EncodeToString()
|
||||
|
||||
val, ok := (*ttlNetCache)(s).cache.Get(strOwner)
|
||||
val, ok := s.cache.Get(strOwner)
|
||||
if !ok {
|
||||
// we could cache the single cnr but in this case we will disperse
|
||||
// with the Sidechain a lot
|
||||
return
|
||||
}
|
||||
|
||||
// panic on typecast below is OK since developer must be careful,
|
||||
// runtime can do nothing with wrong type occurrence
|
||||
item := val.(*valueWithTime).v.(*cacheItemContainerList)
|
||||
item := val.v
|
||||
|
||||
item.mtx.Lock()
|
||||
{
|
||||
|
@ -349,9 +320,11 @@ func (s *ttlContainerLister) update(owner user.ID, cnr cid.ID, add bool) {
|
|||
item.mtx.Unlock()
|
||||
}
|
||||
|
||||
type cachedIRFetcher ttlNetCache
|
||||
type cachedIRFetcher struct {
|
||||
*ttlNetCache[struct{}, [][]byte]
|
||||
}
|
||||
|
||||
func newCachedIRFetcher(f interface{ InnerRingKeys() ([][]byte, error) }) *cachedIRFetcher {
|
||||
func newCachedIRFetcher(f interface{ InnerRingKeys() ([][]byte, error) }) cachedIRFetcher {
|
||||
const (
|
||||
irFetcherCacheSize = 1 // we intend to store only one value
|
||||
|
||||
|
@ -364,25 +337,25 @@ func newCachedIRFetcher(f interface{ InnerRingKeys() ([][]byte, error) }) *cache
|
|||
irFetcherCacheTTL = 30 * time.Second
|
||||
)
|
||||
|
||||
irFetcherCache := newNetworkTTLCache(irFetcherCacheSize, irFetcherCacheTTL,
|
||||
func(key interface{}) (interface{}, error) {
|
||||
irFetcherCache := newNetworkTTLCache[struct{}, [][]byte](irFetcherCacheSize, irFetcherCacheTTL,
|
||||
func(_ struct{}) ([][]byte, error) {
|
||||
return f.InnerRingKeys()
|
||||
},
|
||||
)
|
||||
|
||||
return (*cachedIRFetcher)(irFetcherCache)
|
||||
return cachedIRFetcher{irFetcherCache}
|
||||
}
|
||||
|
||||
// InnerRingKeys returns cached list of Inner Ring keys. If keys are missing in
|
||||
// the cache or expired, then it returns keys from side chain and updates
|
||||
// the cache.
|
||||
func (f *cachedIRFetcher) InnerRingKeys() ([][]byte, error) {
|
||||
val, err := (*ttlNetCache)(f).get("")
|
||||
func (f cachedIRFetcher) InnerRingKeys() ([][]byte, error) {
|
||||
val, err := f.get(struct{}{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return val.([][]byte), nil
|
||||
return val, nil
|
||||
}
|
||||
|
||||
type ttlMaxObjectSizeCache struct {
|
||||
|
|
|
@ -657,8 +657,8 @@ type morphContainerWriter struct {
|
|||
neoClient *cntClient.Client
|
||||
|
||||
cacheEnabled bool
|
||||
eacls *ttlEACLStorage
|
||||
lists *ttlContainerLister
|
||||
eacls ttlEACLStorage
|
||||
lists ttlContainerLister
|
||||
}
|
||||
|
||||
func (m morphContainerWriter) Put(cnr containerCore.Container) (*cid.ID, error) {
|
||||
|
|
5
go.mod
5
go.mod
|
@ -13,7 +13,7 @@ require (
|
|||
github.com/flynn-archive/go-shlex v0.0.0-20150515145356-3f9db97f8568
|
||||
github.com/google/go-github/v39 v39.2.0
|
||||
github.com/google/uuid v1.3.0
|
||||
github.com/hashicorp/golang-lru v0.5.4
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.1
|
||||
github.com/klauspost/compress v1.15.13
|
||||
github.com/mitchellh/go-homedir v1.1.0
|
||||
github.com/mr-tron/base58 v1.2.0
|
||||
|
@ -39,8 +39,6 @@ require (
|
|||
gopkg.in/yaml.v3 v3.0.1
|
||||
)
|
||||
|
||||
require github.com/hashicorp/golang-lru/v2 v2.0.1
|
||||
|
||||
require (
|
||||
github.com/TrueCloudLab/frostfs-crypto v0.5.0 // indirect
|
||||
github.com/TrueCloudLab/rfc6979 v0.3.0 // indirect
|
||||
|
@ -56,6 +54,7 @@ require (
|
|||
github.com/golang/snappy v0.0.3 // indirect
|
||||
github.com/google/go-querystring v1.1.0 // indirect
|
||||
github.com/gorilla/websocket v1.4.2 // indirect
|
||||
github.com/hashicorp/golang-lru v0.5.4 // indirect
|
||||
github.com/hashicorp/hcl v1.0.0 // indirect
|
||||
github.com/holiman/uint256 v1.2.0 // indirect
|
||||
github.com/inconshreveable/mousetrap v1.0.1 // indirect
|
||||
|
|
Loading…
Reference in a new issue