mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
714 lines
20 KiB
Go
714 lines
20 KiB
Go
package datanode
|
|
|
|
import (
|
|
"bytes"
|
|
"fmt"
|
|
"hash/crc32"
|
|
"net"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/cubefs/cubefs/datanode/repl"
|
|
"github.com/cubefs/cubefs/datanode/storage"
|
|
"github.com/cubefs/cubefs/proto"
|
|
"github.com/cubefs/cubefs/util"
|
|
"github.com/cubefs/cubefs/util/exporter"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
type repairWorker struct {
|
|
proto.Packet
|
|
// followersAddrs []string
|
|
// followerPackets []*repl.FollowerPacket
|
|
IsReleased int32 // TODO what is released?
|
|
Object interface{}
|
|
TpObject *exporter.TimePointCount
|
|
NeedReply bool
|
|
OrgBuffer []byte
|
|
|
|
// used locally
|
|
shallDegrade bool
|
|
AfterPre bool
|
|
packChannel chan repl.PacketInterface
|
|
dp *DataPartition
|
|
dstWorker *repairWorker
|
|
role string
|
|
}
|
|
|
|
func (p *repairWorker) IsErrPacket() bool {
|
|
return p.ResultCode != proto.OpOk && p.ResultCode != proto.OpInitResultCode
|
|
}
|
|
|
|
func (p *repairWorker) SetArglen(len uint32) {
|
|
p.ArgLen = len
|
|
}
|
|
|
|
func (p *repairWorker) SetArg(data []byte) {
|
|
p.Arg = data
|
|
}
|
|
|
|
func (p *repairWorker) GetCRC() uint32 {
|
|
return p.CRC
|
|
}
|
|
|
|
func (p *repairWorker) GetStartT() int64 {
|
|
return p.StartT
|
|
}
|
|
|
|
func (p *repairWorker) GetOpcode() uint8 {
|
|
return p.Opcode
|
|
}
|
|
|
|
func (p *repairWorker) GetUniqueLogId() (m string) {
|
|
return ""
|
|
}
|
|
|
|
func (p *repairWorker) GetReqID() int64 {
|
|
return p.ReqID
|
|
}
|
|
|
|
func (p *repairWorker) GetPartitionID() uint64 {
|
|
return p.PartitionID
|
|
}
|
|
|
|
func (p *repairWorker) GetExtentID() uint64 {
|
|
return p.ExtentID
|
|
}
|
|
|
|
func (p *repairWorker) GetSize() uint32 {
|
|
return p.Size
|
|
}
|
|
|
|
func (p *repairWorker) SetSize(size uint32) {
|
|
p.Size = size
|
|
}
|
|
|
|
func (p *repairWorker) GetArg() []byte {
|
|
return p.Arg
|
|
}
|
|
|
|
func (p *repairWorker) GetArgLen() uint32 {
|
|
return p.ArgLen
|
|
}
|
|
|
|
func (p *repairWorker) GetData() []byte {
|
|
return p.Data
|
|
}
|
|
|
|
func (p *repairWorker) GetResultCode() uint8 {
|
|
return p.ResultCode
|
|
}
|
|
|
|
func (p *repairWorker) GetExtentOffset() int64 {
|
|
return p.ExtentOffset
|
|
}
|
|
|
|
func (p *repairWorker) SetResultCode(code uint8) {
|
|
p.ResultCode = code
|
|
}
|
|
|
|
func (p *repairWorker) SetCRC(crc uint32) {
|
|
p.CRC = crc
|
|
}
|
|
|
|
func (p *repairWorker) SetExtentOffset(offset int64) {
|
|
p.ExtentOffset = offset
|
|
}
|
|
|
|
func (p *repairWorker) ShallDegrade() bool {
|
|
return p.shallDegrade
|
|
}
|
|
|
|
func (p *repairWorker) SetStartT(StartT int64) {
|
|
p.StartT = StartT
|
|
}
|
|
|
|
func (p *repairWorker) SetData(data []byte) {
|
|
p.Data = data
|
|
}
|
|
|
|
func (p *repairWorker) SetOpCode(op uint8) {
|
|
p.Opcode = op
|
|
}
|
|
|
|
func (p *repairWorker) PackErrorBody(action, msg string) {
|
|
p.Size = uint32(len([]byte(action + "_" + msg)))
|
|
p.Data = make([]byte, p.Size)
|
|
copy(p.Data[:int(p.Size)], []byte(action+"_"+msg))
|
|
}
|
|
|
|
var sendWorker, recvWorker *repairWorker
|
|
|
|
func (p *repairWorker) WriteToConn(c net.Conn) (err error) {
|
|
pr := new(repl.Packet)
|
|
pr.Opcode = p.Opcode
|
|
pr.Data = make([]byte, len(p.Data))
|
|
copy(pr.Data, p.Data)
|
|
pr.ArgLen = p.ArgLen
|
|
pr.Arg = make([]byte, len(p.Arg))
|
|
copy(pr.Arg, p.Arg)
|
|
pr.Size = p.Size
|
|
pr.ResultCode = p.ResultCode
|
|
pr.StartT = p.StartT
|
|
pr.PartitionID = p.PartitionID
|
|
pr.ExtentID = p.ExtentID
|
|
pr.ExtentOffset = p.ExtentOffset
|
|
pr.ExtentType = p.ExtentType
|
|
pr.ReqID = p.ReqID
|
|
pr.CRC = p.CRC
|
|
|
|
p.dstWorker.packChannel <- pr
|
|
return nil
|
|
}
|
|
|
|
func (p *repairWorker) ReadFromConnWithVer(c net.Conn, timeoutSec int) (err error) {
|
|
pr := <-p.packChannel
|
|
{
|
|
p.CRC = pr.GetCRC()
|
|
p.Data = make([]byte, len(pr.GetData()))
|
|
copy(p.Data, pr.GetData())
|
|
p.ArgLen = pr.GetArgLen()
|
|
p.Arg = make([]byte, p.ArgLen)
|
|
copy(p.Arg, pr.GetArg())
|
|
|
|
p.ExtentID = pr.GetExtentID()
|
|
p.PartitionID = pr.GetPartitionID()
|
|
p.ResultCode = pr.GetResultCode()
|
|
p.Size = pr.GetSize()
|
|
p.ExtentOffset = pr.GetExtentOffset()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func getSrcPathExtentStore(role string) (string, func(), error) {
|
|
pattern := "cfs_storage_extentstore_" + role + "_"
|
|
dir, err := os.MkdirTemp(os.TempDir(), pattern)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
return fmt.Sprintf("%s/extents", dir), func() { os.RemoveAll(dir) }, nil
|
|
}
|
|
|
|
func reciverMakeTinyPacket(partitionID uint64, extentID uint64, offset, size int) (p repl.PacketInterface) {
|
|
return &repairWorker{}
|
|
}
|
|
|
|
func reciverMakeNormalPacket(partitionID uint64, extentID uint64, offset, size int) (p repl.PacketInterface) {
|
|
recvWorker.ExtentOffset = int64(offset)
|
|
recvWorker.Size = uint32(size)
|
|
recvWorker.PartitionID = partitionID
|
|
recvWorker.ExtentID = extentID
|
|
recvWorker.Opcode = proto.OpExtentRepairRead
|
|
|
|
return recvWorker
|
|
}
|
|
|
|
func reciverMakeExtentWithHoleRepairReadPacket(partitionID uint64, extentID uint64, offset, size int) (p repl.PacketInterface) {
|
|
recvWorker.ExtentOffset = int64(offset)
|
|
recvWorker.Size = uint32(size)
|
|
recvWorker.PartitionID = partitionID
|
|
recvWorker.ExtentID = extentID
|
|
recvWorker.Opcode = proto.OpSnapshotExtentRepairRead
|
|
|
|
return recvWorker
|
|
}
|
|
|
|
func reciverNewPacket() (p repl.PacketInterface) {
|
|
pr := recvWorker
|
|
pr.Magic = proto.ProtoMagic
|
|
pr.StartT = time.Now().UnixNano()
|
|
pr.NeedReply = true
|
|
return pr
|
|
}
|
|
|
|
func sendNewNormalReadResponsePacket(requestID int64, partitionID uint64, extentID uint64) (p repl.PacketInterface) {
|
|
pr := sendWorker
|
|
pr.ExtentID = extentID
|
|
pr.PartitionID = partitionID
|
|
pr.Magic = proto.ProtoMagic
|
|
pr.Opcode = proto.OpOk
|
|
pr.ReqID = requestID
|
|
pr.ExtentType = proto.NormalExtentType
|
|
|
|
return pr
|
|
}
|
|
|
|
func mockMakeDp(path string) *DataPartition {
|
|
return &DataPartition{
|
|
volumeID: "dpCfg.VolName",
|
|
clusterID: "clusterID",
|
|
partitionID: 1,
|
|
replicaNum: 3,
|
|
disk: &Disk{},
|
|
dataNode: &DataNode{},
|
|
path: path,
|
|
partitionType: proto.PartitionTypeNormal,
|
|
replicas: make([]string, 0),
|
|
stopC: make(chan bool),
|
|
stopRaftC: make(chan uint64),
|
|
storeC: make(chan uint64, 128),
|
|
snapshot: make([]*proto.File, 0),
|
|
partitionStatus: proto.ReadWrite,
|
|
config: &dataPartitionCfg{},
|
|
raftStatus: RaftStatusStopped,
|
|
DataPartitionCreateType: proto.NormalCreateDataPartition,
|
|
volVersionInfoList: &proto.VolVersionInfoList{},
|
|
}
|
|
}
|
|
|
|
func extentStoreNormalRwTest(t *testing.T, s *storage.ExtentStore, id uint64, crc uint32, data []byte) {
|
|
// append write
|
|
param := &storage.WriteParam{
|
|
ExtentID: id,
|
|
Offset: 0,
|
|
Size: int64(len(data)),
|
|
Data: data,
|
|
Crc: crc,
|
|
WriteType: storage.AppendWriteType,
|
|
IsSync: true,
|
|
IsHole: false,
|
|
IsRepair: false,
|
|
IsBackupWrite: false,
|
|
}
|
|
_, err := s.Write(param)
|
|
require.NoError(t, err)
|
|
actualCrc, err := s.Read(id, 0, int64(len(data)), data, false, false)
|
|
require.NoError(t, err)
|
|
require.EqualValues(t, crc, actualCrc)
|
|
// random write
|
|
param.WriteType = storage.RandomWriteType
|
|
_, err = s.Write(param)
|
|
require.NoError(t, err)
|
|
actualCrc, err = s.Read(id, 0, int64(len(data)), data, false, false)
|
|
require.NoError(t, err)
|
|
require.EqualValues(t, crc, actualCrc)
|
|
// TODO: append random write
|
|
require.NotEqualValues(t, s.GetStoreUsedSize(), 0)
|
|
}
|
|
|
|
func extentReloadCheckNormalCrc(t *testing.T, s *storage.ExtentStore, id uint64, crc uint32) {
|
|
var err error
|
|
// extent crc check
|
|
var e *storage.Extent
|
|
e, err = s.LoadExtentFromDisk(id, true)
|
|
assert.True(t, err == nil)
|
|
extCrc := e.GetCrc(0)
|
|
assert.True(t, crc == extCrc)
|
|
}
|
|
|
|
func extentStoreSnapshotRwTest(t *testing.T, s *storage.ExtentStore, id uint64, crc uint32, data []byte) {
|
|
// append write
|
|
param := &storage.WriteParam{
|
|
ExtentID: id,
|
|
Offset: int64(util.ExtentSize),
|
|
Size: int64(len(data)),
|
|
Data: data,
|
|
Crc: crc,
|
|
IsSync: true,
|
|
IsHole: false,
|
|
IsRepair: false,
|
|
IsBackupWrite: false,
|
|
}
|
|
|
|
offset := int64(util.ExtentSize)
|
|
param.WriteType = storage.AppendRandomWriteType
|
|
_, err := s.Write(param)
|
|
require.NoError(t, err)
|
|
|
|
param.WriteType = storage.AppendRandomWriteType
|
|
param.Offset = 0
|
|
_, err = s.Write(param)
|
|
assert.True(t, err != nil)
|
|
|
|
param.WriteType = storage.AppendWriteType
|
|
param.Offset = offset
|
|
_, err = s.Write(param)
|
|
assert.True(t, err != nil)
|
|
|
|
actualCrc, err := s.Read(id, offset, int64(len(data)), data, false, false)
|
|
require.NoError(t, err)
|
|
require.EqualValues(t, crc, actualCrc)
|
|
// random write
|
|
param.WriteType = storage.RandomWriteType
|
|
_, err = s.Write(param)
|
|
require.NoError(t, err)
|
|
actualCrc, err = s.Read(id, offset, int64(len(data)), data, false, false)
|
|
require.NoError(t, err)
|
|
require.EqualValues(t, crc, actualCrc)
|
|
// TODO: append random write
|
|
require.NotEqualValues(t, s.GetStoreUsedSize(), 0)
|
|
|
|
param.WriteType = storage.AppendRandomWriteType
|
|
_, err = s.Write(param)
|
|
require.NoError(t, err)
|
|
|
|
// extent crc check
|
|
var e *storage.Extent
|
|
e, err = s.LoadExtentFromDisk(id, true)
|
|
assert.True(t, err == nil)
|
|
extCrc := e.GetCrc(offset / util.BlockSize)
|
|
assert.True(t, crc == extCrc)
|
|
|
|
// check
|
|
param.Offset = int64(util.ExtentSize)*2 + util.BlockSize
|
|
_, err = s.Write(param)
|
|
require.NoError(t, err)
|
|
|
|
e, err = s.LoadExtentFromDisk(id, true)
|
|
assert.True(t, err == nil)
|
|
extCrc = e.GetCrc(param.Offset / util.BlockSize)
|
|
assert.True(t, crc == extCrc)
|
|
extCrc = e.GetCrc(param.Offset/util.BlockSize + 1)
|
|
assert.True(t, 0 == extCrc)
|
|
}
|
|
|
|
func extentReloadCheckSnapshotCrc(t *testing.T, path string, id uint64, crc uint32) (s *storage.ExtentStore) {
|
|
var err error
|
|
s, err = storage.NewExtentStore(path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, false)
|
|
require.NoError(t, err)
|
|
|
|
offset := int64(util.ExtentSize)
|
|
// extent crc check
|
|
var e *storage.Extent
|
|
e, err = s.LoadExtentFromDisk(id, true)
|
|
assert.True(t, err == nil)
|
|
extCrc := e.GetCrc(offset / util.BlockSize)
|
|
require.EqualValues(t, crc, extCrc)
|
|
|
|
// check
|
|
offset = int64(util.ExtentSize)*2 + util.BlockSize
|
|
e, err = s.LoadExtentFromDisk(id, true)
|
|
assert.True(t, err == nil)
|
|
extCrc = e.GetCrc(offset / util.BlockSize)
|
|
require.EqualValues(t, crc, extCrc)
|
|
extCrc = e.GetCrc(offset/util.BlockSize + 1)
|
|
assert.True(t, 0 == extCrc)
|
|
return s
|
|
}
|
|
|
|
func mockInitWorker(t *testing.T, role string) *repairWorker {
|
|
worker := &repairWorker{role: role}
|
|
worker.packChannel = make(chan repl.PacketInterface, 100)
|
|
path, _, err := getSrcPathExtentStore(role)
|
|
assert.True(t, err == nil)
|
|
s, err := storage.NewExtentStore(path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, true)
|
|
|
|
require.NoError(t, err)
|
|
worker.dp = mockMakeDp(path)
|
|
// spaceManager := NewSpaceManager(worker.dp.dataNode)
|
|
spaceManager := &SpaceManager{
|
|
dataNode: worker.dp.dataNode,
|
|
partitions: make(map[uint64]*DataPartition),
|
|
}
|
|
worker.dp.disk, err = NewDisk("/tmp", 200, 2000, 10, spaceManager, false)
|
|
require.NoError(t, err)
|
|
spaceManager.partitions[worker.dp.partitionID] = worker.dp
|
|
worker.dp.dataNode.space = spaceManager
|
|
worker.dp.extentStore = s
|
|
worker.dp.dataNode.getRepairConnFunc = func(target string) (net.Conn, error) {
|
|
return new(net.TCPConn), nil
|
|
}
|
|
worker.dp.dataNode.putRepairConnFunc = func(con net.Conn, force bool) {
|
|
_ = struct{}{}
|
|
}
|
|
return worker
|
|
}
|
|
|
|
func genDataAndGetCrc(repeatWord string, size int) (data []byte, crc uint32) {
|
|
repeatWordBytes := []byte(repeatWord)
|
|
repeatWordLen := len(repeatWordBytes)
|
|
numRepeats := size / repeatWordLen
|
|
remainingBytes := size % repeatWordLen
|
|
|
|
data = bytes.Repeat(repeatWordBytes, numRepeats)
|
|
data = append(data, repeatWordBytes[:remainingBytes]...)
|
|
crc = crc32.ChecksumIEEE(data)
|
|
return
|
|
}
|
|
|
|
func workerInit(t *testing.T, id uint64, data []byte, crc uint32) {
|
|
_ = id
|
|
_ = data
|
|
_ = crc
|
|
sendWorker = mockInitWorker(t, "sender")
|
|
recvWorker = mockInitWorker(t, "receiver")
|
|
sendWorker.dstWorker = recvWorker
|
|
recvWorker.dstWorker = sendWorker
|
|
}
|
|
|
|
func senderRepairWorker(t *testing.T, exitCh chan struct{}) {
|
|
con := new(net.TCPConn)
|
|
for {
|
|
select {
|
|
case pr := <-sendWorker.packChannel:
|
|
if pr.GetOpcode() == proto.OpSnapshotExtentRepairRead {
|
|
t.Logf("TestExtentRepair role %v recive packet (%v)", "sender", pr)
|
|
getReplyPacket := func() repl.PacketInterface {
|
|
sendWorker.ExtentID = pr.GetExtentID()
|
|
sendWorker.PartitionID = pr.GetPartitionID()
|
|
sendWorker.Magic = proto.ProtoMagic
|
|
sendWorker.Opcode = proto.OpOk
|
|
sendWorker.ReqID = pr.GetReqID()
|
|
sendWorker.ExtentType = proto.NormalExtentType
|
|
return sendWorker
|
|
}
|
|
sendWorker.dp.ExtentWithHoleRepairRead(pr, con, getReplyPacket)
|
|
} else if pr.GetOpcode() == proto.OpExtentRepairRead {
|
|
t.Logf("TestExtentRepair role %v recive packet (%v)", "sender", pr)
|
|
sendWorker.dp.NormalExtentRepairRead(pr, con, true, nil, sendNewNormalReadResponsePacket)
|
|
}
|
|
|
|
case <-exitCh:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func recvRepairWorker(t *testing.T, id uint64, exitCh chan struct{}) {
|
|
t.Logf("TestExtentRepair role %v start", "reciver")
|
|
ei, err := sendWorker.dp.extentStore.Watermark(id)
|
|
assert.True(t, err == nil)
|
|
t.Logf("TestExtentRepair role %v streamRepairExtent", "reciver")
|
|
repairExtent := &RepairExtentInfo{
|
|
ExtentInfo: *ei,
|
|
Source: "",
|
|
}
|
|
err = recvWorker.dp.streamRepairExtent(repairExtent, reciverMakeTinyPacket, reciverMakeNormalPacket, reciverMakeExtentWithHoleRepairReadPacket, reciverNewPacket)
|
|
assert.True(t, err == nil)
|
|
exitCh <- struct{}{}
|
|
}
|
|
|
|
func testDoRepair(t *testing.T, normalId uint64) {
|
|
var wg sync.WaitGroup
|
|
exitCh := make(chan struct{}, 1)
|
|
wg.Add(1)
|
|
go func(role string) {
|
|
defer func() {
|
|
wg.Done()
|
|
t.Logf("TestExtentRepair role %v finished", role)
|
|
}()
|
|
t.Logf("TestExtentRepair role %v start", role)
|
|
senderRepairWorker(t, exitCh)
|
|
}("sender")
|
|
|
|
wg.Add(1)
|
|
go func(role string) {
|
|
defer func() {
|
|
wg.Done()
|
|
t.Logf("TestExtentRepair role %v finished", role)
|
|
}()
|
|
recvRepairWorker(t, normalId, exitCh)
|
|
}("receiver")
|
|
wg.Wait()
|
|
t.Logf("TestExtentRepair finished")
|
|
}
|
|
|
|
func testDoNormalRepair(t *testing.T, normalId uint64, data []byte, crc uint32, isCreate bool) {
|
|
if isCreate {
|
|
s1 := sendWorker.dp.extentStore
|
|
err := s1.Create(normalId)
|
|
require.NoError(t, err)
|
|
|
|
s2 := recvWorker.dp.extentStore
|
|
err = s2.Create(normalId)
|
|
assert.True(t, err == nil)
|
|
}
|
|
extentStoreNormalRwTest(t, sendWorker.dp.extentStore, normalId, crc, data)
|
|
testDoRepair(t, normalId)
|
|
sendWorker.dp.extentStore.Close()
|
|
recvWorker.dp.extentStore.Close()
|
|
|
|
var err error
|
|
recvWorker.dp.extentStore, err = storage.NewExtentStore(recvWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, false)
|
|
require.NoError(t, err)
|
|
extentReloadCheckNormalCrc(t, recvWorker.dp.extentStore, normalId, crc)
|
|
recvWorker.dp.extentStore.Close()
|
|
}
|
|
|
|
func testDoSnapshotRepair(t *testing.T, normalId uint64, data []byte, crc uint32, isCreate bool) {
|
|
var err error
|
|
|
|
recvWorker.dp.extentStore, err = storage.NewExtentStore(recvWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, false)
|
|
require.NoError(t, err)
|
|
sendWorker.dp.extentStore, err = storage.NewExtentStore(sendWorker.dp.path, 0, 1*util.GB, proto.PartitionTypeNormal, 0, false)
|
|
require.NoError(t, err)
|
|
if isCreate {
|
|
s1 := sendWorker.dp.extentStore
|
|
err := s1.Create(normalId)
|
|
require.NoError(t, err)
|
|
|
|
s2 := recvWorker.dp.extentStore
|
|
err = s2.Create(normalId)
|
|
assert.True(t, err == nil)
|
|
}
|
|
|
|
// write snapshot
|
|
extentStoreSnapshotRwTest(t, sendWorker.dp.extentStore, normalId, crc, data)
|
|
|
|
// do append data large than 128MB
|
|
testDoRepair(t, normalId)
|
|
|
|
// check crc
|
|
recvWorker.dp.extentStore.Close()
|
|
recvWorker.dp.extentStore = extentReloadCheckSnapshotCrc(t, recvWorker.dp.path, normalId, crc)
|
|
}
|
|
|
|
func TestExtentRepair(t *testing.T) {
|
|
proto.InitBufferPool(int64(32768))
|
|
t.Logf("TestExtentRepair initWorker")
|
|
normalId := uint64(1025)
|
|
data, crc := genDataAndGetCrc("normal", util.BlockSize)
|
|
workerInit(t, normalId, data, crc)
|
|
defer func() {
|
|
os.RemoveAll(filepath.Dir(sendWorker.dp.path))
|
|
os.RemoveAll(filepath.Dir(recvWorker.dp.path))
|
|
}()
|
|
testDoNormalRepair(t, normalId, data, crc, true)
|
|
|
|
data, crc = genDataAndGetCrc("snapshot", util.BlockSize)
|
|
testDoSnapshotRepair(t, normalId, data, crc, false)
|
|
}
|
|
|
|
func TestExtentRepair_512KB(t *testing.T) {
|
|
proto.InitBufferPool(int64(32768))
|
|
t.Logf("TestExtentRepair_512KB initWorker")
|
|
normalId := uint64(1025)
|
|
data, crc := genDataAndGetCrc("normal", util.BlockSize)
|
|
workerInit(t, normalId, data, crc)
|
|
// NOTE: set repair size to 512KB
|
|
sendWorker.dp.SetRepairBlockSize(512 * util.KB)
|
|
t.Logf("Repair size is %v", sendWorker.dp.GetRepairBlockSize())
|
|
defer func() {
|
|
os.RemoveAll(filepath.Dir(sendWorker.dp.path))
|
|
os.RemoveAll(filepath.Dir(recvWorker.dp.path))
|
|
}()
|
|
testDoNormalRepair(t, normalId, data, crc, true)
|
|
data, crc = genDataAndGetCrc("snapshot", util.BlockSize)
|
|
testDoSnapshotRepair(t, normalId, data, crc, false)
|
|
}
|
|
|
|
func TestExtentRepairWithIOLimit(t *testing.T) {
|
|
proto.InitBufferPool(int64(32768))
|
|
t.Logf("TestExtentRepairWithIOLimit initWorker")
|
|
|
|
const repairRateLimit = 1 * util.MB
|
|
|
|
// normalIds := []uint64{1025, 1026, 1027, 1028, 1029, 1030, 1031, 1032, 1033, 1034}
|
|
normalIds := make([]uint64, 10)
|
|
for i := 0; i < 10; i++ {
|
|
normalIds[i] = uint64(1025 + i)
|
|
}
|
|
data, crc := genDataAndGetCrc("normal", util.BlockSize)
|
|
|
|
sendWorker = mockInitWorker(t, "sender")
|
|
recvWorker = mockInitWorker(t, "receiver")
|
|
sendWorker.dstWorker = recvWorker
|
|
recvWorker.dstWorker = sendWorker
|
|
|
|
sendWorker.dp.disk.limitAsyncRead = util.NewIOLimiter(repairRateLimit, 0)
|
|
recvWorker.dp.disk.limitAsyncWrite = util.NewIOLimiter(repairRateLimit, 0)
|
|
|
|
for _, id := range normalIds {
|
|
err := sendWorker.dp.extentStore.Create(id)
|
|
require.NoError(t, err)
|
|
err = recvWorker.dp.extentStore.Create(id)
|
|
require.NoError(t, err)
|
|
extentStoreNormalRwTest(t, sendWorker.dp.extentStore, id, crc, data)
|
|
}
|
|
|
|
defer func() {
|
|
os.RemoveAll(filepath.Dir(sendWorker.dp.path))
|
|
os.RemoveAll(filepath.Dir(recvWorker.dp.path))
|
|
}()
|
|
testConcurrentRepairs(t, normalIds, repairRateLimit)
|
|
}
|
|
|
|
func testConcurrentRepairs(t *testing.T, normalIds []uint64, repairRateLimit int64) {
|
|
var wg sync.WaitGroup
|
|
exitCh := make(chan struct{})
|
|
startTime := time.Now()
|
|
|
|
defer func() {
|
|
totalDuration := time.Since(startTime)
|
|
t.Logf("totalDuration: %v", totalDuration)
|
|
expectedMinDuration := time.Duration((len(normalIds)*util.BlockSize)/(int(repairRateLimit))) * time.Second
|
|
t.Logf("expectedMinDuration: %v", expectedMinDuration)
|
|
assert.True(t, totalDuration > expectedMinDuration,
|
|
fmt.Sprintf("actural cost time %v should > expected cost time %v", totalDuration, expectedMinDuration))
|
|
}()
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
senderRepairWorkerWithConcurrent(t, exitCh)
|
|
}()
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
recvRepairWorkerWithConcurrent(t, normalIds, exitCh)
|
|
}()
|
|
|
|
wg.Wait()
|
|
close(exitCh)
|
|
}
|
|
|
|
func senderRepairWorkerWithConcurrent(t *testing.T, exitCh chan struct{}) {
|
|
con := new(net.TCPConn)
|
|
var wg sync.WaitGroup
|
|
sem := make(chan struct{}, 10)
|
|
|
|
for {
|
|
select {
|
|
case pr := <-sendWorker.packChannel:
|
|
if pr.GetOpcode() == proto.OpExtentRepairRead {
|
|
sem <- struct{}{}
|
|
wg.Add(1)
|
|
go func(p repl.PacketInterface) {
|
|
defer func() {
|
|
<-sem
|
|
wg.Done()
|
|
}()
|
|
start := time.Now()
|
|
t.Logf("Sender: limitAsyncRead %v", sendWorker.dp.disk.limitAsyncRead.Status(false))
|
|
sendWorker.dp.NormalExtentRepairRead(pr, con, true, nil, sendNewNormalReadResponsePacket)
|
|
elapsed := time.Since(start)
|
|
t.Logf("request %d read cost time: %v", pr.GetExtentID(), elapsed)
|
|
}(pr)
|
|
}
|
|
case <-exitCh:
|
|
wg.Wait()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func recvRepairWorkerWithConcurrent(t *testing.T, ids []uint64, exitCh chan struct{}) {
|
|
var wg sync.WaitGroup
|
|
for _, id := range ids {
|
|
wg.Add(1)
|
|
go func(eid uint64) {
|
|
defer wg.Done()
|
|
ei, err := sendWorker.dp.extentStore.Watermark(eid)
|
|
require.NoError(t, err)
|
|
t.Logf("Recver: limitAsyncWrite %v", recvWorker.dp.disk.limitAsyncWrite.Status(false))
|
|
repairExtent := &RepairExtentInfo{ExtentInfo: *ei}
|
|
start := time.Now()
|
|
err = recvWorker.dp.streamRepairExtent(repairExtent,
|
|
reciverMakeTinyPacket,
|
|
reciverMakeNormalPacket,
|
|
reciverMakeExtentWithHoleRepairReadPacket,
|
|
reciverNewPacket)
|
|
assert.NoError(t, err)
|
|
elapsed := time.Since(start)
|
|
t.Logf("request %d write cost time: %v", eid, elapsed)
|
|
}(id)
|
|
}
|
|
wg.Wait()
|
|
exitCh <- struct{}{}
|
|
}
|
