a-savchuk/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#811] service/container: Hide cache invalidation logic in Writer interface

Browse source 
Signed-off-by: Alex Vanin <alexey@nspcc.ru>
This commit is contained in:
Alex Vanin 2021-09-08 20:24:16 +03:00 committed by Alex Vanin
parent 49b6b5b49d
commit 01df4ffa61
2 changed files with 64 additions and 67 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
cmd/neofs-node/container.go
View file

@ -58,7 +58,10 @@ func initContainerService(c *cfg) {
}
cnrRdr := new(morphContainerReader)
cnrInvalidator := new(morphContainerInvalidator)
cnrWrt := &morphContainerWriter{
neoClient: wrap,
}
if c.cfgMorph.disableCache {
c.cfgObject.eaclSource = eACLFetcher
@ -79,9 +82,10 @@ func initContainerService(c *cfg) {
cnrRdr.eacl = c.cfgObject.eaclSource
cnrRdr.get = c.cfgObject.cnrSource
cnrInvalidator.lists = cachedContainerLister
cnrInvalidator.eacls = cachedEACLStorage
cnrInvalidator.containers = cachedContainerStorage
cnrWrt.cacheEnabled = true
cnrWrt.lists = cachedContainerLister
cnrWrt.eacls = cachedEACLStorage
cnrWrt.containers = cachedContainerStorage
}
localMetrics := &localStorageLoad{
@ -152,7 +156,7 @@ func initContainerService(c *cfg) {
&c.key.PrivateKey,
containerService.NewResponseService(
&usedSpaceService{
Server: containerService.NewExecutionService(containerMorph.NewExecutor(wrap, cnrRdr, cnrInvalidator)),
Server: containerService.NewExecutionService(containerMorph.NewExecutor(cnrRdr, cnrWrt)),
loadWriterProvider: loadRouter,
loadPlacementBuilder: loadPlacementBuilder,
routeBuilder: routeBuilder,
@ -539,41 +543,56 @@ func (x *morphContainerReader) List(id *owner.ID) ([]*cid.ID, error) {
return x.lister.List(id)
}
type morphContainerInvalidator struct {
containers interface {
InvalidateContainer(*cid.ID)
type morphContainerWriter struct {
neoClient *wrapper.Wrapper
cacheEnabled bool
containers *ttlContainerStorage
eacls *ttlEACLStorage
lists *ttlContainerLister
}
eacls interface {
InvalidateEACL(*cid.ID)
func (m morphContainerWriter) Put(cnr *containerSDK.Container) (*cid.ID, error) {
containerID, err := wrapper.Put(m.neoClient, cnr)
if err != nil {
return nil, err
}
lists interface {
InvalidateContainerList(*owner.ID)
InvalidateContainerListByCID(*cid.ID)
}
if m.cacheEnabled {
m.lists.InvalidateContainerList(cnr.OwnerID())
}
func (x *morphContainerInvalidator) InvalidateContainer(id *cid.ID) {
if x.containers != nil {
x.containers.InvalidateContainer(id)
}
return containerID, nil
}
func (x *morphContainerInvalidator) InvalidateEACL(id *cid.ID) {
if x.eacls != nil {
x.eacls.InvalidateEACL(id)
}
func (m morphContainerWriter) Delete(witness containerCore.RemovalWitness) error {
err := wrapper.Delete(m.neoClient, witness)
if err != nil {
return err
}
func (x *morphContainerInvalidator) InvalidateContainerList(id *owner.ID) {
if x.lists != nil {
x.lists.InvalidateContainerList(id)
}
if m.cacheEnabled {
containerID := witness.ContainerID()
m.containers.InvalidateContainer(containerID)
m.eacls.InvalidateEACL(containerID)
// it is faster to use slower invalidation by CID than making separate
// network request to fetch owner ID of the container.
m.lists.InvalidateContainerListByCID(containerID)
}
func (x *morphContainerInvalidator) InvalidateContainerListByCID(id *cid.ID) {
if x.lists != nil {
x.lists.InvalidateContainerListByCID(id)
return nil
}
func (m morphContainerWriter) PutEACL(table *eaclSDK.Table) error {
err := wrapper.PutEACL(m.neoClient, table)
if err != nil {
return err
}
if m.cacheEnabled {
m.eacls.InvalidateEACL(table.CID())
}
return nil
}

50
pkg/services/container/morph/executor.go
View file

@ -12,17 +12,13 @@ import (
"github.com/nspcc-dev/neofs-api-go/v2/container"
"github.com/nspcc-dev/neofs-api-go/v2/refs"
containercore "github.com/nspcc-dev/neofs-node/pkg/core/container"
"github.com/nspcc-dev/neofs-node/pkg/morph/client/container/wrapper"
containerSvc "github.com/nspcc-dev/neofs-node/pkg/services/container"
"github.com/nspcc-dev/neofs-node/pkg/services/object/acl/eacl"
)
type morphExecutor struct {
wrapper *wrapper.Wrapper
rdr Reader
invalidator Invalidator
wrt Writer
}
// Reader is an interface of read-only container storage.
@ -36,27 +32,20 @@ type Reader interface {
List(*owner.ID) ([]*cid.ID, error)
}
// Invalidator is an interface of local cache invalidator. It removes cached
// values in order to synchronize updated data in the side chain faster.
type Invalidator interface {
// InvalidateContainer from the local container cache if it exists.
InvalidateContainer(*cid.ID)
// InvalidateEACL from the local eACL cache if it exists.
InvalidateEACL(*cid.ID)
// InvalidateContainerList from the local cache of container list results
// if it exists.
InvalidateContainerList(*owner.ID)
// InvalidateContainerListByCID from the local cache of container list
// results if it exists. Container list source uses owner.ID as a key,
// so invalidating cache record by the value requires different approach.
InvalidateContainerListByCID(*cid.ID)
// Writer is an interface of container storage updater.
type Writer interface {
// Put stores specified container in the side chain.
Put(*containerSDK.Container) (*cid.ID, error)
// Delete removes specified container from the side chain.
Delete(containercore.RemovalWitness) error
// PutEACL updates extended ACL table of specified container in the side chain.
PutEACL(*eaclSDK.Table) error
}
func NewExecutor(w *wrapper.Wrapper, rdr Reader, i Invalidator) containerSvc.ServiceExecutor {
func NewExecutor(rdr Reader, wrt Writer) containerSvc.ServiceExecutor {
return &morphExecutor{
wrapper: w,
rdr: rdr,
invalidator: i,
wrt: wrt,
}
}
@ -71,13 +60,11 @@ func (s *morphExecutor) Put(ctx containerSvc.ContextWithToken, body *container.P
session.NewTokenFromV2(ctx.SessionToken),
)
cid, err := wrapper.Put(s.wrapper, cnr)
cid, err := s.wrt.Put(cnr)
if err != nil {
return nil, err
}
s.invalidator.InvalidateContainerList(cnr.OwnerID())
res := new(container.PutResponseBody)
res.SetContainerID(cid.ToV2())
@ -95,18 +82,11 @@ func (s *morphExecutor) Delete(ctx containerSvc.ContextWithToken, body *containe
rmWitness.SetSignature(sig)
rmWitness.SetSessionToken(tok)
err := wrapper.Delete(s.wrapper, rmWitness)
err := s.wrt.Delete(rmWitness)
if err != nil {
return nil, err
}
s.invalidator.InvalidateContainer(id)
s.invalidator.InvalidateEACL(id)
// it is faster to use slower invalidation by CID than making separate
// network request to fetch owner ID of the container.
s.invalidator.InvalidateContainerListByCID(id)
return new(container.DeleteResponseBody), nil
}
@ -155,13 +135,11 @@ func (s *morphExecutor) SetExtendedACL(ctx containerSvc.ContextWithToken, body *
session.NewTokenFromV2(ctx.SessionToken),
)
err := wrapper.PutEACL(s.wrapper, table)
err := s.wrt.PutEACL(table)
if err != nil {
return nil, err
}
s.invalidator.InvalidateEACL(table.CID())
return new(container.SetExtendedACLResponseBody), nil
}