core, eth, event, miner, xeth: fix event post / subscription race

2 files changed, 13 insertions, 8 deletions

diff --git a/miner/miner.go b/miner/miner.go
index b550ed6d6..769db79d1 100644
--- a/miner/miner.go
+++ b/miner/miner.go
@@ -66,7 +66,7 @@ func (self *Miner) update() {
 	events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
 out:
 	for ev := range events.Chan() {
-		switch ev.(type) {
+		switch ev.Data.(type) {
 		case downloader.StartEvent:
 			atomic.StoreInt32(&self.canStart, 0)
 			if self.Mining() {
diff --git a/miner/worker.go b/miner/worker.go
index 8be2db93e..43f6f9909 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -215,13 +215,20 @@ func (self *worker) register(agent Agent) {
 }
 
 func (self *worker) update() {
-	events := self.mux.Subscribe(core.ChainHeadEvent{}, core.ChainSideEvent{}, core.TxPreEvent{})
+	eventSub := self.mux.Subscribe(core.ChainHeadEvent{}, core.ChainSideEvent{}, core.TxPreEvent{})
+	defer eventSub.Unsubscribe()
 
-out:
+	eventCh := eventSub.Chan()
 	for {
 		select {
-		case event := <-events.Chan():
-			switch ev := event.(type) {
+		case event, ok := <-eventCh:
+			if !ok {
+				// Event subscription closed, set the channel to nil to stop spinning
+				eventCh = nil
+				continue
+			}
+			// A real event arrived, process interesting content
+			switch ev := event.Data.(type) {
 			case core.ChainHeadEvent:
 				self.commitNewWork()
 			case core.ChainSideEvent:
@@ -237,11 +244,9 @@ out:
 				}
 			}
 		case <-self.quit:
-			break out
+			return
 		}
 	}
-
-	events.Unsubscribe()
 }
 
 func newLocalMinedBlock(blockNumber uint64, prevMinedBlocks *uint64RingBuffer) (minedBlocks *uint64RingBuffer) {