[#1956] node: Lock shard's mode on its methods switch

Signed-off-by: Pavel Karpy <p.karpy@yadro.com>
This commit is contained in:
Pavel Karpy 2022-12-07 20:42:35 +03:00 committed by Anton Nikiforov
parent 33d279a3f2
commit eea2892109
13 changed files with 94 additions and 22 deletions

View file

@ -67,6 +67,7 @@ Changelog for NeoFS Node
- Synchronizing a tree now longer reports an error for a single-node container (#2154)
- Prevent leaking goroutines in the tree service (#2162)
- Do not search for LOCK objects when delete container when session provided (#2152)
- Race conditions on shard's mode switch (#1956)
### Removed
- `-g` option from `neofs-cli control ...` and `neofs-cli container create` commands (#2089)

View file

@ -28,10 +28,16 @@ func (p *DeletePrm) SetAddresses(addr ...oid.Address) {
// Delete removes data from the shard's writeCache, metaBase and
// blobStor.
func (s *Shard) Delete(prm DeletePrm) (DeleteRes, error) {
m := s.GetMode()
if m.ReadOnly() {
s.m.RLock()
defer s.m.RUnlock()
return s.delete(prm)
}
func (s *Shard) delete(prm DeletePrm) (DeleteRes, error) {
if s.info.Mode.ReadOnly() {
return DeleteRes{}, ErrReadOnlyMode
} else if m.NoMetabase() {
} else if s.info.Mode.NoMetabase() {
return DeleteRes{}, ErrDegradedMode
}

View file

@ -37,7 +37,10 @@ func (s *Shard) Exists(prm ExistsPrm) (ExistsRes, error) {
var exists bool
var err error
if s.GetMode().NoMetabase() {
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
var p common.ExistsPrm
p.Address = prm.addr

View file

@ -187,7 +187,10 @@ func (gc *gc) stop() {
// with GC-marked graves.
// Does nothing if shard is in "read-only" mode.
func (s *Shard) removeGarbage() {
if s.GetMode() != mode.ReadWrite {
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode != mode.ReadWrite {
return
}
@ -221,7 +224,7 @@ func (s *Shard) removeGarbage() {
deletePrm.SetAddresses(buf...)
// delete accumulated objects
_, err = s.Delete(deletePrm)
_, err = s.delete(deletePrm)
if err != nil {
s.log.Warn("could not delete the objects",
zap.String("error", err.Error()),
@ -242,6 +245,13 @@ func (s *Shard) collectExpiredObjects(ctx context.Context, e Event) {
return
}
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return
}
var inhumePrm meta.InhumePrm
inhumePrm.SetAddresses(expired...)
@ -284,17 +294,25 @@ func (s *Shard) collectExpiredTombstones(ctx context.Context, e Event) {
for {
log.Debug("iterating tombstones")
if s.GetMode().NoMetabase() {
s.m.RLock()
if s.info.Mode.NoMetabase() {
s.log.Debug("shard is in a degraded mode, skip collecting expired tombstones")
s.m.RUnlock()
return
}
err := s.metaBase.IterateOverGraveyard(iterPrm)
if err != nil {
log.Error("iterator over graveyard failed", zap.Error(err))
s.m.RUnlock()
return
}
s.m.RUnlock()
tssLen := len(tss)
if tssLen == 0 {
break
@ -332,7 +350,10 @@ func (s *Shard) collectExpiredLocks(ctx context.Context, e Event) {
}
func (s *Shard) getExpiredObjects(ctx context.Context, epoch uint64, typeCond func(object.Type) bool) ([]oid.Address, error) {
if s.GetMode().NoMetabase() {
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return nil, ErrDegradedMode
}

View file

@ -62,6 +62,9 @@ func (r GetRes) HasMeta() bool {
// Returns an error of type apistatus.ObjectAlreadyRemoved if the requested object has been marked as removed in shard.
// Returns the object.ErrObjectIsExpired if the object is presented but already expired.
func (s *Shard) Get(prm GetPrm) (GetRes, error) {
s.m.RLock()
defer s.m.RUnlock()
cb := func(stor *blobstor.BlobStor, id []byte) (*objectSDK.Object, error) {
var getPrm common.GetPrm
getPrm.Address = prm.addr
@ -79,7 +82,7 @@ func (s *Shard) Get(prm GetPrm) (GetRes, error) {
return c.Get(prm.addr)
}
skipMeta := prm.skipMeta || s.GetMode().NoMetabase()
skipMeta := prm.skipMeta || s.info.Mode.NoMetabase()
obj, hasMeta, err := s.fetchObjectData(prm.addr, skipMeta, cb, wc)
return GetRes{
@ -114,7 +117,7 @@ func (s *Shard) fetchObjectData(addr oid.Address, skipMeta bool, cb storFetcher,
mPrm.SetAddress(addr)
mRes, err := s.metaBase.Exists(mPrm)
if err != nil && !s.GetMode().NoMetabase() {
if err != nil && !s.info.Mode.NoMetabase() {
return res, false, err
}
exists = mRes.Exists()

View file

@ -61,10 +61,13 @@ var ErrLockObjectRemoval = meta.ErrLockObjectRemoval
//
// Returns ErrReadOnlyMode error if shard is in "read-only" mode.
func (s *Shard) Inhume(prm InhumePrm) (InhumeRes, error) {
m := s.GetMode()
if m.ReadOnly() {
s.m.RLock()
if s.info.Mode.ReadOnly() {
s.m.RUnlock()
return InhumeRes{}, ErrReadOnlyMode
} else if m.NoMetabase() {
} else if s.info.Mode.NoMetabase() {
s.m.RUnlock()
return InhumeRes{}, ErrDegradedMode
}
@ -91,6 +94,7 @@ func (s *Shard) Inhume(prm InhumePrm) (InhumeRes, error) {
res, err := s.metaBase.Inhume(metaPrm)
if err != nil {
if errors.Is(err, meta.ErrLockObjectRemoval) {
s.m.RUnlock()
return InhumeRes{}, ErrLockObjectRemoval
}
@ -98,9 +102,13 @@ func (s *Shard) Inhume(prm InhumePrm) (InhumeRes, error) {
zap.String("error", err.Error()),
)
s.m.RUnlock()
return InhumeRes{}, fmt.Errorf("metabase inhume: %w", err)
}
s.m.RUnlock()
s.decObjectCounterBy(logical, res.AvailableInhumed())
if deletedLockObjs := res.DeletedLockObjects(); len(deletedLockObjs) != 0 {

View file

@ -64,7 +64,10 @@ func (r ListWithCursorRes) Cursor() *Cursor {
// List returns all objects physically stored in the Shard.
func (s *Shard) List() (res SelectRes, err error) {
if s.GetMode().NoMetabase() {
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return SelectRes{}, ErrDegradedMode
}

View file

@ -15,7 +15,10 @@ import (
//
// Locked list should be unique. Panics if it is empty.
func (s *Shard) Lock(idCnr cid.ID, locker oid.ID, locked []oid.ID) error {
m := s.GetMode()
s.m.RLock()
defer s.m.RUnlock()
m := s.info.Mode
if m.ReadOnly() {
return ErrReadOnlyMode
} else if m.NoMetabase() {

View file

@ -23,7 +23,10 @@ func (p *ToMoveItPrm) SetAddress(addr oid.Address) {
// ToMoveIt calls metabase.ToMoveIt method to mark object as relocatable to
// another shard.
func (s *Shard) ToMoveIt(prm ToMoveItPrm) (ToMoveItRes, error) {
m := s.GetMode()
s.m.RLock()
defer s.m.RUnlock()
m := s.info.Mode
if m.ReadOnly() {
return ToMoveItRes{}, ErrReadOnlyMode
} else if m.NoMetabase() {

View file

@ -30,7 +30,10 @@ func (p *PutPrm) SetObject(obj *object.Object) {
//
// Returns ErrReadOnlyMode error if shard is in "read-only" mode.
func (s *Shard) Put(prm PutPrm) (PutRes, error) {
m := s.GetMode()
s.m.RLock()
defer s.m.RUnlock()
m := s.info.Mode
if m.ReadOnly() {
return PutRes{}, ErrReadOnlyMode
}

View file

@ -67,6 +67,9 @@ func (r RngRes) HasMeta() bool {
// Returns an error of type apistatus.ObjectAlreadyRemoved if the requested object has been marked as removed in shard.
// Returns the object.ErrObjectIsExpired if the object is presented but already expired.
func (s *Shard) GetRange(prm RngPrm) (RngRes, error) {
s.m.RLock()
defer s.m.RUnlock()
cb := func(stor *blobstor.BlobStor, id []byte) (*object.Object, error) {
var getRngPrm common.GetRangePrm
getRngPrm.Address = prm.addr
@ -103,7 +106,7 @@ func (s *Shard) GetRange(prm RngPrm) (RngRes, error) {
return obj, nil
}
skipMeta := prm.skipMeta || s.GetMode().NoMetabase()
skipMeta := prm.skipMeta || s.info.Mode.NoMetabase()
obj, hasMeta, err := s.fetchObjectData(prm.addr, skipMeta, cb, wc)
return RngRes{

View file

@ -40,7 +40,10 @@ func (r SelectRes) AddressList() []oid.Address {
// Returns any error encountered that
// did not allow to completely select the objects.
func (s *Shard) Select(prm SelectPrm) (SelectRes, error) {
if s.GetMode().NoMetabase() {
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.NoMetabase() {
return SelectRes{}, ErrDegradedMode
}

View file

@ -16,7 +16,11 @@ func (s *Shard) TreeMove(d pilorama.CIDDescriptor, treeID string, m *pilorama.Mo
if s.pilorama == nil {
return nil, ErrPiloramaDisabled
}
if s.GetMode().ReadOnly() {
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.ReadOnly() {
return nil, ErrReadOnlyMode
}
return s.pilorama.TreeMove(d, treeID, m)
@ -27,7 +31,11 @@ func (s *Shard) TreeAddByPath(d pilorama.CIDDescriptor, treeID string, attr stri
if s.pilorama == nil {
return nil, ErrPiloramaDisabled
}
if s.GetMode().ReadOnly() {
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.ReadOnly() {
return nil, ErrReadOnlyMode
}
return s.pilorama.TreeAddByPath(d, treeID, attr, path, meta)
@ -38,7 +46,11 @@ func (s *Shard) TreeApply(d pilorama.CIDDescriptor, treeID string, m *pilorama.M
if s.pilorama == nil {
return ErrPiloramaDisabled
}
if s.GetMode().ReadOnly() {
s.m.RLock()
defer s.m.RUnlock()
if s.info.Mode.ReadOnly() {
return ErrReadOnlyMode
}
return s.pilorama.TreeApply(d, treeID, m)