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(data): Use Cursor to optimiza datanode code part3. #1000389775

Browse Source 
Signed-off-by: zhumingze <zhumingze@oppo.com>
This commit is contained in:
zhumingze 2025-11-24 16:39:10 +08:00 committed by 曾雪伟
parent 0362883c96
commit 415984dd59
14 changed files with 118 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
datanode/data_partition_repair.go
View File

@ -223,7 +223,7 @@ func (dp *DataPartition) getLocalExtentInfo(extentType uint8, tinyExtents []uint
err = errors.Trace(err, "getLocalExtentInfo extent DataPartition(%v) GetAllWaterMark", dp.partitionID)
return
}
if len(localExtents) <= 0 {
if len(localExtents) == 0 {
extents = make([]*storage.ExtentInfo, 0)
return
}
@ -239,7 +239,6 @@ func (dp *DataPartition) getRemoteExtentInfo(extentType uint8, tinyExtents []uin
target string,
) (extentFiles []*storage.ExtentInfo, err error) {
p := repl.NewPacketToGetAllWatermarks(dp.partitionID, extentType)
extentFiles = make([]*storage.ExtentInfo, 0)
if proto.IsTinyExtentType(extentType) {
p.Data, err = json.Marshal(tinyExtents)
if err != nil {
@ -270,7 +269,7 @@ func (dp *DataPartition) getRemoteExtentInfo(extentType uint8, tinyExtents []uin
}
data := reply.Data[:reply.Size]
extentFiles, err = storage.UnmarshalBinarySlice(data)
if err != nil && err == storage.ErrNonBytecodeEncode {
if err == storage.ErrNonBytecodeEncode {
extentFiles = make([]*storage.ExtentInfo, 0)
err = json.Unmarshal(data, &extentFiles)
if err != nil {
@ -354,11 +353,11 @@ func (dp *DataPartition) sendAllTinyExtentsToC(extentType uint8, availableTinyEx
}
func (dp *DataPartition) brokenTinyExtents() (brokenTinyExtents []uint64) {
brokenTinyExtents = make([]uint64, 0)
extentsToBeRepaired := MinTinyExtentsToRepair
if dp.extentStore.AvailableTinyExtentCnt() <= MinAvaliTinyExtentCnt {
extentsToBeRepaired = storage.TinyExtentCount
}
brokenTinyExtents = make([]uint64, 0, extentsToBeRepaired)
for i := 0; i < extentsToBeRepaired; i++ {
extentID, err := dp.extentStore.GetBrokenTinyExtent()
if err != nil {
@ -409,7 +408,7 @@ func (dp *DataPartition) prepareRepairTasks(repairTasks []*DataPartitionRepairTa
// Create a new extent if one of the replica is missing.
func (dp *DataPartition) buildExtentCreationTasks(repairTasks []*DataPartitionRepairTask, extentInfoMap map[uint64]*RepairExtentInfo) {
for extentID, extentInfo := range extentInfoMap {
if storage.IsTinyExtent(extentID) {
if storage.IsTinyExtent(extentID) || extentInfo.IsDeleted || dp.ExtentStore().IsDeletedNormalExtent(extentID) {
continue
}
for index := 0; index < len(repairTasks); index++ {
@ -417,16 +416,7 @@ func (dp *DataPartition) buildExtentCreationTasks(repairTasks []*DataPartitionRe
if repairTask == nil {
continue
}
if _, ok := repairTask.extents[extentID]; !ok && !extentInfo.IsDeleted {
if storage.IsTinyExtent(extentID) {
continue
}
if extentInfo.IsDeleted {
continue
}
if dp.ExtentStore().IsDeletedNormalExtent(extentID) {
continue
}
if _, ok := repairTask.extents[extentID]; !ok {
ei := &RepairExtentInfo{Source: extentInfo.Source}
ei.FileID = extentID
ei.SetSize(extentInfo.GetSize())
@ -866,13 +856,17 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *RepairExtentInfo,
log.LogDebugf("streamRepairExtent dp %v extent %v wait for token", dp.partitionID, remoteExtentInfo.FileID)
time.Sleep(time.Second * 5)
return storage.ErrNoDiskReadRepairExtentToken
} else {
err = errors.Trace(fmt.Errorf("unknow result code"),
"streamRepairExtent dp %v extent %v receive opcode error(%v) ,localExtentSize(%v) remoteExtentSize(%v)",
dp.partitionID, remoteExtentInfo.FileID, string(reply.GetData()[:intMin(len(reply.GetData()), int(reply.GetSize()))]), currFixOffset, dstOffset)
log.LogWarnf("%v", err.Error())
return
}
dataLen := len(reply.GetData())
sizeLen := int(reply.GetSize())
if sizeLen < dataLen {
dataLen = sizeLen
}
err = errors.Trace(fmt.Errorf("unknow result code"),
"streamRepairExtent dp %v extent %v receive opcode error(%v) ,localExtentSize(%v) remoteExtentSize(%v)",
dp.partitionID, remoteExtentInfo.FileID, string(reply.GetData()[:dataLen]), currFixOffset, dstOffset)
log.LogWarnf("%v", err.Error())
return
}
if reply.GetReqID() != request.GetReqID() || reply.GetPartitionID() != request.GetPartitionID() ||
@ -955,7 +949,6 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *RepairExtentInfo,
// log.LogDebugf("streamRepairExtent reply size %v, currFixoffset %v, reply %v err %v", reply.Size, currFixOffset, reply, err)
// write to the local extent file
if err != nil {
if isEmptyResponse && storage.IsTinyExtent(localExtentInfo.FileID) && strings.Contains(err.Error(), "offset invalid") {
err = errors.Trace(err, "streamRepairExtent repair data error, "+tinyOffsetInvalid)
return
@ -1013,11 +1006,3 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *RepairExtentInfo,
return
}
func intMin(a, b int) int {
if a < b {
return a
} else {
return b
}
}

4
datanode/data_partition_repair_test.go
View File

@ -363,7 +363,7 @@ func extentStoreSnapshotRwTest(t *testing.T, s *storage.ExtentStore, id uint64,
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)
assert.True(t, extCrc == 0)
}
func extentReloadCheckSnapshotCrc(t *testing.T, path string, id uint64, crc uint32) (s *storage.ExtentStore) {
@ -386,7 +386,7 @@ func extentReloadCheckSnapshotCrc(t *testing.T, path string, id uint64, crc uint
extCrc = e.GetCrc(offset / util.BlockSize)
require.EqualValues(t, crc, extCrc)
extCrc = e.GetCrc(offset/util.BlockSize + 1)
assert.True(t, 0 == extCrc)
assert.True(t, extCrc == 0)
return s
}

4
datanode/disk.go
View File

@ -715,8 +715,8 @@ func (d *Disk) ForceExitRaftStore() {
// DataPartitionList returns a list of the data partitions
func (d *Disk) DataPartitionList() (partitionIDs []uint64) {
d.Lock()
defer d.Unlock()
d.RLock()
defer d.RUnlock()
partitionIDs = make([]uint64, 0, len(d.partitionMap))
for _, dp := range d.partitionMap {
partitionIDs = append(partitionIDs, dp.partitionID)

31
datanode/limit.go
View File

@ -23,17 +23,15 @@ func initRepairLimit() {
}
func requestDoExtentRepair() (err error) {
err = fmt.Errorf("repair limit, cannot do extentRepair")
select {
case <-extentRepairLimitRater:
return nil
default:
return
return fmt.Errorf("repair limit, cannot do extentRepair")
}
}
func fininshDoExtentRepair() {
func finishDoExtentRepair() {
select {
case extentRepairLimitRater <- struct{}{}:
return
@ -55,13 +53,15 @@ func setDoExtentRepair(value int) {
value = MinExtentRepairLimit
}
if CurExtentRepairLimit != value {
CurExtentRepairLimit = value
close(extentRepairLimitRater)
extentRepairLimitRater = make(chan struct{}, CurExtentRepairLimit)
for i := 0; i < CurExtentRepairLimit; i++ {
extentRepairLimitRater <- struct{}{}
}
if CurExtentRepairLimit == value {
return
}
CurExtentRepairLimit = value
close(extentRepairLimitRater)
extentRepairLimitRater = make(chan struct{}, CurExtentRepairLimit)
for i := 0; i < CurExtentRepairLimit; i++ {
extentRepairLimitRater <- struct{}{}
}
}
@ -71,10 +71,9 @@ func DeleteLimiterWait() {
}
func setLimiter(limiter *rate.Limiter, limitValue uint64) {
r := limitValue
l := rate.Limit(r)
if r == 0 {
l = rate.Inf
if limitValue == 0 {
limiter.SetLimit(rate.Inf)
} else {
limiter.SetLimit(rate.Limit(limitValue))
}
limiter.SetLimit(l)
}

4
datanode/limit_test.go
View File

@ -35,7 +35,6 @@ func TestRequestDoExtentRepair(t *testing.T) {
// do nothing
default:
stop = true
break
}
}
@ -62,12 +61,11 @@ func TestFininshDoExtentRepair(t *testing.T) {
// do nothing
default:
stop = true
break
}
}
// finishDoExtentRepair() will send a struct{} to extentRepairLimitRater
fininshDoExtentRepair()
finishDoExtentRepair()
select {
case <-extentRepairLimitRater:

38
datanode/partition.go
View File

@ -631,15 +631,16 @@ func (dp *DataPartition) getVerListFromMaster() (err error) {
}
func (dp *DataPartition) replicasInit() {
replicas := make([]string, 0)
if dp.config.Hosts == nil {
return
}
replicas = append(replicas, dp.config.Hosts...)
dp.replicasLock.Lock()
dp.replicas = replicas
dp.replicas = make([]string, len(dp.config.Hosts))
copy(dp.replicas, dp.config.Hosts)
dp.replicasLock.Unlock()
if dp.config.Hosts != nil && len(dp.config.Hosts) >= 1 {
if len(dp.config.Hosts) >= 1 {
leaderAddr := strings.Split(dp.config.Hosts[0], ":")
if len(leaderAddr) == 2 && strings.TrimSpace(leaderAddr[0]) == LocalIP {
dp.isLeader = true
@ -1096,15 +1097,15 @@ func (dp *DataPartition) updateReplicas(isForce bool) (err error) {
// Compare the fetched replica with the local one.
func (dp *DataPartition) compareReplicas(v1, v2 []string) (equals bool) {
if len(v1) == len(v2) {
for i := 0; i < len(v1); i++ {
if v1[i] != v2[i] {
return false
}
}
return true
if len(v1) != len(v2) {
return false
}
return false
for i := range v1 {
if v1[i] != v2[i] {
return false
}
}
return true
}
type ReplicaInfo struct {
@ -1133,7 +1134,7 @@ func (dp *DataPartition) fetchReplicasFromMaster() (isLeader bool, replicas []st
for _, replica := range partition.Replicas {
infos = append(infos, ReplicaInfo{Addr: replica.Addr, Disk: replica.DiskPath})
}
if partition.Hosts != nil && len(partition.Hosts) >= 1 {
if len(partition.Hosts) >= 1 {
leaderAddr := strings.Split(partition.Hosts[0], ":")
if len(leaderAddr) == 2 && strings.TrimSpace(leaderAddr[0]) == LocalIP {
isLeader = true
@ -1147,18 +1148,12 @@ func (dp *DataPartition) Load() (response *proto.LoadDataPartitionResponse) {
response.PartitionId = uint64(dp.partitionID)
response.PartitionStatus = dp.partitionStatus
response.Used = uint64(dp.Used())
var err error
if dp.loadExtentHeaderStatus != FinishLoadDataPartitionExtentHeader {
response.PartitionSnapshot = make([]*proto.File, 0)
} else {
response.PartitionSnapshot = dp.SnapShot()
}
if err != nil {
response.Status = proto.TaskFailed
response.Result = err.Error()
return
}
return
}
@ -1322,8 +1317,8 @@ func (dp *DataPartition) doStreamFixTinyDeleteRecord(repairTask *DataPartitionRe
}
if p.IsErrPacket() {
logContent := fmt.Sprintf("action[doStreamFixTinyDeleteRecord] %v.",
p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), start, fmt.Errorf(string(p.Data[:p.Size]))))
err = fmt.Errorf(logContent)
p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), start, fmt.Errorf("%s", string(p.Data[:p.Size]))))
err = fmt.Errorf("%s", logContent)
return
}
@ -1391,6 +1386,7 @@ func (dp *DataPartition) canRemoveSelf() (canRemove bool, err error) {
for _, peer := range partition.Peers {
if dp.config.NodeID == peer.ID {
existInPeers = true
break
}
}
if !existInPeers {

59
datanode/partition_op_by_raft.go
View File

@ -95,7 +95,6 @@ func UnmarshalRandWriteRaftLog(raw []byte) (opItem *rndWrtOpItem, err error) {
if err = binary.Read(buff, binary.BigEndian, &version); err != nil {
return
}
if err = binary.Read(buff, binary.BigEndian, &opItem.opcode); err != nil {
return
}
@ -115,15 +114,11 @@ func UnmarshalRandWriteRaftLog(raw []byte) (opItem *rndWrtOpItem, err error) {
if _, err = buff.Read(opItem.data); err != nil {
return
}
return
}
func MarshalRaftCmd(raftOpItem *RaftCmdItem) (raw []byte, err error) {
if raw, err = json.Marshal(raftOpItem); err != nil {
return
}
return
return json.Marshal(raftOpItem)
}
func UnmarshalRaftCmd(raw []byte) (raftOpItem *RaftCmdItem, err error) {
@ -131,9 +126,7 @@ func UnmarshalRaftCmd(raw []byte) (raftOpItem *RaftCmdItem, err error) {
defer func() {
log.LogDebugf("Unmarsh use oldVersion,result %v", err)
}()
if err = json.Unmarshal(raw, raftOpItem); err != nil {
return
}
err = json.Unmarshal(raw, raftOpItem)
return
}
@ -179,14 +172,11 @@ func UnmarshalOldVersionRandWriteOpItem(raw []byte) (result *rndWrtOpItem, err e
// CheckLeader checks if itself is the leader during read
func (dp *DataPartition) CheckLeader(request *repl.Packet, connect net.Conn) (err error) {
// and use another getRaftLeaderAddr() to return the actual address
_, ok := dp.IsRaftLeader()
if !ok {
if _, ok := dp.IsRaftLeader(); !ok {
err = raft.ErrNotLeader
logContent := fmt.Sprintf("action[ReadCheck] %v.", request.LogMessage(request.GetOpMsg(), connect.RemoteAddr().String(), request.StartT, err))
log.LogWarnf(logContent)
return
}
return
}
@ -262,29 +252,38 @@ func (dp *DataPartition) ApplyRandomWrite(command []byte, raftApplyID uint64) (r
log.LogDebugf("[ApplyRandomWrite] ApplyID(%v) Partition(%v)_Extent(%v)_ExtentOffset(%v)_Size(%v)",
raftApplyID, dp.partitionID, opItem.extentID, opItem.offset, opItem.size)
for i := 0; i < 20; i++ {
var syncWrite bool
writeType := storage.RandomWriteType
if opItem.opcode == proto.OpRandomWrite || opItem.opcode == proto.OpSyncRandomWrite {
if dp.verSeq > 0 {
err = storage.ErrVerNotConsistent
log.LogErrorf("action[ApplyRandomWrite] volume [%v] dp [%v] %v,client need update to newest version!", dp.volumeID, dp.partitionID, err)
return
}
} else if opItem.opcode == proto.OpRandomWriteAppend || opItem.opcode == proto.OpSyncRandomWriteAppend {
writeType = storage.AppendRandomWriteType
} else if opItem.opcode == proto.OpTryWriteAppend || opItem.opcode == proto.OpSyncTryWriteAppend {
writeType = storage.AppendWriteType
var syncWrite bool
writeType := storage.RandomWriteType
switch opItem.opcode {
case proto.OpRandomWrite, proto.OpSyncRandomWrite:
if dp.verSeq > 0 {
err = storage.ErrVerNotConsistent
log.LogErrorf("action[ApplyRandomWrite] volume [%v] dp [%v] %v,client need update to newest version!", dp.volumeID, dp.partitionID, err)
return
}
if opItem.opcode == proto.OpSyncRandomWriteAppend || opItem.opcode == proto.OpSyncRandomWrite || opItem.opcode == proto.OpSyncRandomWriteVer {
if opItem.opcode == proto.OpSyncRandomWrite {
syncWrite = true
}
case proto.OpRandomWriteAppend, proto.OpSyncRandomWriteAppend:
writeType = storage.AppendRandomWriteType
if opItem.opcode == proto.OpSyncRandomWriteAppend {
syncWrite = true
}
case proto.OpTryWriteAppend, proto.OpSyncTryWriteAppend:
writeType = storage.AppendWriteType
if opItem.opcode == proto.OpSyncTryWriteAppend {
syncWrite = true
}
case proto.OpSyncRandomWriteVer:
syncWrite = true
}
for i := 0; i < 20; i++ {
dp.disk.diskLimit(OpWrite, uint32(opItem.size), func() {
param := &storage.WriteParam{
ExtentID: uint64(opItem.extentID),
Offset: int64(opItem.offset),
ExtentID: opItem.extentID,
Offset: opItem.offset,
Size: opItem.size,
Data: opItem.data,
Crc: opItem.crc,

4
datanode/partition_raft.go
View File

@ -895,8 +895,8 @@ func (dp *DataPartition) getRemoteAppliedID(target string, p *repl.Packet) (appl
start := time.Now().UnixNano()
defer func() {
if err != nil {
err = fmt.Errorf(p.LogMessage(p.GetOpMsg(), target, start, err))
log.LogErrorf(err.Error())
err = errors.New(p.LogMessage(p.GetOpMsg(), target, start, err))
log.LogError(err.Error())
}
}()

25
datanode/server.go
View File

@ -49,15 +49,17 @@ import (
)
var (
ErrIncorrectStoreType = errors.New("Incorrect store type")
ErrNoSpaceToCreatePartition = errors.New("No disk space to create a data partition")
ErrNewSpaceManagerFailed = errors.New("Creater new space manager failed")
ErrGetMasterDatanodeInfoFailed = errors.New("Failed to get datanode info from master")
ErrIncorrectStoreType = errors.New("incorrect store type")
ErrNoSpaceToCreatePartition = errors.New("no disk space to create a data partition")
ErrNewSpaceManagerFailed = errors.New("creater new space manager failed")
ErrGetMasterDatanodeInfoFailed = errors.New("failed to get datanode info from master")
LocalIP string
gConnPool = util.NewConnectPool()
// MasterClient = masterSDK.NewMasterClient(nil, false)
MasterClient *masterSDK.MasterCLientWithResolver
regexpPort = regexp.MustCompile(`^(\d)+$`)
)
const (
@ -282,7 +284,7 @@ func (s *DataNode) Sync() {
func doStart(server common.Server, cfg *config.Config) (err error) {
s, ok := server.(*DataNode)
if !ok {
return errors.New("Invalid node Type!")
return errors.New("invalid node type")
}
s.stopC = make(chan bool)
s.gogcValue = DefaultGOGCValue
@ -405,18 +407,11 @@ func doShutdown(server common.Server) {
}
func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
var (
port string
regexpPort *regexp.Regexp
)
LocalIP = cfg.GetString(ConfigKeyLocalIP)
port = cfg.GetString(proto.ListenPort)
port := cfg.GetString(proto.ListenPort)
s.bindIp = cfg.GetBool(proto.BindIpKey)
if regexpPort, err = regexp.Compile(`^(\d)+$`); err != nil {
return fmt.Errorf("Err:no port")
}
if !regexpPort.MatchString(port) {
return fmt.Errorf("Err:port must string")
return fmt.Errorf("port must be a string")
}
s.port = port
@ -431,7 +426,7 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
addrs := cfg.GetSlice(proto.MasterAddr)
if len(addrs) == 0 {
return fmt.Errorf("Err:masterAddr unavalid")
return fmt.Errorf("masterAddr invalid")
}
masters := make([]string, 0, len(addrs))
for _, addr := range addrs {

7
datanode/server_handler.go
View File

@ -488,14 +488,11 @@ func (s *DataNode) buildFailureResp(w http.ResponseWriter, code int, msg string)
// Create response for the API request.
func (s *DataNode) buildJSONResp(w http.ResponseWriter, code int, data interface{}, msg string) {
var (
jsonBody []byte
err error
)
w.WriteHeader(code)
w.Header().Set("Content-Type", "application/json")
body := proto.HTTPReply{Code: int32(code), Msg: msg, Data: data}
if jsonBody, err = json.Marshal(body); err != nil {
jsonBody, err := json.Marshal(body)
if err != nil {
return
}
w.Write(jsonBody)

5
datanode/space_manager.go
View File

@ -571,12 +571,13 @@ func (manager *SpaceManager) updateMetrics() {
const DiskSelectMaxStraw = 65536
func (manager *SpaceManager) selectDisk(decommissionedDisks []string) (d *Disk) {
manager.diskMutex.Lock()
defer manager.diskMutex.Unlock()
decommissionedDiskMap := make(map[string]struct{})
for _, disk := range decommissionedDisks {
decommissionedDiskMap[disk] = struct{}{}
}
manager.diskMutex.Lock()
defer manager.diskMutex.Unlock()
maxStraw := float64(0)
for _, disk := range manager.disks {
if _, ok := decommissionedDiskMap[disk.Path]; ok {

28
datanode/stat_log.go
View File

@ -15,16 +15,13 @@ import (
func (d *DataNode) startStat(cfg *config.Config) {
logDir := cfg.GetString(ConfigKeyLogDir)
var err error
var logLeftSpaceLimitRatio float64
logLeftSpaceLimitRatio := log.DefaultLogLeftSpaceLimitRatio
logLeftSpaceLimitRatioStr := cfg.GetString("logLeftSpaceLimitRatio")
if logLeftSpaceLimitRatioStr == "" {
logLeftSpaceLimitRatio = log.DefaultLogLeftSpaceLimitRatio
} else {
logLeftSpaceLimitRatio, err = strconv.ParseFloat(logLeftSpaceLimitRatioStr, 64)
if err != nil {
if logLeftSpaceLimitRatioStr := cfg.GetString("logLeftSpaceLimitRatio"); logLeftSpaceLimitRatioStr != "" {
if val, err := strconv.ParseFloat(logLeftSpaceLimitRatioStr, 64); err == nil {
logLeftSpaceLimitRatio = val
} else {
log.LogWarnf("get log limt ratio failed, err %s", err.Error())
logLeftSpaceLimitRatio = log.DefaultLogLeftSpaceLimitRatio
}
}
@ -58,7 +55,7 @@ func (d *DataNode) getDpOpLog() []proto.OpLog {
}
func (d *DataNode) getOplogs(ops []*stat.Operation) []proto.OpLog {
oplog := make([]proto.OpLog, 0)
oplog := make([]proto.OpLog, 0, len(ops))
for _, op := range ops {
if op.Op == "" {
arr := strings.Split(op.Name, "_")
@ -92,8 +89,8 @@ func (s *DataNode) setOpLog(w http.ResponseWriter, r *http.Request) {
sendMaster, err := strconv.ParseBool(r.FormValue(paramSendMaster))
log.LogDebugf("action[setOpLog] sendMaster:%+v,err:%+v", sendMaster, err)
if err == nil {
stat.DpStat.SetRecordFile(sendMaster)
stat.DiskStat.SetRecordFile(sendMaster)
stat.DpStat.SetSendMaster(sendMaster)
stat.DiskStat.SetSendMaster(sendMaster)
}
s.buildSuccessResp(w, "success")
}
@ -121,10 +118,10 @@ func (s *DataNode) getOpLog(w http.ResponseWriter, r *http.Request) {
case "dp":
stat.DpStat.Lock()
oplogs = s.getOplogs(stat.DpStat.GetPrevOps())
stat.DpStat.Unlock()
if dpId != "" {
oplogs = filterDpId(dpId, oplogs)
}
stat.DpStat.Unlock()
default:
}
if op != "" {
@ -137,7 +134,7 @@ func filterDpId(dpId string, oplogs []proto.OpLog) []proto.OpLog {
if dpId == "" {
return oplogs
}
out := make([]proto.OpLog, 0)
out := make([]proto.OpLog, 0, len(oplogs))
for _, oplog := range oplogs {
arr := strings.Split(oplog.Name, "_")
if len(arr) < 2 {
@ -154,9 +151,10 @@ func filterOp(op string, oplogs []proto.OpLog) []proto.OpLog {
if op == "" {
return oplogs
}
out := make([]proto.OpLog, 0)
out := make([]proto.OpLog, 0, len(oplogs))
opLower := strings.ToLower(op)
for _, oplog := range oplogs {
if !strings.Contains(strings.ToLower(oplog.Op), strings.ToLower(op)) {
if !strings.Contains(strings.ToLower(oplog.Op), opLower) {
continue
}
out = append(out, oplog)

11
datanode/wrap_operator.go
View File

@ -311,7 +311,8 @@ func (s *DataNode) handlePacketToCreateDataPartition(p *repl.Packet) {
}
func (s *DataNode) commitDelVersion(volumeID string, verSeq uint64) (err error) {
for _, partition := range s.space.partitions {
partitions := s.space.getPartitions()
for _, partition := range partitions {
if partition.config.VolName != volumeID {
continue
}
@ -358,7 +359,7 @@ func (s *DataNode) commitCreateVersion(req *proto.MultiVersionOpRequest) (err er
}
s.space.partitionMutex.RLock()
partitions := make([]*DataPartition, 0)
partitions := make([]*DataPartition, 0, len(s.space.partitions))
for _, dp := range s.space.partitions {
partitions = append(partitions, dp)
}
@ -453,7 +454,7 @@ func (s *DataNode) handleUpdateVerPacket(p *repl.Packet) {
}
}()
if !s.clusterEnableSnapshot {
err = fmt.Errorf("cluster not enable snapshot!")
err = fmt.Errorf("cluster not enable snapshot")
return
}
task := &proto.AdminTask{}
@ -739,7 +740,7 @@ func (s *DataNode) asyncLoadDataPartition(task *proto.AdminTask) {
if dp == nil {
response.Status = proto.TaskFailed
response.PartitionId = uint64(request.PartitionId)
err = fmt.Errorf(fmt.Sprintf("DataPartition(%v) not found", request.PartitionId))
err = fmt.Errorf("DataPartition(%v) not found", request.PartitionId)
response.Result = err.Error()
} else {
response = dp.Load()
@ -1173,7 +1174,7 @@ func (s *DataNode) handleExtentRepairReadPacket(p *repl.Packet, connect net.Conn
if err != nil {
return
}
defer fininshDoExtentRepair()
defer finishDoExtentRepair()
partition := p.Object.(*DataPartition)
if !partition.disk.RequireReadExtentToken(partition.partitionID) {
err = storage.ErrNoDiskReadRepairExtentToken

2
datanode/wrap_prepare.go
View File

@ -114,7 +114,7 @@ func (s *DataNode) checkPartition(p *repl.Packet) (err error) {
func (s *DataNode) checkPacketAndPrepare(p *repl.Packet) error {
partition := p.Object.(*DataPartition)
store := p.Object.(*DataPartition).ExtentStore()
store := partition.ExtentStore()
var (
extentID uint64
err error