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

fix(bcache): reduce memory consumption for bcache-service

Browse Source 
close:#22775703

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2024-11-20 11:32:46 +08:00 committed by zhumingze1108
parent 1a20920be2
commit 639a696545
12 changed files with 441 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
blobstore/util/bytespool/pool.go
View File

@ -14,7 +14,10 @@
package bytespool
import "sync"
import (
"bytes"
"sync"
)
const (
zeroSize = 1 << 14 // 16 KB
@ -122,3 +125,17 @@ func msb(size int) byte {
v |= v >> 16
return debruijinPosition[(v*0x07C4ACDD)>>27]
}
func AllocWithZeroLengthBuffer(size int) *bytes.Buffer {
byteSlice := AllocWithZeroLength(size)
buffer := bytes.NewBuffer(byteSlice)
return buffer
}
func AllocWithZeroLength(size int) []byte {
if pool := GetPool(size); pool != nil {
b := pool.Get().([]byte)
return b[:0]
}
return make([]byte, 0, size)
}

46
client/blockcache/bcache/client.go
View File

@ -20,8 +20,10 @@ import (
"sync"
"time"
"github.com/cubefs/cubefs/blobstore/util/bytespool"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/stat"
)
@ -51,13 +53,12 @@ func NewBcacheClient() *BcacheClient {
return client
}
func (c *BcacheClient) Get(key string, buf []byte, offset uint64, size uint32) (int, error) {
func (c *BcacheClient) Get(vol, key string, buf []byte, offset uint64, size uint32) (int, error) {
var err error
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("bcache-get", err, bgTime, 1)
}()
req := &GetCacheRequest{
CacheKey: key,
Offset: offset,
@ -65,11 +66,16 @@ func (c *BcacheClient) Get(key string, buf []byte, offset uint64, size uint32) (
}
packet := NewBlockCachePacket()
packet.Opcode = OpBlockCacheGet
err = packet.MarshalData(req)
data, err := req.Marshal()
if err != nil {
log.LogDebugf("get block cache: req(%v) err(%v)", req.CacheKey, err)
return 0, err
}
defer func() {
bytespool.Free(data)
}()
packet.Data = data
packet.Size = uint32(len(packet.Data))
stat.EndStat("bcache-get-marshal", err, bgTime, 1)
conn, err := c.connPool.Get()
if err != nil {
@ -80,15 +86,18 @@ func (c *BcacheClient) Get(key string, buf []byte, offset uint64, size uint32) (
c.connPool.Put(conn)
}()
stat.EndStat("bcache-get-conn", err, bgTime, 1)
getCachePathMetric := exporter.NewTPCnt("bcache-get-cachepath")
err = packet.WriteToConn(*conn)
if err != nil {
log.LogDebugf("Failed to write to conn, req(%v) err(%v)", req.CacheKey, err)
getCachePathMetric.SetWithLabels(err, map[string]string{exporter.Vol: vol})
return 0, errors.NewErrorf("Failed to write to conn, req(%v) err(%v)", req.CacheKey, err)
}
stat.EndStat("bcache-get-writeconn", err, bgTime, 1)
err = packet.ReadFromConn(*conn, 1)
if err != nil {
log.LogDebugf("Failed to read from conn, req(%v), err(%v)", req.CacheKey, err)
getCachePathMetric.SetWithLabels(err, map[string]string{exporter.Vol: vol})
return 0, errors.NewErrorf("Failed to read from conn, req(%v), err(%v)", req.CacheKey, err)
}
stat.EndStat("bcache-get-readconn", err, bgTime, 1)
@ -97,27 +106,29 @@ func (c *BcacheClient) Get(key string, buf []byte, offset uint64, size uint32) (
if status != statusOK {
err = errors.New(packet.GetResultMsg())
log.LogDebugf("get block cache: req(%v) err(%v) result(%v)", req.CacheKey, err, packet.GetResultMsg())
getCachePathMetric.SetWithLabels(err, map[string]string{exporter.Vol: vol})
return 0, err
}
resp := new(GetCachePathResponse)
err = packet.UnmarshalData(resp)
err = resp.UnmarshalValue(packet.Data)
if err != nil {
log.LogDebugf("get block cache: req(%v) err(%v) PacketData(%v)", req.CacheKey, err, string(packet.Data))
getCachePathMetric.SetWithLabels(err, map[string]string{exporter.Vol: vol})
return 0, err
}
cachePath := resp.CachePath
stat.EndStat("bcache-get-meta", err, bgTime, 1)
getCachePathMetric.SetWithLabels(nil, map[string]string{exporter.Vol: vol})
readBgTime := stat.BeginStat()
subs := strings.Split(cachePath, "/")
if subs[len(subs)-1] != key {
log.LogDebugf("cacheKey(%v) cache path(%v) is not legal",
key, cachePath)
return 0, errors.NewErrorf("cacheKey(%v) cache path is not legal: %v", key, cachePath)
}
readCacheMetric := exporter.NewTPCnt("bcache-read-cachefile")
defer readCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: vol})
f, err := os.Open(cachePath)
if err != nil {
return 0, err
@ -137,25 +148,32 @@ func (c *BcacheClient) Get(key string, buf []byte, offset uint64, size uint32) (
return n, nil
}
func (c *BcacheClient) Put(key string, buf []byte) error {
func (c *BcacheClient) Put(vol, key string, buf []byte) error {
var err error
bgTime := stat.BeginStat()
cacheFileMetric := exporter.NewTPCnt("bcache-put-cachefile")
defer func() {
stat.EndStat("bcache-put", err, bgTime, 1)
cacheFileMetric.SetWithLabels(err, map[string]string{exporter.Vol: vol})
}()
req := &PutCacheRequest{
CacheKey: key,
Data: buf,
VolName: vol,
}
packet := NewBlockCachePacket()
packet.Opcode = OpBlockCachePut
err = packet.MarshalData(req)
data, err := req.Marshal()
defer func() {
bytespool.Free(data)
}()
if err != nil {
log.LogDebugf("put block cache: req(%v) err(%v)", req.CacheKey, err)
return err
}
packet.Data = data
packet.Size = uint32(len(packet.Data))
conn, err := c.connPool.Get()
if err != nil {
log.LogDebugf("put block cache: get Conn failed, req(%v) err(%v)", req.CacheKey, err)
@ -190,12 +208,16 @@ func (c *BcacheClient) Evict(key string) error {
req := &DelCacheRequest{CacheKey: key}
packet := NewBlockCachePacket()
packet.Opcode = OpBlockCacheDel
err := packet.MarshalData(req)
data, err := req.Marshal()
if err != nil {
log.LogDebugf("del block cache: req(%v) err(%v)", req.CacheKey, err)
return err
}
defer func() {
bytespool.Free(data)
}()
packet.Data = data
packet.Size = uint32(len(packet.Data))
conn, err := c.connPool.Get()
if err != nil {
log.LogDebugf("del block cache: get Conn failed, req(%v) err(%v)", req.CacheKey, err)

45
client/blockcache/bcache/manage.go
View File

@ -34,6 +34,7 @@ import (
"time"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/stat"
)
@ -54,7 +55,7 @@ type ReadCloser interface {
}
type BcacheManager interface {
cache(key string, data []byte, direct bool)
cache(vol, key string, data []byte, direct bool)
read(key string, offset uint64, len uint32) (io.ReadCloser, error)
queryCachePath(key string, offset uint64, len uint32) (string, error)
load(key string) (ReadCloser, error)
@ -89,11 +90,13 @@ func newBcacheManager(conf *bcacheConfig) BcacheManager {
}
bm := &bcacheManager{
bstore: make([]*DiskStore, len(cacheDirs)),
bcacheKeys: make(map[string]*list.Element),
lrulist: list.New(),
blockSize: conf.BlockSize,
pending: make(chan waitFlush, 1024),
bstore: make([]*DiskStore, len(cacheDirs)),
bcacheKeys: make(map[string]*list.Element),
lrulist: list.New(),
blockSize: conf.BlockSize,
pending: make(chan waitFlush, 1024),
cacheMetaCount: exporter.NewGaugeVec("bcache_meta_count", "", []string{"volName", "disk", "type"}),
vol: conf.Vol,
}
index := 0
for cacheDir, cacheSize := range dirSizeMap {
@ -126,11 +129,13 @@ type waitFlush struct {
type bcacheManager struct {
sync.RWMutex
bcacheKeys map[string]*list.Element
lrulist *list.List
bstore []*DiskStore
blockSize uint32
pending chan waitFlush
bcacheKeys map[string]*list.Element
lrulist *list.List
bstore []*DiskStore
blockSize uint32
pending chan waitFlush
cacheMetaCount *exporter.GaugeVec
vol string
}
func encryptXOR(data []byte) {
@ -172,7 +177,7 @@ func (bm *bcacheManager) getCachePath(key string) (string, error) {
return cachePath, nil
}
func (bm *bcacheManager) cache(key string, data []byte, direct bool) {
func (bm *bcacheManager) cache(vol, key string, data []byte, direct bool) {
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("Cache:Write", nil, bgTime, 1)
@ -180,20 +185,20 @@ func (bm *bcacheManager) cache(key string, data []byte, direct bool) {
}()
log.LogDebugf("TRACE cache. key(%v) len(%v) direct(%v)", key, len(data), direct)
if direct {
bm.cacheDirect(key, data)
bm.cacheDirect(vol, key, data)
return
}
select {
case bm.pending <- waitFlush{Key: key, Data: data}:
default:
log.LogDebugf("pending chan is full,skip memory. key =%v,len=%v bytes", key, len(data))
bm.cacheDirect(key, data)
bm.cacheDirect(vol, key, data)
}
}
func (bm *bcacheManager) cacheDirect(key string, data []byte) {
func (bm *bcacheManager) cacheDirect(vol, key string, data []byte) {
diskKv := bm.selectDiskKv(key)
if diskKv.flushKey(key, data) == nil {
if diskKv.flushKey(vol, key, data) == nil {
bm.Lock()
item := &cacheItem{
key: key,
@ -319,6 +324,8 @@ func (bm *bcacheManager) spaceManager() {
if 1-useRatio < store.freeLimit || files > int64(store.limit) {
bm.freeSpace(store, 1-useRatio, files)
}
bm.cacheMetaCount.SetWithLabelValues(float64(useRatio), bm.vol, store.dir, "useRatio")
bm.cacheMetaCount.SetWithLabelValues(float64(files), bm.vol, store.dir, "files")
}
case <-tmpTicker.C:
for _, store := range bm.bstore {
@ -469,7 +476,7 @@ func (bm *bcacheManager) flush() {
pending := <-bm.pending
diskKv := bm.selectDiskKv(pending.Key)
log.LogDebugf("flush data,key(%v), dir(%v)", pending.Key, diskKv.dir)
if diskKv.flushKey(pending.Key, pending.Data) == nil {
if diskKv.flushKey(bm.vol, pending.Key, pending.Data) == nil {
bm.Lock()
item := &cacheItem{
key: pending.Key,
@ -584,11 +591,13 @@ func (d *DiskStore) checkBuildCacheDir(dir string) {
//
//}
func (d *DiskStore) flushKey(key string, data []byte) error {
func (d *DiskStore) flushKey(vol, key string, data []byte) error {
var err error
bgTime := stat.BeginStat()
metric := exporter.NewTPCnt("cacheToDisk")
defer func() {
stat.EndStat("Cache:Write:FlushData", err, bgTime, 1)
metric.SetWithLabels(err, map[string]string{exporter.Vol: vol})
}()
cachePath := d.buildCachePath(key, d.dir)
info, err := os.Stat(cachePath)

199
client/blockcache/bcache/packet.go
View File

@ -15,6 +15,7 @@
package bcache
import (
"bytes"
"encoding/binary"
"encoding/json"
"errors"
@ -25,6 +26,7 @@ import (
"syscall"
"time"
"github.com/cubefs/cubefs/blobstore/util/bytespool"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/buf"
)
@ -48,6 +50,86 @@ var Buffers *buf.BufferPool
type PutCacheRequest struct {
CacheKey string `json:"key"`
Data []byte `json:"data"`
VolName string `json:"volName"`
}
func (req *PutCacheRequest) Marshal() (result []byte, err error) {
buff := bytespool.AllocWithZeroLengthBuffer(4 + len(req.CacheKey) + 4 + len(req.Data) + 4 + len(req.VolName))
// cache key
err = binary.Write(buff, binary.BigEndian, uint32(len(req.CacheKey)))
if err != nil {
return
}
_, err = buff.Write([]byte(req.CacheKey))
if err != nil {
return
}
// data
err = binary.Write(buff, binary.BigEndian, uint32(len(req.Data)))
if err != nil {
return
}
_, err = buff.Write(req.Data)
if err != nil {
return
}
// vol
err = binary.Write(buff, binary.BigEndian, uint32(len(req.VolName)))
if err != nil {
return
}
_, err = buff.Write([]byte(req.VolName))
if err != nil {
return
}
return buff.Bytes(), nil
}
func (req *PutCacheRequest) UnmarshalValue(data []byte) (err error) {
reader := bytes.NewReader(data)
var keyLen uint32
err = binary.Read(reader, binary.BigEndian, &keyLen)
if err != nil {
return
}
keyBytes := bytespool.Alloc(int(keyLen))
defer bytespool.Free(keyBytes)
_, err = reader.Read(keyBytes)
if err != nil {
return err
}
req.CacheKey = string(keyBytes)
var dataLen uint32
err = binary.Read(reader, binary.BigEndian, &dataLen)
if err != nil {
return err
}
dataBytes := bytespool.Alloc(int(dataLen))
defer bytespool.Free(dataBytes)
err = binary.Read(reader, binary.BigEndian, &dataBytes)
if err != nil {
return err
}
req.Data = dataBytes
var volNameLen uint32
err = binary.Read(reader, binary.BigEndian, &volNameLen)
if err != nil {
return err
}
volNameBytes := bytespool.Alloc(int(volNameLen))
defer bytespool.Free(volNameBytes)
_, err = reader.Read(volNameBytes)
if err != nil {
return err
}
req.VolName = string(volNameBytes)
return nil
}
type GetCacheRequest struct {
@ -56,18 +138,131 @@ type GetCacheRequest struct {
Size uint32 `json:"size"`
}
func (req *GetCacheRequest) Marshal() (result []byte, err error) {
buff := bytespool.AllocWithZeroLengthBuffer(4 + len(req.CacheKey) + 8 + 4)
// cache key
err = binary.Write(buff, binary.BigEndian, uint32(len(req.CacheKey)))
if err != nil {
return
}
_, err = buff.Write([]byte(req.CacheKey))
if err != nil {
return
}
err = binary.Write(buff, binary.BigEndian, &req.Offset)
if err != nil {
return
}
err = binary.Write(buff, binary.BigEndian, &req.Size)
if err != nil {
return
}
return buff.Bytes(), nil
}
func (req *GetCacheRequest) UnmarshalValue(data []byte) (err error) {
reader := bytes.NewReader(data)
var keyLen uint32
err = binary.Read(reader, binary.BigEndian, &keyLen)
if err != nil {
return
}
keyBytes := bytespool.Alloc(int(keyLen))
defer bytespool.Free(keyBytes)
_, err = reader.Read(keyBytes)
if err != nil {
return err
}
req.CacheKey = string(keyBytes)
err = binary.Read(reader, binary.BigEndian, &req.Offset)
if err != nil {
return err
}
err = binary.Read(reader, binary.BigEndian, &req.Size)
if err != nil {
return err
}
return nil
}
type GetCachePathResponse struct {
CachePath string `json:"path"`
}
type GetCacheDataResponse struct {
Data []byte `json:"data"`
func (req *GetCachePathResponse) Marshal() (result []byte, err error) {
buff := bytespool.AllocWithZeroLengthBuffer(4 + len(req.CachePath))
// cache key
err = binary.Write(buff, binary.BigEndian, uint32(len(req.CachePath)))
if err != nil {
return
}
_, err = buff.Write([]byte(req.CachePath))
if err != nil {
return
}
return buff.Bytes(), nil
}
func (req *GetCachePathResponse) UnmarshalValue(data []byte) (err error) {
reader := bytes.NewReader(data)
var keyLen uint32
err = binary.Read(reader, binary.BigEndian, &keyLen)
if err != nil {
return
}
keyBytes := bytespool.Alloc(int(keyLen))
defer bytespool.Free(keyBytes)
_, err = reader.Read(keyBytes)
if err != nil {
return err
}
req.CachePath = string(keyBytes)
return nil
}
type DelCacheRequest struct {
CacheKey string `json:"key"`
}
func (req *DelCacheRequest) Marshal() (result []byte, err error) {
buff := bytespool.AllocWithZeroLengthBuffer(4 + len(req.CacheKey))
// cache key
err = binary.Write(buff, binary.BigEndian, uint32(len(req.CacheKey)))
if err != nil {
return
}
_, err = buff.Write([]byte(req.CacheKey))
if err != nil {
return
}
return buff.Bytes(), nil
}
func (req *DelCacheRequest) UnmarshalValue(data []byte) (err error) {
reader := bytes.NewReader(data)
var keyLen uint32
err = binary.Read(reader, binary.BigEndian, &keyLen)
if err != nil {
return
}
keyBytes := bytespool.Alloc(int(keyLen))
defer bytespool.Free(keyBytes)
_, err = reader.Read(keyBytes)
if err != nil {
return err
}
req.CacheKey = string(keyBytes)
return nil
}
type BlockCachePacket struct {
Magic uint8
Opcode uint8

88
client/blockcache/bcache/packet_test.go Normal file
View File

@ -0,0 +1,88 @@
package bcache
import (
"reflect"
"testing"
"github.com/cubefs/cubefs/blobstore/util/bytespool"
)
func TestPutCacheRequest(t *testing.T) {
putReq := PutCacheRequest{
CacheKey: "test_key",
Data: []byte("test_data"),
VolName: "test_vol",
}
data, err := putReq.Marshal()
if err != nil {
t.Fatalf("Marshal PutCacheRequest failed: %v", err)
}
var putReq2 PutCacheRequest
err = putReq2.UnmarshalValue(data)
if err != nil {
t.Fatalf("UnmarshalValue PutCacheRequest failed: %v", err)
}
if !reflect.DeepEqual(putReq, putReq2) {
t.Fatalf("result mismatch:\n\tpart1:%v\n\tpart2:%v", putReq, putReq2)
}
bytespool.Free(data)
}
func TestGetCacheRequest(t *testing.T) {
putReq := GetCacheRequest{
CacheKey: "test_key",
Offset: 1,
Size: 1024,
}
data, err := putReq.Marshal()
if err != nil {
t.Fatalf("Marshal PutCacheRequest failed: %v", err)
}
var putReq2 GetCacheRequest
err = putReq2.UnmarshalValue(data)
if err != nil {
t.Fatalf("UnmarshalValue PutCacheRequest failed: %v", err)
}
if !reflect.DeepEqual(putReq, putReq2) {
t.Fatalf("result mismatch:\n\tpart1:%v\n\tpart2:%v", putReq, putReq2)
}
bytespool.Free(data)
}
func TestGetCachePathResponse(t *testing.T) {
putReq := GetCachePathResponse{
CachePath: "test_key",
}
data, err := putReq.Marshal()
if err != nil {
t.Fatalf("Marshal PutCacheRequest failed: %v", err)
}
var putReq2 GetCachePathResponse
err = putReq2.UnmarshalValue(data)
if err != nil {
t.Fatalf("UnmarshalValue PutCacheRequest failed: %v", err)
}
if !reflect.DeepEqual(putReq, putReq2) {
t.Fatalf("result mismatch:\n\tpart1:%v\n\tpart2:%v", putReq, putReq2)
}
bytespool.Free(data)
}
func TestDelCacheRequest(t *testing.T) {
putReq := DelCacheRequest{
CacheKey: "test_key",
}
data, err := putReq.Marshal()
if err != nil {
t.Fatalf("Marshal PutCacheRequest failed: %v", err)
}
var putReq2 DelCacheRequest
err = putReq2.UnmarshalValue(data)
if err != nil {
t.Fatalf("UnmarshalValue PutCacheRequest failed: %v", err)
}
if !reflect.DeepEqual(putReq, putReq2) {
t.Fatalf("result mismatch:\n\tpart1:%v\n\tpart2:%v", putReq, putReq2)
}
bytespool.Free(data)
}

19
client/blockcache/bcache/service.go
View File

@ -15,7 +15,6 @@
package bcache
import (
"encoding/json"
"fmt"
"io"
"net"
@ -25,6 +24,7 @@ import (
"strconv"
"syscall"
"github.com/cubefs/cubefs/blobstore/util/bytespool"
"github.com/cubefs/cubefs/cmd/common"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/config"
@ -41,6 +41,8 @@ const (
CacheLimit = "cacheLimit"
CacheFree = "cacheFree"
BlockSize = "blockSize"
Vol = "vol"
Cluster = "Cluster"
MaxFileSize = 128 << 30
MaxBlockSize = 128 << 20
BigExtentSize = 32 << 20
@ -53,6 +55,7 @@ type bcacheConfig struct {
CacheSize int64
FreeRatio float32
Limit uint32
Vol string
}
type bcacheStore struct {
@ -216,13 +219,13 @@ func (s *bcacheStore) handlePacket(conn net.Conn, p *BlockCachePacket) (err erro
func (s *bcacheStore) opBlockCachePut(conn net.Conn, p *BlockCachePacket) (err error) {
req := &PutCacheRequest{}
if err = json.Unmarshal(p.Data, req); err != nil {
if err = req.UnmarshalValue(p.Data); err != nil {
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
s.response(conn, p)
err = errors.NewErrorf("req[%v],err[%v]", req, err.Error())
return
}
s.bcache.cache(req.CacheKey, req.Data, false)
s.bcache.cache(req.VolName, req.CacheKey, req.Data, false)
p.PacketOkReplay()
s.response(conn, p)
return
@ -230,7 +233,7 @@ func (s *bcacheStore) opBlockCachePut(conn net.Conn, p *BlockCachePacket) (err e
func (s *bcacheStore) opBlockCacheGet(conn net.Conn, p *BlockCachePacket) (err error) {
req := &GetCacheRequest{}
if err = json.Unmarshal(p.Data, req); err != nil {
if err = req.UnmarshalValue(p.Data); err != nil {
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
s.response(conn, p)
err = errors.NewErrorf("req[%v],err[%v]", req, string(p.Data))
@ -250,13 +253,16 @@ func (s *bcacheStore) opBlockCacheGet(conn net.Conn, p *BlockCachePacket) (err e
}
resp := &GetCachePathResponse{CachePath: cachePath}
reply, err := json.Marshal(resp)
reply, err := resp.Marshal()
if err != nil {
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
s.response(conn, p)
err = errors.NewErrorf("req[%v],err[%v]", req, string(p.Data))
return
}
defer func() {
bytespool.Free(reply)
}()
p.PacketOkWithBody(reply)
s.response(conn, p)
return
@ -264,7 +270,7 @@ func (s *bcacheStore) opBlockCacheGet(conn net.Conn, p *BlockCachePacket) (err e
func (s *bcacheStore) opBlockCacheEvict(conn net.Conn, p *BlockCachePacket) (err error) {
req := &DelCacheRequest{}
if err = json.Unmarshal(p.Data, req); err != nil {
if err = req.UnmarshalValue(p.Data); err != nil {
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
s.response(conn, p)
err = errors.NewErrorf("req[%v],err[%v]", req, err.Error())
@ -303,6 +309,7 @@ func (s *bcacheStore) parserConf(cfg *config.Config) (*bcacheConfig, error) {
cacheFree := cfg.GetString(CacheFree)
blockSize := cfg.GetString(BlockSize)
bconf.CacheDir = cacheDir
bconf.Vol = cfg.GetString(Vol)
if cacheDir == "" {
return nil, errors.NewErrorf("cacheDir is required.")
}

19
client/blockcache/cmd.go
View File

@ -33,6 +33,7 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/config"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/stat"
sysutil "github.com/cubefs/cubefs/util/sys"
@ -45,10 +46,11 @@ const (
ConfigKeyLogLevel = "logLevel"
ConfigKeyProfPort = "prof"
ConfigKeyWarnLogDir = "warnLogDir"
RoleBcache = "blockcache"
LoggerOutput = "output.log"
ConfigKeyCluster = "cluster"
ConfigKeyVol = "vol"
RoleBcache = "blockcache"
ModuleName = "bcache-service"
LoggerOutput = "output.log"
)
var (
@ -132,7 +134,13 @@ func main() {
logLevel := cfg.GetString(ConfigKeyLogLevel)
profPort := cfg.GetString(ConfigKeyProfPort)
umpDatadir := cfg.GetString(ConfigKeyWarnLogDir)
cluster := cfg.GetString(ConfigKeyCluster)
vol := cfg.GetString(ConfigKeyVol)
if vol == "" || cluster == "" {
fmt.Println("vol or cluster cannot be nil")
os.Exit(1)
}
// Init server instance with specified role configuration.
var (
server common.Server
@ -195,6 +203,9 @@ func main() {
}
stat.ClearStat()
exporter.Init(ModuleName, cfg)
exporter.RegistConsul(cluster, ModuleName, cfg)
defer func() {
outputFile.Sync()
outputFile.Close()

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

@ -302,7 +302,7 @@ func (reader *Reader) readSliceRange(ctx context.Context, rs *rwSlice) (err erro
// read local cache
if reader.enableBcache {
readN, err = reader.bc.Get(cacheKey, buf, rs.rOffset, rs.rSize)
readN, err = reader.bc.Get(reader.volName, cacheKey, buf, rs.rOffset, rs.rSize)
if err == nil {
reader.ec.BcacheHealth = true
if readN == int(rs.rSize) {
@ -412,7 +412,7 @@ func (reader *Reader) asyncCache(ctx context.Context, cacheKey string, objExtent
}
if reader.needCacheL1() {
reader.bc.Put(cacheKey, buf)
reader.bc.Put(reader.volName, cacheKey, buf)
}
log.LogDebugf("TRACE blobStore asyncCache(L1) Exit. cacheKey=%v", cacheKey)

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

@ -215,7 +215,7 @@ func TestRead(t *testing.T) {
close bool
readConcurrency int
getObjFunc func(*meta.MetaWrapper, uint64) (uint64, uint64, []proto.ExtentKey, []proto.ObjExtentKey, error)
bcacheGetFunc func(*bcache.BcacheClient, string, []byte, uint64, uint32) (int, error)
bcacheGetFunc func(*bcache.BcacheClient, string, string, []byte, uint64, uint32) (int, error)
checkDpExistFunc func(*stream.ExtentClient, uint64) error
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)
@ -385,7 +385,7 @@ func TestReadSliceRange(t *testing.T) {
testCase := []struct {
enableBcache bool
extentKey proto.ExtentKey
bcacheGetFunc func(*bcache.BcacheClient, string, []byte, uint64, uint32) (int, error)
bcacheGetFunc func(*bcache.BcacheClient, string, string, []byte, uint64, uint32) (int, error)
checkDpExistFunc func(*stream.ExtentClient, uint64) error
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)
@ -538,7 +538,7 @@ func MockWriteFalse(client *stream.ExtentClient, inode uint64, offset int, data
return 0, errors.New("Write failed")
}
func MockPutTrue(bc *bcache.BcacheClient, key string, buf []byte) error {
func MockPutTrue(bc *bcache.BcacheClient, vol, key string, buf []byte) error {
return nil
}
@ -546,10 +546,10 @@ func MockPutFalse(bc *bcache.BcacheClient, key string, buf []byte) error {
return errors.New("Bcache put failed")
}
func MockGetTrue(bc *bcache.BcacheClient, key string, buf []byte, offset uint64, size uint32) (int, error) {
func MockGetTrue(bc *bcache.BcacheClient, vol, key string, buf []byte, offset uint64, size uint32) (int, error) {
return int(size), nil
}
func MockGetFalse(bc *bcache.BcacheClient, key string, buf []byte, offset uint64, size uint32) (int, error) {
func MockGetFalse(bc *bcache.BcacheClient, vol, key string, buf []byte, offset uint64, size uint32) (int, error) {
return 0, errors.New("Bcache get failed")
}

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

@ -55,8 +55,8 @@ type (
GetExtentsFunc func(inode uint64, isCache bool, openForWrite bool, isMigration bool) (uint64, uint64, []proto.ExtentKey, error)
TruncateFunc func(inode, size uint64, fullPath string) error
EvictIcacheFunc func(inode uint64)
LoadBcacheFunc func(key string, buf []byte, offset uint64, size uint32) (int, error)
CacheBcacheFunc func(key string, buf []byte) error
LoadBcacheFunc func(vol, key string, buf []byte, offset uint64, size uint32) (int, error)
CacheBcacheFunc func(vol, key string, buf []byte) error
EvictBacheFunc func(key string) error
RenewalForbiddenMigrationFunc func(inode uint64) error
ForbiddenMigrationFunc func(inode uint64) error
@ -849,7 +849,7 @@ func (client *ExtentClient) ReadExtent(inode uint64, ek *proto.ExtentKey, data [
copy(buf, req.Data)
go func() {
log.LogDebugf("ReadExtent L2->L1 Enter cacheKey(%v),client.shouldBcache(%v),needCache(%v)", cacheKey, client.shouldBcache(), needCache)
if err := client.cacheBcache(cacheKey, buf); err != nil {
if err := client.cacheBcache(client.volumeName, cacheKey, buf); err != nil {
client.BcacheHealth = false
log.LogDebugf("ReadExtent L2->L1 failed, err(%v), set BcacheHealth to false.", err)
}

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

@ -716,7 +716,8 @@ 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.storageClass, eh.id); err != nil {
log.LogWarnf("allocateExtent: failed to get write data partition, eh(%v) exclude(%v), clear exclude and try again!", eh, exclude)
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
}

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

@ -18,6 +18,9 @@ import (
"context"
"fmt"
"io"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
@ -219,10 +222,10 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
}
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})
}
//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)",
@ -239,18 +242,22 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
s.inode, proto.StorageClassString(inodeInfo.StorageClass), s.client.bcacheEnable, s.client.bcacheOnlyForNotSSD)
offset := req.FileOffset - int(req.ExtentKey.FileOffset)
if s.client.loadBcache != nil {
readBytes, err = s.client.loadBcache(cacheKey, req.Data, uint64(offset), uint32(req.Size))
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: ino(%v) storageClass(%v) cacheKey(%v) readBytes(%v) err(%v)",
s.inode, inodeInfo.StorageClass, cacheKey, readBytes, err)
log.LogDebugf("TRACE Stream read. hit blockCache: cacheKey(%v) inode(%v) "+
"offset(%v) readBytes(%v) goroutine(%v)", cacheKey, s.inode, offset, readBytes, getGoid())
continue
}
log.LogDebugf("TRACE Stream read. miss blockCache cacheKey(%v) loadBcache(%v)", cacheKey, s.client.loadBcache)
bcacheMissMetric := exporter.NewCounter("fileReadL1CacheMiss")
bcacheMissMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
}
log.LogDebugf("TRACE Stream read. miss blockCache cacheKey(%v) loadBcache(%v)", cacheKey, s.client.loadBcache)
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)
@ -283,11 +290,6 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
log.LogDebugf("Streamer not read from remoteCache, ino(%v) enableRemoteCache(false)", s.inode)
}
if s.needBCache {
bcacheMetric := exporter.NewCounter("fileReadL1CacheMiss")
bcacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
}
// read extent
reader, err = s.GetExtentReader(req.ExtentKey, storageClass)
if err != nil {
@ -307,11 +309,15 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
// 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, storageClass: storageClass}:
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())
}
}
}
@ -344,6 +350,7 @@ func (s *Streamer) asyncBlockCache() {
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
@ -355,8 +362,10 @@ func (s *Streamer) asyncBlockCache() {
}
reader, _ := s.GetExtentReader(ek, pending.storageClass)
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)
@ -366,9 +375,12 @@ func (s *Streamer) asyncBlockCache() {
}
return
}
log.LogDebugf("TRACE read. read blockCache cacheKey(%v) len_buf(%v) cost %v,", cacheKey, len(data), time.Now().Sub(begin).String())
metric.SetWithLabels(err, map[string]string{exporter.Vol: s.client.volumeName})
if s.client.cacheBcache != nil {
log.LogDebugf("TRACE read. write blockCache cacheKey(%v) len_buf(%v),", cacheKey, len(data))
s.client.cacheBcache(cacheKey, data)
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.Now().Sub(begin).String())
}
if ek.Size == bcache.MaxBlockSize {
buf.BCachePool.Put(data)
@ -387,3 +399,11 @@ func (s *Streamer) asyncBlockCache() {
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
}