Merge branch 'dev' into hardfork/pos

helper/concurrentmap/concurrent_map.go

@@ -88,11 +88,12 @@ func (m *concurrentMap) Delete(key interface{}) {
 }
 
 func (m *concurrentMap) Range(f func(key, value interface{}) bool) {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
+	keys := m.Keys()
+
+	for _, key := range keys {
+		load, _ := m.Load(key)
 
-	for key, value := range m.m {
-		if !f(key, value) {
+		if !f(key, load) {
 			break
 		}
 	}

helper/concurrentmap/concurrent_map_test.go

@@ -1,6 +1,8 @@
 package concurrentmap
 
-import "testing"
+import (
+	"testing"
+)
 
 func TestConcurrentMap(t *testing.T) {
 	cmap := NewConcurrentMap()
@@ -50,6 +52,11 @@ func TestConcurrentMap(t *testing.T) {
 			t.Error("Range() failed")
 		}
 
+		// test re-entrant deadlock
+		k, _ := key.(string)
+		v, _ := value.(string)
+		cmap.Store(k+"-re", v+"-re")
+
 		return true
 	})
 

protocol/syncer.go

@@ -203,7 +203,12 @@ func (s *Syncer) Broadcast(b *types.Block) {
 	sendNotify := func(peerID, peer interface{}, req *proto.NotifyReq) {
 		startTime := time.Now()
 
-		if _, err := peer.(*SyncPeer).client.Notify(context.Background(), req); err != nil {
+		syncPeer, ok := peer.(*SyncPeer)
+		if !ok {
+			return
+		}
+
+		if _, err := syncPeer.client.Notify(context.Background(), req); err != nil {
 			s.logger.Error("failed to notify", "err", err)
 
 			return
@@ -684,8 +689,15 @@ func getHeader(clt proto.V1Client, num *uint64, hash *types.Hash) (*types.Header
 
 func (s *Syncer) prunePeerEnqueuedBlocks(block *types.Block) {
 	s.peers.Range(func(key, value interface{}) bool {
-		peerID, _ := key.(peer.ID)
-		syncPeer, _ := value.(*SyncPeer)
+		peerID, ok := key.(peer.ID)
+		if !ok {
+			return true
+		}
+
+		syncPeer, ok := value.(*SyncPeer)
+		if !ok {
+			return true
+		}
 
 		pruned := syncPeer.purgeBlocks(block.Hash())
 

protocol/syncer_test.go

@@ -702,15 +702,7 @@ func createSyncers(count int, servers []*network.Server, blockStores []*mockBloc
 
 // numSyncPeers returns the number of sync peers
 func numSyncPeers(syncer *Syncer) int64 {
-	num := 0
-
-	syncer.peers.Range(func(key, value interface{}) bool {
-		num++
-
-		return true
-	})
-
-	return int64(num)
+	return int64(syncer.peers.Len())
 }
 
 // WaitUntilSyncPeersNumber waits until the number of sync peers reaches a certain number, otherwise it times out

txpool/account.go

@@ -6,21 +6,26 @@ import (
 	"sync/atomic"
 	"time"
 
-	cmap "github.com/dogechain-lab/dogechain/helper/concurrentmap"
 	"github.com/dogechain-lab/dogechain/types"
 )
 
 // Thread safe map of all accounts registered by the pool.
 // Each account (value) is bound to one address (key).
 type accountsMap struct {
-	cmap  cmap.ConcurrentMap
+	// sync.Map is thread-safe and high performance.
+	// We should not use some simple locked implementation map for this complicate usage,
+	// otherwise there might be deadlock.
+	//
+	// PS: accountsMap is never clear
+	// so golang#40999 (https://github.com/golang/go/issues/40999) is no problem,
+	// the price is high memory footprint
+	// but problem in restore blockchain from snapshot will cause high memory use (never release).
+	cmap  sync.Map
 	count uint64
 }
 
 func newAccountsMap() *accountsMap {
-	return &accountsMap{
-		cmap: cmap.NewConcurrentMap(),
-	}
+	return &accountsMap{}
 }
 
 // Intializes an account for the given address.