gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(blobnode): chunk migrate and disk repair support download data in batch

Browse Source 
with: #1000119221

Signed-off-by: JasonHu520 <huzongchao@oppo.com>
This commit is contained in:
JasonHu520 2025-05-14 20:21:13 +08:00 committed by slasher
parent 8037070a38
commit ab621790d8
13 changed files with 261 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
blobstore/api/blobnode/shard.go
View File

@ -49,6 +49,7 @@ type BidInfo struct {
Bid proto.BlobID `json:"bid"`
Size int64 `json:"size"`
Offset int64 `json:"offset"`
Crc uint32 `json:"crc"`
}
type ShardStatus uint8

2
blobstore/blobnode/client/blobnode_cli.go
View File

@ -29,6 +29,8 @@ import (
var defaultFirstStartBid = proto.BlobID(0)
const HeaderSize = 4
// IBlobNode define the interface of blobnode used for worker
type IBlobNode interface {
StatChunk(ctx context.Context, location proto.VunitLocation) (ci *ChunkInfo, err error)

11
blobstore/blobnode/task_runner_migrate.go
View File

@ -42,8 +42,10 @@ type MigrateWorker struct {
bolbNodeCli client.IBlobNode
benchmarkBids []*ShardInfoSimple
bidInfos map[proto.Vuid]*ReplicaBidsRet
downloadShardConcurrency int
forbiddenDirectDownload bool
enableBatchRead bool
}
// MigrateTaskEx migrate task execution machine
@ -51,6 +53,7 @@ type MigrateTaskEx struct {
taskInfo *proto.MigrateTask
downloadShardConcurrency int
enableBatchRead bool
blobNodeCli client.IBlobNode
}
@ -60,6 +63,7 @@ func NewMigrateWorker(task MigrateTaskEx) ITaskWorker {
t: task.taskInfo,
bolbNodeCli: task.blobNodeCli,
downloadShardConcurrency: task.downloadShardConcurrency,
enableBatchRead: task.enableBatchRead,
forbiddenDirectDownload: task.taskInfo.ForbiddenDirectDownload,
}
}
@ -90,13 +94,14 @@ func (w *MigrateWorker) GenTasklets(ctx context.Context) ([]Tasklet, *WorkError)
return nil, OtherError(ErrNotReadyForMigrate)
}
}
migBids, benchmarkBids, err := GenMigrateBids(ctx, w.bolbNodeCli, w.t.Sources, w.t.Destination, w.t.CodeMode, badIdxs)
migBids, benchmarkBids, bidInfos, err := GenMigrateBids(ctx, w.bolbNodeCli, w.t.Sources, w.t.Destination, w.t.CodeMode, badIdxs)
if err != nil {
span.Errorf("gen migrate bids failed: err[%v]", err)
return nil, err
}
w.benchmarkBids = benchmarkBids
w.bidInfos = bidInfos
span.Debugf("task info: taskType[%s], benchmarkBids size[%d], need migrate bids size[%d]", w.TaskType(), len(benchmarkBids), len(migBids))
return BidsSplit(ctx, migBids, workutils.TaskBufPool.GetMigrateBufSize())
@ -106,8 +111,10 @@ func (w *MigrateWorker) GenTasklets(ctx context.Context) ([]Tasklet, *WorkError)
func (w *MigrateWorker) ExecTasklet(ctx context.Context, tasklet Tasklet) *WorkError {
replicas := w.t.Sources
mode := w.t.CodeMode
shardRecover := NewShardRecover(replicas, mode, tasklet.bids, w.bolbNodeCli, w.downloadShardConcurrency, w.t.TaskType)
shardRecover := NewShardRecover(replicas, mode, tasklet.bids, w.bolbNodeCli, w.downloadShardConcurrency,
w.t.TaskType, w.enableBatchRead)
defer shardRecover.ReleaseBuf()
shardRecover.bidInfos = w.bidInfos
return MigrateBids(ctx,
shardRecover,

27
blobstore/blobnode/task_runner_migrate_test.go
View File

@ -16,8 +16,13 @@ package blobnode
import (
"context"
"encoding/binary"
"hash/crc32"
"io"
"testing"
"github.com/cubefs/cubefs/blobstore/blobnode/client"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/stretchr/testify/require"
@ -230,6 +235,28 @@ func TestMigrateExecTasklet(t *testing.T) {
require.Equal(t, crc, crcMap[shard.Bid])
}
}
{
shards, _ := getter.ListShards(context.Background(), replicas[badi])
var bidInfos []api.BidInfo
for _, shard := range shards {
bidInfos = append(bidInfos, api.BidInfo{Size: shard.Size, Offset: shard.Offset, Bid: shard.Bid, Crc: shard.Crc})
}
batchShards, err := getter.GetShards(context.Background(), replicas[badi], bidInfos, api.BackgroundIO)
require.NoError(t, err)
header := make([]byte, client.HeaderSize)
for _, shard := range shards {
crc := crc32.NewIEEE()
_, err = io.ReadFull(batchShards, header)
require.NoError(t, err)
code := binary.BigEndian.Uint32(header)
require.Equal(t, uint32(200), code)
reader := io.TeeReader(batchShards, crc)
_, err = io.CopyN(io.Discard, reader, shard.Size)
require.NoError(t, err)
require.Equal(t, crc.Sum32(), crcMap[shard.Bid])
}
batchShards.Close()
}
workutils.TaskBufPool = nil
_, err := w.GenTasklets(context.Background())
require.Error(t, err)

3
blobstore/blobnode/task_shard_repair.go
View File

@ -112,7 +112,8 @@ func (repairer *ShardRepairer) RepairShard(ctx context.Context, task *proto.Shar
span.Infof("start recover blob: bid[%d], badIdx[%+v]", task.Bid, task.BadIdxs)
bidInfos := []*ShardInfoSimple{{Bid: task.Bid, Size: shardSize}}
shardRecover := NewShardRecover(task.Sources, task.CodeMode, bidInfos, repairer.cli, 1, proto.TaskTypeShardRepair)
shardRecover := NewShardRecover(task.Sources, task.CodeMode, bidInfos, repairer.cli,
1, proto.TaskTypeShardRepair, false)
defer shardRecover.ReleaseBuf()
err = shardRecover.RecoverShards(ctx, task.BadIdxs, false)
if err != nil {

16
blobstore/blobnode/work_shard_getter.go
View File

@ -54,6 +54,12 @@ type ShardInfoWithCrc struct {
Crc32 uint32
}
type ShardInfoWithOffset struct {
Bid proto.BlobID
Size int64
Offset int64
}
// GetSingleVunitNormalBids returns single volume unit bids info
func GetSingleVunitNormalBids(ctx context.Context, cli client.IBlobNode, replica proto.VunitLocation) (bids []*ShardInfoWithCrc, err error) {
shards, err := cli.ListShards(ctx, replica)
@ -130,7 +136,7 @@ func MergeBids(replicasBids map[proto.Vuid]*ReplicaBidsRet) []*ShardInfoSimple {
// GetBenchmarkBids returns bench mark bids
func GetBenchmarkBids(ctx context.Context, cli client.IBlobNode, replicas Vunits,
mode codemode.CodeMode, badIdxs []uint8,
) (bids []*ShardInfoSimple, err error) {
) (bids []*ShardInfoSimple, bidInfos map[proto.Vuid]*ReplicaBidsRet, err error) {
span := trace.SpanFromContextSafe(ctx)
globalReplicas := replicas.IntactGlobalSet(mode, badIdxs)
@ -145,9 +151,9 @@ func GetBenchmarkBids(ctx context.Context, cli client.IBlobNode, replicas Vunits
if wellCnt < minWellReplicasCnt(mode) {
span.Errorf("well replicas cnt is not enough: wellCnt[%d], minWellReplicasCnt[%d]",
wellCnt, minWellReplicasCnt(mode))
return nil, ErrNotEnoughWellReplicaCnt
return nil, nil, ErrNotEnoughWellReplicaCnt
}
bidInfos = replicasBids
allBidsList := MergeBids(replicasBids)
benchMark := []*ShardInfoSimple{}
for _, bid := range allBidsList {
@ -200,10 +206,10 @@ func GetBenchmarkBids(ctx context.Context, cli client.IBlobNode, replicas Vunits
span.Errorf("unexpect when get benchmark bids: vid[%d], bid[%d], existCnt[%d], notExistCnt[%d], allowFailCnt[%d]",
replicas[0].Vuid.Vid(), bid.Bid, existStatus.ExistCnt(), notExistCnt, allowFailCnt(mode))
return nil, ErrUnexpected
return nil, nil, ErrUnexpected
}
return benchMark, nil
return benchMark, bidInfos, nil
}
// minWellReplicasCnt:It is the mini count of well replicas which can determine

10
blobstore/blobnode/work_shard_getter_test.go
View File

@ -161,17 +161,17 @@ func testGetBenchmarkBids(t *testing.T, mode codemode.CodeMode) {
bids := []proto.BlobID{1, 2, 3, 4, 5, 6, 7}
sizes := []int64{10, 1024, 1024, 1024, 1024, 1024, 1024}
getter := NewMockGetterWithBids(replicas, mode, bids, sizes)
benchmarkBids, err := GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
benchmarkBids, _, err := GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
require.NoError(t, err)
bidsEqual(t, benchmarkBids, bids, sizes)
getter.Delete(context.Background(), replicas[0].Vuid, 1)
benchmarkBids, err = GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
benchmarkBids, _, err = GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
require.NoError(t, err)
bidsEqual(t, benchmarkBids, bids, sizes)
getter.MarkDelete(context.Background(), replicas[1].Vuid, 1)
benchmarkBids, err = GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
benchmarkBids, _, err = GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
require.NoError(t, err)
bids1 := []proto.BlobID{2, 3, 4, 5, 6, 7}
sizes1 := []int64{1024, 1024, 1024, 1024, 1024, 1024}
@ -185,7 +185,7 @@ func testGetBenchmarkBids(t *testing.T, mode codemode.CodeMode) {
replica := replicas[idx]
getter.Delete(context.Background(), replica.Vuid, 2)
}
benchmarkBids, err = GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
benchmarkBids, _, err = GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
require.NoError(t, err)
bids2 := []proto.BlobID{3, 4, 5, 6, 7}
sizes2 := []int64{1024, 1024, 1024, 1024, 1024}
@ -206,7 +206,7 @@ func testGetBenchmarkBids(t *testing.T, mode codemode.CodeMode) {
getter.setFail(replica.Vuid, errors.New("fake error"))
}
_, err = GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
_, _, err = GetBenchmarkBids(context.Background(), getter, replicas, mode, []uint8{})
require.Error(t, err)
require.EqualError(t, ErrNotEnoughWellReplicaCnt, err.Error())
}

135
blobstore/blobnode/work_shard_recover.go
View File

@ -16,10 +16,12 @@ package blobnode
import (
"context"
"encoding/binary"
"errors"
"hash/crc32"
"io"
"math/rand"
"sort"
"sync"
"unsafe"
@ -31,6 +33,7 @@ import (
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/util"
"github.com/cubefs/cubefs/blobstore/util/taskpool"
)
@ -310,10 +313,11 @@ func (shards *ShardsBuf) PutShard(bid proto.BlobID, input io.Reader) error {
shards.mu.Unlock()
return errShardSizeNotMatch
}
data := shards.shards[bid].data
shards.mu.Unlock()
// read data from remote is slow,so optimize use of lock
_, err := io.ReadFull(input, shards.shards[bid].data)
_, err := io.ReadFull(input, data)
if err != nil {
return err
}
@ -394,11 +398,15 @@ type ShardRecover struct {
ioType blobnode.IOType
taskType proto.TaskType
ds *downloadStatus
enableBatchRead bool
bidInfos map[proto.Vuid]*ReplicaBidsRet
}
// NewShardRecover returns shard recover
func NewShardRecover(replicas Vunits, mode codemode.CodeMode, bidInfos []*ShardInfoSimple,
shardGetter client.IBlobNode, vunitShardGetConcurrency int, taskType proto.TaskType,
enableBatchRead bool,
) *ShardRecover {
if vunitShardGetConcurrency <= 0 {
vunitShardGetConcurrency = defaultGetConcurrency
@ -415,6 +423,7 @@ func NewShardRecover(replicas Vunits, mode codemode.CodeMode, bidInfos []*ShardI
taskType: taskType,
vunitShardGetConcurrency: vunitShardGetConcurrency,
ds: newDownloadStatus(),
enableBatchRead: enableBatchRead,
}
return &repair
}
@ -599,11 +608,19 @@ func (r *ShardRecover) download(ctx context.Context, repairBids []proto.BlobID,
rep := replica
tp.Run(func() {
defer wg.Done()
if r.enableBatchRead && r.bidInfos[replica.Vuid].RetErr == nil {
r.batchDownloadReplShards(ctxTmp, rep, repairBids)
return
}
r.downloadReplShards(ctxTmp, rep, repairBids)
})
}
wg.Wait()
tp.Close()
// if need download failed bids, forbidden BatchRead
if r.enableBatchRead {
r.enableBatchRead = false
}
}
func (r *ShardRecover) downloadReplShards(ctx context.Context, replica proto.VunitLocation, repairBids []proto.BlobID) {
@ -643,6 +660,71 @@ func (r *ShardRecover) downloadReplShards(ctx context.Context, replica proto.Vun
span.Infof("finish downloadSingle: vuid[%d], idx[%d]", vuid, vuid.Index())
}
func (r *ShardRecover) batchDownloadReplShards(ctx context.Context, replica proto.VunitLocation, repairBids []proto.BlobID) {
span := trace.SpanFromContextSafe(ctx)
vuid := replica.Vuid
span.Infof("start batch downloadSingle: repl idx[%d], len bids[%d]", vuid.Index(), len(repairBids))
var bidInfos []blobnode.BidInfo
for _, bid := range repairBids {
info, ok := r.bidInfos[replica.Vuid].Bids[bid]
if !ok || info.Inline {
continue
}
if info.NopData {
err := r.putShardToBuffer(ctx, replica, info.Bid, io.NopCloser(util.ZeroReader(int(info.Size))), info.Crc)
if err != nil {
span.Errorf("put shard to buffer failed, err: %s", err)
}
continue
}
bidInfos = append(bidInfos, blobnode.BidInfo{Bid: info.Bid, Size: info.Size, Offset: info.Offset, Crc: info.Crc})
}
if len(bidInfos) == 0 {
return
}
sort.Slice(bidInfos, func(i, j int) bool {
return bidInfos[i].Offset < bidInfos[j].Offset
})
r.batchDownloadShard(ctx, replica, bidInfos)
}
func (r *ShardRecover) batchDownloadShard(ctx context.Context, replica proto.VunitLocation, bidinfos []blobnode.BidInfo) {
span := trace.SpanFromContextSafe(ctx)
body, err := r.shardGetter.GetShards(ctx, replica, bidinfos, r.ioType)
if err != nil {
span.Errorf("batch download shard failed, err: %s", err)
return
}
defer body.Close()
header := make([]byte, client.HeaderSize)
for _, info := range bidinfos {
bid := info.Bid
_, err = io.ReadFull(body, header)
if err != nil {
span.Errorf("read headers failed, err: %s", err)
return
}
code := binary.BigEndian.Uint32(header)
if code != uint32(200) {
span.Errorf("download shard failed, errCode: %s", code)
return
}
err = r.putShardToBuffer(ctx, replica, bid, body, info.Crc)
if err != nil {
span.Errorf("put shard to buf failed: replica[%+v], bid[%d], err[%+v]", replica, bid, err)
return
}
select {
case <-ctx.Done():
span.Infof("download cancel: replica[%+v], bids[%d]", replica, bidinfos)
return
default:
}
}
span.Infof("finish batch download: vuid[%d], idx[%d]", replica.Vuid, replica.Vuid.Index())
}
func (r *ShardRecover) downloadShard(ctx context.Context, replica proto.VunitLocation, bid proto.BlobID) error {
span := trace.SpanFromContextSafe(ctx)
@ -657,34 +739,45 @@ func (r *ShardRecover) downloadShard(ctx context.Context, replica proto.VunitLoc
span.Errorf("download failed: replica[%+v], bid[%d], err[%+v]", replica, bid, err)
return err
}
err = r.chunksShardsBuf[replica.Vuid.Index()].PutShard(bid, data)
data.Close()
if err == errBidNotFoundInBuf {
span.Errorf("unexpect put shard failed: err[%+v]", err)
return err
}
if err == errBufHasData {
bufCrc, _ := r.chunksShardsBuf[replica.Vuid.Index()].ShardCrc32(bid)
if bufCrc != crc1 {
span.Errorf("data conflict crc32 not match: bid[%d], bufCrc[%d], crc1[%d]", bid, bufCrc, crc1)
return errCrcNotMatch
}
return nil
}
defer data.Close()
err = r.putShardToBuffer(ctx, replica, bid, data, crc1)
if err != nil {
span.Errorf("blob put shard to buf failed: replica[%+v], bid[%d], err[%+v]", replica, bid, err)
span.Errorf("put shard to buf failed: replica[%+v], bid[%d], err[%+v]", replica, bid, err)
return err
}
}
return nil
}
crc2, _ := r.chunksShardsBuf[replica.Vuid.Index()].ShardCrc32(bid)
if crc1 != crc2 {
span.Errorf("shard crc32 not match: replica[%+v], bid[%d], crc1[%d], crc2[%d]", replica, bid, crc1, crc2)
func (r *ShardRecover) putShardToBuffer(ctx context.Context, replica proto.VunitLocation,
bid proto.BlobID, data io.ReadCloser, crc1 uint32,
) error {
span := trace.SpanFromContextSafe(ctx)
err := r.chunksShardsBuf[replica.Vuid.Index()].PutShard(bid, data)
if err == errBidNotFoundInBuf {
span.Errorf("unexpect put shard failed: err[%+v]", err)
return err
}
if err == errBufHasData {
bufCrc, _ := r.chunksShardsBuf[replica.Vuid.Index()].ShardCrc32(bid)
if bufCrc != crc1 {
span.Errorf("data conflict crc32 not match: bid[%d], bufCrc[%d], crc1[%d]", bid, bufCrc, crc1)
return errCrcNotMatch
}
return nil
}
if err != nil {
span.Errorf("blob put shard to buf failed: replica[%+v], bid[%d], err[%+v]", replica, bid, err)
return err
}
crc2, _ := r.chunksShardsBuf[replica.Vuid.Index()].ShardCrc32(bid)
if crc1 != crc2 {
span.Errorf("shard crc32 not match: replica[%+v], bid[%d], crc1[%d], crc2[%d]", replica, bid, crc1, crc2)
return errCrcNotMatch
}
return nil
}
func (r *ShardRecover) repair(ctx context.Context, repairBids []proto.BlobID, stripe repairStripe) error {

65
blobstore/blobnode/work_shard_recover_test.go
View File

@ -20,6 +20,8 @@ import (
"hash/crc32"
"testing"
"github.com/cubefs/cubefs/blobstore/blobnode/client"
"github.com/stretchr/testify/require"
"github.com/cubefs/cubefs/blobstore/blobnode/base/workutils"
@ -106,7 +108,7 @@ func InitMockRepair(mode codemode.CodeMode) (*ShardRecover, []*ShardInfoSimple,
bidInfos = append(bidInfos, &ele)
}
repair := NewShardRecover(replicas, mode, bidInfos, getter, 3, proto.TaskTypeShardRepair)
repair := NewShardRecover(replicas, mode, bidInfos, getter, 3, proto.TaskTypeShardRepair, false)
return repair, bidInfos, getter, replicas
}
@ -321,7 +323,7 @@ func TestRecoverShards2(t *testing.T) {
func TestLocalStripes(t *testing.T) {
mode := codemode.EC6P10L2
replicas := genMockVol(1, mode)
repair := NewShardRecover(replicas, mode, nil, nil, 4, proto.TaskTypeShardRepair)
repair := NewShardRecover(replicas, mode, nil, nil, 4, proto.TaskTypeShardRepair, false)
for idx, replica := range replicas {
require.Equal(t, idx, int(replica.Vuid.Index()))
}
@ -403,6 +405,65 @@ func TestDownload(t *testing.T) {
}
}
func TestDownloadBatch(t *testing.T) {
ctx := context.Background()
repair, _, getter, replicas := InitMockRepair(codemode.EC6P6)
repairBids := []proto.BlobID{1, 2, 4, 5, 6, 7}
repair.enableBatchRead = true
idxs := replicas.Indexes()
err := repair.allocBuf(ctx, idxs)
require.NoError(t, err)
failVuids := []proto.Vuid{replicas[0].Vuid, replicas[1].Vuid}
for _, fail := range failVuids {
getter.setFail(fail, errors.New("fake error"))
}
bidInfos := make(map[proto.Vuid]*ReplicaBidsRet)
for _, repl := range replicas {
shardInfos, err := getter.ListShards(context.Background(), repl)
if err != nil {
bidInfos[repl.Vuid] = &ReplicaBidsRet{RetErr: err, Bids: nil}
continue
}
si := make(map[proto.BlobID]*client.ShardInfo)
for _, info := range shardInfos {
si[info.Bid] = info
}
bidInfos[repl.Vuid] = &ReplicaBidsRet{RetErr: nil, Bids: si}
}
repair.bidInfos = bidInfos
repair.download(ctx, repairBids, replicas)
for _, fail := range failVuids {
for _, bid := range repairBids {
_, err := repair.GetShard(fail.Index(), bid)
require.Error(t, err)
require.EqualError(t, errShardDataNotPrepared, err.Error())
}
}
var well []proto.Vuid
for _, repl := range replicas {
isFail := false
for _, fail := range failVuids {
if fail == repl.Vuid {
isFail = true
break
}
}
if !isFail {
well = append(well, repl.Vuid)
}
}
for _, repl := range well {
for _, bid := range repairBids {
_, err := repair.GetShard(repl.Index(), bid)
require.NoError(t, err)
}
}
}
func TestDirect(t *testing.T) {
ctx := context.Background()
repair, _, getter, _ := InitMockRepair(codemode.EC6P6)

10
blobstore/blobnode/work_shards_migrate.go
View File

@ -36,13 +36,13 @@ var (
// GenMigrateBids generates migrate blob ids
func GenMigrateBids(ctx context.Context, blobnodeCli client.IBlobNode, srcReplicas Vunits,
dst proto.VunitLocation, mode codemode.CodeMode, badIdxs []uint8,
) (migBids, benchmarkBids []*ShardInfoSimple, wErr *WorkError) {
) (migBids, benchmarkBids []*ShardInfoSimple, bidInfos map[proto.Vuid]*ReplicaBidsRet, wErr *WorkError) {
span := trace.SpanFromContextSafe(ctx)
benchmarkBids, err := GetBenchmarkBids(ctx, blobnodeCli, srcReplicas, mode, badIdxs)
benchmarkBids, bidInfos, err := GetBenchmarkBids(ctx, blobnodeCli, srcReplicas, mode, badIdxs)
if err != nil {
span.Errorf("get benchmark bids failed: err[%v]", err)
return nil, nil, SrcError(err)
return nil, nil, nil, SrcError(err)
}
span.Infof("get benchmark success: len[%d]", len(benchmarkBids))
@ -51,7 +51,7 @@ func GenMigrateBids(ctx context.Context, blobnodeCli client.IBlobNode, srcReplic
destBids, err := GetSingleVunitNormalBids(ctx, blobnodeCli, dst)
if err != nil {
span.Errorf("get single vunit normal bids failed: dst[%+v], err[%+v]", dst, err)
return nil, nil, DstError(err)
return nil, nil, nil, DstError(err)
}
span.Infof("GetSingleVunitNormalBids success: destBids len[%d], idx[%d]", len(destBids), dst.Vuid.Index())
@ -70,7 +70,7 @@ func GenMigrateBids(ctx context.Context, blobnodeCli client.IBlobNode, srcReplic
}
span.Infof("benchmarkBids: len[%d]", len(migBids))
return migBids, benchmarkBids, nil
return migBids, benchmarkBids, bidInfos, nil
}
// MigrateBids migrate the bids data to destination

14
blobstore/blobnode/worker_for_test.go
View File

@ -17,6 +17,7 @@ package blobnode
import (
"bytes"
"context"
"encoding/binary"
"hash/crc32"
"io"
"sync"
@ -206,6 +207,19 @@ func (getter *MockGetter) GetShard(ctx context.Context, location proto.VunitLoca
return io.NopCloser(reader), crc, err
}
func (getter *MockGetter) GetShards(ctx context.Context, location proto.VunitLocation, bids []api.BidInfo, ioType api.IOType) (body io.ReadCloser, err error) {
getter.mu.Lock()
defer getter.mu.Unlock()
buf := make([]byte, 0)
header := make([]byte, client.HeaderSize)
for _, bid := range bids {
binary.BigEndian.PutUint32(header, 200)
buf = append(buf, header...)
buf = append(buf, getter.vunits[location.Vuid].shards[bid.Bid]...)
}
return io.NopCloser(bytes.NewReader(buf)), nil
}
func (getter *MockGetter) MarkDelete(ctx context.Context, vuid proto.Vuid, bid proto.BlobID) {
getter.mu.Lock()
defer getter.mu.Unlock()

6
blobstore/blobnode/worker_service.go
View File

@ -85,8 +85,9 @@ type WorkerConfig struct {
// scheduler client config
Scheduler scheduler.Config `json:"scheduler"`
// blbonode client config
BlobNode bnapi.Config `json:"blobnode"`
ShardNode shardnode.Config `json:"shardnode"`
BlobNode bnapi.Config `json:"blobnode"`
ShardNode shardnode.Config `json:"shardnode"`
EnableBatchRead bool `json:"enable_batch_read"`
DroppedBidRecord *recordlog.Config `json:"dropped_bid_record"`
}
@ -305,6 +306,7 @@ func (s *WorkerService) addBlobNodeTask(ctx context.Context, t *proto.Task) {
taskInfo: task,
downloadShardConcurrency: s.DownloadShardConcurrency,
blobNodeCli: s.blobNodeCli,
enableBatchRead: s.EnableBatchRead,
}); err != nil {
span.Errorf("add task failed: taskID[%s], err[%v]", task.TaskID, err)
return

4
blobstore/blobnode/worker_service_test.go
View File

@ -45,6 +45,10 @@ func getDefaultConfig() WorkerConfig {
type mBlobNodeCli struct{}
func (m *mBlobNodeCli) GetShards(ctx context.Context, location proto.VunitLocation, bids []bnapi.BidInfo, ioType bnapi.IOType) (body io.ReadCloser, err error) {
return
}
func (m *mBlobNodeCli) StatChunk(ctx context.Context, location proto.VunitLocation) (ci *client.ChunkInfo, err error) {
return
}