Merge branch 'blockpool2' of https://github.com/ethersphere/go-ethereum into ethersphere-blockpool2

5 files changed, 89 insertions, 51 deletions

diff --git a/blockpool/blockpool.go b/blockpool/blockpool.go
index 0126d734c..eb3b415dc 100644
--- a/blockpool/blockpool.go
+++ b/blockpool/blockpool.go
@@ -17,7 +17,7 @@ var plog = ethlogger.NewLogger("Blockpool")
 
 var (
 	// max number of block hashes sent in one request
-	blockHashesBatchSize = 512
+	blockHashesBatchSize = 256
 	// max number of blocks sent in one request
 	blockBatchSize = 64
 	// interval between two consecutive block checks (and requests)
@@ -27,11 +27,13 @@ var (
 	// interval between two consecutive block hash checks (and requests)
 	blockHashesRequestInterval = 3 * time.Second
 	// max number of idle iterations, ie., check through a section without new blocks coming in
-	blocksRequestMaxIdleRounds = 100
+	blocksRequestMaxIdleRounds = 20
 	// timeout interval: max time allowed for peer without sending a block hash
 	blockHashesTimeout = 60 * time.Second
 	// timeout interval: max time allowed for peer without sending a block
-	blocksTimeout = 120 * time.Second
+	blocksTimeout = 60 * time.Second
+	//
+	idleBestPeerTimeout = 60 * time.Second
 )
 
 // config embedded in components, by default fall back to constants
@@ -45,6 +47,7 @@ type Config struct {
 	BlocksRequestInterval      time.Duration
 	BlockHashesTimeout         time.Duration
 	BlocksTimeout              time.Duration
+	IdleBestPeerTimeout        time.Duration
 }
 
 // blockpool errors
@@ -53,6 +56,7 @@ const (
 	ErrInvalidPoW
 	ErrUnrequestedBlock
 	ErrInsufficientChainInfo
+	ErrIdleTooLong
 )
 
 var errorToString = map[int]string{
@@ -60,6 +64,7 @@ var errorToString = map[int]string{
 	ErrInvalidPoW:            "Invalid PoW",
 	ErrUnrequestedBlock:      "Unrequested block",
 	ErrInsufficientChainInfo: "Insufficient chain info",
+	ErrIdleTooLong:           "Idle too long",
 }
 
 // init initialises all your laundry
@@ -88,6 +93,9 @@ func (self *Config) init() {
 	if self.BlocksTimeout == 0 {
 		self.BlocksTimeout = blocksTimeout
 	}
+	if self.IdleBestPeerTimeout == 0 {
+		self.IdleBestPeerTimeout = idleBestPeerTimeout
+	}
 }
 
 // node is the basic unit of the internal model of block chain/tree in the blockpool
@@ -149,6 +157,15 @@ func New(
 	}
 }
 
+func severity(code int) ethlogger.LogLevel {
+	switch code {
+	case ErrUnrequestedBlock:
+		return ethlogger.WarnLevel
+	default:
+		return ethlogger.ErrorLevel
+	}
+}
+
 // allows restart
 func (self *BlockPool) Start() {
 	self.lock.Lock()
@@ -169,6 +186,7 @@ func (self *BlockPool) Start() {
 		errors: &errs.Errors{
 			Package: "Blockpool",
 			Errors:  errorToString,
+			Level:   severity,
 		},
 		peers:  make(map[string]*peer),
 		status: self.status,
@@ -363,6 +381,8 @@ LOOP:
 			// check if known block connecting the downloaded chain to our blockchain
 			plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) found block %s in the blockchain", peerId, hex(bestpeer.currentBlockHash), hex(hash))
 			if len(nodes) == 1 {
+				plog.DebugDetailf("AddBlockHashes: singleton section pushed to blockchain peer <%s> (head: %s) found block %s in the blockchain", peerId, hex(bestpeer.currentBlockHash), hex(hash))
+
 				// create new section if needed and push it to the blockchain
 				sec = self.newSection(nodes)
 				sec.addSectionToBlockChain(bestpeer)
@@ -379,6 +399,8 @@ LOOP:
 					and td together with blockBy are recorded on the node
 				*/
 				if len(nodes) == 0 && child != nil {
+					plog.DebugDetailf("AddBlockHashes: child section [%s] pushed to blockchain peer <%s> (head: %s) found block %s in the blockchain", sectionhex(child), peerId, hex(bestpeer.currentBlockHash), hex(hash))
+
 					child.addSectionToBlockChain(bestpeer)
 				}
 			}
@@ -446,10 +468,12 @@ LOOP:
 	*/
 	sec = self.linkSections(nodes, parent, child)
 
-	self.status.lock.Lock()
-	self.status.values.BlockHashes += len(nodes)
-	self.status.lock.Unlock()
-	plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): section [%s] created", peerId, hex(bestpeer.currentBlockHash), sectionhex(sec))
+	if sec != nil {
+		self.status.lock.Lock()
+		self.status.values.BlockHashes += len(nodes)
+		self.status.lock.Unlock()
+		plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): section [%s] created", peerId, hex(bestpeer.currentBlockHash), sectionhex(sec))
+	}
 
 	self.chainLock.Unlock()
 
