TestNewAdjustTimeFail goroutine leaks

[charlesxsh]

System information

OS & Version: Linux
Commit hash : 56e9001

Expected behaviour

All goroutines exit properly

Actual behaviour

//accounts/abi/bind/backends/simulated_test.go
func TestNewAdjustTimeFail(t *testing.T) {
	testAddr := crypto.PubkeyToAddress(testKey.PublicKey)
	sim := simTestBackend(testAddr)

	...
}

Function simTestBackend(testAddr) creates two goroutines. Goroutine-1 blocks in a select and waits for a message from one of the four error channels. If Goroutine-1 can continue, it will unblock goroutine-2.

//accounts/abi/bind/backends/simulated_test.go
func simTestBackend(testAddr common.Address) *SimulatedBackend {
	return NewSimulatedBackend(
		core.GenesisAlloc{
			testAddr: {Balance: big.NewInt(10000000000000000)},
		}, 10000000,
	)
}

//accounts/abi/bind/backends/simulated.go
func NewSimulatedBackend(alloc core.GenesisAlloc, gasLimit uint64) *SimulatedBackend {
	return NewSimulatedBackendWithDatabase(rawdb.NewMemoryDatabase(), alloc, gasLimit)
}

//accounts/abi/bind/backends/simulated.go
func NewSimulatedBackendWithDatabase(database ethdb.Database, alloc core.GenesisAlloc, gasLimit uint64) *SimulatedBackend {
	genesis := core.Genesis{Config: params.AllEthashProtocolChanges, GasLimit: gasLimit, Alloc: alloc}
	genesis.MustCommit(database)
	blockchain, _ := core.NewBlockChain(database, nil, genesis.Config, ethash.NewFaker(), vm.Config{}, nil, nil)

	backend := &SimulatedBackend{
		database:   database,
		blockchain: blockchain,
		config:     genesis.Config,
		events:     filters.NewEventSystem(&filterBackend{database, blockchain}, false),
	}
	backend.rollback(blockchain.CurrentBlock())
	return backend
}

//eth/filters/filter_system.go
func NewEventSystem(backend Backend, lightMode bool) *EventSystem {
	m := &EventSystem{
		backend:       backend,
		lightMode:     lightMode,
		install:       make(chan *subscription),
		uninstall:     make(chan *subscription),
		txsCh:         make(chan core.NewTxsEvent, txChanSize),
		logsCh:        make(chan []*types.Log, logsChanSize),
		rmLogsCh:      make(chan core.RemovedLogsEvent, rmLogsChanSize),
		pendingLogsCh: make(chan []*types.Log, logsChanSize),
		chainCh:       make(chan core.ChainEvent, chainEvChanSize),
	}

	// Subscribe events
	m.txsSub = m.backend.SubscribeNewTxsEvent(m.txsCh)
	m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh)
	m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh)
	m.chainSub = m.backend.SubscribeChainEvent(m.chainCh)
	m.pendingLogsSub = m.backend.SubscribePendingLogsEvent(m.pendingLogsCh)

	...

	go m.eventLoop()  // line 1: create the first goroutine 
	return m
}

The first goroutine is created at line 1. It keeps executing a select in a loop and does not leave from the loop until it receives an err message from one of the four error channels or one of the four error channels is closed.

After it leaves from the loop, it executes a deferred function, which can unblock the second goroutine.

//eth/filters/filter_system.go
func (es *EventSystem) eventLoop() {
	// Ensure all subscriptions get cleaned up
	defer func() {
		es.txsSub.Unsubscribe()
		es.logsSub.Unsubscribe()
		es.rmLogsSub.Unsubscribe()
		es.pendingLogsSub.Unsubscribe()  // unblock the second goroutine
		es.chainSub.Unsubscribe()
	}()

	index := make(filterIndex)
	for i := UnknownSubscription; i < LastIndexSubscription; i++ {
		index[i] = make(map[rpc.ID]*subscription)
	}

	for {
		select {  // blocking
		case ev := <-es.txsCh:
			es.handleTxsEvent(index, ev)
		case ev := <-es.logsCh:
			es.handleLogs(index, ev)
		case ev := <-es.rmLogsCh:
			es.handleRemovedLogs(index, ev)
		case ev := <-es.pendingLogsCh:
			es.handlePendingLogs(index, ev)
		case ev := <-es.chainCh:
			es.handleChainEvent(index, ev)

		case f := <-es.install:
			if f.typ == MinedAndPendingLogsSubscription {
				// the type are logs and pending logs subscriptions
				index[LogsSubscription][f.id] = f
				index[PendingLogsSubscription][f.id] = f
			} else {
				index[f.typ][f.id] = f
			}
			close(f.installed)

		case f := <-es.uninstall:
			if f.typ == MinedAndPendingLogsSubscription {
				// the type are logs and pending logs subscriptions
				delete(index[LogsSubscription], f.id)
				delete(index[PendingLogsSubscription], f.id)
			} else {
				delete(index[f.typ], f.id)
			}
			close(f.err)

		// System stopped
		case <-es.txsSub.Err():
			return
		case <-es.logsSub.Err():
			return
		case <-es.rmLogsSub.Err():
			return
		case <-es.chainSub.Err():
			return
		}
	}
}

