refactor(meta): support batch persist inode atime. #22701675

Signed-off-by: Victor1319 <zengxuewei@oppo.com>

master/topology.go

@@ -1466,6 +1466,7 @@ func (t *topology) allocZonesForNode(rsMgr *rsManager, zoneNumNeed, replicaNum i
 	if excludeZone == nil {
 		excludeZone = make([]string, 0)
 	}
+
 	candidateZones := make([]*Zone, 0)
 	demandWriteNodesCntPerZone := calculateDemandWriteNodes(zoneNumNeed, replicaNum, len(specialZones) > 1)
 

metanode/const.go

@@ -212,6 +212,13 @@ const (
 	opFSMInnerCleanMigrationExtentKeyAfterError   = 93
 )
 
+// new inode opCode
+const (
+	opFSMBatchSyncInodeATime = 11000
+)
+
+var exporterKey string
+
 var (
 	ErrNoLeader   = errors.New("no leader")
 	ErrNotALeader = errors.New("not a leader")
@@ -262,6 +269,7 @@ const (
 	DefaultNameResolveInterval         = 1 // minutes
 	DefaultRaftNumOfLogsToRetain       = 20000 * 2
 	DefaultCreateBlobClientIntervalSec = 30
+	defaultSyncInodeAtimeCnt           = 102400
 )
 
 const (

metanode/partition.go

@@ -595,6 +595,7 @@ type metaPartition struct {
 	statByMigrateStorageClass []*proto.StatOfStorageClass
 	fmList                    *forbiddenMigrationList
 	volStorageClass           uint32
+	syncAtimeCh               chan uint64
 }
 
 func (mp *metaPartition) IsForbidden() bool {
@@ -1033,6 +1034,7 @@ func NewMetaPartition(conf *MetaPartitionConfig, manager *metadataManager) MetaP
 		storeChan:     make(chan *storeMsg, 100),
 		freeList:      newFreeList(),
 		extDelCh:      make(chan []proto.ExtentKey, defaultDelExtentsCnt),
+		syncAtimeCh:   make(chan uint64, defaultSyncInodeAtimeCnt),
 		extReset:      make(chan struct{}),
 		vol:           NewVol(),
 		manager:       manager,
@@ -1050,6 +1052,7 @@ func NewMetaPartition(conf *MetaPartitionConfig, manager *metadataManager) MetaP
 		mp.config.ForbidWriteOpOfProtoVer0 = manager.isVolForbidWriteOpOfProtoVer0(mp.config.VolName)
 	}
 	mp.txProcessor = NewTransactionProcessor(mp)
+	go mp.batchSyncInodeAtime()
 	return mp
 }
 

metanode/partition_fsm.go

@@ -528,6 +528,8 @@ func (mp *metaPartition) Apply(command []byte, index uint64) (resp interface{},
 			return
 		}
 		resp = mp.fsmSyncInodeAccessTime(ino)
+	case opFSMBatchSyncInodeATime:
+		resp = mp.fsmBatchSyncInodeAccessTime(msg.V)
 	}
 	return
 }

metanode/partition_fsmop_inode.go

@@ -171,12 +171,12 @@ func (mp *metaPartition) getInodeTopLayer(ino *Inode) (resp *InodeResponse) {
 	}
 	i := item.(*Inode)
 	// ctime := timeutil.GetCurrentTimeUnix()
-	/*
-	 * FIXME: not protected by lock yet, since nothing is depending on atime.
-	 * Shall add inode lock in the future.
-	 */
+	// /*
+	//  * FIXME: not protected by lock yet, since nothing is depending on atime.
+	//  * Shall add inode lock in the future.
+	//  */
 	// if ctime > i.AccessTime {
-	//	i.AccessTime = ctime
+	// 	i.AccessTime = ctime
 	// }
 
 	resp.Msg = i
@@ -1033,6 +1033,27 @@ func (mp *metaPartition) fsmSyncInodeAccessTime(ino *Inode) (status uint8) {
 	return
 }
 
