[#2057] blobstor: Block operations on a mode change

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
Pavel Karpy 2022-11-15 15:33:48 +03:00 committed by Pavel Karpy
parent 27ca754dc1
commit 479601ceb9
7 changed files with 21 additions and 0 deletions

View file

@ -8,6 +8,9 @@ import (
) )
func (b *BlobStor) Delete(prm common.DeletePrm) (common.DeleteRes, error) { func (b *BlobStor) Delete(prm common.DeletePrm) (common.DeleteRes, error) {
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
if prm.StorageID == nil { if prm.StorageID == nil {
for i := range b.storage { for i := range b.storage {
res, err := b.storage[i].Storage.Delete(prm) res, err := b.storage[i].Storage.Delete(prm)

View file

@ -10,6 +10,9 @@ import (
// Returns any error encountered that did not allow // Returns any error encountered that did not allow
// to completely check object existence. // to completely check object existence.
func (b *BlobStor) Exists(prm common.ExistsPrm) (common.ExistsRes, error) { func (b *BlobStor) Exists(prm common.ExistsPrm) (common.ExistsRes, error) {
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
// If there was an error during existence check below, // If there was an error during existence check below,
// it will be returned unless object was found in blobovnicza. // it will be returned unless object was found in blobovnicza.
// Otherwise, it is logged and the latest error is returned. // Otherwise, it is logged and the latest error is returned.

View file

@ -12,6 +12,9 @@ import (
// If the descriptor is present, only one sub-storage is tried, // If the descriptor is present, only one sub-storage is tried,
// Otherwise, each sub-storage is tried in order. // Otherwise, each sub-storage is tried in order.
func (b *BlobStor) Get(prm common.GetPrm) (common.GetRes, error) { func (b *BlobStor) Get(prm common.GetPrm) (common.GetRes, error) {
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
if prm.StorageID == nil { if prm.StorageID == nil {
for i := range b.storage { for i := range b.storage {
res, err := b.storage[i].Storage.Get(prm) res, err := b.storage[i].Storage.Get(prm)

View file

@ -12,6 +12,9 @@ import (
// If the descriptor is present, only one sub-storage is tried, // If the descriptor is present, only one sub-storage is tried,
// Otherwise, each sub-storage is tried in order. // Otherwise, each sub-storage is tried in order.
func (b *BlobStor) GetRange(prm common.GetRangePrm) (common.GetRangeRes, error) { func (b *BlobStor) GetRange(prm common.GetRangePrm) (common.GetRangeRes, error) {
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
if prm.StorageID == nil { if prm.StorageID == nil {
for i := range b.storage { for i := range b.storage {
res, err := b.storage[i].Storage.GetRange(prm) res, err := b.storage[i].Storage.GetRange(prm)

View file

@ -2,6 +2,9 @@ package blobstor
// DumpInfo returns information about blob stor. // DumpInfo returns information about blob stor.
func (b *BlobStor) DumpInfo() Info { func (b *BlobStor) DumpInfo() Info {
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
sub := make([]SubStorageInfo, len(b.storage)) sub := make([]SubStorageInfo, len(b.storage))
for i := range b.storage { for i := range b.storage {
sub[i].Path = b.storage[i].Storage.Path() sub[i].Path = b.storage[i].Storage.Path()

View file

@ -16,6 +16,9 @@ import (
// //
// If handler returns an error, method wraps and returns it immediately. // If handler returns an error, method wraps and returns it immediately.
func (b *BlobStor) Iterate(prm common.IteratePrm) (common.IterateRes, error) { func (b *BlobStor) Iterate(prm common.IteratePrm) (common.IterateRes, error) {
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
for i := range b.storage { for i := range b.storage {
_, err := b.storage[i].Storage.Iterate(prm) _, err := b.storage[i].Storage.Iterate(prm)
if err != nil && !prm.IgnoreErrors { if err != nil && !prm.IgnoreErrors {

View file

@ -22,6 +22,9 @@ var ErrNoPlaceFound = logicerr.New("couldn't find a place to store an object")
// Returns any error encountered that // Returns any error encountered that
// did not allow to completely save the object. // did not allow to completely save the object.
func (b *BlobStor) Put(prm common.PutPrm) (common.PutRes, error) { func (b *BlobStor) Put(prm common.PutPrm) (common.PutRes, error) {
b.modeMtx.RLock()
defer b.modeMtx.RUnlock()
if prm.Object != nil { if prm.Object != nil {
prm.Address = object.AddressOf(prm.Object) prm.Address = object.AddressOf(prm.Object)
} }