diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index d3086921b9..aafa008561 100644 --- a/cmd/geth/chaincmd.go +++ b/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) diff --git a/eth/backend.go b/eth/backend.go index e4c5fef3e4..52807d110f 100644 --- a/eth/backend.go +++ b/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", diff --git a/eth/downloader/api.go b/eth/downloader/api.go index 91c6322d41..4a786272c2 100644 --- a/eth/downloader/api.go +++ b/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 + d *Downloader + + // 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, + d: d, + 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 ( - sub = api.mux.Subscribe(StartEvent{}, DoneEvent{}, FailedEvent{}) - syncSubscriptions = make(map[chan interface{}]struct{}) - ) + var 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{} - switch event.Data.(type) { - case StartEvent: - notification = &SyncingResult{ - Syncing: true, - Status: api.d.Progress(), - } - case DoneEvent, FailedEvent: - notification = false - } - // broadcast - for c := range syncSubscriptions { - c <- notification - } + case <-api.startCh: + broadcast(&SyncingResult{ + Syncing: true, + Status: api.d.Progress(), + }) + + case <-api.finishCh: + broadcast(false) + + case <-api.startSub.Err(): + return + + case <-api.finishSub.Err(): + return } } } diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index dc23354929..49824289fc 100644 --- a/eth/downloader/downloader.go +++ b/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 - if err != nil { - d.mux.Post(FailedEvent{err}) - } else { - d.mux.Post(DoneEvent{}) - } + d.finishFeed.Send(FinishEvent{Err: err}) }() 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)) +} diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index e85e234c0e..5d11b08c91 100644 --- a/eth/downloader/downloader_test.go +++ b/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 } diff --git a/eth/downloader/events.go b/eth/downloader/events.go index 64905b8f2a..7a76ca716b 100644 --- a/eth/downloader/events.go +++ b/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 +} diff --git a/eth/handler.go b/eth/handler.go index c8f7e13f16..2972946017 100644 --- a/eth/handler.go +++ b/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) diff --git a/les/backend.go b/les/backend.go index 2d8ada7b0e..fdf78ef948 100644 --- a/les/backend.go +++ b/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", diff --git a/les/handler.go b/les/handler.go index 22899eb1bc..be9797b439 100644 --- a/les/handler.go +++ b/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) } diff --git a/miner/miner.go b/miner/miner.go index d9256e9787..6fecc2523b 100644 --- a/miner/miner.go +++ b/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)