diff --git a/proto/distributed_cache.go b/proto/distributed_cache.go index e3c4ebe1d..050bdb236 100644 --- a/proto/distributed_cache.go +++ b/proto/distributed_cache.go @@ -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 +} diff --git a/proto/distributed_cache.pb.go b/proto/distributed_cache.pb.go index da692132a..6c09494a5 100644 --- a/proto/distributed_cache.pb.go +++ b/proto/distributed_cache.pb.go @@ -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 diff --git a/proto/distributed_cache.proto b/proto/distributed_cache.proto index 9f2764207..8790233a0 100644 --- a/proto/distributed_cache.proto +++ b/proto/distributed_cache.proto @@ -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; +} \ No newline at end of file diff --git a/proto/packet.go b/proto/packet.go index 15427198a..ff5d68fab 100644 --- a/proto/packet.go +++ b/proto/packet.go @@ -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 diff --git a/remotecache/flashnode/cachengine/block.go b/remotecache/flashnode/cachengine/block.go index a95710f64..ca9230337 100644 --- a/remotecache/flashnode/cachengine/block.go +++ b/remotecache/flashnode/cachengine/block.go @@ -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 +} diff --git a/remotecache/flashnode/cachengine/block_test.go b/remotecache/flashnode/cachengine/block_test.go index 40c67d711..6ccf40551 100644 --- a/remotecache/flashnode/cachengine/block_test.go +++ b/remotecache/flashnode/cachengine/block_test.go @@ -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() +} diff --git a/remotecache/flashnode/cachengine/engine.go b/remotecache/flashnode/cachengine/engine.go index 5d6299cdc..0eb0bffdf 100644 --- a/remotecache/flashnode/cachengine/engine.go +++ b/remotecache/flashnode/cachengine/engine.go @@ -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) == "" } diff --git a/remotecache/flashnode/cachengine/engine_test.go b/remotecache/flashnode/cachengine/engine_test.go index cf9f92691..4cb7abd74 100644 --- a/remotecache/flashnode/cachengine/engine_test.go +++ b/remotecache/flashnode/cachengine/engine_test.go @@ -31,7 +31,8 @@ import ( ) const ( - testTmpFS = "/cfs_test/tmpfs" + testTmpFS = "/cfs_test/tmpfs" + testTmpFS1 = "/cfs_test/tmpfsblocks" ) var bytesCommon = randTestData(1024) diff --git a/remotecache/flashnode/cachengine/miss_cache.go b/remotecache/flashnode/cachengine/miss_cache.go new file mode 100644 index 000000000..18658ada3 --- /dev/null +++ b/remotecache/flashnode/cachengine/miss_cache.go @@ -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++ + } +} diff --git a/remotecache/flashnode/flashnode.go b/remotecache/flashnode/flashnode.go index 79437a382..c411c45e6 100644 --- a/remotecache/flashnode/flashnode.go +++ b/remotecache/flashnode/flashnode.go @@ -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 diff --git a/remotecache/flashnode/flashnode_op.go b/remotecache/flashnode/flashnode_op.go index c2f887f85..8f44706c6 100644 --- a/remotecache/flashnode/flashnode_op.go +++ b/remotecache/flashnode/flashnode_op.go @@ -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 { diff --git a/remotecache/flashnode/flashnode_op_test.go b/remotecache/flashnode/flashnode_op_test.go index 58a36c5d8..c81a5b350 100644 --- a/remotecache/flashnode/flashnode_op_test.go +++ b/remotecache/flashnode/flashnode_op_test.go @@ -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)) +} diff --git a/remotecache/flashnode/flashnode_test.go b/remotecache/flashnode/flashnode_test.go index 673857907..cc80b766e 100644 --- a/remotecache/flashnode/flashnode_test.go +++ b/remotecache/flashnode/flashnode_test.go @@ -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) diff --git a/sdk/data/stream/extent_flash_cache.go b/sdk/data/stream/extent_flash_cache.go index 517344afb..7dd0b74e4 100644 --- a/sdk/data/stream/extent_flash_cache.go +++ b/sdk/data/stream/extent_flash_cache.go @@ -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 diff --git a/sdk/remotecache/client.go b/sdk/remotecache/client.go index 3339473ed..620c36276 100755 --- a/sdk/remotecache/client.go +++ b/sdk/remotecache/client.go @@ -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