dkirillov/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#235] object/getrange: Fix incorrect processing of range from last child

Browse source 
Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
Leonard Lyubich 2020-12-08 18:02:07 +03:00 committed by Alex Vanin
parent 397d912e19
commit 0fc5ea674c
2 changed files with 30 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
pkg/services/object/get/assemble.go
View file

@ -45,7 +45,7 @@ func (exec *execCtx) assemble() {
// * else go right-to-left with GET and compose in single object before writing // * else go right-to-left with GET and compose in single object before writing
if ok := exec.overtakePayloadInReverse(prev); ok { if ok := exec.overtakePayloadInReverse(prev); ok {
// payload of all children except the last are written, write last payload // payload of all children except the last are written, write last payloa
exec.writeObjectPayload(exec.collectedObject) exec.writeObjectPayload(exec.collectedObject)
} }
} }
@ -61,7 +61,6 @@ func (exec *execCtx) initFromChild(id *objectSDK.ID) (prev *objectSDK.ID, childr
child, ok := exec.getChild(id, nil) child, ok := exec.getChild(id, nil)
if !ok { if !ok {
return return
} }
@ -100,13 +99,20 @@ func (exec *execCtx) initFromChild(id *objectSDK.ID) (prev *objectSDK.ID, childr
childSize := child.PayloadSize() childSize := child.PayloadSize()
if to := seekOff + seekLen; childSize > 0 && to > parSize-childSize { exec.curOff = parSize - childSize
pref := to + childSize - parSize
payload = child.Payload()[:pref] from := uint64(0)
rng.SetLength(rng.GetLength() - pref) if exec.curOff < seekOff {
from = seekOff - exec.curOff
} }
exec.curOff = parSize - childSize to := uint64(0)
if seekOff+seekLen > exec.curOff+from {
to = seekOff + seekLen - exec.curOff
}
payload = child.Payload()[from:to]
rng.SetLength(rng.GetLength() - to + from)
} else { } else {
payload = child.Payload() payload = child.Payload()
} }
@ -140,8 +146,8 @@ func (exec *execCtx) overtakePayloadDirectly(children []*objectSDK.ID, rngs []*o
} }
func (exec *execCtx) overtakePayloadInReverse(prev *objectSDK.ID) bool { func (exec *execCtx) overtakePayloadInReverse(prev *objectSDK.ID) bool {
chain, rngs := exec.buildChainInReverse(prev) chain, rngs, ok := exec.buildChainInReverse(prev)
if len(chain) == 0 { if !ok {
return false return false
} }
@ -164,7 +170,7 @@ func (exec *execCtx) overtakePayloadInReverse(prev *objectSDK.ID) bool {
return true return true
} }
func (exec *execCtx) buildChainInReverse(prev *objectSDK.ID) ([]*objectSDK.ID, []*objectSDK.Range) { func (exec *execCtx) buildChainInReverse(prev *objectSDK.ID) ([]*objectSDK.ID, []*objectSDK.Range, bool) {
var ( var (
chain = make([]*objectSDK.ID, 0) chain = make([]*objectSDK.ID, 0)
rngs = make([]*objectSDK.Range, 0) rngs = make([]*objectSDK.Range, 0)
@ -181,7 +187,7 @@ func (exec *execCtx) buildChainInReverse(prev *objectSDK.ID) ([]*objectSDK.ID, [
head, ok := exec.headChild(prev) head, ok := exec.headChild(prev)
if !ok { if !ok {
return nil, nil return nil, nil, false
} }
if seekRng != nil { if seekRng != nil {
@ -189,7 +195,7 @@ func (exec *execCtx) buildChainInReverse(prev *objectSDK.ID) ([]*objectSDK.ID, [
exec.curOff -= sz exec.curOff -= sz
if exec.curOff < from+to { if exec.curOff < to {
off := uint64(0) off := uint64(0)
if from > exec.curOff { if from > exec.curOff {
off = from - exec.curOff off = from - exec.curOff
@ -214,7 +220,7 @@ func (exec *execCtx) buildChainInReverse(prev *objectSDK.ID) ([]*objectSDK.ID, [
prev = head.PreviousID() prev = head.PreviousID()
} }
return chain, rngs return chain, rngs, true
} }
func equalAddresses(a, b *objectSDK.Address) bool { func equalAddresses(a, b *objectSDK.Address) bool {

11
pkg/services/object/get/get_test.go
View file

@ -1010,6 +1010,17 @@ func TestGetRemoteSmall(t *testing.T) {
err = svc.GetRange(ctx, rngPrm) err = svc.GetRange(ctx, rngPrm)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, payload[off:off+ln], w.object().Payload()) require.Equal(t, payload[off:off+ln], w.object().Payload())
w = newSimpleObjectWriter()
off = payloadSz - 2
ln = 1
rngPrm = newRngPrm(false, w, off, ln)
rngPrm.WithAddress(addr)
err = svc.GetRange(ctx, rngPrm)
require.NoError(t, err)
require.Equal(t, payload[off:off+ln], w.object().Payload())
}) })
}) })
}) })