peer / section interaction redone

- now register peer on active chain: section.controlC now chan *peerInfo
- call requestBlockHashes on this best peer
- stopping sections now simply done via peer.quitC (reassigned via controlC when new best peer promoted)
- now new best peer always registers on all remembered sections
- simplified process start (only switchPeer function remained)
- blockpool.requestBlocks now async goroutine so that peer activation (locking peersLock) cannot cause deadlock by catching section process requesting Blocks and waiting for peersLock

eth/block_pool.go

@@ -91,7 +91,7 @@ type section struct {
 	top         *poolNode
 	bottom      *poolNode
 	nodes       []*poolNode
-	controlC    chan bool
+	controlC    chan *peerInfo
 	suicideC    chan bool
 	blockChainC chan bool
 	forkC       chan chan bool
@@ -287,25 +287,6 @@ func (self *BlockPool) RemovePeer(peerId string) {
 	}
 }
 
-func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) {
-	if newPeer != nil {
-		entry := self.get(newPeer.currentBlock)
-		if entry == nil {
-			poolLogger.Debugf("[%s] head block [%s] not found, requesting hashes", newPeer.id, name(newPeer.currentBlock))
-			newPeer.requestBlockHashes(newPeer.currentBlock)
-		} else {
-			poolLogger.Debugf("[%s] head block [%s] found, activate chain at section [%s]", newPeer.id, name(newPeer.currentBlock), sectionName(entry.section))
-			self.activateChain(entry.section, newPeer)
-		}
-	}
-	if oldPeer != nil {
-		oldPeer.stop(newPeer)
-	}
-	if newPeer != nil {
-		newPeer.start(oldPeer)
-	}
-}
-
 // Entry point for eth protocol to add block hashes received via BlockHashesMsg
 // only hashes from the best peer is handled
 // this method is always responsible to initiate further hash requests until
