From c5e2315c7de89974463b6fb5ad29c922790bd98e Mon Sep 17 00:00:00 2001 From: chihe Date: Tue, 19 Aug 2025 15:32:17 +0800 Subject: [PATCH] fix(flashnode): fixed the infinite loop issue in GetObject. close:#1000151055 Signed-off-by: chihe --- master/admin_task_manager.go | 2 +- proto/packet.go | 36 +++-- .../flashgroupmanager/admin_task_manager.go | 2 +- .../flashnode/cachengine/miss_cache.go | 1 - remotecache/flashnode/extent_reader.go | 2 +- remotecache/flashnode/flashnode.go | 2 +- remotecache/flashnode/flashnode_op.go | 37 +++-- remotecache/flashnode/flashnode_op_test.go | 9 +- sdk/data/stream/extent_flash_cache.go | 11 +- sdk/remotecache/client.go | 92 +++++++----- tool/remotecache-benchmark/main.go | 103 ++++++++----- tool/remotecache-benchmark/storage/localfs.go | 3 + tool/remotecache-benchmark/storage/storage.go | 1 + tool/remotecache-config/cmd/cluster.go | 137 +++++++++++++++++- tool/remotecache-config/main.go | 1 - util/conn_pool.go | 19 ++- 16 files changed, 336 insertions(+), 122 deletions(-) diff --git a/master/admin_task_manager.go b/master/admin_task_manager.go index d23065de5..b26916a17 100644 --- a/master/admin_task_manager.go +++ b/master/admin_task_manager.go @@ -56,7 +56,7 @@ func newAdminTaskManager(targetAddr, clusterID string) (sender *AdminTaskManager clusterID: clusterID, TaskMap: make(map[string]*proto.AdminTask), exitCh: make(chan struct{}, 1), - connPool: util.NewConnectPoolWithTimeout(idleConnTimeout, connectTimeout), + connPool: util.NewConnectPoolWithTimeout(idleConnTimeout, connectTimeout, false), } go sender.process() diff --git a/proto/packet.go b/proto/packet.go index c46316e0c..7b255337a 100644 --- a/proto/packet.go +++ b/proto/packet.go @@ -295,17 +295,17 @@ const ( OpWriteOpOfProtoVerForbidden uint8 = 0x88 OpMetaForbiddenMigration uint8 = 0x89 // Distributed cache related OP codes. - OpFlashNodeHeartbeat uint8 = 0xDA - OpFlashNodeCachePrepare uint8 = 0xDB - OpFlashNodeCacheRead uint8 = 0xDC - OpFlashNodeCachePutBlock uint8 = 0xD8 - OpFlashNodeCacheDelete uint8 = 0xD9 - OpFlashNodeCacheReadObject uint8 = 0xDD - OpFlashNodeSetReadIOLimits uint8 = 0xED - OpFlashNodeSetWriteIOLimits uint8 = 0xEE - OpFlashNodeScan uint8 = 0xD4 - OpFlashNodeTaskCommand uint8 = 0xD5 - OpClientHeartbeat uint8 = 0xD6 + OpFlashNodeHeartbeat uint8 = 0xC1 + OpFlashNodeCachePrepare uint8 = 0xC2 + OpFlashNodeCacheRead uint8 = 0xC3 + OpFlashNodeCachePutBlock uint8 = 0xC4 + OpFlashNodeCacheDelete uint8 = 0xC5 + OpFlashNodeCacheReadObject uint8 = 0xC6 + OpFlashNodeSetReadIOLimits uint8 = 0xC7 + OpFlashNodeSetWriteIOLimits uint8 = 0xC8 + OpFlashNodeScan uint8 = 0xC9 + OpFlashNodeTaskCommand uint8 = 0xCA + OpFlashSDKHeartbeat uint8 = 0xCB ) const ( @@ -756,6 +756,12 @@ func (p *Packet) GetOpMsg() (m string) { m = "OpIsRaftStatusOk" case OpFlashSDKHeartbeat: m = "OpFlashSDKHeartbeat" + case OpFlashNodeCachePutBlock: + m = "OpFlashNodeCachePutBlock" + case OpFlashNodeCacheDelete: + m = "OpFlashNodeCacheDelete" + case OpFlashNodeCacheReadObject: + m = "OpFlashNodeCacheReadObject" default: m = fmt.Sprintf("op:%v not found", p.Opcode) } @@ -1497,11 +1503,11 @@ func (p *Packet) IsForwardPkt() bool { // LogMessage logs the given message. func (p *Packet) LogMessage(action, remote string, start int64, err error) (m string) { if err == nil { - m = fmt.Sprintf("id[%v] isPrimaryBackReplLeader[%v] remote[%v] "+ - " cost[%v] ", p.GetUniqueLogId(), p.IsForwardPkt(), remote, (time.Now().UnixNano()-start)/1e6) + m = fmt.Sprintf("id[%v] action[%v] isPrimaryBackReplLeader[%v] remote[%v] "+ + " cost[%v] ", p.GetUniqueLogId(), action, p.IsForwardPkt(), remote, (time.Now().UnixNano()-start)/1e6) } else { - m = fmt.Sprintf("id[%v] isPrimaryBackReplLeader[%v] remote[%v]"+ - ", err[%v]", p.GetUniqueLogId(), p.IsForwardPkt(), remote, err.Error()) + m = fmt.Sprintf("id[%v] action[%v] isPrimaryBackReplLeader[%v] remote[%v]"+ + ", err[%v]", p.GetUniqueLogId(), action, p.IsForwardPkt(), remote, err.Error()) } return } diff --git a/remotecache/flashgroupmanager/admin_task_manager.go b/remotecache/flashgroupmanager/admin_task_manager.go index 2aae174fb..b6057a982 100644 --- a/remotecache/flashgroupmanager/admin_task_manager.go +++ b/remotecache/flashgroupmanager/admin_task_manager.go @@ -41,7 +41,7 @@ func newAdminTaskManager(targetAddr, clusterID string) (sender *AdminTaskManager clusterID: clusterID, TaskMap: make(map[string]*proto.AdminTask), exitCh: make(chan struct{}, 1), - connPool: util.NewConnectPoolWithTimeout(idleConnTimeout, connectTimeout), + connPool: util.NewConnectPoolWithTimeout(idleConnTimeout, connectTimeout, false), } go sender.process() diff --git a/remotecache/flashnode/cachengine/miss_cache.go b/remotecache/flashnode/cachengine/miss_cache.go index 18658ada3..78e9f60f7 100644 --- a/remotecache/flashnode/cachengine/miss_cache.go +++ b/remotecache/flashnode/cachengine/miss_cache.go @@ -53,7 +53,6 @@ func (mc *MissCache) Increment(key string) int32 { mc.evict(true) } proto.SetMissEntryExpiration(cme, mc.expiration) - element := mc.lruList.PushFront(cme) mc.cache[cme.UniKey] = element if log.EnableDebug() { diff --git a/remotecache/flashnode/extent_reader.go b/remotecache/flashnode/extent_reader.go index 7a0b0436d..91388d13e 100644 --- a/remotecache/flashnode/extent_reader.go +++ b/remotecache/flashnode/extent_reader.go @@ -39,7 +39,7 @@ var _ cachengine.ReadExtentData = ReadExtentData var extentReaderConnPool *util.ConnectPool func initExtentConnPool() { - extentReaderConnPool = util.NewConnectPoolWithTimeoutAndCap(5, 100, _connPoolIdleTimeout, 1) + extentReaderConnPool = util.NewConnectPoolWithTimeoutAndCap(5, 100, _connPoolIdleTimeout, 1, false) } func ReadExtentData(source *proto.DataSource, afterReadFunc cachengine.ReadExtentAfter, timeout int, volume string, ino uint64, clientIP string) (readBytes int, err error) { diff --git a/remotecache/flashnode/flashnode.go b/remotecache/flashnode/flashnode.go index 2aea26a84..d70048f9b 100644 --- a/remotecache/flashnode/flashnode.go +++ b/remotecache/flashnode/flashnode.go @@ -228,7 +228,7 @@ func (f *FlashNode) start(cfg *config.Config) (err error) { } f.initLimiter() initExtentConnPool() - f.connPool = util.NewConnectPoolWithTimeout(_connPoolIdleTimeout, 1) + f.connPool = util.NewConnectPoolWithTimeout(_connPoolIdleTimeout, 1, false) if err = f.startCacheEngine(); err != nil { return } diff --git a/remotecache/flashnode/flashnode_op.go b/remotecache/flashnode/flashnode_op.go index 6c403a1c6..ba55e69ac 100644 --- a/remotecache/flashnode/flashnode_op.go +++ b/remotecache/flashnode/flashnode_op.go @@ -51,7 +51,7 @@ func (f *FlashNode) preHandle(conn net.Conn, p *proto.Packet) error { func (f *FlashNode) handlePacket(conn net.Conn, p *proto.Packet) (err error) { switch p.Opcode { - case proto.OpClientHeartbeat: + case proto.OpFlashSDKHeartbeat: err = f.opClientHeartbeat(conn, p) case proto.OpFlashNodeHeartbeat: err = f.opFlashNodeHeartbeat(conn, p) @@ -424,12 +424,18 @@ func (f *FlashNode) opCacheObjectGet(conn net.Conn, p *proto.Packet) (err error) defer func() { stat.EndStat("FlashNode:opCacheObjectGet", err, bgTime, 1) }() - + // TODO: protobuf + reqID := string(p.Arg) defer func() { if err != nil { if !proto.IsFlashNodeLimitError(err) { - log.LogWarnf("action[opCacheObjectGet] key:[%s], logMsg:%s", key, + log.LogWarnf("action[opCacheObjectGet]reqID[%v] key:[%s], logMsg:%s", reqID, key, p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err)) + } else { + if log.EnableDebug() { + log.LogDebugf("action[opCacheObjectGet]reqID[%v] key:[%s], logMsg:%s", reqID, key, + p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err)) + } } p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) if e := p.WriteToConn(conn); e != nil { @@ -446,7 +452,9 @@ func (f *FlashNode) opCacheObjectGet(conn net.Conn, p *proto.Packet) (err error) return } - log.LogDebugf("opCacheObjectGet req(%v)", req) + if log.EnableDebug() { + log.LogDebugf("opCacheObjectGet req(%v) id(%v) begin", req, reqID) + } uniKey := req.Key pDir := cachengine.MapKeyToDirectory(uniKey) @@ -736,6 +744,7 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req // reply data to client end := offset + int64(req.Size_) + var errInner error for { err = f.limitRead.RunNoWait(proto.CACHE_BLOCK_PACKET_SIZE, false, func() { reply := proto.NewPacket() @@ -762,8 +771,8 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req p.Size = proto.CACHE_BLOCK_PACKET_SIZE p.ExtentOffset = offset - reply.CRC, err = block.Read(ctx, reply.Data[:], alignedOffset, proto.CACHE_BLOCK_PACKET_SIZE, f.waitForCacheBlock, true) - if err != nil { + reply.CRC, errInner = block.Read(ctx, reply.Data[:], alignedOffset, proto.CACHE_BLOCK_PACKET_SIZE, f.waitForCacheBlock, true) + if errInner != nil { bufRelease() return } @@ -776,23 +785,27 @@ func (f *FlashNode) doObjectReadRequest(ctx context.Context, conn net.Conn, req p.ResultCode = proto.OpOk bgTime := stat.BeginStat() - if err = reply.WriteToConnForOCS(conn); err != nil { + if errInner = reply.WriteToConnForOCS(conn); errInner != nil { bufRelease() log.LogErrorf("%s key:[%s] %s", action, req.Key, - reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, err)) + reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, errInner)) return } - stat.EndStat("HitCacheRead:ReplyToClient", err, bgTime, 1) + stat.EndStat("HitCacheRead:ReplyToClient", errInner, bgTime, 1) offset = alignedOffset + proto.CACHE_BLOCK_PACKET_SIZE bufRelease() 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, err), block.String()) + reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, errInner), block.String()) } }) if err != nil { return } + if errInner != nil { + err = errInner + return + } if uint64(offset) >= req.Offset+req.Size_ { break } @@ -978,9 +991,9 @@ func (f *FlashNode) shouldCache(key string) error { count = cachengine.AtomicLoadAndAddWithCAS(&cm.MissCount) } if count == _defaultMissCountThresholdInterval { - return proto.ErrorNotExistShouldCache + return fmt.Errorf("%v, now hit %v", proto.ErrorNotExistShouldCache.Error(), count) } else { - return proto.ErrorNotExistShouldNotCache + return fmt.Errorf("%v, now hit %v", proto.ErrorNotExistShouldNotCache.Error(), count) } } diff --git a/remotecache/flashnode/flashnode_op_test.go b/remotecache/flashnode/flashnode_op_test.go index 9007b81de..2fca3496c 100644 --- a/remotecache/flashnode/flashnode_op_test.go +++ b/remotecache/flashnode/flashnode_op_test.go @@ -17,9 +17,9 @@ package flashnode import ( "encoding/binary" "encoding/json" - "errors" "hash/crc32" "net" + "strings" "sync" "sync/atomic" "testing" @@ -278,12 +278,11 @@ func testShouldCache(t *testing.T) { go func() { defer wg.Done() err := flashServer.shouldCache(key) - switch { - case errors.Is(err, proto.ErrorNotExistShouldCache): + if strings.Contains(err.Error(), proto.ErrorNotExistShouldCache.Error()) { atomic.AddInt64(&countA, 1) - case errors.Is(err, proto.ErrorNotExistShouldNotCache): + } else if strings.Contains(err.Error(), proto.ErrorNotExistShouldNotCache.Error()) { atomic.AddInt64(&countB, 1) - default: + } else { t.Errorf("unexpected error: %v", err) } }() diff --git a/sdk/data/stream/extent_flash_cache.go b/sdk/data/stream/extent_flash_cache.go index f742acda5..724f84bca 100644 --- a/sdk/data/stream/extent_flash_cache.go +++ b/sdk/data/stream/extent_flash_cache.go @@ -219,7 +219,16 @@ func (rc *RemoteCache) Init(client *ExtentClient) (err error) { rc.volname = client.extentConfig.Volume rc.metaWrapper = client.metaWrapper rc.clusterEnable = client.enableRemoteCacheCluster - rc.remoteCacheClient, err = remotecache.NewRemoteCacheClient(client.extentConfig.Masters, proto.CACHE_BLOCK_SIZE, false, "") + cfg := &remotecache.ClientConfig{ + Masters: client.extentConfig.Masters, + BlockSize: proto.CACHE_BLOCK_SIZE, + NeedInitLog: false, + LogLevelStr: "", + LogDir: "", + ConnectTimeout: 500, + FirstPacketTimeout: 1000, + } + rc.remoteCacheClient, err = remotecache.NewRemoteCacheClient(cfg) err = rc.remoteCacheClient.UpdateFlashGroups() if err != nil { diff --git a/sdk/remotecache/client.go b/sdk/remotecache/client.go index 422405099..a74726d88 100755 --- a/sdk/remotecache/client.go +++ b/sdk/remotecache/client.go @@ -69,6 +69,7 @@ type RemoteCacheClient struct { SameZoneTimeout int64 // microsecond SameRegionTimeout int64 // ms AddressPingMap sync.Map + firstPacketTimeout int64 } func (as *AddressPingStats) Add(duration time.Duration) { @@ -95,10 +96,20 @@ func (as *AddressPingStats) Average() time.Duration { return total / time.Duration(len(as.durations)) } -func NewRemoteCacheClient(masters []string, blockSize uint64, needInitLog bool, logLevelStr string) (rc *RemoteCacheClient, err error) { - if needInitLog { - logLevel := log.ParseLogLevel(logLevelStr) - _, err = log.InitLog("/tmp/cfs", "remoteCacheClient", logLevel, nil, log.DefaultLogLeftSpaceLimitRatio) +type ClientConfig struct { + Masters []string + BlockSize uint64 + NeedInitLog bool + LogLevelStr string + LogDir string + ConnectTimeout int64 // ms + FirstPacketTimeout int64 // ms +} + +func NewRemoteCacheClient(config *ClientConfig) (rc *RemoteCacheClient, err error) { + if config.NeedInitLog { + logLevel := log.ParseLogLevel(config.LogLevelStr) + _, err = log.InitLog(config.LogDir, "remoteCacheClient", logLevel, nil, log.DefaultLogLeftSpaceLimitRatio) if err != nil { return nil, errors.New("failed to init log") } @@ -114,15 +125,16 @@ func NewRemoteCacheClient(masters []string, blockSize uint64, needInitLog bool, rc.flashGroups = btree.New(32) rc.ReadTimeout = proto.DefaultRemoteCacheClientReadTimeout rc.WriteTimeout = proto.DefaultRemoteCacheExtentReadTimeout + rc.firstPacketTimeout = config.FirstPacketTimeout - rc.mc = master.NewMasterClient(masters, false) - rc.conns = util.NewConnectPoolWithTimeoutAndCap(5, 500, ConnIdelTimeout, 1) + rc.mc = master.NewMasterClient(config.Masters, false) + rc.conns = util.NewConnectPoolWithTimeoutAndCap(5, 500, ConnIdelTimeout, config.ConnectTimeout, true) rc.TTL = proto.DefaultRemoteCacheTTL rc.RemoteCacheMultiRead = false rc.FlashNodeTimeoutCount = proto.DefaultFlashNodeTimeoutCount rc.SameZoneTimeout = proto.DefaultRemoteCacheSameZoneTimeout rc.SameRegionTimeout = proto.DefaultRemoteCacheSameRegionTimeout - rc.blockSize = blockSize + rc.blockSize = config.BlockSize err = rc.updateRemoteCacheConfig() if err != nil { @@ -350,7 +362,7 @@ func (rc *RemoteCacheClient) refreshHostLatency() { func (rc *RemoteCacheClient) HeartBeat(addr string) (duration time.Duration, err error) { var conn *net.TCPConn packet := proto.NewPacket() - packet.Opcode = proto.OpClientHeartbeat + packet.Opcode = proto.OpFlashSDKHeartbeat defer func() { rc.conns.PutConnect(conn, err != nil) @@ -519,7 +531,7 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re return } replyPacket := NewFlashCacheReply() - if err = replyPacket.ReadFromConnExt(conn, int(rc.WriteTimeout)); err != nil { + if err = replyPacket.ReadFromConnExt(conn, int(rc.firstPacketTimeout)); err != nil { log.LogWarnf("FlashGroup put: reqId(%v) failed to ReadFromConn, replyPacket(%v), fg host(%v) , err(%v)", reqId, replyPacket, addr, err) return } @@ -566,7 +578,7 @@ func (rc *RemoteCacheClient) Put(ctx context.Context, reqId, key string, r io.Re log.LogErrorf("wirte data and crc to flashnode get err %v", err) return err } - if err = replyPacket.ReadFromConnExt(conn, int(rc.WriteTimeout)); err != nil { + if err = replyPacket.ReadFromConnExt(conn, proto.NoReadDeadlineTime); err != nil { log.LogWarnf("FlashGroup put data: reqId(%v) failed to ReadFromConn, replyPacket(%v), fg host(%v) , err(%v)", reqId, replyPacket, addr, err) return } @@ -899,13 +911,15 @@ func (rc *RemoteCacheClient) readObjectFromRemoteCache(ctx context.Context, key req.Slot = uint64(slot)<<32 | uint64(ownerSlot) req.Offset = offset req.Size_ = size - if reader, length, err = rc.ReadObject(ctx, fg, reqId, &req); err != nil { - log.LogWarnf("readObjectFromRemoteCache: flashGroup read failed. offset(%v) size(%v) fg(%v) req(%v) reqId(%v) err(%v)", offset, size, fg, req, reqId, err) + log.LogWarnf("readObjectFromRemoteCache: flashGroup read failed. key(%v) offset(%v) size(%v) fg(%v) req(%v) reqId(%v)"+ + " err(%v)", key, offset, size, fg, req, reqId, err) return } - - log.LogDebugf("readObjectFromRemoteCache: cacheReadRequestBase(%v) key(%v) reqId(%v)", req, key, reqId) + if log.EnableDebug() { + log.LogDebugf("readObjectFromRemoteCache success: cacheReadRequestBase(%v) key(%v) reqId(%v) offset(%v) size(%v)", + req, key, reqId, offset, size) + } return } @@ -959,13 +973,15 @@ type RemoteCacheReader struct { rc *RemoteCacheClient needReadLen int64 alreadyReadLen int64 + reqID string } -func (rc *RemoteCacheClient) NewRemoteCacheReader(conn *net.TCPConn) *RemoteCacheReader { +func (rc *RemoteCacheClient) NewRemoteCacheReader(conn *net.TCPConn, reqID string) *RemoteCacheReader { r := &RemoteCacheReader{ conn: conn, reader: bufio.NewReaderSize(conn, proto.CACHE_BLOCK_PACKET_SIZE), rc: rc, + reqID: reqID, } return r } @@ -981,6 +997,7 @@ func (reader *RemoteCacheReader) Read(p []byte) (n int, err error) { expectLen, err = reader.getReadObjectReply(reply) if err != nil { + log.LogErrorf("RemoteCacheReader:reqID(%v) Read reply err(%v)", reader.reqID, err) return } @@ -989,7 +1006,7 @@ func (reader *RemoteCacheReader) Read(p []byte) (n int, err error) { _, err = io.ReadFull(reader.conn, buff) if err != nil { - log.LogErrorf("RemoteCacheReader:Read err(%v)", err) + log.LogErrorf("RemoteCacheReader:reqID(%v) Read err(%v)", reader.reqID, err) return } @@ -997,8 +1014,8 @@ func (reader *RemoteCacheReader) Read(p []byte) (n int, err error) { actualCrc := crc32.ChecksumIEEE(buff) if actualCrc != reply.CRC { err = fmt.Errorf("inconsistent CRC, offset(%v) extentOffset(%v) expect(%v) actualCrc(%v)", reply.KernelOffset, reply.ExtentOffset, reply.CRC, actualCrc) - log.LogErrorf("RemoteCacheReader:Read check crc failed offset(%v) extentOffset(%v) expect(%v) actualCrc(%v)", - 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 } @@ -1018,7 +1035,7 @@ func (reader *RemoteCacheReader) Close() error { return nil } -func (rc *RemoteCacheClient) ReadObjectFirstReply(conn *net.TCPConn, p *proto.Packet) (length int64, err error) { +func (rc *RemoteCacheClient) ReadObjectFirstReply(conn *net.TCPConn, p *proto.Packet, reqId string) (length int64, err error) { header, err := proto.Buffers.Get(util.PacketHeaderSize) if err != nil { header = make([]byte, util.PacketHeaderSize) @@ -1065,9 +1082,9 @@ func (rc *RemoteCacheClient) ReadObjectFirstReply(conn *net.TCPConn, p *proto.Pa func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, reqId string, req *proto.CacheReadRequestBase) (reader *RemoteCacheReader, length int64, err error) { var ( conn *net.TCPConn - moved bool addr string reqPacket *proto.Packet + logPrefix = fmt.Sprintf("reqID[%v]", reqId) ) bgTime := stat.BeginStat() defer func() { @@ -1087,17 +1104,18 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req addr = fg.getFlashHost() if addr == "" { err = fmt.Errorf("no available host") - log.LogWarnf("FlashGroup Read failed: fg(%v) err(%v)", fg, err) + log.LogWarnf("%v FlashGroup Read failed: fg(%v) err(%v)", logPrefix, fg, err) return } reqPacket = NewRemoteCachePacket(reqId, proto.OpFlashNodeCacheReadObject) if err = reqPacket.MarshalDataPb(req); err != nil { - log.LogWarnf("FlashGroup Read: failed to MarshalData (%+v). err(%v)", req, err) + log.LogWarnf("%v FlashGroup Read: failed to MarshalData (%+v). err(%v)", logPrefix, req, err) return } if conn, err = rc.conns.GetConnect(addr); err != nil { - log.LogWarnf("FlashGroup Read: get connection failed, addr(%v) reqPacket(%v) err(%v) remoteCacheMultiRead(%v)", addr, req, err, rc.RemoteCacheMultiRead) - moved = fg.moveToUnknownRank(addr, err, rc.FlashNodeTimeoutCount) + log.LogWarnf("%v FlashGroup Read: get connection failed, addr(%v) reqPacket(%v) err(%v) remoteCacheMultiRead(%v)", + logPrefix, addr, req, err, rc.RemoteCacheMultiRead) + fg.moveToUnknownRank(addr, err, rc.FlashNodeTimeoutCount) if rc.RemoteCacheMultiRead { log.LogInfof("Retrying due to GetConnect of addr(%v) failure err(%v)", addr, err) continue @@ -1106,9 +1124,10 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req } if err = reqPacket.WriteToConn(conn); err != nil { - log.LogWarnf("FlashGroup Read: failed to write to addr(%v) err(%v) remoteCacheMultiRead(%v)", addr, err, rc.RemoteCacheMultiRead) + log.LogWarnf("%v FlashGroup Read: failed to write to addr(%v) err(%v) remoteCacheMultiRead(%v)", + logPrefix, addr, err, rc.RemoteCacheMultiRead) rc.conns.PutConnect(conn, err != nil) - moved = fg.moveToUnknownRank(addr, err, rc.FlashNodeTimeoutCount) + fg.moveToUnknownRank(addr, err, rc.FlashNodeTimeoutCount) if rc.RemoteCacheMultiRead { log.LogInfof("Retrying due to write to addr(%v) failure err(%v)", addr, err) continue @@ -1117,19 +1136,24 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req } reply := new(proto.Packet) - length, err = rc.ReadObjectFirstReply(conn, reply) + // set time out for first packet + conn.SetReadDeadline(time.Now().Add(time.Millisecond * time.Duration(rc.firstPacketTimeout))) + length, err = rc.ReadObjectFirstReply(conn, reply, reqId) if err != nil { - log.LogWarnf("ReadObject getReadObjectReply from(%v) failed, reply(%v) error(%v)", conn.RemoteAddr(), reply, err) - rc.conns.PutConnect(conn, true) + log.LogWarnf("%v ReadObject getReadObjectReply from(%v) failed, reply(%v) error(%v)", + logPrefix, conn.RemoteAddr(), reply, err) + rc.conns.PutConnect(conn, err != nil) + fg.moveToUnknownRank(addr, err, rc.FlashNodeTimeoutCount) return nil, 0, err } - - reader = rc.NewRemoteCacheReader(conn) + conn.SetReadDeadline(time.Time{}) + reader = rc.NewRemoteCacheReader(conn, reqId) reader.needReadLen = int64(req.Size_) break } - - log.LogDebugf("FlashGroup Read: flashGroup(%v) addr(%v) CacheReadRequest(%v) reqPacket(%v) err(%v) moved(%v) remoteCacheMultiRead(%v)", - fg, addr, req, reqPacket, err, moved, rc.RemoteCacheMultiRead) + if log.EnableDebug() { + log.LogDebugf("%v FlashGroup Read: flashGroup(%v) addr(%v) CacheReadRequest(%v) reqPacket(%v) err(%v)"+ + " remoteCacheMultiRead(%v)", logPrefix, fg, addr, req, reqPacket, err, rc.RemoteCacheMultiRead) + } return } diff --git a/tool/remotecache-benchmark/main.go b/tool/remotecache-benchmark/main.go index 6a07c31a2..633aa8e99 100644 --- a/tool/remotecache-benchmark/main.go +++ b/tool/remotecache-benchmark/main.go @@ -29,13 +29,14 @@ import ( ) type BenchmarkTester struct { - dataDir string - fileCache sync.Map - fileCounts int64 - cacheStorage storage.Storage - nvmeStorage storage.Storage - hddStorage storage.Storage - verify bool + dataDir string + fileCache sync.Map + fileCounts int64 + cacheStorage storage.Storage + nvmeStorage storage.Storage + hddStorage storage.Storage + verify bool + clearPageCache bool } type BenchmarkResult struct { @@ -54,14 +55,15 @@ type BenchmarkResult struct { IOPS float64 } -func NewBenchmarkTester(dataDir, hddBase, nvmeBase string, verify bool, master string) *BenchmarkTester { +func NewBenchmarkTester(dataDir, hddBase, nvmeBase string, verify bool, master string, clearPageCache bool) *BenchmarkTester { os.MkdirAll(dataDir, 0o755) tester := &BenchmarkTester{ - dataDir: dataDir, - cacheStorage: nil, // TODO - nvmeStorage: nil, - hddStorage: nil, - verify: verify, + dataDir: dataDir, + cacheStorage: nil, // TODO + nvmeStorage: nil, + hddStorage: nil, + verify: verify, + clearPageCache: clearPageCache, } if hddBase != "" { fmt.Printf("\nCreate an HDD Tester %v\n", hddBase) @@ -76,7 +78,16 @@ func NewBenchmarkTester(dataDir, hddBase, nvmeBase string, verify bool, master s if master != "" { fmt.Printf("\nCreate a remote cache Tester %v\n", master) var err error - tester.cacheStorage, err = remotecache.NewRemoteCacheClient(strings.Split(master, ","), proto.PageSize, true, "debug") + cfg := &remotecache.ClientConfig{ + Masters: strings.Split(master, ","), + BlockSize: proto.PageSize, + NeedInitLog: true, + LogLevelStr: "debug", + LogDir: "/tmp/cfs", + ConnectTimeout: 500, + FirstPacketTimeout: 1000, + } + tester.cacheStorage, err = remotecache.NewRemoteCacheClient(cfg) if err != nil { fmt.Printf("\nCreate a remote cache Tester failed:%v\n", err) } @@ -194,7 +205,8 @@ func main() { master := flag.String("master", "", "Master address") runPutTest := flag.Bool("put-test", false, "Run the PUT test") runGetTest := flag.Bool("get-test", false, "Run the Get test") - verify := flag.Bool(" ", true, "Verify with the source file") + verify := flag.Bool("verify", true, "Verify with the source file") + clear := flag.Bool("clear-cache", true, "Clear page cache") flag.Parse() fmt.Printf("Parsed command-line arguments:\n") @@ -209,7 +221,8 @@ func main() { fmt.Printf(" -concurrency: %v\n", *concurrency) fmt.Printf(" -verify: %v\n", *verify) fmt.Printf(" -master: %v\n", *master) - tester := NewBenchmarkTester(ensureAbsolutePath(*dataDir), *hddBase, *nvmeBase, *verify, *master) + fmt.Printf(" -clear-cache: %v\n", *clear) + tester := NewBenchmarkTester(ensureAbsolutePath(*dataDir), *hddBase, *nvmeBase, *verify, *master, *clear) if *needGenerate { if err := tester.GenerateTestData(*totalSizeGB, *genConcurrency); err != nil { fmt.Printf("generate test files failed: %v\n", err) @@ -235,6 +248,20 @@ func main() { // Ensure all data has been preheated into the storage system. tester.RunGetBenchmark(ctx, *concurrency) } + tester.Stop() +} + +func (t *BenchmarkTester) Stop() { + if t.hddStorage != nil { + t.hddStorage.Stop() + } + if t.nvmeStorage != nil { + t.nvmeStorage.Stop() + } + + if t.cacheStorage != nil { + t.cacheStorage.Stop() + } } func (t *BenchmarkTester) RunPutBenchmark(ctx context.Context, concurrency int) { @@ -476,7 +503,10 @@ func (t *BenchmarkTester) RunGetBenchmark(ctx context.Context, concurrency int) func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage storage.Storage, concurrency int) *BenchmarkResult { fmt.Printf("Testing %s Get performance...\n", storage.Name()) - clearPageCache() + // if t.clearPageCache { + // clearPageCache() + // } + result := &BenchmarkResult{ StorageType: storage.Name(), OperationType: "Get", @@ -509,7 +539,8 @@ func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage st startTime := time.Now() from := int64(0) to := fh.Size - r, len, _, err := storage.Get(ctx, uuid.New().String(), fh.FileName, from, to) + reqId := uuid.New().String() + r, len, _, err := storage.Get(ctx, reqId, fh.FileName, from, to) var latency time.Duration if err == nil { dataBuf := bytespool.Alloc(proto.CACHE_BLOCK_PACKET_SIZE) @@ -518,9 +549,7 @@ func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage st for { readBytes, err := r.Read(dataBuf) if err != nil { - if err == io.EOF { - break - } + break } copy(tmpBuf[reads:], dataBuf[:readBytes]) reads += readBytes @@ -539,9 +568,9 @@ func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage st } } - if err != nil { + if err != nil && err != io.EOF { atomic.AddInt64(&errorCount, 1) - fmt.Printf("storage %v get %v failed: %v\n", storage.Name(), fh.FileName, err) + fmt.Printf("storage %v get %v failed reqID %v: %v\n", storage.Name(), fh.FileName, reqId, err) } else { atomic.AddInt64(&successCount, 1) atomic.AddInt64(&totalBytes, int64(len)) @@ -562,7 +591,7 @@ func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage st progress := index * 100 / int(t.fileCounts) if progress >= lastProgress+10 { lastProgress = progress - fmt.Printf("\r%s Progress: %d%%", storage.Name(), progress) + fmt.Printf("\r%s Progress: %d%% \n", storage.Name(), progress) } } return true @@ -599,16 +628,16 @@ func (t *BenchmarkTester) runStorageGetBenchmark(ctx context.Context, storage st return result } -func clearPageCache() { - data := []byte("1\n") - file, err := os.OpenFile("/proc/sys/vm/drop_caches", os.O_WRONLY, 0o644) - if err != nil { - fmt.Printf("open /proc/sys/vm/drop_caches failed: %v\n", err) - return - } - defer file.Close() - - if _, err := file.Write(data); err != nil { - fmt.Printf("write to /proc/sys/vm/drop_caches failed: %v\n", err) - } -} +// func clearPageCache() { +// data := []byte("1\n") +// file, err := os.OpenFile("/proc/sys/vm/drop_caches", os.O_WRONLY, 0o644) +// if err != nil { +// fmt.Printf("open /proc/sys/vm/drop_caches failed: %v\n", err) +// return +// } +// defer file.Close() +// +// if _, err := file.Write(data); err != nil { +// fmt.Printf("write to /proc/sys/vm/drop_caches failed: %v\n", err) +// } +// } diff --git a/tool/remotecache-benchmark/storage/localfs.go b/tool/remotecache-benchmark/storage/localfs.go index fee128337..87840264b 100644 --- a/tool/remotecache-benchmark/storage/localfs.go +++ b/tool/remotecache-benchmark/storage/localfs.go @@ -306,3 +306,6 @@ func (fs *LocalFS) LoadTestFiles() error { // TODO: maybe to much handle return nil } + +func (fs *LocalFS) Stop() { +} diff --git a/tool/remotecache-benchmark/storage/storage.go b/tool/remotecache-benchmark/storage/storage.go index ae7ec571f..1628ff9ff 100644 --- a/tool/remotecache-benchmark/storage/storage.go +++ b/tool/remotecache-benchmark/storage/storage.go @@ -9,4 +9,5 @@ type Storage interface { Put(ctx context.Context, reqId, key string, r io.Reader, length int64) (err error) Get(ctx context.Context, reqId, key string, from, to int64) (r io.ReadCloser, length int64, shouldCache bool, err error) Name() string + Stop() } diff --git a/tool/remotecache-config/cmd/cluster.go b/tool/remotecache-config/cmd/cluster.go index 4e7154dff..398fddf8a 100644 --- a/tool/remotecache-config/cmd/cluster.go +++ b/tool/remotecache-config/cmd/cluster.go @@ -1,6 +1,9 @@ package cmd import ( + "fmt" + "strconv" + "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/sdk/master" "github.com/spf13/cobra" @@ -10,7 +13,10 @@ const ( cmdClusterUse = "cluster [COMMAND]" cmdClusterShort = "Manage cluster components" - cmdClusterInfoShort = "Show cluster summary information" + cmdClusterInfoShort = "Show cluster summary information" + cmdClusterSetClusterInfoShort = "Set cluster parameters" + + cmdVolMinRemoteCacheTTL = 10 * 60 ) func newClusterCmd(client *master.MasterClient) *cobra.Command { @@ -20,11 +26,7 @@ func newClusterCmd(client *master.MasterClient) *cobra.Command { } clusterCmd.AddCommand( newClusterInfoCmd(client), - //TODO - //newClusterStatCmd(client), - //newClusterFreezeCmd(client), - //newClusterSetThresholdCmd(client), - //newClusterSetParasCmd(client), + newClusterSetParasCmd(client), ) return clusterCmd } @@ -46,3 +48,126 @@ func newClusterInfoCmd(client *master.MasterClient) *cobra.Command { } return cmd } + +func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command { + var clientIDKey string + var tmp int64 + handleTimeout := "" + readDataNodeTimeout := "" + optRcTTL := "" + optRcReadTimeout := "" + optRemoteCacheMultiRead := "" + optFlashNodeTimeoutCount := "" + optRemoteCacheSameZoneTimeout := "" + optRemoteCacheSameRegionTimeout := "" + cmd := &cobra.Command{ + Use: CliOpSetCluster, + Short: cmdClusterSetClusterInfoShort, + Run: func(cmd *cobra.Command, args []string) { + var err error + defer func() { + errout(err) + }() + if handleTimeout != "" { + if tmp, err = strconv.ParseInt(handleTimeout, 10, 64); err != nil { + err = fmt.Errorf("param (%v) failed, should be int", handleTimeout) + return + } + if tmp <= 0 { + err = fmt.Errorf("handleTimeout (%v) should grater than 0", handleTimeout) + return + } + } + if readDataNodeTimeout != "" { + if tmp, err = strconv.ParseInt(readDataNodeTimeout, 10, 64); err != nil { + err = fmt.Errorf("param (%v) failed, should be int", readDataNodeTimeout) + return + } + if tmp <= 0 { + err = fmt.Errorf("readDataNodeTimeout (%v) should grater than 0", readDataNodeTimeout) + return + } + } + + if optRcTTL != "" { + if tmp, err = strconv.ParseInt(optRcTTL, 10, 64); err != nil { + err = fmt.Errorf("param (%v) failed, should be int", optRcTTL) + return + } + if tmp <= cmdVolMinRemoteCacheTTL { + err = fmt.Errorf("param remoteCacheTTL(%v) must greater than or equal to %v", optRcTTL, cmdVolMinRemoteCacheTTL) + return + } + } + if optRcReadTimeout != "" { + if tmp, err = strconv.ParseInt(optRcReadTimeout, 10, 64); err != nil { + err = fmt.Errorf("param (%v) failed, should be int", optRcReadTimeout) + return + } + if tmp <= 0 { + err = fmt.Errorf("param remoteCacheReadTimeout(%v) must greater than 0", optRcReadTimeout) + return + } + } + if optRemoteCacheMultiRead != "" { + if _, err = strconv.ParseBool(optRemoteCacheMultiRead); err != nil { + err = fmt.Errorf("param remoteCacheMultiRead(%v) should be true or false", optRemoteCacheMultiRead) + return + } + } + if optFlashNodeTimeoutCount != "" { + if tmp, err = strconv.ParseInt(optFlashNodeTimeoutCount, 10, 64); err != nil { + err = fmt.Errorf("param (%v) failed, should be int", optFlashNodeTimeoutCount) + return + } + if tmp <= 0 { + err = fmt.Errorf("param flashNodeTimeoutCount(%v) must greater than 0", optFlashNodeTimeoutCount) + return + } + } + if optRemoteCacheSameZoneTimeout != "" { + if tmp, err = strconv.ParseInt(optRemoteCacheSameZoneTimeout, 10, 64); err != nil { + err = fmt.Errorf("param (%v) failed, should be int", optRemoteCacheSameZoneTimeout) + return + } + if tmp <= 0 { + err = fmt.Errorf("param remoteCacheSameZoneTimeout(%v) must greater than 0", optRemoteCacheSameZoneTimeout) + return + } + } + if optRemoteCacheSameRegionTimeout != "" { + if tmp, err = strconv.ParseInt(optRemoteCacheSameRegionTimeout, 10, 64); err != nil { + err = fmt.Errorf("param (%v) failed, should be int", optRemoteCacheSameRegionTimeout) + return + } + if tmp <= 0 { + err = fmt.Errorf("param remoteCacheSameRegionTimeout(%v) must greater than 0", optRemoteCacheSameRegionTimeout) + return + } + } + + if err = client.AdminAPI().SetClusterParas("", "", "", + "", "", "", "", clientIDKey, + "", "", + "", "", + "", "", "", "", "", "", + "", "", handleTimeout, readDataNodeTimeout, + optRcTTL, optRcReadTimeout, optRemoteCacheMultiRead, optFlashNodeTimeoutCount, + optRemoteCacheSameZoneTimeout, optRemoteCacheSameRegionTimeout); err != nil { + return + } + stdout("Cluster parameters has been set successfully. \n") + }, + } + cmd.Flags().StringVar(&clientIDKey, CliFlagClientIDKey, client.ClientIDKey(), CliUsageClientIDKey) + cmd.Flags().StringVar(&handleTimeout, "flashNodeHandleReadTimeout", "", "Specify flash node handle read timeout (example:1000ms)") + cmd.Flags().StringVar(&readDataNodeTimeout, "flashNodeReadDataNodeTimeout", "", "Specify flash node read data node timeout (example:3000ms)") + cmd.Flags().StringVar(&optRcTTL, CliFlagRemoteCacheTTL, "", "Remote cache ttl[Unit: s](must >= 10min, default 5day)") + cmd.Flags().StringVar(&optRcReadTimeout, CliFlagRemoteCacheReadTimeout, "", "Remote cache read timeout millisecond(must > 0)") + cmd.Flags().StringVar(&optRemoteCacheMultiRead, CliFlagRemoteCacheMultiRead, "", "Remote cache follower read(true|false)") + cmd.Flags().StringVar(&optFlashNodeTimeoutCount, CliFlagFlashNodeTimeoutCount, "", "FlashNode timeout count, flashNode will be removed by client if it's timeout count exceeds this value") + cmd.Flags().StringVar(&optRemoteCacheSameZoneTimeout, CliFlagRemoteCacheSameZoneTimeout, "", "Remote cache same zone timeout microsecond(must > 0)") + cmd.Flags().StringVar(&optRemoteCacheSameRegionTimeout, CliFlagRemoteCacheSameRegionTimeout, "", "Remote cache same region timeout millisecond(must > 0)") + + return cmd +} diff --git a/tool/remotecache-config/main.go b/tool/remotecache-config/main.go index 02cc3ada2..356d93a0f 100644 --- a/tool/remotecache-config/main.go +++ b/tool/remotecache-config/main.go @@ -44,5 +44,4 @@ func main() { os.Exit(1) } log.LogFlush() - } diff --git a/util/conn_pool.go b/util/conn_pool.go index 85ad817bf..328626255 100644 --- a/util/conn_pool.go +++ b/util/conn_pool.go @@ -40,6 +40,7 @@ type ConnectPool struct { connectTimeout int64 closeCh chan struct{} closeOnce sync.Once + useMilliSecond bool } func NewConnectPool() (cp *ConnectPool) { @@ -56,11 +57,11 @@ func NewConnectPool() (cp *ConnectPool) { return cp } -func NewConnectPoolWithTimeout(idleConnTimeout time.Duration, connectTimeout int64) (cp *ConnectPool) { - return NewConnectPoolWithTimeoutAndCap(5, 80, idleConnTimeout, connectTimeout) +func NewConnectPoolWithTimeout(idleConnTimeout time.Duration, connectTimeout int64, useMilliSecond bool) (cp *ConnectPool) { + return NewConnectPoolWithTimeoutAndCap(5, 80, idleConnTimeout, connectTimeout, useMilliSecond) } -func NewConnectPoolWithTimeoutAndCap(minCap, maxCap int, idleConnTimeout time.Duration, connectTimeout int64) (cp *ConnectPool) { +func NewConnectPoolWithTimeoutAndCap(minCap, maxCap int, idleConnTimeout time.Duration, connectTimeout int64, useMilliSecond bool) (cp *ConnectPool) { cp = &ConnectPool{ pools: make(map[string]*Pool), mincap: minCap, @@ -68,6 +69,7 @@ func NewConnectPoolWithTimeoutAndCap(minCap, maxCap int, idleConnTimeout time.Du timeout: int64(idleConnTimeout * time.Second), connectTimeout: connectTimeout, closeCh: make(chan struct{}), + useMilliSecond: useMilliSecond, } go cp.autoRelease() return cp @@ -90,7 +92,7 @@ func (cp *ConnectPool) GetConnect(targetAddr string) (c *net.TCPConn, err error) pool, ok := cp.pools[targetAddr] cp.RUnlock() if !ok { - newPool := NewPool(cp.mincap, cp.maxcap, cp.timeout, cp.connectTimeout, targetAddr) + newPool := NewPool(cp.mincap, cp.maxcap, cp.timeout, cp.connectTimeout, targetAddr, cp.useMilliSecond) cp.Lock() pool, ok = cp.pools[targetAddr] if !ok { @@ -209,9 +211,10 @@ type Pool struct { target string timeout int64 connectTimeout int64 + useMilliSecond bool } -func NewPool(min, max int, timeout, connectTimeout int64, target string) (p *Pool) { +func NewPool(min, max int, timeout, connectTimeout int64, target string, useMilliSecond bool) (p *Pool) { p = new(Pool) p.mincap = min p.maxcap = max @@ -230,7 +233,11 @@ func (p *Pool) initAllConnect() { ) for i := 0; i < p.mincap; i++ { if p.connectTimeout != 0 { - c, err = net.DialTimeout("tcp", p.target, time.Duration(p.connectTimeout)*time.Second) + if p.useMilliSecond { + c, err = net.DialTimeout("tcp", p.target, time.Duration(p.connectTimeout)*time.Millisecond) + } else { + c, err = net.DialTimeout("tcp", p.target, time.Duration(p.connectTimeout)*time.Second) + } } else { c, err = net.Dial("tcp", p.target) }