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

refactor(master): resolve conflicts when picking to master branch. #22906448

Browse Source 
Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2024-12-23 15:50:38 +08:00 committed by AmazingChi
parent 3396accd80
commit dac1a12d0a
30 changed files with 116 additions and 254 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
cli/cmd/disk.go
View File

@ -121,7 +121,7 @@ func newListBadDiskCmd(client *master.MasterClient) *cobra.Command {
Short: cmdCheckBadDisksShort,
Run: func(cmd *cobra.Command, args []string) {
var (
infos *proto.DiskInfos
infos *proto.BadDiskInfos
err error
)
defer func() {
@ -131,12 +131,17 @@ func newListBadDiskCmd(client *master.MasterClient) *cobra.Command {
return
}
stdout("(partitionID=0 means detected by datanode disk checking, not associated with any partition)\n\n[Unavaliable disks]:\n")
sort.SliceStable(infos.Disks, func(i, j int) bool {
return infos.Disks[i].Address < infos.Disks[j].Address
stdout("%v\n", formatBadDiskTableHeader())
sort.SliceStable(infos.BadDisks, func(i, j int) bool {
return infos.BadDisks[i].Address < infos.BadDisks[j].Address
})
stdoutln(formatBadDisks(infos.Disks))
for _, disk := range infos.BadDisks {
stdout("%v\n", formatBadDiskInfoRow(disk))
}
},
}
return cmd
}

10
cli/cmd/fmt.go
View File

@ -1135,6 +1135,16 @@ func formatQuotaInfo(info *proto.QuotaInfo) string {
return ret
}
var badDiskDetailTableRowPattern = "%-18v %-18v %-18v %-18v %-18v"
func formatBadDiskTableHeader() string {
return fmt.Sprintf(badDiskDetailTableRowPattern, "Address", "Path", "TotalPartitionCnt", "DiskErrPartitionCnt", "PartitionIdsWithDiskErr")
}
func formatBadDiskInfoRow(disk proto.BadDiskInfo) string {
msgDpIdList := fmt.Sprintf("%v", disk.DiskErrPartitionList)
return fmt.Sprintf(badDiskDetailTableRowPattern, disk.Address, disk.Path, disk.TotalPartitionCnt, len(disk.DiskErrPartitionList), msgDpIdList)
}
func formatBadDisks(disks []proto.DiskInfo) string {
if len(disks) == 0 {
return ""

1
client/fdstore/fdstore.go
View File

@ -314,6 +314,7 @@ func doDump(filePathes string) {
)
rsize, err = handleListFile.Read(data)
if rsize == 0 || err == io.EOF {
err = nil
break
}

1
client/fs/dir.go
View File

@ -564,6 +564,7 @@ func (d *Dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
}
batchNr := uint64(len(batches))
if batchNr == 0 || (from != "" && batchNr == 1) {
noMore = true
break
} else if batchNr < DefaultReaddirLimit {
noMore = true

2
client/fuse.go
View File

@ -88,7 +88,7 @@ const (
LoggerPrefix = "client"
LoggerOutput = "output.log"
ModuleName = "fuseclient"
// ConfigKeyExporterPort = "exporterKey"
ConfigKeyExporterPort = "exporterKey"
ControlCommandSetRate = "/rate/set"
ControlCommandGetRate = "/rate/get"

2
client/libsdk/libsdk.go
View File

@ -156,6 +156,8 @@ var (
statusEPERM = errorToStatus(syscall.EPERM)
)
var once sync.Once
func init() {
gClientManager = &clientManager{
clients: make(map[int64]*client),

1
datanode/partition_raft.go
View File

@ -31,7 +31,6 @@ import (
raftproto "github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/raftstore"
"github.com/cubefs/cubefs/repl"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/config"
"github.com/cubefs/cubefs/util/errors"

1
datanode/partition_raftfsm.go
View File

@ -26,7 +26,6 @@ import (
"github.com/cubefs/cubefs/depends/tiglabs/raft"
raftproto "github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/storage"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"

1
datanode/repl/packet.go
View File

@ -211,6 +211,7 @@ func (p *Packet) resolveFollowersAddr() (err error) {
p.followersAddrs = followerAddrs[:int(followerNum)]
log.LogInfof("action[resolveFollowersAddr] %v", p.followersAddrs)
}
return
}

1
datanode/space_manager.go
View File

@ -572,6 +572,7 @@ func (s *DataNode) buildHeartBeatResponse(response *proto.DataNodeHeartbeatRespo
response.MaxCapacity = stat.MaxCapacityToCreatePartition
response.RemainingCapacity = stat.RemainingCapacityToCreatePartition
response.BadDisks = make([]string, 0)
response.BadDiskStats = make([]proto.BadDiskStat, 0)
response.DiskStats = make([]proto.DiskStat, 0)
response.StartTime = s.startTime
stat.Unlock()

1
datanode/wrap_operator.go
View File

@ -1145,7 +1145,6 @@ func (s *DataNode) handleExtentRepairReadPacket(p *repl.Packet, connect net.Conn
partition := p.Object.(*DataPartition)
if !partition.disk.RequireReadExtentToken(partition.partitionID) {
err = storage.NoDiskReadRepairExtentTokenError
log.LogWarn("check the source code cos i don't understand the unuesd error,", err)
log.LogDebugf("dp(%v) disk(%v) extent(%v) wait for read extent token",
p.PartitionID, partition.disk.Path, p.ExtentID)
return

1
master/api_service.go
View File

@ -2943,6 +2943,7 @@ func (m *Server) checkCreateVolReq(req *createVolReq) (err error) {
func (m *Server) createVol(w http.ResponseWriter, r *http.Request) {
req := &createVolReq{}
vol := &Vol{}
var err error
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminCreateVol))

83
master/cluster.go
View File

@ -193,6 +193,10 @@ func (mgr *followerReadManager) getVolumeDpView() {
panic(err)
}
if len(volViews) == 0 {
return
}
mgr.rwMutex.Lock()
mgr.volViewMap = make(map[string]*volValue)
for _, vv := range volViews {
@ -211,9 +215,10 @@ func (mgr *followerReadManager) getVolumeDpView() {
return
}
avgSleepTime := time.Second * 5 / time.Duration(len(volViews))
for _, vv := range volViews {
if (vv.Status == proto.VolStatusMarkDelete && !vv.Forbidden) ||
(vv.Status == proto.VolStatusMarkDelete && vv.Forbidden && time.Until(vv.DeleteExecTime) <= 0) {
if (vv.Status == proto.VolStatusMarkDelete && !vv.Forbidden) || (vv.Status == proto.VolStatusMarkDelete && vv.Forbidden && time.Since(vv.DeleteExecTime) <= 0) {
mgr.rwMutex.Lock()
mgr.lastUpdateTick[vv.Name] = time.Now()
mgr.status[vv.Name] = false
@ -226,6 +231,7 @@ func (mgr *followerReadManager) getVolumeDpView() {
log.LogErrorf("followerReadManager.getVolumeDpView %v GetDataPartitions err %v leader(%v)", vv.Name, err, mgr.c.masterClient.Leader())
continue
}
time.Sleep(avgSleepTime)
mgr.updateVolViewFromLeader(vv.Name, view)
}
}
@ -233,12 +239,18 @@ func (mgr *followerReadManager) getVolumeDpView() {
func (mgr *followerReadManager) sendFollowerVolumeDpView() {
var err error
vols := mgr.c.copyVols()
if len(vols) == 0 {
return
}
avgSleepTime := time.Second * 5 / time.Duration(len(vols))
for _, vol := range vols {
log.LogDebugf("followerReadManager.getVolumeDpView %v", vol.Name)
if (vol.Status == proto.VolStatusMarkDelete && !vol.Forbidden) ||
(vol.Status == proto.VolStatusMarkDelete && vol.Forbidden && time.Until(vol.DeleteExecTime) <= 0) {
if (vol.Status == proto.VolStatusMarkDelete && !vol.Forbidden) || (vol.Status == proto.VolStatusMarkDelete && vol.Forbidden && vol.DeleteExecTime.Sub(time.Now()) <= 0) {
continue
}
time.Sleep(avgSleepTime)
var body []byte
if body, err = vol.getDataPartitionsView(); err != nil {
log.LogErrorf("followerReadManager.sendFollowerVolumeDpView err %v", err)
@ -788,35 +800,44 @@ func (c *Cluster) releaseDataPartitionAfterLoad() {
}
func (c *Cluster) scheduleToCheckHeartbeat() {
go func() {
for {
if c.partition != nil && c.partition.IsRaftLeader() {
c.checkLeaderAddr()
c.checkDataNodeHeartbeat()
// update load factor
setOverSoldFactor(c.cfg.ClusterLoadFactor)
}
time.Sleep(time.Second * defaultIntervalToCheckHeartbeat)
}
}()
c.runTask(
&cTask{
tickTime: time.Second * defaultIntervalToCheckHeartbeat,
name: "scheduleToCheckHeartbeat_checkDataNodeHeartbeat",
function: func() (fin bool) {
if c.partition != nil && c.partition.IsRaftLeader() {
c.checkLeaderAddr()
c.checkDataNodeHeartbeat()
// update load factor
setOverSoldFactor(c.cfg.ClusterLoadFactor)
}
return
},
})
go func() {
for {
if c.partition != nil && c.partition.IsRaftLeader() {
c.checkMetaNodeHeartbeat()
}
time.Sleep(time.Second * defaultIntervalToCheckHeartbeat)
}
}()
c.runTask(
&cTask{
tickTime: time.Second * defaultIntervalToCheckHeartbeat,
name: "scheduleToCheckHeartbeat_checkMetaNodeHeartbeat",
function: func() (fin bool) {
if c.partition != nil && c.partition.IsRaftLeader() {
c.checkMetaNodeHeartbeat()
}
return
},
})
go func() {
for {
if c.partition != nil && c.partition.IsRaftLeader() {
c.checkLcNodeHeartbeat()
}
time.Sleep(time.Second * defaultIntervalToCheckHeartbeat)
}
}()
c.runTask(
&cTask{
tickTime: time.Second * defaultIntervalToCheckHeartbeat,
name: "scheduleToCheckHeartbeat_checkLcNodeHeartbeat",
function: func() (fin bool) {
if c.partition != nil && c.partition.IsRaftLeader() {
c.checkLcNodeHeartbeat()
}
return
},
})
}
func (c *Cluster) passAclCheck(ip string) {

4
master/data_partition.go
View File

@ -1518,10 +1518,6 @@ func (partition *DataPartition) Decommission(c *Cluster) bool {
}
errHandler:
// choose other node to create data partition
if resetDecommissionDst {
partition.DecommissionDstAddr = ""
}
// special replica num receive stop signal,donot reset SingleDecommissionStatus for decommission again
if partition.GetDecommissionStatus() == DecommissionPause {
log.LogWarnf("action[decommissionDataPartition] partitionID:%v is stopped", partition.PartitionID)

5
master/metadata_fsm_op.go
View File

@ -1429,12 +1429,13 @@ func (c *Cluster) loadZoneDomain() (ok bool, err error) {
for zoneName, domainId := range c.domainManager.ZoneName2DomainIdMap {
log.LogInfof("action[loadZoneDomain] zoneName %v domainid %v", zoneName, domainId)
if _, ok := c.domainManager.domainId2IndexMap[domainId]; !ok {
if domainIndex, ok := c.domainManager.domainId2IndexMap[domainId]; !ok {
log.LogInfof("action[loadZoneDomain] zoneName %v domainid %v build new domainnodesetgrp manager", zoneName, domainId)
domainGrp := newDomainNodeSetGrpManager()
domainGrp.domainId = domainId
c.domainManager.domainNodeSetGrpVec = append(c.domainManager.domainNodeSetGrpVec, domainGrp)
c.domainManager.domainId2IndexMap[domainId] = len(c.domainManager.domainNodeSetGrpVec) - 1
domainIndex = len(c.domainManager.domainNodeSetGrpVec) - 1
c.domainManager.domainId2IndexMap[domainId] = domainIndex
}
}

1
master/topology.go
View File

@ -2375,6 +2375,7 @@ func (l *DecommissionDataPartitionList) traverse(c *Cluster) {
msg := fmt.Sprintf("ns %v(%p) dp %v decommission success, cost %v",
l.nsId, l, dp.decommissionInfo(), time.Since(dp.RecoverStartTime))
dp.ResetDecommissionStatus()
dp.setRestoreReplicaStop()
err := c.syncUpdateDataPartition(dp)
if err != nil {
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) Remove success dp[%v] failed for %v",

1
metanode/const.go
View File

@ -279,7 +279,6 @@ var (
_ = opFSMUpdateSummaryInfo
_ = (*Dentry).getLastestVer
_ = (*Inode).isEkInRefMap
_ = (*metaPartition).notifyRaftFollowerToFreeInodes
_ = (*metaPartition).decommissionPartition
_ = (*metaPartition).getDentryTree
_ = (*metaPartition).internalHasInode

32
metanode/extend.go
View File

@ -36,12 +36,6 @@ type ExtendMultiSnap struct {
versionMu sync.RWMutex
}
type ExtendMultiSnap struct {
verSeq uint64
multiVers []*Extend
versionMu sync.RWMutex
}
type Extend struct {
inode uint64
dataMap map[string][]byte
@ -71,27 +65,6 @@ func (e *Extend) setVersion(seq uint64) {
e.multiSnap.verSeq = seq
}
func (e *Extend) getVersion() uint64 {
if e.multiSnap == nil {
return 0
}
return e.multiSnap.verSeq
}
func (e *Extend) genSnap() {
if e.multiSnap == nil {
e.multiSnap = &ExtendMultiSnap{}
}
e.multiSnap.multiVers = append([]*Extend{e.Copy().(*Extend)}, e.multiSnap.multiVers...)
}
func (e *Extend) setVersion(seq uint64) {
if e.multiSnap == nil {
e.multiSnap = &ExtendMultiSnap{}
}
e.multiSnap.verSeq = seq
}
func (e *Extend) checkSequence() (err error) {
if e.multiSnap == nil {
return
@ -265,7 +238,6 @@ func (e *Extend) Remove(key []byte) {
return
}
delete(e.dataMap, string(key))
e.mu.Unlock()
}
func (e *Extend) Range(visitor func(key, value []byte) bool) {
@ -305,11 +277,11 @@ func (e *Extend) Copy() btree.Item {
newExt.dataMap[k] = v
}
}
newExt.Quota = make([]byte, len(e.Quota))
copy(newExt.Quota, e.Quota)
if e.multiSnap == nil {
return newExt
}
newExt.Quota = make([]byte, len(e.Quota))
copy(newExt.Quota, e.Quota)
newExt.multiSnap = &ExtendMultiSnap{}
newExt.multiSnap.verSeq = e.multiSnap.verSeq
newExt.multiSnap.multiVers = e.multiSnap.multiVers

4
metanode/metanode.go
View File

@ -281,8 +281,8 @@ func (m *MetaNode) parseConfig(cfg *config.Config) (err error) {
log.LogInfof("[parseConfig] raftRetainLogs[%v]", m.raftRetainLogs)
if cfg.HasKey(cfgRaftSyncSnapFormatVersion) {
raftSyncSnapFormatVersion := uint32(cfg.GetInt(cfgRaftSyncSnapFormatVersion))
if raftSyncSnapFormatVersion > SnapFormatVersion_1 {
raftSyncSnapFormatVersion := uint32(cfg.GetInt64(cfgRaftSyncSnapFormatVersion))
if raftSyncSnapFormatVersion < 0 || raftSyncSnapFormatVersion > SnapFormatVersion_1 {
m.raftSyncSnapFormatVersion = SnapFormatVersion_1
log.LogInfof("invalid config raftSyncSnapFormatVersion, using default[%v]", m.raftSyncSnapFormatVersion)
} else {

4
objectnode/auth.go
View File

@ -206,7 +206,9 @@ func (o *ObjectNode) validateAuthInfo(r *http.Request, auth Auther) (err error)
mux.Vars(r)[ContextKeyAccessKey] = ak
mux.Vars(r)[ContextKeyRequester] = uid
if !o.signatureIgnoredActions.Contains(param.action) {
if !param.action.IsNone() && o.signatureIgnoredActions.Contains(param.action) {
return nil
}
cred := auth.Credential()
if auth.IsSkewed() {
log.LogErrorf("validateAuthInfo: request skewed: requestID(%v) reqTime(%v) servTime(%v)",

149
objectnode/lifecycle_handler.go
View File

@ -1,149 +0,0 @@
// Copyright 2023 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package objectnode
import (
"encoding/xml"
"io"
"net/http"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/log"
)
// API reference: https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/API/API_GetBucketLifecycleConfiguration.html
func (o *ObjectNode) getBucketLifecycleConfigurationHandler(w http.ResponseWriter, r *http.Request) {
var err error
var errorCode *ErrorCode
defer func() {
o.errorResponse(w, r, err, errorCode)
}()
param := ParseRequestParam(r)
if param.Bucket() == "" {
errorCode = InvalidBucketName
return
}
if _, err = o.vm.Volume(param.Bucket()); err != nil {
errorCode = NoSuchBucket
return
}
var lcConf *proto.LcConfiguration
if lcConf, err = o.mc.AdminAPI().GetBucketLifecycle(param.Bucket()); err != nil {
log.LogErrorf("getBucketLifecycle failed: bucket[%v] err(%v)", param.Bucket(), err)
errorCode = NoSuchLifecycleConfiguration
return
}
var lifeCycle = NewLifecycleConfiguration()
lifeCycle.Rules = lcConf.Rules
var data []byte
data, err = xml.Marshal(lifeCycle)
if err != nil {
log.LogErrorf("getBucketLifecycle failed: bucket[%v] err(%v)", param.Bucket(), err)
errorCode = NoSuchLifecycleConfiguration
return
}
writeSuccessResponseXML(w, data)
}
// API reference: https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/API/API_PutBucketLifecycleConfiguration.html
func (o *ObjectNode) putBucketLifecycleConfigurationHandler(w http.ResponseWriter, r *http.Request) {
var err error
var errorCode *ErrorCode
defer func() {
o.errorResponse(w, r, err, errorCode)
}()
param := ParseRequestParam(r)
if param.Bucket() == "" {
errorCode = InvalidBucketName
return
}
if _, err = o.vm.Volume(param.Bucket()); err != nil {
errorCode = NoSuchBucket
return
}
_, errorCode = VerifyContentLength(r, BodyLimit)
if errorCode != nil {
return
}
var requestBody []byte
if requestBody, err = io.ReadAll(r.Body); err != nil && err != io.EOF {
log.LogErrorf("putBucketLifecycle failed: read request body data err: requestID(%v) err(%v)", GetRequestID(r), err)
errorCode = &ErrorCode{
ErrorCode: http.StatusText(http.StatusBadRequest),
ErrorMessage: err.Error(),
StatusCode: http.StatusBadRequest,
}
return
}
var lifeCycle = NewLifecycleConfiguration()
if err = UnmarshalXMLEntity(requestBody, lifeCycle); err != nil {
log.LogWarnf("putBucketLifecycle failed: decode request body err: requestID(%v) err(%v)", GetRequestID(r), err)
errorCode = LifeCycleErrMalformedXML
return
}
ok, errorCode := lifeCycle.Validate()
if !ok {
log.LogErrorf("putBucketLifecycle failed: validate err: requestID(%v) lifeCycle(%v) err(%v)", GetRequestID(r), lifeCycle, errorCode)
return
}
req := proto.LcConfiguration{
VolName: param.Bucket(),
Rules: lifeCycle.Rules,
}
if err = o.mc.AdminAPI().SetBucketLifecycle(&req); err != nil {
log.LogErrorf("putBucketLifecycle failed: SetBucketLifecycle err: bucket[%v] err(%v)", param.Bucket(), err)
return
}
log.LogInfof("putBucketLifecycle success: requestID(%v) volume(%v) lifeCycle(%v)",
GetRequestID(r), param.Bucket(), lifeCycle)
}
// API reference: https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/API/API_DeleteBucketLifecycle.html
func (o *ObjectNode) deleteBucketLifecycleConfigurationHandler(w http.ResponseWriter, r *http.Request) {
var err error
var errorCode *ErrorCode
defer func() {
o.errorResponse(w, r, err, errorCode)
}()
param := ParseRequestParam(r)
if param.Bucket() == "" {
errorCode = InvalidBucketName
return
}
if _, err = o.vm.Volume(param.Bucket()); err != nil {
errorCode = NoSuchBucket
return
}
if err = o.mc.AdminAPI().DelBucketLifecycle(param.Bucket()); err != nil {
log.LogErrorf("deleteBucketLifecycle failed: bucket[%v] err(%v)", param.Bucket(), err)
return
}
w.WriteHeader(http.StatusNoContent)
}

1
proto/admin_proto.go
View File

@ -842,6 +842,7 @@ type DataNodeHeartbeatResponse struct {
Status uint8
Result string
AllDisks []string
DiskStats []DiskStat
BadDisks []string // Keep this old field for compatibility
BadDiskStats []BadDiskStat // key: disk path
CpuUtil float64 `json:"cpuUtil"`

11
proto/model.go
View File

@ -589,3 +589,14 @@ type BadDiskRecoverProgress struct {
BadDataPartitionsNum int
Status string
}
type BadDiskInfo struct {
Address string
Path string
TotalPartitionCnt int
DiskErrPartitionList []uint64
}
type BadDiskInfos struct {
BadDisks []BadDiskInfo
}

6
sdk/data/stream/extent_handler.go
View File

@ -170,8 +170,10 @@ type ExtentHandler struct {
// Signaled in receiver ONLY to exit *sender*.
doneSender chan struct{}
appendLK sync.Mutex
lastKey proto.ExtentKey
// ver update need alloc new extent
verUpdate chan uint64
appendLK sync.Mutex
lastKey proto.ExtentKey
meetLimitedIoError bool

5
sdk/master/api_admin.go
View File

@ -712,8 +712,8 @@ func (api *AdminAPI) GetQuota(volName string, quotaId string) (quotaInfo *proto.
return quotaInfo, err
}
func (api *AdminAPI) QueryBadDisks() (badDisks *proto.DiskInfos, err error) {
badDisks = &proto.DiskInfos{}
func (api *AdminAPI) QueryBadDisks() (badDisks *proto.BadDiskInfos, err error) {
badDisks = &proto.BadDiskInfos{}
err = api.mc.requestWith(badDisks, newRequest(get, proto.QueryBadDisks).Header(api.h))
return
}
@ -725,6 +725,7 @@ func (api *AdminAPI) QueryDisks(addr string) (disks *proto.DiskInfos, err error)
return
}
func (api *AdminAPI) DiskDetail(addr string, diskPath string) (disk *proto.DiskInfo, err error) {
disk = &proto.DiskInfo{}
err = api.mc.requestWith(disk, newRequest(get, proto.QueryDiskDetail).Header(api.h).

1
sdk/meta/meta.go
View File

@ -15,6 +15,7 @@
package meta
import (
"fmt"
"strings"
"sync"
"syscall"

2
tool/preload/preload.go
View File

@ -82,7 +82,7 @@ func main() {
os.Exit(1)
}
if !cli.CheckColdVolume() {
if !cli.CheckVolumeInfoFromMaster() {
fmt.Println("Preload only work in cold volume")
os.Exit(1)
}

22
util/buf/buffer_pool.go
View File

@ -211,16 +211,11 @@ func (bufferP *BufferPool) getHeadProtoVer(id uint64) (data []byte) {
}
}
func (bufferP *BufferPool) getHeadProtoVer(id uint64) (data []byte) {
select {
case data = <-bufferP.headProtoVerPools[id%slotCnt]:
return
default:
return bufferP.headProtoVerPool.Get().([]byte)
}
}
func (bufferP *BufferPool) getNormal(id uint64) (data []byte) {
if NormalBuffersTotalLimit != InvalidLimit && atomic.LoadInt64(&normalBuffersCount) >= NormalBuffersTotalLimit {
ctx := context.Background()
normalBuffersRateLimit.Wait(ctx)
}
select {
case data = <-bufferP.normalPools[id%slotCnt]:
return
@ -308,15 +303,6 @@ func (bufferP *BufferPool) putHeadProtoVer(index int, data []byte) {
}
}
func (bufferP *BufferPool) putHeadProtoVer(index int, data []byte) {
select {
case bufferP.headProtoVerPools[index] <- data:
return
default:
bufferP.headProtoVerPool.Put(data)
}
}
func (bufferP *BufferPool) putNormal(index int, data []byte) {
select {
case bufferP.normalPools[index] <- data:

2
util/log/log.go
View File

@ -35,8 +35,6 @@ import (
"syscall"
"time"
syslog "log"
blog "github.com/cubefs/cubefs/blobstore/util/log"
)

2
util/routinepool/pool_test.go
View File

@ -57,7 +57,7 @@ func TestRoutinePool(t *testing.T) {
func TestNewAndClose(t *testing.T) {
maxRoutineNum := 10
pool := NewRoutinePool(maxRoutineNum)
pool := routinepool.NewRoutinePool(maxRoutineNum)
pool.WaitAndClose()
}