mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
fix(flashnode): fixed the infinite loop issue in GetObject.
close:#1000151055 Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe
parent
037348725c
commit
c5e2315c7d
@ -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()
|
||||
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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()
|
||||
|
||||
|
||||
@ -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() {
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}()
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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)
|
||||
// }
|
||||
// }
|
||||
|
||||
@ -306,3 +306,6 @@ func (fs *LocalFS) LoadTestFiles() error {
|
||||
// TODO: maybe to much handle
|
||||
return nil
|
||||
}
|
||||
|
||||
func (fs *LocalFS) Stop() {
|
||||
}
|
||||
|
||||
@ -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()
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -44,5 +44,4 @@ func main() {
|
||||
os.Exit(1)
|
||||
}
|
||||
log.LogFlush()
|
||||
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
