update for pool

Author: viteshan

common/condTimeout.go

@@ -1,15 +1,19 @@
 package common
 
 import (
-	"github.com/pkg/errors"
+	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
+
+	"github.com/pkg/errors"
 )
 
 var TimeoutErr = errors.New("timeout")
 
 type TimeoutCond struct {
-	cd *sync.Cond
+	cd        *sync.Cond
+	notifyNum uint32
 }
 
 func NewTimeoutCond() *TimeoutCond {
@@ -18,12 +22,21 @@ func NewTimeoutCond() *TimeoutCond {
 }
 
 func (self *TimeoutCond) Wait() {
+	old := atomic.SwapUint32(&self.notifyNum, 0)
+	if old > 0 {
+		return
+	}
 	self.cd.L.Lock()
 	defer self.cd.L.Unlock()
 	self.cd.Wait()
 }
 
 func (self *TimeoutCond) WaitTimeout(t time.Duration) error {
+	old := atomic.SwapUint32(&self.notifyNum, 0)
+	if old > 0 {
+		fmt.Printf("t:%s, num:%d\n", time.Now(), old)
+		return nil
+	}
 	done := make(chan struct{})
 	go func() {
 		self.Wait()
@@ -40,9 +53,10 @@ func (self *TimeoutCond) WaitTimeout(t time.Duration) error {
 }
 
 func (self *TimeoutCond) Broadcast() {
+	atomic.AddUint32(&self.notifyNum, 1)
 	self.cd.Broadcast()
-
 }
 func (self *TimeoutCond) Signal() {
+	atomic.AddUint32(&self.notifyNum, 1)
 	self.cd.Signal()
 }

common/types/address.go

@@ -5,11 +5,12 @@ import (
 	"crypto/rand"
 	"encoding/hex"
 	"fmt"
+	"math/big"
+	"strings"
+
 	"github.com/vitelabs/go-vite/common/helper"
 	vcrypto "github.com/vitelabs/go-vite/crypto"
 	"github.com/vitelabs/go-vite/crypto/ed25519"
-	"math/big"
-	"strings"
 )
 
 const (
@@ -85,6 +86,14 @@ func HexToAddress(hexStr string) (Address, error) {
 	}
 }
 
+func HexToAddressPanic(hexstr string) Address {
+	h, err := HexToAddress(hexstr)
+	if err != nil {
+		panic(err)
+	}
+	return h
+}
+
 func IsValidHexAddress(hexStr string) bool {
 	if len(hexStr) != hexAddressLength || !strings.HasPrefix(hexStr, AddressPrefix) {
 		return false

pool/account_pool.go

@@ -628,8 +628,8 @@ func (self *accountPool) makePackage(q Package, info *offsetInfo) (uint64, error
 	self.pool.RLock()
 	defer self.pool.RUnLock()
 
-	self.rMu.Lock()
-	defer self.rMu.Unlock()
+	self.chainTailMu.Lock()
+	defer self.chainTailMu.Unlock()
 
 	cp := self.chainpool
 	current := cp.current
@@ -679,7 +679,7 @@ func (self *accountPool) makePackage(q Package, info *offsetInfo) (uint64, error
 func (self *accountPool) tryInsertItems(items []*Item, latestSb *ledger.SnapshotBlock) error {
 	// if current size is empty, do nothing.
 	if self.chainpool.current.size() <= 0 {
-		return errors.New("empty chainpool")
+		return errors.Errorf("empty chainpool, but item size:%d", len(items))
 	}
 
 	self.chainTailMu.Lock()
@@ -720,7 +720,7 @@ func (self *accountPool) tryInsertItems(items []*Item, latestSb *ledger.Snapshot
 			fmt.Println(self.address, item.commonBlock.(*accountPoolBlock).block.IsSendBlock())
 			return errors.New("tail not match")
 		}
-		fmt.Printf("try to insert account block[%s]%d-%d success.\n", block.Hash(), i, len(items))
+		self.log.Info(fmt.Sprintf("try to insert account block[%s]%d-%d [latency:%s]success.\n", block.Hash(), i, len(items), block.Latency()))
 	}
 	return nil
 }

pool/pool.go

@@ -79,17 +79,19 @@ type commonBlock interface {
 	resetForkVersion()
 	forkVersion() int
 	Source() types.BlockSource
+	Latency() time.Duration
 	ReferHashes() ([]types.Hash, []types.Hash, *types.Hash)
 }
 
 func newForkBlock(v *ForkVersion, source types.BlockSource) *forkBlock {
-	return &forkBlock{firstV: v.Val(), v: v, source: source}
+	return &forkBlock{firstV: v.Val(), v: v, source: source, nTime: time.Now()}
 }
 
 type forkBlock struct {
 	firstV int
 	v      *ForkVersion
 	source types.BlockSource
+	nTime  time.Time
 }
 
 func (self *forkBlock) forkVersion() int {
@@ -102,6 +104,16 @@ func (self *forkBlock) resetForkVersion() {
 	val := self.v.Val()
 	self.firstV = val
 }
+func (self *forkBlock) Latency() time.Duration {
+	if self.Source() == types.RemoteBroadcast || self.Source() == types.RemoteFetch {
+		return time.Now().Sub(self.nTime)
+	}
+	return time.Duration(0)
+}
+
+func (self *forkBlock) Source() types.BlockSource {
+	return self.source
+}
 
 type pool struct {
 	pendingSc *snapshotPool
@@ -117,6 +129,9 @@ type pool struct {
 	accountSubId  int
 	snapshotSubId int
 
+	newAccBlockCond      *common.TimeoutCond
+	newSnapshotBlockCond *common.TimeoutCond
+
 	rwMutex sync.RWMutex
 	version *ForkVersion
 
@@ -189,6 +204,8 @@ func NewPool(bc chainDb) (*pool, error) {
 	self.addrCache = cache
 
 	self.hashBlacklist, err = NewBlacklist()
+	self.newAccBlockCond = common.NewTimeoutCond()
+	self.newSnapshotBlockCond = common.NewTimeoutCond()
 	if err != nil {
 		return nil, err
 	}
@@ -205,7 +222,7 @@ func (self *pool) Init(s syncer,
 	fe := &snapshotSyncer{fetcher: s, log: self.log.New("t", "snapshot")}
 	v := &snapshotVerifier{v: snapshotV}
 	self.accountVerifier = accountV
-	snapshotPool := newSnapshotPool("snapshotPool", self.version, v, fe, rw, self.hashBlacklist, self.log)
+	snapshotPool := newSnapshotPool("snapshotPool", self.version, v, fe, rw, self.hashBlacklist, self.newSnapshotBlockCond, self.log)
 	snapshotPool.init(
 		newTools(fe, rw),
 		self)
@@ -335,6 +352,8 @@ func (self *pool) AddSnapshotBlock(block *ledger.SnapshotBlock, source types.Blo
 		return
 	}
 	self.pendingSc.AddBlock(newSnapshotPoolBlock(block, self.version, source))
+
+	self.newSnapshotBlockCond.Broadcast()
 }
 
 func (self *pool) AddDirectSnapshotBlock(block *ledger.SnapshotBlock) error {
@@ -375,6 +394,7 @@ func (self *pool) AddAccountBlock(address types.Address, block *ledger.AccountBl
 	ac.AddBlock(newAccountPoolBlock(block, nil, self.version, source))
 	ac.AddReceivedBlock(block)
 
+	self.newAccBlockCond.Broadcast()
 }
 
 func (self *pool) AddDirectAccountBlock(address types.Address, block *vm_db.VmAccountBlock) error {
@@ -408,6 +428,7 @@ func (self *pool) AddAccountBlocks(address types.Address, blocks []*ledger.Accou
 		self.AddAccountBlock(address, b, source)
 	}
 
+	self.newAccBlockCond.Broadcast()
 	return nil
 }
 
@@ -625,24 +646,16 @@ func (self *pool) loopCompact() {
 	self.wg.Add(1)
 	defer self.wg.Done()
 
-	t := time.NewTicker(time.Millisecond * 40)
-	defer t.Stop()
 	sum := 0
 	for {
 		select {
 		case <-self.closed:
 			return
-		case <-t.C:
+		default:
 			if sum == 0 {
-				//self.accountCond.L.Lock()
-				//self.accountCond.Wait()
-				//self.accountCond.L.Unlock()
-				time.Sleep(200 * time.Millisecond)
+				self.newAccBlockCond.WaitTimeout(30 * time.Millisecond)
 			}
 			sum = 0
-
-			sum += self.accountsCompact()
-		default:
 			sum += self.accountsCompact()
 		}
 	}

pool/pool_loop.go

@@ -17,19 +17,29 @@ loop:
 2. insert for queue
 */
 func (self *pool) loopQueue() {
+	self.wg.Add(1)
+	defer self.wg.Done()
 	for {
-		q := self.makeQueue()
-		size := q.Size()
-		if size == 0 {
-			time.Sleep(20 * time.Millisecond)
-			continue
-		}
-		err := self.insertQueue(q)
-		if err != nil {
-			fmt.Printf("insert queue err:%s\n", err)
-			fmt.Printf("all queue:%s\n", q.Info())
-			time.Sleep(time.Second * 2)
-			self.log.Crit("loop pool exit")
+		select {
+		case <-self.closed:
+			return
+		default:
+			t1 := time.Now()
+			q := self.makeQueue()
+			size := q.Size()
+			if size == 0 {
+				time.Sleep(2 * time.Millisecond)
+				continue
+			}
+			err := self.insertQueue(q)
+			if err != nil {
+				fmt.Printf("insert queue err:%s\n", err)
+				fmt.Printf("all queue:%s\n", q.Info())
+				time.Sleep(time.Second * 2)
+				self.log.Crit("loop pool exit")
+			}
+			t2 := time.Now()
+			self.log.Info(fmt.Sprintf("time duration:%s, size:%d", t2.Sub(t1), size))
 		}
 	}
 }

pool/processor_snapshot_package.go

@@ -163,6 +163,12 @@ func (self *snapshotPackage) AddItem(b *Item) error {
 	if max > self.current {
 		self.current = max
 	}
+	if !b.Snapshot() {
+		err := self.accountExistsF(b.Hash())
+		if err == nil {
+			panic(fmt.Sprintf("panic for account block[%s][%d][%s] exist", b.Hash(), b.Height(), b.ownerWrapper))
+		}
+	}
 	return err
 }
 

pool/snapshot_pool.go

@@ -18,16 +18,18 @@ type snapshotPool struct {
 	BCPool
 	//rwMu *sync.RWMutex
 	//consensus consensus.AccountsConsensus
-	closed          chan struct{}
-	wg              sync.WaitGroup
-	pool            *pool
-	rw              *snapshotCh
-	v               *snapshotVerifier
-	f               *snapshotSyncer
-	nextFetchTime   time.Time
-	nextInsertTime  time.Time
-	nextCompactTime time.Time
-	hashBlacklist   Blacklist
+	closed chan struct{}
+	wg     sync.WaitGroup
+	pool   *pool
+	rw     *snapshotCh
+	v      *snapshotVerifier
+	f      *snapshotSyncer
+
+	nextFetchTime        time.Time
+	nextInsertTime       time.Time
+	nextCompactTime      time.Time
+	hashBlacklist        Blacklist
+	newSnapshotBlockCond *common.TimeoutCond
 }
 
 func newSnapshotPoolBlock(block *ledger.SnapshotBlock, version *ForkVersion, source types.BlockSource) *snapshotPoolBlock {
@@ -68,17 +70,14 @@ func (self *snapshotPoolBlock) PrevHash() types.Hash {
 	return self.block.PrevHash
 }
 
-func (self *snapshotPoolBlock) Source() types.BlockSource {
-	return self.source
-}
-
 func newSnapshotPool(
 	name string,
 	version *ForkVersion,
 	v *snapshotVerifier,
 	f *snapshotSyncer,
 	rw *snapshotCh,
 	hashBlacklist Blacklist,
+	cond *common.TimeoutCond,
 	log log15.Logger,
 ) *snapshotPool {
 	pool := &snapshotPool{}
@@ -93,6 +92,7 @@ func newSnapshotPool(
 	pool.nextInsertTime = now
 	pool.nextCompactTime = now
 	pool.hashBlacklist = hashBlacklist
+	pool.newSnapshotBlockCond = cond
 	return pool
 }
 
@@ -305,9 +305,9 @@ func (self *snapshotPool) loop() {
 			s1 := self.nextCompactTime.Sub(n2)
 			s2 := self.nextInsertTime.Sub(n2)
 			if s1 > s2 {
-				time.Sleep(s2)
+				self.newSnapshotBlockCond.WaitTimeout(s2)
 			} else {
-				time.Sleep(s1)
+				self.newSnapshotBlockCond.WaitTimeout(s1)
 			}
 			monitor.LogTime("pool", "snapshotRealSleep", n2)
 			last = time.Now()