cmd/utils, rpc/comms: stop XEth when IPC connection ends

There are a bunch of changes required to make this work:

- in miner: allow unregistering agents, fix RemoteAgent.Stop
- in eth/filters: make FilterSystem.Stop not crash
- in rpc/comms: move listen loop to platform-independent code

Fixes #1930. I ran the shell loop there for a few minutes and didn't see
any changes in the memory profile.

3 files changed, 32 insertions, 20 deletions

diff --git a/miner/miner.go b/miner/miner.go
index 769db79d1..6d4a84f1a 100644
--- a/miner/miner.go
+++ b/miner/miner.go
@@ -133,10 +133,13 @@ func (self *Miner) Register(agent Agent) {
 	if self.Mining() {
 		agent.Start()
 	}
-
 	self.worker.register(agent)
 }
 
+func (self *Miner) Unregister(agent Agent) {
+	self.worker.unregister(agent)
+}
+
 func (self *Miner) Mining() bool {
 	return atomic.LoadInt32(&self.mining) > 0
 }
@@ -146,7 +149,7 @@ func (self *Miner) HashRate() (tot int64) {
 	// do we care this might race? is it worth we're rewriting some
 	// aspects of the worker/locking up agents so we can get an accurate
 	// hashrate?
-	for _, agent := range self.worker.agents {
+	for agent := range self.worker.agents {
 		tot += agent.GetHashRate()
 	}
 	return
diff --git a/miner/remote_agent.go b/miner/remote_agent.go
index 9e4453ce8..18ddf121c 100644
--- a/miner/remote_agent.go
+++ b/miner/remote_agent.go
@@ -48,9 +48,10 @@ type RemoteAgent struct {
 }
 
 func NewRemoteAgent() *RemoteAgent {
-	agent := &RemoteAgent{work: make(map[common.Hash]*Work), hashrate: make(map[common.Hash]hashrate)}
-
-	return agent
+	return &RemoteAgent{
+		work:     make(map[common.Hash]*Work),
+		hashrate: make(map[common.Hash]hashrate),
+	}
 }
 
 func (a *RemoteAgent) SubmitHashrate(id common.Hash, rate uint64) {
@@ -75,8 +76,12 @@ func (a *RemoteAgent) Start() {
 }
 
 func (a *RemoteAgent) Stop() {
-	close(a.quit)
-	close(a.workCh)
+	if a.quit != nil {
+		close(a.quit)
+	}
+	if a.workCh != nil {
+		close(a.workCh)
+	}
 }
 
 // GetHashRate returns the accumulated hashrate of all identifier combined
diff --git a/miner/worker.go b/miner/worker.go
index 3519e1506..2d072ef60 100644
--- a/miner/worker.go
+++ b/miner/worker.go
@@ -92,7 +92,7 @@ type Result struct {
 type worker struct {
 	mu sync.Mutex
 
-	agents []Agent
+	agents map[Agent]struct{}
 	recv   chan *Result
 	mux    *event.TypeMux
 	quit   chan struct{}
@@ -136,6 +136,7 @@ func newWorker(coinbase common.Address, eth core.Backend) *worker {
 		coinbase:       coinbase,
 		txQueue:        make(map[common.Hash]*types.Transaction),
 		quit:           make(chan struct{}),
+		agents:         make(map[Agent]struct{}),
 		fullValidation: false,
 	}
 	go worker.update()
@@ -180,7 +181,7 @@ func (self *worker) start() {
 	atomic.StoreInt32(&self.mining, 1)
 
 	// spin up agents
-	for _, agent := range self.agents {
+	for agent := range self.agents {
 		agent.Start()
 	}
 }
@@ -190,16 +191,14 @@ func (self *worker) stop() {
 	defer self.mu.Unlock()
 
 	if atomic.LoadInt32(&self.mining) == 1 {
-		var keep []Agent
-		// stop all agents
-		for _, agent := range self.agents {
+		// Stop all agents.
+		for agent := range self.agents {
 			agent.Stop()
-			// keep all that's not a cpu agent
-			if _, ok := agent.(*CpuAgent); !ok {
-				keep = append(keep, agent)
+			// Remove CPU agents.
+			if _, ok := agent.(*CpuAgent); ok {
+				delete(self.agents, agent)
 			}
 		}
-		self.agents = keep
 	}
 
 	atomic.StoreInt32(&self.mining, 0)
@@ -209,10 +208,17 @@ func (self *worker) stop() {
 func (self *worker) register(agent Agent) {
 	self.mu.Lock()
 	defer self.mu.Unlock()
-	self.agents = append(self.agents, agent)
+	self.agents[agent] = struct{}{}
 	agent.SetReturnCh(self.recv)
 }
 
+func (self *worker) unregister(agent Agent) {
+	self.mu.Lock()
+	defer self.mu.Unlock()
+	delete(self.agents, agent)
+	agent.Stop()
+}
+
 func (self *worker) update() {
 	eventSub := self.mux.Subscribe(core.ChainHeadEvent{}, core.ChainSideEvent{}, core.TxPreEvent{})
 	defer eventSub.Unsubscribe()
@@ -341,11 +347,9 @@ func (self *worker) push(work *Work) {
 			glog.Infoln("You turn back and abort mining")
 			return
 		}
-
 		// push new work to agents
-		for _, agent := range self.agents {
+		for agent := range self.agents {
 			atomic.AddInt32(&self.atWork, 1)
-
 			if agent.Work() != nil {
 				agent.Work() <- work
 			}