mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
fix(bssdk): fix location size after alloc slices
with: #23053900 Signed-off-by: mawei029 <mawei2@oppo.com>
This commit is contained in:
mawei029
slasher
parent
f7472eeb2d
commit
4b75603a06
@ -988,6 +988,7 @@ func (s *sdkHandler) putBlobs(ctx context.Context, args *acapi.PutBlobArgs) (pro
|
||||
failLoc := proto.Location{ClusterID: created.Location.ClusterID}
|
||||
loc, err := fixLocationSize(created.Location, args.Size)
|
||||
if err != nil {
|
||||
span.Errorf("fail to fix location size, loc:%+v, err:%+v", loc, err)
|
||||
return failLoc, nil, err
|
||||
}
|
||||
|
||||
@ -1030,13 +1031,21 @@ func (s *sdkHandler) putBlobs(ctx context.Context, args *acapi.PutBlobArgs) (pro
|
||||
return failLoc, nil, err
|
||||
}
|
||||
|
||||
span.Debugf("retry:%d, location:%+v, blobIdx:%d, sliceIdx:%d, remainSize:%d", retryCnt, loc, blobIdx, sliceIdx, remainSize)
|
||||
allocs, err1 := s.retryAllocSlice(ctx, args, loc, blobIdx, sliceIdx, remainSize)
|
||||
if err1 != nil {
|
||||
return failLoc, nil, err1
|
||||
}
|
||||
loc.Slices, blobIdx = s.updateLocationSlices(loc.Slices, allocs, blobIdx, sliceIdx, remainSize)
|
||||
// we should fix location size, after create/alloc
|
||||
allocLoc, err1 := fixLocationSize(proto.Location{Slices: allocs, SliceSize: loc.SliceSize}, remainSize)
|
||||
if err1 != nil {
|
||||
return failLoc, nil, err1
|
||||
}
|
||||
|
||||
loc.Slices, blobIdx = s.updateLocationSlices(loc.Slices, allocLoc.Slices, blobIdx, sliceIdx, remainSize)
|
||||
needRead = false
|
||||
buf = buf1 // prevent repeated io read
|
||||
span.Debugf("update location slices:%+v, blobIdx:%d", loc.Slices, blobIdx)
|
||||
}
|
||||
|
||||
for alg, hasher := range hasherMap {
|
||||
@ -1048,12 +1057,14 @@ func (s *sdkHandler) putBlobs(ctx context.Context, args *acapi.PutBlobArgs) (pro
|
||||
func (s *sdkHandler) putOneSlice(ctx context.Context, args *acapi.PutBlobArgs, loc proto.Location,
|
||||
blobIdx int, needRead bool, buf []byte,
|
||||
) ([]byte, int, uint64, error) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
slice := loc.Slices[blobIdx]
|
||||
var err error
|
||||
|
||||
// todo: dont concurrency, support concurrency next version
|
||||
for cnt, remainSize := 0, slice.ValidSize; uint32(cnt) < slice.Count; {
|
||||
if remainSize <= 0 {
|
||||
span.Errorf("err location size, blobIdx:%d, loc:%+v, err:%+v", blobIdx, loc, err)
|
||||
return []byte{}, 0, 0, errcode.ErrIllegalLocationSize
|
||||
}
|
||||
|
||||
@ -1210,7 +1221,7 @@ func fixLocationSize(loc proto.Location, size uint64) (proto.Location, error) {
|
||||
if remainSize > uint64(loc.Slices[i].Count*loc.SliceSize) {
|
||||
return proto.Location{}, errcode.ErrIllegalLocationSize
|
||||
}
|
||||
loc.Slices[i].ValidSize = remainSize
|
||||
loc.Slices[i].ValidSize = remainSize // ValidSize is 0, so set it
|
||||
break
|
||||
}
|
||||
|
||||
@ -1218,7 +1229,7 @@ func fixLocationSize(loc proto.Location, size uint64) (proto.Location, error) {
|
||||
if validSize >= remainSize {
|
||||
return proto.Location{}, errcode.ErrIllegalLocationSize
|
||||
}
|
||||
loc.Slices[i].ValidSize = validSize
|
||||
loc.Slices[i].ValidSize = validSize // ValidSize is 0, so set it
|
||||
remainSize -= validSize
|
||||
}
|
||||
|
||||
|
||||
@ -618,14 +618,14 @@ func TestSdkBlob_Put(t *testing.T) {
|
||||
args.Size = uint64(len(data))
|
||||
loca := &proto.Location{
|
||||
ClusterID: 1,
|
||||
Size_: uint64(len(data)),
|
||||
// Size_: uint64(len(data)),
|
||||
SliceSize: 4,
|
||||
Slices: []proto.Slice{
|
||||
{
|
||||
MinSliceID: 1,
|
||||
Vid: 10,
|
||||
Count: 3,
|
||||
ValidSize: uint64(len(data)),
|
||||
// ValidSize: uint64(len(data)),
|
||||
},
|
||||
},
|
||||
}
|
||||
@ -670,14 +670,14 @@ func TestSdkBlob_Put(t *testing.T) {
|
||||
args.NeedSeal = true
|
||||
loc2 := &proto.Location{
|
||||
ClusterID: 1,
|
||||
Size_: uint64(len(data)),
|
||||
// Size_: uint64(len(data)),
|
||||
SliceSize: uint32(len(data)),
|
||||
Slices: []proto.Slice{
|
||||
{
|
||||
MinSliceID: 1,
|
||||
Vid: 10,
|
||||
Count: 1,
|
||||
ValidSize: uint64(len(data)),
|
||||
// ValidSize: uint64(len(data)),
|
||||
},
|
||||
},
|
||||
}
|
||||
@ -690,15 +690,43 @@ func TestSdkBlob_Put(t *testing.T) {
|
||||
require.Equal(t, proto.ClusterID(1), cid)
|
||||
require.Equal(t, 0, len(hashes))
|
||||
|
||||
// create, fix location size fail
|
||||
loc2.SliceSize = 1
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().CreateBlob(gAny, gAny).Return(loc2, nil)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().DeleteBlob(gAny, gAny).Return(nil)
|
||||
cid, _, err = hd.PutBlob(ctx, args)
|
||||
require.NotNil(t, err)
|
||||
require.ErrorIs(t, err, errcode.ErrIllegalLocationSize)
|
||||
require.Equal(t, proto.ClusterID(1), cid)
|
||||
|
||||
// alloc fix location size fail
|
||||
hd.conf.MaxRetry = 2
|
||||
loc2.SliceSize = 9
|
||||
args.Body = bytes.NewBuffer([]byte(data))
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().CreateBlob(gAny, gAny).Return(loc2, nil)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().PutAt(gAny, gAny, gAny, gAny, gAny, gAny, gAny).Return(errMock).Times(1)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().Delete(gAny, gAny).Return(nil).Times(1)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().AllocSlice(gAny, gAny).Return(shardnode.AllocSliceRet{
|
||||
Slices: []proto.Slice{{
|
||||
MinSliceID: 2, Vid: 1, Count: 0,
|
||||
}},
|
||||
}, nil)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().DeleteBlob(gAny, gAny).Return(nil)
|
||||
cid, _, err = hd.PutBlob(ctx, args)
|
||||
require.NotNil(t, err)
|
||||
require.ErrorIs(t, err, errcode.ErrIllegalLocationSize)
|
||||
require.Equal(t, proto.ClusterID(1), cid)
|
||||
|
||||
// put fail, max retry, don't need to seal
|
||||
hd.conf.MaxRetry = 2
|
||||
loc2.SliceSize = 9
|
||||
args.Body = bytes.NewBuffer([]byte(data))
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().CreateBlob(gAny, gAny).Return(loc2, nil)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().PutAt(gAny, gAny, gAny, gAny, gAny, gAny, gAny).Return(errMock).Times(2)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().Delete(gAny, gAny).Return(nil).Times(2)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().AllocSlice(gAny, gAny).Return(shardnode.AllocSliceRet{
|
||||
Slices: []proto.Slice{{
|
||||
Vid: 1, Count: 1, ValidSize: uint64(len(data)),
|
||||
MinSliceID: 2, Vid: 1, Count: 1, // ValidSize: uint64(len(data)),
|
||||
}},
|
||||
}, nil)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().DeleteBlob(gAny, gAny).Return(nil)
|
||||
@ -742,7 +770,7 @@ func TestSdkBlob_Put(t *testing.T) {
|
||||
MinSliceID: 4, // 2,3 -> 4,5
|
||||
Vid: 3,
|
||||
Count: 2,
|
||||
ValidSize: 4,
|
||||
// ValidSize: 4,
|
||||
}},
|
||||
}, nil)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().PutAt(gAny, gAny, gAny, gAny, gAny, gAny, gAny).DoAndReturn(
|
||||
@ -779,7 +807,7 @@ func TestSdkBlob_Put(t *testing.T) {
|
||||
MinSliceID: 4, // 1,2,3 -> 4,5,6
|
||||
Vid: 3,
|
||||
Count: 3,
|
||||
ValidSize: 6,
|
||||
// ValidSize: 6,
|
||||
}},
|
||||
}, nil)
|
||||
wt = bytes.NewBuffer(nil)
|
||||
@ -822,7 +850,7 @@ func TestSdkBlob_Put(t *testing.T) {
|
||||
MinSliceID: 12, // 11 -> 12
|
||||
Vid: 3,
|
||||
Count: 1,
|
||||
ValidSize: 1,
|
||||
// ValidSize: 1,
|
||||
}},
|
||||
}, nil)
|
||||
hd.handler.(*mocks.MockStreamHandler).EXPECT().PutAt(gAny, gAny, gAny, gAny, gAny, gAny, gAny).DoAndReturn(
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