@@ -314,7 +295,8 @@ func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) {
 // this function needs to run asynchronously for one peer since the message is discarded???
 func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string) {
 
-	// check if this peer is the best
+	// register with peer manager loop
+
 	peer, best := self.getPeer(peerId)
 	if !best {
 		return
@@ -322,13 +304,9 @@ func (self *BlockPool) AddBlockHashes(next func() ([]byte, bool), peerId string)
 	// peer is still the best
 	poolLogger.Debugf("adding hashes for best peer %s", peerId)
 
-	// self.wg.Add(1)
-	// self.procWg.Add(1)
-
-	// go func() {
 	var size, n int
 	var hash []byte
-	var ok bool = true
+	var ok bool
 	var section, child, parent *section
 	var entry *poolEntry
 	var nodes []*poolNode
@@ -339,9 +317,10 @@ LOOP:
 		n++
 		select {
 		case <-self.quit:
-			break LOOP
+			return
 		case <-peer.quitC:
 			// if the peer is demoted, no more hashes taken
+			peer = nil
 			break LOOP
 		default:
 		}
@@ -393,7 +372,7 @@ LOOP:
 
 	if parent != nil && entry != nil && entry.node != parent.top {
 		poolLogger.Debugf("[%s] fork section", sectionName(parent))
-		parent.controlC <- false
+		parent.controlC <- nil
 		waiter := make(chan bool)
 		parent.forkC <- waiter
 		chain := parent.nodes
@@ -402,7 +381,7 @@ LOOP:
 		orphan := newSection()
 		self.link(orphan, parent.child)
 		self.processSection(orphan, chain[0:entry.index])
-		orphan.controlC <- false
+		orphan.controlC <- nil
 		close(waiter)
 	}
 
@@ -419,7 +398,7 @@ LOOP:
 	self.chainLock.Unlock()
 	poolLogger.Debugf("[%s] unlock chain lock", sectionName(section))
 
-	if parent != nil {
+	if parent != nil && peer != nil {
 		poolLogger.Debugf("[%s] activating parent chain [%s]...", name(parent.top.hash), sectionName(parent))
 		self.activateChain(parent, peer)
 		poolLogger.Debugf("[%s] activated parent chain [%s]. done", name(parent.top.hash), sectionName(parent))
@@ -428,12 +407,8 @@ LOOP:
 	if section != nil {
 		poolLogger.Debugf("[%s] activate new section process", sectionName(section))
 		peer.addSection(section.top.hash, section)
-		section.controlC <- true
+		section.controlC <- peer
 	}
-	// 	self.procWg.Done()
-	// 	self.wg.Done()
-
-	// }()
 }
 
 func name(hash []byte) (name string) {
@@ -556,15 +531,9 @@ LOOP:
 		poolLogger.Debugf("[%s] register section with peer %s", sectionName(section), peer.id)
 		peer.addSection(section.top.hash, section)
 		poolLogger.Debugf("[%s] activate section process", sectionName(section))
-		section.controlC <- true
+		section.controlC <- peer
 		i++
-		// section.lock.RLock()
-		// parent := section.parent
-		// section.lock.RUnlock()
-		// section = parent
-		poolLogger.Debugf(" before")
 		section = self.getParent(section)
-		poolLogger.Debugf(" after")
 		select {
 		case <-peer.quitC:
 			break LOOP
@@ -601,6 +570,8 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 		// absolute time after which sub-chain is killed if not complete (some blocks are missing)
 		suicideTimer := time.After(blockTimeout * time.Minute)
 
+		var peer, newPeer *peerInfo
+
 		var blocksRequestTimer, blockHashesRequestTimer <-chan time.Time
 		var blocksRequestTime, blockHashesRequestTime bool
 		var blocksRequests, blockHashesRequests int
@@ -613,8 +584,9 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 
 		var i, total, missing, lastMissing, depth int
 		var idle int
-		var init, done, same, running, ready bool
+		var init, done, same, ready bool
 		var insertChain bool
+		var quitC chan bool
 
 		var blockChainC = section.blockChainC
 
@@ -709,7 +681,7 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 				} else {
 					blockHashesRequests++
 					poolLogger.Debugf("[%s] hash request on root (%v total attempts)\n", sectionName(section), blockHashesRequests)
-					self.requestBlockHashes(section.bottom.hash)
+					peer.requestBlockHashes(section.bottom.hash)
 					blockHashesRequestTimer = time.After(blockHashesRequestInterval * time.Millisecond)
 				}
 				blockHashesRequestTime = false
@@ -723,6 +695,13 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 			case <-self.quit:
 				break LOOP
 
+			case <-quitC:
+				// peer quit or demoted, put section in idle mode
+				quitC = nil
+				go func() {
+					section.controlC <- nil
+				}()
+
 			case <-self.purgeC:
 				suicideTimer = time.After(0)
 
@@ -733,15 +712,18 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 			case <-section.suicideC:
 				poolLogger.Debugf("[%s] suicide", sectionName(section))
 
+				// first delink from child and parent under chainlock
 				self.chainLock.Lock()
 				self.link(nil, section)
 				self.link(section, nil)
 				self.chainLock.Unlock()
+				// delete node entries from pool index under pool lock
 				self.lock.Lock()
 				for _, node := range section.nodes {
 					delete(self.pool, string(node.hash))
 				}
 				self.lock.Unlock()
+
 				break LOOP
 
 			case <-blocksRequestTimer:
@@ -752,16 +734,16 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 				poolLogger.Debugf("[%s] hash request time again", sectionName(section))
 				blockHashesRequestTime = true
 
-			case r := <-section.controlC:
+			case newPeer = <-section.controlC:
 
-				if running && !r {
+				// active -> idle
+				if peer != nil && newPeer == nil {
 					self.procWg.Done()
 					poolLogger.Debugf("[%s] idle mode", sectionName(section))
 					if init {
 						poolLogger.Debugf("[%s] off (%v total attempts): missing %v/%v/%v", sectionName(section), blocksRequests, missing, total, depth)
 					}
 
-					running = false
 					blocksRequestTime = false
 					blocksRequestTimer = nil
 					blockHashesRequestTime = false
@@ -771,9 +753,10 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 						processC = nil
 					}
 				}
-				if !running && r {
+
+				// idle -> active
+				if peer == nil && newPeer != nil {
 					self.procWg.Add(1)
-					running = true
 
 					poolLogger.Debugf("[%s] active mode", sectionName(section))
 					poolLogger.Debugf("[%s] check if complete", sectionName(section))
@@ -786,7 +769,6 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 						blockHashesRequestTime = true
 					}
 					if !init {
-						// if not run at least once fully, launch iterator
 						processC = make(chan *poolNode, blockHashesBatchSize)
 						missingC = make(chan *poolNode, blockHashesBatchSize)
 						poolLogger.Debugf("[%s] initialise section", sectionName(section))
@@ -798,6 +780,7 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 						self.wg.Add(1)
 						self.procWg.Add(1)
 						depth = len(section.nodes)
+						// if not run at least once fully, launch iterator
 						go func() {
 							var node *poolNode
 						IT:
@@ -817,8 +800,14 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 						processC = offC
 					}
 				}
+				// reset quitC to current best peer
+				if newPeer != nil {
+					quitC = newPeer.quitC
+				}
+				peer = newPeer
 
 			case waiter := <-section.forkC:
+				// this case just blocks the process until section is split at the fork
 				poolLogger.Debugf("[%s] locking for fork", sectionName(section))
 				<-waiter
 				poolLogger.Debugf("[%s] unlocking for fork", sectionName(section))
@@ -897,7 +886,7 @@ func (self *BlockPool) processSection(section *section, nodes []*poolNode) {
 		poolLogger.Debugf("[%s] process complete done", sectionName(section))
 
 		self.wg.Done()
-		if running {
+		if peer != nil {
 			self.procWg.Done()
 		}
 	}()
@@ -913,16 +902,10 @@ func (self *BlockPool) peerError(peerId string, code int, format string, params
 	}
 }
 
-func (self *BlockPool) requestBlockHashes(hash []byte) {
-	self.peersLock.Lock()
-	defer self.peersLock.Unlock()
-	if self.peer != nil {
-		poolLogger.Debugf("request hashes starting on %x from best peer %s", hash[:4], self.peer.id)
-		self.peer.requestBlockHashes(hash)
-	}
-}
-
 func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) {
+	self.wg.Add(1)
+	self.procWg.Add(1)
+	go func() {
 		// distribute block request among known peers
 		poolLogger.Debugf("request blocks")
 		self.peersLock.Lock()
@@ -951,7 +934,9 @@ func (self *BlockPool) requestBlocks(attempts int, hashes [][]byte) {
 			i++
 		}
 		poolLogger.Debugf("done requesting blocks")
-
+		self.wg.Done()
+		self.procWg.Done()
+	}()
 }
 
 func (self *BlockPool) getPeer(peerId string) (*peerInfo, bool) {
@@ -974,54 +959,32 @@ func (self *peerInfo) addSection(hash []byte, section *section) {
 	self.sections[string(hash)] = section
 }
 
-// (re)starts processes registered for this peer (self)
+func (self *BlockPool) switchPeer(oldPeer, newPeer *peerInfo) {
-func (self *peerInfo) start(peer *peerInfo) {
+	if newPeer != nil {
-	self.lock.Lock()
+		entry := self.get(newPeer.currentBlock)
-	defer self.lock.Unlock()
+		if entry == nil {
-	self.quitC = make(chan bool)
+			poolLogger.Debugf("[%s] head block [%s] not found, requesting hashes", newPeer.id, name(newPeer.currentBlock))
-	poolLogger.Debugf("[%s] activate section processes", self.id)
+			newPeer.requestBlockHashes(newPeer.currentBlock)
-	self.controlSections(peer, true)
+		} else {
+			poolLogger.Debugf("[%s] head block [%s] found, activate chain at section [%s]", newPeer.id, name(newPeer.currentBlock), sectionName(entry.section))
+			self.activateChain(entry.section, newPeer)
 		}
-
+		poolLogger.Debugf("[%s] activate section processes", newPeer.id)
-//  (re)starts process without requests, only suicide timer
+		for hash, section := range newPeer.sections {
-func (self *peerInfo) stop(peer *peerInfo) {
-	self.lock.RLock()
-	defer self.lock.RUnlock()
-	close(self.quitC)
-	poolLogger.Debugf("[%s] inactivate section processes", self.id)
-	self.controlSections(peer, false)
-}
-
-func (self *peerInfo) controlSections(peer *peerInfo, on bool) {
-	if peer != nil {
-		peer.lock.RLock()
-		defer peer.lock.RUnlock()
-	}
-
-	for hash, section := range self.sections {
-
 			if section.off {
-			poolLogger.Debugf("[%s][%x] section process complete - remove", self.id, hash[:4])
+				poolLogger.Debugf("[%s][%x] section process complete - remove", newPeer.id, hash[:4])
-			delete(self.sections, hash)
+				delete(newPeer.sections, hash)
 				continue
 			}
-		var found bool
+			// this will block if section process is waiting for peer lock
-		if peer != nil {
+			poolLogger.Debugf("[%s][%x] registering peer with section", newPeer.id, hash[:4])
-			_, found = peer.sections[hash]
+			section.controlC <- newPeer
+			poolLogger.Debugf("[%s][%x] registered peer with section", newPeer.id, hash[:4])
 		}
-
+		newPeer.quitC = make(chan bool)
-		// switch on processes not found in old peer
-		// and switch off processes not found in new peer
-		if !found {
-			if on {
-				// self is best peer
-				poolLogger.Debugf("[%s][%s] section process -> active", self.id, sectionName(section))
-			} else {
-				//  (re)starts process without requests, only suicide timer
-				poolLogger.Debugf("[%s][%s] section process -> inactive", self.id, sectionName(section))
-			}
-			section.controlC <- on
 	}
+	if oldPeer != nil {
+		close(oldPeer.quitC)
 	}
 }
 
@@ -1041,14 +1004,15 @@ func (self *BlockPool) getChild(sec *section) *section {
 
 func newSection() (sec *section) {
 	sec = &section{
-		controlC:    make(chan bool, 1),
+		controlC:    make(chan *peerInfo),
-		suicideC:    make(chan bool, 1),
+		suicideC:    make(chan bool),
-		blockChainC: make(chan bool, 1),
+		blockChainC: make(chan bool),
 		forkC:       make(chan chan bool),
 	}
 	return
 }
 
+// link should only be called under chainLock
 func (self *BlockPool) link(parent *section, child *section) {
 	if parent != nil {
 		exChild := parent.child