mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
feat(flashnode): support put block to flashnode #1000150848
Signed-off-by: chihe <chihe@oppo.com>
(cherry picked from commit 0929e28a07)
This commit is contained in:
chihe
parent
e95b3c523e
commit
9e5daafdb8
@ -36,6 +36,11 @@ const (
|
||||
FlashManualCheckAction = "check"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrorNotExistShouldCache = fmt.Errorf("cache miss: should store and cache the key")
|
||||
ErrorNotExistShouldNotCache = fmt.Errorf("only cache the miss after reaching the miss times")
|
||||
)
|
||||
|
||||
type FlashGroupStatus int
|
||||
|
||||
type SlotStatus int
|
||||
@ -247,6 +252,13 @@ func (pr *CachePrepareRequest) String() string {
|
||||
return fmt.Sprintf("cachePrepareRequest[Volume(%v) Inode(%v) FixedFileOffset(%v) Sources(%v) TTL(%v)]", pr.CacheRequest.Volume, pr.CacheRequest.Inode, pr.CacheRequest.FixedFileOffset, len(pr.CacheRequest.Sources), pr.CacheRequest.TTL)
|
||||
}
|
||||
|
||||
func (pbh *PutBlockHead) String() string {
|
||||
if pbh == nil {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf("PutBlockHead[UniKey(%v) BlockLen(%v) TTL(%v)]", pbh.UniKey, pbh.BlockLen, pbh.TTL)
|
||||
}
|
||||
|
||||
type FlashGroupsAdminView struct {
|
||||
FlashGroups []FlashGroupAdminView
|
||||
}
|
||||
@ -283,6 +295,14 @@ type FlashNodeStat struct {
|
||||
CacheStatus []*CacheStatus
|
||||
}
|
||||
|
||||
type FlashWriteParam struct {
|
||||
Offset int64
|
||||
Size int64
|
||||
Data []byte
|
||||
Crc []byte
|
||||
DataSize int64
|
||||
}
|
||||
|
||||
type CacheStatus struct {
|
||||
DataPath string `json:"data_path"`
|
||||
Medium string `json:"medium"`
|
||||
@ -424,3 +444,21 @@ func (flt *FlashManualTask) SetResponse(taskRsp *FlashNodeManualTaskResponse) {
|
||||
flt.ManualTaskStatistics.LastCacheSize = taskRsp.LastCacheSize
|
||||
flt.Done = taskRsp.Done
|
||||
}
|
||||
|
||||
type CacheMissEntry struct {
|
||||
UniKey string
|
||||
MissCount int32
|
||||
expiration int64
|
||||
}
|
||||
|
||||
func CacheMissExpired(info *CacheMissEntry) bool {
|
||||
return time.Now().UnixNano() > info.expiration
|
||||
}
|
||||
|
||||
func SetMissEntryExpiration(info *CacheMissEntry, t time.Duration) {
|
||||
info.expiration = time.Now().Add(t).UnixNano()
|
||||
}
|
||||
|
||||
func GetMissEntryExpiration(info *CacheMissEntry) int64 {
|
||||
return info.expiration
|
||||
}
|
||||
|
||||
@ -43,11 +43,9 @@ func (*DataSource) ProtoMessage() {}
|
||||
func (*DataSource) Descriptor() ([]byte, []int) {
|
||||
return fileDescriptor_9a29821e5022ef34, []int{0}
|
||||
}
|
||||
|
||||
func (m *DataSource) XXX_Unmarshal(b []byte) error {
|
||||
return m.Unmarshal(b)
|
||||
}
|
||||
|
||||
func (m *DataSource) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
|
||||
if deterministic {
|
||||
return xxx_messageInfo_DataSource.Marshal(b, m, deterministic)
|
||||
@ -60,15 +58,12 @@ func (m *DataSource) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
|
||||
return b[:n], nil
|
||||
}
|
||||
}
|
||||
|
||||
func (m *DataSource) XXX_Merge(src proto.Message) {
|
||||
xxx_messageInfo_DataSource.Merge(m, src)
|
||||
}
|
||||
|
||||
func (m *DataSource) XXX_Size() int {
|
||||
return m.Size()
|
||||
}
|
||||
|
||||
func (m *DataSource) XXX_DiscardUnknown() {
|
||||
xxx_messageInfo_DataSource.DiscardUnknown(m)
|
||||
}
|
||||
@ -135,11 +130,9 @@ func (*CacheRequest) ProtoMessage() {}
|
||||
func (*CacheRequest) Descriptor() ([]byte, []int) {
|
||||
return fileDescriptor_9a29821e5022ef34, []int{1}
|
||||
}
|
||||
|
||||
func (m *CacheRequest) XXX_Unmarshal(b []byte) error {
|
||||
return m.Unmarshal(b)
|
||||
}
|
||||
|
||||
func (m *CacheRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
|
||||
if deterministic {
|
||||
return xxx_messageInfo_CacheRequest.Marshal(b, m, deterministic)
|
||||
@ -152,15 +145,12 @@ func (m *CacheRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
|
||||
return b[:n], nil
|
||||
}
|
||||
}
|
||||
|
||||
func (m *CacheRequest) XXX_Merge(src proto.Message) {
|
||||
xxx_messageInfo_CacheRequest.Merge(m, src)
|
||||
}
|
||||
|
||||
func (m *CacheRequest) XXX_Size() int {
|
||||
return m.Size()
|
||||
}
|
||||
|
||||
func (m *CacheRequest) XXX_DiscardUnknown() {
|
||||
xxx_messageInfo_CacheRequest.DiscardUnknown(m)
|
||||
}
|
||||
@ -230,11 +220,9 @@ func (*CacheReadRequest) ProtoMessage() {}
|
||||
func (*CacheReadRequest) Descriptor() ([]byte, []int) {
|
||||
return fileDescriptor_9a29821e5022ef34, []int{2}
|
||||
}
|
||||
|
||||
func (m *CacheReadRequest) XXX_Unmarshal(b []byte) error {
|
||||
return m.Unmarshal(b)
|
||||
}
|
||||
|
||||
func (m *CacheReadRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
|
||||
if deterministic {
|
||||
return xxx_messageInfo_CacheReadRequest.Marshal(b, m, deterministic)
|
||||
@ -247,15 +235,12 @@ func (m *CacheReadRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, er
|
||||
return b[:n], nil
|
||||
}
|
||||
}
|
||||
|
||||
func (m *CacheReadRequest) XXX_Merge(src proto.Message) {
|
||||
xxx_messageInfo_CacheReadRequest.Merge(m, src)
|
||||
}
|
||||
|
||||
func (m *CacheReadRequest) XXX_Size() int {
|
||||
return m.Size()
|
||||
}
|
||||
|
||||
func (m *CacheReadRequest) XXX_DiscardUnknown() {
|
||||
xxx_messageInfo_CacheReadRequest.DiscardUnknown(m)
|
||||
}
|
||||
@ -296,11 +281,9 @@ func (*CachePrepareRequest) ProtoMessage() {}
|
||||
func (*CachePrepareRequest) Descriptor() ([]byte, []int) {
|
||||
return fileDescriptor_9a29821e5022ef34, []int{3}
|
||||
}
|
||||
|
||||
func (m *CachePrepareRequest) XXX_Unmarshal(b []byte) error {
|
||||
return m.Unmarshal(b)
|
||||
}
|
||||
|
||||
func (m *CachePrepareRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
|
||||
if deterministic {
|
||||
return xxx_messageInfo_CachePrepareRequest.Marshal(b, m, deterministic)
|
||||
@ -313,15 +296,12 @@ func (m *CachePrepareRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte,
|
||||
return b[:n], nil
|
||||
}
|
||||
}
|
||||
|
||||
func (m *CachePrepareRequest) XXX_Merge(src proto.Message) {
|
||||
xxx_messageInfo_CachePrepareRequest.Merge(m, src)
|
||||
}
|
||||
|
||||
func (m *CachePrepareRequest) XXX_Size() int {
|
||||
return m.Size()
|
||||
}
|
||||
|
||||
func (m *CachePrepareRequest) XXX_DiscardUnknown() {
|
||||
xxx_messageInfo_CachePrepareRequest.DiscardUnknown(m)
|
||||
}
|
||||
@ -342,44 +322,109 @@ func (m *CachePrepareRequest) GetFlashNodes() []string {
|
||||
return nil
|
||||
}
|
||||
|
||||
type PutBlockHead struct {
|
||||
UniKey string `protobuf:"bytes,1,opt,name=UniKey,proto3" json:"UniKey,omitempty"`
|
||||
BlockLen int64 `protobuf:"varint,2,opt,name=BlockLen,proto3" json:"BlockLen,omitempty"`
|
||||
TTL uint64 `protobuf:"varint,3,opt,name=TTL,proto3" json:"TTL,omitempty"`
|
||||
XXX_NoUnkeyedLiteral struct{} `json:"-"`
|
||||
XXX_unrecognized []byte `json:"-"`
|
||||
XXX_sizecache int32 `json:"-"`
|
||||
}
|
||||
|
||||
func (m *PutBlockHead) Reset() { *m = PutBlockHead{} }
|
||||
func (*PutBlockHead) ProtoMessage() {}
|
||||
func (*PutBlockHead) Descriptor() ([]byte, []int) {
|
||||
return fileDescriptor_9a29821e5022ef34, []int{4}
|
||||
}
|
||||
func (m *PutBlockHead) XXX_Unmarshal(b []byte) error {
|
||||
return m.Unmarshal(b)
|
||||
}
|
||||
func (m *PutBlockHead) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
|
||||
if deterministic {
|
||||
return xxx_messageInfo_PutBlockHead.Marshal(b, m, deterministic)
|
||||
} else {
|
||||
b = b[:cap(b)]
|
||||
n, err := m.MarshalToSizedBuffer(b)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return b[:n], nil
|
||||
}
|
||||
}
|
||||
func (m *PutBlockHead) XXX_Merge(src proto.Message) {
|
||||
xxx_messageInfo_PutBlockHead.Merge(m, src)
|
||||
}
|
||||
func (m *PutBlockHead) XXX_Size() int {
|
||||
return m.Size()
|
||||
}
|
||||
func (m *PutBlockHead) XXX_DiscardUnknown() {
|
||||
xxx_messageInfo_PutBlockHead.DiscardUnknown(m)
|
||||
}
|
||||
|
||||
var xxx_messageInfo_PutBlockHead proto.InternalMessageInfo
|
||||
|
||||
func (m *PutBlockHead) GetUniKey() string {
|
||||
if m != nil {
|
||||
return m.UniKey
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (m *PutBlockHead) GetBlockLen() int64 {
|
||||
if m != nil {
|
||||
return m.BlockLen
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (m *PutBlockHead) GetTTL() uint64 {
|
||||
if m != nil {
|
||||
return m.TTL
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func init() {
|
||||
proto.RegisterType((*DataSource)(nil), "proto.DataSource")
|
||||
proto.RegisterType((*CacheRequest)(nil), "proto.CacheRequest")
|
||||
proto.RegisterType((*CacheReadRequest)(nil), "proto.CacheReadRequest")
|
||||
proto.RegisterType((*CachePrepareRequest)(nil), "proto.CachePrepareRequest")
|
||||
proto.RegisterType((*PutBlockHead)(nil), "proto.PutBlockHead")
|
||||
}
|
||||
|
||||
func init() { proto.RegisterFile("distributed_cache.proto", fileDescriptor_9a29821e5022ef34) }
|
||||
|
||||
var fileDescriptor_9a29821e5022ef34 = []byte{
|
||||
// 421 bytes of a gzipped FileDescriptorProto
|
||||
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x52, 0x4d, 0x6f, 0xd3, 0x40,
|
||||
0x10, 0xed, 0xd6, 0xb1, 0x43, 0x27, 0x45, 0x84, 0x2d, 0x82, 0x55, 0x0f, 0xc6, 0xca, 0xc9, 0x12,
|
||||
0x22, 0x95, 0xca, 0x01, 0x89, 0x23, 0x84, 0x8a, 0x48, 0x08, 0xaa, 0x6d, 0xd5, 0x2b, 0xf2, 0xc7,
|
||||
0x24, 0x59, 0xc9, 0xf5, 0x16, 0xef, 0x5a, 0xaa, 0x90, 0xf8, 0x1f, 0x3d, 0xf2, 0x4b, 0x38, 0x73,
|
||||
0xe4, 0xca, 0x0d, 0x85, 0x3f, 0x82, 0x3c, 0xbb, 0xa1, 0x6e, 0x39, 0xf6, 0xe4, 0x79, 0x6f, 0xc6,
|
||||
0x33, 0xef, 0x3d, 0x1b, 0x9e, 0x94, 0xca, 0xd8, 0x46, 0xe5, 0xad, 0xc5, 0xf2, 0x53, 0x91, 0x15,
|
||||
0x2b, 0x9c, 0x5e, 0x34, 0xda, 0x6a, 0x1e, 0xd2, 0x63, 0xff, 0xf9, 0x52, 0xd9, 0x55, 0x9b, 0x4f,
|
||||
0x0b, 0x7d, 0x7e, 0xb0, 0xd4, 0x4b, 0x7d, 0x40, 0x74, 0xde, 0x2e, 0x08, 0x11, 0xa0, 0xca, 0xbd,
|
||||
0x35, 0xf9, 0xce, 0x00, 0x66, 0x99, 0xcd, 0x4e, 0x74, 0xdb, 0x14, 0xc8, 0x63, 0x80, 0x23, 0x55,
|
||||
0xe1, 0xc7, 0xc5, 0xc2, 0xa0, 0x15, 0x2c, 0x61, 0xe9, 0x40, 0xf6, 0x18, 0x9e, 0xc0, 0xe8, 0x38,
|
||||
0x6b, 0xac, 0xb2, 0x4a, 0xd7, 0xf3, 0x99, 0xd8, 0xa6, 0x81, 0x3e, 0xc5, 0xf7, 0xe1, 0xde, 0xdb,
|
||||
0x4b, 0x8b, 0xb5, 0x9d, 0xcf, 0x44, 0x40, 0xed, 0x7f, 0x98, 0x4f, 0x60, 0xd7, 0xd5, 0x7e, 0xff,
|
||||
0x80, 0xfa, 0x37, 0x38, 0xce, 0x61, 0x70, 0xa2, 0xbe, 0xa0, 0x08, 0xa9, 0x47, 0x35, 0x7f, 0x04,
|
||||
0xe1, 0x3b, 0x6d, 0xac, 0x11, 0x51, 0x12, 0xa4, 0x3b, 0xd2, 0x81, 0x57, 0x83, 0xab, 0x6f, 0x4f,
|
||||
0xb7, 0x26, 0xbf, 0x18, 0xec, 0xbe, 0xe9, 0x62, 0x90, 0xf8, 0xb9, 0x45, 0x63, 0xf9, 0x63, 0x88,
|
||||
0xce, 0x74, 0xd5, 0x9e, 0x23, 0xc9, 0xdf, 0x91, 0x1e, 0x75, 0x4b, 0xe6, 0xb5, 0x2e, 0xd1, 0x8b,
|
||||
0x76, 0x80, 0xa7, 0xf0, 0xe0, 0x48, 0x5d, 0x62, 0xd9, 0x73, 0xed, 0x54, 0xdf, 0xa6, 0xb9, 0x80,
|
||||
0xe1, 0x19, 0x36, 0x46, 0xe9, 0x9a, 0x74, 0xdf, 0x97, 0x1b, 0xc8, 0x9f, 0xc1, 0xd0, 0xc5, 0x67,
|
||||
0x44, 0x98, 0x04, 0xe9, 0xe8, 0xf0, 0xa1, 0x0b, 0x77, 0x7a, 0x1d, 0xac, 0xdc, 0x4c, 0xf0, 0x31,
|
||||
0x04, 0xa7, 0xa7, 0xef, 0x45, 0x94, 0xb0, 0x34, 0x90, 0x5d, 0x49, 0x8e, 0x2b, 0x6d, 0xc5, 0xd0,
|
||||
0x3b, 0xae, 0xb4, 0xf5, 0xde, 0xbe, 0xc2, 0xd8, 0x5b, 0xcb, 0xca, 0x8d, 0xbd, 0x97, 0x37, 0xed,
|
||||
0x92, 0xc9, 0xd1, 0xe1, 0x9e, 0xbf, 0xd8, 0x6f, 0xc9, 0xff, 0x72, 0xf1, 0x06, 0x5d, 0x00, 0xd1,
|
||||
0xad, 0xc0, 0x83, 0xeb, 0xc0, 0xfd, 0x79, 0x0b, 0x7b, 0xb4, 0xe1, 0xb8, 0xc1, 0x8b, 0xac, 0xc1,
|
||||
0x3b, 0x2b, 0xe8, 0x7e, 0xae, 0x2a, 0x33, 0xab, 0x0f, 0xba, 0x44, 0x23, 0xb6, 0xe9, 0x5b, 0xf6,
|
||||
0x18, 0x77, 0xf5, 0xf5, 0xf8, 0xc7, 0x3a, 0x66, 0x3f, 0xd7, 0x31, 0xfb, 0xbd, 0x8e, 0xd9, 0xd5,
|
||||
0x9f, 0x78, 0x2b, 0x8f, 0x68, 0xf3, 0x8b, 0xbf, 0x01, 0x00, 0x00, 0xff, 0xff, 0xf3, 0x2e, 0xf9,
|
||||
0xd8, 0xfb, 0x02, 0x00, 0x00,
|
||||
// 463 bytes of a gzipped FileDescriptorProto
|
||||
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x52, 0xc1, 0x6e, 0xd3, 0x40,
|
||||
0x10, 0xad, 0xeb, 0xc4, 0xa1, 0x93, 0x20, 0xc2, 0x16, 0xc1, 0xaa, 0x07, 0x63, 0xe5, 0x64, 0x09,
|
||||
0x91, 0x4a, 0xe5, 0x80, 0xc4, 0xb1, 0x84, 0xaa, 0x11, 0x15, 0x44, 0xdb, 0xd2, 0x03, 0x17, 0xe4,
|
||||
0xd8, 0x93, 0x64, 0x85, 0xeb, 0x2d, 0xde, 0xb5, 0x54, 0x90, 0xf8, 0x8f, 0x1e, 0xf9, 0x12, 0xce,
|
||||
0x1c, 0xb9, 0x72, 0x43, 0xe1, 0x47, 0x90, 0x67, 0xd7, 0xad, 0x5b, 0x8e, 0x9c, 0x3c, 0xef, 0xcd,
|
||||
0x7a, 0xe6, 0xbd, 0xb7, 0x0b, 0x8f, 0x32, 0xa9, 0x4d, 0x29, 0xe7, 0x95, 0xc1, 0xec, 0x43, 0x9a,
|
||||
0xa4, 0x2b, 0x1c, 0x9f, 0x97, 0xca, 0x28, 0xd6, 0xa5, 0xcf, 0xce, 0xd3, 0xa5, 0x34, 0xab, 0x6a,
|
||||
0x3e, 0x4e, 0xd5, 0xd9, 0xee, 0x52, 0x2d, 0xd5, 0x2e, 0xd1, 0xf3, 0x6a, 0x41, 0x88, 0x00, 0x55,
|
||||
0xf6, 0xaf, 0xd1, 0x77, 0x0f, 0x60, 0x92, 0x98, 0xe4, 0x58, 0x55, 0x65, 0x8a, 0x2c, 0x04, 0x38,
|
||||
0x90, 0x39, 0xbe, 0x5d, 0x2c, 0x34, 0x1a, 0xee, 0x45, 0x5e, 0xdc, 0x11, 0x2d, 0x86, 0x45, 0xd0,
|
||||
0x9f, 0x25, 0xa5, 0x91, 0x46, 0xaa, 0x62, 0x3a, 0xe1, 0x9b, 0x74, 0xa0, 0x4d, 0xb1, 0x1d, 0xb8,
|
||||
0xf3, 0xea, 0xc2, 0x60, 0x61, 0xa6, 0x13, 0xee, 0x53, 0xfb, 0x0a, 0xb3, 0x11, 0x0c, 0x6c, 0xed,
|
||||
0xe6, 0x77, 0xa8, 0x7f, 0x83, 0x63, 0x0c, 0x3a, 0xc7, 0xf2, 0x0b, 0xf2, 0x2e, 0xf5, 0xa8, 0x66,
|
||||
0x0f, 0xa0, 0x7b, 0xa8, 0xb4, 0xd1, 0x3c, 0x88, 0xfc, 0x78, 0x4b, 0x58, 0xf0, 0xa2, 0x73, 0xf9,
|
||||
0xed, 0xf1, 0xc6, 0xe8, 0x97, 0x07, 0x83, 0x97, 0x75, 0x0c, 0x02, 0x3f, 0x55, 0xa8, 0x0d, 0x7b,
|
||||
0x08, 0xc1, 0xa9, 0xca, 0xab, 0x33, 0x24, 0xf9, 0x5b, 0xc2, 0xa1, 0x7a, 0xc8, 0xb4, 0x50, 0x19,
|
||||
0x3a, 0xd1, 0x16, 0xb0, 0x18, 0xee, 0x1d, 0xc8, 0x0b, 0xcc, 0x5a, 0xae, 0xad, 0xea, 0xdb, 0x34,
|
||||
0xe3, 0xd0, 0x3b, 0xc5, 0x52, 0x4b, 0x55, 0x90, 0xee, 0xbb, 0xa2, 0x81, 0xec, 0x09, 0xf4, 0x6c,
|
||||
0x7c, 0x9a, 0x77, 0x23, 0x3f, 0xee, 0xef, 0xdd, 0xb7, 0xe1, 0x8e, 0xaf, 0x83, 0x15, 0xcd, 0x09,
|
||||
0x36, 0x04, 0xff, 0xe4, 0xe4, 0x88, 0x07, 0x91, 0x17, 0xfb, 0xa2, 0x2e, 0xc9, 0x71, 0xae, 0x0c,
|
||||
0xef, 0x39, 0xc7, 0xb9, 0x32, 0xce, 0xdb, 0x57, 0x18, 0x3a, 0x6b, 0x49, 0xd6, 0xd8, 0x7b, 0x7e,
|
||||
0xd3, 0x2e, 0x99, 0xec, 0xef, 0x6d, 0xbb, 0x8d, 0xed, 0x96, 0xf8, 0x27, 0x17, 0x67, 0xd0, 0x06,
|
||||
0x10, 0xdc, 0x0a, 0xdc, 0xbf, 0x0e, 0xdc, 0xad, 0x37, 0xb0, 0x4d, 0x13, 0x66, 0x25, 0x9e, 0x27,
|
||||
0x25, 0xfe, 0xb7, 0x82, 0xfa, 0x71, 0xe5, 0x89, 0x5e, 0xbd, 0x51, 0x19, 0x6a, 0xbe, 0x49, 0x77,
|
||||
0xd9, 0x62, 0xdc, 0xd6, 0xf7, 0x30, 0x98, 0x55, 0x66, 0x3f, 0x57, 0xe9, 0xc7, 0x43, 0x4c, 0xb2,
|
||||
0x5a, 0xf7, 0xbb, 0x42, 0xbe, 0xc6, 0xcf, 0xcd, 0x7d, 0x5a, 0x54, 0x3f, 0x34, 0x3a, 0x74, 0x84,
|
||||
0x05, 0x39, 0xf2, 0xc5, 0x15, 0x6e, 0x42, 0xb6, 0x96, 0xea, 0xd2, 0xce, 0xde, 0x1f, 0xfe, 0x58,
|
||||
0x87, 0xde, 0xcf, 0x75, 0xe8, 0xfd, 0x5e, 0x87, 0xde, 0xe5, 0x9f, 0x70, 0x63, 0x1e, 0x90, 0xea,
|
||||
0x67, 0x7f, 0x03, 0x00, 0x00, 0xff, 0xff, 0x42, 0xb3, 0xc8, 0x05, 0x57, 0x03, 0x00, 0x00,
|
||||
}
|
||||
|
||||
func (m *DataSource) Marshal() (dAtA []byte, err error) {
|
||||
@ -613,6 +658,50 @@ func (m *CachePrepareRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
|
||||
return len(dAtA) - i, nil
|
||||
}
|
||||
|
||||
func (m *PutBlockHead) Marshal() (dAtA []byte, err error) {
|
||||
size := m.Size()
|
||||
dAtA = make([]byte, size)
|
||||
n, err := m.MarshalToSizedBuffer(dAtA[:size])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return dAtA[:n], nil
|
||||
}
|
||||
|
||||
func (m *PutBlockHead) MarshalTo(dAtA []byte) (int, error) {
|
||||
size := m.Size()
|
||||
return m.MarshalToSizedBuffer(dAtA[:size])
|
||||
}
|
||||
|
||||
func (m *PutBlockHead) MarshalToSizedBuffer(dAtA []byte) (int, error) {
|
||||
i := len(dAtA)
|
||||
_ = i
|
||||
var l int
|
||||
_ = l
|
||||
if m.XXX_unrecognized != nil {
|
||||
i -= len(m.XXX_unrecognized)
|
||||
copy(dAtA[i:], m.XXX_unrecognized)
|
||||
}
|
||||
if m.TTL != 0 {
|
||||
i = encodeVarintDistributedCache(dAtA, i, uint64(m.TTL))
|
||||
i--
|
||||
dAtA[i] = 0x18
|
||||
}
|
||||
if m.BlockLen != 0 {
|
||||
i = encodeVarintDistributedCache(dAtA, i, uint64(m.BlockLen))
|
||||
i--
|
||||
dAtA[i] = 0x10
|
||||
}
|
||||
if len(m.UniKey) > 0 {
|
||||
i -= len(m.UniKey)
|
||||
copy(dAtA[i:], m.UniKey)
|
||||
i = encodeVarintDistributedCache(dAtA, i, uint64(len(m.UniKey)))
|
||||
i--
|
||||
dAtA[i] = 0xa
|
||||
}
|
||||
return len(dAtA) - i, nil
|
||||
}
|
||||
|
||||
func encodeVarintDistributedCache(dAtA []byte, offset int, v uint64) int {
|
||||
offset -= sovDistributedCache(v)
|
||||
base := offset
|
||||
@ -624,7 +713,6 @@ func encodeVarintDistributedCache(dAtA []byte, offset int, v uint64) int {
|
||||
dAtA[offset] = uint8(v)
|
||||
return base
|
||||
}
|
||||
|
||||
func (m *DataSource) Size() (n int) {
|
||||
if m == nil {
|
||||
return 0
|
||||
@ -739,14 +827,34 @@ func (m *CachePrepareRequest) Size() (n int) {
|
||||
return n
|
||||
}
|
||||
|
||||
func (m *PutBlockHead) Size() (n int) {
|
||||
if m == nil {
|
||||
return 0
|
||||
}
|
||||
var l int
|
||||
_ = l
|
||||
l = len(m.UniKey)
|
||||
if l > 0 {
|
||||
n += 1 + l + sovDistributedCache(uint64(l))
|
||||
}
|
||||
if m.BlockLen != 0 {
|
||||
n += 1 + sovDistributedCache(uint64(m.BlockLen))
|
||||
}
|
||||
if m.TTL != 0 {
|
||||
n += 1 + sovDistributedCache(uint64(m.TTL))
|
||||
}
|
||||
if m.XXX_unrecognized != nil {
|
||||
n += len(m.XXX_unrecognized)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func sovDistributedCache(x uint64) (n int) {
|
||||
return (math_bits.Len64(x|1) + 6) / 7
|
||||
}
|
||||
|
||||
func sozDistributedCache(x uint64) (n int) {
|
||||
return sovDistributedCache(uint64((x << 1) ^ uint64((int64(x) >> 63))))
|
||||
}
|
||||
|
||||
func (m *DataSource) Unmarshal(dAtA []byte) error {
|
||||
l := len(dAtA)
|
||||
iNdEx := 0
|
||||
@ -925,7 +1033,6 @@ func (m *DataSource) Unmarshal(dAtA []byte) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *CacheRequest) Unmarshal(dAtA []byte) error {
|
||||
l := len(dAtA)
|
||||
iNdEx := 0
|
||||
@ -1138,7 +1245,6 @@ func (m *CacheRequest) Unmarshal(dAtA []byte) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *CacheReadRequest) Unmarshal(dAtA []byte) error {
|
||||
l := len(dAtA)
|
||||
iNdEx := 0
|
||||
@ -1264,7 +1370,6 @@ func (m *CacheReadRequest) Unmarshal(dAtA []byte) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *CachePrepareRequest) Unmarshal(dAtA []byte) error {
|
||||
l := len(dAtA)
|
||||
iNdEx := 0
|
||||
@ -1384,7 +1489,127 @@ func (m *CachePrepareRequest) Unmarshal(dAtA []byte) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (m *PutBlockHead) Unmarshal(dAtA []byte) error {
|
||||
l := len(dAtA)
|
||||
iNdEx := 0
|
||||
for iNdEx < l {
|
||||
preIndex := iNdEx
|
||||
var wire uint64
|
||||
for shift := uint(0); ; shift += 7 {
|
||||
if shift >= 64 {
|
||||
return ErrIntOverflowDistributedCache
|
||||
}
|
||||
if iNdEx >= l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
b := dAtA[iNdEx]
|
||||
iNdEx++
|
||||
wire |= uint64(b&0x7F) << shift
|
||||
if b < 0x80 {
|
||||
break
|
||||
}
|
||||
}
|
||||
fieldNum := int32(wire >> 3)
|
||||
wireType := int(wire & 0x7)
|
||||
if wireType == 4 {
|
||||
return fmt.Errorf("proto: PutBlockHead: wiretype end group for non-group")
|
||||
}
|
||||
if fieldNum <= 0 {
|
||||
return fmt.Errorf("proto: PutBlockHead: illegal tag %d (wire type %d)", fieldNum, wire)
|
||||
}
|
||||
switch fieldNum {
|
||||
case 1:
|
||||
if wireType != 2 {
|
||||
return fmt.Errorf("proto: wrong wireType = %d for field UniKey", wireType)
|
||||
}
|
||||
var stringLen uint64
|
||||
for shift := uint(0); ; shift += 7 {
|
||||
if shift >= 64 {
|
||||
return ErrIntOverflowDistributedCache
|
||||
}
|
||||
if iNdEx >= l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
b := dAtA[iNdEx]
|
||||
iNdEx++
|
||||
stringLen |= uint64(b&0x7F) << shift
|
||||
if b < 0x80 {
|
||||
break
|
||||
}
|
||||
}
|
||||
intStringLen := int(stringLen)
|
||||
if intStringLen < 0 {
|
||||
return ErrInvalidLengthDistributedCache
|
||||
}
|
||||
postIndex := iNdEx + intStringLen
|
||||
if postIndex < 0 {
|
||||
return ErrInvalidLengthDistributedCache
|
||||
}
|
||||
if postIndex > l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
m.UniKey = string(dAtA[iNdEx:postIndex])
|
||||
iNdEx = postIndex
|
||||
case 2:
|
||||
if wireType != 0 {
|
||||
return fmt.Errorf("proto: wrong wireType = %d for field BlockLen", wireType)
|
||||
}
|
||||
m.BlockLen = 0
|
||||
for shift := uint(0); ; shift += 7 {
|
||||
if shift >= 64 {
|
||||
return ErrIntOverflowDistributedCache
|
||||
}
|
||||
if iNdEx >= l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
b := dAtA[iNdEx]
|
||||
iNdEx++
|
||||
m.BlockLen |= int64(b&0x7F) << shift
|
||||
if b < 0x80 {
|
||||
break
|
||||
}
|
||||
}
|
||||
case 3:
|
||||
if wireType != 0 {
|
||||
return fmt.Errorf("proto: wrong wireType = %d for field TTL", wireType)
|
||||
}
|
||||
m.TTL = 0
|
||||
for shift := uint(0); ; shift += 7 {
|
||||
if shift >= 64 {
|
||||
return ErrIntOverflowDistributedCache
|
||||
}
|
||||
if iNdEx >= l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
b := dAtA[iNdEx]
|
||||
iNdEx++
|
||||
m.TTL |= uint64(b&0x7F) << shift
|
||||
if b < 0x80 {
|
||||
break
|
||||
}
|
||||
}
|
||||
default:
|
||||
iNdEx = preIndex
|
||||
skippy, err := skipDistributedCache(dAtA[iNdEx:])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthDistributedCache
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
|
||||
iNdEx += skippy
|
||||
}
|
||||
}
|
||||
|
||||
if iNdEx > l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func skipDistributedCache(dAtA []byte) (n int, err error) {
|
||||
l := len(dAtA)
|
||||
iNdEx := 0
|
||||
|
||||
@ -41,3 +41,11 @@ message CachePrepareRequest {
|
||||
CacheRequest CacheRequest = 1;
|
||||
repeated string FlashNodes = 2;
|
||||
}
|
||||
|
||||
message PutBlockHead {
|
||||
option (gogoproto.goproto_stringer) = false;
|
||||
|
||||
string UniKey = 1;
|
||||
int64 BlockLen = 2;
|
||||
uint64 TTL = 3;
|
||||
}
|
||||
@ -298,6 +298,8 @@ const (
|
||||
OpFlashNodeHeartbeat uint8 = 0xDA
|
||||
OpFlashNodeCachePrepare uint8 = 0xDB
|
||||
OpFlashNodeCacheRead uint8 = 0xDC
|
||||
OpFlashNodeCachePutBlock uint8 = 0xD8
|
||||
OpFlashNodeCacheDelete uint8 = 0xD9
|
||||
OpFlashNodeSetReadIOLimits uint8 = 0xED
|
||||
OpFlashNodeSetWriteIOLimits uint8 = 0xEE
|
||||
OpFlashNodeScan uint8 = 0xD4
|
||||
|
||||
@ -23,12 +23,14 @@ import (
|
||||
"path"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util"
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
"github.com/cubefs/cubefs/util/stat"
|
||||
)
|
||||
@ -36,11 +38,15 @@ import (
|
||||
const (
|
||||
_cacheBlockOpenOpt = os.O_CREATE | os.O_RDWR
|
||||
HeaderSize = 40
|
||||
SourceTypeDefault = ""
|
||||
SourceTypeBlock = "blocks"
|
||||
CRCLen = 4
|
||||
)
|
||||
|
||||
type CacheBlock struct {
|
||||
rootPath string
|
||||
filePath string
|
||||
sourceType string
|
||||
blockKey string
|
||||
cacheEngine *CacheEngine
|
||||
ttl int64
|
||||
@ -74,9 +80,10 @@ func NewCacheBlock(rootPath string, volume string, inode, fixedOffset uint64, ve
|
||||
cb.inode = inode
|
||||
cb.fixedOffset = fixedOffset
|
||||
cb.version = version
|
||||
cb.sourceType = SourceTypeDefault
|
||||
cb.blockKey = GenCacheBlockKey(volume, inode, fixedOffset, version)
|
||||
cb.updateAllocSize(int64(allocSize))
|
||||
cb.filePath = path.Join(rootPath, cb.blockKey)
|
||||
cb.filePath = path.Join(rootPath+SourceTypeDefault, cb.blockKey)
|
||||
cb.rootPath = rootPath
|
||||
cb.sourceReader = reader
|
||||
cb.readyCh = make(chan struct{})
|
||||
@ -248,7 +255,7 @@ func (cb *CacheBlock) writeCacheBlockFileHeader(file *os.File) (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
func (cb *CacheBlock) checkCacheBlockFileHeader(file *os.File) (allocSize, usedSize int64, expiredTime time.Time, err error) {
|
||||
func (cb *CacheBlock) checkCacheBlockFileHeader(file *os.File, sourceType string) (allocSize, usedSize int64, expiredTime time.Time, err error) {
|
||||
var stat os.FileInfo
|
||||
var seconds int64
|
||||
var reserved1, reserved2 uint64
|
||||
@ -279,9 +286,15 @@ func (cb *CacheBlock) checkCacheBlockFileHeader(file *os.File) (allocSize, usedS
|
||||
return
|
||||
}
|
||||
|
||||
if usedSize+HeaderSize != stat.Size() {
|
||||
if SourceTypeDefault == sourceType && usedSize+HeaderSize != stat.Size() {
|
||||
err = fmt.Errorf("usedSize + headerSize[%v] != file real size[%v]", usedSize+HeaderSize, stat.Size())
|
||||
return
|
||||
} else if SourceTypeBlock == sourceType {
|
||||
crcSize := (usedSize + proto.PageSize - 1) / proto.PageSize * CRCLen
|
||||
if usedSize+HeaderSize+crcSize != stat.Size() {
|
||||
err = fmt.Errorf("usedSize + headerSize + crsSize [%v] != file real size[%v]", usedSize+HeaderSize+crcSize, stat.Size())
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if err = binary.Read(file, binary.BigEndian, &seconds); err != nil {
|
||||
@ -310,7 +323,7 @@ func (cb *CacheBlock) initFilePath(isLoad bool) (err error) {
|
||||
}()
|
||||
|
||||
var file *os.File
|
||||
blockParent := path.Join(cb.rootPath, cb.volume)
|
||||
blockParent := path.Join(cb.rootPath+cb.sourceType, cb.volume)
|
||||
|
||||
if _, err = os.Stat(blockParent); err != nil {
|
||||
if !os.IsNotExist(err.(*os.PathError)) {
|
||||
@ -346,7 +359,7 @@ func (cb *CacheBlock) initFilePath(isLoad bool) (err error) {
|
||||
} else {
|
||||
var allocSize, usedSize int64
|
||||
var expiredTime time.Time
|
||||
if allocSize, usedSize, expiredTime, err = cb.checkCacheBlockFileHeader(file); err != nil {
|
||||
if allocSize, usedSize, expiredTime, err = cb.checkCacheBlockFileHeader(file, cb.sourceType); err != nil {
|
||||
file.Close()
|
||||
return fmt.Errorf("initFilePath check file header failed: %s", err.Error())
|
||||
}
|
||||
@ -497,7 +510,9 @@ func (cb *CacheBlock) ready(ctx context.Context, waitForBlock bool) error {
|
||||
}
|
||||
|
||||
func (cb *CacheBlock) notifyClose() {
|
||||
cb.closeOnce.Do(func() { close(cb.closeCh) })
|
||||
cb.closeOnce.Do(func() {
|
||||
close(cb.closeCh)
|
||||
})
|
||||
}
|
||||
|
||||
func (cb *CacheBlock) notifyReady() {
|
||||
@ -525,7 +540,7 @@ func (cb *CacheBlock) InitOnce(engine *CacheEngine, sources []*proto.DataSource)
|
||||
cb.initOnce.Do(func() { cb.Init(sources, engine.readDataNodeTimeout) })
|
||||
select {
|
||||
case <-cb.closeCh:
|
||||
engine.deleteCacheBlock(cb.blockKey)
|
||||
engine.DeleteCacheBlock(cb.blockKey)
|
||||
auditlog.LogFlashNodeOp("BlockInit", fmt.Sprintf("%v is closed", cb.info()), nil)
|
||||
default:
|
||||
}
|
||||
@ -567,7 +582,7 @@ func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto
|
||||
cb.InitForCacheRead(sources, engine.readDataNodeTimeout)
|
||||
select {
|
||||
case <-cb.closeCh:
|
||||
engine.deleteCacheBlock(cb.blockKey)
|
||||
engine.DeleteCacheBlock(cb.blockKey)
|
||||
auditlog.LogFlashNodeOp("BlockInit", fmt.Sprintf("%v is closed", cb.info()), nil)
|
||||
default:
|
||||
}
|
||||
@ -646,3 +661,212 @@ func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNode
|
||||
func (cb *CacheBlock) info() string {
|
||||
return fmt.Sprintf("path(%v)_from(%v)_size(%v)", cb.filePath, cb.clientIP, cb.allocSize)
|
||||
}
|
||||
|
||||
func NewCacheBlockV2(rootPath string, volume string, uniKey string, allocSize uint64, clientIP string, d *Disk,
|
||||
) (cb *CacheBlock) {
|
||||
cb = new(CacheBlock)
|
||||
cb.volume = volume
|
||||
cb.blockKey = GenCacheBlockKeyV2(volume, uniKey)
|
||||
cb.updateAllocSize(int64(allocSize))
|
||||
cb.sourceType = SourceTypeBlock
|
||||
cb.filePath = path.Join(rootPath+SourceTypeBlock, cb.blockKey)
|
||||
cb.rootPath = rootPath
|
||||
cb.readyCh = make(chan struct{})
|
||||
cb.closeCh = make(chan struct{})
|
||||
cb.clientIP = clientIP
|
||||
cb.disk = d
|
||||
return
|
||||
}
|
||||
|
||||
// CreateBlockV2 todo merge create block v2 to default
|
||||
func (c *CacheEngine) CreateBlockV2(pDir string, uniKey string, ttl uint64, size uint32, clientIP string) (block *CacheBlock, err error, created bool) {
|
||||
if block, err, created = c.createCacheBlockV2(pDir, uniKey, int64(ttl), uint64(size), clientIP); err != nil {
|
||||
log.LogWarnf("action[CreateBlock] createCacheBlock(%v) failed err %v ",
|
||||
GenCacheBlockKeyV2(pDir, uniKey), err)
|
||||
c.DeleteCacheBlock(GenCacheBlockKeyV2(pDir, uniKey))
|
||||
return nil, err, created
|
||||
}
|
||||
return block, nil, created
|
||||
}
|
||||
|
||||
func (c *CacheEngine) createCacheBlockV2(pDir string, uniKey string, ttl int64, allocSize uint64, clientIP string) (block *CacheBlock, err error, created bool) {
|
||||
if allocSize == 0 {
|
||||
return nil, fmt.Errorf("alloc size is zero"), false
|
||||
}
|
||||
key := GenCacheBlockKeyV2(pDir, uniKey)
|
||||
v, ok := c.keyToDiskMap.Load(key)
|
||||
if ok {
|
||||
cacheItem := v.(*lruCacheItem)
|
||||
if atomic.LoadInt32(&cacheItem.disk.Status) == proto.ReadWrite {
|
||||
if blockValue, got := cacheItem.lruCache.Peek(key); got {
|
||||
block = blockValue.(*CacheBlock)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
value, loaded := c.creatingCacheBlockMap.LoadOrStore(key, make(chan struct{}))
|
||||
ch := value.(chan struct{})
|
||||
if loaded {
|
||||
created = true
|
||||
<-ch
|
||||
v, ok = c.keyToDiskMap.Load(key)
|
||||
if ok {
|
||||
cacheItem := v.(*lruCacheItem)
|
||||
if atomic.LoadInt32(&cacheItem.disk.Status) == proto.ReadWrite {
|
||||
if blockValue, got := v.(*lruCacheItem).lruCache.Peek(key); got {
|
||||
block = blockValue.(*CacheBlock)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("unable to get created cacheblock"), created
|
||||
} else {
|
||||
defer func() {
|
||||
close(ch)
|
||||
c.creatingCacheBlockMap.Delete(key)
|
||||
}()
|
||||
}
|
||||
|
||||
v, ok = c.keyToDiskMap.Load(key)
|
||||
if ok {
|
||||
cacheItem := v.(*lruCacheItem)
|
||||
if atomic.LoadInt32(&cacheItem.disk.Status) == proto.ReadWrite {
|
||||
if blockValue, got := cacheItem.lruCache.Peek(key); got {
|
||||
block = blockValue.(*CacheBlock)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var cacheItem *lruCacheItem
|
||||
if cacheItem, err = c.selectAvailableLruCache(); err == nil {
|
||||
block = NewCacheBlockV2(cacheItem.config.Path, pDir, uniKey, allocSize, clientIP, cacheItem.disk)
|
||||
if ttl <= 0 {
|
||||
ttl = proto.DefaultCacheTTLSec
|
||||
}
|
||||
block.cacheEngine = c
|
||||
block.ttl = ttl
|
||||
|
||||
defer func() {
|
||||
if err != nil {
|
||||
block.Delete(fmt.Sprintf("create block failed %v", err))
|
||||
}
|
||||
}()
|
||||
if _, err = cacheItem.lruCache.CheckDiskSpace(block.rootPath, block.blockKey, block.getAllocSize()); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if err = block.initFilePath(false); err != nil {
|
||||
return
|
||||
}
|
||||
if _, err = cacheItem.lruCache.Set(key, block, time.Duration(ttl)*time.Second); err != nil {
|
||||
return
|
||||
}
|
||||
c.keyToDiskMap.Store(key, cacheItem)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (c *CacheEngine) createCacheBlockFromExistV2(dataPath string, volume string, uniKey string, allocSize uint64, clientIP string) (block *CacheBlock, err error) {
|
||||
key := GenCacheBlockKeyV2(volume, uniKey)
|
||||
v, ok := c.keyToDiskMap.Load(key)
|
||||
if ok {
|
||||
cacheItem := v.(*lruCacheItem)
|
||||
if atomic.LoadInt32(&cacheItem.disk.Status) == proto.ReadWrite {
|
||||
if blockValue, got := cacheItem.lruCache.Peek(key); got {
|
||||
block = blockValue.(*CacheBlock)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
v, ok = c.lruCacheMap.Load(dataPath)
|
||||
if !ok {
|
||||
return nil, errors.NewErrorf("no lru cache item related to dataPath(%v)", dataPath)
|
||||
}
|
||||
cacheItem := v.(*lruCacheItem)
|
||||
if atomic.LoadInt32(&cacheItem.disk.Status) == proto.Unavailable {
|
||||
return nil, errors.NewErrorf("lru cache item related to dataPath(%v) is unavailable", dataPath)
|
||||
}
|
||||
block = NewCacheBlockV2(cacheItem.config.Path, volume, uniKey, allocSize, clientIP, cacheItem.disk)
|
||||
block.cacheEngine = c
|
||||
defer func() {
|
||||
if err != nil {
|
||||
block.Delete(fmt.Sprintf("create block from exist failed %v", err))
|
||||
}
|
||||
}()
|
||||
|
||||
if err = block.initFilePath(true); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if _, err = cacheItem.lruCache.Set(key, block, time.Duration(block.ttl)*time.Second); err != nil {
|
||||
return
|
||||
}
|
||||
c.keyToDiskMap.Store(key, cacheItem)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (cb *CacheBlock) WriteAtV2(writeParam *proto.FlashWriteParam) (err error) {
|
||||
var file *os.File
|
||||
bgTime := stat.BeginStat()
|
||||
startTime := time.Now()
|
||||
defer func() {
|
||||
if err != nil {
|
||||
if IsDiskErr(err.Error()) {
|
||||
log.LogWarnf("[checkIsDiskError] data path(%v) meet io error", cb.filePath)
|
||||
cb.cacheEngine.triggerCacheError(cb.blockKey, cb.rootPath)
|
||||
}
|
||||
}
|
||||
stat.EndStat("CacheBlock:WriteAtV2", err, bgTime, 1)
|
||||
elapsed := time.Since(startTime)
|
||||
if elapsed > time.Second {
|
||||
log.LogWarnf("[WriteAtV2] WriteAt function (%v) cost %v", cb.filePath, elapsed.String())
|
||||
}
|
||||
}()
|
||||
usedSize := cb.getUsedSize()
|
||||
if allocSize := cb.getAllocSize(); writeParam.Offset != usedSize || writeParam.Offset >= allocSize || writeParam.DataSize == 0 || writeParam.Offset+writeParam.DataSize > allocSize {
|
||||
return fmt.Errorf("parameter offset=%d size=%d allocSize:%d blockUsedSize:%d", writeParam.Offset, writeParam.DataSize, allocSize, usedSize)
|
||||
}
|
||||
if file, err = cb.GetOrOpenFileHandler(); err != nil {
|
||||
log.LogWarnf("[WriteAtV2] GetOrOpenFileHandler (%v) err %v", cb.filePath, err)
|
||||
return
|
||||
}
|
||||
if _, err = file.WriteAt(writeParam.Data, writeParam.Offset+HeaderSize); err != nil {
|
||||
log.LogWarnf("[WriteAtV2] WriteAt (%v) data offset %v err %v", cb.filePath, writeParam.Offset+HeaderSize, err)
|
||||
return
|
||||
}
|
||||
n := writeParam.Offset/proto.PageSize*CRCLen + cb.allocSize + HeaderSize
|
||||
if log.EnableDebug() {
|
||||
log.LogDebugf("[WriteAtV2] file offset %v crc index %v and datasize %v", writeParam.Offset, n, writeParam.DataSize)
|
||||
}
|
||||
if _, err = file.WriteAt(writeParam.Crc[:CRCLen], n); err != nil {
|
||||
log.LogWarnf("[WriteAtV2] WriteAt (%v) crc offset %v err %v", cb.filePath, n, err)
|
||||
return
|
||||
}
|
||||
cb.maybeUpdateUsedSize(writeParam.Offset + writeParam.DataSize)
|
||||
return
|
||||
}
|
||||
|
||||
func (cb *CacheBlock) MaybeWriteCompleted() (err error) {
|
||||
if cb.usedSize != cb.allocSize {
|
||||
return
|
||||
}
|
||||
var file *os.File
|
||||
if file, err = cb.GetOrOpenFileHandler(); err != nil {
|
||||
log.LogErrorf("action[MaybeWriteCompleted], block:%s, get file handler err:%v", cb.blockKey, err)
|
||||
return
|
||||
}
|
||||
if err = cb.writeCacheBlockFileHeader(file); err != nil {
|
||||
log.LogErrorf("action[MaybeWriteCompleted], block:%s, write file header err:%v", cb.blockKey, err)
|
||||
return
|
||||
}
|
||||
if err = file.Sync(); err != nil {
|
||||
return
|
||||
}
|
||||
cb.notifyReady()
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -16,9 +16,12 @@ package cachengine
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"hash/crc32"
|
||||
"math/rand"
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@ -330,3 +333,214 @@ func testParallelOperation(t *testing.T) {
|
||||
close(stopCh)
|
||||
time.Sleep(time.Millisecond * 5)
|
||||
}
|
||||
|
||||
func initTestTmpfs1() (umount func() error, err error) {
|
||||
os.MkdirAll(testTmpFS1, 0o777)
|
||||
if !enabledTmpfs() {
|
||||
return func() error { return os.RemoveAll(testTmpFS1) }, nil
|
||||
}
|
||||
_, err = os.Stat(testTmpFS1)
|
||||
if err == nil {
|
||||
if tmpfs.IsTmpfs(testTmpFS1) {
|
||||
if err = tmpfs.Umount(testTmpFS1); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if !os.IsNotExist(err) {
|
||||
return
|
||||
}
|
||||
_ = os.MkdirAll(testTmpFS1, 0o777)
|
||||
}
|
||||
if err = tmpfs.MountTmpfs(testTmpFS1, 200*util.MB); err != nil {
|
||||
return
|
||||
}
|
||||
return func() error { return tmpfs.Umount(testTmpFS1) }, nil
|
||||
}
|
||||
|
||||
func TestBlockWriteCacheV2(t *testing.T) {
|
||||
umount, err := initTestTmpfs1()
|
||||
require.NoError(t, err)
|
||||
defer func() { require.NoError(t, umount()) }()
|
||||
|
||||
testWriteSingleFileV2(t)
|
||||
testWriteSingleFileErrorV2(t)
|
||||
testWriteCacheBlockFullV2(t)
|
||||
testWriteMultiCacheBlockV2(t)
|
||||
}
|
||||
|
||||
func testWriteSingleFileV2(t *testing.T) {
|
||||
disk := new(Disk)
|
||||
disk.Path = testTmpFS
|
||||
disk.Status = proto.ReadWrite
|
||||
uniKey := t.Name()
|
||||
pDir := MapKeyToDirectory(uniKey)
|
||||
cacheBlock := NewCacheBlockV2(testTmpFS, pDir, uniKey, 5120, "", disk)
|
||||
cacheBlock.cacheEngine = &CacheEngine{}
|
||||
cacheBlock.cacheEngine.lruFhCache = NewCache(LRUFileHandleCacheType, 1, -1, time.Hour,
|
||||
func(v interface{}, reason string) error {
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
},
|
||||
func(v interface{}) error {
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
})
|
||||
require.NoError(t, cacheBlock.initFilePath(false))
|
||||
defer func() { require.NoError(t, cacheBlock.Delete("test")) }()
|
||||
bytes := randTestData(4096)
|
||||
crcBuf := make([]byte, 4)
|
||||
crcSum1 := crc32.ChecksumIEEE(bytes)
|
||||
binary.BigEndian.PutUint32(crcBuf[:4], crcSum1)
|
||||
require.NoError(t, cacheBlock.WriteAtV2(&proto.FlashWriteParam{
|
||||
Offset: 0,
|
||||
Size: 5120,
|
||||
Data: bytes,
|
||||
Crc: crcBuf[:4],
|
||||
DataSize: 4096,
|
||||
}))
|
||||
bytes = randTestData(1024)
|
||||
crcSum2 := crc32.ChecksumIEEE(bytes)
|
||||
binary.BigEndian.PutUint32(crcBuf[:4], crcSum2)
|
||||
require.NoError(t, cacheBlock.WriteAtV2(&proto.FlashWriteParam{
|
||||
Offset: 4096,
|
||||
Size: 5120,
|
||||
Data: bytes,
|
||||
Crc: crcBuf[:4],
|
||||
DataSize: 1024,
|
||||
}))
|
||||
t.Logf("testWriteSingleFileV2, test:%s cacheBlock.datasize:%d", t.Name(), cacheBlock.usedSize)
|
||||
_ = cacheBlock.MaybeWriteCompleted()
|
||||
file, _ := cacheBlock.GetOrOpenFileHandler()
|
||||
_, _ = file.ReadAt(crcBuf[:4], 5120+HeaderSize)
|
||||
require.Equal(t, crcSum1, binary.BigEndian.Uint32(crcBuf[:4]))
|
||||
_, _ = file.ReadAt(crcBuf[:4], 5120+HeaderSize+4)
|
||||
require.Equal(t, crcSum2, binary.BigEndian.Uint32(crcBuf[:4]))
|
||||
}
|
||||
|
||||
func testWriteSingleFileErrorV2(t *testing.T) {
|
||||
disk := new(Disk)
|
||||
disk.Path = testTmpFS
|
||||
disk.Status = proto.ReadWrite
|
||||
uniKey := t.Name()
|
||||
pDir := MapKeyToDirectory(uniKey)
|
||||
cacheBlock := NewCacheBlockV2(testTmpFS, pDir, uniKey, 1024, "", disk)
|
||||
cacheBlock.cacheEngine = &CacheEngine{}
|
||||
cacheBlock.cacheEngine.lruFhCache = NewCache(LRUFileHandleCacheType, 1, -1, time.Hour,
|
||||
func(v interface{}, reason string) error {
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
},
|
||||
func(v interface{}) error {
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
})
|
||||
require.NoError(t, cacheBlock.initFilePath(false))
|
||||
defer func() { require.NoError(t, cacheBlock.Delete("test")) }()
|
||||
bytes := randTestData(1024)
|
||||
require.NoError(t, cacheBlock.WriteAtV2(&proto.FlashWriteParam{
|
||||
Offset: 0,
|
||||
Size: 1024,
|
||||
Data: bytes,
|
||||
Crc: make([]byte, CRCLen),
|
||||
DataSize: 1024,
|
||||
}))
|
||||
require.Error(t, cacheBlock.WriteAtV2(&proto.FlashWriteParam{
|
||||
Offset: proto.CACHE_BLOCK_SIZE,
|
||||
Size: 2048,
|
||||
Data: bytes,
|
||||
Crc: make([]byte, CRCLen),
|
||||
DataSize: 1024,
|
||||
}))
|
||||
t.Logf("testWriteSingleFileErrorV2, test:%s cacheBlock.datasize:%d", t.Name(), cacheBlock.usedSize)
|
||||
}
|
||||
|
||||
func testWriteCacheBlockFullV2(t *testing.T) {
|
||||
var err error
|
||||
disk := new(Disk)
|
||||
disk.Path = testTmpFS
|
||||
disk.Status = proto.ReadWrite
|
||||
uniKey := t.Name()
|
||||
pDir := MapKeyToDirectory(uniKey)
|
||||
cacheBlock := NewCacheBlockV2(testTmpFS, pDir, uniKey, proto.PageSize*1024, "", disk)
|
||||
cacheBlock.cacheEngine = &CacheEngine{}
|
||||
cacheBlock.cacheEngine.lruFhCache = NewCache(LRUFileHandleCacheType, 1, -1, time.Hour,
|
||||
func(v interface{}, reason string) error {
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
},
|
||||
func(v interface{}) error {
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
})
|
||||
require.NoError(t, cacheBlock.initFilePath(false))
|
||||
defer func() { require.NoError(t, cacheBlock.Delete("test")) }()
|
||||
bytes := randTestData(4096)
|
||||
var offset int64
|
||||
for {
|
||||
if err = cacheBlock.WriteAtV2(&proto.FlashWriteParam{
|
||||
Offset: offset,
|
||||
Size: proto.PageSize * 1024,
|
||||
Data: bytes,
|
||||
Crc: make([]byte, CRCLen),
|
||||
DataSize: 4096,
|
||||
}); err != nil {
|
||||
break
|
||||
}
|
||||
offset += 4096
|
||||
if offset/4096%1024 == 0 {
|
||||
t.Logf("testWriteCacheBlockFullV2, offset:%d cacheBlock.datasize:%d", offset, cacheBlock.usedSize)
|
||||
}
|
||||
}
|
||||
require.Equal(t, offset, int64(proto.PageSize*1024))
|
||||
}
|
||||
|
||||
func testWriteMultiCacheBlockV2(t *testing.T) {
|
||||
disk := new(Disk)
|
||||
disk.Path = testTmpFS
|
||||
disk.Status = proto.ReadWrite
|
||||
count := 100
|
||||
wg := sync.WaitGroup{}
|
||||
for i := 0; i < 10; i++ {
|
||||
wg.Add(1)
|
||||
go func(index int) {
|
||||
defer wg.Done()
|
||||
var err error
|
||||
uniKey := t.Name() + strconv.Itoa(index)
|
||||
pDir := MapKeyToDirectory(uniKey)
|
||||
cacheBlock := NewCacheBlockV2(testTmpFS, pDir, uniKey, 4096*100, "", disk)
|
||||
defer func() { require.NoError(t, cacheBlock.Delete("test")) }()
|
||||
cacheBlock.cacheEngine = &CacheEngine{}
|
||||
cacheBlock.cacheEngine.lruFhCache = NewCache(LRUFileHandleCacheType, 1, -1, time.Hour,
|
||||
func(v interface{}, reason string) error {
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
},
|
||||
func(v interface{}) error {
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
})
|
||||
err = cacheBlock.initFilePath(false)
|
||||
require.NoError(t, err)
|
||||
bytes := randTestData(4096)
|
||||
offset := int64(0)
|
||||
for j := 0; j < count; j++ {
|
||||
if err = cacheBlock.WriteAtV2(&proto.FlashWriteParam{
|
||||
Offset: offset,
|
||||
Size: 4096 * 100,
|
||||
Data: bytes,
|
||||
Crc: make([]byte, CRCLen),
|
||||
DataSize: 4096,
|
||||
}); err != nil {
|
||||
break
|
||||
}
|
||||
offset += 4096
|
||||
if j%4096 == 0 {
|
||||
t.Logf("testWriteMultiCacheBlock, pdir:%v, write count:%v, cacheBlock.datasize:%d", pDir, j, cacheBlock.usedSize)
|
||||
}
|
||||
}
|
||||
require.Equal(t, offset, int64(4096*100))
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
@ -16,6 +16,7 @@ package cachengine
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"hash/crc32"
|
||||
syslog "log"
|
||||
"math"
|
||||
"os"
|
||||
@ -42,6 +43,7 @@ import (
|
||||
|
||||
const (
|
||||
EnvDockerTmpfs = "DOCKER_FLASHNODE_TMPFS_OFF"
|
||||
DirMod = 512
|
||||
|
||||
DefaultExpireTime = 60 * 60
|
||||
InitFileName = "flash.init"
|
||||
@ -73,15 +75,17 @@ type cachePrepareTask struct {
|
||||
}
|
||||
|
||||
type cacheLoadTask struct {
|
||||
volume string
|
||||
dataPath string
|
||||
volume string
|
||||
dataPath string
|
||||
sourceType string
|
||||
}
|
||||
|
||||
type cacheLoadFile struct {
|
||||
volume string
|
||||
dataPath string
|
||||
fullPath string
|
||||
fileName string
|
||||
volume string
|
||||
dataPath string
|
||||
fullPath string
|
||||
fileName string
|
||||
sourceType string
|
||||
}
|
||||
|
||||
type lruCacheItem struct {
|
||||
@ -182,15 +186,17 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
|
||||
file := v.(*os.File)
|
||||
return file.Close()
|
||||
})
|
||||
|
||||
if _, err = os.Stat(fullPath); err != nil {
|
||||
if !os.IsNotExist(err.(*os.PathError)) {
|
||||
return
|
||||
}
|
||||
if err = os.Mkdir(fullPath, 0o755); err != nil {
|
||||
if !os.IsExist(err) {
|
||||
pPaths := []string{fullPath, fullPath + SourceTypeDefault, fullPath + SourceTypeBlock}
|
||||
for _, pPath := range pPaths {
|
||||
if _, err = os.Stat(pPath); err != nil {
|
||||
if !os.IsNotExist(err.(*os.PathError)) {
|
||||
return
|
||||
}
|
||||
if err = os.Mkdir(pPath, 0o755); err != nil {
|
||||
if !os.IsExist(err) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -227,16 +233,20 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
|
||||
s.lruCacheMap.Store(fullPath, &lruCacheItem{lruCache: cache, config: diskCacheConfig, disk: d})
|
||||
s.totalCacheNum++
|
||||
|
||||
if _, err = os.Stat(fullPath); err != nil {
|
||||
if !os.IsNotExist(err.(*os.PathError)) {
|
||||
return
|
||||
}
|
||||
if err = os.Mkdir(fullPath, 0o755); err != nil {
|
||||
if !os.IsExist(err) {
|
||||
pPaths := []string{fullPath, fullPath + SourceTypeDefault, fullPath + SourceTypeBlock}
|
||||
for _, pPath := range pPaths {
|
||||
if _, err = os.Stat(pPath); err != nil {
|
||||
if !os.IsNotExist(err.(*os.PathError)) {
|
||||
return
|
||||
}
|
||||
if err = os.Mkdir(pPath, 0o755); err != nil {
|
||||
if !os.IsExist(err) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
s.lruFhCache = NewCache(LRUFileHandleCacheType, fhCapacity, -1, expireTime,
|
||||
func(v interface{}, reason string) error {
|
||||
@ -308,6 +318,7 @@ func (c *CacheEngine) LoadDisk(diskPath string) (err error) {
|
||||
cbNum atomicutil.Int64
|
||||
errorCbNum atomicutil.Int64
|
||||
)
|
||||
storages := []string{SourceTypeDefault, SourceTypeBlock}
|
||||
begin := time.Now()
|
||||
defer func() {
|
||||
msg := fmt.Sprintf("[LoadDisk] dataPath(%v) load all cacheBlock(%v) using time(%v), unloaded cacheBlock num is (%v)", diskPath, cbNum.Load(), time.Since(begin), errorCbNum.Load())
|
||||
@ -344,26 +355,29 @@ func (c *CacheEngine) LoadDisk(diskPath string) (err error) {
|
||||
}
|
||||
for _, fileInfo := range fileInfoList {
|
||||
filename := fileInfo.Name()
|
||||
if !c.isCacheBlockFileName(filename) {
|
||||
if task.sourceType == SourceTypeDefault && !c.isCacheBlockFileName(filename) {
|
||||
log.LogWarnf("[LoadDisk] find invalid cacheBlock file[%v] on dataPath(%v)", filename, fullPath)
|
||||
continue
|
||||
}
|
||||
filePathChan <- cacheLoadFile{volume: volume, dataPath: diskPath, fullPath: fullPath, fileName: filename}
|
||||
filePathChan <- cacheLoadFile{volume: volume, dataPath: diskPath, fullPath: fullPath, fileName: filename, sourceType: task.sourceType}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
log.LogDebugf("action[LoadDisk] load cacheBlock from path(%v).", diskPath)
|
||||
entries, err := os.ReadDir(diskPath)
|
||||
if err != nil {
|
||||
log.LogErrorf("action[LoadDisk] read dir(%v) err(%v).", diskPath, err)
|
||||
close(cacheLoadTaskCh)
|
||||
close(filePathChan)
|
||||
return
|
||||
}
|
||||
for _, volEntry := range entries {
|
||||
cacheLoadTaskCh <- cacheLoadTask{volume: volEntry.Name(), dataPath: diskPath}
|
||||
for _, s := range storages {
|
||||
sPath := diskPath + s
|
||||
entries, err1 := os.ReadDir(sPath)
|
||||
if err1 != nil {
|
||||
log.LogErrorf("action[LoadDisk] read dir(%v) err(%v).", sPath, err1)
|
||||
close(cacheLoadTaskCh)
|
||||
close(filePathChan)
|
||||
return err1
|
||||
}
|
||||
for _, volEntry := range entries {
|
||||
cacheLoadTaskCh <- cacheLoadTask{volume: volEntry.Name(), dataPath: sPath, sourceType: s}
|
||||
}
|
||||
}
|
||||
close(cacheLoadTaskCh)
|
||||
dirScanWg.Wait()
|
||||
@ -373,21 +387,33 @@ func (c *CacheEngine) LoadDisk(diskPath string) (err error) {
|
||||
}
|
||||
|
||||
func (c *CacheEngine) handlerFile(file *cacheLoadFile, cbNum *atomicutil.Int64, errorCbNum *atomicutil.Int64) {
|
||||
inode, offset, version, err := unmarshalCacheBlockName(file.fileName)
|
||||
if err != nil {
|
||||
log.LogErrorf("action[LoadDisk] unmarshal cacheBlockName(%v) from dataPath(%v) volume(%v) err(%v) ",
|
||||
file.fileName, file.fullPath, file.volume, err.Error())
|
||||
return
|
||||
}
|
||||
log.LogDebugf("acton[LoadDisk] dataPath(%v) cacheBlockName(%v) volume(%v) inode(%v) offset(%v) version(%v).",
|
||||
file.fullPath, file.fileName, file.volume, inode, offset, version)
|
||||
if SourceTypeDefault == file.sourceType {
|
||||
inode, offset, version, err := unmarshalCacheBlockName(file.fileName)
|
||||
if err != nil {
|
||||
log.LogErrorf("action[LoadDisk] unmarshal cacheBlockName(%v) from dataPath(%v) volume(%v) err(%v) ",
|
||||
file.fileName, file.fullPath, file.volume, err.Error())
|
||||
return
|
||||
}
|
||||
log.LogDebugf("acton[LoadDisk] dataPath(%v) cacheBlockName(%v) volume(%v) inode(%v) offset(%v) version(%v).",
|
||||
file.fullPath, file.fileName, file.volume, inode, offset, version)
|
||||
|
||||
if _, err := c.createCacheBlockFromExist(file.dataPath, file.volume, inode, offset, version, 0, ""); err != nil {
|
||||
c.deleteCacheBlock(GenCacheBlockKey(file.volume, inode, offset, version))
|
||||
log.LogInfof("action[LoadDisk] createCacheBlock(%v) from dataPath(%v) volume(%v) err(%v) ",
|
||||
file.fileName, file.fullPath, file.volume, err.Error())
|
||||
errorCbNum.Add(1)
|
||||
return
|
||||
if _, err := c.createCacheBlockFromExist(file.dataPath, file.volume, inode, offset, version, 0, ""); err != nil {
|
||||
c.DeleteCacheBlock(GenCacheBlockKey(file.volume, inode, offset, version))
|
||||
log.LogInfof("action[LoadDisk] createCacheBlock(%v) from dataPath(%v) volume(%v) err(%v) ",
|
||||
file.fileName, file.fullPath, file.volume, err.Error())
|
||||
errorCbNum.Add(1)
|
||||
return
|
||||
}
|
||||
} else {
|
||||
log.LogDebugf("acton[LoadDisk] dataPath(%v) cacheBlockName(%v) volume(%v)",
|
||||
file.fullPath, file.fileName, file.volume)
|
||||
if _, err := c.createCacheBlockFromExistV2(file.dataPath, file.volume, file.fileName, 0, ""); err != nil {
|
||||
c.DeleteCacheBlock(GenCacheBlockKeyV2(file.volume, file.fileName))
|
||||
log.LogInfof("action[LoadDisk] createCacheBlock(%v) from dataPath(%v) volume(%v) err(%v) ",
|
||||
file.fileName, file.fullPath, file.volume, err.Error())
|
||||
errorCbNum.Add(1)
|
||||
return
|
||||
}
|
||||
}
|
||||
cbNum.Add(1)
|
||||
}
|
||||
@ -497,7 +523,7 @@ func (c *CacheEngine) initTmpfs() (err error) {
|
||||
return fd.Close()
|
||||
}
|
||||
|
||||
func (c *CacheEngine) deleteCacheBlock(key string) {
|
||||
func (c *CacheEngine) DeleteCacheBlock(key string) {
|
||||
value, ok := c.keyToDiskMap.Load(key)
|
||||
if ok {
|
||||
cacheItem := value.(*lruCacheItem)
|
||||
@ -780,7 +806,7 @@ func (c *CacheEngine) CreateBlock(req *proto.CacheRequest, clientIP string, isPr
|
||||
if block, err = c.createCacheBlock(req.Volume, req.Inode, req.FixedFileOffset, req.Version, req.TTL, computeAllocSize(req.Sources), clientIP, isPrepare); err != nil {
|
||||
log.LogWarnf("action[CreateBlock] createCacheBlock(%v) failed err %v ",
|
||||
GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version), err)
|
||||
c.deleteCacheBlock(GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version))
|
||||
c.DeleteCacheBlock(GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version))
|
||||
return nil, err
|
||||
}
|
||||
return block, nil
|
||||
@ -885,6 +911,15 @@ func GenCacheBlockKey(volume string, inode, offset uint64, version uint32) strin
|
||||
return path.Join(volume, u(inode, 10)+"#"+u(offset, 10)+"#"+u(uint64(version), 10))
|
||||
}
|
||||
|
||||
func GenCacheBlockKeyV2(pDir string, key string) string {
|
||||
return path.Join(pDir, key)
|
||||
}
|
||||
|
||||
func MapKeyToDirectory(key string) string {
|
||||
dirNum := crc32.ChecksumIEEE([]byte(key)) & 0xFFF % DirMod
|
||||
return strconv.Itoa(int(dirNum))
|
||||
}
|
||||
|
||||
func enabledTmpfs() bool {
|
||||
return os.Getenv(EnvDockerTmpfs) == ""
|
||||
}
|
||||
|
||||
@ -31,7 +31,8 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
testTmpFS = "/cfs_test/tmpfs"
|
||||
testTmpFS = "/cfs_test/tmpfs"
|
||||
testTmpFS1 = "/cfs_test/tmpfsblocks"
|
||||
)
|
||||
|
||||
var bytesCommon = randTestData(1024)
|
||||
|
||||
190
remotecache/flashnode/cachengine/miss_cache.go
Normal file
190
remotecache/flashnode/cachengine/miss_cache.go
Normal file
@ -0,0 +1,190 @@
|
||||
package cachengine
|
||||
|
||||
import (
|
||||
"container/list"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
)
|
||||
|
||||
const (
|
||||
// MinMissCacheEvictNum is used in the foreground eviction.
|
||||
// When clearing the miss keys from the cache, it stops as soon as 10 inodes have been evicted.
|
||||
MinMissCacheEvictNum = 10
|
||||
// MaxMissCacheEvictNum is used in the background. We can evict 200000 inodes at max.
|
||||
MaxMissCacheEvictNum = 100000
|
||||
|
||||
BgEvictionInterval = 1 * time.Minute
|
||||
)
|
||||
|
||||
// MissCache defines the structure of the missed key cache.
|
||||
type MissCache struct {
|
||||
sync.RWMutex
|
||||
cache map[string]*list.Element
|
||||
lruList *list.List
|
||||
expiration time.Duration
|
||||
maxElements int
|
||||
}
|
||||
|
||||
func NewMissCache(exp time.Duration, maxElements int) *MissCache {
|
||||
mc := &MissCache{
|
||||
cache: make(map[string]*list.Element),
|
||||
lruList: list.New(),
|
||||
expiration: exp,
|
||||
maxElements: maxElements,
|
||||
}
|
||||
go mc.backgroundEviction()
|
||||
return mc
|
||||
}
|
||||
|
||||
func (mc *MissCache) Increment(key string) int32 {
|
||||
mc.Lock()
|
||||
defer mc.Unlock()
|
||||
e, ok := mc.cache[key]
|
||||
if !ok {
|
||||
cme := &proto.CacheMissEntry{
|
||||
UniKey: key,
|
||||
MissCount: 1,
|
||||
}
|
||||
if mc.lruList.Len() >= mc.maxElements {
|
||||
mc.evict(true)
|
||||
}
|
||||
proto.SetMissEntryExpiration(cme, mc.expiration)
|
||||
|
||||
element := mc.lruList.PushFront(cme)
|
||||
mc.cache[cme.UniKey] = element
|
||||
if log.EnableDebug() {
|
||||
log.LogDebugf("MissCache put miss cache: unikey(%v) expire(%v)",
|
||||
cme.UniKey, proto.GetMissEntryExpiration(cme))
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
info := e.Value.(*proto.CacheMissEntry)
|
||||
if proto.CacheMissExpired(info) {
|
||||
mc.lruList.Remove(e)
|
||||
delete(mc.cache, info.UniKey)
|
||||
cme := &proto.CacheMissEntry{
|
||||
UniKey: key,
|
||||
MissCount: 1,
|
||||
}
|
||||
if mc.lruList.Len() >= mc.maxElements {
|
||||
mc.evict(true)
|
||||
}
|
||||
proto.SetMissEntryExpiration(cme, mc.expiration)
|
||||
|
||||
element := mc.lruList.PushFront(cme)
|
||||
mc.cache[cme.UniKey] = element
|
||||
return 1
|
||||
}
|
||||
return AtomicLoadAndAddWithCAS(&info.MissCount)
|
||||
}
|
||||
|
||||
func AtomicLoadAndAddWithCAS(addr *int32) int32 {
|
||||
for {
|
||||
old := atomic.LoadInt32(addr)
|
||||
newVal := old + 1
|
||||
if atomic.CompareAndSwapInt32(addr, old, newVal) {
|
||||
return newVal
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (mc *MissCache) Get(uniKey string) *proto.CacheMissEntry {
|
||||
mc.RLock()
|
||||
element, ok := mc.cache[uniKey]
|
||||
if !ok {
|
||||
mc.RUnlock()
|
||||
log.LogDebugf("Miss Cache not found %v", uniKey)
|
||||
return nil
|
||||
}
|
||||
|
||||
info := element.Value.(*proto.CacheMissEntry)
|
||||
if proto.CacheMissExpired(info) {
|
||||
mc.RUnlock()
|
||||
log.LogDebugf("Miss Cache %v expired", info)
|
||||
return nil
|
||||
}
|
||||
mc.RUnlock()
|
||||
return info
|
||||
}
|
||||
|
||||
func (mc *MissCache) backgroundEviction() {
|
||||
t := time.NewTicker(BgEvictionInterval)
|
||||
defer t.Stop()
|
||||
|
||||
for range t.C {
|
||||
log.LogInfof("MissCache: start BG evict")
|
||||
start := time.Now()
|
||||
mc.Lock()
|
||||
mc.evict(false)
|
||||
mc.Unlock()
|
||||
elapsed := time.Since(start)
|
||||
log.LogInfof("MissCache: total miss cache(%d), cost(%d)ns", mc.lruList.Len(), elapsed.Nanoseconds())
|
||||
}
|
||||
}
|
||||
|
||||
func (mc *MissCache) Delete(uniKey string) {
|
||||
log.LogDebugf("MissCache Delete: key(%v)", uniKey)
|
||||
mc.Lock()
|
||||
element, ok := mc.cache[uniKey]
|
||||
if ok {
|
||||
mc.lruList.Remove(element)
|
||||
delete(mc.cache, uniKey)
|
||||
}
|
||||
mc.Unlock()
|
||||
}
|
||||
|
||||
// Foreground eviction cares more about the speed.
|
||||
// Background eviction evicts all expired items from the cache.
|
||||
// The caller should grab the WRITE lock of the inode cache.
|
||||
func (mc *MissCache) evict(foreground bool) {
|
||||
var count int
|
||||
|
||||
for i := 0; i < MinMissCacheEvictNum; i++ {
|
||||
element := mc.lruList.Back()
|
||||
if element == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// For background eviction, if all expired items have been evicted, just return
|
||||
// But for foreground eviction, we need to evict at least MinInodeCacheEvictNum inodes.
|
||||
// The foreground eviction, does not need to care if the inode has expired or not.
|
||||
info := element.Value.(*proto.CacheMissEntry)
|
||||
log.LogDebugf("MissCache check miss key(%v)", info.UniKey)
|
||||
if !foreground && !proto.CacheMissExpired(info) {
|
||||
log.LogDebugf("MissCache check key(%v) expired(%v)",
|
||||
info.UniKey, proto.CacheMissExpired(info))
|
||||
return
|
||||
}
|
||||
|
||||
log.LogDebugf("MissCache remove key(%v)", info.UniKey)
|
||||
mc.lruList.Remove(element)
|
||||
delete(mc.cache, info.UniKey)
|
||||
count++
|
||||
}
|
||||
// For background eviction, we need to continue evict all expired items from the cache
|
||||
if foreground {
|
||||
return
|
||||
}
|
||||
|
||||
for i := 0; i < MaxMissCacheEvictNum; i++ {
|
||||
element := mc.lruList.Back()
|
||||
if element == nil {
|
||||
break
|
||||
}
|
||||
info := element.Value.(*proto.CacheMissEntry)
|
||||
if !proto.CacheMissExpired(info) {
|
||||
log.LogDebugf("MissCache check key(%v) expired(%v)",
|
||||
info.UniKey, proto.CacheMissExpired(info))
|
||||
break
|
||||
}
|
||||
log.LogDebugf("MissCache remove key(%v)", info.UniKey)
|
||||
mc.lruList.Remove(element)
|
||||
delete(mc.cache, info.UniKey)
|
||||
count++
|
||||
}
|
||||
}
|
||||
@ -74,6 +74,9 @@ const (
|
||||
_defaultManualScanLimitBurst = 1000
|
||||
_slotStatValidPeriod = 10 * time.Minute // min
|
||||
_defaultPrepareRoutineNum = 20
|
||||
_defaultMissEntryExpiration = 2 * time.Minute
|
||||
_defaultMaxMissEntryCache = 100000
|
||||
_defaultMissCountThresholdInterval = 5
|
||||
)
|
||||
|
||||
// Configuration keys
|
||||
@ -107,6 +110,7 @@ const (
|
||||
cfgPrepareLimitPerSecond = "prepareLimitPerSecond"
|
||||
cfgWaitForBlockCache = "waitForBlockCache"
|
||||
cfgPrepareLoadRoutineNum = "prepareLoadRoutineNum"
|
||||
cfgMissEntryTimeout = "missEntryTimeout"
|
||||
paramIocc = "iocc"
|
||||
paramFlow = "flow"
|
||||
paramFactor = "factor"
|
||||
@ -171,6 +175,7 @@ type FlashNode struct {
|
||||
|
||||
slotMap sync.Map // [uint32]*SlotStat
|
||||
readCount uint64
|
||||
missCache *cachengine.MissCache
|
||||
}
|
||||
|
||||
// Start starts up the flash node with the specified configuration.
|
||||
@ -426,6 +431,13 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) {
|
||||
log.LogInfof("[parseConfig] load diskDataPath[%v] totalSize[%d] capacity[%d]", d.Path, d.TotalSpace, d.Capacity)
|
||||
}
|
||||
|
||||
missCacheEntryExpiration := _defaultMissEntryExpiration
|
||||
missCacheTimeOut := cfg.GetInt(cfgMissEntryTimeout)
|
||||
if missCacheTimeOut > 0 {
|
||||
missCacheEntryExpiration = time.Duration(missCacheTimeOut) * time.Second
|
||||
}
|
||||
f.missCache = cachengine.NewMissCache(missCacheEntryExpiration, _defaultMaxMissEntryCache)
|
||||
|
||||
taskCountLimit := cfg.GetInt(cfgNodeTaskCountLimit)
|
||||
if taskCountLimit <= 0 {
|
||||
taskCountLimit = _defaultFlashNodeTaskCountLimit
|
||||
|
||||
@ -19,9 +19,11 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
@ -29,6 +31,7 @@ import (
|
||||
"github.com/cubefs/cubefs/sdk/data/stream"
|
||||
"github.com/cubefs/cubefs/sdk/meta"
|
||||
"github.com/cubefs/cubefs/util"
|
||||
"github.com/cubefs/cubefs/util/bytespool"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
"github.com/cubefs/cubefs/util/exporter"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
@ -56,6 +59,10 @@ func (f *FlashNode) handlePacket(conn net.Conn, p *proto.Packet) (err error) {
|
||||
err = f.opCachePrepare(conn, p)
|
||||
case proto.OpFlashNodeCacheRead:
|
||||
err = f.opCacheRead(conn, p)
|
||||
case proto.OpFlashNodeCachePutBlock:
|
||||
err = f.opCachePutBlock(conn, p)
|
||||
case proto.OpFlashNodeCacheDelete:
|
||||
err = f.opCacheDelete(conn, p)
|
||||
case proto.OpFlashNodeSetReadIOLimits:
|
||||
err = f.opSetReadIOLimits(conn, p)
|
||||
case proto.OpFlashNodeSetWriteIOLimits:
|
||||
@ -249,6 +256,135 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
func (f *FlashNode) opCacheDelete(conn net.Conn, p *proto.Packet) (err error) {
|
||||
data := p.Data
|
||||
defer func() {
|
||||
if err != nil {
|
||||
log.LogWarnf("action[opCacheDelete] end, logMsg:%s",
|
||||
p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err))
|
||||
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
|
||||
} else {
|
||||
p.PacketOkReply()
|
||||
}
|
||||
if e := p.WriteToConn(conn); e != nil {
|
||||
log.LogErrorf("action[opCacheDelete] write to conn %v", e)
|
||||
}
|
||||
}()
|
||||
if p.Size == 0 {
|
||||
return fmt.Errorf("no cache delete request")
|
||||
}
|
||||
uniKey := string(data)
|
||||
pDir := cachengine.MapKeyToDirectory(uniKey)
|
||||
f.cacheEngine.DeleteCacheBlock(cachengine.GenCacheBlockKeyV2(pDir, uniKey))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error) {
|
||||
defer func() {
|
||||
if err != nil {
|
||||
log.LogWarnf("action[opCachePutBlock] end, logMsg:%s",
|
||||
p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err))
|
||||
p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error()))
|
||||
if e := p.WriteToConn(conn); e != nil {
|
||||
log.LogErrorf("action[opCachePutBlock] write to conn %v", e)
|
||||
}
|
||||
}
|
||||
}()
|
||||
if len(p.Data) == 0 {
|
||||
return fmt.Errorf("unable to build a key from the packet due to lack of available parameters")
|
||||
}
|
||||
req := new(proto.PutBlockHead)
|
||||
if err = p.UnmarshalDataPb(req); err != nil {
|
||||
return fmt.Errorf("error parsing cache write head structure")
|
||||
}
|
||||
uniKey := req.UniKey
|
||||
pDir := cachengine.MapKeyToDirectory(uniKey)
|
||||
blockKey := cachengine.GenCacheBlockKeyV2(pDir, uniKey)
|
||||
_, err1 := f.cacheEngine.PeekCacheBlock(blockKey)
|
||||
if err1 == nil {
|
||||
if log.EnableDebug() {
|
||||
log.LogDebug("action[opCachePutBlock] check block key:"+uniKey+" logMsg:%s",
|
||||
p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err))
|
||||
}
|
||||
err = fmt.Errorf("block %v already exsit", uniKey)
|
||||
return err
|
||||
}
|
||||
err1 = nil
|
||||
if log.EnableDebug() {
|
||||
log.LogDebug("action[opCachePutBlock] create block key:"+uniKey+" logMsg:%s",
|
||||
p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err))
|
||||
}
|
||||
missTaskDone := make(chan struct{})
|
||||
err = f.limitWrite.TryRunAsync(context.Background(), int(req.BlockLen), false, func() {
|
||||
defer func() {
|
||||
close(missTaskDone)
|
||||
}()
|
||||
if cb, err2, created := f.cacheEngine.CreateBlockV2(pDir, uniKey, req.TTL, uint32(req.BlockLen), conn.RemoteAddr().String()); err2 != nil || created {
|
||||
err = fmt.Errorf("already create block(%v)", cachengine.GenCacheBlockKeyV2(pDir, uniKey))
|
||||
return
|
||||
} else {
|
||||
p.PacketOkReply()
|
||||
if e := p.WriteToConn(conn); e != nil {
|
||||
log.LogErrorf("action[opCachePutBlock] write to conn %v", e)
|
||||
}
|
||||
var (
|
||||
totalWritten int64
|
||||
n int
|
||||
readSize int
|
||||
)
|
||||
buf := bytespool.Alloc(proto.PageSize)
|
||||
defer bytespool.Free(buf)
|
||||
crcBuf := bytespool.Alloc(cachengine.CRCLen)
|
||||
defer bytespool.Free(crcBuf)
|
||||
for {
|
||||
readSize = proto.PageSize
|
||||
if totalWritten+int64(readSize) > req.BlockLen {
|
||||
readSize = int(req.BlockLen - totalWritten)
|
||||
}
|
||||
if n, err1 = io.ReadFull(conn, buf[:readSize]); err1 != nil {
|
||||
return
|
||||
}
|
||||
if n != readSize {
|
||||
err1 = syscall.EBADMSG
|
||||
}
|
||||
if n, err1 = io.ReadFull(conn, crcBuf[:cachengine.CRCLen]); err1 != nil {
|
||||
return
|
||||
}
|
||||
if n != cachengine.CRCLen {
|
||||
err1 = syscall.EBADMSG
|
||||
}
|
||||
err1 = cb.WriteAtV2(&proto.FlashWriteParam{
|
||||
Offset: totalWritten,
|
||||
Size: req.BlockLen,
|
||||
Data: buf[:readSize],
|
||||
Crc: crcBuf[:cachengine.CRCLen],
|
||||
DataSize: int64(readSize),
|
||||
})
|
||||
err1 = cb.MaybeWriteCompleted()
|
||||
if err1 != nil {
|
||||
return
|
||||
}
|
||||
totalWritten += int64(readSize)
|
||||
if totalWritten == req.BlockLen {
|
||||
if log.EnableDebug() {
|
||||
log.LogDebugf("action[opCachePutBlock] total write %v", totalWritten)
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
if err == nil {
|
||||
<-missTaskDone
|
||||
}
|
||||
if err1 != nil {
|
||||
f.cacheEngine.DeleteCacheBlock(blockKey)
|
||||
err = err1
|
||||
return
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (f *FlashNode) opFlashNodeScan(conn net.Conn, p *proto.Packet) (err error) {
|
||||
data := p.Data
|
||||
responseAckOKToMaster(conn, p)
|
||||
@ -644,6 +780,22 @@ func (f *FlashNode) opSetWriteIOLimits(conn net.Conn, p *proto.Packet) (err erro
|
||||
return
|
||||
}
|
||||
|
||||
//nolint:unused // used for new read operation
|
||||
func (f *FlashNode) shouldCache(key string) error {
|
||||
var count int32
|
||||
cm := f.missCache.Get(key)
|
||||
if cm == nil {
|
||||
count = f.missCache.Increment(key)
|
||||
} else {
|
||||
count = cachengine.AtomicLoadAndAddWithCAS(&cm.MissCount)
|
||||
}
|
||||
if (count-1) == 0 || count == _defaultMissCountThresholdInterval {
|
||||
return proto.ErrorNotExistShouldCache
|
||||
} else {
|
||||
return proto.ErrorNotExistShouldNotCache
|
||||
}
|
||||
}
|
||||
|
||||
func responseAckOKToMaster(conn net.Conn, p *proto.Packet) {
|
||||
p.PacketOkReply()
|
||||
if err := p.WriteToConn(conn); err != nil {
|
||||
|
||||
@ -15,8 +15,13 @@
|
||||
package flashnode
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"hash/crc32"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@ -48,6 +53,8 @@ func testTCP(t *testing.T) {
|
||||
t.Run("CachePrepare", testTCPCachePrepare)
|
||||
t.Run("CacheRead", testTCPCacheRead)
|
||||
t.Run("ManualScan", testTCPManualScan)
|
||||
t.Run("CachePutBlock", testTCPCachePutBlock)
|
||||
t.Run("CacheDelete", testTCPCacheDelete)
|
||||
}
|
||||
|
||||
func testTCPHeartbeat(t *testing.T) {
|
||||
@ -169,6 +176,53 @@ func testTCPCacheRead(t *testing.T) {
|
||||
require.Equal(t, uint32(blockSize), r.Size)
|
||||
}
|
||||
|
||||
func testTCPCachePutBlock(t *testing.T) {
|
||||
conn := newTCPConn(t)
|
||||
defer conn.Close()
|
||||
p := proto.NewPacketReqID()
|
||||
r := proto.NewPacket()
|
||||
p.Opcode = proto.OpFlashNodeCachePutBlock
|
||||
require.NoError(t, p.WriteToConn(conn))
|
||||
require.NoError(t, r.ReadFromConn(conn, 3))
|
||||
require.Equal(t, proto.OpErr, r.ResultCode) // lack parameter
|
||||
key := "testTCPCacheWrite_key"
|
||||
req := &proto.PutBlockHead{
|
||||
UniKey: key,
|
||||
BlockLen: 1,
|
||||
TTL: 0,
|
||||
}
|
||||
_ = p.MarshalDataPb(req)
|
||||
p.Opcode = proto.OpFlashNodeCachePutBlock
|
||||
require.NoError(t, p.WriteToConn(conn))
|
||||
require.NoError(t, r.ReadFromConn(conn, 3))
|
||||
require.Equal(t, proto.OpOk, r.ResultCode)
|
||||
buf := []byte{'{'}
|
||||
n := len(buf)
|
||||
_, err := conn.Write(buf[:n])
|
||||
require.NoError(t, err)
|
||||
crcBuf := make([]byte, 4)
|
||||
binary.BigEndian.PutUint32(crcBuf, crc32.ChecksumIEEE(buf[:n]))
|
||||
_, err = conn.Write(crcBuf[:4])
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func testTCPCacheDelete(t *testing.T) {
|
||||
conn := newTCPConn(t)
|
||||
defer conn.Close()
|
||||
p := proto.NewPacketReqID()
|
||||
r := proto.NewPacket()
|
||||
p.Opcode = proto.OpFlashNodeCacheDelete
|
||||
require.NoError(t, p.WriteToConn(conn))
|
||||
require.NoError(t, r.ReadFromConn(conn, 3))
|
||||
require.Equal(t, proto.OpErr, r.ResultCode)
|
||||
|
||||
p.Data = ([]byte)("testTCPCacheWrite_key")
|
||||
p.Size = uint32(len(p.Data))
|
||||
require.NoError(t, p.WriteToConn(conn))
|
||||
require.NoError(t, r.ReadFromConn(conn, 3))
|
||||
require.Equal(t, proto.OpOk, r.ResultCode)
|
||||
}
|
||||
|
||||
func testTCPManualScan(t *testing.T) {
|
||||
conn := newTCPConn(t)
|
||||
p := proto.NewPacketReqID()
|
||||
@ -215,3 +269,31 @@ func testTCPManualScan(t *testing.T) {
|
||||
require.Equal(t, proto.OpOk, r.ResultCode)
|
||||
_ = conn.Close()
|
||||
}
|
||||
|
||||
func testShouldCache(t *testing.T) {
|
||||
var wg sync.WaitGroup
|
||||
key := "testCacheKey"
|
||||
goroutines := 10
|
||||
var countA, countB int64
|
||||
for i := 0; i < goroutines; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
err := flashServer.shouldCache(key)
|
||||
switch {
|
||||
case errors.Is(err, proto.ErrorNotExistShouldCache):
|
||||
atomic.AddInt64(&countA, 1)
|
||||
case errors.Is(err, proto.ErrorNotExistShouldNotCache):
|
||||
atomic.AddInt64(&countB, 1)
|
||||
default:
|
||||
t.Errorf("unexpected error: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
flashServer.missCache.Delete(key)
|
||||
t.Logf("error not exist should cache count: %d", countA)
|
||||
t.Logf("error not exist should not cache count: %d", countB)
|
||||
require.Equal(t, countA, int64(2))
|
||||
require.Equal(t, countB, int64(8))
|
||||
}
|
||||
|
||||
@ -28,6 +28,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/remotecache/flashnode/cachengine"
|
||||
"github.com/cubefs/cubefs/sdk/master"
|
||||
"github.com/cubefs/cubefs/util/config"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
@ -98,12 +99,13 @@ func handleMaster(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
func newFlashnode() *FlashNode {
|
||||
return &FlashNode{mc: master.NewMasterClient([]string{masterAddr}, false)}
|
||||
return &FlashNode{mc: master.NewMasterClient([]string{masterAddr}, false), missCache: cachengine.NewMissCache(_defaultMissEntryExpiration, _defaultMaxMissEntryCache)}
|
||||
}
|
||||
|
||||
func TestFlashNode(t *testing.T) {
|
||||
t.Run("New", testNew)
|
||||
t.Run("Config", testConfig)
|
||||
t.Run("MissCache", testShouldCache)
|
||||
t.Run("TCP", testTCP)
|
||||
t.Run("HTTP", testHTTP)
|
||||
t.Run("ManualScan", testManualScanner)
|
||||
|
||||
@ -16,7 +16,6 @@ package stream
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@ -216,7 +215,6 @@ func (rc *RemoteCache) Init(client *ExtentClient) (err error) {
|
||||
}
|
||||
|
||||
log.LogDebugf("RemoteCache: Init")
|
||||
fmt.Println("RemoteCache: Init")
|
||||
rc.cluster = client.dataWrapper.ClusterName
|
||||
rc.volname = client.extentConfig.Volume
|
||||
rc.metaWrapper = client.metaWrapper
|
||||
|
||||
@ -3,6 +3,7 @@ package remotecache
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"hash/crc32"
|
||||
"io"
|
||||
@ -17,6 +18,7 @@ import (
|
||||
"github.com/cubefs/cubefs/util"
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/btree"
|
||||
"github.com/cubefs/cubefs/util/bytespool"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
"github.com/cubefs/cubefs/util/exporter"
|
||||
"github.com/cubefs/cubefs/util/iputil"
|
||||
@ -472,6 +474,112 @@ func (rc *RemoteCacheClient) Get(ctx context.Context, reqId, key string, from, t
|
||||
return nil, 0, false, nil
|
||||
}
|
||||
|
||||
func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Reader, length int64) (err error) {
|
||||
if length <= 0 {
|
||||
log.LogWarnf("put data length is less and equal 0")
|
||||
return nil
|
||||
}
|
||||
slot := proto.ComputeCacheBlockSlot(key, 0, 0)
|
||||
fg, _ := rc.GetFlashGroupBySlot(slot)
|
||||
if fg == nil {
|
||||
err = fmt.Errorf("FlashGroup put failed: cannot find any flashGroups")
|
||||
return
|
||||
}
|
||||
addr := fg.getFlashHost()
|
||||
if addr == "" {
|
||||
err = fmt.Errorf("getFlashHost failed: can not find host")
|
||||
log.LogWarnf("FlashGroup reqId(%v) put failed: err(%v)", reqId, err)
|
||||
return
|
||||
}
|
||||
var conn *net.TCPConn
|
||||
if conn, err = rc.conns.GetConnect(addr); err != nil {
|
||||
log.LogWarnf("FlashGroup put: reqId(%v) get connection to curr addr failed, addr(%v) key(%v) err(%v)", reqId, addr, key, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
rc.conns.PutConnect(conn, err != nil)
|
||||
}()
|
||||
req := &proto.PutBlockHead{
|
||||
UniKey: key,
|
||||
BlockLen: length,
|
||||
TTL: uint64(rc.TTL),
|
||||
}
|
||||
reqPacket := proto.NewPacketReqID()
|
||||
_ = reqPacket.MarshalDataPb(req)
|
||||
reqPacket.Opcode = proto.OpFlashNodeCachePutBlock
|
||||
if err = reqPacket.WriteToConn(conn); err != nil {
|
||||
log.LogWarnf("FlashGroup put: reqId(%v) failed to write header to addr(%v) err(%v)", reqId, addr, err)
|
||||
return
|
||||
}
|
||||
replyPacket := NewFlashCacheReply()
|
||||
if err = replyPacket.ReadFromConnExt(conn, int(rc.ReadTimeout)); err != nil {
|
||||
log.LogWarnf("FlashGroup put: reqId(%v) failed to ReadFromConn, replyPacket(%v), fg host(%v) , err(%v)", reqId, replyPacket, addr, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if err != nil {
|
||||
rc.deleteRemoteBlock(key, conn)
|
||||
}
|
||||
}()
|
||||
var (
|
||||
totalWritten int64
|
||||
n int
|
||||
)
|
||||
buf := bytespool.Alloc(proto.PageSize)
|
||||
defer bytespool.Free(buf)
|
||||
crcBuf := bytespool.Alloc(4)
|
||||
defer bytespool.Free(crcBuf)
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
readSize := proto.PageSize
|
||||
if totalWritten+int64(readSize) > length {
|
||||
readSize = int(length - totalWritten)
|
||||
}
|
||||
n, err = r.Read(buf[:readSize])
|
||||
if n != readSize {
|
||||
return fmt.Errorf("expected to read %d bytes, but only read %d", readSize, n)
|
||||
}
|
||||
if n > 0 {
|
||||
conn.SetWriteDeadline(time.Now().Add(proto.WriteDeadlineTime * time.Second))
|
||||
if _, err = conn.Write(buf[:n]); err != nil {
|
||||
log.LogErrorf("wirte data to flashnode get err %v", err)
|
||||
return err
|
||||
}
|
||||
binary.BigEndian.PutUint32(crcBuf, crc32.ChecksumIEEE(buf[:n]))
|
||||
if _, err = conn.Write(crcBuf[:4]); err != nil {
|
||||
log.LogErrorf("wirte crc to flashnode get err %v", err)
|
||||
return err
|
||||
}
|
||||
totalWritten += int64(n)
|
||||
}
|
||||
if err == io.EOF {
|
||||
err = nil
|
||||
break
|
||||
} else if err != nil {
|
||||
return err
|
||||
} else if totalWritten == length {
|
||||
break
|
||||
}
|
||||
}
|
||||
if totalWritten != length {
|
||||
return fmt.Errorf("unexpected data length: expected %d bytes, got %d", length, totalWritten)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rc *RemoteCacheClient) deleteRemoteBlock(key string, conn *net.TCPConn) {
|
||||
p := proto.NewPacketReqID()
|
||||
p.Data = ([]byte)(key)
|
||||
p.Opcode = proto.OpFlashNodeCacheDelete
|
||||
p.Size = uint32(len(p.Data))
|
||||
if err := p.WriteToConn(conn); err != nil {
|
||||
log.LogWarnf("FlashGroup put: failed to write to addr(%v) err(%v)", conn.RemoteAddr().String(), err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
type CacheReadRequest struct {
|
||||
proto.CacheReadRequest
|
||||
Data []byte
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
