From 97612b6b20b03eef2bd59c4640d72fbb643b7804 Mon Sep 17 00:00:00 2001 From: baihailong Date: Thu, 27 Mar 2025 15:57:12 +0800 Subject: [PATCH] feat(flashnode): enable statistic slot hit rate.#1000037792 Signed-off-by: baihailong --- cli/cmd/flashnode.go | 26 ++++++++ cli/cmd/fmt.go | 18 ++++++ flashnode/flashnode.go | 67 ++++++++++++++++++- flashnode/flashnode_op.go | 7 +- flashnode/serve_http.go | 9 +++ proto/distributed_cache.go | 14 ++++ proto/distributed_cache.pb.go | 90 ++++++++++++++++++-------- proto/distributed_cache.proto | 1 + sdk/data/stream/extent_flash_cache.go | 10 +-- sdk/data/stream/stream_remote_cache.go | 11 ++-- sdk/httpclient/flashnode.go | 6 ++ 11 files changed, 221 insertions(+), 38 deletions(-) diff --git a/cli/cmd/flashnode.go b/cli/cmd/flashnode.go index af316dd6f..dca1055b3 100644 --- a/cli/cmd/flashnode.go +++ b/cli/cmd/flashnode.go @@ -44,6 +44,7 @@ func newFlashNodeCmd(client *master.MasterClient) *cobra.Command { newCmdFlashNodeHTTPStatAll(client), newCmdFlashNodeHTTPEvict(client), newCmdFlashNodeHTTPInactiveDisk(client), + newCmdFlashNodeHTTPSlotStat(client), ) return cmd } @@ -218,6 +219,31 @@ func newCmdFlashNodeHTTPStatAll(client *master.MasterClient) *cobra.Command { } } +func newCmdFlashNodeHTTPSlotStat(client *master.MasterClient) *cobra.Command { + return &cobra.Command{ + Use: "httpSlotStat" + _flashnodeAddr, + Short: "show flashnode slot stat", + Args: cobra.MinimumNArgs(1), + RunE: func(_ *cobra.Command, args []string) (err error) { + // check flashnode whether exist + _, err = client.NodeAPI().GetFlashNode(args[0]) + if err != nil { + return + } + stat, err := httpclient.New().Addr(addr2Prof(args[0])).FlashNode().SlotStat() + if err != nil { + return + } + + sort.SliceStable(stat.SlotStat, func(i, j int) bool { + return stat.SlotStat[i].SlotId < stat.SlotStat[j].SlotId + }) + stdout("%v\n", formatFlashNodeSlotStat(&stat)) + return + }, + } +} + func newCmdFlashNodeHTTPEvict(client *master.MasterClient) *cobra.Command { return &cobra.Command{ Use: "httpEvict" + _flashnodeAddr + " [volume]", diff --git a/cli/cmd/fmt.go b/cli/cmd/fmt.go index 8e45f666f..17b0efe57 100644 --- a/cli/cmd/fmt.go +++ b/cli/cmd/fmt.go @@ -1325,3 +1325,21 @@ func formatFlashGroupView(fg *proto.FlashGroupAdminView) string { fmt.Sprintf(" Step:%v\n", fg.Step) + fmt.Sprintf(" FlashNodeCount:%v\n", fg.FlashNodeCount) } + +var ( + flashnodeSlotStatTablePattern = "%-12v %-12v %-12v %-12v %-12v\n" + flashnodeSlotStatTableHeader = fmt.Sprintf(flashnodeSlotStatTablePattern, + "SlotId", "OwnerSlotId", "HitCount", "HitRate", "RecentTime") +) + +func formatFlashNodeSlotStat(stat *proto.FlashNodeSlotStat) string { + sb := strings.Builder{} + sb.WriteString(flashnodeSlotStatTableHeader) + for _, slot := range stat.SlotStat { + hitrate := fmt.Sprintf("%.2f%%", slot.HitRate*100) + recentTime := formatTimeToString(slot.RecentTime) + sb.WriteString(fmt.Sprintf(flashnodeSlotStatTablePattern, slot.SlotId, slot.OwnerSlotId, slot.HitCount, hitrate, recentTime)) + } + + return sb.String() +} diff --git a/flashnode/flashnode.go b/flashnode/flashnode.go index 124d5e7ae..6631c28db 100644 --- a/flashnode/flashnode.go +++ b/flashnode/flashnode.go @@ -21,6 +21,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "syscall" "time" @@ -71,6 +72,7 @@ const ( _defaultManualScanLimitPerSecond = 10000 _defaultPrepareLimitPerSecond = 1000 _defaultManualScanLimitBurst = 1000 + _slotStatValidPeriod = 10 * time.Minute // min ) // Configuration keys @@ -161,8 +163,11 @@ type FlashNode struct { manualScanLimitPerSecond int64 prepareLimitPerSecond int64 scannerMutex sync.RWMutex - manualScanners sync.Map //[string]*ManualScanner + manualScanners sync.Map // [string]*ManualScanner waitForCacheBlock bool + + slotMap sync.Map // [uint32]*SlotStat + readCount uint64 } // Start starts up the flash node with the specified configuration. @@ -230,6 +235,7 @@ func (f *FlashNode) start(cfg *config.Config) (err error) { if err != nil { return } + f.startSlotStat() return nil } @@ -544,3 +550,62 @@ func (f *FlashNode) respondToMaster(task *proto.AdminTask) { } }() } + +func (f *FlashNode) startSlotStat() { + log.LogInfof("startSlotStat") + go func() { + tick := time.NewTicker(time.Second * 60) + defer tick.Stop() + for { + f.replaceSlotStat() + select { + case <-tick.C: + case <-f.stopCh: + log.LogInfof("exit slotStat") + return + } + } + }() +} + +func (f *FlashNode) replaceSlotStat() { + readCount := atomic.SwapUint64(&f.readCount, 0) + f.slotMap.Range(func(_, value interface{}) bool { + slotStat := value.(*proto.SlotStat) + if slotStat.RecentTime.Before(time.Now().Add(-_slotStatValidPeriod)) { + f.slotMap.Delete(slotStat.SlotId) + } else { + hitCount := atomic.SwapUint32(&slotStat.HitCount, 0) + if readCount == 0 { + slotStat.HitRate = 0 + } else { + slotStat.HitRate = float64(hitCount) / float64(readCount) + } + } + return true + }) +} + +func (f *FlashNode) updateSlotStat(reqSlot uint64) { + atomic.AddUint64(&f.readCount, 1) + slotId := uint32((reqSlot >> 32) & 0xFFFFFFFF) + ownerSlotId := uint32(reqSlot & 0xFFFFFFFF) + if value, ok := f.slotMap.Load(slotId); ok { + slotStat := value.(*proto.SlotStat) + atomic.AddUint32(&slotStat.HitCount, 1) + slotStat.RecentTime = time.Now() + } else { + slotStat := &proto.SlotStat{SlotId: slotId, OwnerSlotId: ownerSlotId, HitCount: 1, RecentTime: time.Now()} + f.slotMap.Store(slotId, slotStat) + } +} + +func (f *FlashNode) GetFlashNodeSlotStat() []*proto.SlotStat { + slotStats := make([]*proto.SlotStat, 0) + f.slotMap.Range(func(_, value interface{}) bool { + slotStat := value.(*proto.SlotStat) + slotStats = append(slotStats, slotStat) + return true + }) + return slotStats +} diff --git a/flashnode/flashnode_op.go b/flashnode/flashnode_op.go index dc3c8e496..1c15cf265 100644 --- a/flashnode/flashnode_op.go +++ b/flashnode/flashnode_op.go @@ -170,9 +170,12 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) { err = fmt.Errorf("no cache read request") return } - volume = req.CacheRequest.Volume + volume = req.CacheRequest.Volume cr := req.CacheRequest + + f.updateSlotStat(cr.Slot) + block, err := f.cacheEngine.GetCacheBlockForRead(volume, cr.Inode, cr.FixedFileOffset, cr.Version, req.Size_) if err != nil { hitRateMap := f.cacheEngine.GetHitRate() @@ -479,6 +482,8 @@ func (f *FlashNode) opCachePrepare(conn net.Conn, p *proto.Packet) (err error) { err = fmt.Errorf("no cache prepare request") return } + + f.updateSlotStat(req.CacheRequest.Slot) volume = req.CacheRequest.Volume if err = f.cacheEngine.PrepareCache(p.ReqID, req.CacheRequest, conn.RemoteAddr().String()); err != nil { diff --git a/flashnode/serve_http.go b/flashnode/serve_http.go index c23d215a0..d20791e3a 100644 --- a/flashnode/serve_http.go +++ b/flashnode/serve_http.go @@ -35,6 +35,7 @@ func (f *FlashNode) registerAPIHandler() { http.HandleFunc("/getDiskQos", f.handleGetDiskQos) http.HandleFunc("/scannerControl", f.handleScannerCommand) http.HandleFunc("/setWaitForCacheBlock", f.handleSetWaitForCacheBlock) + http.HandleFunc("/slotStat", f.handleSlotStat) } func (f *FlashNode) handleStat(w http.ResponseWriter, r *http.Request) { @@ -235,3 +236,11 @@ func (f *FlashNode) handleSetWaitForCacheBlock(w http.ResponseWriter, r *http.Re f.waitForCacheBlock = val replyOK(w, r, nil) } + +func (f *FlashNode) handleSlotStat(w http.ResponseWriter, r *http.Request) { + replyOK(w, r, proto.FlashNodeSlotStat{ + NodeId: f.nodeID, + Addr: f.localAddr, + SlotStat: f.GetFlashNodeSlotStat(), + }) +} diff --git a/proto/distributed_cache.go b/proto/distributed_cache.go index 298b406ad..a1818a95c 100644 --- a/proto/distributed_cache.go +++ b/proto/distributed_cache.go @@ -298,6 +298,20 @@ type CacheStatus struct { Status int `json:"status"` } +type SlotStat struct { + SlotId uint32 `json:"slot_id"` + OwnerSlotId uint32 `json:"owner_slot_id"` + HitCount uint32 `json:"hit_count"` + HitRate float64 `json:"hit_rate"` + RecentTime time.Time `json:"recent_time"` +} + +type FlashNodeSlotStat struct { + NodeId uint64 + Addr string + SlotStat []*SlotStat +} + func ComputeSourcesVersion(sources []*DataSource, gen uint64) (version uint32) { if len(sources) == 0 { return 0 diff --git a/proto/distributed_cache.pb.go b/proto/distributed_cache.pb.go index 1ab12de24..c9cccadd0 100644 --- a/proto/distributed_cache.pb.go +++ b/proto/distributed_cache.pb.go @@ -116,6 +116,7 @@ type CacheRequest struct { Version uint32 `protobuf:"varint,4,opt,name=Version,proto3" json:"Version,omitempty"` Sources []*DataSource `protobuf:"bytes,5,rep,name=Sources,proto3" json:"Sources,omitempty"` TTL int64 `protobuf:"varint,6,opt,name=TTL,proto3" json:"TTL,omitempty"` + Slot uint64 `protobuf:"varint,7,opt,name=Slot,proto3" json:"Slot,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` @@ -195,6 +196,13 @@ func (m *CacheRequest) GetTTL() int64 { return 0 } +func (m *CacheRequest) GetSlot() uint64 { + if m != nil { + return m.Slot + } + return 0 +} + type CacheReadRequest struct { CacheRequest *CacheRequest `protobuf:"bytes,1,opt,name=CacheRequest,proto3" json:"CacheRequest,omitempty"` Offset uint64 `protobuf:"varint,2,opt,name=Offset,proto3" json:"Offset,omitempty"` @@ -321,33 +329,34 @@ func init() { func init() { proto.RegisterFile("distributed_cache.proto", fileDescriptor_9a29821e5022ef34) } var fileDescriptor_9a29821e5022ef34 = []byte{ - // 412 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x52, 0x4d, 0x6b, 0x14, 0x41, - 0x14, 0x4c, 0x67, 0x3e, 0x62, 0xde, 0x46, 0x5c, 0x3b, 0xa2, 0x4d, 0x0e, 0xe3, 0xb0, 0xa7, 0x01, - 0x71, 0x03, 0xf1, 0x20, 0x78, 0xd4, 0x35, 0xb8, 0x20, 0x1a, 0x3a, 0x21, 0x57, 0x99, 0x8f, 0xb7, - 0xbb, 0x0d, 0x9b, 0xe9, 0x38, 0xdd, 0x03, 0x41, 0xf0, 0x7f, 0xe4, 0xe8, 0x2f, 0xf1, 0x26, 0x78, - 0xf4, 0x27, 0xc8, 0xfa, 0x47, 0x64, 0x5e, 0xf7, 0x9a, 0xc9, 0x7a, 0xcc, 0x69, 0x5e, 0xd5, 0xeb, - 0xa9, 0xae, 0xaa, 0x19, 0x78, 0x52, 0x29, 0x63, 0x1b, 0x55, 0xb4, 0x16, 0xab, 0x4f, 0x65, 0x5e, - 0x2e, 0x70, 0x7c, 0xd9, 0x68, 0xab, 0x79, 0x44, 0x8f, 0x83, 0xe7, 0x73, 0x65, 0x17, 0x6d, 0x31, - 0x2e, 0xf5, 0xc5, 0xe1, 0x5c, 0xcf, 0xf5, 0x21, 0xd1, 0x45, 0x3b, 0x23, 0x44, 0x80, 0x26, 0xf7, - 0xd6, 0xe8, 0x3b, 0x03, 0x98, 0xe4, 0x36, 0x3f, 0xd5, 0x6d, 0x53, 0x22, 0x4f, 0x00, 0x8e, 0xd5, - 0x12, 0x3f, 0xce, 0x66, 0x06, 0xad, 0x60, 0x29, 0xcb, 0x42, 0xd9, 0x63, 0x78, 0x0a, 0x83, 0x93, - 0xbc, 0xb1, 0xca, 0x2a, 0x5d, 0x4f, 0x27, 0x62, 0x9b, 0x0e, 0xf4, 0x29, 0x7e, 0x00, 0xf7, 0xde, - 0x5e, 0x59, 0xac, 0xed, 0x74, 0x22, 0x02, 0x5a, 0xff, 0xc3, 0x7c, 0x04, 0x7b, 0x6e, 0xf6, 0xfa, - 0x21, 0xed, 0x6f, 0x71, 0x9c, 0x43, 0x78, 0xaa, 0xbe, 0xa0, 0x88, 0x68, 0x47, 0x33, 0x7f, 0x04, - 0xd1, 0x3b, 0x6d, 0xac, 0x11, 0x71, 0x1a, 0x64, 0xbb, 0xd2, 0x81, 0x57, 0xe1, 0xf5, 0xb7, 0xa7, - 0x5b, 0xa3, 0x1f, 0x0c, 0xf6, 0xde, 0x74, 0x35, 0x48, 0xfc, 0xdc, 0xa2, 0xb1, 0xfc, 0x31, 0xc4, - 0xe7, 0x7a, 0xd9, 0x5e, 0x20, 0xd9, 0xdf, 0x95, 0x1e, 0x75, 0x22, 0xd3, 0x5a, 0x57, 0xe8, 0x4d, - 0x3b, 0xc0, 0x33, 0x78, 0x70, 0xac, 0xae, 0xb0, 0xea, 0xa5, 0x76, 0xae, 0x37, 0x69, 0x2e, 0x60, - 0xe7, 0x1c, 0x1b, 0xa3, 0x74, 0x4d, 0xbe, 0xef, 0xcb, 0x35, 0xe4, 0xcf, 0x60, 0xc7, 0xd5, 0x67, - 0x44, 0x94, 0x06, 0xd9, 0xe0, 0xe8, 0xa1, 0x2b, 0x77, 0x7c, 0x53, 0xac, 0x5c, 0x9f, 0xe0, 0x43, - 0x08, 0xce, 0xce, 0xde, 0x8b, 0x38, 0x65, 0x59, 0x20, 0xbb, 0xd1, 0xe7, 0xf8, 0x0a, 0x43, 0x1f, - 0x23, 0xaf, 0xd6, 0x51, 0x5e, 0xde, 0x8e, 0x46, 0x81, 0x06, 0x47, 0xfb, 0x5e, 0xbd, 0xbf, 0x92, - 0xff, 0x75, 0xe0, 0xc3, 0xb8, 0xb0, 0xf1, 0x46, 0xb9, 0xc1, 0x4d, 0xb9, 0xfe, 0x7a, 0x0b, 0xfb, - 0xa4, 0x70, 0xd2, 0xe0, 0x65, 0xde, 0xe0, 0x9d, 0x1d, 0x74, 0x3f, 0xd2, 0x32, 0x37, 0x8b, 0x0f, - 0xba, 0x42, 0x23, 0xb6, 0xe9, 0xbb, 0xf5, 0x18, 0x77, 0xeb, 0xeb, 0xe1, 0xcf, 0x55, 0xc2, 0x7e, - 0xad, 0x12, 0xf6, 0x7b, 0x95, 0xb0, 0xeb, 0x3f, 0xc9, 0x56, 0x11, 0x93, 0xf2, 0x8b, 0xbf, 0x01, - 0x00, 0x00, 0xff, 0xff, 0x8a, 0x2d, 0x06, 0xa1, 0xe7, 0x02, 0x00, 0x00, + // 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, } func (m *DataSource) Marshal() (dAtA []byte, err error) { @@ -435,6 +444,11 @@ func (m *CacheRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) { i -= len(m.XXX_unrecognized) copy(dAtA[i:], m.XXX_unrecognized) } + if m.Slot != 0 { + i = encodeVarintDistributedCache(dAtA, i, uint64(m.Slot)) + i-- + dAtA[i] = 0x38 + } if m.TTL != 0 { i = encodeVarintDistributedCache(dAtA, i, uint64(m.TTL)) i-- @@ -648,6 +662,9 @@ func (m *CacheRequest) Size() (n int) { if m.TTL != 0 { n += 1 + sovDistributedCache(uint64(m.TTL)) } + if m.Slot != 0 { + n += 1 + sovDistributedCache(uint64(m.Slot)) + } if m.XXX_unrecognized != nil { n += len(m.XXX_unrecognized) } @@ -1053,6 +1070,25 @@ func (m *CacheRequest) Unmarshal(dAtA []byte) error { break } } + case 7: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Slot", wireType) + } + m.Slot = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowDistributedCache + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Slot |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := skipDistributedCache(dAtA[iNdEx:]) diff --git a/proto/distributed_cache.proto b/proto/distributed_cache.proto index 2975e3f5e..9f2764207 100644 --- a/proto/distributed_cache.proto +++ b/proto/distributed_cache.proto @@ -24,6 +24,7 @@ message CacheRequest { uint32 Version = 4; repeated DataSource Sources = 5; int64 TTL = 6; + uint64 Slot = 7; } message CacheReadRequest { diff --git a/sdk/data/stream/extent_flash_cache.go b/sdk/data/stream/extent_flash_cache.go index e9dc69d18..650235556 100644 --- a/sdk/data/stream/extent_flash_cache.go +++ b/sdk/data/stream/extent_flash_cache.go @@ -578,7 +578,7 @@ func (rc *RemoteCache) updateHostLatency(hosts []string) { } } -func (rc *RemoteCache) GetFlashGroupBySlot(slot uint32) *FlashGroup { +func (rc *RemoteCache) GetFlashGroupBySlot(slot uint32) (*FlashGroup, uint32) { var item *SlotItem pivot := &SlotItem{slot: slot} @@ -591,7 +591,7 @@ func (rc *RemoteCache) GetFlashGroupBySlot(slot uint32) *FlashGroup { if item == nil { return rc.getMinFlashGroup() } - return item.FlashGroup + return item.FlashGroup, item.slot } func (rc *RemoteCache) getFlashHostsMap() map[string]bool { @@ -624,14 +624,14 @@ func (rc *RemoteCache) rangeFlashGroups(pivot *SlotItem, rangeFunc func(item btr } } -func (rc *RemoteCache) getMinFlashGroup() *FlashGroup { +func (rc *RemoteCache) getMinFlashGroup() (*FlashGroup, uint32) { flashGroups := rc.flashGroups if flashGroups.Len() > 0 { item := flashGroups.Min().(*SlotItem) if item != nil { - return item.FlashGroup + return item.FlashGroup, item.slot } } - return nil + return nil, 0 } diff --git a/sdk/data/stream/stream_remote_cache.go b/sdk/data/stream/stream_remote_cache.go index b03329560..1ed0afc54 100644 --- a/sdk/data/stream/stream_remote_cache.go +++ b/sdk/data/stream/stream_remote_cache.go @@ -81,12 +81,13 @@ func (s *Streamer) prepareRemoteCache(ctx context.Context, ek *proto.ExtentKey, } for _, req := range cReadRequests { - fg := s.getFlashGroup(req.CacheRequest.FixedFileOffset) + slot, fg, ownerSlot := s.getFlashGroup(req.CacheRequest.FixedFileOffset) if fg == nil { err = fmt.Errorf("cannot find any flashGroups") log.LogWarnf("Streamer prepareRemoteCache failed: %v", err) break } + req.CacheRequest.Slot = uint64(slot)<<32 | uint64(ownerSlot) prepareReq := &proto.CachePrepareRequest{ CacheRequest: req.CacheRequest, FlashNodes: fg.Hosts, @@ -113,11 +114,12 @@ func (s *Streamer) readFromRemoteCache(ctx context.Context, offset, size uint64, total += int(req.Size_) continue } - fg := s.getFlashGroup(req.CacheRequest.FixedFileOffset) + slot, fg, ownerSlot := s.getFlashGroup(req.CacheRequest.FixedFileOffset) if fg == nil { err = fmt.Errorf("readFromRemoteCache failed: cannot find any flashGroups") return } + req.CacheRequest.Slot = uint64(slot)<<32 | uint64(ownerSlot) if read, err = s.client.RemoteCache.Read(ctx, fg, s.inode, req); err != nil { if !proto.IsFlashNodeLimitError(err) { log.LogWarnf("readFromRemoteCache: flashGroup read failed. offset(%v) size(%v) fg(%v) req(%v) err(%v)", offset, size, fg, req, err) @@ -131,9 +133,10 @@ func (s *Streamer) readFromRemoteCache(ctx context.Context, offset, size uint64, return total, nil } -func (s *Streamer) getFlashGroup(fixedFileOffset uint64) *FlashGroup { +func (s *Streamer) getFlashGroup(fixedFileOffset uint64) (uint32, *FlashGroup, uint32) { slot := proto.ComputeCacheBlockSlot(s.client.dataWrapper.VolName, s.inode, fixedFileOffset) - return s.client.RemoteCache.GetFlashGroupBySlot(slot) + fg, ownerSlot := s.client.RemoteCache.GetFlashGroupBySlot(slot) + return slot, fg, ownerSlot } func (s *Streamer) getDataSource(start, size, fixedFileOffset uint64, isRead bool) ([]*proto.DataSource, error) { diff --git a/sdk/httpclient/flashnode.go b/sdk/httpclient/flashnode.go index 2df4ee4b6..ac67b9cdf 100644 --- a/sdk/httpclient/flashnode.go +++ b/sdk/httpclient/flashnode.go @@ -24,6 +24,7 @@ type FlashNode interface { Stat() (proto.FlashNodeStat, error) StatAll() (proto.FlashNodeStat, error) InactiveDisk(dataPath string) error + SlotStat() (proto.FlashNodeSlotStat, error) } type flashNode struct { @@ -59,3 +60,8 @@ func (f *flashNode) InactiveDisk(dataPath string) error { r.params.Add("dataPath", dataPath) return f.client.serveWith(nil, r) } + +func (f *flashNode) SlotStat() (st proto.FlashNodeSlotStat, err error) { + err = f.client.serveWith(&st, newRequest(get, "/slotStat")) + return +}