From 184b6a15355e93390c49698dc380c453addd69aa Mon Sep 17 00:00:00 2001 From: clinx Date: Fri, 1 Aug 2025 16:15:40 +0800 Subject: [PATCH] feat(tool): remove specific key #1000250043 Signed-off-by: clinx (cherry picked from commit 6fe627bd870d588b9502939fc497aa35c53ae04a) --- proto/distributed_cache.go | 36 +++++++- remotecache/flashnode/cachengine/block.go | 19 ++-- remotecache/flashnode/cachengine/engine.go | 2 +- remotecache/flashnode/flashnode_op.go | 38 ++++---- sdk/remotecache/client.go | 90 ++++++++++++------- tool/remotecache-benchmark/main.go | 22 ++++- tool/remotecache-benchmark/storage/localfs.go | 5 ++ tool/remotecache-benchmark/storage/storage.go | 1 + 8 files changed, 151 insertions(+), 62 deletions(-) diff --git a/proto/distributed_cache.go b/proto/distributed_cache.go index 13b24cb66..603cff7ba 100644 --- a/proto/distributed_cache.go +++ b/proto/distributed_cache.go @@ -40,8 +40,40 @@ const ( ) var ( - ErrorNotExistShouldCache = fmt.Errorf("cache miss: should store and cache the key") - ErrorNotExistShouldNotCache = fmt.Errorf("only cache the miss after reaching the miss times") + ErrorNotExistShouldCache = fmt.Errorf("cache miss: should store and cache the key") + ErrorNotExistShouldNotCache = fmt.Errorf("only cache the miss after reaching the miss times") + ErrorUnableToBuildKeyFromPacket = fmt.Errorf("unable to build a key from the packet due to lack of available parameters") + ErrorParsingCacheWriteHead = fmt.Errorf("error parsing cache write head structure") + ErrorNoCacheDeleteRequest = fmt.Errorf("no cache delete request") + ErrorNoCacheReadRequest = fmt.Errorf("no cache read request") + ErrorNoCachePrepareRequest = fmt.Errorf("no cache prepare request") + ErrorDeleteBlockKeyNil = fmt.Errorf("delete block key is nil") + ErrorRequireDataIsCaching = fmt.Errorf("require data is caching") + ErrorReadTimeout = fmt.Errorf("read timeout") + ErrorNoAvailableHost = fmt.Errorf("no available host") + ErrorContextDeadLine = fmt.Errorf("context deadline exceeded") + ErrorRequestOutOfRange = fmt.Errorf("the requested read size is out of range") + ErrorUnableGetCreatedBlock = fmt.Errorf("unable to get created cacheblock") + ErrorFlashGroupDeleteFailed = fmt.Errorf("FlashGroup delete failed: cannot find any flashGroups") + ErrorFlashGroupPutFailed = fmt.Errorf("FlashGroup put failed: cannot find any flashGroups") + ErrorReadObjectFromRemoteCacheFailed = fmt.Errorf("readObjectFromRemoteCache failed: cannot find any flashGroups") +) + +var ( + ErrorResultCodeNOKTpl = "ResultCode NOK (%v)" + ErrorUnknownOpcodeTpl = "unknown Opcode:%d" + ErrorTaskIDNotExistTpl = "task id(%v) not exist" + ErrorBlockAlreadyExistsTpl = "block %v already exist" + ErrorCreateBlockFailedTpl = "create block(%v) error %v" + ErrorBlockAlreadyCreatedTpl = "block(%v) already created" + ErrorPutDataLengthInvalidTpl = "put data length %v is leq 0 or gt 4M" + ErrorInconsistentCRCTpl = "inconsistent CRC, expect(%v) reply(%v)" + ErrorExpectedReadBytesMismatchTpl = "expected to read %d bytes, but only read %d" + ErrorUnexpectedDataLengthTpl = "unexpected data length: expected %d bytes, got %d" + ErrorWriteDataAndCRCToFlashNodeTpl = "write data and crc to flashNode get err %v" + ErrorReadFromCloseReaderTpl = "read from close reader reqID(%v)" + ErrorInconsistentCRCObjectTpl = "inconsistent CRC, offset(%v) extentOffset(%v) expect(%v) actualCrc(%v)" + ErrorInvalidRangeTpl = "invalid range: from(%v) to(%v)" ) type FlashGroupStatus int diff --git a/remotecache/flashnode/cachengine/block.go b/remotecache/flashnode/cachengine/block.go index dcc01f7e7..382749f85 100644 --- a/remotecache/flashnode/cachengine/block.go +++ b/remotecache/flashnode/cachengine/block.go @@ -96,6 +96,10 @@ func (cb *CacheBlock) String() string { return fmt.Sprintf("volume(%s) inode(%d) offset(%d) version(%d)", cb.volume, cb.inode, cb.fixedOffset, cb.version) } +func (cb *CacheBlock) GetBlockKey() string { + return cb.blockKey +} + // Close this extent and release FD. func (cb *CacheBlock) Close() (err error) { cb.notifyClose() @@ -530,7 +534,7 @@ func (cb *CacheBlock) ready(ctx context.Context, waitForBlock bool) error { return ctx.Err() default: if !waitForBlock { - return fmt.Errorf("require data is caching") + return proto.ErrorRequireDataIsCaching } } } @@ -583,7 +587,9 @@ func (cb *CacheBlock) maybeUpdateUsedSize(size int64) { cb.sizeLock.Lock() defer cb.sizeLock.Unlock() if cb.usedSize < size { - log.LogDebugf("maybeUpdateUsedSize, cache block:%v, old:%v, new:%v", cb.blockKey, cb.usedSize, size) + if log.EnableDebug() { + log.LogDebugf("maybeUpdateUsedSize, cache block:%v, old:%v, new:%v", cb.blockKey, cb.usedSize, size) + } cb.usedSize = size } } @@ -747,7 +753,7 @@ func (c *CacheEngine) createCacheBlockV2(pDir string, uniKey string, ttl int64, } } } - return nil, fmt.Errorf("unable to get created cacheblock"), created + return nil, proto.ErrorUnableGetCreatedBlock, created } else { defer func() { close(ch) @@ -761,6 +767,7 @@ func (c *CacheEngine) createCacheBlockV2(pDir string, uniKey string, ttl int64, if atomic.LoadInt32(&cacheItem.disk.Status) == proto.ReadWrite { if blockValue, got := cacheItem.lruCache.Peek(key); got { block = blockValue.(*CacheBlock) + created = true return } } @@ -917,12 +924,10 @@ func (cb *CacheBlock) VerifyObjectReq(offset, size uint64) error { log.LogErrorf("invalid range offset(%v) size(%v)", offset, size) return fmt.Errorf("invalid range offset(%v) size(%v)", offset, size) } - if uint64(cb.usedSize) <= end { - log.LogWarnf("block is not read, usedSize(%v) offset(%v) size(%v)", cb.usedSize, offset, size) - return fmt.Errorf("block is not ready") + log.LogWarnf("the requested read size is out of range, usedSize(%v) offset(%v) size(%v)", cb.usedSize, offset, size) + return proto.ErrorRequestOutOfRange } - return nil } diff --git a/remotecache/flashnode/cachengine/engine.go b/remotecache/flashnode/cachengine/engine.go index 2e63f6c60..bd2fc232f 100644 --- a/remotecache/flashnode/cachengine/engine.go +++ b/remotecache/flashnode/cachengine/engine.go @@ -676,7 +676,7 @@ func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64, } } } - return nil, fmt.Errorf("unable to get created cacheblock") + return nil, proto.ErrorUnableGetCreatedBlock } else { defer func() { close(ch) diff --git a/remotecache/flashnode/flashnode_op.go b/remotecache/flashnode/flashnode_op.go index a35971639..988aebf54 100644 --- a/remotecache/flashnode/flashnode_op.go +++ b/remotecache/flashnode/flashnode_op.go @@ -75,7 +75,7 @@ func (f *FlashNode) handlePacket(conn net.Conn, p *proto.Packet) (err error) { case proto.OpFlashNodeTaskCommand: err = f.opFlashNodeTaskCommand(conn, p) default: - err = fmt.Errorf("unknown Opcode:%d", p.Opcode) + err = fmt.Errorf(proto.ErrorUnknownOpcodeTpl, p.Opcode) } return @@ -200,7 +200,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) { return } if req.CacheRequest == nil { - err = fmt.Errorf("no cache read request") + err = proto.ErrorNoCacheReadRequest return } @@ -240,7 +240,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) { } if !f.waitForCacheBlock { stat.EndStat("MissCacheRead:Data is caching", nil, bgTime2, 1) - return fmt.Errorf("require data is caching") + return proto.ErrorRequireDataIsCaching } select { case <-ctx.Done(): @@ -284,7 +284,7 @@ func (f *FlashNode) opCacheDelete(conn net.Conn, p *proto.Packet) (err error) { } }() if p.Size == 0 { - return fmt.Errorf("no cache delete request") + return proto.ErrorNoCacheDeleteRequest } uniKey := string(data) pDir := cachengine.MapKeyToDirectory(uniKey) @@ -310,11 +310,11 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error) } }() if len(p.Data) == 0 { - return fmt.Errorf("unable to build a key from the packet due to lack of available parameters") + return proto.ErrorUnableToBuildKeyFromPacket } req := new(proto.PutBlockHead) if err = p.UnmarshalDataPb(req); err != nil { - return fmt.Errorf("error parsing cache write head structure") + return proto.ErrorParsingCacheWriteHead } uniKey := req.UniKey pDir := cachengine.MapKeyToDirectory(uniKey) @@ -322,10 +322,10 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error) _, err1 := f.cacheEngine.GetCacheBlockForReadByKey(blockKey) if err1 == nil { if log.EnableDebug() { - log.LogDebug(logPrefix+" check block key:"+uniKey+" logMsg:%s", + log.LogDebug(logPrefix+" check block key:"+uniKey+" logMsg:", p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err)) } - err = fmt.Errorf("block %v already exsit", uniKey) + err = fmt.Errorf(proto.ErrorBlockAlreadyExistsTpl, uniKey) return err } var allocSize int @@ -334,14 +334,14 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error) } err1 = nil if log.EnableDebug() { - log.LogDebug(logPrefix+" create block key:"+uniKey+" logMsg:%s", + log.LogDebug(logPrefix+" create block key:"+uniKey+" logMsg:", p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err)) } if cb, err2, created := f.cacheEngine.CreateBlockV2(pDir, uniKey, req.TTL, uint32(allocSize), conn.RemoteAddr().String()); err2 != nil || created { if err2 != nil { - err = fmt.Errorf("create block(%v) error %v", cachengine.GenCacheBlockKeyV2(pDir, uniKey), err2.Error()) + err = fmt.Errorf(proto.ErrorCreateBlockFailedTpl, cachengine.GenCacheBlockKeyV2(pDir, uniKey), err2.Error()) } else { - err = fmt.Errorf("block(%v) already created", cachengine.GenCacheBlockKeyV2(pDir, uniKey)) + err = fmt.Errorf(proto.ErrorBlockAlreadyCreatedTpl, cachengine.GenCacheBlockKeyV2(pDir, uniKey)) } return } else { @@ -516,7 +516,7 @@ func (f *FlashNode) opCacheObjectGet(conn net.Conn, p *proto.Packet) (err error) bgTime2 := stat.BeginStat() // reply to client as quick as possible if hit cache - err = f.doObjectReadRequest(ctx, conn, req, p, block) + err = f.doObjectReadRequest(ctx, conn, req, p, block, reqID) if err != nil { stat.EndStat("HitCacheRead", err, bgTime2, 1) return @@ -578,7 +578,7 @@ func (f *FlashNode) opFlashNodeTaskCommand(conn net.Conn, p *proto.Packet) error } mScanner, ok := f.manualScanners.Load(req.ID) if !ok { - err = fmt.Errorf("task id(%v) not exist", req.ID) + err = fmt.Errorf(proto.ErrorTaskIDNotExistTpl, req.ID) return err } scanner := mScanner.(*ManualScanner) @@ -754,15 +754,15 @@ func (f *FlashNode) doStreamReadRequest(ctx context.Context, conn net.Conn, req } func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req *proto.CacheReadRequestBase, p *proto.Packet, - block *cachengine.CacheBlock, + block *cachengine.CacheBlock, reqId string, ) (err error) { - const action = "action[doObjectReadRequest]" + action := fmt.Sprintf("action[doObjectReadRequest] reqId(%v)", reqId) offset := int64(req.Offset) defer func() { if err != nil { // too many caching logs if strings.Compare(err.Error(), "require data is caching") != 0 { - log.LogWarnf("%s cache block(%v) err:%v", action, block.String(), err) + log.LogWarnf("%s cache block(%v) err:%v", action, block.GetBlockKey(), err) } } else { f.metrics.updateReadCountMetric(block.GetRootPath()) @@ -827,7 +827,7 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req offset = alignedOffset + proto.CACHE_BLOCK_PACKET_SIZE if log.EnableInfo() { log.LogInfof("%s ReqID[%d] key:[%s] reply[%s] block[%s]", action, p.ReqID, req.Key, - reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, errInner), block.String()) + reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, errInner), block.GetBlockKey()) } }) if err != nil { @@ -866,7 +866,7 @@ func (f *FlashNode) opCachePrepare(conn net.Conn, p *proto.Packet) (err error) { return } if req.CacheRequest == nil { - err = fmt.Errorf("no cache prepare request") + err = proto.ErrorNoCachePrepareRequest return } @@ -939,7 +939,7 @@ func (f *FlashNode) sendPrepareRequest(addr string, req *proto.CachePrepareReque } if reply.ResultCode != proto.OpOk { log.LogWarnf("%s reply(%v) from addr(%s) ResultCode(%d)", action, reply, addr, reply.ResultCode) - return fmt.Errorf("ResultCode(%v)", reply.ResultCode) + return fmt.Errorf(proto.ErrorResultCodeNOKTpl, reply.ResultCode) } return nil } diff --git a/sdk/remotecache/client.go b/sdk/remotecache/client.go index d756aaa45..f0854bac2 100755 --- a/sdk/remotecache/client.go +++ b/sdk/remotecache/client.go @@ -483,20 +483,44 @@ func (rc *RemoteCacheClient) getFlashGroupByKey(key string) (uint32, *FlashGroup return slot, fg, ownerSlot } -func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Reader, length int64) (err error) { - if length <= 0 || length > proto.CACHE_OBJECT_BLOCK_SIZE { - log.LogWarnf("put data length %v is leq 0 or gt 4M", length) - return fmt.Errorf("put data length %v is leq 0 or gt 4M", length) +func (rc *RemoteCacheClient) Delete(ctx context.Context, key string) (err error) { + if key == "" { + log.LogWarnf("delete block key is nil") + return proto.ErrorDeleteBlockKeyNil } slot := proto.ComputeCacheBlockSlot(key, 0, 0) fg, _ := rc.GetFlashGroupBySlot(slot) if fg == nil { - err = fmt.Errorf("FlashGroup put failed: cannot find any flashGroups") + err = proto.ErrorFlashGroupDeleteFailed + return + } + addr := fg.getFlashHost() + if log.EnableDebug() { + log.LogDebugf("FlashGroup addr(%v) delete key(%v)", addr, key) + } + if addr == "" { + err = proto.ErrorNoAvailableHost + log.LogWarnf("FlashGroup fg(%v) delete failed: err(%v)", fg, err) + return + } + rc.deleteRemoteBlock(key, addr) + return +} + +func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Reader, length int64) (err error) { + if length <= 0 || length > proto.CACHE_OBJECT_BLOCK_SIZE { + log.LogWarnf("put data length %v is leq 0 or gt 4M", length) + return fmt.Errorf(proto.ErrorPutDataLengthInvalidTpl, length) + } + slot := proto.ComputeCacheBlockSlot(key, 0, 0) + fg, _ := rc.GetFlashGroupBySlot(slot) + if fg == nil { + err = proto.ErrorFlashGroupPutFailed return } addr := fg.getFlashHost() if addr == "" { - err = fmt.Errorf("getFlashHost failed: can not find host") + err = proto.ErrorNoAvailableHost log.LogWarnf("FlashGroup fg(%v) reqId(%v) put failed: err(%v)", fg, reqId, err) return } @@ -539,7 +563,7 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re return } if replyPacket.ResultCode != proto.OpOk { - err = fmt.Errorf("%v", string(replyPacket.Data)) + err = fmt.Errorf(string(replyPacket.Data)) if !proto.IsFlashNodeLimitError(err) { log.LogWarnf("getPutBlockReply: ResultCode NOK, req(%v) reply(%v) ResultCode(%v)", reqPacket, replyPacket, replyPacket.ResultCode) } @@ -596,26 +620,28 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re } else if err != nil { return err } else if bufferOffset == readSize { - log.LogDebugf(" total written %v write size %v to fg", totalWritten, readSize) + if log.EnableDebug() { + log.LogDebugf(" total written %v write size %v to fg", totalWritten, readSize) + } break } } if bufferOffset != readSize { - log.LogWarnf("FlashGroup put: expected to read %d bytes, but only read %d", readSize, n) - return fmt.Errorf("expected to read %d bytes, but only read %d", readSize, n) + log.LogWarnf("FlashGroup put: expected to read %d bytes, but only read %d", readSize, bufferOffset) + return fmt.Errorf(proto.ErrorExpectedReadBytesMismatchTpl, readSize, bufferOffset) } - if readSize > 0 { + if bufferOffset > 0 { if log.EnableDebug() { - log.LogDebugf("FlashGroup put: write %d bytes total(%d) to fg", n, totalWritten) + log.LogDebugf("FlashGroup put: write %d bytes total(%d) to fg", bufferOffset, totalWritten) } binary.BigEndian.PutUint32(buf[proto.CACHE_BLOCK_PACKET_SIZE:], crc32.ChecksumIEEE(buf[:proto.CACHE_BLOCK_PACKET_SIZE])) conn.SetWriteDeadline(time.Now().Add(proto.WriteDeadlineTime * time.Second)) if _, err = conn.Write(buf[:writeLen]); err != nil { log.LogErrorf("wirte data and crc to flashnode get err %v", err) - return err + return fmt.Errorf(proto.ErrorWriteDataAndCRCToFlashNodeTpl, err) } ch.WaitAckChan <- struct{}{} - totalWritten += int64(readSize) + totalWritten += int64(bufferOffset) } if err == io.EOF { err = nil @@ -623,7 +649,9 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re } else if err != nil { return err } else if totalWritten == length { - log.LogDebugf(" total written %v write size %v to fg", totalWritten, readSize) + if log.EnableDebug() { + log.LogDebugf("key(%v) total written(%v) equal to lengh(%v)", key, totalWritten, length) + } break } } @@ -636,7 +664,7 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re return ch.RemoteError } if totalWritten != length { - return fmt.Errorf("unexpected data length: expected %d bytes, got %d", length, totalWritten) + return fmt.Errorf(proto.ErrorUnexpectedDataLengthTpl, length, totalWritten) } if log.EnableDebug() { log.LogDebugf("put data success key(%v) addr(%v) totalWriten(%v)", key, addr, totalWritten) @@ -655,7 +683,7 @@ func (rc *RemoteCacheClient) processPutReply(conn *net.TCPConn, ch *proto.CoonHa return } if replyPacket.ResultCode != proto.OpOk { - err = fmt.Errorf("%v", string(replyPacket.Data)) + err = fmt.Errorf(string(replyPacket.Data)) if !proto.IsFlashNodeLimitError(err) { log.LogWarnf("getPutBlockReply: put data ResultCode NOK, reqId(%v) reply(%v) ResultCode(%v)", reqId, replyPacket, replyPacket.ResultCode) } @@ -754,7 +782,7 @@ func (rc *RemoteCacheClient) Read(ctx context.Context, fg *FlashGroup, reqId int forceClose := err != nil && !proto.IsFlashNodeLimitError(err) rc.conns.PutConnect(conn, forceClose) if err != nil && strings.Contains(err.Error(), "timeout") { - err = fmt.Errorf("read timeout") + err = proto.ErrorReadTimeout } parts := strings.Split(addr, ":") if len(parts) > 0 && addr != "" { @@ -765,7 +793,7 @@ func (rc *RemoteCacheClient) Read(ctx context.Context, fg *FlashGroup, reqId int for { addr = fg.getFlashHost() if addr == "" { - err = fmt.Errorf("no available host") + err = proto.ErrorNoAvailableHost log.LogWarnf("FlashGroup Read failed: fg(%v) err(%v)", fg, err) return } @@ -822,7 +850,7 @@ func (rc *RemoteCacheClient) Prepare(ctx context.Context, fg *FlashGroup, req *p ) addr := fg.getFlashHost() if addr == "" { - err = fmt.Errorf("getFlashHost failed: can not find host") + err = proto.ErrorNoAvailableHost log.LogWarnf("FlashGroup prepare failed: err(%v)", err) return } @@ -856,7 +884,7 @@ func (rc *RemoteCacheClient) Prepare(ctx context.Context, fg *FlashGroup, req *p } if replyPacket.ResultCode != proto.OpOk { log.LogWarnf("FlashGroup Prepare: ResultCode NOK, replyPacket(%v), fg host(%v), ResultCode(%v)", replyPacket, addr, replyPacket.ResultCode) - err = fmt.Errorf("ResultCode NOK (%v)", replyPacket.ResultCode) + err = fmt.Errorf(proto.ErrorResultCodeNOKTpl, replyPacket.ResultCode) return } log.LogDebugf("FlashGroup Prepare successful: flashGroup(%v) addr(%v) CachePrepareRequest(%v) reqPacket(%v) replyPacket(%v) err(%v) moved(%v)", fg, addr, req, reqPacket, replyPacket, err, moved) @@ -874,7 +902,7 @@ func (rc *RemoteCacheClient) getReadReply(conn *net.TCPConn, reqPacket *proto.Pa return } if replyPacket.ResultCode != proto.OpOk { - err = fmt.Errorf("%v", string(replyPacket.Data)) + err = fmt.Errorf(string(replyPacket.Data)) if !proto.IsFlashNodeLimitError(err) { log.LogWarnf("getReadReply: ResultCode NOK, req(%v) reply(%v) ResultCode(%v)", reqPacket, replyPacket, replyPacket.ResultCode) } @@ -882,7 +910,7 @@ func (rc *RemoteCacheClient) getReadReply(conn *net.TCPConn, reqPacket *proto.Pa } expectCrc := crc32.ChecksumIEEE(replyPacket.Data[:replyPacket.Size]) if replyPacket.CRC != expectCrc { - err = fmt.Errorf("inconsistent CRC, expect(%v) reply(%v)", expectCrc, replyPacket.CRC) + err = fmt.Errorf(proto.ErrorInconsistentCRCTpl, expectCrc, replyPacket.CRC) log.LogWarnf("getReadReply: req(%v) err(%v)", req, err) return } @@ -899,7 +927,7 @@ func (rc *RemoteCacheClient) Name() string { func (rc *RemoteCacheClient) Get(ctx context.Context, reqId, key string, from, to int64) (r io.ReadCloser, length int64, shouldCache bool, err error) { log.LogDebugf("RemoteCacheClient Get: key(%v) reqId(%v) from(%v) to(%v)", key, reqId, from, to) if from < 0 || to-from <= 0 { - return nil, 0, false, fmt.Errorf("invalid range: from(%v) to(%v)", from, to) + return nil, 0, false, fmt.Errorf(proto.ErrorInvalidRangeTpl, from, to) } remoteCacheMetric := exporter.NewCounter("readRemoteCache") remoteCacheMetric.AddWithLabels(1, map[string]string{"key": key}) @@ -961,7 +989,7 @@ func (rc *RemoteCacheClient) readObjectFromRemoteCache(ctx context.Context, key slot, fg, ownerSlot := rc.getFlashGroupByKey(key) if fg == nil { - err = fmt.Errorf("readObjectFromRemoteCache failed: cannot find any flashGroups") + err = proto.ErrorReadObjectFromRemoteCacheFailed return } @@ -1018,7 +1046,7 @@ func (reader *RemoteCacheReader) getReadObjectReply(p *proto.Packet) (length int if _, err = io.ReadFull(reader.conn, p.Data[:size]); err != nil { return 0, err } - err = fmt.Errorf("%v", string(p.Data)) + err = fmt.Errorf(string(p.Data)) if !proto.IsFlashNodeLimitError(err) { log.LogWarnf("getReadObjectReply: ResultCode NOK, err(%v) ResultCode(%v)", err, p.ResultCode) } @@ -1082,7 +1110,7 @@ func (reader *RemoteCacheReader) read(p []byte) (n int, err error) { // check crc actualCrc := crc32.ChecksumIEEE(p) if actualCrc != reply.CRC { - err = fmt.Errorf("inconsistent CRC, offset(%v) extentOffset(%v) expect(%v) actualCrc(%v)", reply.KernelOffset, reply.ExtentOffset, reply.CRC, actualCrc) + err = fmt.Errorf(proto.ErrorInconsistentCRCObjectTpl, reply.KernelOffset, reply.ExtentOffset, reply.CRC, actualCrc) log.LogErrorf("RemoteCacheReader:Read check crc failed reqID(%v) offset(%v) extentOffset(%v) expect(%v) actualCrc(%v)", reader.reqID, reply.KernelOffset, reply.ExtentOffset, reply.CRC, actualCrc) return @@ -1107,7 +1135,7 @@ func (reader *RemoteCacheReader) Read(p []byte) (n int, err error) { } if reader.closed { log.LogErrorf("read from close reader reqID(%v)", reader.reqID) - return 0, fmt.Errorf("read from close reader reqID(%v)", reader.reqID) + return 0, fmt.Errorf(proto.ErrorReadFromCloseReaderTpl, reader.reqID) } if reader.ctx.Err() != nil { log.LogErrorf("RemoteCacheReader:reqID(%v) err(%v)", reader.reqID, err) @@ -1192,7 +1220,7 @@ func (rc *RemoteCacheClient) ReadObjectFirstReply(conn *net.TCPConn, p *proto.Pa if _, err = io.ReadFull(conn, p.Data[:size]); err != nil { return 0, err } - err = fmt.Errorf("%v", string(p.Data)) + err = fmt.Errorf(string(p.Data)) if !proto.IsFlashNodeLimitError(err) { log.LogWarnf("ReadObjectFirstReply: ResultCode NOK, err(%v) ResultCode(%v)", err, p.ResultCode) } @@ -1211,7 +1239,7 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req bgTime := stat.BeginStat() defer func() { if err != nil && strings.Contains(err.Error(), "timeout") { - err = fmt.Errorf("read timeout") + err = proto.ErrorReadTimeout } parts := strings.Split(addr, ":") if len(parts) > 0 && addr != "" { @@ -1225,7 +1253,7 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req } addr = fg.getFlashHost() if addr == "" { - err = fmt.Errorf("no available host") + err = proto.ErrorNoAvailableHost log.LogWarnf("%v FlashGroup Read failed: fg(%v) err(%v)", logPrefix, fg, err) return } diff --git a/tool/remotecache-benchmark/main.go b/tool/remotecache-benchmark/main.go index 23a6277c1..669cbaa4a 100644 --- a/tool/remotecache-benchmark/main.go +++ b/tool/remotecache-benchmark/main.go @@ -207,6 +207,8 @@ func main() { runGetTest := flag.Bool("get-test", false, "Run the Get test") verify := flag.Bool("verify", true, "Verify with the source file") clear := flag.Bool("clear-cache", true, "Clear page cache") + removeKey := flag.Bool("remove-key", false, "Remove specific key") + blockKey := flag.String("block-key", "", "Block unique key") flag.Parse() fmt.Printf("Parsed command-line arguments:\n") @@ -222,6 +224,8 @@ func main() { fmt.Printf(" -verify: %v\n", *verify) fmt.Printf(" -master: %v\n", *master) fmt.Printf(" -clear-cache: %v\n", *clear) + fmt.Printf(" -remove-key: %v\n", *removeKey) + fmt.Printf(" -block-key: %v\n", *blockKey) tester := NewBenchmarkTester(ensureAbsolutePath(*dataDir), *hddBase, *nvmeBase, *verify, *master, *clear) if *needGenerate { if err := tester.GenerateTestData(*totalSizeGB, *genConcurrency); err != nil { @@ -248,6 +252,10 @@ func main() { // Ensure all data has been preheated into the storage system. tester.RunGetBenchmark(ctx, *concurrency) } + + if *removeKey && *blockKey != "" { + tester.RunDeleteOperation(*blockKey) + } tester.Stop() } @@ -426,6 +434,16 @@ func (t *BenchmarkTester) LoadExistingTestFiles(concurrency int) error { return nil } +func (t *BenchmarkTester) RunDeleteOperation(blockKey string) { + fmt.Printf("\n=== Starting Delete operation for key: %s ===\n", blockKey) + if t.cacheStorage != nil { + err := t.cacheStorage.Delete(context.Background(), blockKey) + if err != nil { + fmt.Printf("Delete operation for key: %s failed err %v \n", blockKey, err) + } + } +} + func (t *BenchmarkTester) printBenchmarkResult(result *BenchmarkResult) { if result == nil { fmt.Printf("\n--- no invalid BenchmarkResult ---\n") @@ -561,9 +579,9 @@ func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage st break } } - if reads != int(len1) { + if err == nil && reads != int(len1) { err = fmt.Errorf("wrong len %v:expected[%v]", reads, len1) - } else { + } else if err == nil { latency = time.Since(startTime) if t.verify { readCrc := crc32.ChecksumIEEE(tmpBuf[:reads]) diff --git a/tool/remotecache-benchmark/storage/localfs.go b/tool/remotecache-benchmark/storage/localfs.go index 6123e4da8..d208f2600 100644 --- a/tool/remotecache-benchmark/storage/localfs.go +++ b/tool/remotecache-benchmark/storage/localfs.go @@ -309,3 +309,8 @@ func (fs *LocalFS) LoadTestFiles() error { func (fs *LocalFS) Stop() { } + +// Delete removes a file from local filesystem +func (fs *LocalFS) Delete(ctx context.Context, key string) error { + return nil +} diff --git a/tool/remotecache-benchmark/storage/storage.go b/tool/remotecache-benchmark/storage/storage.go index 1628ff9ff..ca2280bc5 100644 --- a/tool/remotecache-benchmark/storage/storage.go +++ b/tool/remotecache-benchmark/storage/storage.go @@ -10,4 +10,5 @@ type Storage interface { Get(ctx context.Context, reqId, key string, from, to int64) (r io.ReadCloser, length int64, shouldCache bool, err error) Name() string Stop() + Delete(ctx context.Context, key string) error }