mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-09 05:13:10 +00:00
440 lines
17 KiB
Go
440 lines
17 KiB
Go
// Copyright 2018 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 stream
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"runtime"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/cubefs/cubefs/client/blockcache/bcache"
|
|
"github.com/cubefs/cubefs/proto"
|
|
"github.com/cubefs/cubefs/util"
|
|
"github.com/cubefs/cubefs/util/buf"
|
|
"github.com/cubefs/cubefs/util/exporter"
|
|
"github.com/cubefs/cubefs/util/log"
|
|
"github.com/cubefs/cubefs/util/stat"
|
|
)
|
|
|
|
// One inode corresponds to one streamer. All the requests to the same inode will be queued.
|
|
// TODO rename streamer here is not a good name as it also handles overwrites, not just stream write.
|
|
type Streamer struct {
|
|
client *ExtentClient
|
|
inode uint64
|
|
parentInode uint64
|
|
status int32
|
|
refcnt int32
|
|
idle int // how long there is no new request
|
|
traversed int // how many times the streamer is traversed
|
|
extents *ExtentCache
|
|
once sync.Once
|
|
handler *ExtentHandler // current open handler
|
|
dirtylist *DirtyExtentList // dirty handlers
|
|
dirty bool // whether current open handler is in the dirty list
|
|
isOpen bool
|
|
needBCache bool
|
|
request chan interface{} // request channel, write/flush/close
|
|
done chan struct{} // stream writer is being closed
|
|
writeLock sync.Mutex
|
|
inflightEvictL1cache sync.Map
|
|
pendingCache chan bcacheKey
|
|
verSeq uint64
|
|
needUpdateVer int32
|
|
isCache bool
|
|
openForWrite bool
|
|
rdonly bool
|
|
aheadReadEnable bool
|
|
aheadReadWindow *AheadReadWindow
|
|
fullPath string
|
|
}
|
|
|
|
type bcacheKey struct {
|
|
cacheKey string
|
|
extentKey *proto.ExtentKey
|
|
storageClass uint32
|
|
}
|
|
|
|
// NewStreamer returns a new streamer.
|
|
func NewStreamer(client *ExtentClient, inode uint64, openForWrite, isCache bool, fullPath string) *Streamer {
|
|
s := new(Streamer)
|
|
s.client = client
|
|
s.inode = inode
|
|
s.parentInode = 0
|
|
s.extents = NewExtentCache(inode)
|
|
s.request = make(chan interface{}, reqChanSize)
|
|
s.done = make(chan struct{})
|
|
s.dirtylist = NewDirtyExtentList()
|
|
s.isOpen = true
|
|
s.pendingCache = make(chan bcacheKey, 1)
|
|
s.verSeq = client.multiVerMgr.latestVerSeq
|
|
s.extents.verSeq = client.multiVerMgr.latestVerSeq
|
|
s.openForWrite = openForWrite
|
|
s.isCache = isCache
|
|
s.fullPath = fullPath
|
|
|
|
if log.EnableDebug() {
|
|
log.LogDebugf("NewStreamer: streamer(%v), reqChSize %d", s, reqChanSize)
|
|
}
|
|
if s.openForWrite {
|
|
err := s.client.forbiddenMigration(s.inode)
|
|
if err != nil {
|
|
log.LogWarnf("ino(%v) forbiddenMigration failed err %v", s.inode, err.Error())
|
|
s.setError()
|
|
}
|
|
}
|
|
if client.AheadRead != nil {
|
|
s.aheadReadEnable = client.AheadRead.enable
|
|
s.aheadReadWindow = NewAheadReadWindow(client.AheadRead, s)
|
|
}
|
|
go s.server()
|
|
go s.asyncBlockCache()
|
|
return s
|
|
}
|
|
|
|
func (s *Streamer) SetParentInode(inode uint64) {
|
|
s.parentInode = inode
|
|
}
|
|
|
|
func (s *Streamer) SetFullPath(fullPath string) {
|
|
s.fullPath = fullPath
|
|
}
|
|
|
|
// String returns the string format of the streamer.
|
|
func (s *Streamer) String() string {
|
|
return fmt.Sprintf("Streamer{ino(%v), fullPath(%v), refcnt(%v), isOpen(%v) openForWrite(%v), inflight(%v), eh(%v) addr(%p)}",
|
|
s.inode, s.fullPath, atomic.LoadInt32(&s.refcnt), s.isOpen, s.openForWrite, len(s.request), s.handler, s)
|
|
}
|
|
|
|
// TODO should we call it RefreshExtents instead?
|
|
func (s *Streamer) GetExtents(isMigration bool) error {
|
|
if s.client.disableMetaCache || !s.needBCache {
|
|
return s.extents.RefreshForce(s.inode, false, s.client.getExtents, s.isCache, s.openForWrite, isMigration)
|
|
}
|
|
|
|
return s.extents.Refresh(s.inode, s.client.getExtents, s.isCache, s.openForWrite, isMigration)
|
|
}
|
|
|
|
func (s *Streamer) GetExtentsForce() error {
|
|
return s.extents.RefreshForce(s.inode, false, s.client.getExtents, s.isCache, s.openForWrite, false)
|
|
}
|
|
|
|
func (s *Streamer) GetExtentsForceRefresh() error {
|
|
return s.extents.RefreshForce(s.inode, true, s.client.getExtents, s.isCache, s.openForWrite, false)
|
|
}
|
|
|
|
// GetExtentReader returns the extent reader.
|
|
// TODO: use memory pool
|
|
func (s *Streamer) GetExtentReader(ek *proto.ExtentKey, storageClass uint32) (*ExtentReader, error) {
|
|
partition, err := s.client.dataWrapper.GetDataPartition(ek.PartitionId)
|
|
if err != nil {
|
|
if strings.Contains(err.Error(), "no writable data partition") {
|
|
partition, err = s.client.dataWrapper.GetDataPartition(ek.PartitionId)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
} else {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
if partition.IsDiscard {
|
|
log.LogWarnf("GetExtentReader: datapartition %v is discard", partition.PartitionID)
|
|
return nil, DpDiscardError
|
|
}
|
|
|
|
retryRead := true
|
|
if proto.IsCold(s.client.volumeType) || proto.IsStorageClassBlobStore(storageClass) {
|
|
retryRead = false
|
|
}
|
|
|
|
enableFollowerRead := s.client.dataWrapper.FollowerRead() && !s.client.dataWrapper.InnerReq()
|
|
reader := NewExtentReader(s.inode, ek, partition, enableFollowerRead, retryRead)
|
|
reader.maxRetryTimeout = s.client.streamRetryTimeout
|
|
return reader, nil
|
|
}
|
|
|
|
func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32) (total int, err error) {
|
|
var (
|
|
readBytes int
|
|
reader *ExtentReader
|
|
requests []*ExtentRequest
|
|
revisedRequests []*ExtentRequest
|
|
)
|
|
log.LogDebugf("action[streamer.read] ino(%v) offset %v size %v", s.inode, offset, size)
|
|
ctx := context.Background()
|
|
if s.client.readLimit() {
|
|
s.client.readLimiter.Wait(ctx)
|
|
}
|
|
s.client.LimitManager.ReadAlloc(ctx, size)
|
|
requests = s.extents.PrepareReadRequests(offset, size, data)
|
|
for _, req := range requests {
|
|
if req.ExtentKey == nil {
|
|
continue
|
|
}
|
|
if req.ExtentKey.PartitionId == 0 || req.ExtentKey.ExtentId == 0 {
|
|
s.writeLock.Lock()
|
|
if err = s.IssueFlushRequest(); err != nil {
|
|
log.LogErrorf("[read] failed to issue flush request, ino(%v) offset(%v) size(%v) err(%v) req(%v)", s.inode, offset, size, err, req)
|
|
s.writeLock.Unlock()
|
|
return 0, err
|
|
}
|
|
revisedRequests = s.extents.PrepareReadRequests(offset, size, data)
|
|
s.writeLock.Unlock()
|
|
break
|
|
}
|
|
}
|
|
|
|
if revisedRequests != nil {
|
|
requests = revisedRequests
|
|
}
|
|
|
|
filesize, _ := s.extents.Size()
|
|
log.LogDebugf("read: ino(%v) requests(%v) filesize(%v)", s.inode, requests, filesize)
|
|
for _, req := range requests {
|
|
log.LogDebugf("action[streamer.read] req %v", req)
|
|
if req.ExtentKey == nil {
|
|
zeros := make([]byte, len(req.Data))
|
|
copy(req.Data, zeros)
|
|
|
|
if req.FileOffset+req.Size > filesize {
|
|
if req.FileOffset > filesize {
|
|
return
|
|
}
|
|
req.Size = filesize - req.FileOffset
|
|
total += req.Size
|
|
err = io.EOF
|
|
return
|
|
}
|
|
|
|
// Reading a hole, just fill zero
|
|
total += req.Size
|
|
log.LogDebugf("Stream read hole: ino(%v) req(%v) total(%v)", s.inode, req, total)
|
|
} else {
|
|
log.LogDebugf("Stream read: ino(%v) req(%v) s.needBCache(%v) s.client.bcacheEnable(%v) aheadReadEnable(%v)", s.inode, req, s.needBCache, s.client.bcacheEnable, s.aheadReadEnable)
|
|
if s.aheadReadEnable && req.ExtentKey.Size > util.CacheReadBlockSize {
|
|
readBytes, err = s.aheadRead(req)
|
|
if err == nil && readBytes == req.Size {
|
|
total += readBytes
|
|
continue
|
|
}
|
|
log.LogDebugf("aheadRead inode(%v) FileOffset(%v) readBytes(%v) reqSize(%v) err(%v)", s.inode, req.FileOffset, readBytes, req.Size, err)
|
|
}
|
|
// if s.needBCache {
|
|
// bcacheMetric := exporter.NewCounter("fileReadL1Cache")
|
|
// bcacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
|
|
// }
|
|
|
|
// skip hole,ek is not nil,read block cache firstly
|
|
log.LogDebugf("Stream read: ino(%v) req(%v) s.client.bcacheEnable(%v) s.client.bcacheOnlyForNotSSD(%v) s.needBCache(%v)",
|
|
s.inode, req, s.client.bcacheEnable, s.client.bcacheOnlyForNotSSD, s.needBCache)
|
|
if s.client.bcacheEnable && s.needBCache && filesize <= bcache.MaxFileSize {
|
|
cacheKey := util.GenerateRepVolKey(s.client.volumeName, s.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, req.ExtentKey.FileOffset)
|
|
inodeInfo, err := s.client.getInodeInfo(s.inode)
|
|
if err != nil {
|
|
log.LogErrorf("Streamer read: getInodeInfo failed. ino(%v) req(%v) err(%v)", s.inode, req, err)
|
|
return 0, err
|
|
}
|
|
if !s.client.bcacheOnlyForNotSSD || (s.client.bcacheOnlyForNotSSD && inodeInfo.StorageClass != proto.StorageClass_Replica_SSD) {
|
|
log.LogDebugf("Streamer read from bcache, ino(%v) storageClass(%v) s.client.bcacheEnable(%v) bcacheOnlyForNotSSD(%v)",
|
|
s.inode, proto.StorageClassString(inodeInfo.StorageClass), s.client.bcacheEnable, s.client.bcacheOnlyForNotSSD)
|
|
offset := req.FileOffset - int(req.ExtentKey.FileOffset)
|
|
if s.client.loadBcache != nil {
|
|
bcacheMetric := exporter.NewCounter("fileReadL1Cache")
|
|
bcacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
|
|
readBytes, err = s.client.loadBcache(s.client.volumeName, cacheKey, req.Data, uint64(offset), uint32(req.Size))
|
|
if err == nil && readBytes == req.Size {
|
|
total += req.Size
|
|
bcacheMetric := exporter.NewCounter("fileReadL1CacheHit")
|
|
bcacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
|
|
log.LogDebugf("TRACE Stream read. hit blockCache: cacheKey(%v) inode(%v) "+
|
|
"offset(%v) readBytes(%v) goroutine(%v)", cacheKey, s.inode, offset, readBytes, getGoid())
|
|
continue
|
|
}
|
|
bcacheMissMetric := exporter.NewCounter("fileReadL1CacheMiss")
|
|
bcacheMissMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
|
|
}
|
|
log.LogDebugf("TRACE Stream read. miss blockCache cacheKey(%v) inode(%v) offset(%v) size(%v)"+
|
|
"goroutine(%v)", cacheKey, s.inode, offset, req.Size, getGoid())
|
|
} else {
|
|
log.LogDebugf("Streamer not read from bcache, ino(%v) storageClass(%v) s.client.bcacheEnable(%v) bcacheOnlyForNotSSD(%v)",
|
|
s.inode, proto.StorageClassString(inodeInfo.StorageClass), s.client.bcacheEnable, s.client.bcacheOnlyForNotSSD)
|
|
}
|
|
log.LogDebugf("TRACE Stream read. miss blockCache cacheKey(%v) loadBcache(%v)", cacheKey, s.client.loadBcache)
|
|
} else if s.enableRemoteCache() {
|
|
inodeInfo, err := s.client.getInodeInfo(s.inode)
|
|
if err != nil {
|
|
log.LogErrorf("Streamer read: getInodeInfo failed. ino(%v) req(%v) err(%v)", s.inode, req, err)
|
|
return 0, err
|
|
}
|
|
|
|
if s.client.forceRemoteCache || !s.client.RemoteCache.remoteCacheOnlyForNotSSD || (s.client.RemoteCache.remoteCacheOnlyForNotSSD && inodeInfo.StorageClass != proto.StorageClass_Replica_SSD) {
|
|
log.LogDebugf("Streamer read from remoteCache, ino(%v) enableRemoteCache(true) storageClass(%v) remoteCacheOnlyForNotSSD(%v) forceRemoteCache(%v)",
|
|
s.inode, proto.StorageClassString(inodeInfo.StorageClass), s.client.RemoteCache.remoteCacheOnlyForNotSSD, s.client.forceRemoteCache)
|
|
var cacheReadRequests []*CacheReadRequest
|
|
cacheReadRequests, err = s.prepareCacheRequests(uint64(offset), uint64(size), data, inodeInfo.Generation)
|
|
if err == nil {
|
|
var read int
|
|
remoteCacheMetric := exporter.NewCounter("readRemoteCache")
|
|
remoteCacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
|
|
if read, err = s.readFromRemoteCache(ctx, uint64(offset), uint64(size), cacheReadRequests); err == nil {
|
|
remoteCacheHitMetric := exporter.NewCounter("readRemoteCacheHit")
|
|
remoteCacheHitMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
|
|
return read, err
|
|
}
|
|
}
|
|
if !proto.IsFlashNodeLimitError(err) {
|
|
log.LogWarnf("Stream read: readFromRemoteCache failed: ino(%v) offset(%v) size(%v), err(%v)", s.inode, offset, size, err)
|
|
}
|
|
} else {
|
|
log.LogDebugf("Streamer not read from remoteCache, ino(%v) enableRemoteCache(true) storageClass(%v) remoteCacheOnlyForNotSSD(%v)",
|
|
s.inode, proto.StorageClassString(inodeInfo.StorageClass), s.client.RemoteCache.remoteCacheOnlyForNotSSD)
|
|
}
|
|
} else {
|
|
log.LogDebugf("Streamer not read from remoteCache, ino(%v) enableRemoteCache(false)", s.inode)
|
|
}
|
|
|
|
// read extent
|
|
reader, err = s.GetExtentReader(req.ExtentKey, storageClass)
|
|
if err != nil {
|
|
log.LogErrorf("action[streamer.read] req %v err %v", req, err)
|
|
break
|
|
}
|
|
|
|
if s.client.bcacheEnable && s.needBCache && filesize <= bcache.MaxFileSize {
|
|
inodeInfo, err := s.client.getInodeInfo(s.inode)
|
|
if err != nil {
|
|
log.LogErrorf("Streamer read: getInodeInfo failed. ino(%v) req(%v) err(%v)", s.inode, req, err)
|
|
return 0, err
|
|
}
|
|
cacheKey := util.GenerateRepVolKey(s.client.volumeName, s.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, req.ExtentKey.FileOffset)
|
|
// limit big block cache
|
|
if s.exceedBlockSize(req.ExtentKey.Size) && atomic.LoadInt32(&s.client.inflightL1BigBlock) > 10 {
|
|
// do nothing
|
|
} else if !s.client.bcacheOnlyForNotSSD || (s.client.bcacheOnlyForNotSSD && inodeInfo.StorageClass != proto.StorageClass_Replica_SSD) {
|
|
select {
|
|
case s.pendingCache <- bcacheKey{cacheKey: cacheKey, extentKey: req.ExtentKey}:
|
|
log.LogDebugf("action[streamer.read] blockCache send cacheKey %v for ino(%v) offset %v size %v goroutine(%v)",
|
|
cacheKey, s.inode, req.FileOffset-int(req.ExtentKey.FileOffset), req.Size, getGoid())
|
|
if s.exceedBlockSize(req.ExtentKey.Size) {
|
|
atomic.AddInt32(&s.client.inflightL1BigBlock, 1)
|
|
}
|
|
default:
|
|
log.LogDebugf("action[streamer.read] blockCache discard cacheKey %v for ino(%v) offset %v size %v goroutine(%v)",
|
|
cacheKey, s.inode, req.FileOffset-int(req.ExtentKey.FileOffset), req.Size, getGoid())
|
|
}
|
|
}
|
|
}
|
|
bgTime := stat.BeginStat()
|
|
readBytes, err = reader.Read(req)
|
|
stat.EndStat("ReadFromDataNode", err, bgTime, 1)
|
|
log.LogDebugf("TRACE Stream read: ino(%v) req(%v) readBytes(%v) err(%v)", s.inode, req, readBytes, err)
|
|
|
|
total += readBytes
|
|
|
|
if err != nil || readBytes < req.Size {
|
|
if total == 0 {
|
|
log.LogErrorf("Stream read: ino(%v) req(%v) readBytes(%v) err(%v)", s.inode, req, readBytes, err)
|
|
}
|
|
break
|
|
}
|
|
}
|
|
}
|
|
log.LogDebugf("action[streamer.read] offset %v size %v exit", offset, size)
|
|
return
|
|
}
|
|
|
|
func (s *Streamer) asyncBlockCache() {
|
|
if !s.needBCache || !s.isOpen {
|
|
return
|
|
}
|
|
t := time.NewTicker(3 * time.Second)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case pending := <-s.pendingCache:
|
|
ek := pending.extentKey
|
|
cacheKey := pending.cacheKey
|
|
begin := time.Now()
|
|
log.LogDebugf("asyncBlockCache: cacheKey=(%v) ek=(%v)", cacheKey, ek)
|
|
|
|
// read full extent
|
|
var data []byte
|
|
if ek.Size == bcache.MaxBlockSize {
|
|
data = buf.BCachePool.Get()
|
|
} else {
|
|
data = make([]byte, ek.Size)
|
|
}
|
|
reader, err := s.GetExtentReader(ek, pending.storageClass)
|
|
if err != nil {
|
|
log.LogErrorf("asyncBlockCache: GetExtentReader err %v", err)
|
|
return
|
|
}
|
|
fullReq := NewExtentRequest(int(ek.FileOffset), int(ek.Size), data, ek)
|
|
metric := exporter.NewTPCnt("bcache-read-cachedata")
|
|
readBytes, err := reader.Read(fullReq)
|
|
if err != nil || readBytes != len(data) {
|
|
metric.SetWithLabels(err, map[string]string{exporter.Vol: s.client.volumeName})
|
|
log.LogWarnf("asyncBlockCache: Stream read full extent error. fullReq(%v) readBytes(%v) err(%v)", fullReq, readBytes, err)
|
|
if ek.Size == bcache.MaxBlockSize {
|
|
buf.BCachePool.Put(data)
|
|
}
|
|
if s.exceedBlockSize(ek.Size) {
|
|
atomic.AddInt32(&s.client.inflightL1BigBlock, -1)
|
|
}
|
|
return
|
|
}
|
|
log.LogDebugf("TRACE read. read blockCache cacheKey(%v) len_buf(%v) cost %v,", cacheKey, len(data), time.Since(begin).String())
|
|
metric.SetWithLabels(err, map[string]string{exporter.Vol: s.client.volumeName})
|
|
if s.client.cacheBcache != nil {
|
|
begin = time.Now()
|
|
s.client.cacheBcache(s.client.volumeName, cacheKey, data)
|
|
log.LogDebugf("TRACE read. read blockCache cacheKey(%v) len_buf(%v) cost %v,", cacheKey, len(data), time.Since(begin).String())
|
|
}
|
|
if ek.Size == bcache.MaxBlockSize {
|
|
buf.BCachePool.Put(data)
|
|
}
|
|
if s.exceedBlockSize(ek.Size) {
|
|
atomic.AddInt32(&s.client.inflightL1BigBlock, -1)
|
|
}
|
|
case <-t.C:
|
|
if !s.isOpen {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Streamer) exceedBlockSize(size uint32) bool {
|
|
return size > bcache.BigExtentSize
|
|
}
|
|
|
|
func getGoid() int {
|
|
var buf [64]byte
|
|
n := runtime.Stack(buf[:], false)
|
|
idField := strings.Fields(string(buf[:n]))[1]
|
|
gid, _ := strconv.Atoi(idField)
|
|
return gid
|
|
}
|
|
|
|
func (s *Streamer) UpdateStringPath(fullPath string) {
|
|
s.fullPath = fullPath
|
|
}
|
