refactor(sdk): remove dp cache code from sdk, client, cli. #1000063488

Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2025-04-24 15:19:09 +08:00
parent 1bcef5b139
commit a819193dcd
52 changed files with 87 additions and 1775 deletions

View File

@ -1,23 +0,0 @@
version: 1
env:
- CGO_ENABLED=0
flags:
- -trimpath
goos: linux
goarch: amd64
# (Optional) Entrypoint to compile.
main: ./preload/preload.go
binary: cfs-preload-{{ .Os }}-{{ .Arch }}
ldflags:
- "-X github.com/cubefs/cubefs/proto.Version={{ .Env.VERSION }}"
- "-X github.com/cubefs/cubefs/proto.CommitID={{ .Env.COMMIT_ID }}"
- "-X github.com/cubefs/cubefs/proto.BranchName={{ .Env.BRANCH_NAME }}"
- "-X github.com/cubefs/cubefs/proto.BuildTime={{ .Env.BUILD_TIME }}"
- "-X github.com/cubefs/cubefs/blobstore/util/version.version={{ .Env.BRANCH_NAME }}/{{ .Env.COMMIT_ID }}"
- "-w -s"

View File

@ -110,24 +110,6 @@ builds:
- -X {{.Env.PROTO}}.BuildTime={{.Date}}
- -X {{.Env.VV}}={{.Branch}}/{{.Date}}
- -w -s
- id: "preload"
main: ./preload
binary: cfs-preload
env:
- CGO_ENABLED=0
goos:
- linux
goarch:
- amd64
flags:
- -trimpath
ldflags:
- -X {{.Env.PROTO}}.Version={{.Version}}
- -X {{.Env.PROTO}}.CommitID={{.FullCommit}}
- -X {{.Env.PROTO}}.BranchName={{.Branch}}
- -X {{.Env.PROTO}}.BuildTime={{.Date}}
- -X {{.Env.VV}}={{.Branch}}/{{.Date}}
- -w -s
- id: "server"
main: ./cmd

View File

@ -8,8 +8,8 @@ default: all
phony := all
all: build
phony += build server authtool client cli libsdkpre libsdk fsck fdstore preload bcache blobstore deploy
build: server authtool client cli libsdk fsck fdstore preload bcache blobstore deploy
phony += build server authtool client cli libsdkpre libsdk fsck fdstore bcache blobstore deploy
build: server authtool client cli libsdk fsck fdstore bcache blobstore deploy
server:
@build/build.sh server $(GOMOD) --threads=$(threads)
@ -43,9 +43,6 @@ libsdk:
fdstore:
@build/build.sh fdstore $(GOMOD) --threads=$(threads)
preload:
@build/build.sh preload $(GOMOD) --threads=$(threads)
bcache:
@build/build.sh bcache $(GOMOD) --threads=$(threads)

View File

