From aa8dd18b078b9325049c575dfcac5ab51f2ca0f3 Mon Sep 17 00:00:00 2001 From: Alex Vanin Date: Tue, 1 Dec 2020 13:18:25 +0300 Subject: [PATCH] [#222] Update Put method in storage engine Signed-off-by: Alex Vanin --- pkg/local_object_storage/engine/delete.go | 2 +- pkg/local_object_storage/engine/get.go | 2 +- pkg/local_object_storage/engine/head.go | 2 +- pkg/local_object_storage/engine/inhume.go | 2 +- pkg/local_object_storage/engine/put.go | 97 ++++++++++++++--------- pkg/local_object_storage/engine/select.go | 2 +- pkg/local_object_storage/engine/shards.go | 25 +++++- 7 files changed, 86 insertions(+), 46 deletions(-) diff --git a/pkg/local_object_storage/engine/delete.go b/pkg/local_object_storage/engine/delete.go index 8b1aad28..b73999f8 100644 --- a/pkg/local_object_storage/engine/delete.go +++ b/pkg/local_object_storage/engine/delete.go @@ -33,7 +33,7 @@ func (e *StorageEngine) Delete(prm *DeletePrm) (*DeleteRes, error) { shPrm := new(shard.DeletePrm). WithAddress(prm.addr) - e.iterateOverSortedShards(prm.addr, func(sh *shard.Shard) (stop bool) { + e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) { _, err := sh.Delete(shPrm) if err != nil { // TODO: smth wrong with shard, need to be processed diff --git a/pkg/local_object_storage/engine/get.go b/pkg/local_object_storage/engine/get.go index 92aaed34..a42fceb7 100644 --- a/pkg/local_object_storage/engine/get.go +++ b/pkg/local_object_storage/engine/get.go @@ -69,7 +69,7 @@ func (e *StorageEngine) Get(prm *GetPrm) (*GetRes, error) { shPrm = shPrm.WithRange(prm.off, int64(prm.ln)) } - e.iterateOverSortedShards(prm.addr, func(sh *shard.Shard) (stop bool) { + e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) { res, err := sh.Get(shPrm) if err != nil { if !errors.Is(err, object.ErrNotFound) { diff --git a/pkg/local_object_storage/engine/head.go b/pkg/local_object_storage/engine/head.go index 19272ed5..cff15b23 100644 --- a/pkg/local_object_storage/engine/head.go +++ b/pkg/local_object_storage/engine/head.go @@ -48,7 +48,7 @@ func (e *StorageEngine) Head(prm *HeadPrm) (*HeadRes, error) { shPrm := new(shard.GetPrm). WithAddress(prm.addr) - e.iterateOverSortedShards(prm.addr, func(sh *shard.Shard) (stop bool) { + e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) { res, err := sh.Get(shPrm) if err != nil { if !errors.Is(err, object.ErrNotFound) { diff --git a/pkg/local_object_storage/engine/inhume.go b/pkg/local_object_storage/engine/inhume.go index 18bd1bc0..94bf1537 100644 --- a/pkg/local_object_storage/engine/inhume.go +++ b/pkg/local_object_storage/engine/inhume.go @@ -30,7 +30,7 @@ func (p *InhumePrm) WithTarget(addr, tombstone *objectSDK.Address) *InhumePrm { func (e *StorageEngine) Inhume(prm *InhumePrm) (*InhumeRes, error) { shPrm := new(shard.InhumePrm).WithTarget(prm.addr, prm.tombstone) - e.iterateOverSortedShards(prm.addr, func(sh *shard.Shard) (stop bool) { + e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) { _, err := sh.Inhume(shPrm) if err != nil { // TODO: smth wrong with shard, need to be processed diff --git a/pkg/local_object_storage/engine/put.go b/pkg/local_object_storage/engine/put.go index 97c7d330..07147719 100644 --- a/pkg/local_object_storage/engine/put.go +++ b/pkg/local_object_storage/engine/put.go @@ -4,6 +4,7 @@ import ( "errors" "github.com/nspcc-dev/neofs-node/pkg/core/object" + meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase/v2" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard" "go.uber.org/zap" ) @@ -34,60 +35,78 @@ func (p *PutPrm) WithObject(obj *object.Object) *PutPrm { // Returns any error encountered that // did not allow to completely save the object. func (e *StorageEngine) Put(prm *PutPrm) (*PutRes, error) { - // choose shards through sorting by weight - sortedShards := e.sortShardsByWeight(prm.obj.Address()) + alreadyRemoved := false // first check if object has not been marked as removed - // check object existence - if e.objectExists(prm.obj, sortedShards) { - return nil, nil - } + existPrm := new(shard.ExistsPrm) + existPrm.WithAddress(prm.obj.Address()) - shPrm := new(shard.PutPrm) + // todo: make this check parallel + e.iterateOverUnsortedShards(func(s *shard.Shard) (stop bool) { + _, err := s.Exists(existPrm) + if err != nil && errors.Is(err, meta.ErrAlreadyRemoved) { + alreadyRemoved = true - // save the object into the "largest" possible shard - for _, sh := range sortedShards { - _, err := sh.sh.Put( - shPrm.WithObject(prm.obj), - ) - - if err != nil { - // TODO: smth wrong with shard, need to be processed - e.log.Warn("could not save object in shard", - zap.Stringer("shard", sh.sh.ID()), - zap.String("error", err.Error()), - ) - } else { - return nil, nil + return true } + + return false + }) + + if alreadyRemoved { + return nil, meta.ErrAlreadyRemoved } - return nil, errPutShard -} - -func (e *StorageEngine) objectExists(obj *object.Object, shards []hashedShard) bool { - exists := false - - for _, sh := range shards { - res, err := sh.sh.Exists( - new(shard.ExistsPrm). - WithAddress(obj.Address()), - ) + finished := false + e.iterateOverSortedShards(prm.obj.Address(), func(ind int, s *shard.Shard) (stop bool) { + exists, err := s.Exists(existPrm) if err != nil { - // TODO: smth wrong with shard, need to be processed - e.log.Warn("could not check object existence", + return false // this is not ErrAlreadyRemoved error so we can go to the next shard + } + + if exists.Exists() { + if ind != 0 { + toMoveItPrm := new(shard.ToMoveItPrm) + toMoveItPrm.WithAddress(prm.obj.Address()) + + _, err = s.ToMoveIt(toMoveItPrm) + if err != nil { + e.log.Warn("could not mark object for shard relocation", + zap.Stringer("shard", s.ID()), + zap.String("error", err.Error()), + ) + } + } + + finished = true + + return true + } + + putPrm := new(shard.PutPrm) + putPrm.WithObject(prm.obj) + + _, err = s.Put(putPrm) + if err != nil { + e.log.Warn("could not put object in shard", + zap.Stringer("shard", s.ID()), zap.String("error", err.Error()), ) - continue + return false } - if exists = res.Exists(); exists { - break - } + finished = true + return true + }) + + var err error = nil + + if !finished { + err = errPutShard } - return exists + return nil, err } // Put writes provided object to local storage. diff --git a/pkg/local_object_storage/engine/select.go b/pkg/local_object_storage/engine/select.go index 5ed46631..0272a637 100644 --- a/pkg/local_object_storage/engine/select.go +++ b/pkg/local_object_storage/engine/select.go @@ -39,7 +39,7 @@ func (e *StorageEngine) Select(prm *SelectPrm) (*SelectRes, error) { shPrm := new(shard.SelectPrm). WithFilters(prm.filters) - e.iterateOverSortedShards(nil, func(sh *shard.Shard) (stop bool) { + e.iterateOverSortedShards(nil, func(_ int, sh *shard.Shard) (stop bool) { res, err := sh.Select(shPrm) if err != nil { // TODO: smth wrong with shard, need to be processed diff --git a/pkg/local_object_storage/engine/shards.go b/pkg/local_object_storage/engine/shards.go index 5bde2759..d125562d 100644 --- a/pkg/local_object_storage/engine/shards.go +++ b/pkg/local_object_storage/engine/shards.go @@ -71,8 +71,29 @@ func (e *StorageEngine) sortShardsByWeight(objAddr fmt.Stringer) []hashedShard { return shards } -func (e *StorageEngine) iterateOverSortedShards(addr *object.Address, handler func(*shard.Shard) (stop bool)) { - for _, sh := range e.sortShardsByWeight(addr) { +func (e *StorageEngine) unsortedShards() []hashedShard { + e.mtx.RLock() + defer e.mtx.RUnlock() + + shards := make([]hashedShard, 0, len(e.shards)) + + for _, sh := range e.shards { + shards = append(shards, hashedShard{sh}) + } + + return shards +} + +func (e *StorageEngine) iterateOverSortedShards(addr *object.Address, handler func(int, *shard.Shard) (stop bool)) { + for i, sh := range e.sortShardsByWeight(addr) { + if handler(i, sh.sh) { + break + } + } +} + +func (e *StorageEngine) iterateOverUnsortedShards(handler func(*shard.Shard) (stop bool)) { + for _, sh := range e.unsortedShards() { if handler(sh.sh) { break }