//event/subscription.go
func (s *funcSub) Unsubscribe() {
	s.mu.Lock()
	if s.unsubscribed {
		s.mu.Unlock()
		return
	}
	s.unsubscribed = true
	close(s.unsub) // unblock the second goroutine
	s.mu.Unlock()
	// Wait for producer shutdown.
	<-s.err
}

The following call chain leads the second goroutine to be created.

//accounts/abi/bind/backends/simulated.go
func (fb *filterBackend) SubscribePendingLogsEvent(ch chan<- []*types.Log) event.Subscription {
	return nullSubscription()
}

//accounts/abi/bind/backends/simulated.go
func nullSubscription() event.Subscription {
	return event.NewSubscription(func(quit <-chan struct{}) error {
		<-quit // blocking
		return nil
	})
}

//event/subscription.go
func NewSubscription(producer func(<-chan struct{}) error) Subscription {
	s := &funcSub{unsub: make(chan struct{}), err: make(chan error, 1)}
	go func() {  // line 2: second goroutine
		defer close(s.err)       // line 4
		err := producer(s.unsub) // line 3
		s.mu.Lock()
		defer s.mu.Unlock()
		if !s.unsubscribed {
			if err != nil {
				s.err <- err
			}
			s.unsubscribed = true
		}
	}()
	return s
}

The second goroutine is created in function NewSubscription() at line 2. Function producer() called at line 3 is the anonymous function created in function nullSubscription(). The second goroutine is waiting to pull a message from channel quit, which is the unsub field of the funcSub struct created in function NewSubscription().

Steps to reproduce the behaviour

Running TestNewAdjustTimeFail at github.com/ethereum/go-ethereum/accounts/abi/bind/backends package.

Backtrace

[backtrace]
...

goroutine 31 [chan receive]:
github.com/ethereum/go-ethereum/accounts/abi/bind/backends.nullSubscription.func1(0xc00008cf80, 0x0, 0x0)
	/fuzz/target/accounts/abi/bind/backends/simulated.go:1546 +0x85
github.com/ethereum/go-ethereum/event.NewSubscription.func1(0xc000520150, 0xc8c020)
	/fuzz/target/event/subscription.go:54 +0x69
created by github.com/ethereum/go-ethereum/event.NewSubscription
	/fuzz/target/event/subscription.go:52 +0x151

goroutine 32 [chan receive]:
github.com/ethereum/go-ethereum/accounts/abi/bind/backends.nullSubscription.func1(0xc00008d300, 0x0, 0x0)
	/fuzz/target/accounts/abi/bind/backends/simulated.go:1546 +0x85
github.com/ethereum/go-ethereum/event.NewSubscription.func1(0xc0005201c0, 0xc8c020)
	/fuzz/target/event/subscription.go:54 +0x69
created by github.com/ethereum/go-ethereum/event.NewSubscription
	/fuzz/target/event/subscription.go:52 +0x151

goroutine 33 [select]:
github.com/ethereum/go-ethereum/eth/filters.(*EventSystem).eventLoop(0xc0001be210)
	/fuzz/target/eth/filters/filter_system.go:1122 +0x1b05
created by github.com/ethereum/go-ethereum/eth/filters.NewEventSystem
	/fuzz/target/eth/filters/filter_system.go:139 +0x57c

When submitting logs: please submit them as text and not screenshots.

[ligi]

Would you be willing to submit a PR to fix the leak?