@@ -549,6 +573,7 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
 			self.status.lock.Unlock()
 		} else {
 			plog.DebugDetailf("AddBlock: head block %s for peer <%s> (head: %s) already known", hex(hash), peerId, hex(sender.currentBlockHash))
+			sender.currentBlockC <- block
 		}
 	} else {
 
@@ -644,11 +669,15 @@ LOOP:
 		  we need to relink both complete and incomplete sections
 		  the latter could have been blockHashesRequestsComplete before being delinked from its parent
 		*/
-		if parent == nil && sec.bottom.block != nil {
-			if entry := self.get(sec.bottom.block.ParentHash()); entry != nil {
-				parent = entry.section
-				plog.DebugDetailf("activateChain: [%s]-[%s] relink", sectionhex(parent), sectionhex(sec))
-				link(parent, sec)
+		if parent == nil {
+			if sec.bottom.block != nil {
+				if entry := self.get(sec.bottom.block.ParentHash()); entry != nil {
+					parent = entry.section
+					plog.DebugDetailf("activateChain: [%s]-[%s] link", sectionhex(parent), sectionhex(sec))
+					link(parent, sec)
+				}
+			} else {
+				plog.DebugDetailf("activateChain: section [%s] activated by peer <%s> has missing root block", sectionhex(sec), p.id)
 			}
 		}
 		sec = parent
@@ -704,9 +733,15 @@ func (self *BlockPool) remove(sec *section) {
 	// delete node entries from pool index under pool lock
 	self.lock.Lock()
 	defer self.lock.Unlock()
+
 	for _, node := range sec.nodes {
 		delete(self.pool, string(node.hash))
 	}
+	if sec.initialised && sec.poolRootIndex != 0 {
+		self.status.lock.Lock()
+		self.status.values.BlocksInPool -= len(sec.nodes) - sec.missing
+		self.status.lock.Unlock()
+	}
 }
 
 func (self *BlockPool) getHashSlice() (s [][]byte) {
diff --git a/blockpool/config_test.go b/blockpool/config_test.go
index c06649024..d5540c864 100644
--- a/blockpool/config_test.go
+++ b/blockpool/config_test.go
@@ -20,12 +20,13 @@ func TestBlockPoolConfig(t *testing.T) {
 	test.CheckDuration("BlocksRequestInterval", c.BlocksRequestInterval, blocksRequestInterval, t)
 	test.CheckDuration("BlockHashesTimeout", c.BlockHashesTimeout, blockHashesTimeout, t)
 	test.CheckDuration("BlocksTimeout", c.BlocksTimeout, blocksTimeout, t)
+	test.CheckDuration("IdleBestPeerTimeout", c.IdleBestPeerTimeout, idleBestPeerTimeout, t)
 }
 
 func TestBlockPoolOverrideConfig(t *testing.T) {
 	test.LogInit()
 	blockPool := &BlockPool{Config: &Config{}}
-	c := &Config{128, 32, 1, 0, 300 * time.Millisecond, 100 * time.Millisecond, 90 * time.Second, 0}
+	c := &Config{128, 32, 1, 0, 300 * time.Millisecond, 100 * time.Millisecond, 90 * time.Second, 0, 30 * time.Second}
 
 	blockPool.Config = c
 	blockPool.Start()
@@ -37,4 +38,5 @@ func TestBlockPoolOverrideConfig(t *testing.T) {
 	test.CheckDuration("BlocksRequestInterval", c.BlocksRequestInterval, 100*time.Millisecond, t)
 	test.CheckDuration("BlockHashesTimeout", c.BlockHashesTimeout, 90*time.Second, t)
 	test.CheckDuration("BlocksTimeout", c.BlocksTimeout, blocksTimeout, t)
+	test.CheckDuration("IdleBestPeerTimeout", c.IdleBestPeerTimeout, 30*time.Second, t)
 }
diff --git a/blockpool/peers.go b/blockpool/peers.go
index 5f1b2017c..b4aa48f49 100644
--- a/blockpool/peers.go
+++ b/blockpool/peers.go
@@ -37,7 +37,7 @@ type peer struct {
 	currentBlockC chan *types.Block
 	headSectionC  chan *section
 
-	// channels to signal peers witch and peer quit
+	// channels to signal peer switch and peer quit to section processes
 	idleC   chan bool
 	switchC chan bool
 
@@ -47,7 +47,7 @@ type peer struct {
 	// timers for head section process
 	blockHashesRequestTimer <-chan time.Time
 	blocksRequestTimer      <-chan time.Time
-	suicide                 <-chan time.Time
+	suicideC                <-chan time.Time
 
 	idle bool
 }
@@ -214,6 +214,7 @@ func (self *peers) addPeer(
 	// check peer current head
 	if self.bp.hasBlock(currentBlockHash) {
 		// peer not ahead
+		plog.Debugf("addPeer: peer <%v> with td %v and current block %s is behind", id, td, hex(currentBlockHash))
 		return false
 	}
 
@@ -285,8 +286,7 @@ func (self *peers) removePeer(id string) {
 	}
 }
 
-// switchPeer launches section processes based on information about
-// shared interest and legacy of peers
+// switchPeer launches section processes
 func (self *BlockPool) switchPeer(oldp, newp *peer) {
 
 	// first quit AddBlockHashes, requestHeadSection and activateChain
@@ -371,16 +371,16 @@ func (self *peer) handleSection(sec *section) {
 			self.bp.syncing()
 		}
 
-		self.suicide = time.After(self.bp.Config.BlockHashesTimeout)
+		self.suicideC = time.After(self.bp.Config.BlockHashesTimeout)
 
 		plog.DebugDetailf("HeadSection: <%s> head block hash changed (mined block received). New head %s", self.id, hex(self.currentBlockHash))
 	} else {
 		if !self.idle {
 			self.idle = true
-			self.suicide = nil
 			self.bp.wg.Done()
 		}
 		plog.DebugDetailf("HeadSection: <%s> head section [%s] created", self.id, sectionhex(sec))
+		self.suicideC = time.After(self.bp.Config.IdleBestPeerTimeout)
 	}
 }
 
@@ -450,7 +450,7 @@ func (self *peer) getBlockHashes() {
 	self.blockHashesRequestTimer = nil
 	if !self.idle {
 		self.idle = true
-		self.suicide = nil
+		self.suicideC = time.After(self.bp.Config.IdleBestPeerTimeout)
 		self.bp.wg.Done()
 	}
 }
@@ -466,7 +466,7 @@ func (self *peer) run() {
 	self.blockHashesRequestTimer = nil
 
 	self.blocksRequestTimer = time.After(0)
-	self.suicide = time.After(self.bp.Config.BlockHashesTimeout)
+	self.suicideC = time.After(self.bp.Config.BlockHashesTimeout)
 
 	var quit chan bool
 
@@ -475,9 +475,20 @@ func (self *peer) run() {
 LOOP:
 	for {
 		select {
+		// to minitor section process behaviou
 		case <-ping.C:
 			plog.Debugf("HeadSection: <%s> section with head %s, idle: %v", self.id, hex(self.currentBlockHash), self.idle)
 
+		// idle timer started when process goes idle
+		case <-self.idleC:
+			if self.idle {
+				self.peerError(self.bp.peers.errors.New(ErrIdleTooLong, "timed out without providing new blocks...quitting", currentBlockHash))
+
+				self.bp.status.lock.Lock()
+				self.bp.status.badPeers[self.id]++
+				self.bp.status.lock.Unlock()
+			}
+
 		// signal from AddBlockHashes that head section for current best peer is created
 		// if sec == nil, it signals that chain info has updated (new block message)
 		case sec := <-self.headSectionC:
@@ -502,8 +513,8 @@ LOOP:
 			self.getCurrentBlock(nil)
 
 		// quitting on timeout
-		case <-self.suicide:
-			self.peerError(self.bp.peers.errors.New(ErrInsufficientChainInfo, "timed out without providing block hashes or head block", currentBlockHash))
+		case <-self.suicideC:
+			self.peerError(self.bp.peers.errors.New(ErrInsufficientChainInfo, "timed out without providing block hashes or head block %x", currentBlockHash))
 
 			self.bp.status.lock.Lock()
 			self.bp.status.badPeers[self.id]++
diff --git a/blockpool/section.go b/blockpool/section.go
index 48ea15d31..03c4f5cc6 100644
--- a/blockpool/section.go
+++ b/blockpool/section.go
@@ -138,7 +138,7 @@ func (self *section) addSectionToBlockChain(p *peer) {
 			plog.Warnf("penalise peers %v (hash), %v (block)", node.hashBy, node.blockBy)
 
 			// or invalid block and the entire chain needs to be removed
-			self.removeInvalidChain()
+			self.removeChain()
 		} else {
 			// if all blocks inserted in this section
 			// then need to try to insert blocks in child section
@@ -235,16 +235,14 @@ LOOP:
 
 		// timebomb - if section is not complete in time, nuke the entire chain
 		case <-self.suicideTimer:
-			self.suicide()
+			self.removeChain()
 			plog.Debugf("[%s] timeout. (%v total attempts): missing %v/%v/%v...suicide", sectionhex(self), self.blocksRequests, self.missing, self.lastMissing, self.depth)
 			self.suicideTimer = nil
+			break LOOP
 
 		// closing suicideC triggers section suicide: removes section nodes from pool and terminates section process
 		case <-self.suicideC:
-			plog.DebugDetailf("[%s] suicide", sectionhex(self))
-			self.unlink()
-			self.bp.remove(self)
-			plog.DebugDetailf("[%s] done", sectionhex(self))
+			plog.DebugDetailf("[%s] quit", sectionhex(self))
 			break LOOP
 
 		// alarm for checking blocks in the section
@@ -283,7 +281,7 @@ LOOP:
 				checking = false
 				break
 			}
-			plog.DebugDetailf("[%s] section proc step %v: missing %v/%v/%v", sectionhex(self), self.step, self.missing, self.lastMissing, self.depth)
+			// plog.DebugDetailf("[%s] section proc step %v: missing %v/%v/%v", sectionhex(self), self.step, self.missing, self.lastMissing, self.depth)
 			if !checking {
 				self.step = 0
 				self.missing = 0
@@ -522,7 +520,7 @@ func (self *section) checkRound() {
 				// too many idle rounds
 				if self.idle >= self.bp.Config.BlocksRequestMaxIdleRounds {
 					plog.DebugDetailf("[%s] block requests had %v idle rounds (%v total attempts): missing %v/%v/%v\ngiving up...", sectionhex(self), self.idle, self.blocksRequests, self.missing, self.lastMissing, self.depth)
-					self.suicide()
+					self.removeChain()
 				}
 			} else {
 				self.idle = 0
@@ -602,10 +600,12 @@ func (self *BlockPool) linkSections(nodes []*node, parent, child *section) (sec
 		link(parent, sec)
 		link(sec, child)
 	} else {
-		// now this can only happen if we allow response to hash request to include <from> hash
-		// in this case we just link parent and child (without needing root block of child section)
-		plog.Debugf("[%s]->[%s] connecting known sections", sectionhex(parent), sectionhex(child))
-		link(parent, child)
+		if parent != nil && child != nil {
+			// now this can only happen if we allow response to hash request to include <from> hash
+			// in this case we just link parent and child (without needing root block of child section)
+			plog.Debugf("[%s]->[%s] connecting known sections", sectionhex(parent), sectionhex(child))
+			link(parent, child)
+		}
 	}
 	return
 }
@@ -614,6 +614,7 @@ func (self *section) activate(p *peer) {
 	self.bp.wg.Add(1)
 	select {
 	case <-self.offC:
+		plog.DebugDetailf("[%s] completed section process. cannot activate for peer <%s>", sectionhex(self), p.id)
 		self.bp.wg.Done()
 	case self.controlC <- p:
 		plog.DebugDetailf("[%s] activate section process for peer <%s>", sectionhex(self), p.id)
@@ -625,22 +626,10 @@ func (self *section) deactivate() {
 	self.controlC <- nil
 }
 
-func (self *section) suicide() {
-	select {
-	case <-self.suicideC:
-		return
-	default:
-	}
-	close(self.suicideC)
-}
-
 // removes this section exacly
 func (self *section) remove() {
 	select {
 	case <-self.offC:
-		// section is complete, no process
-		self.unlink()
-		self.bp.remove(self)
 		close(self.suicideC)
 		plog.DebugDetailf("[%s] remove: suicide", sectionhex(self))
 	case <-self.suicideC:
@@ -649,21 +638,23 @@ func (self *section) remove() {
 		plog.DebugDetailf("[%s] remove: suicide", sectionhex(self))
 		close(self.suicideC)
 	}
+	self.unlink()
+	self.bp.remove(self)
 	plog.DebugDetailf("[%s] removed section.", sectionhex(self))
 
 }
 
 // remove a section and all its descendents from the pool
-func (self *section) removeInvalidChain() {
+func (self *section) removeChain() {
 	// need to get the child before removeSection delinks the section
 	self.bp.chainLock.RLock()
 	child := self.child
 	self.bp.chainLock.RUnlock()
 
-	plog.DebugDetailf("[%s] remove invalid chain", sectionhex(self))
+	plog.DebugDetailf("[%s] remove chain", sectionhex(self))
 	self.remove()
 	if child != nil {
-		child.removeInvalidChain()
+		child.removeChain()
 	}
 }
 
diff --git a/blockpool/status.go b/blockpool/status.go
index 0dd874232..4529c77fe 100644
--- a/blockpool/status.go
+++ b/blockpool/status.go
@@ -51,7 +51,6 @@ type Status struct {
 func (self *BlockPool) Status() *Status {
 	self.status.lock.Lock()
 	defer self.status.lock.Unlock()
-	self.status.values.BlockHashesInPool = len(self.pool)
 	self.status.values.ActivePeers = len(self.status.activePeers)
 	self.status.values.BestPeers = len(self.status.bestPeers)
 	self.status.values.BadPeers = len(self.status.badPeers)