Found data race on Close() (in Router and GoChannel). Version v1.0.0-rc.1

[aMFeriuSS]

Hello! I found some data races (probably).
Am I doing something wrong maybe (wrong closing)?

Code:

type Server struct {
	serverPubSub *gochannel.GoChannel
	clientPubSub *gochannel.GoChannel
}
...

logger := log.NewWatermillLogAdapter(lggr)
ServerPubSub = gochannel.NewGoChannel(gochannel.Config{}, logger)
ClientPubSub = gochannel.NewGoChannel(gochannel.Config{}, logger)

...
defer s.Stop()

func (s *Server) Stop() {
	err := s.clientPubSub.Close()
	if err != nil {
		panic(err)
	}
	err = s.serverPubSub.Close()
	if err != nil {
		panic(err)
	}
}
...

Logs for gochannel.Gochannel.Close() (for message.Publisher)
go test -failfast -count 1 -race ./ledger/light/integration -run '^Test_BootstrapCalls$'

==================
WARNING: DATA RACE
Write at 0x00c0000fa288 by goroutine 7:
  github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/pubsub/gochannel.(*GoChannel).Close()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/pubsub/gochannel/pubsub.go:272 +0xad
  github.com/insolar/insolar/ledger/light/integration_test.(*Server).Stop()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/ledger/light/integration/server_test.go:483 +0x122
  github.com/insolar/insolar/ledger/light/integration_test.Test_BootstrapCalls()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/ledger/light/integration/light_test.go:64 +0x499
  testing.tRunner()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:865 +0x163

Previous read at 0x00c0000fa288 by goroutine 81:
  github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/pubsub/gochannel.(*GoChannel).Publish()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/pubsub/gochannel/pubsub.go:83 +0x61
  github.com/insolar/insolar/insolar/bus.(*Bus).sendTarget()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/insolar/bus/bus.go:261 +0xf81
  github.com/insolar/insolar/insolar/bus.(*Bus).SendRole()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/insolar/bus/bus.go:179 +0x5ba
  github.com/insolar/insolar/ledger/light/proc.(*HotObjects).sendConfirmationToHeavy()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/ledger/light/proc/hot_data.go:175 +0x64f
  github.com/insolar/insolar/ledger/light/proc.(*HotObjects).Proceed()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/ledger/light/proc/hot_data.go:157 +0x1368
  github.com/insolar/insolar/insolar/flow/internal/thread.(*Thread).procedure.func1()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/insolar/flow/internal/thread/thread.go:143 +0x56

Goroutine 7 (running) created at:
  testing.(*T).Run()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:916 +0x65a
  testing.runTests.func1()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:1157 +0xa8
  testing.tRunner()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:865 +0x163
  testing.runTests()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:1155 +0x523
  testing.(*M).Run()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:1072 +0x2eb
  main.main()
      _testmain.go:76 +0x222

And this - data race for message.Router.Close():

		inRouter, _ := message.NewRouter(message.RouterConfig{}, logger)
		outRouter,_ := message.NewRouter(message.RouterConfig{}, logger)
...

defer s.Stop()
...

func (s *Server) Stop() {
	err := s. inRouter.Close()
	if err != nil {
		panic(err)
	}
	err = s. outRouter.Close()
	if err != nil {
		panic(err)
	}
}

Logs for message.Router.Close() (for message.Publisher)
go test -failfast -count 1 -race ./ledger/light/integration -run '^Test_BasicOperations$'

==================
WARNING: DATA RACE
Read at 0x00c00058e5a0 by goroutine 141:
  github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/message.(*Router).Close()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/message/router.go:350 +0x4f
  github.com/insolar/insolar/ledger/light/integration_test.(*Server).Stop()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/ledger/light/integration/server_test.go:499 +0x122
  github.com/insolar/insolar/ledger/light/integration_test.Test_BasicOperations()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/ledger/light/integration/light_test.go:195 +0x4cf
  testing.tRunner()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:865 +0x163

Previous write at 0x00c00058e5a0 by goroutine 229:
  github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/message.(*Router).Close()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/message/router.go:353 +0x6c
  github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/message.(*Router).closeWhenAllHandlersStopped()
      /Users/ilyamarkin/go/src/github.com/insolar/insolar/vendor/github.com/ThreeDotsLabs/watermill/message/router.go:334 +0x168

Goroutine 141 (running) created at:
  testing.(*T).Run()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:916 +0x65a
  testing.runTests.func1()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:1157 +0xa8
  testing.tRunner()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:865 +0x163
  testing.runTests()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:1155 +0x523
  testing.(*M).Run()
      /usr/local/Cellar/go/1.12.7/libexec/src/testing/testing.go:1072 +0x2eb
  main.main()
      _testmain.go:76 +0x222

[m110]

Hey @aMFeriuSS, thanks for posting these. Your code is fine, it's an issue with checking if the pub/sub is closed. I'm looking into it and will post a PR along with additional tests.

[aromancev]

Hey @m110, thanks for fast reply. How is it going there? We are having problems with data races in our tests. Would be awesome to see it fixed soon!

[m110]

Hey @andreyromancev. Currently working on it, I expect to have all races fixed within few days.

[m110]

Fixed in v1.0.0-rc.3.