Change isValidator to isBonded and add lock (#312)

monitor/common.go

@@ -124,23 +124,37 @@ func (m *Monitor) getAccount(addr sdk.AccAddress) (exported.Account, error) {
 }
 
 func (m *Monitor) dbGet(key []byte) ([]byte, error) {
-	m.dbLock.RLock()
-	defer m.dbLock.RUnlock()
+	m.lock.RLock()
+	defer m.lock.RUnlock()
 	return m.db.Get(key)
 }
 
 func (m *Monitor) dbSet(key, val []byte) error {
-	m.dbLock.Lock()
-	defer m.dbLock.Unlock()
+	m.lock.Lock()
+	defer m.lock.Unlock()
 	return m.db.Set(key, val)
 }
 
 func (m *Monitor) dbDelete(key []byte) error {
-	m.dbLock.Lock()
-	defer m.dbLock.Unlock()
+	m.lock.Lock()
+	defer m.lock.Unlock()
 	return m.db.Delete(key)
 }
 
-func chanInfoKey(cid mainchain.CidType, receiver mainchain.Addr) string {
-	return mainchain.Cid2Hex(cid) + ":" + mainchain.Addr2Hex(receiver)
+func (m *Monitor) isBonded() bool {
+	m.lock.RLock()
+	defer m.lock.RUnlock()
+	return m.bonded
+}
+
+func (m *Monitor) setBonded() {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.bonded = true
+}
+
+func (m *Monitor) setUnbonded() {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.bonded = false
 }

monitor/guard.go

@@ -54,7 +54,7 @@ func unmarshalChanInfo(input []byte) *ChanInfo {
 
 func (m *Monitor) processGuardQueue() {
 	var keys, vals [][]byte
-	m.dbLock.RLock()
+	m.lock.RLock()
 	iterator, err := m.db.Iterator(GuardKeyPrefix, storetypes.PrefixEndBytes(GuardKeyPrefix))
 	if err != nil {
 		log.Errorln("Create db iterator err", err)
@@ -65,7 +65,7 @@ func (m *Monitor) processGuardQueue() {
 		vals = append(vals, iterator.Value())
 	}
 	iterator.Close()
-	m.dbLock.RUnlock()
+	m.lock.RUnlock()
 
 	for i, key := range keys {
 		chanInfo := unmarshalChanInfo(vals[i])
@@ -100,18 +100,18 @@ func (m *Monitor) processGuardQueue() {
 				continue
 			}
 			if guarded {
-				m.dbLock.Lock()
+				m.lock.Lock()
 				exist, err := m.db.Has(key)
 				if err != nil {
 					log.Errorln("db Get err:", err)
-					m.dbLock.Unlock()
+					m.lock.Unlock()
 					continue
 				}
 				if exist {
 					val, err2 := m.db.Get(key)
 					if err2 != nil {
 						log.Errorln("db Get err", err2)
-						m.dbLock.Unlock()
+						m.lock.Unlock()
 						continue
 					}
 					chanInfo = unmarshalChanInfo(val)
@@ -127,7 +127,7 @@ func (m *Monitor) processGuardQueue() {
 				if err != nil {
 					log.Errorln("db Set err", err)
 				}
-				m.dbLock.Unlock()
+				m.lock.Unlock()
 			}
 		}
 	}
@@ -257,8 +257,8 @@ func (m *Monitor) setGuardEvent(eLog ethtypes.Log, state uint8) {
 
 func (m *Monitor) setChanInfo(cid mainchain.CidType, simplexReceiver mainchain.Addr, state uint8, seqNum uint64) {
 	key := GetGuardKey(cid, simplexReceiver)
-	m.dbLock.Lock()
-	defer m.dbLock.Unlock()
+	m.lock.Lock()
+	defer m.lock.Unlock()
 	var chanInfo *ChanInfo
 	exist, err := m.db.Has(key)
 	if err != nil {

monitor/monitor.go

@@ -35,9 +35,9 @@ type Monitor struct {
 	ledgerContract  monitor.Contract
 	verifiedChanges *bigcache.BigCache
 	sidechainAcct   sdk.AccAddress
-	isValidator     bool
+	bonded          bool
 	executeSlash    bool
-	dbLock          sync.RWMutex
+	lock            sync.RWMutex
 }
 
 func NewMonitor(ethClient *mainchain.EthClient, operator *transactor.Transactor, db dbm.DB) {
@@ -75,7 +75,7 @@ func NewMonitor(ethClient *mainchain.EthClient, operator *transactor.Transactor,
 		sgnContract:     sgnContract,
 		ledgerContract:  ledgerContract,
 		verifiedChanges: verifiedChanges,
-		isValidator:     mainchain.IsBonded(dposCandidateInfo),
+		bonded:          mainchain.IsBonded(dposCandidateInfo),
 		executeSlash:    viper.GetBool(common.FlagSgnExecuteSlash),
 	}
 	m.sidechainAcct, err = sdk.AccAddressFromBech32(viper.GetString(common.FlagSgnOperator))
@@ -138,7 +138,7 @@ func (m *Monitor) monitorSGNUpdateSidechainAddr() {
 			if dberr != nil {
 				log.Errorln("db Set err", dberr)
 			}
-			if !m.isValidator {
+			if !m.isBonded() {
 				e, perr := m.ethClient.SGN.ParseUpdateSidechainAddr(eLog)
 				if perr != nil {
 					log.Errorln("parse event err", perr)
@@ -216,15 +216,15 @@ func (m *Monitor) monitorDPoSValidatorChange() {
 				// self init sync if add validator
 				if validatorChange.EthAddr == m.ethClient.Address {
 					log.Infof("%s. Init my own validator.", logmsg)
-					m.isValidator = true
+					m.setBonded()
 					m.syncValidator(validatorChange.EthAddr)
 					m.setTransactors()
 				}
 			} else {
 				// self only put removal event to puller queue
 				log.Infof("%s, eth addr: %x", logmsg, validatorChange.EthAddr)
 				if validatorChange.EthAddr == m.ethClient.Address {
-					m.isValidator = false
+					m.setUnbonded()
 				}
 				event := NewEvent(ValidatorChange, eLog)
 				dberr := m.dbSet(GetPullerKey(eLog.TxHash), event.MustMarshal())
@@ -333,7 +333,7 @@ func (m *Monitor) handleDPoSDelegate(delegate *mainchain.DPoSDelegate) {
 		delegate.Delegator, delegate.Candidate, delegate.NewStake.String(), delegate.StakingPool.String())
 	m.syncDelegator(delegate.Candidate, delegate.Delegator)
 
-	if m.isValidator {
+	if m.isBonded() {
 		m.syncValidator(delegate.Candidate)
 	} else if m.shouldClaimValidator() {
 		m.claimValidatorOnMainchain()

monitor/penalty.go

@@ -20,7 +20,7 @@ func (m *Monitor) processPenaltyQueue() {
 		return
 	}
 	var keys, vals [][]byte
-	m.dbLock.RLock()
+	m.lock.RLock()
 	iterator, err := m.db.Iterator(PenaltyKeyPrefix, storetypes.PrefixEndBytes(PenaltyKeyPrefix))
 	if err != nil {
 		log.Errorln("Create db iterator err", err)
@@ -31,7 +31,7 @@ func (m *Monitor) processPenaltyQueue() {
 		vals = append(vals, iterator.Value())
 	}
 	iterator.Close()
-	m.dbLock.RUnlock()
+	m.lock.RUnlock()
 
 	for i, key := range keys {
 		event := NewPenaltyEventFromBytes(vals[i])

monitor/puller.go

@@ -21,7 +21,7 @@ func (m *Monitor) processPullerQueue() {
 		return
 	}
 	var keys, vals [][]byte
-	m.dbLock.RLock()
+	m.lock.RLock()
 	iterator, err := m.db.Iterator(PullerKeyPrefix, storetypes.PrefixEndBytes(PullerKeyPrefix))
 	if err != nil {
 		log.Errorln("Create db iterator err", err)
@@ -32,7 +32,7 @@ func (m *Monitor) processPullerQueue() {
 		vals = append(vals, iterator.Value())
 	}
 	iterator.Close()
-	m.dbLock.RUnlock()
+	m.lock.RUnlock()
 
 	for i, key := range keys {
 		event := NewEventFromBytes(vals[i])

monitor/sidechain.go

@@ -21,7 +21,7 @@ var (
 
 func (m *Monitor) monitorSidechainWithdrawReward() {
 	m.monitorTendermintEvent(initiateWithdrawRewardEvent, func(e abci.Event) {
-		if !m.isValidator {
+		if !m.isBonded() {
 			return
 		}
 
@@ -34,7 +34,7 @@ func (m *Monitor) monitorSidechainWithdrawReward() {
 
 func (m *Monitor) monitorSidechainSlash() {
 	m.monitorTendermintEvent(slashEvent, func(e abci.Event) {
-		if !m.isValidator {
+		if !m.isBonded() {
 			return
 		}