eth/downloader, les, miner: replace mux with feed

cmd/geth/chaincmd.go

@@ -33,7 +33,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/ethdb"
-	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/trie"
 	"github.com/syndtr/goleveldb/leveldb/util"
@@ -373,7 +372,7 @@ func copyDb(ctx *cli.Context) error {
 	chain, chainDb := utils.MakeChain(ctx, stack)
 
 	syncmode := *utils.GlobalTextMarshaler(ctx, utils.SyncModeFlag.Name).(*downloader.SyncMode)
-	dl := downloader.New(syncmode, chainDb, new(event.TypeMux), chain, nil, nil)
+	dl := downloader.New(syncmode, chainDb, chain, nil, nil)
 
 	// Create a source peer to satisfy downloader requests from
 	db, err := ethdb.NewLDBDatabase(ctx.Args().First(), ctx.GlobalInt(utils.CacheFlag.Name), 256)

eth/backend.go

@@ -262,7 +262,7 @@ func (s *Ethereum) APIs() []rpc.API {
 		}, {
 			Namespace: "eth",
 			Version:   "1.0",
-			Service:   downloader.NewPublicDownloaderAPI(s.protocolManager.downloader, s.eventMux),
+			Service:   downloader.NewPublicDownloaderAPI(s.protocolManager.downloader),
 			Public:    true,
 		}, {
 			Namespace: "miner",

eth/downloader/api.go

@@ -22,30 +22,53 @@ import (
 
 	ethereum "github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
+const (
+	// startChanSize is the size of channel listening to StartEvent.
+	startChanSize = 10
+	// finishChanSize is the size of channel listening to FinishEvent.
+	finishChanSize = 10
+)
+
 // PublicDownloaderAPI provides an API which gives information about the current synchronisation status.
 // It offers only methods that operates on data that can be available to anyone without security risks.
 type PublicDownloaderAPI struct {
 	d *Downloader
-	mux                       *event.TypeMux
+
+	// Channels
+	startCh                   chan StartEvent
+	finishCh                  chan FinishEvent
 	installSyncSubscription   chan chan interface{}
 	uninstallSyncSubscription chan *uninstallSyncSubscriptionRequest
+
+	// Subscriptions
+	startSub  event.Subscription
+	finishSub event.Subscription
 }
 
 // NewPublicDownloaderAPI create a new PublicDownloaderAPI. The API has an internal event loop that
 // listens for events from the downloader through the global event mux. In case it receives one of
 // these events it broadcasts it to all syncing subscriptions that are installed through the
 // installSyncSubscription channel.
-func NewPublicDownloaderAPI(d *Downloader, m *event.TypeMux) *PublicDownloaderAPI {
+func NewPublicDownloaderAPI(d *Downloader) *PublicDownloaderAPI {
 	api := &PublicDownloaderAPI{
 		d:                         d,
-		mux: m,
+		startCh:                   make(chan StartEvent, startChanSize),
+		finishCh:                  make(chan FinishEvent, finishChanSize),
 		installSyncSubscription:   make(chan chan interface{}),
 		uninstallSyncSubscription: make(chan *uninstallSyncSubscriptionRequest),
 	}
 
+	// Subscribe downloader events
+	api.startSub = d.SubscribeStartEvent(api.startCh)
+	api.finishSub = d.SubscribeFinishEvent(api.finishCh)
+	if api.startSub == nil || api.finishSub == nil {
+		log.Crit("Subscribe downloader events failed")
+	}
+
 	go api.eventLoop()
 
 	return api
@@ -54,37 +77,42 @@ func NewPublicDownloaderAPI(d *Downloader, m *event.TypeMux) *PublicDownloaderAP
 // eventLoop runs a loop until the event mux closes. It will install and uninstall new
 // sync subscriptions and broadcasts sync status updates to the installed sync subscriptions.
 func (api *PublicDownloaderAPI) eventLoop() {
-	var (
+	var syncSubscriptions = make(map[chan interface{}]struct{})
-		sub               = api.mux.Subscribe(StartEvent{}, DoneEvent{}, FailedEvent{})
+
-		syncSubscriptions = make(map[chan interface{}]struct{})
+	defer func() {
-	)
+		api.startSub.Unsubscribe()
+		api.finishSub.Unsubscribe()
+	}()
+
+	broadcast := func(notification interface{}) {
+		for c := range syncSubscriptions {
+			c <- notification
+		}
+	}
 
 	for {
 		select {
 		case i := <-api.installSyncSubscription:
 			syncSubscriptions[i] = struct{}{}
+
 		case u := <-api.uninstallSyncSubscription:
 			delete(syncSubscriptions, u.c)
 			close(u.uninstalled)
-		case event := <-sub.Chan():
-			if event == nil {
-				return
-			}
 
-			var notification interface{}
+		case <-api.startCh:
-			switch event.Data.(type) {
+			broadcast(&SyncingResult{
-			case StartEvent:
-				notification = &SyncingResult{
 				Syncing: true,
 				Status:  api.d.Progress(),
-				}
+			})
-			case DoneEvent, FailedEvent:
+
-				notification = false
+		case <-api.finishCh:
-			}
+			broadcast(false)
-			// broadcast
+
-			for c := range syncSubscriptions {
+		case <-api.startSub.Err():
-				c <- notification
+			return
-			}
+
+		case <-api.finishSub.Err():
+			return
 		}
 	}
 }

eth/downloader/downloader.go

@@ -143,6 +143,11 @@ type Downloader struct {
 	quitCh   chan struct{} // Quit channel to signal termination
 	quitLock sync.RWMutex  // Lock to prevent double closes
 
+	// Subscriptions
+	startFeed  event.Feed              // Feed for downloader synchronization start subscription.
+	finishFeed event.Feed              // Feed for downloader synchronization finish subscription.
+	feedScope  event.SubscriptionScope // Manager of all feed subscriptions which can unsubscribe all at once
+
 	// Testing hooks
 	syncInitHook     func(uint64, uint64)  // Method to call upon initiating a new sync run
 	bodyFetchHook    func([]*types.Header) // Method to call upon starting a block body fetch
@@ -198,7 +203,7 @@ type BlockChain interface {
 }
 
 // New creates a new downloader to fetch hashes and blocks from remote peers.
-func New(mode SyncMode, stateDb ethdb.Database, mux *event.TypeMux, chain BlockChain, lightchain LightChain, dropPeer peerDropFn) *Downloader {
+func New(mode SyncMode, stateDb ethdb.Database, chain BlockChain, lightchain LightChain, dropPeer peerDropFn) *Downloader {
 	if lightchain == nil {
 		lightchain = chain
 	}
@@ -206,7 +211,6 @@ func New(mode SyncMode, stateDb ethdb.Database, mux *event.TypeMux, chain BlockC
 	dl := &Downloader{
 		mode:           mode,
 		stateDB:        stateDb,
-		mux:            mux,
 		queue:          newQueue(),
 		peers:          newPeerSet(),
 		rttEstimate:    uint64(rttMaxEstimate),
@@ -402,14 +406,9 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode
 // syncWithPeer starts a block synchronization based on the hash chain from the
 // specified peer and head hash.
 func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.Int) (err error) {
-	d.mux.Post(StartEvent{})
+	d.startFeed.Send(StartEvent{})
 	defer func() {
-		// reset on error
+		d.finishFeed.Send(FinishEvent{Err: err})
-		if err != nil {
-			d.mux.Post(FailedEvent{err})
-		} else {
-			d.mux.Post(DoneEvent{})
-		}
 	}()
 	if p.version < 62 {
 		return errTooOld
@@ -539,6 +538,7 @@ func (d *Downloader) Terminate() {
 
 	// Cancel any pending download requests
 	d.Cancel()
+	d.feedScope.Close()
 }
 
 // fetchHeight retrieves the head header of the remote peer to aid in estimating
@@ -1640,3 +1640,13 @@ func (d *Downloader) requestTTL() time.Duration {
 	}
 	return ttl
 }
+
+// SubscribeStartEvent registers a subscription of StartEvent.
+func (d *Downloader) SubscribeStartEvent(ch chan<- StartEvent) event.Subscription {
+	return d.feedScope.Track(d.startFeed.Subscribe(ch))
+}
+
+// SubscribeFinishEvent registers a subscription of FinishEvent.
+func (d *Downloader) SubscribeFinishEvent(ch chan<- FinishEvent) event.Subscription {
+	return d.feedScope.Track(d.finishFeed.Subscribe(ch))
+}

eth/downloader/downloader_test.go

@@ -31,7 +31,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
-	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/trie"
 )
@@ -96,7 +95,7 @@ func newTester() *downloadTester {
 	tester.stateDb, _ = ethdb.NewMemDatabase()
 	tester.stateDb.Put(genesis.Root().Bytes(), []byte{0x00})
 
-	tester.downloader = New(FullSync, tester.stateDb, new(event.TypeMux), tester, nil, tester.dropPeer)
+	tester.downloader = New(FullSync, tester.stateDb, tester, nil, tester.dropPeer)
 
 	return tester
 }

eth/downloader/events.go

@@ -16,6 +16,11 @@
 
 package downloader
 
-type DoneEvent struct{}
+// StartEvent is posted when downloader start to synchronization
 type StartEvent struct{}
-type FailedEvent struct{ Err error }
+
+// FinishEvent is posted when downloader finish synchronization.
+// If synchronization is not successful, a relevant error will be returned.
+type FinishEvent struct {
+	Err error
+}

eth/handler.go

@@ -159,7 +159,7 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
 		return nil, errIncompatibleConfig
 	}
 	// Construct the different synchronisation mechanisms
-	manager.downloader = downloader.New(mode, chaindb, manager.eventMux, blockchain, nil, manager.removePeer)
+	manager.downloader = downloader.New(mode, chaindb, blockchain, nil, manager.removePeer)
 
 	validator := func(header *types.Header) error {
 		return engine.VerifyHeader(blockchain, header, true)

les/backend.go

@@ -186,7 +186,7 @@ func (s *LightEthereum) APIs() []rpc.API {
 		}, {
 			Namespace: "eth",
 			Version:   "1.0",
-			Service:   downloader.NewPublicDownloaderAPI(s.protocolManager.downloader, s.eventMux),
+			Service:   downloader.NewPublicDownloaderAPI(s.protocolManager.downloader),
 			Public:    true,
 		}, {
 			Namespace: "eth",

les/handler.go

@@ -205,7 +205,7 @@ func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, protoco
 	}
 
 	if lightSync {
-		manager.downloader = downloader.New(downloader.LightSync, chainDb, manager.eventMux, nil, blockchain, removePeer)
+		manager.downloader = downloader.New(downloader.LightSync, chainDb, nil, blockchain, removePeer)
 		manager.peers.notify((*downloaderPeerNotify)(manager))
 		manager.fetcher = newLightFetcher(manager)
 	}

miner/miner.go

@@ -76,7 +76,7 @@ func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine con
 // the loop is exited. This to prevent a major security vuln where external parties can DOS you with blocks
 // and halt your mining operation for as long as the DOS continues.
 func (self *Miner) update() {
-	events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
+	events := self.mux.Subscribe(downloader.StartEvent{}, downloader.FinishEvent{})
 out:
 	for ev := range events.Chan() {
 		switch ev.Data.(type) {
@@ -87,7 +87,7 @@ out:
 				atomic.StoreInt32(&self.shouldStart, 1)
 				log.Info("Mining aborted due to sync")
 			}
-		case downloader.DoneEvent, downloader.FailedEvent:
+		case downloader.FinishEvent:
 			shouldStart := atomic.LoadInt32(&self.shouldStart) == 1
 
 			atomic.StoreInt32(&self.canStart, 1)