+func (mp *metaPartition) fsmBatchSyncInodeAccessTime(bufSlice []byte) (status uint8) {
+	status = proto.OpOk
+	start := time.Now()
+	inode := NewInode(0, 0)
+	for idx := 0; idx+8 <= len(bufSlice); idx += 8 {
+		inode.Inode = binary.BigEndian.Uint64(bufSlice[idx : idx+8])
+		item := mp.inodeTree.CopyGet(inode)
+		if item == nil {
+			continue
+		}
+
+		i := item.(*Inode)
+		i.AccessTime = timeutil.GetCurrentTimeUnix()
+		log.LogDebugf("fsmBatchSyncInodeAccessTime: mp(%d) inode [%v] AccessTime update success.",
+			mp.config.PartitionId, i.Inode)
+	}
+
+	log.LogDebugf("fsmBatchSyncInodeAccessTime: batch inode accessTime finish. mp(%d), cnt(%d), cost(%d)us", mp.config.PartitionId, len(bufSlice)/8, time.Since(start).Milliseconds())
+	return
+}
+
 func (mp *metaPartition) internalFreeForbiddenMigrationInode(val []byte) (err error) {
 	if len(val) == 0 {
 		return

metanode/partition_op_extent.go

@@ -15,9 +15,9 @@
 package metanode
 
 import (
+	"encoding/binary"
 	"encoding/json"
 	"fmt"
-	"net"
 	"os"
 	"sort"
 	"sync/atomic"
@@ -105,7 +105,7 @@ func (mp *metaPartition) ExtentAppendWithCheck(req *proto.AppendExtentKeyWithChe
 		p.PacketErrorWithBody(status, reply)
 		return
 	}
-	if req.IsCache == true && req.IsMigration == true {
+	if req.IsCache && req.IsMigration {
 		log.LogErrorf("ExtentAppendWithCheck parameter IsCache and IsMigration can not be ture at the same time")
 		err = errors.New("ExtentAppendWithCheck parameter IsCache and IsMigration can not be ture at the same time")
 		reply := []byte(err.Error())
@@ -806,64 +806,126 @@ func (mp *metaPartition) persistInodeAccessTime(inode uint64, p *Packet) {
 	if !mp.enablePersistAccessTime {
 		return
 	}
+
 	i := NewInode(inode, 0)
 	item := mp.inodeTree.Get(i)
 	if item == nil {
 		log.LogWarnf("persistInodeAccessTime inode %v is not found", inode)
 		return
 	}
+
 	ino := item.(*Inode)
 	ctime := timeutil.GetCurrentTimeUnix()
-	at := time.Unix(ino.AccessTime, 0)
-	if !(ctime > ino.AccessTime && time.Now().Sub(at) > mp.GetAccessTimeValidInterval()*time.Second) {
-		log.LogDebugf("%v %v %v", ctime > ino.AccessTime,
-			time.Now().Sub(at) > mp.GetAccessTimeValidInterval(), mp.GetAccessTimeValidInterval())
+	atime := ino.AccessTime
+	interval := mp.GetAccessTimeValidInterval()
+
+	if ctime <= atime || ctime-atime < int64(interval) {
+		log.LogDebugf("persistInodeAccessTime: no need to persit atime, ino %d, ctime %d, atime %d, interval %d",
+			inode, ctime, atime, interval)
 		return
 	}
-	var (
-		err   error
-		val   []byte
-		mConn *net.TCPConn
-		m     = mp.manager
-	)
+
 	// always update local AccessTime
 	ino.AccessTime = ctime
-	log.LogDebugf("persistInodeAccessTime ino(%v) persist at to %v", ino.Inode, at)
-	if leaderAddr, ok := mp.IsLeader(); ok {
-		// sync AccessTime to followers
-		val, err = ino.Marshal()
+	log.LogDebugf("persistInodeAccessTime ino(%v) persist at to %v", ino.Inode, atime)
+
+	leaderAddr, ok := mp.IsLeader()
+	if ok {
+		retry := 0
+		// sync AccessTime to followers, retry 3 times
+		for retry <= 3 {
+			select {
+			case mp.syncAtimeCh <- ino.Inode:
+				return
+			default:
+				log.LogWarnf("persistInodeAccessTime: inode accessTime ch maybe blocked, mp %d, ino %d, retry %d",
+					mp.config.PartitionId, ino.Inode, retry)
+				retry++
+				time.Sleep(time.Second)
+			}
+		}
+		return
+	}
+
+	if leaderAddr == "" {
+		log.LogWarnf("persistInodeAccessTime ino(%v) sync InodeAccessTime failed for no leader", ino.Inode)
+		return
+	}
+	m := mp.manager
+	mConn, err := m.connPool.GetConnect(leaderAddr)
+	if err != nil {
+		log.LogWarnf("persistInodeAccessTime ino(%v) get connect to leader failed %v", ino.Inode, err)
+		m.connPool.PutConnect(mConn, ForceClosedConnect)
+		return
+	}
+	packetCopy := p.GetCopy()
+	if err = packetCopy.WriteToConn(mConn); err != nil {
+		log.LogWarnf("persistInodeAccessTime ino(%v) write to  connect failed %v", ino.Inode, err)
+		m.connPool.PutConnect(mConn, ForceClosedConnect)
+		return
+	}
+	if err = packetCopy.ReadFromConn(mConn, proto.ReadDeadlineTime); err != nil {
+		log.LogWarnf("persistInodeAccessTime ino(%v) read from  connect failed %v", ino.Inode, err)
+		m.connPool.PutConnect(mConn, ForceClosedConnect)
+		return
+	}
+	m.connPool.PutConnect(mConn, NoClosedConnect)
+	log.LogDebugf("persistInodeAccessTime ino(%v) notify leader to %v sync accessTime", ino.Inode, leaderAddr)
+}
+
+func (mp *metaPartition) batchSyncInodeAtime() {
+	batchCount := 1024
+	maxRetryCnt := 10
+	inodes := make([]uint64, 0, batchCount)
+	bufSlice := make([]byte, 0, 8*batchCount)
+	mpId := mp.config.PartitionId
+
+	defer func() {
+		close(mp.syncAtimeCh)
+	}()
+
+	for {
+		inodes = inodes[:0]
+		bufSlice = bufSlice[:0]
+		retry := 0
+
+		ino := <-mp.syncAtimeCh
+		log.LogDebugf("batchSyncInodeAtime: mp(%d) recive inode id %d", mpId, ino)
+		inodes = append(inodes, ino)
+
+		// try send msgs in 10ms at time or batchCount once, avoid too many req for one mp
+		for len(inodes) < batchCount && retry < maxRetryCnt {
+			select {
+			case ino := <-mp.syncAtimeCh:
+				inodes = append(inodes, ino)
+				log.LogDebugf("batchSyncInodeAtime: mp(%d) recive inode id %d, cnt(%d)", mpId, ino, len(inodes))
+			case <-mp.stopC:
+				log.LogWarnf("batchSyncInodeAtime: recive stop signal, exit, mp(%d), cnt(%d)", mpId, len(inodes))
+				return
+			default:
+				retry++
+				log.LogDebugf("batchSyncInodeAtime: mp(%d) no new inode req at now, cnt(%d) retry(%d)",
+					mpId, len(inodes), retry)
+				time.Sleep(time.Millisecond)
+			}
+		}
+
+		if log.EnableDebug() {
+			log.LogDebugf("batchSyncInodeAtime: mp(%d) try to batch sync inode atime, inods %+v, retry %d",
+				mpId, len(inodes), retry)
+		}
+
+		buf := make([]byte, 8)
+		for _, ino := range inodes {
+			binary.BigEndian.PutUint64(buf, ino)
+			bufSlice = append(bufSlice, buf...)
+		}
+
+		_, err := mp.submit(opFSMBatchSyncInodeATime, bufSlice)
 		if err != nil {
-			log.LogWarnf("persistInodeAccessTime marshal ino(%v) failed %v", ino.Inode, err)
-			return
+			log.LogWarnf("batchSyncInodeAtime: mp(%d) cnt (%d), opFSMBatchSyncInodeATime submit failed %v",
+				mpId, len(inodes), err)
+			continue
 		}
-		_, err = mp.submit(opFSMSyncInodeAccessTime, val)
-		if err != nil {
-			log.LogWarnf("persistInodeAccessTime ino(%v)  submit opFSMSyncInodeAccessTime failed %v", ino.Inode, err)
-			return
-		}
-	} else {
-		if leaderAddr == "" {
-			log.LogWarnf("persistInodeAccessTime ino(%v) sync InodeAccessTime failed for no leader", ino.Inode)
-			return
-		}
-		mConn, err = m.connPool.GetConnect(leaderAddr)
-		if err != nil {
-			log.LogWarnf("persistInodeAccessTime ino(%v) get connect to leader failed %v", ino.Inode, err)
-			m.connPool.PutConnect(mConn, ForceClosedConnect)
-			return
-		}
-		packetCopy := p.GetCopy()
-		if err = packetCopy.WriteToConn(mConn); err != nil {
-			log.LogWarnf("persistInodeAccessTime ino(%v) write to  connect failed %v", ino.Inode, err)
-			m.connPool.PutConnect(mConn, ForceClosedConnect)
-			return
-		}
-		if err = packetCopy.ReadFromConn(mConn, proto.NoReadDeadlineTime); err != nil {
-			log.LogWarnf("persistInodeAccessTime ino(%v) read from  connect failed %v", ino.Inode, err)
-			m.connPool.PutConnect(mConn, ForceClosedConnect)
-			return
-		}
-		m.connPool.PutConnect(mConn, NoClosedConnect)
-		log.LogDebugf("persistInodeAccessTime ino(%v) notify leader to %v sync accessTime", ino.Inode, leaderAddr)
 	}
 }

metanode/partition_op_inode_test.go

@@ -2,10 +2,12 @@ package metanode
 
 import (
 	"encoding/json"
+	"math/rand"
 	"testing"
 	"time"
 
 	"github.com/cubefs/cubefs/proto"
+	"github.com/cubefs/cubefs/util/timeutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -46,3 +48,30 @@ func TestInodeGet(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, resp.Info.AccessTime.Unix() == ino.AccessTime)
 }
+
+func TestInodeGetPerf(t *testing.T) {
+	initMp(t)
+	cnt := 10240000
+	for idx := 1; idx < cnt; idx++ {
+		ino := NewInode(uint64(idx), 0)
+		mp.inodeTree.ReplaceOrInsert(ino, true)
+	}
+
+	testNum := 1024
+	for i := 1; i < 10; i++ {
+		ids := make([]uint64, 0, testNum)
+		for i := 0; i < testNum; i++ {
+			ids = append(ids, rand.Uint64()%uint64(testNum)+1)
+		}
+
+		start := time.Now()
+		for _, id := range ids {
+			ino := NewInode(id, 0)
+			item := mp.inodeTree.CopyGet(ino)
+			require.NotNil(t, item)
+			newIno := item.(*Inode)
+			newIno.AccessTime = timeutil.GetCurrentTimeUnix()
+		}
+		t.Logf("TestInodeGetPerf: cnt %d, cost %dus", testNum, time.Since(start).Microseconds())
+	}
+}