Repository: mantlenetworkio/mantle (language: go)

Issue:
Sequencer: Add sequencer code comments
During the development of the sequencer logic code, there is no corresponding comment on the specific method to describe the purpose of this method. At this stage, the development work has been basically completed, and we need to improve the comments of such methods to facilitate other developers to review.


Discussion / hints:

Fix patch:
diff --git a/l2geth/consensus/clique/scheduler.go b/l2geth/consensus/clique/scheduler.go
index 8bf9179ce..c9360108e 100644
--- a/l2geth/consensus/clique/scheduler.go
+++ b/l2geth/consensus/clique/scheduler.go
@@ -21,11 +21,15 @@ import (
 )
 
 const (
-	// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
+	// chainHeadChanSize is the size of channel listening to ChainHeadEvent
 	chainHeadChanSize = 10
 
-	defaultBatchSize  = int64(100)
+	// default config of scheduler consensus
+	// defaultBatchSize is the size of batch tx num
+	defaultBatchSize = int64(100)
+	// defaultExpireTime is the default config of a batch timeout
 	defaultExpireTime = int64(60)
+	// defaultBatchEpoch is the epoch time of the scheduler synchronizing the sequencer collection from l1, default 1 day
 	defaultBatchEpoch = time.Duration(86400)
 )
 
@@ -33,12 +37,17 @@ var (
 	scale = big.NewInt(1e18)
 )
 
+// Config sets the parameters required for the scheduler service
 type Config struct {
 	BatchSize  int64
 	BatchTime  int64
 	BatchEpoch int64
 }
 
