[#1523] local_object_storage: Unify parameters for the GetRange operation

Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
Evgenii Stratonikov 2022-07-06 14:54:23 +03:00 committed by fyrchik
parent d9a2f280c9
commit 36c88f0dc8
6 changed files with 69 additions and 86 deletions

View file

@ -317,11 +317,11 @@ func (b *Blobovniczas) Delete(prm DeleteSmallPrm) (res DeleteSmallRes, err error
// //
// If blobocvnicza ID is specified, only this blobovnicza is processed. // If blobocvnicza ID is specified, only this blobovnicza is processed.
// Otherwise, all Blobovniczas are processed descending weight. // Otherwise, all Blobovniczas are processed descending weight.
func (b *Blobovniczas) GetRange(prm GetRangeSmallPrm) (res GetRangeSmallRes, err error) { func (b *Blobovniczas) GetRange(prm common.GetRangePrm) (res common.GetRangeRes, err error) {
if prm.blobovniczaID != nil { if prm.BlobovniczaID != nil {
blz, err := b.openBlobovnicza(prm.blobovniczaID.String()) blz, err := b.openBlobovnicza(prm.BlobovniczaID.String())
if err != nil { if err != nil {
return GetRangeSmallRes{}, err return common.GetRangeRes{}, err
} }
return b.getObjectRange(blz, prm) return b.getObjectRange(blz, prm)
@ -330,7 +330,7 @@ func (b *Blobovniczas) GetRange(prm GetRangeSmallPrm) (res GetRangeSmallRes, err
activeCache := make(map[string]struct{}) activeCache := make(map[string]struct{})
objectFound := false objectFound := false
err = b.iterateSortedLeaves(&prm.addr, func(p string) (bool, error) { err = b.iterateSortedLeaves(&prm.Address, func(p string) (bool, error) {
dirPath := filepath.Dir(p) dirPath := filepath.Dir(p)
_, ok := activeCache[dirPath] _, ok := activeCache[dirPath]
@ -361,7 +361,7 @@ func (b *Blobovniczas) GetRange(prm GetRangeSmallPrm) (res GetRangeSmallRes, err
// not found in any blobovnicza // not found in any blobovnicza
var errNotFound apistatus.ObjectNotFound var errNotFound apistatus.ObjectNotFound
return GetRangeSmallRes{}, errNotFound return common.GetRangeRes{}, errNotFound
} }
return return
@ -493,7 +493,7 @@ func (b *Blobovniczas) getObjectFromLevel(prm blobovnicza.GetPrm, blzPath string
// tries to read range of object payload data from particular blobovnicza. // tries to read range of object payload data from particular blobovnicza.
// //
// returns error if object could not be read from any blobovnicza of the same level. // returns error if object could not be read from any blobovnicza of the same level.
func (b *Blobovniczas) getRangeFromLevel(prm GetRangeSmallPrm, blzPath string, tryActive bool) (GetRangeSmallRes, error) { func (b *Blobovniczas) getRangeFromLevel(prm common.GetRangePrm, blzPath string, tryActive bool) (common.GetRangeRes, error) {
lvlPath := filepath.Dir(blzPath) lvlPath := filepath.Dir(blzPath)
// try to read from blobovnicza if it is opened // try to read from blobovnicza if it is opened
@ -551,13 +551,13 @@ func (b *Blobovniczas) getRangeFromLevel(prm GetRangeSmallPrm, blzPath string, t
var errNotFound apistatus.ObjectNotFound var errNotFound apistatus.ObjectNotFound
return GetRangeSmallRes{}, errNotFound return common.GetRangeRes{}, errNotFound
} }
// open blobovnicza (cached inside) // open blobovnicza (cached inside)
blz, err := b.openBlobovnicza(blzPath) blz, err := b.openBlobovnicza(blzPath)
if err != nil { if err != nil {
return GetRangeSmallRes{}, err return common.GetRangeRes{}, err
} }
return b.getObjectRange(blz, prm) return b.getObjectRange(blz, prm)
@ -602,9 +602,9 @@ func (b *Blobovniczas) getObject(blz *blobovnicza.Blobovnicza, prm blobovnicza.G
} }
// reads range of object payload data from blobovnicza and returns GetRangeSmallRes. // reads range of object payload data from blobovnicza and returns GetRangeSmallRes.
func (b *Blobovniczas) getObjectRange(blz *blobovnicza.Blobovnicza, prm GetRangeSmallPrm) (GetRangeSmallRes, error) { func (b *Blobovniczas) getObjectRange(blz *blobovnicza.Blobovnicza, prm common.GetRangePrm) (common.GetRangeRes, error) {
var gPrm blobovnicza.GetPrm var gPrm blobovnicza.GetPrm
gPrm.SetAddress(prm.addr) gPrm.SetAddress(prm.Address)
// we don't use GetRange call for now since blobovnicza // we don't use GetRange call for now since blobovnicza
// stores data that is compressed on BlobStor side. // stores data that is compressed on BlobStor side.
@ -612,35 +612,33 @@ func (b *Blobovniczas) getObjectRange(blz *blobovnicza.Blobovnicza, prm GetRange
// we can start using GetRange. // we can start using GetRange.
res, err := blz.Get(gPrm) res, err := blz.Get(gPrm)
if err != nil { if err != nil {
return GetRangeSmallRes{}, err return common.GetRangeRes{}, err
} }
// decompress the data // decompress the data
data, err := b.Decompress(res.Object()) data, err := b.Decompress(res.Object())
if err != nil { if err != nil {
return GetRangeSmallRes{}, fmt.Errorf("could not decompress object data: %w", err) return common.GetRangeRes{}, fmt.Errorf("could not decompress object data: %w", err)
} }
// unmarshal the object // unmarshal the object
obj := objectSDK.New() obj := objectSDK.New()
if err := obj.Unmarshal(data); err != nil { if err := obj.Unmarshal(data); err != nil {
return GetRangeSmallRes{}, fmt.Errorf("could not unmarshal the object: %w", err) return common.GetRangeRes{}, fmt.Errorf("could not unmarshal the object: %w", err)
} }
from := prm.rng.GetOffset() from := prm.Range.GetOffset()
to := from + prm.rng.GetLength() to := from + prm.Range.GetLength()
payload := obj.Payload() payload := obj.Payload()
if pLen := uint64(len(payload)); to < from || pLen < from || pLen < to { if pLen := uint64(len(payload)); to < from || pLen < from || pLen < to {
var errOutOfRange apistatus.ObjectOutOfRange var errOutOfRange apistatus.ObjectOutOfRange
return GetRangeSmallRes{}, errOutOfRange return common.GetRangeRes{}, errOutOfRange
} }
return GetRangeSmallRes{ return common.GetRangeRes{
rangeData{ Data: payload[from:to],
data: payload[from:to],
},
}, nil }, nil
} }

View file

@ -93,30 +93,27 @@ func TestBlobovniczas(t *testing.T) {
require.Equal(t, obj, res.Object) require.Equal(t, obj, res.Object)
// get range w/ blobovnicza ID // get range w/ blobovnicza ID
var rngPrm GetRangeSmallPrm var rngPrm common.GetRangePrm
rngPrm.SetBlobovniczaID(id) rngPrm.BlobovniczaID = id
rngPrm.SetAddress(addr) rngPrm.Address = addr
payload := obj.Payload() payload := obj.Payload()
pSize := uint64(len(obj.Payload())) pSize := uint64(len(obj.Payload()))
rng := objectSDK.NewRange()
rngPrm.SetRange(rng)
off, ln := pSize/3, 2*pSize/3 off, ln := pSize/3, 2*pSize/3
rng.SetOffset(off) rngPrm.Range.SetOffset(off)
rng.SetLength(ln) rngPrm.Range.SetLength(ln)
rngRes, err := b.GetRange(rngPrm) rngRes, err := b.GetRange(rngPrm)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, payload[off:off+ln], rngRes.RangeData()) require.Equal(t, payload[off:off+ln], rngRes.Data)
// get range w/o blobovnicza ID // get range w/o blobovnicza ID
rngPrm.SetBlobovniczaID(nil) rngPrm.BlobovniczaID = nil
rngRes, err = b.GetRange(rngPrm) rngRes, err = b.GetRange(rngPrm)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, payload[off:off+ln], rngRes.RangeData()) require.Equal(t, payload[off:off+ln], rngRes.Data)
} }
var dPrm DeleteSmallPrm var dPrm DeleteSmallPrm

View file

@ -1,13 +0,0 @@
package blobovniczatree
// GetRangeSmallPrm groups the parameters of GetRangeSmall operation.
type GetRangeSmallPrm struct {
address
rwRange
rwBlobovniczaID
}
// GetRangeSmallRes groups the resulting values of GetRangeSmall operation.
type GetRangeSmallRes struct {
rangeData
}

View file

@ -0,0 +1,17 @@
package common
import (
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobovnicza"
objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
)
type GetRangePrm struct {
Address oid.Address
Range objectSDK.Range
BlobovniczaID *blobovnicza.ID
}
type GetRangeRes struct {
Data []byte
}

View file

@ -4,24 +4,12 @@ import (
"errors" "errors"
"fmt" "fmt"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/blobovniczatree"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/fstree" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/fstree"
apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status" apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
objectSDK "github.com/nspcc-dev/neofs-sdk-go/object" objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
) )
// GetRangeBigPrm groups the parameters of GetRangeBig operation.
type GetRangeBigPrm struct {
address
rwRange
}
// GetRangeBigRes groups the resulting values of GetRangeBig operation.
type GetRangeBigRes struct {
rangeData
}
// GetRangeBig reads data of object payload range from shallow dir of BLOB storage. // GetRangeBig reads data of object payload range from shallow dir of BLOB storage.
// //
// Returns any error encountered that // Returns any error encountered that
@ -29,43 +17,41 @@ type GetRangeBigRes struct {
// //
// Returns ErrRangeOutOfBounds if the requested object range is out of bounds. // Returns ErrRangeOutOfBounds if the requested object range is out of bounds.
// Returns an error of type apistatus.ObjectNotFound if object is missing. // Returns an error of type apistatus.ObjectNotFound if object is missing.
func (b *BlobStor) GetRangeBig(prm GetRangeBigPrm) (GetRangeBigRes, error) { func (b *BlobStor) GetRangeBig(prm common.GetRangePrm) (common.GetRangeRes, error) {
// get compressed object data // get compressed object data
data, err := b.fsTree.Get(common.GetPrm{Address: prm.addr}) data, err := b.fsTree.Get(common.GetPrm{Address: prm.Address})
if err != nil { if err != nil {
if errors.Is(err, fstree.ErrFileNotFound) { if errors.Is(err, fstree.ErrFileNotFound) {
var errNotFound apistatus.ObjectNotFound var errNotFound apistatus.ObjectNotFound
return GetRangeBigRes{}, errNotFound return common.GetRangeRes{}, errNotFound
} }
return GetRangeBigRes{}, fmt.Errorf("could not read object from fs tree: %w", err) return common.GetRangeRes{}, fmt.Errorf("could not read object from fs tree: %w", err)
} }
data, err = b.Decompress(data) data, err = b.Decompress(data)
if err != nil { if err != nil {
return GetRangeBigRes{}, fmt.Errorf("could not decompress object data: %w", err) return common.GetRangeRes{}, fmt.Errorf("could not decompress object data: %w", err)
} }
// unmarshal the object // unmarshal the object
obj := objectSDK.New() obj := objectSDK.New()
if err := obj.Unmarshal(data); err != nil { if err := obj.Unmarshal(data); err != nil {
return GetRangeBigRes{}, fmt.Errorf("could not unmarshal the object: %w", err) return common.GetRangeRes{}, fmt.Errorf("could not unmarshal the object: %w", err)
} }
payload := obj.Payload() payload := obj.Payload()
ln, off := prm.rng.GetLength(), prm.rng.GetOffset() ln, off := prm.Range.GetLength(), prm.Range.GetOffset()
if pLen := uint64(len(payload)); ln+off < off || pLen < off || pLen < ln+off { if pLen := uint64(len(payload)); ln+off < off || pLen < off || pLen < ln+off {
var errOutOfRange apistatus.ObjectOutOfRange var errOutOfRange apistatus.ObjectOutOfRange
return GetRangeBigRes{}, errOutOfRange return common.GetRangeRes{}, errOutOfRange
} }
return GetRangeBigRes{ return common.GetRangeRes{
rangeData: rangeData{ Data: payload[off : off+ln],
data: payload[off : off+ln],
},
}, nil }, nil
} }
@ -79,6 +65,6 @@ func (b *BlobStor) GetRangeBig(prm GetRangeBigPrm) (GetRangeBigRes, error) {
// //
// Returns ErrRangeOutOfBounds if the requested object range is out of bounds. // Returns ErrRangeOutOfBounds if the requested object range is out of bounds.
// Returns an error of type apistatus.ObjectNotFound if the requested object is missing in blobovnicza(s). // Returns an error of type apistatus.ObjectNotFound if the requested object is missing in blobovnicza(s).
func (b *BlobStor) GetRangeSmall(prm blobovniczatree.GetRangeSmallPrm) (blobovniczatree.GetRangeSmallRes, error) { func (b *BlobStor) GetRangeSmall(prm common.GetRangePrm) (common.GetRangeRes, error) {
return b.blobovniczas.GetRange(prm) return b.blobovniczas.GetRange(prm)
} }

View file

@ -3,7 +3,7 @@ package shard
import ( import (
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobovnicza" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobovnicza"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/blobovniczatree" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/writecache" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/writecache"
apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status" apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
"github.com/nspcc-dev/neofs-sdk-go/object" "github.com/nspcc-dev/neofs-sdk-go/object"
@ -69,14 +69,11 @@ func (r RngRes) HasMeta() bool {
func (s *Shard) GetRange(prm RngPrm) (RngRes, error) { func (s *Shard) GetRange(prm RngPrm) (RngRes, error) {
var big, small storFetcher var big, small storFetcher
rng := object.NewRange()
rng.SetOffset(prm.off)
rng.SetLength(prm.ln)
big = func(stor *blobstor.BlobStor, _ *blobovnicza.ID) (*object.Object, error) { big = func(stor *blobstor.BlobStor, _ *blobovnicza.ID) (*object.Object, error) {
var getRngBigPrm blobstor.GetRangeBigPrm var getRngBigPrm common.GetRangePrm
getRngBigPrm.SetAddress(prm.addr) getRngBigPrm.Address = prm.addr
getRngBigPrm.SetRange(rng) getRngBigPrm.Range.SetOffset(prm.off)
getRngBigPrm.Range.SetLength(prm.ln)
res, err := stor.GetRangeBig(getRngBigPrm) res, err := stor.GetRangeBig(getRngBigPrm)
if err != nil { if err != nil {
@ -84,16 +81,17 @@ func (s *Shard) GetRange(prm RngPrm) (RngRes, error) {
} }
obj := object.New() obj := object.New()
obj.SetPayload(res.RangeData()) obj.SetPayload(res.Data)
return obj, nil return obj, nil
} }
small = func(stor *blobstor.BlobStor, id *blobovnicza.ID) (*object.Object, error) { small = func(stor *blobstor.BlobStor, id *blobovnicza.ID) (*object.Object, error) {
var getRngSmallPrm blobovniczatree.GetRangeSmallPrm var getRngSmallPrm common.GetRangePrm
getRngSmallPrm.SetAddress(prm.addr) getRngSmallPrm.Address = prm.addr
getRngSmallPrm.SetRange(rng) getRngSmallPrm.Range.SetOffset(prm.off)
getRngSmallPrm.SetBlobovniczaID(id) getRngSmallPrm.Range.SetLength(prm.ln)
getRngSmallPrm.BlobovniczaID = id
res, err := stor.GetRangeSmall(getRngSmallPrm) res, err := stor.GetRangeSmall(getRngSmallPrm)
if err != nil { if err != nil {
@ -101,7 +99,7 @@ func (s *Shard) GetRange(prm RngPrm) (RngRes, error) {
} }
obj := object.New() obj := object.New()
obj.SetPayload(res.RangeData()) obj.SetPayload(res.Data)
return obj, nil return obj, nil
} }
@ -113,8 +111,8 @@ func (s *Shard) GetRange(prm RngPrm) (RngRes, error) {
} }
payload := res.Payload() payload := res.Payload()
from := rng.GetOffset() from := prm.off
to := from + rng.GetLength() to := from + prm.ln
if pLen := uint64(len(payload)); to < from || pLen < from || pLen < to { if pLen := uint64(len(payload)); to < from || pLen < from || pLen < to {
return nil, apistatus.ObjectOutOfRange{} return nil, apistatus.ObjectOutOfRange{}
} }