From a3e89bcfaf71f1abde53d2bbc65a438b3a7bd40a Mon Sep 17 00:00:00 2001 From: slasher Date: Thu, 21 Dec 2023 10:36:25 +0800 Subject: [PATCH] feat(flashnode): refinement flashnode . #22200504 #22200509 Signed-off-by: slasher --- cmd/cfg/flashnode.json | 19 ++ cmd/cmd.go | 6 + flashnode/cachengine/engine.go | 8 +- flashnode/extent_reader.go | 196 +++++++++++++++ flashnode/flashnode.go | 157 ++++++------ flashnode/flashnode_op.go | 257 ++++++++------------ flashnode/nodeinfo.go | 101 -------- flashnode/packet.go | 119 --------- flashnode/read_source.go | 208 ---------------- flashnode/{api_handler.go => serve_http.go} | 17 +- flashnode/{server.go => serve_tcp.go} | 44 ++-- 11 files changed, 449 insertions(+), 683 deletions(-) create mode 100644 cmd/cfg/flashnode.json create mode 100644 flashnode/extent_reader.go delete mode 100644 flashnode/nodeinfo.go delete mode 100644 flashnode/packet.go delete mode 100644 flashnode/read_source.go rename flashnode/{api_handler.go => serve_http.go} (78%) rename flashnode/{server.go => serve_tcp.go} (63%) diff --git a/cmd/cfg/flashnode.json b/cmd/cfg/flashnode.json new file mode 100644 index 000000000..abc966270 --- /dev/null +++ b/cmd/cfg/flashnode.json @@ -0,0 +1,19 @@ +{ + "role": "flashnode", + "listen": "6100", + "prof": "6101", + "logLevel": "info", + "logDir": "/export/Logs/flashnode", + "warnLogDir":"/export/home/tomcat/UMP-Monitor/logs/", + "consulAddr": "http://consul.prometheus-cfs.local", + "exporterPort": 9514, + "masterAddr": [ + "192.168.31.173:80", + "192.168.31.141:80", + "192.168.30.200:80" + ], + "zoneName": "zone", + "memTotal": 0, + "memPercent": 0.5, + "readRps": 512 +} diff --git a/cmd/cmd.go b/cmd/cmd.go index e403ce5dc..2f74d6a96 100644 --- a/cmd/cmd.go +++ b/cmd/cmd.go @@ -35,6 +35,7 @@ import ( "github.com/cubefs/cubefs/cmd/common" "github.com/cubefs/cubefs/console" "github.com/cubefs/cubefs/datanode" + "github.com/cubefs/cubefs/flashnode" "github.com/cubefs/cubefs/lcnode" "github.com/cubefs/cubefs/master" "github.com/cubefs/cubefs/metanode" @@ -73,6 +74,7 @@ const ( RoleObject = "objectnode" RoleConsole = "console" RoleLifeCycle = "lcnode" + RoleFlash = "flashnode" ) const ( @@ -83,6 +85,7 @@ const ( ModuleObject = "objectNode" ModuleConsole = "console" ModuleLifeCycle = "lcnode" + ModuleFlash = "flashNode" ) const ( @@ -218,6 +221,9 @@ func main() { case RoleLifeCycle: server = lcnode.NewServer() module = ModuleLifeCycle + case RoleFlash: + server = flashnode.NewServer() + module = ModuleFlash default: err = errors.NewErrorf("Fatal: role mismatch: %s", role) fmt.Println(err) diff --git a/flashnode/cachengine/engine.go b/flashnode/cachengine/engine.go index 874c8899d..3823d6ed4 100644 --- a/flashnode/cachengine/engine.go +++ b/flashnode/cachengine/engine.go @@ -73,7 +73,8 @@ type CacheEngine struct { } type ( - ReadExtentData func(source *proto.DataSource, afterReadFunc func([]byte, int64) error) (n int, err error) + ReadExtentAfter func([]byte, int64) error + ReadExtentData func(source *proto.DataSource, afterReadFunc ReadExtentAfter) (n int, err error) ) func NewCacheEngine(dataDir string, totalSize int64, maxUseRatio float64, @@ -81,6 +82,9 @@ func NewCacheEngine(dataDir string, totalSize int64, maxUseRatio float64, ) (s *CacheEngine, err error) { s = new(CacheEngine) s.dataPath = dataDir + if maxUseRatio < 1e-1 { + maxUseRatio = DefaultCacheMaxUsedRatio + } s.config = CacheConfig{ MaxAlloc: int64(float64(totalSize) * maxUseRatio), Total: totalSize, @@ -313,7 +317,7 @@ func (c *CacheEngine) EvictCacheByVolume(evictVol string) (failedKeys []interfac func (c *CacheEngine) EvictCacheAll() { c.lruCache.EvictAll() - log.LogWarnf("action[EvictCacheAll] evict all finish") + log.LogWarn("action[EvictCacheAll] evict all finish") } func GenCacheBlockKey(volume string, inode, offset uint64, version uint32) string { diff --git a/flashnode/extent_reader.go b/flashnode/extent_reader.go new file mode 100644 index 000000000..fcbbe7a4c --- /dev/null +++ b/flashnode/extent_reader.go @@ -0,0 +1,196 @@ +// Copyright 2018 The CubeFS Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +// implied. See the License for the specific language governing +// permissions and limitations under the License. + +package flashnode + +import ( + "fmt" + "hash/crc32" + "net" + "strings" + "time" + + "github.com/cubefs/cubefs/flashnode/cachengine" + "github.com/cubefs/cubefs/proto" + "github.com/cubefs/cubefs/util" + "github.com/cubefs/cubefs/util/errors" + "github.com/cubefs/cubefs/util/exporter" + "github.com/cubefs/cubefs/util/log" +) + +var _ cachengine.ReadExtentData = ReadExtentData + +var extentReaderConnPool *util.ConnectPool + +func initExtentConnPool() { + extentReaderConnPool = util.NewConnectPoolWithTimeoutAndCap(0, 10, _connPoolIdleTimeout, 1) +} + +func ReadExtentData(source *proto.DataSource, afterReadFunc cachengine.ReadExtentAfter) (readBytes int, err error) { + reqPacket := newReadPacket(&proto.ExtentKey{ + PartitionId: source.PartitionID, + ExtentId: source.ExtentID, + }, int(source.ExtentOffset), int(source.Size_), source.FileOffset, true) + + readBytes, err = extentReadWithRetry(reqPacket, source.Hosts, afterReadFunc) + if err != nil { + log.LogErrorf("read extent err(%v)", err) + } + return +} + +func extentReadWithRetry(reqPacket *proto.Packet, hosts []string, afterReadFunc cachengine.ReadExtentAfter) (readBytes int, err error) { + errMap := make(map[string]error) + startTime := time.Now() + for try := range [_extentReadMaxRetry]struct{}{} { + try++ + for _, addr := range hosts { + if addr == "" { + continue + } + log.LogDebugf("extentReadWithRetry: try(%d) addr(%s) reqPacket(%v)", try, addr, reqPacket) + if readBytes, err = readFromDataPartition(addr, reqPacket, afterReadFunc); err == nil { + return + } + errMap[addr] = err + log.LogWarnf("extentReadWithRetry: try(%d) addr(%s) reqPacket(%v) err(%v)", try, addr, reqPacket, err) + if strings.Contains(err.Error(), proto.ErrTmpfsNoSpace.Error()) { + break + } + } + if time.Since(startTime) > _extentReadTimeoutSec*time.Second { + log.LogWarnf("extentReadWithRetry: retry timeout req(%v) time(%v)", reqPacket, time.Since(startTime)) + break + } + log.LogWarnf("extentReadWithRetry: errMap(%v) reqPacket(%v) try the next round", errMap, reqPacket) + time.Sleep(_extentReadInterval) + } + err = errors.NewErrorf("FollowerRead: tried %d times reqPacket(%v) errMap(%v)", _extentReadMaxRetry, reqPacket, errMap) + return +} + +func readFromDataPartition(addr string, reqPacket *proto.Packet, afterReadFunc cachengine.ReadExtentAfter) (readBytes int, err error) { + var conn *net.TCPConn + var why string + defer func() { + extentReaderConnPool.PutConnect(conn, err != nil) + if err != nil { + log.LogWarnf("readFromDataPartition: %s addr(%s) reqPacket(%v) err(%v)", why, addr, reqPacket, err) + } + }() + if conn, err = extentReaderConnPool.GetConnect(addr); err != nil { + why = "get connection" + return + } + + log.LogDebugf("readFromDataPartition: addr(%s) reqPacket(%v)", addr, reqPacket) + if reqPacket.Opcode == proto.OpStreamFollowerRead { + reqPacket.StartT = time.Now().UnixNano() + } + if err = reqPacket.WriteToConn(conn); err != nil { + why = "set to connection" + return + } + if readBytes, err = getReadReply(conn, reqPacket, afterReadFunc); err != nil { + why = "get reply" + return + } + return +} + +func getReadReply(conn *net.TCPConn, reqPacket *proto.Packet, afterReadFunc cachengine.ReadExtentAfter) (readBytes int, err error) { + buf, bufErr := proto.Buffers.Get(util.ReadBlockSize) + defer func() { + if bufErr == nil { + proto.Buffers.Put(buf[:util.ReadBlockSize]) + } + if err != nil { + log.LogWarnf("getReadReply: req(%v) readBytes(%v) err(%v)", reqPacket, readBytes, err) + } + }() + if bufErr != nil { + buf = make([]byte, reqPacket.Size) + } + + for readBytes < int(reqPacket.Size) { + reply := newReplyPacket(reqPacket.ReqID, reqPacket.PartitionID, reqPacket.ExtentID) + bufSize := util.Min(util.ReadBlockSize, int(reqPacket.Size)-readBytes) + reply.Data = buf[readBytes : readBytes+bufSize] + if err = reply.ReadFromConn(conn, _extentReadTimeoutSec); err != nil { + return + } + + if err = checkReadReplyValid(reqPacket, reply); err != nil { + return + } + if afterReadFunc != nil { + if err = afterReadFunc(reply.Data[:bufSize], int64(bufSize)); err != nil { + return + } + } + if reply.Size != uint32(bufSize) { + exporter.Warning(fmt.Sprintf("action[getReadReply] reply size not valid, "+ + "ReqID(%v) PartitionID(%v) Extent(%v) buffSize(%v) ReplySize(%v)", + reqPacket.ReqID, reqPacket.PartitionID, reqPacket.ExtentID, bufSize, reply.Size)) + } + readBytes += int(reply.Size) + } + return readBytes, nil +} + +func checkReadReplyValid(request *proto.Packet, reply *proto.Packet) (err error) { + if reply.ResultCode != proto.OpOk { + return errors.NewErrorf("ResultCode(%v) NOTOK", reply.GetResultMsg()) + } + if !isValidReadReply(request, reply) { + return errors.NewErrorf("inconsistent req and reply, req(%v) reply(%v)", request, reply) + } + expectCrc := crc32.ChecksumIEEE(reply.Data[:reply.Size]) + if reply.CRC != expectCrc { + return errors.NewErrorf("inconsistent CRC, expectCRC(%v) replyCRC(%v)", expectCrc, reply.CRC) + } + return nil +} + +func isValidReadReply(p *proto.Packet, q *proto.Packet) bool { + return p.ReqID == q.ReqID && p.PartitionID == q.PartitionID && p.ExtentID == q.ExtentID +} + +func newReadPacket(key *proto.ExtentKey, extentOffset, size int, fileOffset uint64, followerRead bool) *proto.Packet { + p := new(proto.Packet) + p.ExtentID = key.ExtentId + p.PartitionID = key.PartitionId + p.Magic = proto.ProtoMagic + p.ReqID = proto.GenerateRequestID() + p.ExtentOffset = int64(extentOffset) + p.Size = uint32(size) + p.ExtentType = proto.NormalExtentType + p.RemainingFollowers = 0 + p.KernelOffset = fileOffset + p.Opcode = proto.OpStreamRead + if followerRead { + p.Opcode = proto.OpStreamFollowerRead + } + return p +} + +func newReplyPacket(reqID int64, partitionID uint64, extentID uint64) *proto.Packet { + p := new(proto.Packet) + p.ReqID = reqID + p.PartitionID = partitionID + p.ExtentID = extentID + p.Magic = proto.ProtoMagic + p.ExtentType = proto.NormalExtentType + return p +} diff --git a/flashnode/flashnode.go b/flashnode/flashnode.go index 32519e87e..5ffce7331 100644 --- a/flashnode/flashnode.go +++ b/flashnode/flashnode.go @@ -15,12 +15,14 @@ package flashnode import ( + "fmt" "net" - "strconv" "strings" "sync" "time" + "golang.org/x/time/rate" + "github.com/cubefs/cubefs/cmd/common" "github.com/cubefs/cubefs/flashnode/cachengine" "github.com/cubefs/cubefs/proto" @@ -30,7 +32,6 @@ import ( "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/exporter" "github.com/cubefs/cubefs/util/log" - "golang.org/x/time/rate" ) //TODO: remove this later. @@ -40,25 +41,22 @@ const ( TmpfsPath = "/cfs/tmpfs" moduleName = "flashNode" - volumePara = "volume" - LruCacheDefaultCapacity = 400000 - UpdateRateLimitInfoInterval = 60 * time.Second - ServerTimeout = 60 * 5 - ConnectPoolIdleConnTimeout = 60 * time.Second - DefaultBurst = 512 - - ExtentReadMaxRetry = 3 - ExtentReadTimeoutSec = 3 - ExtentReadSleepInterval = 100 * time.Millisecond - IdleConnTimeoutData = 30 * time.Second + _defaultReadBurst = 512 + _defaultLRUCapacity = 400000 + _tcpServerTimeoutSec = 60 * 5 + _connPoolIdleTimeout = 60 * time.Second + _extentReadMaxRetry = 3 + _extentReadTimeoutSec = 3 + _extentReadInterval = 100 * time.Millisecond ) // Configuration keys const ( - cfgLocalIP = "localIP" - cfgTotalMem = "totalMem" - cfgZoneName = "zoneName" + cfgMemTotal = "memTotal" + cfgMemPercent = "memPercent" + cfgZoneName = "zoneName" + cfgReadRps = "readRps" ) // The FlashNode manages the inode block cache to speed the file reading. @@ -75,19 +73,16 @@ type FlashNode struct { clusterID string nodeID uint64 - nodeLimit uint64 - nodeLimiter *rate.Limiter - volLimitMap map[string]uint64 // volume -> limit - volLimiterMap map[string]*rate.Limiter // volume -> *Limiter - control common.Control stopOnce sync.Once stopCh chan struct{} connPool *util.ConnectPool - netListener net.Listener - readSource *ReadSource + tcpListener net.Listener cacheEngine *cachengine.CacheEngine + + readRps int + readLimiter *rate.Limiter } // Start starts up the flash node with the specified configuration. @@ -122,15 +117,14 @@ func doStart(s common.Server, cfg *config.Config) (err error) { return } f.initLimiter() - f.connPool = util.NewConnectPoolWithTimeout(ConnectPoolIdleConnTimeout, 1) + initExtentConnPool() + f.connPool = util.NewConnectPoolWithTimeout(_connPoolIdleTimeout, 1) f.registerAPIHandler() - go f.startUpdateScheduler() - if err = f.startTcpServer(); err != nil { + if err = f.startCacheEngine(); err != nil { return } - f.readSource = NewReadSource() - if err = f.startCacheEngine(); err != nil { + if err = f.startTcpServer(); err != nil { return } @@ -156,26 +150,39 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) { if cfg == nil { return errors.New("invalid configuration") } - f.localAddr = cfg.GetString(cfgLocalIP) f.listen = strings.TrimSpace(cfg.GetString(proto.ListenPort)) if f.listen == "" { return errors.New("bad listen config") } - f.zoneName = cfg.GetString(cfgZoneName) - f.total, err = strconv.ParseUint(cfg.GetString(cfgTotalMem), 10, 64) - if err != nil { - return err + if f.zoneName = cfg.GetString(cfgZoneName); f.zoneName == "" { + return errors.New("bad zoneName config") } - if f.total == 0 { - return errors.New("recommended to be configured as 80 percent of physical machine memory") + f.readRps = cfg.GetInt(cfgReadRps) + if f.readRps <= 0 { + f.readRps = _defaultReadBurst } - total, _, err := util.GetMemInfo() - if err == nil && f.total > uint64(float64(total)*0.8) { - return errors.New("recommended to be configured as 80 percent of physical machine memory") + + mem := cfg.GetInt64(cfgMemTotal) + if mem <= 0 { + percent := cfg.GetFloat(cfgMemPercent) + if percent <= 1e-2 || percent > 0.8 { + return errors.NewErrorf("recommended to physical memory %s=0.8 %.2f", cfgMemPercent, percent) + } + total, _, err := util.GetMemInfo() + if err != nil { + return errors.NewErrorf("get physical memory %v", err) + } + mem = int64(float64(total) * percent) } - log.LogInfof("[parseConfig] load localAddr[%v].", f.localAddr) - log.LogInfof("[parseConfig] load listen[%v].", f.listen) - log.LogInfof("[parseConfig] load zoneName[%v].", f.zoneName) + if mem < 32*(1<<20) { + return errors.NewErrorf("low physical memory %d", mem) + } + f.total = uint64(mem) + + log.LogInfof("[parseConfig] load listen[%s].", f.listen) + log.LogInfof("[parseConfig] load zoneName[%s].", f.zoneName) + log.LogInfof("[parseConfig] load totalMem[%d].", f.total) + log.LogInfof("[parseConfig] load readRps[%d].", f.readRps) f.mc = master.NewMasterClient(cfg.GetStringSlice(proto.MasterAddr), false) if len(f.mc.Nodes()) == 0 { @@ -186,16 +193,16 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) { func (f *FlashNode) stopCacheEngine() { if f.cacheEngine != nil { - err := f.cacheEngine.Stop() - if err != nil { - log.LogErrorf("stopCacheEngine: err:%v", err) + if err := f.cacheEngine.Stop(); err != nil { + log.LogErrorf("stopCacheEngine err:%v", err) } } } func (f *FlashNode) startCacheEngine() (err error) { - if f.cacheEngine, err = cachengine.NewCacheEngine(f.tmpPath, int64(f.total), cachengine.DefaultCacheMaxUsedRatio, LruCacheDefaultCapacity, time.Hour, f.readSource.ReadExtentData); err != nil { - log.LogErrorf("start CacheEngine failed: %v", err) + if f.cacheEngine, err = cachengine.NewCacheEngine(f.tmpPath, int64(f.total), + 0, _defaultLRUCapacity, time.Hour, ReadExtentData); err != nil { + log.LogErrorf("startCacheEngine failed:%v", err) return } f.cacheEngine.Start() @@ -203,33 +210,43 @@ func (f *FlashNode) startCacheEngine() (err error) { } func (f *FlashNode) initLimiter() { - f.nodeLimit = 0 - f.nodeLimiter = rate.NewLimiter(rate.Inf, DefaultBurst) - f.volLimitMap = make(map[string]uint64) - f.volLimiterMap = make(map[string]*rate.Limiter) + f.readLimiter = rate.NewLimiter(rate.Limit(f.readRps), 2*f.readRps) } -func (f *FlashNode) register() (err error) { - // TODO: xxx - // var ( - // regInfo = &master.RegNodeInfoReq{ - // Role: proto.RoleFlash, - // ZoneName: f.zoneName, - // Version: NodeLatestVersion, - // SrvPort: f.listen, - // } - // regRsp *proto.RegNodeRsp - // ) +func (f *FlashNode) register() error { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for { + log.LogInfo("to register flashnode") + for { + ci, err := f.mc.AdminAPI().GetClusterInfo() + if err != nil { + log.LogErrorf("action[register] cannot get ip from master err(%v)", err) + break + } - // if regRsp, err = masterClient.RegNodeInfo(proto.AuthFilePath, regInfo); err != nil { - // return - // } + localIP := ci.Ip + if !util.IsIPV4(localIP) { + log.LogErrorf("action[register] got an invalid local ip(%s) from master", localIP) + break + } + f.clusterID = ci.Cluster + f.localAddr = fmt.Sprintf("%s:%v", localIP, f.listen) - // f.nodeID= regRsp.Id - // ipAddr := strings.Split(regRsp.Addr, ":")[0] - // f.localAddr = ipAddr - // f.clusterID = regRsp.Cluster + nodeID, err := f.mc.NodeAPI().AddFlashNode(f.localAddr, f.zoneName, "") + if err != nil { + log.LogErrorf("action[register] cannot register flashnode to master err(%v).", err) + break + } + f.nodeID = nodeID + log.LogInfof("action[register] flashnode(%d) cluster(%s) localAddr(%s)", f.nodeID, f.clusterID, f.localAddr) + return nil + } - f.nodeID = 0 - return + select { + case <-ticker.C: + case <-f.stopCh: + return fmt.Errorf("stopped") + } + } } diff --git a/flashnode/flashnode_op.go b/flashnode/flashnode_op.go index 5109140f7..7f4d69dd1 100644 --- a/flashnode/flashnode_op.go +++ b/flashnode/flashnode_op.go @@ -18,8 +18,6 @@ import ( "context" "fmt" "net" - "runtime/debug" - "strings" "sync" "time" @@ -31,190 +29,171 @@ import ( "github.com/cubefs/cubefs/util/log" ) -func (f *FlashNode) preHandle(conn net.Conn, p *Packet) error { - if p.Opcode == proto.OpCachePrepare || p.Opcode == proto.OpCacheRead { - if f.cacheEngine == nil { - return errors.New("cache engine not started") - } - } - // request rate limit for entire flash node - if !(p.Opcode == proto.OpCacheRead && f.nodeLimit != 0) { - return nil - } - if !f.nodeLimiter.Allow() { - err := errors.NewErrorf("flashnode request is limited(%d)", f.nodeLimit) +func (f *FlashNode) preHandle(conn net.Conn, p *proto.Packet) error { + if p.Opcode == proto.OpCacheRead && !f.readLimiter.Allow() { metric := exporter.NewTPCnt("NodeReqLimit") - log.LogWarnf("action[preHandle] %s, remote address:%s", err.Error(), conn.RemoteAddr()) metric.Set(nil) + err := errors.NewErrorf("%s", "flashnode read request was been limited") + log.LogWarnf("action[preHandle] %s, remote address:%s", err.Error(), conn.RemoteAddr()) return err } return nil } -// handlePacket handles the tcp packet operations. -func (f *FlashNode) handlePacket(conn net.Conn, p *Packet, remoteAddr string) (err error) { +func (f *FlashNode) handlePacket(conn net.Conn, p *proto.Packet) (err error) { metric := exporter.NewTPCnt(p.GetOpMsg()) - defer func() { - metric.Set(err) - }() + defer func() { metric.Set(err) }() + switch p.Opcode { case proto.OpFlashNodeHeartbeat: - err = f.opFlashNodeHeartbeat(conn, p, remoteAddr) + err = f.opFlashNodeHeartbeat(conn, p) case proto.OpCacheRead: - err = f.opCacheRead(conn, p, remoteAddr) + err = f.opCacheRead(conn, p) case proto.OpCachePrepare: - err = f.opPrepare(conn, p, remoteAddr) + err = f.opPrepare(conn, p) default: err = fmt.Errorf("unknown Opcode:%d", p.Opcode) } if err != nil { - err = errors.NewErrorf("%s [%s] req: %d - %s", remoteAddr, p.GetOpMsg(), p.GetReqID(), err.Error()) + err = errors.NewErrorf("%s [%s] req: %d - %s", conn.RemoteAddr().String(), + p.GetOpMsg(), p.GetReqID(), err.Error()) } return } -func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *Packet, remoteAddr string) (err error) { +func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *proto.Packet) (err error) { p.PacketOkReply() if err = p.WriteToConn(conn); err != nil { log.LogErrorf("ack master response: %s", err.Error()) return err } - log.LogInfof("%s [opMasterHeartbeat] ", remoteAddr) + log.LogInfof("[opMasterHeartbeat] master:%s", conn.RemoteAddr().String()) return } -func (f *FlashNode) opCacheRead(conn net.Conn, p *Packet, remoteAddr string) (err error) { - var ( - block *cachengine.CacheBlock - req *proto.CacheReadRequest - volume string - ) - +func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) { + var volume string defer func() { if err != nil { logContent := fmt.Sprintf("action[opCacheRead] volume:[%v], logMsg:%v.", volume, - p.LogMessage(p.GetOpMsg(), remoteAddr, p.StartT, err)) - log.LogErrorf(logContent) + p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err)) + log.LogError(logContent) p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) - _ = respondToClient(conn, p) + p.WriteToConn(conn) } }() + ctx, ctxCancel := context.WithTimeout(context.Background(), proto.ReadCacheTimeout*time.Second*2) defer ctxCancel() - if req, err = UnMarshalPacketToCacheRead(p); err != nil { + + req := new(proto.CacheReadRequest) + if err = p.UnmarshalDataPb(req); err != nil { return } volume = req.CacheRequest.Volume - if !f.volLimitAllow(volume) { - err = errors.NewErrorf("volume(%s) request is limited(%d)", volume, f.volLimitMap[volume]) - log.LogWarnf("action[preHandle] %s, remote address:%s", err.Error(), remoteAddr) - metric := exporter.NewTPCnt("VolReqLimit") - p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) - _ = respondToClient(conn, p) - err = nil - metric.Set(nil) - return - } - if block, err = f.cacheEngine.GetCacheBlockForRead(volume, req.CacheRequest.Inode, req.CacheRequest.FixedFileOffset, req.CacheRequest.Version, req.Size_); err != nil { - if block, err = f.cacheEngine.CreateBlock(req.CacheRequest); err != nil { + + cr := req.CacheRequest + block, err := f.cacheEngine.GetCacheBlockForRead(volume, cr.Inode, cr.FixedFileOffset, cr.Version, req.Size_) + if err != nil { + if block, err = f.cacheEngine.CreateBlock(cr); err != nil { return err } - go block.InitOnce(f.cacheEngine, req.CacheRequest.Sources) - } - if err = f.doStreamReadRequest(ctx, conn, req, p, block); err != nil { - return + go block.InitOnce(f.cacheEngine, cr.Sources) } + err = f.doStreamReadRequest(ctx, conn, req, p, block) return } -func (f *FlashNode) doStreamReadRequest(ctx context.Context, conn net.Conn, req *proto.CacheReadRequest, p *Packet, block *cachengine.CacheBlock) (err error) { +func (f *FlashNode) doStreamReadRequest(ctx context.Context, conn net.Conn, req *proto.CacheReadRequest, p *proto.Packet, block *cachengine.CacheBlock) (err error) { + const action = "action[doStreamReadRequest]" needReplySize := uint32(req.Size_) offset := int64(req.Offset) defer func() { if err != nil { - err = fmt.Errorf("action[doStreamReadRequest] cache block(%v) err:%v", block.String(), err) + err = fmt.Errorf("%s cache block(%v) err:%v", action, block.String(), err) } }() - for { - if needReplySize <= 0 { - break - } + + for needReplySize > 0 { err = nil - reply := NewCacheReply() + reply := proto.NewPacket() reply.ReqID = p.ReqID reply.StartT = p.StartT + currReadSize := uint32(util.Min(int(needReplySize), util.ReadBlockSize)) - if currReadSize == util.ReadBlockSize { - reply.Data, _ = proto.Buffers.Get(util.ReadBlockSize) - } else { - reply.Data = make([]byte, currReadSize) + + var bufOnce sync.Once + buf, bufErr := proto.Buffers.Get(util.ReadBlockSize) + bufRelease := func() { + bufOnce.Do(func() { + if bufErr == nil { + proto.Buffers.Put(reply.Data[:util.ReadBlockSize]) + } + }) } + if bufErr != nil { + buf = make([]byte, currReadSize) + } + reply.Data = buf[:currReadSize] reply.ExtentOffset = offset p.Size = currReadSize p.ExtentOffset = offset - err = func() error { - var storeErr error - reply.CRC, storeErr = block.Read(ctx, reply.Data[0:currReadSize], offset, int64(currReadSize)) - return storeErr - }() - p.CRC = reply.CRC + reply.CRC, err = block.Read(ctx, reply.Data[:], offset, int64(currReadSize)) if err != nil { - if currReadSize == util.ReadBlockSize { - proto.Buffers.Put(reply.Data) - } + bufRelease() return } + p.CRC = reply.CRC + reply.Size = currReadSize reply.ResultCode = proto.OpOk reply.Opcode = p.Opcode p.ResultCode = proto.OpOk - - err = func() error { - return reply.WriteToConn(conn) - }() - if err != nil { - if currReadSize == util.ReadBlockSize { - proto.Buffers.Put(reply.Data) - } - logContent := fmt.Sprintf("action[doStreamReadRequest] volume:[%v] %v.", req.CacheRequest.Volume, + if err = reply.WriteToConn(conn); err != nil { + bufRelease() + log.LogErrorf("%s volume:[%v] %v.", action, req.CacheRequest.Volume, reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, err)) - log.LogErrorf(logContent) return } + needReplySize -= currReadSize offset += int64(currReadSize) - if currReadSize == util.ReadBlockSize { - proto.Buffers.Put(reply.Data) - } - logContent := fmt.Sprintf("action[doStreamReadRequest] ReqID[%v] volume:[%v] reply[%v] block[%v] .", p.ReqID, req.CacheRequest.Volume, + bufRelease() + log.LogInfof("%s ReqID[%v] volume:[%v] reply[%v] block[%v] .", action, p.ReqID, req.CacheRequest.Volume, reply.LogMessage(reply.GetOpMsg(), conn.RemoteAddr().String(), reply.StartT, err), block.String()) - log.LogReadf(logContent) } p.PacketOkReply() return } -func (f *FlashNode) opPrepare(conn net.Conn, p *Packet, remoteAddr string) (err error) { - var req *proto.CachePrepareRequest +func (f *FlashNode) opPrepare(conn net.Conn, p *proto.Packet) (err error) { + action := "action[opPrepare]" var volume string defer func() { if err != nil { - logContent := fmt.Sprintf("action[opPrepare] volume:[%v] %v.", volume, p.LogMessage(p.GetOpMsg(), remoteAddr, p.StartT, err)) - log.LogErrorf(logContent) + log.LogErrorf("%s volume:[%v] %v.", action, volume, + p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err)) } }() - reqID := p.ReqID - if req, err = UnMarshalPacketToCachePrepare(p); err != nil { + + req := new(proto.CachePrepareRequest) + if err = p.UnmarshalDataPb(req); err != nil { p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) - _ = respondToClient(conn, p) - return err + if e := p.WriteToConn(conn); e != nil { + log.LogErrorf("%s write to conn %v.", action, e) + } + return } - p.PacketOkReply() - _ = respondToClient(conn, p) volume = req.CacheRequest.Volume - if err = f.cacheEngine.PrepareCache(reqID, req.CacheRequest); err != nil { + + p.PacketOkReply() + if e := p.WriteToConn(conn); e != nil { + log.LogErrorf("%s write to conn volume:%s %v.", action, volume, e) + } + + if err = f.cacheEngine.PrepareCache(p.ReqID, req.CacheRequest); err != nil { return err } if len(req.FlashNodes) > 0 { @@ -229,75 +208,51 @@ func (f *FlashNode) dispatchRequestToFollowers(request *proto.CachePrepareReques FlashNodes: make([]string, 0), } wg := sync.WaitGroup{} - for _, n := range request.FlashNodes { - if strings.Split(n, ":")[0] == f.localAddr { + for idx, addr := range request.FlashNodes { + if addr == f.localAddr { continue } wg.Add(1) + log.LogDebugf("dispatchRequestToFollowers: try to prepare on addr:%s (%d/%d)", addr, idx, len(request.FlashNodes)) go func(addr string) { defer wg.Done() - if err := f.sendPrepareRequest(req, addr); err != nil { + if err := f.sendPrepareRequest(addr, req); err != nil { log.LogErrorf("dispatchRequestToFollowers: failed to distribute request to addr(%v) err(%v)", addr, err) } - }(n) + }(addr) } wg.Wait() } -func (f *FlashNode) sendPrepareRequest(req *proto.CachePrepareRequest, target string) (err error) { - var conn *net.TCPConn - conn, err = f.connPool.GetConnect(target) +func (f *FlashNode) sendPrepareRequest(addr string, req *proto.CachePrepareRequest) (err error) { + action := "action[sendPrepareRequest]" + conn, err := f.connPool.GetConnect(addr) if err != nil { return err } defer func() { f.connPool.PutConnect(conn, err != nil) }() + log.LogDebugf("%s to addr:%s request:%v", action, addr, req) - // TODO: xxx - // log.LogDebugf("action[sendPrepareRequest] request:%v", req) - // followerPacket, err := MarshalCachePrepareRequestToPacket(req) - // if err != nil { - // return err - // } - // if err = followerPacket.WriteToConnNs(conn, CacheReqWriteTimeoutMilliSec*1e6); err != nil { - // log.LogWarnf("action[sendPrepareRequest]: failed to write to addr(%v) err(%v)", target, err) - // return - // } - // replyPacket := common.NewCacheReply(followerPacket.Ctx()) - // // replyPacket = repl.NewReplProtocol(followerPacket.Ctx()) - // if err = replyPacket.ReadFromConnNs(conn, CacheReqReadTimeoutMilliSec*1e6); err != nil { - // log.LogWarnf("action[sendPrepareRequest]: failed to ReadFromConn, replyPacket(%v), fg host(%v), err(%v)", replyPacket, target, err) - // return - // } - // if replyPacket.ResultCode != proto.OpOk { - // log.LogWarnf("action[sendPrepareRequest]: ResultCode NOK, replyPacket(%v), fg host(%v), ResultCode(%v)", replyPacket, target, replyPacket.ResultCode) - // err = fmt.Errorf("ResultCode NOK (%v)", replyPacket.ResultCode) - // return - // } - return -} - -// Reply data through tcp connection to the client. -func respondToClient(conn net.Conn, p *Packet) (err error) { - // Handle panic - defer func() { - if r := recover(); r != nil { - log.LogErrorf("respondToClient: panic occurred: %v\n%v", r, string(debug.Stack())) - switch data := r.(type) { - case error: - err = data - default: - err = errors.New(data.(string)) - } - } - }() - - // process data and send reply though specified tcp connection. - err = p.WriteToConn(conn) - if err != nil { - log.LogErrorf("response to client[%s], request[%s], response packet[%s]", - err.Error(), p.GetOpMsg(), p.GetResultMsg()) + followerPacket := proto.NewPacketReqID() + followerPacket.Opcode = proto.OpCachePrepare + if err = followerPacket.MarshalDataPb(req); err != nil { + log.LogWarnf("%s failed to MarshalDataPb (%+v) err(%v)", action, followerPacket, err) + return err } - return + if err = followerPacket.WriteToNoDeadLineConn(conn); err != nil { + log.LogWarnf("%s failed to write to addr(%s) err(%v)", action, addr, err) + return err + } + reply := proto.NewPacket() + if err = reply.ReadFromConn(conn, 30); err != nil { + log.LogWarnf("%s read reply(%v) from addr(%s) err(%v)", action, reply, addr, err) + return err + } + 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 nil } diff --git a/flashnode/nodeinfo.go b/flashnode/nodeinfo.go deleted file mode 100644 index 9194f807f..000000000 --- a/flashnode/nodeinfo.go +++ /dev/null @@ -1,101 +0,0 @@ -// Copyright 2018 The CubeFS Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or -// implied. See the License for the specific language governing -// permissions and limitations under the License. - -package flashnode - -import ( - "time" - - "github.com/cubefs/cubefs/util/log" -) - -func (f *FlashNode) startUpdateScheduler() { - limitTicker := time.NewTicker(UpdateRateLimitInfoInterval) - defer limitTicker.Stop() - for { - f.updateFlashNodeRateLimit() - select { - case <-f.stopCh: - log.LogInfo("flashNode startUpdateScheduler goroutine stopped") - return - case <-limitTicker.C: - } - } -} - -func (f *FlashNode) updateFlashNodeRateLimit() { - // limitInfo, err := f.mc.AdminAPI().GetLimitInfo("") - // if err != nil { - // log.LogWarnf("[updateRateLimitInfo] get limit info err: %s", err.Error()) - // return - // } - // f.updateZoneLimiter(limitInfo) - // f.updateZoneVolLimiter(limitInfo) -} - -// // updateZoneLimiter update limiter for zone -// func (f *FlashNode) updateZoneLimiter(limitInfo *proto.LimitInfo) { -// r, ok := limitInfo.FlashNodeLimitMap[f.zoneName] -// if !ok { -// r, ok = limitInfo.FlashNodeLimitMap[""] -// } -// if !ok { -// f.nodeLimit = 0 -// f.nodeLimiter.SetLimit(rate.Inf) -// return -// } -// if r == 0 && f.nodeLimit == 0 { -// return -// } -// if f.nodeLimit != r { -// f.nodeLimit = r -// l := rate.Inf -// if r > 0 { -// l = rate.Limit(r) -// } -// f.nodeLimiter.SetLimit(l) -// } -// } - -// func (f *FlashNode) updateZoneVolLimiter(limitInfo *proto.LimitInfo) { -// volLimitMap, ok := limitInfo.FlashNodeVolLimitMap[f.zoneName] -// if !ok { -// volLimitMap, ok = limitInfo.FlashNodeVolLimitMap[""] -// } -// if !ok { -// f.volLimitMap = make(map[string]uint64) -// f.volLimiterMap = make(map[string]*rate.Limiter) -// return -// } -// if !reflect.DeepEqual(f.volLimitMap, volLimitMap) { -// f.volLimitMap = volLimitMap -// tmpVolLimiterMap := make(map[string]*rate.Limiter) -// for vol, r := range volLimitMap { -// l := rate.Limit(r) -// tmpVolLimiterMap[vol] = rate.NewLimiter(l, DefaultBurst) -// } -// f.volLimiterMap = tmpVolLimiterMap -// } -// } - -func (f *FlashNode) volLimitAllow(vol string) bool { - if len(f.volLimitMap) == 0 { - return true - } - limiter, ok := f.volLimiterMap[vol] - if !ok { - return true - } - return limiter.Allow() -} diff --git a/flashnode/packet.go b/flashnode/packet.go deleted file mode 100644 index 8ad9439c4..000000000 --- a/flashnode/packet.go +++ /dev/null @@ -1,119 +0,0 @@ -// Copyright 2018 The CubeFS Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or -// implied. See the License for the specific language governing -// permissions and limitations under the License. - -package flashnode - -import ( - "io" - "syscall" - "time" - - "github.com/cubefs/cubefs/proto" - "github.com/cubefs/cubefs/util" - "github.com/cubefs/cubefs/util/errors" - "github.com/cubefs/cubefs/util/log" -) - -type Packet struct { - proto.Packet -} - -func (p *Packet) ReadFromReader(reader io.Reader) (err error) { - header, err := proto.Buffers.Get(util.PacketHeaderSize) - if err != nil { - header = make([]byte, util.PacketHeaderSize) - } - defer proto.Buffers.Put(header) - var n int - if n, err = io.ReadFull(reader, header); err != nil { - return - } - if n != util.PacketHeaderSize { - return syscall.EBADMSG - } - if err = p.UnmarshalHeader(header); err != nil { - return - } - - if p.ArgLen > 0 { - p.Arg = make([]byte, int(p.ArgLen)) - if _, err = io.ReadFull(reader, p.Arg[:int(p.ArgLen)]); err != nil { - return err - } - } - - size := p.Size - if (p.Opcode == proto.OpRead || p.Opcode == proto.OpStreamRead || p.Opcode == proto.OpExtentRepairRead || p.Opcode == proto.OpStreamFollowerRead) && p.ResultCode == proto.OpInitResultCode { - size = 0 - } - var buffErr error - if p.Data, buffErr = proto.Buffers.Get(int(size)); buffErr != nil { - p.Data = make([]byte, size) - } - if n, err = io.ReadFull(reader, p.Data[:size]); err != nil { - return err - } - if n != int(size) { - return syscall.EBADMSG - } - return nil -} - -func NewPacket() *Packet { - p := &Packet{} - p.Magic = proto.ProtoMagic - p.StartT = time.Now().UnixNano() - return p -} - -func NewCacheReply() *Packet { - p := new(Packet) - p.Magic = proto.ProtoMagic - return p -} - -func NewPreparePacket() *Packet { - p := new(Packet) - p.Magic = proto.ProtoMagic - p.ReqID = proto.GenerateRequestID() - p.Opcode = proto.OpCachePrepare - return p -} - -func MarshalCachePrepareRequestToPacket(req *proto.CachePrepareRequest) (packet *Packet, err error) { - packet = NewPreparePacket() - if err = packet.MarshalDataPb(req); err != nil { - log.LogWarnf("MarshalCachePrepareRequestToPacket: failed to MarshalDataPb (%+v). err(%v)", packet, err) - return - } - return packet, err -} - -func UnMarshalPacketToCachePrepare(p *Packet) (req *proto.CachePrepareRequest, err error) { - req = &proto.CachePrepareRequest{} - if err = p.UnmarshalDataPb(req); err != nil { - err = errors.NewErrorf("UnmarshalDataPb data length[%v] err[%v]", len(p.Data), err) - return - } - return -} - -func UnMarshalPacketToCacheRead(p *Packet) (req *proto.CacheReadRequest, err error) { - req = &proto.CacheReadRequest{} - if err = p.UnmarshalDataPb(req); err != nil { - err = errors.NewErrorf("UnmarshalDataPb data length[%v] err[%v]", len(p.Data), err) - return - } - return -} diff --git a/flashnode/read_source.go b/flashnode/read_source.go deleted file mode 100644 index 9c20c37eb..000000000 --- a/flashnode/read_source.go +++ /dev/null @@ -1,208 +0,0 @@ -// Copyright 2018 The CubeFS Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or -// implied. See the License for the specific language governing -// permissions and limitations under the License. - -package flashnode - -import ( - "fmt" - "hash/crc32" - "net" - "strings" - "time" - - "github.com/cubefs/cubefs/proto" - "github.com/cubefs/cubefs/util" - "github.com/cubefs/cubefs/util/errors" - "github.com/cubefs/cubefs/util/exporter" - "github.com/cubefs/cubefs/util/log" -) - -var extentReaderConnPool *util.ConnectPool - -type ReadSource struct{} - -func NewReadSource() *ReadSource { - extentReaderConnPool = util.NewConnectPoolWithTimeoutAndCap(0, 10, IdleConnTimeoutData, 1) - return &ReadSource{} -} - -func (reader *ReadSource) ReadExtentData(source *proto.DataSource, afterReadFunc func([]byte, int64) error) (readBytes int, err error) { - reqPacket := newReadPacket(&proto.ExtentKey{ - PartitionId: source.PartitionID, - ExtentId: source.ExtentID, - }, int(source.ExtentOffset), int(source.Size_), source.FileOffset, true) - - readBytes, err = extentReadWithRetry(reqPacket, source.Hosts, afterReadFunc) - if err != nil { - log.LogErrorf("read error: err(%v)", err) - return readBytes, err - } - return -} - -func extentReadWithRetry(reqPacket *proto.Packet, hosts []string, afterReadFunc func([]byte, int64) error) (readBytes int, err error) { - errMap := make(map[string]error) - startTime := time.Now() - retryCount := 0 - for i := 0; i < ExtentReadMaxRetry; i++ { - retryCount++ - for _, addr := range hosts { - log.LogDebugf("extentReadWithRetry: try addr(%v) reqPacket(%v)", addr, reqPacket) - readBytes, err = sendReadCmdToDataPartition(addr, reqPacket, afterReadFunc) - if err == nil { - return - } - errMap[addr] = err - log.LogWarnf("extentReadWithRetry: try addr(%v) failed! reqPacket(%v) err(%v)", addr, reqPacket, err) - if strings.Contains(err.Error(), proto.ErrTmpfsNoSpace.Error()) { - break - } - } - if time.Since(startTime) > time.Duration(ExtentReadTimeoutSec*int64(time.Second)) { - log.LogWarnf("extentReadWithRetry: retry timeout req(%v) time(%v)", reqPacket, time.Since(startTime)) - break - } - log.LogWarnf("extentReadWithRetry: errMap(%v), reqPacket(%v), try the next round", errMap, reqPacket) - time.Sleep(ExtentReadSleepInterval) - } - err = errors.New(fmt.Sprintf("FollowerRead: failed %v times, reqPacket(%v) errMap(%v)", ExtentReadMaxRetry, reqPacket, errMap)) - return -} - -func sendReadCmdToDataPartition(addr string, reqPacket *proto.Packet, afterReadFunc func([]byte, int64) error) (readBytes int, err error) { - if addr == "" { - err = errors.New(fmt.Sprintf("sendReadCmdToDataPartition: failed, current address is null, reqPacket(%v)", reqPacket)) - return - } - var conn *net.TCPConn - defer func() { - extentReaderConnPool.PutConnect(conn, err != nil) - }() - if conn, err = extentReaderConnPool.GetConnect(addr); err != nil { - log.LogWarnf("sendToDataPartition: get connection to curr addr failed, addr(%v) reqPacket(%v) err(%v)", addr, reqPacket, err) - return - } - if err = sendToDataPartition(conn, reqPacket, addr); err != nil { - log.LogWarnf("sendReadCmdToDataPartition: send to curr addr failed, addr(%v) reqPacket(%v) err(%v)", addr, reqPacket, err) - return - } - if readBytes, err = getReadReply(conn, reqPacket, afterReadFunc); err != nil { - log.LogWarnf("sendReadCmdToDataPartition: getReply error and RETURN, addr(%v) reqPacket(%v) err(%v)", addr, reqPacket, err) - return - } - return -} - -func sendToDataPartition(conn *net.TCPConn, req *proto.Packet, addr string) (err error) { - log.LogDebugf("sendToDataPartition: send to addr(%v)", addr) - if req.Opcode == proto.OpStreamFollowerRead { - req.StartT = time.Now().UnixNano() - } - // TODO: xxx - // if err = req.WriteToConnNs(conn, ExtentReadTimeoutSec*int64(time.Second)); err != nil { - // log.LogWarnf("sendToDataPartition: failed to write to addr(%v) err(%v)", addr, err) - // return - // } - log.LogDebugf("sendToDataPartition exit: send to addr(%v) successfully", addr) - return -} - -func getReadReply(conn *net.TCPConn, reqPacket *proto.Packet, afterReadFunc func([]byte, int64) error) (readBytes int, err error) { - readBytes = 0 - tmpData := make([]byte, reqPacket.Size) - for readBytes < int(reqPacket.Size) { - replyPacket := newReplyPacket(reqPacket.ReqID, reqPacket.PartitionID, reqPacket.ExtentID) - bufSize := util.Min(util.ReadBlockSize, int(reqPacket.Size)-readBytes) - replyPacket.Data = tmpData[readBytes : readBytes+bufSize] - e := replyPacket.ReadFromConn(conn, ExtentReadTimeoutSec) - if e != nil { - log.LogWarnf("getReadReply: failed to read from connect, req(%v) readBytes(%v) err(%v)", reqPacket, readBytes, e) - // Upon receiving TryOtherAddrError, other hosts will be retried. - return readBytes, e - } - - e = checkReadReplyValid(reqPacket, replyPacket) - if e != nil { - // Don't change the error message, since the caller will - // check if it is NotLeaderErr. - return readBytes, e - } - if afterReadFunc != nil { - e = afterReadFunc(replyPacket.Data[:bufSize], int64(bufSize)) - if e != nil { - return readBytes, e - } - } - if replyPacket.Size != uint32(bufSize) { - exporter.Warning(fmt.Sprintf("action[getReadReply] reply size not valid, ReqID(%v) PartitionID(%v) Extent(%v) buffSize(%v) ReplySize(%v)", reqPacket.ReqID, reqPacket.PartitionID, reqPacket.ExtentID, bufSize, replyPacket.Size)) - } - readBytes += int(replyPacket.Size) - } - return readBytes, nil -} - -func checkReadReplyValid(request *proto.Packet, reply *proto.Packet) (err error) { - if reply.ResultCode != proto.OpOk { - err = errors.New(fmt.Sprintf("checkReadReplyValid: ResultCode(%v) NOK", reply.GetResultMsg())) - return - } - if !isValidReadReply(request, reply) { - err = errors.New(fmt.Sprintf("checkReadReplyValid: inconsistent req and reply, req(%v) reply(%v)", request, reply)) - return - } - expectCrc := crc32.ChecksumIEEE(reply.Data[:reply.Size]) - if reply.CRC != expectCrc { - err = errors.New(fmt.Sprintf("checkReadReplyValid: inconsistent CRC, expectCRC(%v) replyCRC(%v)", expectCrc, reply.CRC)) - return - } - return nil -} - -func isValidReadReply(p *proto.Packet, q *proto.Packet) bool { - if p.ReqID == q.ReqID && p.PartitionID == q.PartitionID && p.ExtentID == q.ExtentID { - return true - } - return false -} - -// newReadPacket returns a new read packet. -func newReadPacket(key *proto.ExtentKey, extentOffset, size int, fileOffset uint64, followerRead bool) *proto.Packet { - p := new(proto.Packet) - p.ExtentID = key.ExtentId - p.PartitionID = key.PartitionId - p.Magic = proto.ProtoMagic - p.ExtentOffset = int64(extentOffset) - p.Size = uint32(size) - if followerRead { - p.Opcode = proto.OpStreamFollowerRead - } else { - p.Opcode = proto.OpStreamRead - } - p.ExtentType = proto.NormalExtentType - p.ReqID = proto.GenerateRequestID() - p.RemainingFollowers = 0 - p.KernelOffset = fileOffset - return p -} - -// newReplyPacket returns a new reply packet. -func newReplyPacket(reqID int64, partitionID uint64, extentID uint64) *proto.Packet { - p := new(proto.Packet) - p.ReqID = reqID - p.PartitionID = partitionID - p.ExtentID = extentID - p.Magic = proto.ProtoMagic - p.ExtentType = proto.NormalExtentType - return p -} diff --git a/flashnode/api_handler.go b/flashnode/serve_http.go similarity index 78% rename from flashnode/api_handler.go rename to flashnode/serve_http.go index 7ea3668b4..15f76593d 100644 --- a/flashnode/api_handler.go +++ b/flashnode/serve_http.go @@ -25,22 +25,21 @@ import ( ) func (f *FlashNode) registerAPIHandler() { - http.HandleFunc("/stat", f.getCacheStatHandler) - http.HandleFunc("/evictVol", f.evictVolumeCacheHandler) - http.HandleFunc("/evictAll", f.evictAllCacheHandler) + http.HandleFunc("/stat", f.handleStat) + http.HandleFunc("/evictVol", f.handleEvictVolume) + http.HandleFunc("/evictAll", f.handleEvictAll) } -func (f *FlashNode) getCacheStatHandler(w http.ResponseWriter, r *http.Request) { +func (f *FlashNode) handleStat(w http.ResponseWriter, r *http.Request) { replyOK(w, r, proto.FlashNodeStat{ - NodeLimit: f.nodeLimit, - VolLimit: f.volLimitMap, + NodeLimit: uint64(f.readLimiter.Limit()), CacheStatus: f.cacheEngine.Status(), }) } -func (f *FlashNode) evictVolumeCacheHandler(w http.ResponseWriter, r *http.Request) { +func (f *FlashNode) handleEvictVolume(w http.ResponseWriter, r *http.Request) { r.ParseForm() - volume := r.FormValue(volumePara) + volume := r.FormValue("volume") if volume == "" { replyErr(w, r, http.StatusBadRequest, "volume name can not be empty", nil) return @@ -48,7 +47,7 @@ func (f *FlashNode) evictVolumeCacheHandler(w http.ResponseWriter, r *http.Reque replyOK(w, r, f.cacheEngine.EvictCacheByVolume(volume)) } -func (f *FlashNode) evictAllCacheHandler(w http.ResponseWriter, r *http.Request) { +func (f *FlashNode) handleEvictAll(w http.ResponseWriter, r *http.Request) { f.cacheEngine.EvictCacheAll() replyOK(w, r, nil) } diff --git a/flashnode/server.go b/flashnode/serve_tcp.go similarity index 63% rename from flashnode/server.go rename to flashnode/serve_tcp.go index 545caf965..c3f6d9cd9 100644 --- a/flashnode/server.go +++ b/flashnode/serve_tcp.go @@ -15,7 +15,6 @@ package flashnode import ( - "bufio" "fmt" "io" "net" @@ -33,30 +32,30 @@ func NewServer() *FlashNode { // StartTcpService binds and listens to the specified port. func (f *FlashNode) startTcpServer() (err error) { - f.netListener, err = net.Listen("tcp", ":"+f.listen) + f.tcpListener, err = net.Listen("tcp", ":"+f.listen) if err != nil { return } go func() { - defer f.netListener.Close() + defer f.tcpListener.Close() var latestAlarm time.Time for { - conn, err1 := f.netListener.Accept() + conn, err1 := f.tcpListener.Accept() select { case <-f.stopCh: - log.LogWarnf("http server stopped") + log.LogWarn("flashnode tcp server stopped") return default: } if err1 != nil { - log.LogErrorf("action[startTcpServer] failed to accept, err:%s", err.Error()) + log.LogErrorf("action[tcpAccept] failed err:%s", err.Error()) // Alarm only once at 1 minute if time.Since(latestAlarm) > time.Minute { warnMsg := fmt.Sprintf("SERVER ACCEPT CONNECTION FAILED!\n"+ - "Failed on accept connection from %v and will retry after 10s.\n"+ - "Error message: %s", f.netListener.Addr(), err.Error()) + "Failed on accept connection from %v and will retry after 1s.\n"+ + "Error message: %s", f.tcpListener.Addr(), err.Error()) exporter.Warning(warnMsg) latestAlarm = time.Now() } @@ -66,45 +65,44 @@ func (f *FlashNode) startTcpServer() (err error) { go f.serveConn(conn) } }() - log.LogInfof("start tcp server over...") + log.LogInfo("started tcp server") return } func (f *FlashNode) stopServer() { - f.netListener.Close() + f.tcpListener.Close() } -// serveConn Read data from the specified tcp connection until the connection is closed by the remote or the tcp service is down. func (f *FlashNode) serveConn(conn net.Conn) { defer conn.Close() c := conn.(*net.TCPConn) - _ = c.SetKeepAlive(true) // Ignore error - _ = c.SetNoDelay(true) // Ignore error + c.SetKeepAlive(true) + c.SetNoDelay(true) + remoteAddr := conn.RemoteAddr().String() - connReader := bufio.NewReader(c) for { + c.SetReadDeadline(time.Now().Add(time.Second * _tcpServerTimeoutSec)) select { case <-f.stopCh: return default: } - p := NewPacket() - _ = c.SetReadDeadline(time.Now().Add(time.Second * ServerTimeout)) - if err := p.ReadFromReader(connReader); err != nil { + + p := proto.NewPacketReqID() + if err := p.ReadFromConn(c, -1); err != nil { if err != io.EOF { - log.LogError("serve FlashNode: ", err.Error()) + log.LogError("flashnode read from remote", err.Error()) } return } if err := f.preHandle(conn, p); err != nil { - logContent := fmt.Sprintf("preHandle %v.", p.LogMessage(p.GetOpMsg(), remoteAddr, p.StartT, err)) - log.LogWarnf(logContent) + log.LogWarn("preHandle", p.LogMessage(p.GetOpMsg(), remoteAddr, p.StartT, err)) p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) - _ = respondToClient(conn, p) + p.WriteToConn(conn) continue } - if err := f.handlePacket(conn, p, remoteAddr); err != nil { - log.LogErrorf("serve handlePacket fail: %v", err) + if err := f.handlePacket(conn, p); err != nil { + log.LogError("handlePacket", err) } } }