+// Scheduler service has the ability to update the sequencer set
+// and select a producer from the sequencer set according to the priority algorithm.
+// The scheduler then generates blocks and broadcasts them based
+// on the transaction set sent by the producer
 type Scheduler struct {
 	sequencerSetMtx sync.Mutex // The lock used to protect the sequencerSet field
 
@@ -77,6 +86,7 @@ type Scheduler struct {
 	syncer *synchronizer.Synchronizer
 }
 
+// NewScheduler will create a scheduler server
 func NewScheduler(db ethdb.Database, config *Config, schedulerAddress common.Address, clique *Clique, blockchain *core.BlockChain, txpool *core.TxPool, eventMux *event.TypeMux) (*Scheduler, error) {
 	log.Info("Create Sequencer Server")
 
@@ -129,28 +139,37 @@ func NewScheduler(db ethdb.Database, config *Config, schedulerAddress common.Add
 	return schedulerInst, nil
 }
 
+// SetWallet set signer and wallet, so that scheduler servier
 func (schedulerInst *Scheduler) SetWallet(wallet accounts.Wallet, acc accounts.Account) {
 	schedulerInst.wallet = wallet
 	schedulerInst.signAccount = acc
 }
 
+// Scheduler return scheduler address
 func (schedulerInst *Scheduler) Scheduler() common.Address {
 	return schedulerInst.schedulerAddr
 }
 
+// CurrentStartMsg get the startMsg of this batch
 func (schedulerInst *Scheduler) CurrentStartMsg() types.BatchPeriodStartMsg {
 	return schedulerInst.currentStartMsg
 }
 
+// Start initializes and starts the sub-thread services of the scheduler
 func (schedulerInst *Scheduler) Start() {
 	if schedulerInst.wallet == nil || len(schedulerInst.signAccount.Address.Bytes()) == 0 {
 		panic("Sequencer server need wallet to sign msgs")
 	}
+
+	// channel init
 	schedulerInst.exitCh = make(chan struct{}, 1)
 	schedulerInst.batchDone = make(chan uint64, 2)
 
+	// Subscribe to events PeerAddEvent
 	schedulerInst.addPeerSub = schedulerInst.eventMux.Subscribe(core.PeerAddEvent{})
+	// Subscribe to events BatchEndEvent
 	schedulerInst.batchEndSub = schedulerInst.eventMux.Subscribe(core.BatchEndEvent(0))
+	// Subscribe to events chainHeadSub
 	schedulerInst.chainHeadSub = schedulerInst.blockchain.SubscribeChainHeadEvent(schedulerInst.chainHeadCh)
 
 	go schedulerInst.syncSequencerSetRoutine()
@@ -162,16 +181,19 @@ func (schedulerInst *Scheduler) Start() {
 	atomic.StoreInt32(&schedulerInst.running, 1)
 }
 
+// Stop set schedulerInst the indicator whether schedulerInst is running or not to 1
+// and call the method Close to unsubscribe events and close channel exitCh
 func (schedulerInst *Scheduler) Stop() {
 	atomic.StoreInt32(&schedulerInst.running, 0)
 	schedulerInst.Close()
 }
 
-// IsRunning returns an indicator whether schedulerInst is running or not.
+// IsRunning returns an indicator whether schedulerInst is running or not
 func (schedulerInst *Scheduler) IsRunning() bool {
 	return atomic.LoadInt32(&schedulerInst.running) == 1
 }
 
+// Close cancel all subscriptions and close channel exitCh
 func (schedulerInst *Scheduler) Close() {
 	schedulerInst.chainHeadSub.Unsubscribe()
 	schedulerInst.addPeerSub.Unsubscribe()
@@ -179,6 +201,11 @@ func (schedulerInst *Scheduler) Close() {
 	close(schedulerInst.exitCh)
 }
 
+// addPeerCheckLoop will not be called before schedulerInst starts.
+// When the miner starts, the schedulerInst service will be started synchronously
+// and the scheduler will use addPeerCheckLoop to check the connected p2p nodes.
+// The p2p connected after schedulerInst starts will call addPeerCheckLoop to check
+// whether it is in the sequencer set
 func (schedulerInst *Scheduler) addPeerCheckLoop() {
 	// automatically stops if unsubscribe
 	for obj := range schedulerInst.addPeerSub.Chan() {
@@ -196,6 +223,8 @@ func (schedulerInst *Scheduler) addPeerCheckLoop() {
 	}
 }
 
+// batchEndLoop subscription BatchEndEvent then end this batch,
+// this will happen when the batch has tx failed
 func (schedulerInst *Scheduler) batchEndLoop() {
 	// automatically stops if unsubscribe
 	for obj := range schedulerInst.batchEndSub.Chan() {
@@ -205,6 +234,14 @@ func (schedulerInst *Scheduler) batchEndLoop() {
 	}
 }
 
+// schedulerRoutine the main process of schedulerInst
+// uses the batch parameters in the config to rotate each round of batches.
+// Each round of batches will conduct a health check based on the results of
+// the previous round of batches first, and then perform L1ToL2Tx processing.
+// According to the sequencer set The priority selects the producer,
+// generates BatchPeriodStartMsg and broadcasts, then updates the priority of the sequencer set.
+// The batch will end only when it times out or receives batchDone signal,
+// and the process will be interrupted when the exitCh signal is received
 func (schedulerInst *Scheduler) schedulerRoutine() {
 	batchSize := defaultBatchSize
 	expireTime := defaultExpireTime
@@ -240,6 +277,7 @@ func (schedulerInst *Scheduler) schedulerRoutine() {
 				return fmt.Errorf("empty sequencer set")
 			}
 
+			// get producer
 			seq := schedulerInst.sequencerSet.getSeqWithMostPriority()
 			currentBlock := schedulerInst.blockchain.CurrentBlock()
 			currentIndex := rawdb.ReadStartMsgIndex(schedulerInst.db)
@@ -262,9 +300,12 @@ func (schedulerInst *Scheduler) schedulerRoutine() {
 				return fmt.Errorf("get minimum block count failed %s", err.Error())
 			}
 			schedulerInst.expectMinTxsCount = expectMinTxsCount
+			// store msg as currentStartMsg
 			schedulerInst.currentStartMsg = msg
+			// set BatchIndex to db
 			rawdb.WriteCurrentBatchPeriodIndex(schedulerInst.db, msg.BatchIndex)
 
+			// broadcast BatchPeriodStartMsg
 			err = schedulerInst.eventMux.Post(core.BatchPeriodStartEvent{
 				Msg:   &msg,
 				ErrCh: nil,
@@ -299,15 +340,18 @@ func (schedulerInst *Scheduler) schedulerRoutine() {
 	}
 }
 
+// handleChainHeadEventLoop checks whether the current block height reaches the currentStartMsg.maxHeight,
+// if it reaches the maxHeight, send batchIndex to channel batchDone.
 func (schedulerInst *Scheduler) handleChainHeadEventLoop() {
 	for {
 		select {
 		case chainHead := <-schedulerInst.chainHeadCh:
+			// Transactions of type QueueOriginL1ToL2 are not included in the batch
 			if chainHead.Block.Transactions().Len() != 0 && chainHead.Block.Transactions()[0].GetMeta() != nil && chainHead.Block.Transactions()[0].QueueOrigin() == types.QueueOriginL1ToL2 {
 				log.Debug("chainHead", "block_number", chainHead.Block.NumberU64(), "extra_data", hex.EncodeToString(chainHead.Block.Extra()))
 				continue
 			}
-			log.Info("schedulerInst handle chain head", "current_height", schedulerInst.blockchain.CurrentBlock().NumberU64(), "max_height", schedulerInst.currentStartMsg.MaxHeight)
+			log.Debug("schedulerInst handle chain head", "current_height", schedulerInst.blockchain.CurrentBlock().NumberU64(), "max_height", schedulerInst.currentStartMsg.MaxHeight)
 			if schedulerInst.blockchain.CurrentBlock().NumberU64() == schedulerInst.currentStartMsg.MaxHeight {
 				schedulerInst.batchDone <- schedulerInst.CurrentStartMsg().BatchIndex
 			}
@@ -318,6 +362,8 @@ func (schedulerInst *Scheduler) handleChainHeadEventLoop() {
 	}
 }
 
+// syncSequencerSetRoutine uses the batchEpoch parameter set in the config to regularly obtain
+// the sequencer set from L1 and update the schedulerInst.sequencerSet
 func (schedulerInst *Scheduler) syncSequencerSetRoutine() {
 	for {
 		select {
@@ -331,6 +377,8 @@ func (schedulerInst *Scheduler) syncSequencerSetRoutine() {
 					return fmt.Errorf("get sequencer set failed %s", err.Error())
 				}
 				schedulerInst.setSequencerHealthPoints(seqSet)
+
+				// todo do we need get changes then use UpdateWithChangeSet to update sequencer set or just replace schedulerInst.sequencerSet
 				// get changes
 				changes := compareSequencerSet(schedulerInst.sequencerSet.Sequencers, seqSet)
 				log.Info(fmt.Sprintf("Get sequencer set success, have changes: %d", len(changes)))
@@ -343,7 +391,7 @@ func (schedulerInst *Scheduler) syncSequencerSetRoutine() {
 				return err
 			}()
 			if err != nil {
-				log.Error("update sequencer set error", "err_msg", err.Error())
+				log.Error("syncSequencerSetRoutine update sequencer set error", "err_msg", err.Error())
 				continue
 			}
 		case <-schedulerInst.exitCh:
@@ -391,6 +439,7 @@ func compareSequencerSet(preSeqs []*Sequencer, newSeq synchronizer.SequencerSequ
 	return changes
 }
 
+// bindToSeq replaces the sequencer structure from SequencerSequencerInfos to Sequencer
 func bindToSeq(binds synchronizer.SequencerSequencerInfos) []*Sequencer {
 	var seqs []*Sequencer
 	for _, v := range binds {
diff --git a/l2geth/consensus/clique/scheduler_health.go b/l2geth/consensus/clique/scheduler_health.go
index e841fff53..47e9a5c95 100644
--- a/l2geth/consensus/clique/scheduler_health.go
+++ b/l2geth/consensus/clique/scheduler_health.go
@@ -31,6 +31,8 @@ func (schedulerInst *Scheduler) setSequencerHealthPoints(seqSets synchronizer.Se
 	}
 }
 
+// checkSequencer checks the working status of the previous producer at the beginning
+// of the new batch and deduct points of the sequencer that did not achieve the expected goal
 func (schedulerInst *Scheduler) checkSequencer() {
 	sequencer := schedulerInst.currentStartMsg.Sequencer
 	if sequencer.String() == (common.Address{}).String() {
@@ -80,6 +82,7 @@ func (schedulerInst *Scheduler) punishSequencer(sequencer common.Address) {
 	}
 }
 
+// deductPoints deduct points for the specified sequencer
 func (schedulerInst *Scheduler) deductPoints(sequencer common.Address) {
 	if schedulerInst.sequencerAssessor.SequencersPoints[sequencer] > 0 {
 		schedulerInst.sequencerAssessor.SequencersPoints[sequencer] = schedulerInst.sequencerAssessor.SequencersPoints[sequencer] - 1
@@ -97,6 +100,8 @@ func (schedulerInst *Scheduler) zeroPoints(sequencer common.Address) bool {
 	return schedulerInst.sequencerAssessor.SequencersPoints[sequencer] == 0
 }
 
+// getExpectMinTxsCount returns the minimum amount of transactions that the producer should
+// complete according to the number of transactions contained in the current txpool
 func (schedulerInst *Scheduler) getExpectMinTxsCount(batchSize uint64) (uint64, error) {
 	var pendingTxCount uint64
 	pendingTxs, err := schedulerInst.txpool.Pending()
diff --git a/l2geth/consensus/clique/synchronizer/server.go b/l2geth/consensus/clique/synchronizer/server.go
index 0d0242db2..a6deade54 100644
--- a/l2geth/consensus/clique/synchronizer/server.go
+++ b/l2geth/consensus/clique/synchronizer/server.go
@@ -10,9 +10,9 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/mantlenetworkio/mantle/l2geth/log"
 
 	binding "github.com/mantlenetworkio/mantle/l2geth/consensus/clique/synchronizer/bindings"
+	"github.com/mantlenetworkio/mantle/l2geth/log"
 )
 
 const (
@@ -24,26 +24,34 @@ const (
 	GET_SEQUENCER_TIMEOUT          = 10
 )
 
-// set infos for sort
+// SequencerSequencerInfos implementation the sort interface
+var _ sort.Interface = SequencerSequencerInfos{}
+
+// SequencerSequencerInfos set infos for sort
 type SequencerSequencerInfos []binding.SequencerSequencerInfo
 
-func (seqs SequencerSequencerInfos) Len() int {
-	return len(seqs)
+// Len for sort
+func (s SequencerSequencerInfos) Len() int {
+	return len(s)
 }
 
+// Less for sort
 func (s SequencerSequencerInfos) Less(i, j int) bool {
 	return s[i].Amount.Int64() < s[j].Amount.Int64()
 }
 
+// Swap for sort
 func (s SequencerSequencerInfos) Swap(i, j int) {
 	s[i], s[j] = s[j], s[i]
 }
 
+// Synchronizer set for connect with l1 sequencer contract
 type Synchronizer struct {
 	sequencerContractAddr common.Address
 	rpcUrl                string
 }
 
+// NewSynchronizer get sequencer address and l1 rpc address from os env then create Synchronizer
 func NewSynchronizer() *Synchronizer {
 	seqAddr := os.Getenv(ENV_SEQUENCER_CONTRACT_ADDRESS)
 	rpcUrl := os.Getenv(ENV_SEQUENCER_L1_RPC)
@@ -53,6 +61,7 @@ func NewSynchronizer() *Synchronizer {
 	}
 }
 
+// GetSchedulerAddr query scheduler address from l1 sequencer contract
 func (sync *Synchronizer) GetSchedulerAddr() (common.Address, error) {
 	// Create eth client
 	ctx, cancel := context.WithTimeout(context.Background(), ctxTimeout)
@@ -77,7 +86,7 @@ func (sync *Synchronizer) GetSchedulerAddr() (common.Address, error) {
 	return schedulerAddr, err
 }
 
-// GetSequencerSet will return the validator set
+// GetSequencerSet will query the sequencer set from l1 sequencer contract
 func (sync *Synchronizer) GetSequencerSet() (SequencerSequencerInfos, error) {
 	// Create eth client
 	ctx, cancel := context.WithTimeout(context.Background(), ctxTimeout)
diff --git a/l2geth/eth/consensus_handler.go b/l2geth/eth/consensus_handler.go
index 8a6912fdf..4a6fc3951 100644
--- a/l2geth/eth/consensus_handler.go
+++ b/l2geth/eth/consensus_handler.go
@@ -32,7 +32,8 @@ func (pm *ProtocolManager) makeConsensusProtocol(version uint) p2p.Protocol {
 	}
 }
 
-func (pm *ProtocolManager) removePeerTmp(id string) {
+// removeConsensusPeer disconnect with consensus peers
+func (pm *ProtocolManager) removeConsensusPeer(id string) {
 	// Short circuit if the peer was already removed
 	peer := pm.consensusPeers.Peer(id)
 	if peer == nil {
@@ -49,6 +50,8 @@ func (pm *ProtocolManager) removePeerTmp(id string) {
 	}
 }
 
+// consensusHandler add a new peer to consensusPeers, and start a
+// thread for this peer node to handle consensus messages between nodes
 func (pm *ProtocolManager) consensusHandler(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
 	p := pm.newPeer(int(eth64), peer, rw)
 	select {
@@ -68,7 +71,7 @@ func (pm *ProtocolManager) consensusHandler(peer *p2p.Peer, rw p2p.MsgReadWriter
 			p.Log().Error("Ethereum peer registration failed", "err", err)
 			return err
 		}
-		defer pm.removePeerTmp(p.id)
+		defer pm.removeConsensusPeer(p.id)
 
 		// Handle incoming messages until the connection is torn down
 		for {
@@ -86,6 +89,7 @@ func (pm *ProtocolManager) consensusHandler(peer *p2p.Peer, rw p2p.MsgReadWriter
 	}
 }
 
+// checkPeer check if peer is registered in the sequencer set, only check when schedulerInst running
 func (pm *ProtocolManager) checkPeer(p *peer) error {
 	if !pm.schedulerInst.IsRunning() {
 		return nil
@@ -113,6 +117,7 @@ func (pm *ProtocolManager) checkPeer(p *peer) error {
 	return nil
 }
 
+// handleConsensusMsg node received and handle consensus msg from p2p connection
 func (pm *ProtocolManager) handleConsensusMsg(p *peer) error {
 	// Read the next message from the remote peer, and ensure it's fully consumed
 	msg, err := p.rw.ReadMsg()
@@ -136,6 +141,7 @@ func (pm *ProtocolManager) handleConsensusMsg(p *peer) error {
 			return nil
 		}
 
+		// todo mark as know, keep the format consistent
 		p.knowBatchPeriodStartMsg.Add(bs.BatchIndex)
 		erCh := make(chan error, 1)
 		pm.eventMux.Post(core.BatchPeriodStartEvent{
@@ -155,6 +161,7 @@ func (pm *ProtocolManager) handleConsensusMsg(p *peer) error {
 		if !types.VerifySigner(bpa, pm.schedulerInst.CurrentStartMsg().Sequencer) {
 			return nil
 		}
+		// todo mark as know, keep the format consistent
 		p.knowBatchPeriodAnswerMsg.Add(bpa.Hash())
 		erCh := make(chan error, 1)
 		pm.eventMux.Post(core.BatchPeriodAnswerEvent{
@@ -171,7 +178,7 @@ func (pm *ProtocolManager) handleConsensusMsg(p *peer) error {
 
 // ---------------------------- Consensus Control Messages ----------------------------
 
-// BatchPeriodStartMsg will
+// BatchPeriodStartMsg get BatchPeriodStartEvent by subscription batchStartMsgSub then broadcast it
 func (pm *ProtocolManager) batchPeriodStartMsgBroadcastLoop() {
 	log.Info("Start batchPeriodStartMsg broadcast routine")
 	// automatically stops if unsubscribe
@@ -186,6 +193,7 @@ func (pm *ProtocolManager) batchPeriodStartMsgBroadcastLoop() {
 	}
 }
 
+// BroadcastBatchPeriodStartMsg broadcast BatchPeriodStartMsg
 func (pm *ProtocolManager) BroadcastBatchPeriodStartMsg(msg *types.BatchPeriodStartMsg) {
 	peers := pm.consensusPeers.PeersWithoutStartMsg(msg.BatchIndex)
 	for _, p := range peers {
@@ -194,6 +202,7 @@ func (pm *ProtocolManager) BroadcastBatchPeriodStartMsg(msg *types.BatchPeriodSt
 	log.Trace("Broadcast batch period start msg")
 }
 
+// AsyncSendBatchPeriodStartMsg set BatchPeriodStartMsg as known and clean up the old
 func (p *peer) AsyncSendBatchPeriodStartMsg(msg *types.BatchPeriodStartMsg) {
 	select {
 	case p.queuedBatchStartMsg <- msg:
@@ -207,7 +216,7 @@ func (p *peer) AsyncSendBatchPeriodStartMsg(msg *types.BatchPeriodStartMsg) {
 	}
 }
 
-// BatchPeriodAnswerMsg
+// BatchPeriodAnswerMsg get BatchPeriodAnswerEvent by subscription batchAnswerMsgSub then broadcast it
 func (pm *ProtocolManager) batchPeriodAnswerMsgBroadcastLoop() {
 	log.Info("Start batchPeriodAnswerMsg broadcast routine")
 	for obj := range pm.batchAnswerMsgSub.Chan() {
@@ -218,6 +227,7 @@ func (pm *ProtocolManager) batchPeriodAnswerMsgBroadcastLoop() {
 	}
 }
 
+// BroadcastBatchPeriodAnswerMsg broadcast BatchPeriodAnswerMsg
 func (pm *ProtocolManager) BroadcastBatchPeriodAnswerMsg(msg *types.BatchPeriodAnswerMsg) {
 	peers := pm.consensusPeers.PeersWithoutEndMsg(msg.Hash())
 	for _, p := range peers {
@@ -226,6 +236,7 @@ func (pm *ProtocolManager) BroadcastBatchPeriodAnswerMsg(msg *types.BatchPeriodA
 	log.Trace("Broadcast batch period answer msg")
 }
 
+// AsyncSendBatchPeriodAnswerMsg set queuedBatchAnswerMsg as known and clean up the old
 func (p *peer) AsyncSendBatchPeriodAnswerMsg(msg *types.BatchPeriodAnswerMsg) {
 	select {
 	case p.queuedBatchAnswerMsg <- msg:
@@ -241,12 +252,12 @@ func (p *peer) AsyncSendBatchPeriodAnswerMsg(msg *types.BatchPeriodAnswerMsg) {
 
 // ---------------------------- Proposers ----------------------------
 
-// SendBatchPeriodStart sends a batch of transaction receipts, corresponding to the
-// ones requested from an already RLP encoded format.
+// SendBatchPeriodStart sends BatchPeriodStartMsg
 func (p *peer) SendBatchPeriodStart(bps *types.BatchPeriodStartMsg) error {
 	return p2p.Send(p.rw, BatchPeriodStartMsg, bps)
 }
 
+// SendBatchPeriodAnswer sends BatchPeriodAnswerMsg
 func (p *peer) SendBatchPeriodAnswer(bpa *types.BatchPeriodAnswerMsg) error {
 	return p2p.Send(p.rw, BatchPeriodAnswerMsg, bpa)
 }
diff --git a/l2geth/eth/handler.go b/l2geth/eth/handler.go
index dfea7a430..f92953046 100644
--- a/l2geth/eth/handler.go
+++ b/l2geth/eth/handler.go
@@ -210,7 +210,7 @@ func (pm *ProtocolManager) setEtherBase(etherBase common.Address) {
 	pm.etherbase = etherBase
 	for k, v := range pm.consensusPeers.peers {
 		if err := pm.checkPeer(v); err != nil {
-			pm.removePeerTmp(k)
+			pm.removeConsensusPeer(k)
 		}
 	}
 }
@@ -283,10 +283,9 @@ func (pm *ProtocolManager) Start(maxPeers int) {
 	pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
 	go pm.minedBroadcastLoop()
 
-	// broadcast producers
+	// broadcast consensus messages
 	pm.batchStartMsgSub = pm.eventMux.Subscribe(core.BatchPeriodStartEvent{})
 	go pm.batchPeriodStartMsgBroadcastLoop()
-
 	pm.batchAnswerMsgSub = pm.eventMux.Subscribe(core.BatchPeriodAnswerEvent{})
 	go pm.batchPeriodAnswerMsgBroadcastLoop()
 


Test patch: