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

refactor(clientmetanode): hybrid cloud check inode/vol storage class when checking vol type

Browse Source 
Signed-off-by: chihe <chi.he@oppo.com>
This commit is contained in:
chihe 2023-11-23 04:31:41 -08:00 committed by AmazingChi
parent 1f9dab0eda
commit 347208f189
20 changed files with 189 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
client/fs/dir.go
View File

@ -178,7 +178,11 @@ func (d *Dir) Create(ctx context.Context, req *fuse.CreateRequest, resp *fuse.Cr
if req.Flags&0x0f != syscall.O_RDONLY {
openForWrite = true
}
d.super.ec.OpenStream(info.Inode, openForWrite)
var isCache = false
if proto.IsStorageClassBlobStore(info.StorageClass) {
isCache = true
}
d.super.ec.OpenStream(info.Inode, openForWrite, isCache)
d.super.fslock.Lock()
d.super.nodeCache[info.Inode] = child
d.super.fslock.Unlock()

50
client/fs/file.go
View File

@ -68,7 +68,7 @@ var (
// NewFile returns a new file.
func NewFile(s *Super, i *proto.InodeInfo, flag uint32, pino uint64, filename string) fs.Node {
if proto.IsCold(s.volType) || i.StorageClass == proto.StorageClass_BlobStore {
if proto.IsCold(s.volType) || proto.IsStorageClassBlobStore(i.StorageClass) {
var (
fReader *blobstore.Reader
fWriter *blobstore.Writer
@ -229,10 +229,14 @@ func (f *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenR
if req.Flags&0x0f != syscall.O_RDONLY {
openForWrite = true
}
var isCache = false
if proto.IsStorageClassBlobStore(f.info.StorageClass) {
isCache = true
}
if needBCache {
f.super.ec.OpenStreamWithCache(ino, needBCache, openForWrite)
f.super.ec.OpenStreamWithCache(ino, needBCache, openForWrite, isCache)
} else {
f.super.ec.OpenStream(ino, openForWrite)
f.super.ec.OpenStream(ino, openForWrite, isCache)
}
log.LogDebugf("TRACE open ino(%v) f.super.bcacheDir(%v) needBCache(%v)", ino, f.super.bcacheDir, needBCache)
@ -241,7 +245,7 @@ func (f *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenR
if f.super.keepCache && resp != nil {
resp.Flags |= fuse.OpenKeepCache
}
if proto.IsCold(f.super.volType) || f.info.StorageClass == proto.StorageClass_BlobStore {
if proto.IsCold(f.super.volType) || proto.IsStorageClassBlobStore(f.info.StorageClass) {
log.LogDebugf("TRANCE open ino(%v) info(%v)", ino, f.info)
fileSize, _ := f.fileSizeVersion2(ino)
clientConf := blobstore.ClientConfig{
@ -336,8 +340,8 @@ func (f *File) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadR
metric.SetWithLabels(err, map[string]string{exporter.Vol: f.super.volname})
}()
var size int
if proto.IsHot(f.super.volType) {
size, err = f.super.ec.Read(f.info.Inode, resp.Data[fuse.OutHeaderSize:], int(req.Offset), req.Size)
if proto.IsHot(f.super.volType) || proto.IsStorageClassReplica(f.info.StorageClass) {
size, err = f.super.ec.Read(f.info.Inode, resp.Data[fuse.OutHeaderSize:], int(req.Offset), req.Size, f.info.StorageClass)
} else {
size, err = f.fReader.Read(ctx, resp.Data[fuse.OutHeaderSize:], int(req.Offset), req.Size)
}
@ -392,7 +396,7 @@ func (f *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.Wri
reqlen := len(req.Data)
log.LogDebugf("TRACE Write enter: ino(%v) offset(%v) len(%v) flags(%v) fileflags(%v) quotaIds(%v) req(%v)",
ino, req.Offset, reqlen, req.Flags, req.FileFlags, f.info.QuotaInfos, req)
if proto.IsHot(f.super.volType) || f.isStoredInReplicaSystem() {
if proto.IsHot(f.super.volType) || proto.IsStorageClassReplica(f.info.StorageClass) {
filesize, _ := f.fileSize(ino)
if req.Offset > int64(filesize) && reqlen == 1 && req.Data[0] == 0 {
@ -419,13 +423,13 @@ func (f *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.Wri
if f.super.enSyncWrite {
flags |= proto.FlagsSyncWrite
}
if proto.IsCold(f.super.volType) {
if proto.IsCold(f.super.volType) || proto.IsStorageClassBlobStore(f.info.StorageClass) {
waitForFlush = false
flags |= proto.FlagsSyncWrite
}
}
if req.FileFlags&fuse.OpenAppend != 0 || proto.IsCold(f.super.volType) {
if req.FileFlags&fuse.OpenAppend != 0 || proto.IsCold(f.super.volType) || proto.IsStorageClassBlobStore(f.info.StorageClass) {
flags |= proto.FlagsAppend
}
@ -452,12 +456,12 @@ func (f *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.Wri
return nil
}
var size int
if proto.IsHot(f.super.volType) || f.isStoredInReplicaSystem() {
if proto.IsHot(f.super.volType) || proto.IsStorageClassReplica(f.info.StorageClass) {
f.super.ec.GetStreamer(ino).SetParentInode(f.parentIno)
if size, err = f.super.ec.Write(ino, int(req.Offset), req.Data, flags, checkFunc, f.info.StorageClass, false); err == ParseError(syscall.ENOSPC) {
return
}
} else if f.isStoredInEbsSystem() {
} else if proto.IsStorageClassBlobStore(f.info.StorageClass) {
atomic.StoreInt32(&f.idle, 0)
size, err = f.fWriter.Write(context.Background(), int(req.Offset), req.Data, flags)
} else {
@ -513,7 +517,7 @@ func (f *File) Flush(ctx context.Context, req *fuse.FlushRequest) (err error) {
defer func() {
metric.SetWithLabels(err, map[string]string{exporter.Vol: f.super.volname})
}()
if proto.IsHot(f.super.volType) {
if proto.IsHot(f.super.volType) || proto.IsStorageClassReplica(f.info.StorageClass) {
err = f.super.ec.Flush(f.info.Inode)
} else {
f.Lock()
@ -547,7 +551,7 @@ func (f *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) (err error) {
log.LogDebugf("TRACE Fsync enter: ino(%v)", f.info.Inode)
start := time.Now()
if proto.IsHot(f.super.volType) {
if proto.IsHot(f.super.volType) || proto.IsStorageClassReplica(f.info.StorageClass) {
err = f.super.ec.Flush(f.info.Inode)
} else {
err = f.fWriter.Flush(f.info.Inode, context.Background())
@ -577,10 +581,14 @@ func (f *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse
if req.Flags&0x0f != syscall.O_RDONLY {
openForWrite = true
}
if req.Valid.Size() && proto.IsHot(f.super.volType) {
var isCache = false
if proto.IsStorageClassBlobStore(f.info.StorageClass) {
isCache = true
}
if req.Valid.Size() && (proto.IsHot(f.super.volType) || proto.IsStorageClassReplica(f.info.StorageClass)) {
// when use trunc param in open request through nfs client and mount on cfs mountPoint, cfs client may not recv open message but only setAttr,
// the streamer may not open and cause io error finally,so do a open no matter the stream be opened or not
if err := f.super.ec.OpenStream(ino, openForWrite); err != nil {
if err := f.super.ec.OpenStream(ino, openForWrite, isCache); err != nil {
log.LogErrorf("Setattr: OpenStream ino(%v) size(%v) err(%v)", ino, req.Size, err)
return ParseError(err)
}
@ -605,7 +613,7 @@ func (f *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse
return ParseError(err)
}
if req.Valid.Size() && proto.IsHot(f.super.volType) {
if req.Valid.Size() && (proto.IsHot(f.super.volType) || proto.IsStorageClassReplica(f.info.StorageClass)) {
if req.Size != info.Size {
log.LogWarnf("Setattr: truncate ino(%v) reqSize(%v) inodeSize(%v)", ino, req.Size, info.Size)
}
@ -763,7 +771,7 @@ func (f *File) fileSize(ino uint64) (size int, gen uint64) {
func (f *File) fileSizeVersion2(ino uint64) (size int, gen uint64) {
size, gen, valid := f.super.ec.FileSize(ino)
if proto.IsCold(f.super.volType) {
if proto.IsCold(f.super.volType) || proto.IsStorageClassBlobStore(f.info.StorageClass) {
valid = false
}
if !valid {
@ -803,11 +811,3 @@ func (f *File) filterFilesSuffix(filterFiles string) bool {
}
return false
}
func (f *File) isStoredInReplicaSystem() bool {
return f.info.StorageClass == proto.StorageClass_Replica_SSD || f.info.StorageClass == proto.StorageClass_Replica_HDD
}
func (f *File) isStoredInEbsSystem() bool {
return f.info.StorageClass == proto.StorageClass_BlobStore
}

6
client/fs/super.go
View File

@ -248,6 +248,7 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
StreamRetryTimeout: opt.StreamRetryTimeout,
OnRenewalForbiddenMigration: s.mw.RenewalForbiddenMigration,
CacheDpStorageClass: s.cacheDpStorageClass,
AllowedStorageClass: opt.AllowedStorageClass,
}
s.ec, err = stream.NewExtentClient(extentConfig)
@ -255,7 +256,8 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
return nil, errors.Trace(err, "NewExtentClient failed!")
}
s.mw.VerReadSeq = s.ec.GetReadVer()
if proto.IsCold(opt.VolType) {
if proto.IsCold(opt.VolType) || proto.VolSupportsBlobStore(opt.AllowedStorageClass) {
s.ebsc, err = blobstore.NewEbsClient(access.Config{
ConnMode: access.NoLimitConnMode,
Consul: access.ConsulConfig{
@ -281,7 +283,7 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
}
s.suspendCh = make(chan interface{})
if proto.IsCold(opt.VolType) {
if proto.IsCold(opt.VolType) || proto.VolSupportsBlobStore(opt.AllowedStorageClass) {
go s.scheduleFlush()
}
if s.mw.EnableSummary {

5
client/fuse.go
View File

@ -435,7 +435,7 @@ func main() {
proto.InitBufferPoolEx(opt.BuffersTotalLimit, int(opt.BufferChanSize))
log.LogInfof("InitBufferPoolEx: total limit %d, chan size %d", opt.BuffersTotalLimit, opt.BufferChanSize)
if proto.IsCold(opt.VolType) {
if proto.IsCold(opt.VolType) || proto.VolSupportsBlobStore(opt.AllowedStorageClass) {
buf.InitCachePool(opt.EbsBlockSize)
}
if opt.EnableBcache {
@ -815,7 +815,7 @@ func mount(opt *proto.MountOptions) (fsConn *fuse.Conn, super *cfs.Super, err er
os.Exit(1)
}
super.SetTransaction(volumeInfo.EnableTransactionV1, volumeInfo.TxTimeout, volumeInfo.TxConflictRetryNum, volumeInfo.TxConflictRetryInterval)
if proto.IsCold(opt.VolType) {
if proto.IsCold(opt.VolType) || proto.VolSupportsBlobStore(opt.AllowedStorageClass) {
super.CacheAction = volumeInfo.CacheAction
super.CacheThreshold = volumeInfo.CacheThreshold
super.EbsBlockSize = volumeInfo.ObjBlockSize
@ -1064,6 +1064,7 @@ func loadConfFromMaster(opt *proto.MountOptions) (err error) {
opt.TxConflictRetryNum = volumeInfo.TxConflictRetryNum
opt.TxConflictRetryInterval = volumeInfo.TxConflictRetryInterval
opt.CacheDpStorageClass = volumeInfo.CacheDpStorageClass
opt.AllowedStorageClass = volumeInfo.AllowedStorageClass
var clusterInfo *proto.ClusterInfo
clusterInfo, err = mc.AdminAPI().GetClusterInfo()

20
libsdk/libsdk.go
View File

@ -822,11 +822,11 @@ func cfs_write(id C.int64_t, fd C.int, buf unsafe.Pointer, size C.size_t, off C.
var wait bool
if f.flags&uint32(C.O_DIRECT) != 0 || f.flags&uint32(C.O_SYNC) != 0 || f.flags&uint32(C.O_DSYNC) != 0 {
if proto.IsHot(c.volType) {
if proto.IsHot(c.volType) || proto.IsStorageClassReplica(f.storageClass) {
wait = true
}
}
if f.flags&uint32(C.O_APPEND) != 0 || proto.IsCold(c.volType) {
if f.flags&uint32(C.O_APPEND) != 0 || proto.IsCold(c.volType) || proto.IsStorageClassBlobStore(f.storageClass) {
flags |= proto.FlagsAppend
flags |= proto.FlagsSyncWrite
}
@ -1499,7 +1499,7 @@ func (c *client) allocFD(ino uint64, flags, mode uint32, fileCache bool, fileSiz
if flags&0x0f != syscall.O_RDONLY {
f.openForWrite = true
}
if proto.IsCold(c.volType) {
if proto.IsCold(c.volType) || proto.IsStorageClassBlobStore(storageClass) {
clientConf := blobstore.ClientConfig{
VolName: c.volName,
VolType: c.volType,
@ -1621,7 +1621,11 @@ func (c *client) mkdir(pino uint64, name string, mode uint32, fullPath string) (
}
func (c *client) openStream(f *file) {
_ = c.ec.OpenStream(f.ino, f.openForWrite)
var isCache = false
if proto.IsStorageClassBlobStore(f.storageClass) {
isCache = true
}
_ = c.ec.OpenStream(f.ino, f.openForWrite, isCache)
}
func (c *client) closeStream(f *file) {
@ -1633,7 +1637,7 @@ func (c *client) closeStream(f *file) {
}
func (c *client) flush(f *file) error {
if proto.IsHot(c.volType) {
if proto.IsHot(c.volType) || proto.IsStorageClassReplica(f.storageClass) {
return c.ec.Flush(f.ino)
} else {
if f.fileWriter != nil {
@ -1652,7 +1656,7 @@ func (c *client) truncate(f *file, size int) error {
}
func (c *client) write(f *file, offset int, data []byte, flags int) (n int, err error) {
if proto.IsHot(c.volType) {
if proto.IsHot(c.volType) || proto.IsStorageClassReplica(f.storageClass) {
c.ec.GetStreamer(f.ino).SetParentInode(f.pino) // set the parent inode
checkFunc := func() error {
if !c.mw.EnableQuota {
@ -1679,8 +1683,8 @@ func (c *client) write(f *file, offset int, data []byte, flags int) (n int, err
}
func (c *client) read(f *file, offset int, data []byte) (n int, err error) {
if proto.IsHot(c.volType) {
n, err = c.ec.Read(f.ino, data, offset, len(data))
if proto.IsHot(c.volType) || proto.IsStorageClassReplica(f.storageClass) {
n, err = c.ec.Read(f.ino, data, offset, len(data), f.storageClass)
} else {
n, err = f.fileReader.Read(c.ctx(c.id, f.ino), data, offset, len(data))
}

7
metanode/partition_op_extent.go
View File

@ -114,6 +114,13 @@ func (mp *metaPartition) ExtentAppendWithCheck(req *proto.AppendExtentKeyWithChe
return
}
if !proto.IsStorageClassReplica(req.StorageClass) {
log.LogErrorf("ExtentAppendWithCheck wrong storage class [%v]", req.StorageClass)
err = errors.New(fmt.Sprintf("ExtentAppendWithCheck wrong storage class [%v]", req.StorageClass))
reply := []byte(err.Error())
p.PacketErrorWithBody(status, reply)
return
}
//TODO:if storage type is not ssd , update extent key by CacheExtentAppendWithCheck
// check volume's Type: if volume's type is cold, cbfs' extent can be modify/add only when objextent exist
//if proto.IsCold(mp.volType) {

2
objectnode/api_handler_multipart.go
View File

@ -382,7 +382,7 @@ func (o *ObjectNode) uploadPartCopyHandler(w http.ResponseWriter, r *http.Reques
}
reader, writer := io.Pipe()
go func() {
err = srcVol.readFile(srcFileInfo.Inode, size, srcObject, writer, fb, cl)
err = srcVol.readFile(srcFileInfo.Inode, size, srcObject, writer, fb, cl, srcFileInfo.StorageClass)
if err != nil {
log.LogErrorf("uploadPartCopyHandler: read srcObj err(%v): requestId(%v) srcVol(%v) path(%v)",
err, GetRequestID(r), srcBucket, srcObject)

2
objectnode/api_handler_object.go
View File

@ -290,7 +290,7 @@ func (o *ObjectNode) getObjectHandler(w http.ResponseWriter, r *http.Request) {
// read file
start = time.Now()
err = vol.readFile(fileInfo.Inode, fileSize, param.Object(), writer, offset, size)
err = vol.readFile(fileInfo.Inode, fileSize, param.Object(), writer, offset, size, fileInfo.StorageClass)
span.AppendTrackLog("file.r", start, err)
if err != nil {
log.LogErrorf("getObjectHandler: read file fail: requestID(%v) volume(%v) path(%v) offset(%v) size(%v) err(%v)",

1
objectnode/fs.go
View File

@ -34,6 +34,7 @@ type FSFileInfo struct {
Expires string
Metadata map[string]string `graphql:"-"` // User-defined metadata
RetainUntilDate string
StorageClass uint32
}
type Prefixes []string

92
objectnode/fs_volume.go
View File

@ -700,7 +700,11 @@ func (v *Volume) PutObject(path string, reader io.Reader, opt *PutFileOption) (f
}()
md5Hash := md5.New()
if err = v.ec.OpenStream(invisibleTempDataInode.Inode, true); err != nil {
var isCache = false
if proto.IsStorageClassBlobStore(invisibleTempDataInode.StorageClass) {
isCache = true
}
if err = v.ec.OpenStream(invisibleTempDataInode.Inode, true, isCache); err != nil {
log.LogErrorf("PutObject: open stream fail: volume(%v) path(%v) inode(%v) err(%v)",
v.name, path, invisibleTempDataInode.Inode, err)
return
@ -712,7 +716,7 @@ func (v *Volume) PutObject(path string, reader io.Reader, opt *PutFileOption) (f
}
}()
if proto.IsCold(v.volType) {
if proto.IsCold(v.volType) || proto.IsStorageClassBlobStore(invisibleTempDataInode.StorageClass) {
if _, err = v.ebsWrite(invisibleTempDataInode.Inode, reader, md5Hash); err != nil {
log.LogErrorf("PutObject: ebs write fail: volume(%v) path(%v) inode(%v) err(%v)",
v.name, path, invisibleTempDataInode.Inode, err)
@ -804,7 +808,8 @@ func (v *Volume) PutObject(path string, reader io.Reader, opt *PutFileOption) (f
}
// apply new inode to dentry
err = v.applyInodeToDEntry(parentId, lastPathItem.Name, invisibleTempDataInode.Inode, false, fixedPath)
err = v.applyInodeToDEntry(parentId, lastPathItem.Name, invisibleTempDataInode.Inode, false,
fixedPath, invisibleTempDataInode.StorageClass)
if err != nil {
log.LogErrorf("PutObject: apply new inode to dentry fail: parentID(%v) name(%v) inode(%v) err(%v)",
parentId, lastPathItem.Name, invisibleTempDataInode.Inode, err)
@ -818,7 +823,8 @@ func (v *Volume) PutObject(path string, reader io.Reader, opt *PutFileOption) (f
return fsInfo, nil
}
func (v *Volume) applyInodeToDEntry(parentId uint64, name string, inode uint64, isCompleteMultipart bool, fullPath string) (err error) {
func (v *Volume) applyInodeToDEntry(parentId uint64, name string, inode uint64, isCompleteMultipart bool,
fullPath string, storageClass uint32) (err error) {
var existMode uint32
_, existMode, err = v.mw.Lookup_ll(parentId, name) // exist object inode
if err != nil && err != syscall.ENOENT {
@ -844,7 +850,7 @@ func (v *Volume) applyInodeToDEntry(parentId uint64, name string, inode uint64,
// current implementation doesn't support object versioning, so uploading a object with a key already existed in bucket
// is implemented with replacing the old one instead.
// refer: https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html
if err = v.applyInodeToExistDentry(parentId, name, inode, isCompleteMultipart, fullPath); err != nil {
if err = v.applyInodeToExistDentry(parentId, name, inode, isCompleteMultipart, fullPath, storageClass); err != nil {
log.LogErrorf("applyInodeToDEntry: apply inode to exist dentry fail: parentID(%v) name(%v) inode(%v) err(%v)",
parentId, name, inode, err)
return
@ -1035,7 +1041,11 @@ func (v *Volume) WritePart(path string, multipartId string, partId uint16, reade
etag string
md5Hash = md5.New()
)
if err = v.ec.OpenStream(tempInodeInfo.Inode, true); err != nil {
var isCache = false
if proto.IsStorageClassBlobStore(tempInodeInfo.StorageClass) {
isCache = true
}
if err = v.ec.OpenStream(tempInodeInfo.Inode, true, isCache); err != nil {
log.LogErrorf("WritePart: data open stream fail: volume(%v) path(%v) multipartID(%v) partID(%v) inode(%v) err(%v)",
v.name, path, multipartId, partId, tempInodeInfo.Inode, err)
return nil, err
@ -1046,7 +1056,7 @@ func (v *Volume) WritePart(path string, multipartId string, partId uint16, reade
v.name, path, multipartId, partId, tempInodeInfo.Inode, closeErr)
}
}()
if proto.IsCold(v.volType) {
if proto.IsCold(v.volType) || proto.IsStorageClassBlobStore(tempInodeInfo.StorageClass) {
if size, err = v.ebsWrite(tempInodeInfo.Inode, reader, md5Hash); err != nil {
log.LogErrorf("WritePart: ebs write fail: volume(%v) inode(%v) multipartID(%v) partID(%v) err(%v)",
v.name, tempInodeInfo.Inode, multipartId, partId, err)
@ -1196,7 +1206,7 @@ func (v *Volume) CompleteMultipart(path, multipartID string, multipartInfo *prot
// merge complete extent keys
var size uint64
var fileOffset uint64
if proto.IsCold(v.volType) {
if proto.IsCold(v.volType) || proto.IsStorageClassBlobStore(completeInodeInfo.StorageClass) {
completeObjExtentKeys := make([]proto.ObjExtentKey, 0)
for _, part := range parts {
var objExtents []proto.ObjExtentKey
@ -1293,7 +1303,8 @@ func (v *Volume) CompleteMultipart(path, multipartID string, multipartInfo *prot
}
// apply new inode to dentry
if err = v.applyInodeToDEntry(parentId, filename, completeInodeInfo.Inode, true, path); err != nil {
if err = v.applyInodeToDEntry(parentId, filename, completeInodeInfo.Inode, true,
path, completeInodeInfo.StorageClass); err != nil {
log.LogErrorf("CompleteMultipart: apply inode to dentry fail: volume(%v) multipartID(%v) parentId(%v) "+
"fileName(%v) inode(%v) err(%v)", v.name, multipartID, parentId, filename, completeInodeInfo.Inode, err)
return
@ -1403,7 +1414,7 @@ func (v *Volume) streamWrite(inode uint64, reader io.Reader, h hash.Hash, storag
}
func (v *Volume) appendInodeHash(h hash.Hash, inode uint64, total uint64, preAllocatedBuf []byte) (err error) {
if err = v.ec.OpenStream(inode, false); err != nil {
if err = v.ec.OpenStream(inode, false, false); err != nil {
log.LogErrorf("appendInodeHash: data open stream fail: inode(%v) err(%v)",
inode, err)
return
@ -1431,7 +1442,8 @@ func (v *Volume) appendInodeHash(h hash.Hash, inode uint64, total uint64, preAll
if uint64(size) > rest {
size = int(rest)
}
n, err = v.ec.Read(inode, buf, offset, size)
//no reference to this appendInodeHash
n, err = v.ec.Read(inode, buf, offset, size, proto.StorageClass_Unspecified)
if err != nil && err != io.EOF {
log.LogErrorf("appendInodeHash: data read fail, inode(%v) offset(%v) size(%v) err(%v)", inode, offset, size, err)
return
@ -1460,7 +1472,8 @@ func (v *Volume) applyInodeToNewDentry(parentID uint64, name string, inode uint6
return
}
func (v *Volume) applyInodeToExistDentry(parentID uint64, name string, inode uint64, isCompleteMultipart bool, fullPath string) (err error) {
func (v *Volume) applyInodeToExistDentry(parentID uint64, name string, inode uint64, isCompleteMultipart bool,
fullPath string, storageClass uint32) (err error) {
var oldInode uint64
oldInode, err = v.mw.DentryUpdate_ll(parentID, name, inode, fullPath)
if err != nil {
@ -1476,7 +1489,7 @@ func (v *Volume) applyInodeToExistDentry(parentID uint64, name string, inode uin
// concurrent completeMultipart request: temporary data security check
if isCompleteMultipart {
isSameExtent, err := v.referenceExtentKey(oldInode, inode)
isSameExtent, err := v.referenceExtentKey(oldInode, inode, storageClass)
if err != nil {
return err
}
@ -1532,8 +1545,12 @@ func (v *Volume) loadUserDefinedMetadata(inode uint64) (metadata map[string]stri
return
}
func (v *Volume) readFile(inode, inodeSize uint64, path string, writer io.Writer, offset, size uint64) (err error) {
if err = v.ec.OpenStream(inode, false); err != nil {
func (v *Volume) readFile(inode, inodeSize uint64, path string, writer io.Writer, offset, size uint64, storageClass uint32) (err error) {
var isCache = false
if proto.IsStorageClassBlobStore(storageClass) {
isCache = true
}
if err = v.ec.OpenStream(inode, false, isCache); err != nil {
log.LogErrorf("readFile: data open stream fail, Inode(%v) err(%v)", inode, err)
return err
}
@ -1542,9 +1559,8 @@ func (v *Volume) readFile(inode, inodeSize uint64, path string, writer io.Writer
log.LogErrorf("readFile: data close stream fail: inode(%v) err(%v)", inode, closeErr)
}
}()
if proto.IsHot(v.volType) {
return v.read(inode, inodeSize, path, writer, offset, size)
if proto.IsHot(v.volType) || proto.IsStorageClassReplica(storageClass) {
return v.read(inode, inodeSize, path, writer, offset, size, storageClass)
} else {
return v.readEbs(inode, inodeSize, path, writer, offset, size)
}
@ -1599,7 +1615,7 @@ func (v *Volume) readEbs(inode, inodeSize uint64, path string, writer io.Writer,
return nil
}
func (v *Volume) read(inode, inodeSize uint64, path string, writer io.Writer, offset, size uint64) error {
func (v *Volume) read(inode, inodeSize uint64, path string, writer io.Writer, offset, size uint64, storageClass uint32) error {
upper := size + offset
if upper > inodeSize {
upper = inodeSize - offset
@ -1620,7 +1636,7 @@ func (v *Volume) read(inode, inodeSize uint64, path string, writer io.Writer, of
if err != nil {
return err
}
n, err = v.ec.Read(inode, tmp, off, readSize)
n, err = v.ec.Read(inode, tmp, off, readSize, storageClass)
if err != nil && err != io.EOF {
log.LogErrorf("ReadFile: data read fail: volume(%v) path(%v) inode(%v) offset(%v) size(%v) err(%v)",
v.name, path, inode, offset, size, err)
@ -1659,7 +1675,7 @@ func (v *Volume) ReadFile(path string, writer io.Writer, offset, size uint64) er
return err
}
return v.readFile(ino, inoInfo.Size, path, writer, offset, size)
return v.readFile(ino, inoInfo.Size, path, writer, offset, size, inoInfo.StorageClass)
}
func (v *Volume) ObjectMeta(path string) (info *FSFileInfo, xattr *proto.XAttrInfo, err error) {
@ -1787,6 +1803,7 @@ func (v *Volume) ObjectMeta(path string) (info *FSFileInfo, xattr *proto.XAttrIn
Expires: expires,
Metadata: metadata,
RetainUntilDate: retainUntilDate,
StorageClass: inoInfo.StorageClass,
}
return
}
@ -2604,7 +2621,11 @@ func (v *Volume) CopyFile(sv *Volume, sourcePath, targetPath, metaDirective stri
log.LogErrorf("CopyFile: copy source path file size greater than 5GB, source path(%v), target path(%v)", sourcePath, targetPath)
return nil, syscall.EFBIG
}
if err = sv.ec.OpenStream(sInode, false); err != nil {
var isCache = false
if proto.IsStorageClassBlobStore(sInodeInfo.StorageClass) {
isCache = true
}
if err = sv.ec.OpenStream(sInode, false, isCache); err != nil {
log.LogErrorf("CopyFile: open source path stream fail, source path(%v) source path inode(%v) err(%v)",
sourcePath, sInode, err)
return
@ -2768,7 +2789,11 @@ func (v *Volume) CopyFile(sv *Volume, sourcePath, targetPath, metaDirective stri
}
}()
if err = v.ec.OpenStream(tInodeInfo.Inode, true); err != nil {
isCache = false
if proto.IsStorageClassBlobStore(tInodeInfo.StorageClass) {
isCache = true
}
if err = v.ec.OpenStream(tInodeInfo.Inode, true, isCache); err != nil {
return
}
defer func() {
@ -2797,11 +2822,11 @@ func (v *Volume) CopyFile(sv *Volume, sourcePath, targetPath, metaDirective stri
var ebsReader *blobstore.Reader
var tctx context.Context
var ebsWriter *blobstore.Writer
if proto.IsCold(sv.volType) {
if proto.IsCold(sv.volType) || proto.IsStorageClassBlobStore(sInodeInfo.StorageClass) {
sctx = context.Background()
ebsReader = v.getEbsReader(sInode)
}
if proto.IsCold(v.volType) {
if proto.IsCold(v.volType) || proto.IsStorageClassBlobStore(tInodeInfo.StorageClass) {
tctx = context.Background()
ebsWriter = v.getEbsWriter(tInodeInfo.Inode)
}
@ -2815,16 +2840,16 @@ func (v *Volume) CopyFile(sv *Volume, sourcePath, targetPath, metaDirective stri
readSize = rest
}
buf = buf[:readSize]
if proto.IsCold(sv.volType) {
if proto.IsCold(sv.volType) || proto.IsStorageClassBlobStore(sInodeInfo.StorageClass) {
readN, err = ebsReader.Read(sctx, buf, readOffset, readSize)
} else {
readN, err = sv.ec.Read(sInode, buf, readOffset, readSize)
readN, err = sv.ec.Read(sInode, buf, readOffset, readSize, sInodeInfo.StorageClass)
}
if err != nil && err != io.EOF {
return
}
if readN > 0 {
if proto.IsCold(v.volType) {
if proto.IsCold(v.volType) || proto.IsStorageClassBlobStore(tInodeInfo.StorageClass) {
writeN, err = ebsWriter.WriteWithoutPool(tctx, writeOffset, buf[:readN])
} else {
writeN, err = v.ec.Write(tInodeInfo.Inode, writeOffset, buf[:readN], 0, nil, tInodeInfo.StorageClass, false)
@ -2846,7 +2871,7 @@ func (v *Volume) CopyFile(sv *Volume, sourcePath, targetPath, metaDirective stri
}
}
// flush
if proto.IsCold(v.volType) {
if proto.IsCold(v.volType) || proto.IsStorageClassBlobStore(tInodeInfo.StorageClass) {
err = ebsWriter.FlushWithoutPool(tInodeInfo.Inode, tctx)
} else {
v.ec.Flush(tInodeInfo.Inode)
@ -2960,7 +2985,8 @@ func (v *Volume) CopyFile(sv *Volume, sourcePath, targetPath, metaDirective stri
}
// apply new inode to dentry
err = v.applyInodeToDEntry(tParentId, tLastName, tInodeInfo.Inode, false, targetPath)
err = v.applyInodeToDEntry(tParentId, tLastName, tInodeInfo.Inode, false,
targetPath, tInodeInfo.StorageClass)
if err != nil {
log.LogErrorf("CopyFile: apply inode to new dentry fail: path(%v) parentID(%v) name(%v) inode(%v) err(%v)",
targetPath, tParentId, tLastName, tInodeInfo.Inode, err)
@ -3029,7 +3055,7 @@ func NewVolume(config *VolumeConfig) (*Volume, error) {
OnTruncate: metaWrapper.Truncate,
OnRenewalForbiddenMigration: metaWrapper.RenewalForbiddenMigration,
}
if proto.IsCold(volumeInfo.VolType) {
if proto.IsCold(volumeInfo.VolType) || proto.VolSupportsBlobStore(volumeInfo.AllowedStorageClass) {
if blockCache != nil {
extentConfig.BcacheEnable = true
extentConfig.OnLoadBcache = blockCache.Get
@ -3150,9 +3176,9 @@ func safeConvertStrToUint16(str string) (uint16, error) {
return uint16(parsed), nil
}
func (v *Volume) referenceExtentKey(oldInode, inode uint64) (bool, error) {
func (v *Volume) referenceExtentKey(oldInode, inode uint64, storageClass uint32) (bool, error) {
// cold volume
if proto.IsCold(v.volType) {
if proto.IsCold(v.volType) || proto.IsStorageClassBlobStore(storageClass) {
_, _, _, oldObjExtents, err := v.mw.GetObjExtents(oldInode)
if err != nil {
log.LogErrorf("referenceExtentKey: meta get oldInode objextents fail: volume(%v) inode(%v) err(%v)",

2
preload/sdk/preloadsdk.go
View File

@ -447,7 +447,7 @@ func (c *PreLoadClient) preloadFileWorker(id int64, jobs <-chan fileInfo, wg *sy
log.LogDebugf("worker %v ready to preload(%v)", id, job.name)
ino := job.ino
//#1 open
c.ec.OpenStream(ino, true)
c.ec.OpenStream(ino, true, true)
//#2 write
var (
objExtents []proto.ObjExtentKey

9
proto/admin_proto.go
View File

@ -1458,6 +1458,15 @@ func IsStorageClassBlobStore(storageClass uint32) bool {
return storageClass == StorageClass_BlobStore
}
func VolSupportsBlobStore(allowedStorageClass []uint32) bool {
for _, storageClass := range allowedStorageClass {
if storageClass == StorageClass_BlobStore {
return true
}
}
return false
}
func GetVolTypeByStorageClass(storageClass uint32) (err error, volType int) {
if IsStorageClassReplica(storageClass) {
volType = VolumeTypeHot

1
proto/mount_options.go
View File

@ -343,4 +343,5 @@ type MountOptions struct {
DisableMountSubtype bool
// stream retry timeout
StreamRetryTimeout int
AllowedStorageClass []uint32
}

6
sdk/data/blobstore/reader.go
View File

@ -327,7 +327,8 @@ func (reader *Reader) readSliceRange(ctx context.Context, rs *rwSlice) (err erro
// 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))
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.
@ -401,9 +402,6 @@ func (reader *Reader) asyncCache(ctx context.Context, cacheKey string, objExtent
if streamer == nil {
log.LogWarnf("[asyncCache(L2)] streamer for ino %v is nil ", reader.ino)
return
} else {
//for get cache key in future
streamer.WorkAsCache()
}
reader.ec.Write(reader.ino, int(objExtentKey.FileOffset), buf, proto.FlagsCache, nil, reader.ec.CacheDpStorageClass, false)

8
sdk/data/blobstore/reader_test.go
View File

@ -217,7 +217,7 @@ func TestRead(t *testing.T) {
getObjFunc func(*meta.MetaWrapper, uint64) (uint64, uint64, []proto.ExtentKey, []proto.ObjExtentKey, error)
bcacheGetFunc func(*bcache.BcacheClient, string, []byte, uint64, uint32) (int, error)
checkDpExistFunc func(*stream.ExtentClient, uint64) error
readExtentFunc func(*stream.ExtentClient, uint64, *proto.ExtentKey, []byte, int, int) (int, error, bool)
readExtentFunc func(*stream.ExtentClient, uint64, *proto.ExtentKey, []byte, int, int, uint32) (int, error, bool)
ebsReadFunc func(*BlobStoreClient, context.Context, string, []byte, uint64, uint64, proto.ObjExtentKey) (int, error)
expectError error
}{
@ -387,7 +387,7 @@ func TestReadSliceRange(t *testing.T) {
extentKey proto.ExtentKey
bcacheGetFunc func(*bcache.BcacheClient, string, []byte, uint64, uint32) (int, error)
checkDpExistFunc func(*stream.ExtentClient, uint64) error
readExtentFunc func(*stream.ExtentClient, uint64, *proto.ExtentKey, []byte, int, int) (int, error, bool)
readExtentFunc func(*stream.ExtentClient, uint64, *proto.ExtentKey, []byte, int, int, uint32) (int, error, bool)
ebsReadFunc func(*BlobStoreClient, context.Context, string, []byte, uint64, uint64, proto.ObjExtentKey) (int, error)
expectError error
}{
@ -507,13 +507,13 @@ func MockEbscReadFalse(ebsc *BlobStoreClient, ctx context.Context, volName strin
}
func MockReadExtentTrue(client *stream.ExtentClient, inode uint64, ek *proto.ExtentKey,
data []byte, offset int, size int,
data []byte, offset int, size int, storageClass uint32,
) (read int, err error, b bool) {
return len("Hello world"), nil, true
}
func MockReadExtentFalse(client *stream.ExtentClient, inode uint64, ek *proto.ExtentKey,
data []byte, offset int, size int,
data []byte, offset int, size int, storageClass uint32,
) (read int, err error) {
return 0, errors.New("Read extent failed")
}

20
sdk/data/stream/extent_client.go
View File

@ -147,6 +147,7 @@ type ExtentConfig struct {
OnRenewalForbiddenMigration RenewalForbiddenMigrationFunc
CacheDpStorageClass uint32
AllowedStorageClass []uint32
}
type MultiVerMgr struct {
@ -266,7 +267,8 @@ func NewExtentClient(config *ExtentConfig) (client *ExtentClient, err error) {
limit := 0
retry:
client.dataWrapper, err = wrapper.NewDataPartitionWrapper(client, config.Volume, config.Masters, config.Preload, config.MinWriteAbleDataPartitionCnt, config.VerReadSeq)
client.dataWrapper, err = wrapper.NewDataPartitionWrapper(client, config.Volume, config.Masters,
config.Preload, config.MinWriteAbleDataPartitionCnt, config.VerReadSeq, config.AllowedStorageClass)
if err != nil {
log.LogErrorf("NewExtentClient: new data partition wrapper failed: volume(%v) mayRetry(%v) err(%v)",
config.Volume, limit, err)
@ -424,22 +426,22 @@ func (client *ExtentClient) UpdateLatestVer(verList *proto.VolVersionInfoList) (
}
// Open request shall grab the lock until request is sent to the request channel
func (client *ExtentClient) OpenStream(inode uint64, openForWrite bool) error {
func (client *ExtentClient) OpenStream(inode uint64, openForWrite, isCache bool) error {
client.streamerLock.Lock()
s, ok := client.streamers[inode]
if !ok {
s = NewStreamer(client, inode, openForWrite)
s = NewStreamer(client, inode, openForWrite, isCache)
client.streamers[inode] = s
}
return s.IssueOpenRequest()
}
// Open request shall grab the lock until request is sent to the request channel
func (client *ExtentClient) OpenStreamWithCache(inode uint64, needBCache, openForWrite bool) error {
func (client *ExtentClient) OpenStreamWithCache(inode uint64, needBCache, openForWrite, isCache bool) error {
client.streamerLock.Lock()
s, ok := client.streamers[inode]
if !ok {
s = NewStreamer(client, inode, openForWrite)
s = NewStreamer(client, inode, openForWrite, isCache)
client.streamers[inode] = s
if !client.disableMetaCache && needBCache {
client.streamerList.PushFront(inode)
@ -607,7 +609,7 @@ func (client *ExtentClient) Flush(inode uint64) error {
return s.IssueFlushRequest()
}
func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int) (read int, err error) {
func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int, storageClass uint32) (read int, err error) {
// log.LogErrorf("======> ExtentClient Read Enter, inode(%v), len(data)=(%v), offset(%v), size(%v).", inode, len(data), offset, size)
// t1 := time.Now()
beg := time.Now()
@ -646,14 +648,14 @@ func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int
clientMetric.WithLabelValues("Read_Flush").Observe(float64(time.Since(beg).Microseconds()))
beg = time.Now()
read, err = s.read(data, offset, size)
read, err = s.read(data, offset, size, storageClass)
clientMetric.WithLabelValues("Read_read").Observe(float64(time.Since(beg).Microseconds()))
// log.LogErrorf("======> ExtentClient Read Exit, inode(%v), time[%v us].", inode, time.Since(t1).Microseconds())
readReqCountMetric.Dec()
return
}
func (client *ExtentClient) ReadExtent(inode uint64, ek *proto.ExtentKey, data []byte, offset int, size int) (read int, err error, isStream bool) {
func (client *ExtentClient) ReadExtent(inode uint64, ek *proto.ExtentKey, data []byte, offset int, size int, storageClass uint32) (read int, err error, isStream bool) {
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("read-extent", err, bgTime, 1)
@ -674,7 +676,7 @@ func (client *ExtentClient) ReadExtent(inode uint64, ek *proto.ExtentKey, data [
if err != nil {
return
}
reader, err = s.GetExtentReader(ek)
reader, err = s.GetExtentReader(ek, storageClass)
if err != nil {
return
}

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

@ -116,7 +116,7 @@ type ExtentHandler struct {
stop chan struct{}
sync.Once
mediaType uint32
storageClass uint32
}
// NewExtentHandler returns a new extent handler.
@ -137,7 +137,7 @@ func NewExtentHandler(stream *Streamer, offset int, storeMode int, size int, sto
stop: make(chan struct{}),
meetLimitedIoError: false,
verUpdate: make(chan uint64),
mediaType: proto.GetMediaTypeByStorageClass(storageClass),
storageClass: proto.GetMediaTypeByStorageClass(storageClass),
}
go eh.receiver()
@ -171,7 +171,7 @@ func (eh *ExtentHandler) write(data []byte, offset, size int, direct bool) (ek *
// If this write request is not continuous, and cannot be merged
// into the extent handler, just close it and return error.
// In this case, the caller should try to create a new extent handler.
if proto.IsHot(eh.stream.client.volumeType) {
if proto.IsHot(eh.stream.client.volumeType) || proto.IsStorageClassReplica(eh.storageClass) {
if eh.fileOffset+eh.size != offset || eh.size+size > util.ExtentSize ||
(eh.storeMode == proto.TinyExtentType && eh.size+size > blksize) {
@ -459,14 +459,15 @@ func (eh *ExtentHandler) appendExtentKey() (err error) {
if eh.key != nil {
if eh.dirty {
if proto.IsCold(eh.stream.client.volumeType) && eh.status == ExtentStatusError {
if (proto.IsCold(eh.stream.client.volumeType) || proto.IsStorageClassBlobStore(eh.storageClass)) &&
eh.status == ExtentStatusError {
return
}
var status int
ekey := *eh.key
doAppend := func() (err error) {
discard := eh.stream.extents.Append(&ekey, true)
status, err = eh.stream.client.appendExtentKey(eh.stream.parentInode, eh.inode, ekey, discard, eh.stream.isCache, proto.GetStorageClassByMediaType(eh.mediaType))
status, err = eh.stream.client.appendExtentKey(eh.stream.parentInode, eh.inode, ekey, discard, eh.stream.isCache, eh.storageClass)
if atomic.LoadInt32(&eh.stream.needUpdateVer) > 0 {
if errUpdateExtents := eh.stream.GetExtentsForceRefresh(); errUpdateExtents != nil {
log.LogErrorf("action[appendExtentKey] inode %v GetExtents err %v errUpdateExtents %v", eh.stream.inode, err, errUpdateExtents)
@ -562,12 +563,9 @@ func (eh *ExtentHandler) waitForFlush() (err error) {
}
func (eh *ExtentHandler) recoverPacket(packet *Packet) error {
// NOTE: if last result code is limited io error
// allow client to retry indefinitely
if !eh.meetLimitedIoError {
packet.errCount++
}
if packet.errCount >= MaxPacketErrorCount || proto.IsCold(eh.stream.client.volumeType) {
packet.errCount++
if packet.errCount >= MaxPacketErrorCount || proto.IsCold(eh.stream.client.volumeType) ||
proto.IsStorageClassBlobStore(eh.storageClass) {
return errors.New(fmt.Sprintf("recoverPacket failed: reach max error limit, eh(%v) packet(%v)", eh, packet))
}
@ -582,7 +580,7 @@ func (eh *ExtentHandler) recoverPacket(packet *Packet) error {
if eh.meetLimitedIoError {
extentType = eh.storeMode
}
handler = NewExtentHandler(eh.stream, int(packet.KernelOffset), extentType, 0, eh.mediaType)
handler = NewExtentHandler(eh.stream, int(packet.KernelOffset), extentType, 0, eh.storageClass)
handler.setClosed()
}
handler.pushToRequest(packet)
@ -614,7 +612,7 @@ func (eh *ExtentHandler) allocateExtent() (err error) {
for i := 0; i < MaxSelectDataPartitionForWrite; i++ {
if eh.key == nil {
if dp, err = eh.stream.client.dataWrapper.GetDataPartitionForWrite(exclude, eh.mediaType); err != nil {
if dp, err = eh.stream.client.dataWrapper.GetDataPartitionForWrite(exclude, eh.storageClass); err != nil {
log.LogWarnf("allocateExtent: failed to get write data partition, eh(%v) exclude(%v), clear exclude and try again!", eh, exclude)
exclude = make(map[string]struct{})
continue
@ -745,8 +743,8 @@ func (eh *ExtentHandler) setRecovery() bool {
}
func (eh *ExtentHandler) setError() bool {
// log.LogDebugf("action[ExtentHandler.setError] stack (%v)", string(debug.Stack()))
if proto.IsHot(eh.stream.client.volumeType) {
// log.LogDebugf("action[ExtentHandler.setError] stack (%v)", string(debug.Stack()))
if proto.IsHot(eh.stream.client.volumeType) || proto.IsStorageClassReplica(eh.storageClass) {
atomic.StoreInt32(&eh.stream.status, StreamerError)
}
return atomic.CompareAndSwapInt32(&eh.status, ExtentStatusRecovery, ExtentStatusError)

29
sdk/data/stream/stream_reader.go
View File

@ -59,12 +59,13 @@ type Streamer struct {
}
type bcacheKey struct {
cacheKey string
extentKey *proto.ExtentKey
cacheKey string
extentKey *proto.ExtentKey
storageClass uint32
}
// NewStreamer returns a new streamer.
func NewStreamer(client *ExtentClient, inode uint64, openForWrite bool) *Streamer {
func NewStreamer(client *ExtentClient, inode uint64, openForWrite, isCache bool) *Streamer {
s := new(Streamer)
s.client = client
s.inode = inode
@ -78,6 +79,7 @@ func NewStreamer(client *ExtentClient, inode uint64, openForWrite bool) *Streame
s.verSeq = client.multiVerMgr.latestVerSeq
s.extents.verSeq = client.multiVerMgr.latestVerSeq
s.openForWrite = openForWrite
s.isCache = isCache
go s.server()
go s.asyncBlockCache()
return s
@ -111,7 +113,7 @@ func (s *Streamer) GetExtentsForceRefresh() error {
// GetExtentReader returns the extent reader.
// TODO: use memory pool
func (s *Streamer) GetExtentReader(ek *proto.ExtentKey) (*ExtentReader, error) {
func (s *Streamer) GetExtentReader(ek *proto.ExtentKey, storageClass uint32) (*ExtentReader, error) {
partition, err := s.client.dataWrapper.GetDataPartition(ek.PartitionId)
if err != nil {
return nil, err
@ -123,7 +125,7 @@ func (s *Streamer) GetExtentReader(ek *proto.ExtentKey) (*ExtentReader, error) {
}
retryRead := true
if proto.IsCold(s.client.volumeType) {
if proto.IsCold(s.client.volumeType) || proto.IsStorageClassBlobStore(storageClass) {
retryRead = false
}
@ -132,7 +134,7 @@ func (s *Streamer) GetExtentReader(ek *proto.ExtentKey) (*ExtentReader, error) {
return reader, nil
}
func (s *Streamer) read(data []byte, offset int, size int) (total int, err error) {
func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32) (total int, err error) {
var (
readBytes int
reader *ExtentReader
@ -217,7 +219,7 @@ func (s *Streamer) read(data []byte, offset int, size int) (total int, err error
}
// read extent
reader, err = s.GetExtentReader(req.ExtentKey)
reader, err = s.GetExtentReader(req.ExtentKey, storageClass)
if err != nil {
log.LogErrorf("action[streamer.read] req %v err %v", req, err)
break
@ -229,7 +231,7 @@ func (s *Streamer) read(data []byte, offset int, size int) (total int, err error
// do nothing
} else {
select {
case s.pendingCache <- bcacheKey{cacheKey: cacheKey, extentKey: req.ExtentKey}:
case s.pendingCache <- bcacheKey{cacheKey: cacheKey, extentKey: req.ExtentKey, storageClass: storageClass}:
if s.exceedBlockSize(req.ExtentKey.Size) {
atomic.AddInt32(&s.client.inflightL1BigBlock, 1)
}
@ -277,7 +279,7 @@ func (s *Streamer) asyncBlockCache() {
} else {
data = make([]byte, ek.Size)
}
reader, _ := s.GetExtentReader(ek)
reader, err := s.GetExtentReader(ek, pending.storageClass)
fullReq := NewExtentRequest(int(ek.FileOffset), int(ek.Size), data, ek)
readBytes, err := reader.Read(fullReq)
if err != nil || readBytes != len(data) {
@ -310,9 +312,8 @@ func (s *Streamer) asyncBlockCache() {
}
func (s *Streamer) exceedBlockSize(size uint32) bool {
return size > bcache.BigExtentSize
}
func (s *Streamer) WorkAsCache() {
s.isCache = true
if size > bcache.BigExtentSize {
return true
}
return false
}

10
sdk/data/stream/stream_writer.go
View File

@ -376,7 +376,7 @@ begin:
log.LogDebugf("action[streamer.write] inode [%v] latest seq [%v] extentkey seq [%v] info [%v] before compare seq",
s.inode, s.verSeq, req.ExtentKey.GetSeq(), req.ExtentKey)
if req.ExtentKey.GetSeq() == s.verSeq {
writeSize, err = s.doOverwrite(req, direct)
writeSize, err = s.doOverwrite(req, direct, storageClass)
if err == proto.ErrCodeVersionOp {
log.LogDebugf("action[streamer.write] write need version update")
if err = s.GetExtentsForceRefresh(); err != nil {
@ -464,7 +464,7 @@ func (s *Streamer) doDirectWriteByAppend(req *ExtentRequest, direct bool, op uin
}
retry := true
if proto.IsCold(s.client.volumeType) {
if proto.IsCold(s.client.volumeType) || proto.IsStorageClassBlobStore(storageClass) {
retry = false
}
log.LogDebugf("action[doDirectWriteByAppend] inode %v data process", s.inode)
@ -609,7 +609,7 @@ func (s *Streamer) doDirectWriteByAppend(req *ExtentRequest, direct bool, op uin
return
}
func (s *Streamer) doOverwrite(req *ExtentRequest, direct bool) (total int, err error) {
func (s *Streamer) doOverwrite(req *ExtentRequest, direct bool, storageClass uint32) (total int, err error) {
var dp *wrapper.DataPartition
err = s.flush()
@ -637,7 +637,7 @@ func (s *Streamer) doOverwrite(req *ExtentRequest, direct bool) (total int, err
}
retry := true
if proto.IsCold(s.client.volumeType) {
if proto.IsCold(s.client.volumeType) || proto.IsStorageClassBlobStore(storageClass) {
retry = false
}
@ -838,7 +838,7 @@ func (s *Streamer) doWriteAppendEx(data []byte, offset, size int, direct bool, r
storeMode = s.GetStoreMod(offset, size)
log.LogDebugf("doWriteAppendEx enter: ino(%v) offset(%v) size(%v) storeMode(%v)", s.inode, offset, size, storeMode)
if proto.IsHot(s.client.volumeType) {
if proto.IsHot(s.client.volumeType) || proto.IsStorageClassReplica(storageClass) {
if reUseEk {
if isLastEkVerNotEqual := s.tryInitExtentHandlerByLastEk(offset, size); isLastEkVerNotEqual {
log.LogDebugf("doWriteAppendEx enter: ino(%v) tryInitExtentHandlerByLastEk worked but seq not equal", s.inode)

15
sdk/data/wrapper/wrapper.go
View File

@ -84,10 +84,13 @@ type Wrapper struct {
SimpleClient SimpleClientInfo
IsSnapshotEnabled bool
allowedStorageClass []uint32
}
// NewDataPartitionWrapper returns a new data partition wrapper.
func NewDataPartitionWrapper(client SimpleClientInfo, volName string, masters []string, preload bool, minWritableDataPartitionCnt int, verReadSeq uint64) (w *Wrapper, err error) {
func NewDataPartitionWrapper(client SimpleClientInfo, volName string, masters []string, preload bool,
minWriteAbleDataPartitionCnt int, verReadSeq uint64, allowedStorageClass []uint32) (w *Wrapper, err error) {
log.LogInfof("action[NewDataPartitionWrapper] verReadSeq %v", verReadSeq)
w = new(Wrapper)
@ -103,6 +106,7 @@ func NewDataPartitionWrapper(client SimpleClientInfo, volName string, masters []
if w.minWritableDataPartitionCnt < 0 {
w.minWritableDataPartitionCnt = DefaultMinWritableDataPartitionCnt
}
w.allowedStorageClass = allowedStorageClass
if w.LocalIp, err = ump.GetLocalIpAddr(); err != nil {
err = errors.Trace(err, "NewDataPartitionWrapper:")
@ -369,7 +373,7 @@ func (w *Wrapper) updateDataPartitionByRsp(forceUpdate bool, refreshPolicy Refre
ClientWrapper: w,
}
}
if proto.IsCold(w.volType) {
if proto.IsCold(w.volType) || proto.VolSupportsBlobStore(w.allowedStorageClass) {
w.clearPartitions()
}
rwPartitionGroups := make([]*DataPartition, 0)
@ -384,8 +388,9 @@ func (w *Wrapper) updateDataPartitionByRsp(forceUpdate bool, refreshPolicy Refre
}
log.LogInfof("updateDataPartition: dp(%v)", dp)
w.replaceOrInsertPartition(dp)
// do not insert preload dp in cold vol
if proto.IsCold(w.volType) && proto.IsPreLoadDp(dp.PartitionType) {
if (proto.IsCold(w.volType) || proto.VolSupportsBlobStore(w.allowedStorageClass)) && proto.IsPreLoadDp(dp.PartitionType) {
continue
}
if dp.Status == proto.ReadWrite {
@ -491,7 +496,7 @@ func (w *Wrapper) AllocatePreLoadDataPartition(volName string, count int, capaci
rwPartitionGroups := make([]*DataPartition, 0)
for _, partition := range dpv.DataPartitions {
dp := convert(partition)
if proto.IsCold(w.volType) && !proto.IsPreLoadDp(dp.PartitionType) {
if (proto.IsCold(w.volType) || proto.VolSupportsBlobStore(w.allowedStorageClass)) && !proto.IsPreLoadDp(dp.PartitionType) {
continue
}
log.LogInfof("updateDataPartition: dp(%v)", dp)
@ -533,7 +538,7 @@ func (w *Wrapper) replaceOrInsertPartition(dp *DataPartition) {
// GetDataPartition returns the data partition based on the given partition ID.
func (w *Wrapper) GetDataPartition(partitionID uint64) (*DataPartition, error) {
dp, ok := w.tryGetPartition(partitionID)
if !ok && !proto.IsCold(w.volType) { // cache miss && hot volume
if !ok && (!proto.IsCold(w.volType) || !proto.VolSupportsBlobStore(w.allowedStorageClass)) { // cache miss && hot volume
err := w.getDataPartitionFromMaster(partitionID)
if err == nil {
dp, ok = w.tryGetPartition(partitionID)