les: implement request distributor, fix blocking issues (#3660)

* les: implement request distributor, fix blocking issues
* core: moved header validation before chain mutex lock

2 files changed, 23 insertions, 11 deletions

diff --git a/light/lightchain.go b/light/lightchain.go
index 4370dc0fc..4715d47ab 100644
--- a/light/lightchain.go
+++ b/light/lightchain.go
@@ -20,6 +20,7 @@ import (
 	"math/big"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
@@ -369,9 +370,17 @@ func (self *LightChain) postChainEvents(events []interface{}) {
 // In the case of a light chain, InsertHeaderChain also creates and posts light
 // chain events when necessary.
 func (self *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) (int, error) {
+	start := time.Now()
+	if i, err := self.hc.ValidateHeaderChain(chain, checkFreq); err != nil {
+		return i, err
+	}
+
 	// Make sure only one thread manipulates the chain at once
 	self.chainmu.Lock()
-	defer self.chainmu.Unlock()
+	defer func() {
+		self.chainmu.Unlock()
+		time.Sleep(time.Millisecond * 10) // ugly hack; do not hog chain lock in case syncing is CPU-limited by validation
+	}()
 
 	self.wg.Add(1)
 	defer self.wg.Done()
@@ -397,7 +406,7 @@ func (self *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int)
 		}
 		return err
 	}
-	i, err := self.hc.InsertHeaderChain(chain, checkFreq, whFunc)
+	i, err := self.hc.InsertHeaderChain(chain, whFunc, start)
 	go self.postChainEvents(events)
 	return i, err
 }
diff --git a/light/txpool.go b/light/txpool.go
index 28c8d8ca5..5eb1ba801 100644
--- a/light/txpool.go
+++ b/light/txpool.go
@@ -276,15 +276,17 @@ func (pool *TxPool) setNewHead(ctx context.Context, newHeader *types.Header) (tx
 	// clear old mined tx entries of old blocks
 	if idx := newHeader.Number.Uint64(); idx > pool.clearIdx+txPermanent {
 		idx2 := idx - txPermanent
-		for i := pool.clearIdx; i < idx2; i++ {
-			hash := core.GetCanonicalHash(pool.chainDb, i)
-			if list, ok := pool.mined[hash]; ok {
-				hashes := make([]common.Hash, len(list))
-				for i, tx := range list {
-					hashes[i] = tx.Hash()
+		if len(pool.mined) > 0 {
+			for i := pool.clearIdx; i < idx2; i++ {
+				hash := core.GetCanonicalHash(pool.chainDb, i)
+				if list, ok := pool.mined[hash]; ok {
+					hashes := make([]common.Hash, len(list))
+					for i, tx := range list {
+						hashes[i] = tx.Hash()
+					}
+					pool.relay.Discard(hashes)
+					delete(pool.mined, hash)
 				}
-				pool.relay.Discard(hashes)
-				delete(pool.mined, hash)
 			}
 		}
 		pool.clearIdx = idx2
@@ -303,15 +305,16 @@ func (pool *TxPool) eventLoop() {
 	for ev := range pool.events.Chan() {
 		switch ev.Data.(type) {
 		case core.ChainHeadEvent:
+			head := pool.chain.CurrentHeader()
 			pool.mu.Lock()
 			ctx, _ := context.WithTimeout(context.Background(), blockCheckTimeout)
-			head := pool.chain.CurrentHeader()
 			txc, _ := pool.setNewHead(ctx, head)
 			m, r := txc.getLists()
 			pool.relay.NewHead(pool.head, m, r)
 			pool.homestead = pool.config.IsHomestead(head.Number)
 			pool.signer = types.MakeSigner(pool.config, head.Number)
 			pool.mu.Unlock()
+			time.Sleep(time.Millisecond) // hack in order to avoid hogging the lock; this part will be replaced by a subsequent PR
 		}
 	}
 }