@ -254,7 +254,7 @@ run_test() {
export JENKINS_TEST=1
ulimit -n 65536
echo -n "${TPATH}"
go test -cover -v -coverprofile=cover.output $(go list ./... | grep -v depends | grep master) | tee cubefs_unittest.output
go test -cover -v -coverprofile=cover.output $(go list ./... | grep -v depends) | tee cubefs_unittest.output
ret=$?
popd >/dev/null
exit $ret
@ -415,12 +415,6 @@ build_fdstore() {
popd >/dev/null
}
build_preload() {
pushd $SrcPath >/dev/null
echo -n "build cfs-preload "
CGO_ENABLED=0 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-preload ${SrcPath}/tool/preload/*.go && echo "success" || echo "failed"
}
build_bcache(){
pushd $SrcPath >/dev/null
echo -n "build cfs-blockcache "
@ -524,9 +518,6 @@ case "$cmd" in
"fdstore")
build_fdstore
;;
"preload")
build_preload
;;
"bcache")
build_bcache
;;

View File

@ -95,13 +95,7 @@ const (
CliFlagMaximallyRead = "maximally-read"
CliFlagCacheRuleKey = "cache-rule-key"
CliFlagEbsBlkSize = "ebs-blk-size"
CliFlagCacheCapacity = "cache-capacity"
CliFlagCacheAction = "cache-action"
CliFlagCacheThreshold = "cache-threshold"
CliFlagCacheTTL = "cache-ttl"
CliFlagCacheHighWater = "cache-high-water"
CliFlagCacheLowWater = "cache-low-water"
CliFlagCacheLRUInterval = "cache-lru-interval"
CliFlagCacheRule = "cache-rule"
CliFlagThreshold = "threshold"
CliFlagAddress = "addr"

View File

@ -282,17 +282,10 @@ func formatSimpleVolView(svv *proto.SimpleVolView) string {
}
sb.WriteString(fmt.Sprintf(" VolStorageClass : %v\n", proto.StorageClassString(svv.VolStorageClass)))
sb.WriteString(fmt.Sprintf(" AllowedStorageClass : %v\n", allowedStorageClassStr))
sb.WriteString(fmt.Sprintf(" CacheDpStorageClass : %v\n", proto.StorageClassString(svv.CacheDpStorageClass)))
if svv.VolType == 1 {
sb.WriteString(fmt.Sprintf(" ObjBlockSize : %v byte\n", svv.ObjBlockSize))
sb.WriteString(fmt.Sprintf(" CacheCapacity : %v G\n", svv.CacheCapacity))
sb.WriteString(fmt.Sprintf(" CacheAction : %v\n", svv.CacheAction))
sb.WriteString(fmt.Sprintf(" CacheThreshold : %v byte\n", svv.CacheThreshold))
sb.WriteString(fmt.Sprintf(" CacheLruInterval : %v min\n", svv.CacheLruInterval))
sb.WriteString(fmt.Sprintf(" CacheTtl : %v day\n", svv.CacheTtl))
sb.WriteString(fmt.Sprintf(" CacheLowWater : %v\n", svv.CacheLowWater))
sb.WriteString(fmt.Sprintf(" CacheHighWater : %v\n", svv.CacheHighWater))
sb.WriteString(fmt.Sprintf(" CacheRule : %v\n", svv.CacheRule))
}

View File

@ -107,13 +107,7 @@ const (
cmdVolDefaultBusiness = ""
cmdVolDefaultCacheRuleKey = ""
cmdVolDefaultEbsBlkSize = 8 * 1024 * 1024
cmdVolDefaultCacheCapacity = 0
cmdVolDefaultCacheAction = 0
cmdVolDefaultCacheThreshold = 10 * 1024 * 1024
cmdVolDefaultCacheTTL = 30
cmdVolDefaultCacheHighWater = 80
cmdVolDefaultCacheLowWater = 60
cmdVolDefaultCacheLRUInterval = 5
cmdVolDefaultDpReadOnlyWhenVolFull = "false"
cmdVolDefaultAllowedStorageClass = ""
cmdVolMinRemoteCacheTTL = 10 * 60
@ -138,13 +132,7 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
var optZoneName string
var optCacheRuleKey string
var optEbsBlkSize int
var optCacheCap int
var optCacheAction int
var optCacheThreshold int
var optCacheTTL int
var optCacheHighWater int
var optCacheLowWater int
var optCacheLRUInterval int
var optDpReadOnlyWhenVolFull string
var optEnableQuota string
var optTxMask string
@ -265,13 +253,7 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
stdout(" zoneName : %v\n", optZoneName)
stdout(" cacheRuleKey : %v\n", optCacheRuleKey)
stdout(" ebsBlkSize : %v byte\n", optEbsBlkSize)
stdout(" cacheCapacity : %v G\n", optCacheCap)
stdout(" cacheAction : %v\n", optCacheAction)
stdout(" cacheThreshold : %v byte\n", optCacheThreshold)
stdout(" cacheTTL : %v day\n", optCacheTTL)
stdout(" cacheHighWater : %v\n", optCacheHighWater)
stdout(" cacheLowWater : %v\n", optCacheLowWater)
stdout(" cacheLRUInterval : %v min\n", optCacheLRUInterval)
stdout(" TransactionMask : %v\n", optTxMask)
stdout(" TransactionTimeout : %v min\n", optTxTimeout)
stdout(" TxConflictRetryNum : %v\n", optTxConflictRetryNum)
@ -305,9 +287,7 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
err = client.AdminAPI().CreateVolName(
volumeName, userID, optCapacity, optDeleteLockTime, crossZone, normalZonesFirst, optBusiness,
optMPCount, optDPCount, int(replicaNum), optDPSize, followerRead,
optZoneName, optCacheRuleKey, optEbsBlkSize, optCacheCap,
optCacheAction, optCacheThreshold, optCacheTTL, optCacheHighWater,
optCacheLowWater, optCacheLRUInterval, dpReadOnlyWhenVolFull,
optZoneName, optCacheRuleKey, optEbsBlkSize, optCacheThreshold, dpReadOnlyWhenVolFull,
optTxMask, optTxTimeout, optTxConflictRetryNum, optTxConflictRetryInterval, optEnableQuota, clientIDKey,
optVolStorageClass, optAllowedStorageClass, optMetaFollowerRead, optMaximallyRead,
optRcEnable, optRcAutoPrepare, optRcPath, optRcTTL, optRcReadTimeout, optRemoteCacheMaxFileSizeGB,
@ -334,13 +314,7 @@ func newVolCreateCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&optZoneName, CliFlagZoneName, cmdVolDefaultZoneName, "Specify volume zone name")
cmd.Flags().StringVar(&optCacheRuleKey, CliFlagCacheRuleKey, cmdVolDefaultCacheRuleKey, "Anything that match this field will be written to the cache")
cmd.Flags().IntVar(&optEbsBlkSize, CliFlagEbsBlkSize, cmdVolDefaultEbsBlkSize, "Specify ebsBlk Size[Unit: byte]")
cmd.Flags().IntVar(&optCacheCap, CliFlagCacheCapacity, cmdVolDefaultCacheCapacity, "Specify low volume capacity[Unit: GB]")
cmd.Flags().IntVar(&optCacheAction, CliFlagCacheAction, cmdVolDefaultCacheAction, "Specify low volume cacheAction (default 0)")
cmd.Flags().IntVar(&optCacheThreshold, CliFlagCacheThreshold, cmdVolDefaultCacheThreshold, "Specify cache threshold[Unit: byte]")
cmd.Flags().IntVar(&optCacheTTL, CliFlagCacheTTL, cmdVolDefaultCacheTTL, "Specify cache expiration time[Unit: day]")
cmd.Flags().IntVar(&optCacheHighWater, CliFlagCacheHighWater, cmdVolDefaultCacheHighWater, "")
cmd.Flags().IntVar(&optCacheLowWater, CliFlagCacheLowWater, cmdVolDefaultCacheLowWater, "")
cmd.Flags().IntVar(&optCacheLRUInterval, CliFlagCacheLRUInterval, cmdVolDefaultCacheLRUInterval, "Specify interval expiration time[Unit: min]")
cmd.Flags().StringVar(&optDpReadOnlyWhenVolFull, CliDpReadOnlyWhenVolFull, cmdVolDefaultDpReadOnlyWhenVolFull,
"Enable volume becomes read only when it is full")
cmd.Flags().StringVar(&clientIDKey, CliFlagClientIDKey, client.ClientIDKey(), CliUsageClientIDKey)
@ -386,13 +360,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
var optMaximallyRead string
var optDirectRead string
var optEbsBlkSize int
var optCacheCap string
var optCacheAction string
var optCacheThreshold int
var optCacheTTL int
var optCacheHighWater int
var optCacheLowWater int
var optCacheLRUInterval int
var optDpReadOnlyWhenVolFull string
var clientIDKey string
var optRcEnable string
@ -551,18 +519,6 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
} else {
confirmString.WriteString(fmt.Sprintf(" EbsBlkSize : %v byte\n", vv.ObjBlockSize))
}
if optCacheCap != "" {
if vv.VolStorageClass != proto.StorageClass_BlobStore {
err = fmt.Errorf("cache-capacity can not be set because vol storageClass is not blobstore\n")
return
}
isChange = true
confirmString.WriteString(fmt.Sprintf(" CacheCap : %v GB -> %v GB\n", vv.CacheCapacity, optCacheCap))
intNum, _ := strconv.Atoi(optCacheCap)
vv.CacheCapacity = uint64(intNum)
} else {
confirmString.WriteString(fmt.Sprintf(" CacheCap : %v GB\n", vv.CacheCapacity))
}
if optEnableQuota != "" {
if optEnableQuota == "false" {
@ -676,20 +632,6 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
confirmString.WriteString(fmt.Sprintf(" Tx Operation limit : %v\n", vv.TxOpLimit))
}
if optCacheAction != "" {
if vv.VolStorageClass != proto.StorageClass_BlobStore {
err = fmt.Errorf("cache-action can not be set because vol storageClass is not blobstore\n")
return
}
isChange = true
confirmString.WriteString(fmt.Sprintf(" CacheAction : %v -> %v \n", vv.CacheAction, optCacheAction))
vv.CacheAction, err = strconv.Atoi(optCacheAction)
if err != nil {
return
}
} else {
confirmString.WriteString(fmt.Sprintf(" CacheAction : %v \n", vv.CacheAction))
}
if optCacheRule != "" {
if vv.VolStorageClass != proto.StorageClass_BlobStore {
err = fmt.Errorf("cache-rule can not be set because vol storageClass is not blobstore\n")
@ -699,7 +641,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
confirmString.WriteString(fmt.Sprintf(" CacheRule : %v -> %v \n", vv.CacheRule, optCacheRule))
vv.CacheRule = optCacheRule
} else {
confirmString.WriteString(fmt.Sprintf(" CacheRule : %v \n", vv.CacheAction))
confirmString.WriteString(fmt.Sprintf(" CacheRule : %v \n", vv.CacheRule))
}
if optCacheThreshold > 0 {
if vv.VolStorageClass != proto.StorageClass_BlobStore {
@ -712,50 +654,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
} else {
confirmString.WriteString(fmt.Sprintf(" CacheThreshold : %v byte\n", vv.CacheThreshold))
}
if optCacheTTL > 0 {
if vv.VolStorageClass != proto.StorageClass_BlobStore {
err = fmt.Errorf("cache-ttl can not be set because vol storageClass is not blobstore\n")
return
}
isChange = true
confirmString.WriteString(fmt.Sprintf(" CacheTTL : %v day -> %v day \n", vv.CacheTtl, optCacheTTL))
vv.CacheTtl = optCacheTTL
} else {
confirmString.WriteString(fmt.Sprintf(" CacheTTL : %v day\n", vv.CacheTtl))
}
if optCacheHighWater > 0 {
if vv.VolStorageClass != proto.StorageClass_BlobStore {
err = fmt.Errorf("cache-high-water can not be set because vol storageClass is not blobstore\n")
return
}
isChange = true
confirmString.WriteString(fmt.Sprintf(" CacheHighWater : %v -> %v \n", vv.CacheHighWater, optCacheHighWater))
vv.CacheHighWater = optCacheHighWater
} else {
confirmString.WriteString(fmt.Sprintf(" CacheHighWater : %v \n", vv.CacheHighWater))
}
if optCacheLowWater > 0 {
if vv.VolStorageClass != proto.StorageClass_BlobStore {
err = fmt.Errorf("cache-low-water can not be set because vol storageClass is not blobstore\n")
return
}
isChange = true
confirmString.WriteString(fmt.Sprintf(" CacheLowWater : %v -> %v \n", vv.CacheLowWater, optCacheLowWater))
vv.CacheLowWater = optCacheLowWater
} else {
confirmString.WriteString(fmt.Sprintf(" CacheLowWater : %v \n", vv.CacheLowWater))
}
if optCacheLRUInterval > 0 {
if vv.VolStorageClass != proto.StorageClass_BlobStore {
err = fmt.Errorf("cache-lru-interval can not be set because vol storageClass is not blobstore\n")
return
}
isChange = true
confirmString.WriteString(fmt.Sprintf(" CacheLRUInterval : %v min -> %v min \n", vv.CacheLruInterval, optCacheLRUInterval))
vv.CacheLruInterval = optCacheLRUInterval
} else {
confirmString.WriteString(fmt.Sprintf(" CacheLRUInterval : %v min\n", vv.CacheLruInterval))
}
if optDpReadOnlyWhenVolFull != "" {
isChange = true
var enable bool
@ -1022,14 +921,8 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&optDirectRead, "directRead", "", "Enable read direct from disk (true|false, default false)")
cmd.Flags().StringVar(&optMaximallyRead, CliFlagMaximallyRead, "", "Enable read more hosts (true|false, default false)")
cmd.Flags().IntVar(&optEbsBlkSize, CliFlagEbsBlkSize, 0, "Specify ebsBlk Size[Unit: byte]")
cmd.Flags().StringVar(&optCacheCap, CliFlagCacheCapacity, "", "Specify low volume capacity[Unit: GB]")
cmd.Flags().StringVar(&optCacheAction, CliFlagCacheAction, "", "Specify low volume cacheAction (default 0)")
cmd.Flags().IntVar(&optCacheThreshold, CliFlagCacheThreshold, 0, "Specify cache threshold[Unit: byte] (default 10M)")
cmd.Flags().IntVar(&optCacheTTL, CliFlagCacheTTL, 0, "Specify cache expiration time[Unit: day] (default 30)")
cmd.Flags().IntVar(&optCacheHighWater, CliFlagCacheHighWater, 0, " (default 80)")
cmd.Flags().IntVar(&optCacheLowWater, CliFlagCacheLowWater, 0, " (default 60)")
cmd.Flags().StringVar(&optCacheRule, CliFlagCacheRule, "", "Specify cache rule")
cmd.Flags().IntVar(&optCacheLRUInterval, CliFlagCacheLRUInterval, 0, "Specify interval expiration time[Unit: min] (default 5)")
cmd.Flags().StringVar(&optDpReadOnlyWhenVolFull, CliDpReadOnlyWhenVolFull, "", "Enable volume becomes read only when it is full")
cmd.Flags().BoolVarP(&optYes, "yes", "y", false, "Answer yes for all questions")
cmd.Flags().StringVar(&optTxMask, CliTxMask, "", "Enable transaction for specified operation: \"create|mkdir|remove|rename|mknod|symlink|link\" or \"off\" or \"all\"")

View File

@ -89,7 +89,6 @@ func NewFile(s *Super, i *proto.InodeInfo, flag uint32, pino uint64, filename st
EnableBcache: s.enableBcache,
WConcurrency: s.writeThreads,
ReadConcurrency: s.readThreads,
CacheAction: s.CacheAction,
FileCache: false,
FileSize: i.Size,
CacheThreshold: s.CacheThreshold,
@ -268,7 +267,6 @@ func (f *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenR
EnableBcache: f.super.enableBcache,
WConcurrency: f.super.writeThreads,
ReadConcurrency: f.super.readThreads,
CacheAction: f.super.CacheAction,
FileCache: false,
FileSize: uint64(fileSize),
CacheThreshold: f.super.CacheThreshold,

View File

@ -72,7 +72,6 @@ func (s *Super) InodeGet(ino uint64) (info *proto.InodeInfo, err error) {
EnableBcache: f.super.enableBcache,
WConcurrency: f.super.writeThreads,
ReadConcurrency: f.super.readThreads,
CacheAction: f.super.CacheAction,
FileCache: false,
FileSize: uint64(fileSize),
CacheThreshold: f.super.CacheThreshold,

View File

@ -75,7 +75,6 @@ type Super struct {
// data lake
volType int
ebsEndpoint string
CacheAction int
CacheThreshold int
EbsBlockSize int
enableBcache bool
@ -93,8 +92,6 @@ type Super struct {
taskPool []common.TaskPool
closeC chan struct{}
cacheDpStorageClass uint32
}
// Functions that Super needs to implement
@ -210,7 +207,6 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
s.volType = opt.VolType
s.ebsEndpoint = opt.EbsEndpoint
s.CacheAction = opt.CacheAction
s.CacheThreshold = opt.CacheThreshold
s.EbsBlockSize = opt.EbsBlockSize
s.enableBcache = opt.EnableBcache
@ -222,8 +218,6 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
s.bc = bcache.NewBcacheClient()
}
s.cacheDpStorageClass = opt.VolCacheDpStorageClass
extentConfig := &stream.ExtentConfig{
Volume: opt.Volname,
Masters: masters,
@ -252,7 +246,6 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
OnRenewalForbiddenMigration: s.mw.RenewalForbiddenMigration,
VolStorageClass: opt.VolStorageClass,
VolAllowedStorageClass: opt.VolAllowedStorageClass,
VolCacheDpStorageClass: s.cacheDpStorageClass,
OnForbiddenMigration: s.mw.ForbiddenMigration,
OnGetInodeInfo: s.InodeGet,
@ -324,8 +317,8 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
atomic.StoreUint32((*uint32)(&s.state), uint32(fs.FSStatRestore))
}
log.LogInfof("NewSuper: cluster(%v) volname(%v) icacheExpiration(%v) LookupValidDuration(%v) AttrValidDuration(%v) state(%v) cacheDpStorageClass(%v)",
s.cluster, s.volname, inodeExpiration, LookupValidDuration, AttrValidDuration, s.state, s.cacheDpStorageClass)
log.LogInfof("NewSuper: cluster(%v) volname(%v) icacheExpiration(%v) LookupValidDuration(%v) AttrValidDuration(%v) state(%v)",
s.cluster, s.volname, inodeExpiration, LookupValidDuration, AttrValidDuration, s.state)
stat.PrintModuleStat = func(writer *bufio.Writer) {
fmt.Fprintf(writer, "ic:%d dc:%d nodecache:%d dircache:%d\n", s.ic.lruList.Len(), s.dc.lruList.Len(), len(s.nodeCache), s.mw.DirCacheLen())
}

3
client/fuse.go Normal file → Executable file
View File

@ -821,7 +821,6 @@ func mount(opt *proto.MountOptions) (fsConn *fuse.Conn, super *cfs.Super, err er
}
super.SetTransaction(volumeInfo.EnableTransactionV1, volumeInfo.TxTimeout, volumeInfo.TxConflictRetryNum, volumeInfo.TxConflictRetryInterval)
if proto.IsCold(opt.VolType) || proto.IsStorageClassBlobStore(opt.VolStorageClass) {
super.CacheAction = volumeInfo.CacheAction
super.CacheThreshold = volumeInfo.CacheThreshold
super.EbsBlockSize = volumeInfo.ObjBlockSize
} else if proto.IsVolSupportStorageClass(opt.VolAllowedStorageClass, proto.StorageClass_BlobStore) {
@ -1128,7 +1127,6 @@ func loadConfFromMaster(opt *proto.MountOptions) (err error) {
}
opt.VolType = volumeInfo.VolType
opt.EbsBlockSize = volumeInfo.ObjBlockSize
opt.CacheAction = volumeInfo.CacheAction
opt.CacheThreshold = volumeInfo.CacheThreshold
opt.EnableQuota = volumeInfo.EnableQuota
opt.EnableTransaction = volumeInfo.EnableTransactionV1
@ -1137,7 +1135,6 @@ func loadConfFromMaster(opt *proto.MountOptions) (err error) {
opt.TxConflictRetryInterval = volumeInfo.TxConflictRetryInterval
opt.VolStorageClass = volumeInfo.VolStorageClass
opt.VolAllowedStorageClass = volumeInfo.AllowedStorageClass
opt.VolCacheDpStorageClass = volumeInfo.CacheDpStorageClass
var clusterInfo *proto.ClusterInfo
clusterInfo, err = mc.AdminAPI().GetClusterInfo()

View File

@ -70,7 +70,6 @@ type (
ebsEndpoint string
servicePath string
volType int
cacheAction int
ebsBlockSize int
cacheRuleKey string
cacheThreshold int
@ -81,7 +80,6 @@ type (
// hybrid cloud
volStorageClass uint32
volAllowedStorageClass []uint32
volCacheDpStorageClass uint32
// runtime context
cwd string // current working directory
@ -279,7 +277,6 @@ func (c *Client) Start() (err error) {
DisableMetaCache: true,
VolStorageClass: c.volStorageClass,
VolAllowedStorageClass: c.volAllowedStorageClass,
VolCacheDpStorageClass: c.volCacheDpStorageClass,
OnRenewalForbiddenMigration: mw.RenewalForbiddenMigration,
OnForbiddenMigration: mw.ForbiddenMigration,
MetaWrapper: mw,
@ -1154,7 +1151,6 @@ func (c *Client) allocFD(ino uint64, flags int, mode uint32, fileCache bool, fil
EnableBcache: c.cfg.EnableBcache,
WConcurrency: c.cfg.WriteBlockThread,
ReadConcurrency: c.cfg.ReadBlockThread,
CacheAction: c.cacheAction,
FileCache: fileCache,
FileSize: fileSize,
CacheThreshold: c.cacheThreshold,
@ -1196,13 +1192,10 @@ func (c *Client) loadConfFromMaster(masters []string) (err error) {
}
c.volType = volumeInfo.VolType
c.ebsBlockSize = volumeInfo.ObjBlockSize
c.cacheAction = volumeInfo.CacheAction
c.cacheRuleKey = volumeInfo.CacheRule
c.cacheThreshold = volumeInfo.CacheThreshold
c.volStorageClass = volumeInfo.VolStorageClass
c.volAllowedStorageClass = volumeInfo.AllowedStorageClass
c.volCacheDpStorageClass = volumeInfo.CacheDpStorageClass
var clusterInfo *proto.ClusterInfo
clusterInfo, err = mc.AdminAPI().GetClusterInfo()
if err != nil {

View File

@ -255,7 +255,6 @@ type client struct {
ebsEndpoint string
servicePath string
volType int
cacheAction int
ebsBlockSize int
enableBcache bool
readBlockThread int
@ -272,7 +271,6 @@ type client struct {
enableAudit bool
volStorageClass uint32
volAllowedStorageClass []uint32
cacheDpStorageClass uint32
enableInnerReq bool
// runtime context
@ -1626,7 +1624,6 @@ func (c *client) start() (err error) {
DisableMetaCache: true,
VolStorageClass: c.volStorageClass,
VolAllowedStorageClass: c.volAllowedStorageClass,
VolCacheDpStorageClass: c.cacheDpStorageClass,
OnRenewalForbiddenMigration: mw.RenewalForbiddenMigration,
OnForbiddenMigration: mw.ForbiddenMigration,
MetaWrapper: mw,
@ -1700,7 +1697,6 @@ func (c *client) allocFD(ino uint64, flags, mode uint32, fileCache bool, fileSiz
EnableBcache: c.enableBcache,
WConcurrency: c.writeBlockThread,
ReadConcurrency: c.readBlockThread,
CacheAction: c.cacheAction,
FileCache: fileCache,
FileSize: fileSize,
CacheThreshold: c.cacheThreshold,
@ -1898,12 +1894,10 @@ func (c *client) loadConfFromMaster(masters []string) (err error) {
}
c.volType = volumeInfo.VolType
c.ebsBlockSize = volumeInfo.ObjBlockSize
c.cacheAction = volumeInfo.CacheAction
c.cacheRuleKey = volumeInfo.CacheRule
c.cacheThreshold = volumeInfo.CacheThreshold
c.volStorageClass = volumeInfo.VolStorageClass
c.volAllowedStorageClass = volumeInfo.AllowedStorageClass
c.cacheDpStorageClass = volumeInfo.CacheDpStorageClass
var clusterInfo *proto.ClusterInfo
clusterInfo, err = mc.AdminAPI().GetClusterInfo()

View File

@ -113,7 +113,7 @@ create_cold_volume() {
return
fi
md5=`echo -n ${Owner} | md5sum | cut -d ' ' -f1`
curl -v "http://192.168.0.11:17010/admin/createVol?name=${VolName}&volStorageClass=1&cacheCap=8&cacheAction=2&capacity=10&owner=${Owner}&mpCount=3"
curl -v "http://192.168.0.11:17010/admin/createVol?name=${VolName}&volStorageClass=1&capacity=10&owner=${Owner}&mpCount=3"
curl -v "http://192.168.0.11:17010/client/vol?name=${VolName}&authKey=${md5}" | python -m json.tool
}

View File

@ -34,13 +34,7 @@ CubeFS以 **Owner** 参数作为用户ID。
| zoneName | string | 指定区域 | 否 | 如果 crossZone 设为 false则默认值为 default |
| cacheRuleKey | string | 纠删码卷使用 | 否 | 非空时,匹配该字段的才会写入 cache空 |
| ebsBlkSize | int | 每个块的大小,单位 byte | 否 | 默认8M |
| cacheCap | int | 纠删码卷 cache 容量的大小,单位 GB | 否 | 纠删码卷开启缓存必填 |
| cacheAction | int | 纠删码卷写 cache 的场景0-不写 cache, 1-读数据回写 cache, 2-读写数据都写到 cache | 否 | 0 |
| cacheThreshold | int | 纠删码卷小于该值时,才写入到 cache 中,单位 byte | 否 | 默认10M |
| cacheTTL | int | 纠删码卷 cache 淘汰时间,单位 天 | 否 | 默认30 |
| cacheHighWater | int | 纠删码卷 cache 淘汰的阈值dp 内容量淘汰上水位,达到该值时,触发淘汰 | 否 | 默认80即120G*80/100=96G时dp开始淘汰数据 |
| cacheLowWater | int | dp 上容量淘汰下水位,达到该值时,不再淘汰 | 否 | 默认60即120G*60/100=72Gdp不再淘汰数据 |
| cacheLRUInterval | int | 低容量淘汰检测周期,单位 分钟 | 否 | 默认5分钟 |
## 删除
@ -84,14 +78,8 @@ curl -v "http://10.196.59.198:17010/admin/getVol?name=test" | python -m json.too
``` json
{
"Authenticate": false,
"CacheAction": 0,
"CacheCapacity": 0,
"CacheHighWater": 80,
"CacheLowWater": 60,
"CacheLruInterval": 5,
"CacheRule": "",
"CacheThreshold": 10485760,
"CacheTtl": 30,
"Capacity": 10,
"CreateTime": "2022-03-31 16:08:31",
"CrossZone": false,

View File

@ -47,8 +47,6 @@ Create a new volume:
zoneName :
cacheRuleKey :
ebsBlkSize : 8388608 byte
cacheCapacity : 0 G
cacheAction : 0
cacheThreshold : 10485760 byte
cacheTTL : 30 day
cacheHighWater : 80

View File

@ -77,28 +77,4 @@ curl -v "http://127.0.0.1:17010/vol/update?name=test&cacheCap=100&cacheAction=1&
## 混合云云上节点做缓存
在混合云 ML 场景,为保证数据安全性和一致性,通常会将训练数据保存在私有云,公有云上的计算节点通过专线或公网访问私有云上的数据。这种跨云数据读写方式会导致较高的读写延时和较大的带宽开销,同时训练耗时更长也会导致算力资源浪费。可通过 CubeFS 的本地缓存及分布式缓存机制,将训练数据缓存到公有云节点,减少数据的跨云数据传输,从而提升训练迭代效率。
![Architecture](./pic/cfs-bache-hybridcloud.png)
如果训练任务的数据集存储路径固定,可通过预热的方式,将训练数据提前加载到多副本子系统。
![Architecture](./pic/cfs-preload.png)
``` bash
./cfs-preload -c config.json
```
配置文件中各参数的含义如下表所示:
| 参数 | 类型 | 含义 | 必需 |
|--------------|--------------|--------------------------------------|-----|
| volumeName | string | 预热数据所在卷| 是 |
| masterAddr | string | 预热数据所在集群的master地址| 是 |
| target | string | 预热数据在卷中的存储路径 | 是 |
| logDir | string | 日志存储目录 | 是 |
| logLevel | string | 日志等级| 是 |
| ttl | string | 预热数据的生存周期,单位秒 | 是 |
| action | string | 预热操作: "preload"执行预热操作;"clear"执行预热数据清理 | 是 |
| action | string | 遍历预热数据目录的最大并发数 | 否 |
| action | string | 预热数据文件大小的最大值,低于该值的文件不会被预热 | 否 |
| action | string | 预热数据文件的最大并发数 | 否 |
同时,计算节点可开启本地缓存,将多副本子系统中的已预热的数据再缓存到本地磁盘/内存,以进一步提高数据的访问效率。
![Architecture](./pic/cfs-bache-hybridcloud.png)

View File

@ -16,30 +16,24 @@ For more information, please refer to: [User Guide](./user.md)
Parameter List
| Parameter | Type | Description | Required | Default Value |
|------------------|--------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------------------------------------------------------------------------------------------------------|
| name | string | Volume name | Yes | None |
| volType | int | Volume type: 0: replica volume, 1: erasure-coded volume | No | 0 |
| capacity | int | Volume quota, in GB | Yes | None |
| owner | string | Volume owner, also the user ID | Yes | None |
| mpCount | int | Number of initialized metadata shards | No | 3 |
| dpCount | int | Number of initialized data shards | No | default 10, maximum limit 200 |
| replicaNum | int | Number of replicas | No | 3 for replica volume (supports 1, 3), 1 for erasure-coded volume (supports 1-16) |
| dpSize | int | Maximum data shard size, in GB | No | 120 |
| enablePosixAcl | bool | Whether to configure POSIX permission restrictions | No | false |
| followerRead | bool | Whether to allow reading data from followers, true by default for erasure-coded volume. If set to true, the client also needs to configure this field to true | No | false |
| crossZone | bool | Whether to cross regions. If set to true, the zoneName parameter cannot be set | No | false |
| normalZonesFirst | bool | Whether to prioritize writing to normal domains | No | false |
| zoneName | string | Specify the region | No | default if crossZone is set to false |
| cacheRuleKey | string | Used for erasure-coded volume | No | Only data matching this field will be written to the cache if it is not empty |
| ebsBlkSize | int | Size of each block, in bytes | No | Default 8M |
| cacheCap | int | Size of the erasure-coded volume cache, in GB | No | Required if the cache is enabled for the erasure-coded volume |
| cacheAction | int | The scenario for writing the erasure-coded volume cache: 0 - do not write to the cache, 1 - read data and write back to the cache, 2 - read and write data to the cache | No | 0 |
| cacheThreshold | int | The minimum size of data to be written to the cache, in bytes | No | Default 10M |
| cacheTTL | int | The erasure-coded volume cache eviction time, in days | No | Default 30 |
| cacheHighWater | int | The threshold for erasure-coded volume cache eviction, the upper limit of the content to be evicted, when it reaches this value, the eviction is triggered | No | Default 80, i.e., when the content of dp reaches 96G (120G * 80/100), the dp starts to evict data |
| cacheLowWater | int | The lower limit of the capacity to be evicted when it reaches this value, the dp will no longer evict data | No | Default 60, i.e., when the content of dp reaches 72G (120G * 60/100), the dp will no longer evict data |
| cacheLRUInterval | int | The detection cycle for low-capacity eviction, in minutes | No | Default 5 minutes |
| Parameter | Type | Description | Required | Default Value |
| ---------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | -------------------------------------------------------------------------------- |
| name | string | Volume name | Yes | None |
| volType | int | Volume type: 0: replica volume, 1: erasure-coded volume | No | 0 |
| capacity | int | Volume quota, in GB | Yes | None |
| owner | string | Volume owner, also the user ID | Yes | None |
| mpCount | int | Number of initialized metadata shards | No | 3 |
| dpCount | int | Number of initialized data shards | No | default 10, maximum limit 200 |
| replicaNum | int | Number of replicas | No | 3 for replica volume (supports 1, 3), 1 for erasure-coded volume (supports 1-16) |
| dpSize | int | Maximum data shard size, in GB | No | 120 |
| enablePosixAcl | bool | Whether to configure POSIX permission restrictions | No | false |
| followerRead | bool | Whether to allow reading data from followers, true by default for erasure-coded volume. If set to true, the client also needs to configure this field to true | No | false |
| crossZone | bool | Whether to cross regions. If set to true, the zoneName parameter cannot be set | No | false |
| normalZonesFirst | bool | Whether to prioritize writing to normal domains | No | false |
| zoneName | string | Specify the region | No | default if crossZone is set to false |
| cacheRuleKey | string | Used for erasure-coded volume | No | Only data matching this field will be written to the cache if it is not empty |
| ebsBlkSize | int | Size of each block, in bytes | No | Default 8M |
| cacheThreshold | int | The minimum size of data to be written to the cache, in bytes | No | Default 10M |
## Delete
@ -59,7 +53,7 @@ When deleting a volume, all permission information related to the volume will be
Parameter List
| Parameter | Type | Description |
|-----------|--------|----------------------------------------------------------------------------------------|
| --------- | ------ | -------------------------------------------------------------------------------------- |
| name | string | Volume name |
| authKey | string | Calculate the 32-bit MD5 value of the owner field of vol as authentication information |
@ -74,7 +68,7 @@ Displays basic information about the volume, including the name of the volume, i
Parameter List
| Parameter | Type | Description |
|-----------|--------|-------------|
| --------- | ------ | ----------- |
| name | string | Volume name |
Response Example
@ -82,14 +76,8 @@ Response Example
``` json
{
"Authenticate": false,
"CacheAction": 0,
"CacheCapacity": 0,
"CacheHighWater": 80,
"CacheLowWater": 60,
"CacheLruInterval": 5,
"CacheRule": "",
"CacheThreshold": 10485760,
"CacheTtl": 30,
"Capacity": 10,
"CreateTime": "2022-03-31 16:08:31",
"CrossZone": false,
@ -131,7 +119,7 @@ Displays information about all data shards of the volume.
Parameter List
| Parameter | Type | Description |
|-----------|--------|-------------|
| --------- | ------ | ----------- |
| name | string | Volume name |
Response Example
@ -163,7 +151,7 @@ Displays information about all metadata shards of the volume.
Parameter List
| Parameter | Type | Description |
|-----------|--------|-------------|
| --------- | ------ | ----------- |
| name | string | Volume name |
Response Example
@ -198,7 +186,7 @@ Displays the total space size, used space size, and whether read-write token con
Parameter List
| Parameter | Type | Description |
|-----------|--------|---------------------------------------------------------------------------------------------------------------------------------------|
| --------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------- |
| name | string | Volume name |
| version | int | Volume version, 0: replica volume, 1: erasure-coded volume, default 0 for replica volume, required for accessing erasure-coded volume |
@ -228,25 +216,19 @@ Increases the quota of the volume and adjusts other related parameters.
Parameter List
| Parameter | Type | Description | Required |
|------------------|--------|----------------------------------------------------------------------------------------------------------------------------------|----------|
| name | string | Volume name | Yes |
| description | string | Volume description information | No |
| authKey | string | Calculate the 32-bit MD5 value of the owner field of vol as authentication information | Yes |
| capacity | int | Update the datanode capacity of the volume, in GB. The replica volume cannot be less than the used capacity | No |
| zoneName | string | The region where the volume is located after the update. If not set, it will be updated to the default region | Yes |
| followerRead | bool | Whether to allow reading data from followers | No |
| enablePosixAcl | bool | Whether to configure POSIX permission restrictions | No |
| emptyCacheRule | string | Whether to empty the cacheRule | No |
| cacheRuleKey | string | Cache rule, used for erasure-coded volume. Only data that meets the corresponding rule will be cached | No |
| ebsBlkSize | int | The size of each block of the erasure-coded volume | No |
| cacheCap | int | The capacity of the cache when the erasure-coded volume uses the secondary cache | No |
| cacheAction | int | For erasure-coded volume, 0: do not write to the cache, 1: read data and write to the cache, 2: read and write data to the cache | No |
| cacheThreshold | int | The size limit of the cached file. Only files smaller than this value will be written to the cache | No |
| cacheTTL | int | Cache expiration time, in days | No |
| cacheHighWater | int | Eviction high water mark | No |
| cacheLowWater | int | Cache eviction low water mark | No |
| cacheLRUInterval | int | Cache detection cycle, in minutes | No |
| Parameter | Type | Description | Required |
| -------------- | ------ | ------------------------------------------------------------------------------------------------------------- | -------- |
| name | string | Volume name | Yes |
| description | string | Volume description information | No |
| authKey | string | Calculate the 32-bit MD5 value of the owner field of vol as authentication information | Yes |
| capacity | int | Update the datanode capacity of the volume, in GB. The replica volume cannot be less than the used capacity | No |
| zoneName | string | The region where the volume is located after the update. If not set, it will be updated to the default region | Yes |
| followerRead | bool | Whether to allow reading data from followers | No |
| enablePosixAcl | bool | Whether to configure POSIX permission restrictions | No |
| emptyCacheRule | string | Whether to empty the cacheRule | No |
| cacheRuleKey | string | Cache rule, used for erasure-coded volume. Only data that meets the corresponding rule will be cached | No |
| ebsBlkSize | int | The size of each block of the erasure-coded volume | No |
| cacheThreshold | int | The size limit of the cached file. Only files smaller than this value will be written to the cache | No |
## Get Volume List
@ -259,7 +241,7 @@ Gets a list of all volumes, filtered by keyword.
Parameter List
| Parameter | Type | Description | Required |
|-----------|--------|--------------------------------------------------------------------|----------|
| --------- | ------ | ------------------------------------------------------------------ | -------- |
| keywords | string | Get the information of the volume whose name contains this keyword | No |
Response Example
@ -296,7 +278,7 @@ Expands the specified volume to the specified capacity.
Parameter List
| Parameter | Type | Description | Required |
|-----------|--------|----------------------------------------------------------------------------------------|----------|
| --------- | ------ | -------------------------------------------------------------------------------------- | -------- |
| name | string | Volume name | Yes |
| authKey | string | Calculate the 32-bit MD5 value of the owner field of vol as authentication information | Yes |
| capacity | int | The quota of the volume after expansion, in GB | Yes |
@ -312,7 +294,7 @@ Reduces the specified volume to the specified capacity.
Parameter List
| Parameter | Type | Description | Required |
|-----------|--------|----------------------------------------------------------------------------------------|----------|
| --------- | ------ | -------------------------------------------------------------------------------------- | -------- |
| name | string | Volume name | Yes |
| authKey | string | Calculate the 32-bit MD5 value of the owner field of vol as authentication information | Yes |
| capacity | int | The quota of the volume after compression, in GB | Yes |
@ -327,11 +309,11 @@ Enable/Disable trash feature for the specified volume.
Parameter List
| Parameter | Type | Description | Required |
|-----------|--------|----------------------------------------------------------------------------------------|----------|
| name | string | Volume name | Yes |
| authKey | string | Calculate the 32-bit MD5 value of the owner field of vol as authentication information | Yes |
| trashInterval | int | The time interval for cleaning expired data in the trash is specified in minutes. A value of 0 indicates that the trash is disabled, while any other positive value indicates that the trash is enabled. | Yes
| Parameter | Type | Description | Required |
| ------------- | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- |
| name | string | Volume name | Yes |
| authKey | string | Calculate the 32-bit MD5 value of the owner field of vol as authentication information | Yes |
| trashInterval | int | The time interval for cleaning expired data in the trash is specified in minutes. A value of 0 indicates that the trash is disabled, while any other positive value indicates that the trash is enabled. | Yes |
## Two Replicas

View File

@ -46,13 +46,7 @@ Create a new volume:
zoneName :
cacheRuleKey :
ebsBlkSize : 8388608 byte
cacheCapacity : 0 G
cacheAction : 0
cacheThreshold : 10485760 byte
cacheTTL : 30 day
cacheHighWater : 80
cacheLowWater : 60
cacheLRUInterval : 5 min
TransactionMask :
TransactionTimeout : 1 min
TxConflictRetryNum : 0

View File

@ -79,25 +79,5 @@ curl -v "http://127.0.0.1:17010/vol/update?name=test&cacheCap=100&cacheAction=1&
In hybrid cloud ML scenarios, to ensure data security and consistency, training data is usually stored in private cloud, and the computing nodes in public cloud access the data on the private cloud through dedicated lines or public networks. Such cross-cloud data reading and writing approach is prone to high latency and large bandwidth overhead, while longer training time can also lead to wasted computational resources. By using CubeFS's local and distributed cache mechanisms, training data can be cached on public cloud nodes, reducing cross-cloud data transmission and improving training iteration efficiency.
![Architecture](./pic/cfs-bache-hybridcloud.png)
If the storage path of the training data set is fixed, the training data can be preloaded into the replicate subsystem through warm-up to improve the training performance.
![Architecture](./pic/cfs-preload.png)
``` bash
./cfs-preload -c config.json
```
The meanings of each parameter in the configuration file are as shown in the following table:
| Parameter | Type | Meaning | Required |
|--------------|--------------|--------------------------------------|-----|
| volumeName | string | Name of the volume where preload data exists| Yes |
| masterAddr | string | Master address of the cluster where preload data resides| Yes |
| target | string | Storage path of the preload data within the volume | Yes |
| logDir | string | Directory for storing logs | Yes |
| logLevel | string | Log level| Yes |
| ttl | string | TTL for preload data in seconds | Yes |
| action | string | Preload action: "preload" for preloading operation; "clear" for preload data clearing | Yes |
| action | string | Maximum concurrency for traversing the preload data directory | No |
| action | string | Maximum file size for preheating, files smaller than this size will not be preloaded | No |
| action | string | Maximum concurrency for preloading data files | No |
In addition, computing nodes can enable local caching to further improve data access efficiency by caching the preloaded data from the replica subsystem to local disk/memory.

View File

@ -36,7 +36,6 @@ The response is as follows:
"adminclusterstat": "/cluster/stat",
"admincreatedatapartition": "/dataPartition/create",
"admincreatemetapartition": "/metaPartition/create",
"admincreatepreloaddatapartition": "/dataPartition/createPreLoad",
"admincreatevol": "/admin/createVol",
"admindatapartitionchangeleader": "/dataPartition/changeleader",
"admindecommissiondatapartition": "/dataPartition/decommission",

View File

@ -374,7 +374,6 @@ func (f *FlashNode) getValidViewInfo(req *proto.FlashNodeManualTaskRequest) (met
OnRenewalForbiddenMigration: metaWrapper.RenewalForbiddenMigration,
VolStorageClass: volumeInfo.VolStorageClass,
VolAllowedStorageClass: volumeInfo.AllowedStorageClass,
VolCacheDpStorageClass: volumeInfo.CacheDpStorageClass,
OnForbiddenMigration: metaWrapper.ForbiddenMigration,
MetaWrapper: metaWrapper,
NeedRemoteCache: true,

View File

@ -147,7 +147,6 @@ func NewS3Scanner(adminTask *proto.AdminTask, l *LcNode) (*LcScanner, error) {
OnRenewalForbiddenMigration: metaWrapper.RenewalForbiddenMigration,
VolStorageClass: volumeInfo.VolStorageClass,
VolAllowedStorageClass: volumeInfo.AllowedStorageClass,
VolCacheDpStorageClass: volumeInfo.CacheDpStorageClass,
OnForbiddenMigration: metaWrapper.ForbiddenMigration,
InnerReq: true,
MetaWrapper: metaWrapper,

View File

@ -449,16 +449,11 @@ func parseColdVolUpdateArgs(r *http.Request, vol *Vol) (args *coldVolArgs, err e
if vol.volStorageClass != proto.StorageClass_BlobStore {
log.LogInfof("[parseColdVolUpdateArgs] vol(%v) storageClass(%v) is not blobstore, skip parse cache args",
vol.Name, proto.StorageClassString(vol.volStorageClass))
args.cacheAction = vol.CacheAction
args.cacheThreshold = vol.CacheThreshold
args.cacheRule = vol.CacheRule
return
}
if args.cacheAction, err = extractUintWithDefault(r, cacheActionKey, vol.CacheAction); err != nil {
return
}
if args.cacheThreshold, err = extractUintWithDefault(r, cacheThresholdKey, vol.CacheThreshold); err != nil {
return
}
@ -473,10 +468,6 @@ func parseColdVolUpdateArgs(r *http.Request, vol *Vol) (args *coldVolArgs, err e
args.cacheRule = ""
}
if args.cacheAction < proto.NoCache || args.cacheAction > proto.RWCache {
return nil, fmt.Errorf("cache action is illegal (%d)", args.cacheAction)
}
return
}
@ -732,7 +723,6 @@ func (qos *qosArgs) isArgsWork() bool {
type coldVolArgs struct {
objBlockSize int
cacheAction int
cacheThreshold int
cacheRule string
accessTimeValidInterval int64
@ -776,7 +766,6 @@ type createVolReq struct {
// hybrid cloud
volStorageClass uint32
allowedStorageClass []uint32
cacheDpStorageClass uint32
// remote cache
remoteCacheEnable bool
remoteCacheAutoPrepare bool
@ -791,14 +780,6 @@ type createVolReq struct {
remoteCacheSameRegionTimeout int64
}
func checkCacheAction(action int) error {
if action != proto.NoCache && action != proto.RCache && action != proto.RWCache {
return fmt.Errorf("cache action is not legal, action [%d]", action)
}
return nil
}
func parseColdArgs(r *http.Request) (args coldVolArgs, err error) {
args.cacheRule = extractStr(r, cacheRuleKey)
@ -806,10 +787,6 @@ func parseColdArgs(r *http.Request) (args coldVolArgs, err error) {
return
}
if args.cacheAction, err = extractUint(r, cacheActionKey); err != nil {
return
}
if args.cacheThreshold, err = extractUint(r, cacheThresholdKey); err != nil {
return
}

View File

@ -2701,11 +2701,6 @@ func (m *Server) checkStorageClassForCreateVolReq(req *createVolReq) (err error)
log.LogInfof("[checkStorageClassForCreateVol] create vol(%v), volStorageClass not specified, auto set as: %v",
req.name, proto.StorageClassString(req.volStorageClass))
} else if proto.IsStorageClassBlobStore(req.volStorageClass) {
req.cacheDpStorageClass = m.cluster.GetFastestReplicaStorageClassInCluster(resourceChecker, req.zoneName)
log.LogInfof("[checkStorageClassForCreateVol] create vol(%v) volStorageClass(%v) set cacheDpStorageClass: %v",
req.name, proto.StorageClassString(req.volStorageClass), proto.StorageClassString(req.cacheDpStorageClass))
}
if !proto.IsValidStorageClass(req.volStorageClass) {
@ -2863,10 +2858,6 @@ func (m *Server) checkCreateVolReq(req *createVolReq) (err error) {
args.objBlockSize = defaultEbsBlkSize
}
if err = checkCacheAction(args.cacheAction); err != nil {
return
}
if args.cacheThreshold == 0 {
args.cacheThreshold = defaultCacheThreshold
}
@ -3075,7 +3066,6 @@ func newSimpleView(vol *Vol) (view *proto.SimpleVolView) {
DpReadOnlyWhenVolFull: vol.DpReadOnlyWhenVolFull,
VolType: vol.VolType,
ObjBlockSize: vol.EbsBlkSize,
CacheAction: vol.CacheAction,
CacheThreshold: vol.CacheThreshold,
CacheRule: vol.CacheRule,
TrashInterval: vol.TrashInterval,
@ -3089,7 +3079,6 @@ func newSimpleView(vol *Vol) (view *proto.SimpleVolView) {
EnablePersistAccessTime: vol.EnablePersistAccessTime,
VolStorageClass: vol.volStorageClass,
CacheDpStorageClass: vol.cacheDpStorageClass,
ForbidWriteOpOfProtoVer0: vol.ForbidWriteOpOfProtoVer0.Load(),
QuotaOfStorageClass: quotaOfClass,
@ -5748,7 +5737,6 @@ func volStat(vol *Vol, countByMeta bool) (stat *proto.VolStatInfo) {
stat.TrashInterval = vol.TrashInterval
stat.DefaultStorageClass = vol.volStorageClass
stat.CacheDpStorageClass = vol.cacheDpStorageClass
stat.StatByStorageClass = vol.StatByStorageClass
stat.StatMigrateStorageClass = vol.StatMigrateStorageClass
stat.StatByDpMediaType = vol.StatByDpMediaType

View File

@ -580,13 +580,7 @@ func TestUpdateVol(t *testing.T) {
assert.True(t, view.Authenticate == view2.Authenticate)
assert.True(t, view.FollowerRead == view2.FollowerRead)
assert.True(t, view.ObjBlockSize == view2.ObjBlockSize)
assert.True(t, view.CacheCapacity == view2.CacheCapacity)
assert.True(t, view.CacheAction == view2.CacheAction)
assert.True(t, view.CacheThreshold == view2.CacheThreshold)
assert.True(t, view.CacheTtl == view2.CacheTtl)
assert.True(t, view.CacheHighWater == view2.CacheHighWater)
assert.True(t, view.CacheLowWater == view2.CacheLowWater)
assert.True(t, view.CacheLruInterval == view2.CacheLruInterval)
assert.True(t, view.CacheRule == view2.CacheRule)
// update
@ -602,7 +596,6 @@ func TestUpdateVol(t *testing.T) {
checkParam(authenticateKey, proto.AdminUpdateVol, req, "tt", true, t)
checkParam(followerReadKey, proto.AdminUpdateVol, req, "test", true, t)
checkParam(ebsBlkSizeKey, proto.AdminUpdateVol, req, "-1", blkSize, t)
checkParam(cacheActionKey, proto.AdminUpdateVol, req, "3", proto.RWCache, t)
checkParam(cacheThresholdKey, proto.AdminUpdateVol, req, "-1", threshold, t)
setParam(cacheRuleKey, proto.AdminUpdateVol, req, rule, t)
checkParam("remoteCacheEnable", proto.AdminUpdateVol, req, "not-bool", true, t)
@ -620,7 +613,6 @@ func TestUpdateVol(t *testing.T) {
// LF vol always be true
assert.True(t, view.FollowerRead)
assert.True(t, view.ObjBlockSize == blkSize)
assert.True(t, view.CacheAction == proto.RWCache)
assert.True(t, view.CacheThreshold == threshold)
assert.True(t, view.CacheRule == rule)
require.True(t, view.RemoteCacheEnable)

View File

@ -3926,7 +3926,6 @@ func (c *Cluster) doCreateVol(req *createVolReq) (vol *Vol, err error) {
VolType: req.volType,
EbsBlkSize: req.coldArgs.objBlockSize,
CacheAction: req.coldArgs.cacheAction,
CacheThreshold: req.coldArgs.cacheThreshold,
CacheRule: req.coldArgs.cacheRule,
@ -3944,7 +3943,6 @@ func (c *Cluster) doCreateVol(req *createVolReq) (vol *Vol, err error) {
VolStorageClass: req.volStorageClass,
AllowedStorageClass: req.allowedStorageClass,
CacheDpStorageClass: req.cacheDpStorageClass,
RemoteCacheEnable: req.remoteCacheEnable,
RemoteCacheAutoPrepare: req.remoteCacheAutoPrepare,

View File

@ -54,7 +54,6 @@ const (
forceDelVolKey = "forceDelVol"
ebsBlkSizeKey = "ebsBlkSize"
cacheActionKey = "cacheAction"
cacheThresholdKey = "cacheThreshold"
clientVersion = "version"
domainIdKey = "domainId"

View File

@ -330,7 +330,6 @@ type volValue struct {
VolType int
EbsBlkSize int
CacheAction int
CacheThreshold int
CacheRule string
@ -359,7 +358,6 @@ type volValue struct {
VolStorageClass uint32
AllowedStorageClass []uint32
CacheDpStorageClass uint32
ForbidWriteOpOfProtoVer0 bool
QuotaOfClass []*proto.StatOfStorageClass
@ -423,7 +421,6 @@ func newVolValue(vol *Vol) (vv *volValue) {
VolType: vol.VolType,
EbsBlkSize: vol.EbsBlkSize,
CacheAction: vol.CacheAction,
CacheThreshold: vol.CacheThreshold,
CacheRule: vol.CacheRule,
VolQosEnable: vol.qosManager.qosEnable,
@ -451,7 +448,6 @@ func newVolValue(vol *Vol) (vv *volValue) {
EnablePersistAccessTime: vol.EnablePersistAccessTime,
VolStorageClass: vol.volStorageClass,
CacheDpStorageClass: vol.cacheDpStorageClass,
ForbidWriteOpOfProtoVer0: vol.ForbidWriteOpOfProtoVer0.Load(),
RemoteCacheEnable: vol.remoteCacheEnable,
@ -1711,7 +1707,6 @@ func (c *Cluster) setStorageClassForLegacyVol(vv *Vol) {
if proto.IsHot(vv.VolType) {
vv.volStorageClass = proto.GetStorageClassByMediaType(c.legacyDataMediaType)
vv.allowedStorageClass = []uint32{vv.volStorageClass}
vv.cacheDpStorageClass = proto.StorageClass_Unspecified
log.LogInfof("legacy vol(%v), set volStorageClass(%v) by cluster LegacyDataMediaType",
vv.Name, proto.StorageClassString(vv.volStorageClass))
return
@ -1719,10 +1714,6 @@ func (c *Cluster) setStorageClassForLegacyVol(vv *Vol) {
vv.volStorageClass = proto.StorageClass_BlobStore
vv.allowedStorageClass = []uint32{vv.volStorageClass}
vv.cacheDpStorageClass = proto.GetStorageClassByMediaType(c.legacyDataMediaType)
log.LogWarnf("legacy cold vol(%v), set cacheDpStorageClass(%v) by cluster LegacyDataMediaType",
vv.Name, proto.StorageClassString(vv.cacheDpStorageClass))
}
func (c *Cluster) loadVols() (err error) {

View File

@ -82,7 +82,6 @@ type VolVarargs struct {
// nolint: structcheck
type CacheSubItem struct {
EbsBlkSize int
CacheAction int
CacheThreshold int
CacheRule string
}
@ -198,7 +197,6 @@ type Vol struct {
// hybrid cloud
allowedStorageClass []uint32 // specifies which storageClasses the vol use, a cluster may have multiple StorageClasses
volStorageClass uint32 // specifies which storageClass is written, unless dirStorageClass is set in file path
cacheDpStorageClass uint32 // for SDK those access cache/preload dp of cold volume
StatByStorageClass []*proto.StatOfStorageClass
StatMigrateStorageClass []*proto.StatOfStorageClass
StatByDpMediaType []*proto.StatOfStorageClass
@ -241,7 +239,6 @@ func newVol(vv volValue) (vol *Vol) {
vol.VolType = vv.VolType
vol.EbsBlkSize = vv.EbsBlkSize
vol.CacheAction = vv.CacheAction
vol.CacheThreshold = vv.CacheThreshold
vol.CacheRule = vv.CacheRule
vol.Status = vv.Status
@ -286,7 +283,6 @@ func newVol(vv volValue) (vol *Vol) {
vol.allowedStorageClass = make([]uint32, len(vv.AllowedStorageClass))
copy(vol.allowedStorageClass, vv.AllowedStorageClass)
vol.volStorageClass = vv.VolStorageClass
vol.cacheDpStorageClass = vv.CacheDpStorageClass
vol.StatByStorageClass = make([]*proto.StatOfStorageClass, 0)
vol.StatMigrateStorageClass = make([]*proto.StatOfStorageClass, 0)
vol.ForbidWriteOpOfProtoVer0.Store(defaultVolForbidWriteOpOfProtoVersion0)
@ -1211,12 +1207,6 @@ func (vol *Vol) needCreateDataPartition() (ok bool, err error) {
return
}
// cold
if vol.CacheAction == proto.NoCache && vol.CacheRule == "" {
err = proto.ErrVolNoCacheAndRule
return
}
ok = true
return
}
@ -1829,7 +1819,6 @@ func setVolFromArgs(args *VolVarargs, vol *Vol) {
if args.volStorageClass == proto.StorageClass_BlobStore {
coldArgs := args.coldArgs
vol.CacheThreshold = coldArgs.cacheThreshold
vol.CacheAction = coldArgs.cacheAction
vol.CacheRule = coldArgs.cacheRule
vol.EbsBlkSize = coldArgs.objBlockSize
}
@ -1869,7 +1858,6 @@ func setVolFromArgs(args *VolVarargs, vol *Vol) {
func getVolVarargs(vol *Vol) *VolVarargs {
args := &coldVolArgs{
objBlockSize: vol.EbsBlkSize,
cacheAction: vol.CacheAction,
cacheThreshold: vol.CacheThreshold,
cacheRule: vol.CacheRule,
accessTimeValidInterval: vol.AccessTimeValidInterval,

View File

@ -128,7 +128,6 @@ func TestCreateColdVol(t *testing.T) {
require.NoError(t, err)
require.EqualValues(t, "", vol.CacheRule)
require.EqualValues(t, defaultEbsBlkSize, vol.EbsBlkSize)
require.EqualValues(t, proto.NoCache, vol.CacheAction)
require.EqualValues(t, vol.CacheThreshold, defaultCacheThreshold)
require.EqualValues(t, 0, vol.dpReplicaNum)
require.True(t, vol.FollowerRead)
@ -143,7 +142,6 @@ func TestCreateColdVol(t *testing.T) {
// check with illegal args
checkCreateVolParam(ebsBlkSizeKey, req, -1, blkSize, t)
checkCreateVolParam(cacheActionKey, req, "3", proto.NoCache, t)
checkCreateVolParam(cacheThresholdKey, req, -1, threshold, t)
checkCreateVolParam(followerReadKey, req, -1, true, t)
@ -152,7 +150,6 @@ func TestCreateColdVol(t *testing.T) {
view := getSimpleVol(volName3, true, t)
assert.True(t, view.ObjBlockSize == blkSize)
assert.True(t, view.CacheThreshold == threshold)
assert.True(t, view.CacheAction == proto.NoCache)
delVol(volName3, t)

View File

@ -117,7 +117,7 @@ func (i *Inode) GetExtents() *SortedExtents {
if proto.IsStorageClassReplica(i.StorageClass) && i.HybridCloudExtents.sortedEks != nil {
return i.HybridCloudExtents.sortedEks.(*SortedExtents)
}
i.HybridCloudExtents.sortedEks = NewSortedExtents()
return i.HybridCloudExtents.sortedEks.(*SortedExtents)
}

View File

@ -142,7 +142,6 @@ type Volume struct {
volType int
ebsBlockSize int
cacheAction int
cacheThreshold int
closeOnce sync.Once
@ -3064,7 +3063,6 @@ func NewVolume(config *VolumeConfig) (*Volume, error) {
OnRenewalForbiddenMigration: metaWrapper.RenewalForbiddenMigration,
VolStorageClass: volumeInfo.VolStorageClass,
VolAllowedStorageClass: volumeInfo.AllowedStorageClass,
VolCacheDpStorageClass: volumeInfo.CacheDpStorageClass,
OnForbiddenMigration: metaWrapper.ForbiddenMigration,
MetaWrapper: metaWrapper,
}
@ -3094,7 +3092,6 @@ func NewVolume(config *VolumeConfig) (*Volume, error) {
createTime: metaWrapper.VolCreateTime(),
volType: volumeInfo.VolType,
ebsBlockSize: volumeInfo.ObjBlockSize,
cacheAction: volumeInfo.CacheAction,
cacheThreshold: volumeInfo.CacheThreshold,
closeCh: make(chan struct{}),
onAsyncTaskError: func(err error) {
@ -3130,7 +3127,6 @@ func (v *Volume) getEbsWriter(ino uint64, storageClass uint32) (writer *blobstor
EnableBcache: enableBlockcache,
WConcurrency: writeThreads,
ReadConcurrency: readThreads,
CacheAction: v.cacheAction,
FileCache: false,
FileSize: 0,
CacheThreshold: v.cacheThreshold,
@ -3155,7 +3151,6 @@ func (v *Volume) getEbsReader(ino uint64, storageClass uint32) (reader *blobstor
EnableBcache: enableBlockcache,
WConcurrency: writeThreads,
ReadConcurrency: readThreads,
CacheAction: v.cacheAction,
FileCache: false,
FileSize: 0,
CacheThreshold: v.cacheThreshold,

View File

@ -44,7 +44,6 @@ const (
AdminGetDataPartition = "/dataPartition/get"
AdminLoadDataPartition = "/dataPartition/load"
AdminCreateDataPartition = "/dataPartition/create"
AdminCreatePreLoadDataPartition = "/dataPartition/createPreLoad"
AdminDecommissionDataPartition = "/dataPartition/decommission"
AdminDiagnoseDataPartition = "/dataPartition/diagnose"
AdminResetDataPartitionDecommissionStatus = "/dataPartition/resetDecommissionStatus"
@ -328,7 +327,6 @@ var GApiInfo map[string]string = map[string]string{
"admingetdatapartition": AdminGetDataPartition,
"adminloaddatapartition": AdminLoadDataPartition,
"admincreatedatapartition": AdminCreateDataPartition,
"admincreatepreloaddatapartition": AdminCreatePreLoadDataPartition,
"admindecommissiondatapartition": AdminDecommissionDataPartition,
"admindiagnosedatapartition": AdminDiagnoseDataPartition,
"admindeletedatareplica": AdminDeleteDataReplica,
@ -1317,19 +1315,12 @@ type SimpleVolView struct {
DpReadOnlyWhenVolFull bool
LeaderRetryTimeout int64
VolType int
ObjBlockSize int
CacheCapacity uint64
CacheAction int
CacheThreshold int
CacheHighWater int
CacheLowWater int
CacheLruInterval int
CacheTtl int
CacheRule string
PreloadCapacity uint64
Uids []UidSimpleInfo
TrashInterval int64
VolType int
ObjBlockSize int
CacheThreshold int
CacheRule string
Uids []UidSimpleInfo
TrashInterval int64
// multi version snapshot
LatestVer uint64
@ -1344,7 +1335,6 @@ type SimpleVolView struct {
// hybrid cloud
VolStorageClass uint32
AllowedStorageClass []uint32
CacheDpStorageClass uint32
ForbidWriteOpOfProtoVer0 bool
QuotaOfStorageClass []*StatOfStorageClass
@ -1464,23 +1454,10 @@ type OpLogView struct {
}
const (
PartitionTypeNormal = 0
PartitionTypeCache = 1
PartitionTypePreLoad = 2
PartitionTypeNormal = 0
PartitionTypeCache = 1
)
func GetDpType(volType int, isPreload bool) int {
if volType == VolumeTypeHot {
return PartitionTypeNormal
}
if isPreload {
return PartitionTypePreLoad
}
return PartitionTypeCache
}
func IsCacheDp(typ int) bool {
return typ == PartitionTypeCache
}
@ -1489,10 +1466,6 @@ func IsNormalDp(typ int) bool {
return typ == PartitionTypeNormal
}
func IsPreLoadDp(typ int) bool {
return typ == PartitionTypePreLoad
}
const (
VolumeTypeInvalid = -1
VolumeTypeHot = 0

View File

@ -306,7 +306,6 @@ type VolStatInfo struct {
DpReadOnlyWhenVolFull bool
TrashInterval int64 `json:"TrashIntervalV2"`
DefaultStorageClass uint32
CacheDpStorageClass uint32
MetaFollowerRead bool
MaximallyRead bool
LeaderRetryTimeOut int

View File

@ -56,7 +56,6 @@ const (
VolType
EbsEndpoint
EbsServerPath
CacheAction
EbsBlockSize
EnableBcache
BcacheDir
@ -160,7 +159,6 @@ func InitMountOptions(opts []MountOption) {
opts[VolType] = MountOption{"volType", "volume type", "", int64(0)}
opts[EbsEndpoint] = MountOption{"ebsEndpoint", "Ebs service address", "", ""}
opts[EbsServerPath] = MountOption{"ebsServerPath", "Ebs service path", "", ""}
opts[CacheAction] = MountOption{"cacheAction", "Cold cache action", "", int64(0)}
opts[EbsBlockSize] = MountOption{"ebsBlockSize", "Ebs object size", "", ""}
// opts[EnableBcache] = MountOption{"enableBcache", "Enable block cache", "", false}
opts[BcacheDir] = MountOption{"bcacheDir", "block cache dir", "", ""}
@ -332,7 +330,6 @@ type MountOptions struct {
VolType int
EbsEndpoint string
EbsServicePath string
CacheAction int
CacheThreshold int
EbsBlockSize int
EnableBcache bool
@ -368,7 +365,6 @@ type MountOptions struct {
// hybrid cloud
VolStorageClass uint32
VolAllowedStorageClass []uint32
VolCacheDpStorageClass uint32
AheadReadEnable bool
AheadReadTotalMem int64

View File

@ -56,17 +56,16 @@ type rwSlice struct {
rSize uint32
read int
Data []byte
extentKey proto.ExtentKey
objExtentKey proto.ObjExtentKey
}
func (s rwSlice) String() string {
return fmt.Sprintf("rwSlice{fileOffset(%v),size(%v),rOffset(%v),rSize(%v),read(%v),extentKey(%v),objExtentKey(%v)}", s.fileOffset, s.size, s.rOffset, s.rSize, s.read, s.extentKey, s.objExtentKey)
return fmt.Sprintf("rwSlice{fileOffset(%v),size(%v),rOffset(%v),rSize(%v),read(%v),objExtentKey(%v)}", s.fileOffset, s.size, s.rOffset, s.rSize, s.read, s.objExtentKey)
}
func (reader *Reader) String() string {
return fmt.Sprintf("Reader{address(%v),volName(%v),volType(%v),ino(%v),fileSize(%v),enableBcache(%v),cacheAction(%v),fileCache(%v),cacheThreshold(%v)},readConcurrency(%v)",
&reader, reader.volName, reader.volType, reader.ino, reader.fileLength, reader.enableBcache, reader.cacheAction, reader.fileCache, reader.cacheThreshold, reader.readConcurrency)
return fmt.Sprintf("Reader{address(%v),volName(%v),volType(%v),ino(%v),fileSize(%v),enableBcache(%v),fileCache(%v),cacheThreshold(%v)},readConcurrency(%v)",
&reader, reader.volName, reader.volType, reader.ino, reader.fileLength, reader.enableBcache, reader.fileCache, reader.cacheThreshold, reader.readConcurrency)
}
type Reader struct {
@ -82,17 +81,16 @@ type Reader struct {
wg sync.WaitGroup
once sync.Once
sync.Mutex
close bool
extentKeys []proto.ExtentKey
objExtentKeys []proto.ObjExtentKey
enableBcache bool
cacheAction int
fileCache bool
cacheThreshold int
fileLength uint64
valid bool
inflightL2cache sync.Map
limitManager *manager.LimitManager
close bool
extentKeys []proto.ExtentKey
objExtentKeys []proto.ObjExtentKey
enableBcache bool
fileCache bool
cacheThreshold int
fileLength uint64
valid bool
inflightCache sync.Map
limitManager *manager.LimitManager
}
type ClientConfig struct {
@ -107,7 +105,6 @@ type ClientConfig struct {
EnableBcache bool
WConcurrency int
ReadConcurrency int
CacheAction int
FileCache bool
FileSize uint64
CacheThreshold int
@ -121,20 +118,15 @@ func NewReader(config ClientConfig) (reader *Reader) {
reader.volType = config.VolType
reader.ino = config.Ino
reader.bc = config.Bc
reader.ec = config.Ec
reader.ebs = config.Ebsc
reader.mw = config.Mw
reader.ec = config.Ec
reader.enableBcache = config.EnableBcache
reader.readConcurrency = config.ReadConcurrency
reader.cacheAction = config.CacheAction
reader.fileCache = config.FileCache
reader.cacheThreshold = config.CacheThreshold
if proto.IsCold(reader.volType) || proto.IsStorageClassBlobStore(config.StorageClass) {
reader.ec.UpdateDataPartitionForColdVolume()
}
reader.limitManager = reader.ec.LimitManager
reader.limitManager = config.Ec.LimitManager
return
}
@ -248,7 +240,6 @@ func (reader *Reader) prepareEbsSlice(offset int, size uint32) ([]*rwSlice, erro
}
if selected {
rs.objExtentKey = oek
reader.buildExtentKey(rs)
rs.Data = make([]byte, rs.rSize)
start = oek.FileOffset + oek.Size
chunks = append(chunks, rs)
@ -262,28 +253,6 @@ func (reader *Reader) prepareEbsSlice(offset int, size uint32) ([]*rwSlice, erro
return chunks, nil
}
func (reader *Reader) buildExtentKey(rs *rwSlice) {
if len(reader.extentKeys) <= 0 {
rs.extentKey = proto.ExtentKey{}
} else {
low := 0
high := len(reader.extentKeys) - 1
for low <= high {
mid := (high + low) / 2
target := reader.extentKeys[mid]
if target.FileOffset == rs.objExtentKey.FileOffset {
rs.extentKey = target
return
} else if target.FileOffset > rs.objExtentKey.FileOffset {
high = mid - 1
} else {
low = mid + 1
}
}
rs.extentKey = proto.ExtentKey{}
}
}
func (reader *Reader) readSliceRange(ctx context.Context, rs *rwSlice) (err error) {
defer reader.wg.Done()
log.LogDebugf("TRACE blobStore readSliceRange Enter. ino(%v) rs.fileOffset(%v),rs.rOffset(%v),rs.rSize(%v) ", reader.ino, rs.fileOffset, rs.rOffset, rs.rSize)
@ -322,32 +291,6 @@ func (reader *Reader) readSliceRange(ctx context.Context, rs *rwSlice) (err erro
}
readLimitOn := false
// read cfs and cache to bcache
if rs.extentKey != (proto.ExtentKey{}) {
// check if dp is exist in preload sence
err = reader.ec.CheckDataPartitionExsit(rs.extentKey.PartitionId)
if err == nil || ctx.Value("objectnode") != nil {
readN, err, readLimitOn = reader.ec.ReadExtent(reader.ino, &rs.extentKey, buf, int(rs.rOffset),
int(rs.rSize), proto.StorageClass_BlobStore)
if err == nil && readN == int(rs.rSize) {
// L2 cache hit.
metric := exporter.NewTPCnt("L2CacheGetHit")
stat.EndStat("CacheHit-L2", nil, bgTime, 1)
defer func() {
metric.SetWithLabels(err, map[string]string{exporter.Vol: reader.volName})
}()
copy(rs.Data, buf)
reader.err <- nil
return
}
} else {
log.LogDebugf("checkDataPartitionExsit failed (%v)", err)
}
log.LogDebugf("TRACE blobStore readSliceRange. cfs block miss.extentKey=%v,err=%v", rs.extentKey, err)
}
if !readLimitOn {
reader.limitManager.ReadAlloc(ctx, int(rs.rSize))
}
@ -361,7 +304,7 @@ func (reader *Reader) readSliceRange(ctx context.Context, rs *rwSlice) (err erro
reader.err <- nil
// cache full block
if !reader.needCacheL1() && !reader.needCacheL2() || reader.ec.IsPreloadMode() {
if !reader.needCacheL1() {
log.LogDebugf("TRACE blobStore readSliceRange exit without cache. read counter=%v", read)
return nil
}
@ -383,12 +326,12 @@ func (reader *Reader) asyncCache(ctx context.Context, cacheKey string, objExtent
log.LogDebugf("TRACE blobStore asyncCache Enter. cacheKey=%v", cacheKey)
// block is go loading.
if _, ok := reader.inflightL2cache.Load(cacheKey); ok {
if _, ok := reader.inflightCache.Load(cacheKey); ok {
return
}
reader.inflightL2cache.Store(cacheKey, true)
defer reader.inflightL2cache.Delete(cacheKey)
reader.inflightCache.Store(cacheKey, true)
defer reader.inflightCache.Delete(cacheKey)
buf := make([]byte, objExtentKey.Size)
read, err := reader.ebs.Read(ctx, reader.volName, buf, 0, uint64(len(buf)), objExtentKey)
@ -398,19 +341,6 @@ func (reader *Reader) asyncCache(ctx context.Context, cacheKey string, objExtent
return
}
if reader.needCacheL2() {
streamer := reader.ec.GetStreamer(reader.ino)
if streamer == nil {
log.LogWarnf("[asyncCache(L2)] streamer for ino %v is nil ", reader.ino)
return
}
reader.ec.Write(reader.ino, int(objExtentKey.FileOffset), buf, proto.FlagsCache, nil, reader.ec.CacheDpStorageClass, false)
log.LogDebugf("TRACE blobStore asyncCache(L2) Exit. storageClass(%v) cacheKey=%v",
proto.StorageClassString(reader.ec.CacheDpStorageClass), cacheKey)
return
}
if reader.needCacheL1() {
reader.bc.Put(reader.volName, cacheKey, buf)
}
@ -418,13 +348,6 @@ func (reader *Reader) asyncCache(ctx context.Context, cacheKey string, objExtent
log.LogDebugf("TRACE blobStore asyncCache(L1) Exit. cacheKey=%v", cacheKey)
}
func (reader *Reader) needCacheL2() bool {
if reader.cacheAction > proto.NoCache && reader.fileLength < uint64(reader.cacheThreshold) || reader.fileCache {
return true
}
return false
}
func (reader *Reader) needCacheL1() bool {
return reader.enableBcache
}

View File

@ -48,7 +48,6 @@ func TestNewReader(t *testing.T) {
EnableBcache: false,
WConcurrency: 0,
ReadConcurrency: 0,
CacheAction: 0,
FileCache: false,
FileSize: 0,
CacheThreshold: 0,
@ -64,37 +63,6 @@ func TestNewReader(t *testing.T) {
assert.NotEmpty(t, reader, nil)
}
func TestBuildExtentKey(t *testing.T) {
testCase := []struct {
eks []proto.ExtentKey
expectEk proto.ExtentKey
}{
{nil, proto.ExtentKey{}},
{[]proto.ExtentKey{
{FileOffset: uint64(0), Size: uint32(100)},
{FileOffset: uint64(100), Size: uint32(100)},
{FileOffset: uint64(200), Size: uint32(100)},
{FileOffset: uint64(300), Size: uint32(100)},
}, proto.ExtentKey{FileOffset: uint64(100), Size: uint32(100)}},
{[]proto.ExtentKey{
{FileOffset: uint64(0), Size: uint32(1)},
{FileOffset: uint64(1), Size: uint32(1)},
{FileOffset: uint64(2), Size: uint32(1)},
{FileOffset: uint64(3), Size: uint32(1)},
}, proto.ExtentKey{}},
}
rs := &rwSlice{}
rs.objExtentKey = proto.ObjExtentKey{FileOffset: 100, Size: 100}
for _, tc := range testCase {
reader := Reader{}
reader.limitManager = manager.NewLimitManager(nil)
reader.extentKeys = tc.eks
reader.buildExtentKey(rs)
assert.Equal(t, tc.expectEk, rs.extentKey)
}
}
func TestFileSize(t *testing.T) {
testCase := []struct {
valid bool
@ -260,7 +228,6 @@ func TestRead(t *testing.T) {
reader.mw = mw
reader.ebs = ebsc
reader.bc = bc
reader.ec = ec
ctx := context.Background()
buf := make([]byte, 500)
@ -324,45 +291,10 @@ func TestAsyncCache(t *testing.T) {
t.Fatalf("Hook advance instance method failed:%s", err.Error())
}
reader.fileLength = tc.fileSize
reader.cacheAction = tc.cacheAction
reader.asyncCache(ctx, "cacheKey", objEk)
}
}
func TestNeedCacheL2(t *testing.T) {
testCase := []struct {
cacheAction int
fileLength uint64
cacheThreshold int
fileCache bool
expectCache bool
}{
{proto.NoCache, 10, 100, false, false},
{proto.NoCache, 10, 100, true, true},
{proto.NoCache, 101, 100, true, true},
{proto.NoCache, 101, 100, false, false},
{proto.RCache, 10, 100, false, true},
{proto.RCache, 10, 100, true, true},
{proto.RCache, 101, 100, false, false},
{proto.RCache, 101, 100, true, true},
{proto.RWCache, 10, 100, false, true},
{proto.RWCache, 10, 100, true, true},
{proto.RWCache, 101, 100, false, false},
{proto.RWCache, 101, 100, true, true},
}
for _, tc := range testCase {
reader := Reader{}
reader.limitManager = manager.NewLimitManager(nil)
reader.cacheAction = tc.cacheAction
reader.fileLength = tc.fileLength
reader.cacheThreshold = tc.cacheThreshold
reader.fileCache = tc.fileCache
got := reader.needCacheL2()
assert.Equal(t, tc.expectCache, got)
}
}
func TestNeedCacheL1(t *testing.T) {
testCase := []struct {
enableCache bool
@ -433,11 +365,11 @@ func TestReadSliceRange(t *testing.T) {
reader.ino = 12407
reader.fileLength = 10
reader.cacheThreshold = 100
reader.ec = ec
reader.err = make(chan error)
rs := &rwSlice{}
rs.rSize = uint32(len("Hello world"))
rs.Data = make([]byte, len("Hello world"))
rs.extentKey = tc.extentKey
reader.enableBcache = tc.enableBcache
err := gohook.HookMethod(ebsc, "Read", tc.ebsReadFunc, nil)
@ -457,7 +389,6 @@ func TestReadSliceRange(t *testing.T) {
panic(fmt.Sprintf("Hook advance instance method failed:%s", err.Error()))
}
reader.ebs = ebsc
reader.ec = ec
reader.bc = bc
ctx := context.Background()

View File

@ -26,13 +26,11 @@ import (
func TestNew(t *testing.T) {
objEks := make([]proto.ObjExtentKey, 0)
eks := make([]proto.ExtentKey, 0)
objEkLen := rand.Intn(20)
expectedFileSize := 0
for i := 0; i < objEkLen; i++ {
size := rand.Intn(1000)
objEks = append(objEks, proto.ObjExtentKey{Size: uint64(size), FileOffset: uint64(expectedFileSize)})
eks = append(eks, proto.ExtentKey{FileOffset: uint64(expectedFileSize), Size: uint32(size)})
expectedFileSize += size
}
@ -46,7 +44,6 @@ func TestNew(t *testing.T) {
rSize: 0,
read: 0,
Data: nil,
extentKey: eks[i],
objExtentKey: objEks[i],
})
}

View File

@ -27,7 +27,6 @@ import (
"github.com/cubefs/cubefs/client/blockcache/bcache"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/data/manager"
"github.com/cubefs/cubefs/sdk/data/stream"
"github.com/cubefs/cubefs/sdk/meta"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/buf"
@ -53,14 +52,12 @@ type Writer struct {
err chan *wSliceErr
bc *bcache.BcacheClient
mw *meta.MetaWrapper
ec *stream.ExtentClient
ebsc *BlobStoreClient
wConcurrency int
wg sync.WaitGroup
once sync.Once
sync.RWMutex
enableBcache bool
cacheAction int
buf []byte
fileOffset int
fileCache bool
@ -81,27 +78,25 @@ func NewWriter(config ClientConfig) (writer *Writer) {
writer.err = nil
writer.bc = config.Bc
writer.mw = config.Mw
writer.ec = config.Ec
writer.ebsc = config.Ebsc
writer.wConcurrency = config.WConcurrency
writer.wg = sync.WaitGroup{}
writer.once = sync.Once{}
writer.RWMutex = sync.RWMutex{}
writer.enableBcache = config.EnableBcache
writer.cacheAction = config.CacheAction
writer.fileCache = config.FileCache
writer.fileSize = config.FileSize
writer.cacheThreshold = config.CacheThreshold
writer.dirty = false
writer.allocateCache()
writer.limitManager = writer.ec.LimitManager
writer.limitManager = config.Ec.LimitManager
return
}
func (writer *Writer) String() string {
return fmt.Sprintf("Writer{address(%v),volName(%v),volType(%v),ino(%v),blockSize(%v),fileSize(%v),enableBcache(%v),cacheAction(%v),fileCache(%v),cacheThreshold(%v)},wConcurrency(%v)",
&writer, writer.volName, writer.volType, writer.ino, writer.blockSize, writer.fileSize, writer.enableBcache, writer.cacheAction, writer.fileCache, writer.cacheThreshold, writer.wConcurrency)
return fmt.Sprintf("Writer{address(%v),volName(%v),volType(%v),ino(%v),blockSize(%v),fileSize(%v),enableBcache(%v),fileCache(%v),cacheThreshold(%v)},wConcurrency(%v)",
&writer, writer.volName, writer.volType, writer.ino, writer.blockSize, writer.fileSize, writer.enableBcache, writer.fileCache, writer.cacheThreshold, writer.wConcurrency)
}
func (writer *Writer) WriteWithoutPool(ctx context.Context, offset int, data []byte) (size int, err error) {
@ -189,22 +184,9 @@ func (writer *Writer) doParallelWrite(ctx context.Context, data []byte, offset i
}
atomic.AddUint64(&writer.fileSize, uint64(size))
for _, wSlice := range wSlices {
writer.cacheLevel2(wSlice)
}
return
}
func (writer *Writer) cacheLevel2(wSlice *rwSlice) {
if writer.cacheAction == proto.RWCache && (wSlice.fileOffset+uint64(wSlice.size)) < uint64(writer.cacheThreshold) || writer.fileCache {
buf := make([]byte, wSlice.size)
offSet := int(wSlice.fileOffset)
copy(buf, wSlice.Data)
go writer.asyncCache(writer.ino, offSet, buf)
}
}
func (writer *Writer) WriteFromReader(ctx context.Context, reader io.Reader, h hash.Hash) (size uint64, err error) {
var (
tmp = buf.ClodVolWriteBufPool.Get().([]byte)
@ -258,8 +240,6 @@ func (writer *Writer) WriteFromReader(ctx context.Context, reader io.Reader, h h
oeksLock.Lock()
oeks = append(oeks, wSlice.objExtentKey)
oeksLock.Unlock()
writer.cacheLevel2(wSlice)
}
exec.Run(write)
@ -500,19 +480,6 @@ func (writer *Writer) writeSlice(ctx context.Context, wSlice *rwSlice, wg bool)
return
}
func (writer *Writer) asyncCache(ino uint64, offset int, data []byte) {
var err error
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("write-async-cache", err, bgTime, 1)
}()
log.LogDebugf("TRACE asyncCache Enter,fileOffset(%v) len(%v), storageClass(%v)",
offset, len(data), proto.StorageClassString(writer.ec.CacheDpStorageClass))
write, err := writer.ec.Write(ino, offset, data, proto.FlagsCache, nil, writer.ec.CacheDpStorageClass, false)
log.LogDebugf("TRACE asyncCache Exit,write(%v) err(%v)", write, err)
}
func (writer *Writer) resetBufferWithoutPool() {
writer.buf = writer.buf[:0]
}
@ -560,8 +527,6 @@ func (writer *Writer) flushWithoutPool(inode uint64, ctx context.Context, flushF
return
}
writer.resetBufferWithoutPool()
writer.cacheLevel2(wSlice)
return
}
@ -603,8 +568,6 @@ func (writer *Writer) flush(inode uint64, ctx context.Context, flushFlag bool) (
return
}
writer.resetBuffer()
writer.cacheLevel2(wSlice)
return
}

View File

@ -60,7 +60,6 @@ func init() {
EnableBcache: false,
WConcurrency: 10,
ReadConcurrency: 10,
CacheAction: 2,
FileCache: false,
FileSize: 0,
CacheThreshold: 0,
@ -207,111 +206,6 @@ func TestParallelWrite(t *testing.T) {
writer.doParallelWrite(ctx, data, offset)
}
func TestCacheL2(t *testing.T) {
testCase := []struct {
cacheAction int
fileOffset uint64
cacheThreshold int
fileCache bool
}{
{proto.NoCache, 0, 10000, false},
{proto.NoCache, 0, 10000, true},
{proto.RCache, 0, 10000, false},
{proto.RWCache, 0, 10000, false},
{proto.RCache, 10001, 10000, false},
{proto.RWCache, 10001, 10000, false},
{proto.RCache, 10001, 10000, true},
{proto.RWCache, 10001, 10000, true},
}
wSlice := &rwSlice{
index: 0,
fileOffset: 0,
size: 1000,
rOffset: 0,
rSize: 1000,
read: 0,
Data: make([]byte, 1000),
extentKey: proto.ExtentKey{},
objExtentKey: proto.ObjExtentKey{},
}
ec := &stream.ExtentClient{}
err := gohook.HookMethod(ec, "Write", MockWriteTrue, nil)
if err != nil {
panic(fmt.Sprintf("Hook advance instance method failed:%s", err.Error()))
}
writer.ec = ec
for _, tc := range testCase {
writer.cacheAction = tc.cacheAction
writer.cacheThreshold = tc.cacheThreshold
writer.fileCache = tc.fileCache
wSlice.fileOffset = tc.fileOffset
writer.cacheLevel2(wSlice)
}
}
func TestWriterAsyncCache(t *testing.T) {
testCase := []struct {
ino uint64
offset int
data []byte
}{
{1, 0, []byte("hello world")},
{2, 10, []byte("hello world")},
}
ec := &stream.ExtentClient{}
err := gohook.HookMethod(ec, "Write", MockWriteTrue, nil)
if err != nil {
panic(fmt.Sprintf("Hook advance instance method failed:%s", err.Error()))
}
writer.ec = ec
for _, tc := range testCase {
writer.asyncCache(tc.ino, tc.offset, tc.data)
}
}
func TestFlush(t *testing.T) {
testCase := []struct {
buf []byte
dirty bool
flushFlag bool
appendObjEkFunc func(*meta.MetaWrapper, uint64, []proto.ObjExtentKey) error
expectError error
}{
{make([]byte, 0), true, true, MockAppendObjExtentKeysTrue, nil},
{make([]byte, 0), false, true, MockAppendObjExtentKeysTrue, nil},
{nil, true, true, MockAppendObjExtentKeysTrue, nil},
{[]byte("hello world"), false, true, MockAppendObjExtentKeysTrue, nil},
{[]byte("hello world"), false, true, MockAppendObjExtentKeysFalse, syscall.EIO},
{[]byte("hello world"), false, false, MockAppendObjExtentKeysTrue, nil},
{[]byte("hello world"), false, false, MockAppendObjExtentKeysFalse, syscall.EIO},
}
ec := &stream.ExtentClient{}
err := gohook.HookMethod(ec, "Write", MockWriteTrue, nil)
if err != nil {
panic(fmt.Sprintf("Hook advance instance method failed:%s", err.Error()))
}
writer.ec = ec
for _, tc := range testCase {
mw := &meta.MetaWrapper{}
err = gohook.HookMethod(mw, "AppendObjExtentKeys", tc.appendObjEkFunc, nil)
if err != nil {
panic(fmt.Sprintf("Hook advance instance method failed:%s", err.Error()))
}
writer.mw = mw
writer.buf = tc.buf
writer.dirty = tc.dirty
ctx := context.Background()
_ = writer.flush(1, ctx, tc.flushFlag)
// assert.Equal(t, tc.expectError, gotError)
}
}
func TestNewWriter(t *testing.T) {
config := ClientConfig{
VolName: "cfs",
@ -325,7 +219,6 @@ func TestNewWriter(t *testing.T) {
EnableBcache: false,
WConcurrency: 10,
ReadConcurrency: 10,
CacheAction: 2,
FileCache: false,
FileSize: 0,
CacheThreshold: 0,
@ -384,7 +277,6 @@ func TestWriteSlice(t *testing.T) {
rSize: 100,
read: 0,
Data: make([]byte, 100),
extentKey: proto.ExtentKey{},
objExtentKey: proto.ObjExtentKey{},
}

View File

@ -156,7 +156,6 @@ type ExtentConfig struct {
VolStorageClass uint32
VolAllowedStorageClass []uint32
VolCacheDpStorageClass uint32
OnGetInodeInfo GetInodeInfoFunc
BcacheOnlyForNotSSD bool
@ -209,7 +208,6 @@ type ExtentClient struct {
multiVerMgr *MultiVerMgr
renewalForbiddenMigration RenewalForbiddenMigrationFunc
forbiddenMigration ForbiddenMigrationFunc
CacheDpStorageClass uint32
getInodeInfo GetInodeInfoFunc
bcacheOnlyForNotSSD bool
AheadRead *AheadReadCache
@ -348,7 +346,6 @@ retry:
client.preload = config.Preload
client.disableMetaCache = config.DisableMetaCache
client.renewalForbiddenMigration = config.OnRenewalForbiddenMigration
client.CacheDpStorageClass = config.VolCacheDpStorageClass
client.forbiddenMigration = config.OnForbiddenMigration
client.getInodeInfo = config.OnGetInodeInfo
@ -951,10 +948,6 @@ func (client *ExtentClient) Close() error {
return nil
}
func (client *ExtentClient) AllocatePreLoadDataPartition(volName string, count int, capacity, ttl uint64, zones string) (err error) {
return client.dataWrapper.AllocatePreLoadDataPartition(volName, count, capacity, ttl, zones)
}
func (client *ExtentClient) CheckDataPartitionExsit(partitionID uint64) error {
_, err := client.dataWrapper.GetDataPartition(partitionID)
return err

View File

@ -428,10 +428,6 @@ func (w *Wrapper) updateDataPartitionByRsp(forceUpdate bool, refreshPolicy Refre
hddDpCount += 1
}
// do not insert preload dp in cold vol
if (proto.IsCold(w.volType) || proto.IsStorageClassBlobStore(w.volStorageClass)) && proto.IsPreLoadDp(dp.PartitionType) {
continue
}
if dp.Status == proto.ReadWrite {
dp.MetricsRefresh()
rwPartitionGroups = append(rwPartitionGroups, dp)
@ -553,35 +549,6 @@ func (w *Wrapper) clearPartitions() {
w.partitions = make(map[uint64]*DataPartition)
}
func (w *Wrapper) AllocatePreLoadDataPartition(volName string, count int, capacity, ttl uint64, zones string) (err error) {
var dpv *proto.DataPartitionsView
if dpv, err = w.mc.AdminAPI().CreatePreLoadDataPartition(volName, count, capacity, ttl, zones); err != nil {
log.LogWarnf("CreatePreLoadDataPartition fail: err(%v)", err)
return
}
convert := func(response *proto.DataPartitionResponse) *DataPartition {
return &DataPartition{
DataPartitionResponse: *response,
ClientWrapper: w,
}
}
rwPartitionGroups := make([]*DataPartition, 0)
for _, partition := range dpv.DataPartitions {
dp := convert(partition)
if (proto.IsCold(w.volType) || proto.IsStorageClassBlobStore(w.volStorageClass)) && !proto.IsPreLoadDp(dp.PartitionType) {
continue
}
log.LogInfof("updateDataPartition: dp(%v)", dp)
w.replaceOrInsertPartition(dp)
dp.MetricsRefresh()
rwPartitionGroups = append(rwPartitionGroups, dp)
}
w.refreshDpSelector(MergeDpPolicy, rwPartitionGroups)
return nil
}
func (w *Wrapper) replaceOrInsertPartition(dp *DataPartition) {
var oldstatus int8
w.Lock.Lock()

View File

@ -298,13 +298,7 @@ func (api *AdminAPI) UpdateVolume(
request.addParam(proto.VolEnableDirectRead, strconv.FormatBool(vv.DirectRead))
request.addParam(proto.MaximallyReadKey, strconv.FormatBool(vv.MaximallyRead))
request.addParam("ebsBlkSize", strconv.Itoa(vv.ObjBlockSize))
request.addParam("cacheCap", strconv.FormatUint(vv.CacheCapacity, 10))
request.addParam("cacheAction", strconv.Itoa(vv.CacheAction))
request.addParam("cacheThreshold", strconv.Itoa(vv.CacheThreshold))
request.addParam("cacheTTL", strconv.Itoa(vv.CacheTtl))
request.addParam("cacheHighWater", strconv.Itoa(vv.CacheHighWater))
request.addParam("cacheLowWater", strconv.Itoa(vv.CacheLowWater))
request.addParam("cacheLRUInterval", strconv.Itoa(vv.CacheLruInterval))
request.addParam("cacheRuleKey", vv.CacheRule)
request.addParam("dpReadOnlyWhenVolFull", strconv.FormatBool(vv.DpReadOnlyWhenVolFull))
request.addParam("replicaNum", strconv.FormatUint(uint64(vv.DpReplicaNum), 10))
@ -396,7 +390,7 @@ func (api *AdminAPI) VolAddAllowedStorageClass(volName string, addAllowedStorage
func (api *AdminAPI) CreateVolName(volName, owner string, capacity uint64, deleteLockTime int64, crossZone, normalZonesFirst bool,
business string, mpCount, dpCount, replicaNum, dpSize int, followerRead bool, zoneName, cacheRuleKey string, ebsBlkSize,
cacheCapacity, cacheAction, cacheThreshold, cacheTTL, cacheHighWater, cacheLowWater, cacheLRUInterval int,
cacheThreshold int,
dpReadOnlyWhenVolFull bool, txMask string, txTimeout uint32, txConflictRetryNum int64, txConflictRetryInterval int64, optEnableQuota string,
clientIDKey string, volStorageClass uint32, allowedStorageClass string, optMetaFollowerRead string, optMaximallyRead string,
remoteCacheEnable string, remoteCacheAutoPrepare string, remoteCachePath string, remoteCacheTTL int64, remoteCacheReadTimeout int64,
@ -421,13 +415,7 @@ func (api *AdminAPI) CreateVolName(volName, owner string, capacity uint64, delet
request.addParam("zoneName", zoneName)
request.addParam("cacheRuleKey", cacheRuleKey)
request.addParam("ebsBlkSize", strconv.Itoa(ebsBlkSize))
request.addParam("cacheCap", strconv.Itoa(cacheCapacity))
request.addParam("cacheAction", strconv.Itoa(cacheAction))
request.addParam("cacheThreshold", strconv.Itoa(cacheThreshold))
request.addParam("cacheTTL", strconv.Itoa(cacheTTL))
request.addParam("cacheHighWater", strconv.Itoa(cacheHighWater))
request.addParam("cacheLowWater", strconv.Itoa(cacheLowWater))
request.addParam("cacheLRUInterval", strconv.Itoa(cacheLRUInterval))
request.addParam("dpReadOnlyWhenVolFull", strconv.FormatBool(dpReadOnlyWhenVolFull))
request.addParam("enableQuota", optEnableQuota)
request.addParam("clientIDKey", clientIDKey)
@ -670,18 +658,6 @@ func (api *AdminAPI) GetClusterParas() (delParas map[string]string, err error) {
return
}
func (api *AdminAPI) CreatePreLoadDataPartition(volName string, count int, capacity, ttl uint64, zongs string) (view *proto.DataPartitionsView, err error) {
view = &proto.DataPartitionsView{}
err = api.mc.requestWith(view, newRequest(get, proto.AdminCreatePreLoadDataPartition).Header(api.h).Param(
anyParam{"name", volName},
anyParam{"replicaNum", count},
anyParam{"capacity", capacity},
anyParam{"cacheTTL", ttl},
anyParam{"zoneName", zongs},
))
return
}
func (api *AdminAPI) ListQuota(volName string) (quotaInfo []*proto.QuotaInfo, err error) {
resp := &proto.ListMasterQuotaResponse{}
if err = api.mc.requestWith(resp, newRequest(get, proto.QuotaList).

View File

@ -190,7 +190,6 @@ type MetaWrapper struct {
Client wrapper.SimpleClientInfo
IsSnapshotEnabled bool
DefaultStorageClass uint32
CacheDpStorageClass uint32
InnerReq bool
FollowerRead bool

View File

@ -25,7 +25,6 @@ type cfsOption struct {
BcacheFilterFiles string `json:"bcacheFilterFiles"`
BuffersTotalLimit string `json:"buffersTotalLimit"`
C string `json:"c"`
CacheAction string `json:"cacheAction"`
CertFile string `json:"certFile"`
ClientKey string `json:"clientKey"`

View File

@ -1,15 +0,0 @@
#!/usr/bin/env bash
Version=`git describe --abbrev=0 --tags`
BranchName=`git rev-parse --abbrev-ref HEAD`
CommitID=`git rev-parse HEAD`
BuildTime=`date +%Y-%m-%d\ %H:%M`
[[ "-$GOPATH" == "-" ]] && { echo "GOPATH not set"; exit 1; }
go build -ldflags "\
-X github.com/cubefs/cubefs/proto.Version=${Version} \
-X github.com/cubefs/cubefs/proto.CommitID=${CommitID} \
-X github.com/cubefs/cubefs/proto.BranchName=${BranchName} \
-X 'github.com/cubefs/cubefs/proto.BuildTime=${BuildTime}'" \
-o cfs-preload

View File

@ -1,16 +0,0 @@
{
"target":"/",
"volumeName": "cold4",
"masterAddr": "10.177.69.105:17010,10.177.69.106:17010,10.177.117.108:17010",
"logDir": "/mnt/hgfs/share/cfs-client-test",
"logLevel": "debug",
"ttl": "100",
"replicaNum": "1",
"zones": "",
"action":"clear",
"traverseDirConcurrency":"4",
"preloadFileConcurrency":"10",
"preloadFileSizeLimit":"10737418240",
"readBlockConcurrency":"10"
"prof":"27520"
}

View File

@ -1,124 +0,0 @@
package main
import (
"flag"
"fmt"
"os"
"strconv"
"strings"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/tool/preload/sdk"
"github.com/cubefs/cubefs/util/config"
)
var (
configFile = flag.String("c", "", "config file path")
configVersion = flag.Bool("v", false, "show version")
)
const (
Role = "Client"
)
func main() {
defer sdk.FlushLog()
flag.Parse()
if *configVersion {
fmt.Print(proto.DumpVersion(Role))
os.Exit(0)
}
// TODO: why *configFlie
cfg, err := config.LoadConfigFile(*configFile)
if err != nil {
fmt.Println("LoadConfigFile failed")
os.Exit(1)
}
if !checkConfig(cfg) {
os.Exit(1)
}
masters := strings.Split(cfg.GetString("masterAddr"), ",")
travereDirConcurrency, _ := strconv.ParseInt(cfg.GetString("traverseDirConcurrency"), 10, 64)
preloadFileConcurrency, _ := strconv.ParseInt(cfg.GetString("preloadFileConcurrency"), 10, 64)
preloadFileSizeLimit, _ := strconv.ParseInt(cfg.GetString("preloadFileSizeLimit"), 10, 64)
readBlockConcurrency, _ := strconv.ParseInt(cfg.GetString("readBlockConcurrency"), 10, 32)
clearFileConcurrency, _ := strconv.ParseInt(cfg.GetString("clearFileConcurrency"), 10, 64)
buffersTotalLimit, _ := strconv.ParseInt(cfg.GetString("buffersTotalLimit"), 10, 64)
replicaNum, _ := strconv.ParseInt(cfg.GetString("replicaNum"), 10, 64)
ttl, _ := strconv.ParseInt(cfg.GetString("ttl"), 10, 64)
if replicaNum > 16 || replicaNum < 1 {
fmt.Println("replicaNum must between [1,16]")
os.Exit(1)
}
if buffersTotalLimit < 0 {
fmt.Println("buffersTotalLimit cannot less than 0")
os.Exit(1)
} else if buffersTotalLimit == 0 {
buffersTotalLimit = int64(32768)
}
proto.InitBufferPool(buffersTotalLimit)
config := sdk.PreloadConfig{
Volume: cfg.GetString("volumeName"),
Masters: masters,
LogDir: cfg.GetString("logDir"),
LogLevel: cfg.GetString("logLevel"),
ProfPort: cfg.GetString("prof"),
LimitParam: sdk.LimitParameters{
TraverseDirConcurrency: travereDirConcurrency,
PreloadFileConcurrency: preloadFileConcurrency,
ReadBlockConcurrency: int32(readBlockConcurrency),
PreloadFileSizeLimit: preloadFileSizeLimit,
ClearFileConcurrency: clearFileConcurrency,
},
}
cli := sdk.NewClient(config)
if cli == nil {
fmt.Println("Preload client created failed")
os.Exit(1)
}
if !cli.CheckVolumeInfoFromMaster() {
fmt.Println("Preload only work in cold volume")
os.Exit(1)
}
fmt.Printf("conf is %v\n", config)
action := cfg.GetString("action")
if action == "preload" {
if err := cli.PreloadDir(cfg.GetString("target"), int(replicaNum), uint64(ttl), cfg.GetString("zones")); err != nil {
total, succeed := cli.GetPreloadResult()
fmt.Printf("Preload failed:%v\n", err)
fmt.Printf("Result: total[%v], succeed[%v]\n", total, succeed)
} else {
fmt.Println("Preload succeed")
}
} else if action == "clear" {
cli.ClearPreloadDP(cfg.GetString("target"))
} else {
fmt.Printf("action[%v] is not support\n", action)
os.Exit(1)
}
}
func checkConfig(cfg *config.Config) bool {
masters := cfg.GetString("masterAddr")
target := cfg.GetString("target")
vol := cfg.GetString("volumeName")
logDir := cfg.GetString("logDir")
logLevel := cfg.GetString("logLevel")
ttl := cfg.GetString("ttl")
action := cfg.GetString("action")
if len(masters) == 0 || len(target) == 0 || len(vol) == 0 || len(logDir) == 0 ||
len(logLevel) == 0 || len(ttl) == 0 || len(action) == 0 {
fmt.Println("masterAddr, target, volumeName, logDir, logLevel, ttl , action cannot be empty")
return false
}
return true
}

View File

@ -1,141 +0,0 @@
// Copyright 2022 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 main
import (
"testing"
"github.com/cubefs/cubefs/util/config"
)
func TestCheckConfMaster(t *testing.T) {
t.Run("masters", func(t *testing.T) {
cfgJSON := `{"target":"/",
"volumeName": "cold3",
"masterAddr": "10.177.69.105:17010, 10.177.69.106:17010, 10.177.117.108:17010",
"logDir": "/home/ADC/80256477/adls/log/preload",
"logLevel": "debug",
"ttl": "10",
"action":"preload"}`
cfg := config.LoadConfigString(cfgJSON)
if checkConfig(cfg) == false {
t.Fatalf("expected true, but got false")
}
})
t.Run("masters_empty", func(t *testing.T) {
cfgJSON := `{"target":"/",
"volumeName": "cold3",
"logDir": "/home/ADC/80256477/adls/log/preload",
"logLevel": "debug",
"ttl": "10",
"action":"preload"}`
cfg := config.LoadConfigString(cfgJSON)
if checkConfig(cfg) == true {
t.Fatalf("expected false, but got true")
}
})
}
func TestCheckConfTarget(t *testing.T) {
t.Run("target_empty", func(t *testing.T) {
cfgJSON := `{
"volumeName": "cold3",
"masterAddr": "10.177.69.105:17010, 10.177.69.106:17010, 10.177.117.108:17010",
"logDir": "/home/ADC/80256477/adls/log/preload",
"logLevel": "debug",
"ttl": "10",
"action":"preload"}`
cfg := config.LoadConfigString(cfgJSON)
if checkConfig(cfg) == true {
t.Fatalf("expected false, but got true")
}
})
}
func TestCheckConfVol(t *testing.T) {
t.Run("vol_empty", func(t *testing.T) {
cfgJSON := `{"target":"/",
"masterAddr": "10.177.69.105:17010, 10.177.69.106:17010, 10.177.117.108:17010",
"logDir": "/home/ADC/80256477/adls/log/preload",
"logLevel": "debug",
"ttl": "10",
"action":"preload"}`
cfg := config.LoadConfigString(cfgJSON)
if checkConfig(cfg) == true {
t.Fatalf("expected false, but got true")
}
})
}
func TestCheckConfLogDir(t *testing.T) {
t.Run("logDir_empty", func(t *testing.T) {
cfgJSON := `{"target":"/",
"volumeName": "cold3",
"masterAddr": "10.177.69.105:17010, 10.177.69.106:17010, 10.177.117.108:17010",
"logLevel": "debug",
"ttl": "10",
"action":"preload"}`
cfg := config.LoadConfigString(cfgJSON)
if checkConfig(cfg) == true {
t.Fatalf("expected false, but got true")
}
})
}
func TestCheckConfLogLevel(t *testing.T) {
t.Run("logLevel_empty", func(t *testing.T) {
cfgJSON := `{"target":"/",
"volumeName": "cold3",
"masterAddr": "10.177.69.105:17010, 10.177.69.106:17010, 10.177.117.108:17010",
"logDir": "/home/ADC/80256477/adls/log/preload",
"ttl": "10",
"action":"preload"}`
cfg := config.LoadConfigString(cfgJSON)
if checkConfig(cfg) == true {
t.Fatalf("expected false, but got true")
}
})
}
func TestCheckConfTTL(t *testing.T) {
t.Run("ttl_empty", func(t *testing.T) {
cfgJSON := `{"target":"/",
"volumeName": "cold3",
"masterAddr": "10.177.69.105:17010, 10.177.69.106:17010, 10.177.117.108:17010",
"logDir": "/home/ADC/80256477/adls/log/preload",
"logLevel": "debug",
"action":"preload"}`
cfg := config.LoadConfigString(cfgJSON)
if checkConfig(cfg) == true {
t.Fatalf("expected false, but got true")
}
})
}
func TestCheckConfAction(t *testing.T) {
t.Run("action_empty", func(t *testing.T) {
cfgJSON := `{"target":"/",
"volumeName": "cold3",
"masterAddr": "10.177.69.105:17010, 10.177.69.106:17010, 10.177.117.108:17010",
"logDir": "/home/ADC/80256477/adls/log/preload",
"logLevel": "debug",
"ttl": "10"}`
cfg := config.LoadConfigString(cfgJSON)
if checkConfig(cfg) == true {
t.Fatalf("expected false, but got true")
}
})
}

View File

@ -1,607 +0,0 @@
// Copyright 2022 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 sdk
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/pprof"
"path"
gopath "path"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/cubefs/cubefs/blobstore/api/access"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/data/blobstore"
"github.com/cubefs/cubefs/sdk/data/stream"
masterSDK "github.com/cubefs/cubefs/sdk/master"
"github.com/cubefs/cubefs/sdk/meta"
"github.com/cubefs/cubefs/util/buf"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/stat"
)
type LimitParameters struct {
TraverseDirConcurrency int64
PreloadFileConcurrency int64
ReadBlockConcurrency int32
PreloadFileSizeLimit int64
ClearFileConcurrency int64
}
type PreLoadClient struct {
mw *meta.MetaWrapper
ec *stream.ExtentClient
mc *masterSDK.MasterClient
ebsc *blobstore.BlobStoreClient
sync.RWMutex
fileCache []fileInfo // file list for target dir
vol string
limitParam LimitParameters
cacheAction int
cacheThreshold int
ebsBlockSize int
preloadFileNumTotal int64
preloadFileNumSucceed int64
CacheDpStorageClass uint32
}
type fileInfo struct {
ino uint64
name string
size uint64
}
type PreloadConfig struct {
Volume string
Masters []string
LogDir string
LogLevel string
ProfPort string
LimitParam LimitParameters
}
func FlushLog() {
log.LogFlush()
}
func NewClient(config PreloadConfig) *PreLoadClient {
defer log.LogFlush()
c := &PreLoadClient{
vol: config.Volume,
}
if config.LogDir != "" {
log.InitLog(config.LogDir, "preload", convertLogLevel(config.LogLevel), nil, log.DefaultLogLeftSpaceLimitRatio)
stat.NewStatistic(config.LogDir, "preload", int64(stat.DefaultStatLogSize), stat.DefaultTimeOutUs, true)
}
if config.ProfPort != "" {
go func() {
mainMux := http.NewServeMux()
mux := http.NewServeMux()
http.HandleFunc(log.SetLogLevelPath, log.SetLogLevel)
mux.Handle("/debug/pprof", http.HandlerFunc(pprof.Index))
mux.Handle("/debug/pprof/cmdline", http.HandlerFunc(pprof.Cmdline))
mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
mux.Handle("/debug/", http.HandlerFunc(pprof.Index))
mainHandler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
if strings.HasPrefix(req.URL.Path, "/debug/") {
mux.ServeHTTP(w, req)
} else {
http.DefaultServeMux.ServeHTTP(w, req)
}
})
mainMux.Handle("/", mainHandler)
e := http.ListenAndServe(fmt.Sprintf(":%v", config.ProfPort), mainMux)
if e != nil {
log.LogWarnf("newClient newEBSClient cannot listen pprof (%v)", config.ProfPort)
}
}()
}
var (
mw *meta.MetaWrapper
ec *stream.ExtentClient
err error
)
c.mc = masterSDK.NewMasterClient(config.Masters, false)
if err = c.newEBSClient(config.Masters, config.LogDir); err != nil {
log.LogErrorf("newClient newEBSClient failed(%v)", err)
return nil
}
var view *proto.SimpleVolView
if view, err = c.mc.AdminAPI().GetVolumeSimpleInfo(c.vol); err != nil {
log.LogErrorf("CheckVolumeInfoFromMaster: get volume simple info fail: volume(%v) err(%v)", c.vol, err)
return nil
}
if mw, err = meta.NewMetaWrapper(&meta.MetaConfig{
Volume: config.Volume,
Masters: config.Masters,
ValidateOwner: false,
}); err != nil {
log.LogErrorf("newClient NewMetaWrapper failed(%v)", err)
return nil
}
if ec, err = stream.NewExtentClient(&stream.ExtentConfig{
Volume: config.Volume,
Masters: config.Masters,
Preload: true,
OnAppendExtentKey: mw.AppendExtentKey,
OnSplitExtentKey: mw.SplitExtentKey,
OnGetExtents: mw.GetExtents,
OnTruncate: mw.Truncate,
VolStorageClass: view.VolStorageClass,
VolAllowedStorageClass: view.AllowedStorageClass,
VolCacheDpStorageClass: view.CacheDpStorageClass,
OnRenewalForbiddenMigration: mw.RenewalForbiddenMigration,
OnForbiddenMigration: mw.ForbiddenMigration,
MetaWrapper: mw,
}); err != nil {
log.LogErrorf("newClient NewExtentClient failed(%v)", err)
return nil
}
c.mw = mw
c.ec = ec
c.limitParam.PreloadFileConcurrency = config.LimitParam.PreloadFileConcurrency
if c.limitParam.PreloadFileConcurrency == 0 {
c.limitParam.PreloadFileConcurrency = 3
}
c.limitParam.TraverseDirConcurrency = config.LimitParam.TraverseDirConcurrency
if c.limitParam.PreloadFileConcurrency == 0 {
c.limitParam.PreloadFileConcurrency = 4
}
c.limitParam.PreloadFileSizeLimit = config.LimitParam.PreloadFileSizeLimit
c.limitParam.ReadBlockConcurrency = config.LimitParam.ReadBlockConcurrency
if c.limitParam.ReadBlockConcurrency == 0 {
c.limitParam.ReadBlockConcurrency = 3
}
c.limitParam.ClearFileConcurrency = config.LimitParam.ClearFileConcurrency
if c.limitParam.ClearFileConcurrency == 0 {
c.limitParam.ClearFileConcurrency = 4
}
log.LogDebugf("Client is created:(%v)", c)
return c
}
func convertLogLevel(level string) log.Level {
switch level {
case "debug":
return log.DebugLevel
case "info":
return log.InfoLevel
case "warn":
return log.WarnLevel
case "error":
return log.ErrorLevel
case "fatal":
return log.FatalLevel
case "critical":
return log.CriticalLevel
case "read":
return log.ReadLevel
case "update":
return log.UpdateLevel
default:
return log.DebugLevel
}
}
func (c *PreLoadClient) newEBSClient(masters []string, logDir string) (err error) {
var (
volumeInfo *proto.SimpleVolView
clusterInfo *proto.ClusterInfo
ebsc *blobstore.BlobStoreClient
)
volumeInfo, err = c.mc.AdminAPI().GetVolumeSimpleInfo(c.vol)
if err != nil {
return
}
c.ebsBlockSize = volumeInfo.ObjBlockSize
c.cacheAction = volumeInfo.CacheAction
c.cacheThreshold = volumeInfo.CacheThreshold
clusterInfo, err = c.mc.AdminAPI().GetClusterInfo()
if err != nil {
return
}
ebsEndpoint := clusterInfo.EbsAddr
if ebsc, err = blobstore.NewEbsClient(access.Config{
ConnMode: access.NoLimitConnMode,
Consul: access.ConsulConfig{
Address: ebsEndpoint,
},
MaxSizePutOnce: int64(c.ebsBlockSize),
Logger: &access.Logger{
Filename: gopath.Join(logDir, "ebs/ebs.log"),
},
}); err != nil {
log.LogErrorf("newClient newEBSClient failed(%v)", err)
return
}
c.ebsc = ebsc
buf.InitCachePool(c.ebsBlockSize)
return
}
func (c *PreLoadClient) ctx(cid int64, ino uint64) context.Context {
_, ctx := trace.StartSpanFromContextWithTraceID(context.Background(), "", fmt.Sprintf("cid=%v,ino=%v", cid, ino))
return ctx
}
func (c *PreLoadClient) walker(dir string, wg *sync.WaitGroup, currentGoroutineNum *int64, ch chan<- fileInfo, newGoroutine bool) (err error) {
defer func() {
if newGoroutine {
atomic.AddInt64(currentGoroutineNum, -1)
wg.Done()
}
}()
var (
ino uint64
info *proto.InodeInfo
)
ino, err = c.mw.LookupPath(gopath.Clean(dir))
if err != nil {
log.LogErrorf("LookupPath path(%v) faild(%v)", dir, err)
return err
}
info, err = c.mw.InodeGet_ll(ino)
if err != nil {
log.LogErrorf("InodeGet_ll path(%v) faild(%v)", dir, err)
return err
}
if proto.IsRegular(info.Mode) {
fInfo := fileInfo{ino: info.Inode, name: gopath.Base(dir)}
log.LogDebugf("Target is a file:(%v)", fInfo)
ch <- fInfo
return err
}
children, err := c.mw.ReadDir_ll(info.Inode)
if err != nil {
log.LogErrorf("ReadDir_ll path(%v) faild(%v)", dir, err)
return err
}
for _, child := range children {
if proto.IsDir(child.Type) {
log.LogDebugf("preloadTravrseDir:(%v) is dir", path.Join(dir, child.Name))
if atomic.LoadInt64(currentGoroutineNum) < c.limitParam.PreloadFileConcurrency {
wg.Add(1)
atomic.AddInt64(currentGoroutineNum, 1)
go c.walker(path.Join(dir, child.Name), wg, currentGoroutineNum, ch, true)
} else {
c.walker(path.Join(dir, child.Name), wg, currentGoroutineNum, ch, false)
}
} else if proto.IsRegular(child.Type) {
fInfo := fileInfo{ino: child.Inode, name: child.Name}
log.LogDebugf("preloadTravrseDir send:(%v)", fInfo)
ch <- fInfo
}
}
return nil
}
func (c *PreLoadClient) allocatePreloadDPWorker(ch <-chan fileInfo) uint64 {
var (
inodes []uint64
total uint64
cache map[uint64]fileInfo
)
cache = make(map[uint64]fileInfo)
for fInfo := range ch {
inodes = append(inodes, fInfo.ino)
cache[fInfo.ino] = fInfo
if len(inodes) < 100 {
continue
}
infos := c.mw.BatchInodeGet(inodes)
for _, info := range infos {
if int64(info.Size) <= c.limitParam.PreloadFileSizeLimit {
total += info.Size
cachefino := cache[info.Inode]
cachefino.size = info.Size
// append
c.fileCache = append(c.fileCache, cachefino)
log.LogDebugf("allocatePreloadDPWorker append:%v", cachefino)
}
}
inodes = inodes[:0]
for key := range cache {
delete(cache, key)
}
}
// flush cache
if len(inodes) != 0 {
infos := c.mw.BatchInodeGet(inodes)
for _, info := range infos {
if int64(info.Size) <= c.limitParam.PreloadFileSizeLimit {
total += info.Size
cachefino := cache[info.Inode]
cachefino.size = info.Size
// append
c.fileCache = append(c.fileCache, cachefino)
log.LogDebugf("allocatePreloadDPWorker append#2:%v", cachefino)
}
}
}
return total
}
func (c *PreLoadClient) allocatePreloadDP(target string, count int, ttl uint64, zones string) (err error) {
log.LogDebugf("allocatePreloadDP enter")
ch := make(chan fileInfo, 100)
var (
wg sync.WaitGroup
currentGoroutineNum int64 = 0
)
wg.Add(1)
atomic.AddInt64(&currentGoroutineNum, 1)
go c.walker(target, &wg, &currentGoroutineNum, ch, true)
go func() {
wg.Wait()
close(ch)
}()
need := c.allocatePreloadDPWorker(ch)
if need == 0 {
return errors.New("No file would be preloaded")
}
//#test1
if (need % (1024 * 1024 * 1024)) == 0 {
need = uint64(need / (1024 * 1024 * 1024))
} else {
need = uint64(need/(1024*1024*1024)) + 1
}
log.LogDebugf("preloadFile:need total space %v GB, ttl %v second, zones=%v", need, ttl, zones)
total, used := c.getVolumeCacheCapacity()
if float64(need+used) > float64(total) {
log.LogErrorf("AllocatePreLoadDataPartition failed: need space (%v) GB, volume total(%v)GB used(%v)GB", need, total, used)
return fmt.Errorf("AllocatePreLoadDataPartition failed: need space (%v) GB, volume total(%v)GB used(%v)GB", need, total, used)
}
err = c.ec.AllocatePreLoadDataPartition(c.vol, count, need, ttl, zones)
if err != nil {
log.LogErrorf("AllocatePreLoadDataPartition failed:%v", err)
return err
}
log.LogDebugf("allocatePreloadDP leave")
return nil
}
func (c *PreLoadClient) clearPreloadDPWorker(ch <-chan fileInfo) {
var routineNum int64 = 0
var wg sync.WaitGroup
for fInfo := range ch {
if atomic.LoadInt64(&routineNum) < c.limitParam.ClearFileConcurrency {
wg.Add(1)
go func(f fileInfo) {
defer func() {
atomic.AddInt64(&routineNum, -1)
wg.Done()
}()
log.LogDebugf("clearPreloadDPWorker:clear file %v", f.name)
c.mw.InodeClearPreloadCache_ll(f.ino)
}(fInfo)
atomic.AddInt64(&routineNum, 1)
} else {
log.LogDebugf("clearPreloadDPWorker:clear file %v", fInfo.name)
c.mw.InodeClearPreloadCache_ll(fInfo.ino)
}
}
wg.Wait()
}
func (c *PreLoadClient) ClearPreloadDP(target string) (err error) {
ch := make(chan fileInfo, 100)
var (
wg sync.WaitGroup
currentGoroutineNum int64 = 0
)
wg.Add(1)
atomic.AddInt64(&currentGoroutineNum, 1)
go c.walker(target, &wg, &currentGoroutineNum, ch, true)
go func() {
wg.Wait()
close(ch)
}()
c.clearPreloadDPWorker(ch)
return nil
}
func (c *PreLoadClient) preloadFileWorker(id int64, jobs <-chan fileInfo, wg *sync.WaitGroup) {
defer wg.Done()
var total int64 = 0
var succeed int64 = 0
noWritableDP := false
for job := range jobs {
if noWritableDP {
log.LogWarnf("no writable dp,ingnore (%v) to cbfs", job.name)
continue // consume the job
}
total += 1
log.LogDebugf("worker %v ready to preload(%v)", id, job.name)
ino := job.ino
//#1 open
c.ec.OpenStream(ino, true, true, "")
//#2 write
var (
objExtents []proto.ObjExtentKey
err error
)
if _, _, _, objExtents, err = c.mw.GetObjExtents(ino); err != nil {
log.LogWarnf("GetObjExtents (%v) faild(%v)", job.name, err)
continue
}
clientConf := blobstore.ClientConfig{
VolName: c.vol,
VolType: proto.VolumeTypeCold,
Ino: ino,
Mw: c.mw,
Ec: c.ec,
Ebsc: c.ebsc,
EnableBcache: false,
ReadConcurrency: int(c.limitParam.ReadBlockConcurrency),
CacheAction: c.cacheAction,
FileCache: false,
CacheThreshold: c.cacheThreshold,
}
fileReader := blobstore.NewReader(clientConf)
subErr := false
for _, objExtent := range objExtents {
size := objExtent.Size
buf := make([]byte, size)
var n int
n, err = fileReader.Read(c.ctx(0, ino), buf, int(objExtent.FileOffset), int(size))
if err != nil {
subErr = true
log.LogWarnf("Read (%v) from ebs failed (%v)", objExtent, err)
continue
}
if uint64(n) != size {
log.LogWarnf("Read (%v) wrong size:(%v)", objExtent, n)
continue
}
_, err = c.ec.Write(ino, int(objExtent.FileOffset), buf, 0, nil, c.CacheDpStorageClass, false)
// in preload mode,once extend_hander set to error, streamer is set to error
// so write should fail immediately
if err != nil {
subErr = true
log.LogWarnf("preload (%v) to cbfs failed (%v)", job.name, err)
if err = c.ec.GetDataPartitionForWrite(c.CacheDpStorageClass); err != nil {
log.LogErrorf("worker %v end for %v", id, err)
noWritableDP = true
}
break
}
}
c.ec.CloseStream(ino)
if !subErr {
log.LogInfof("worker %v preload (%v) to cbfs success", id, job.name)
succeed += 1
}
}
atomic.AddInt64(&c.preloadFileNumTotal, total)
atomic.AddInt64(&c.preloadFileNumSucceed, succeed)
log.LogInfof("worker %v end:total %v, succeed %v", id, total, succeed)
}
func (c *PreLoadClient) preloadFile() error {
log.LogDebug("preloadFile enter")
var (
wg sync.WaitGroup
w int64
)
jobs := make(chan fileInfo, 100)
for w = 1; w <= c.limitParam.PreloadFileConcurrency; w++ {
wg.Add(1)
go c.preloadFileWorker(w, jobs, &wg)
}
for _, fileInfo := range c.fileCache {
jobs <- fileInfo
}
close(jobs)
wg.Wait()
log.LogInfof("preloadFile end:total %v, succeed %v", c.preloadFileNumTotal, c.preloadFileNumSucceed)
if c.preloadFileNumTotal == c.preloadFileNumSucceed {
return nil
} else if c.preloadFileNumSucceed < c.preloadFileNumTotal {
return errors.New("Preload partially succeed")
} else {
return errors.New("Preload failed")
}
}
func (c *PreLoadClient) CheckVolumeInfoFromMaster() bool {
var (
err error
view *proto.SimpleVolView
)
if view, err = c.mc.AdminAPI().GetVolumeSimpleInfo(c.vol); err != nil {
log.LogErrorf("CheckVolumeInfoFromMaster: get volume simple info fail: volume(%v) err(%v)", c.vol, err)
return false
}
if view.CacheDpStorageClass == proto.StorageClass_Unspecified {
log.LogErrorf("CheckVolumeInfoFromMaster: volume(%v) cacheDpStorageClass not specified", c.vol)
return false
}
c.CacheDpStorageClass = view.CacheDpStorageClass
log.LogInfof("CheckVolumeInfoFromMaster: volume(%v) cacheDpStorageClass: %v",
c.vol, proto.StorageClassString(c.CacheDpStorageClass))
return view.VolType == proto.VolumeTypeCold
}
func (c *PreLoadClient) getVolumeCacheCapacity() (total uint64, used uint64) {
var (
err error
view *proto.SimpleVolView
)
if view, err = c.mc.AdminAPI().GetVolumeSimpleInfo(c.vol); err != nil {
log.LogErrorf("getSimpleVolView: get volume simple info fail: volume(%v) err(%v)", c.vol, err)
return 0, 0
}
return view.CacheCapacity, view.PreloadCapacity
}
func (c *PreLoadClient) PreloadDir(target string, count int, ttl uint64, zones string) (err error) {
log.LogDebugf("PreloadDir (%v)", target)
if err = c.allocatePreloadDP(target, count, ttl, zones); err != nil {
log.LogErrorf("PreloadDir failed(%v)", err)
return
}
log.LogDebugf("Wait 100s for preload dp get ready")
time.Sleep(time.Duration(100) * time.Second)
log.LogDebugf("Sleep end")
// Step3.2 preload the file
return c.preloadFile()
}
func (c *PreLoadClient) GetPreloadResult() (int64, int64) {
return c.preloadFileNumTotal, c.preloadFileNumSucceed
}

View File

@ -1,34 +0,0 @@
// Copyright 2022 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 sdk
import (
"testing"
"github.com/cubefs/cubefs/util/log"
)
func TestConvertDebugLevel(t *testing.T) {
t.Run("warn", func(t *testing.T) {
if convertLogLevel("warn") != log.WarnLevel {
t.Fatalf("expected WarnLevel")
}
})
t.Run("warn2", func(t *testing.T) {
if convertLogLevel("warn2") != log.DebugLevel {
t.Fatalf("expected DebugLevel")
}
})
}