GODRIVER-898 make topology update serverDescriptions synchronously

Change-Id: I0a9e1719adced1ad315d32d82d04e9d1585717e1

Author: Isabella Siu

mongo/transactions_test.go

@@ -142,7 +142,6 @@ func runTransactionTestFile(t *testing.T, filepath string) {
 
 func runTransactionsTestCase(t *testing.T, test *transTestCase, testfile transTestFile, dbAdmin *Database) {
 	t.Run(test.Description, func(t *testing.T) {
-
 		// kill sessions from previously failed tests
 		killSessions(t, dbAdmin.client)
 

x/mongo/driver/topology/connection.go

@@ -60,7 +60,7 @@ func (sc *sconn) processErr(err error) {
 		return
 	}
 
-	ne, ok := err.(connection.NetworkError)
+	ne, ok := err.(connection.Error)
 	if !ok {
 		return
 	}

x/mongo/driver/topology/connection_test.go

@@ -33,7 +33,7 @@ func (n netErr) Temporary() bool {
 }
 
 type connect struct {
-	err *connection.NetworkError
+	err *connection.Error
 }
 
 func (c connect) WriteWireMessage(ctx context.Context, wm wiremessage.WireMessage) error {
@@ -58,7 +58,7 @@ func (c connect) ID() string {
 // Test case for sconn processErr
 func TestConnectionProcessErrSpec(t *testing.T) {
 	ctx := context.Background()
-	s, err := NewServer(address.Address("localhost"))
+	s, err := NewServer(address.Address("localhost"), nil)
 	require.NoError(t, err)
 
 	desc := s.Description()
@@ -67,7 +67,7 @@ func TestConnectionProcessErrSpec(t *testing.T) {
 	s.connectionstate = connected
 
 	innerErr := netErr{}
-	connectErr := connection.NetworkError{"blah", innerErr}
+	connectErr := connection.Error{ConnectionID: "blah", Wrapped: innerErr}
 	c := connect{&connectErr}
 	sc := sconn{c, s, 1}
 	err = sc.WriteWireMessage(ctx, nil)

x/mongo/driver/topology/server.go

@@ -99,12 +99,14 @@ type Server struct {
 	currentSubscriberID uint64
 
 	subscriptionsClosed bool
+
+	updateTopologyCallback atomic.Value
 }
 
 // ConnectServer creates a new Server and then initializes it using the
 // Connect method.
-func ConnectServer(ctx context.Context, addr address.Address, opts ...ServerOption) (*Server, error) {
-	srvr, err := NewServer(addr, opts...)
+func ConnectServer(ctx context.Context, addr address.Address, topo func(description.Server), opts ...ServerOption) (*Server, error) {
+	srvr, err := NewServer(addr, topo, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -117,7 +119,7 @@ func ConnectServer(ctx context.Context, addr address.Address, opts ...ServerOpti
 
 // NewServer creates a new server. The mongodb server at the address will be monitored
 // on an internal monitoring goroutine.
-func NewServer(addr address.Address, opts ...ServerOption) (*Server, error) {
+func NewServer(addr address.Address, topo func(description.Server), opts ...ServerOption) (*Server, error) {
 	cfg, err := newServerConfig(opts...)
 	if err != nil {
 		return nil, err
@@ -133,6 +135,7 @@ func NewServer(addr address.Address, opts ...ServerOption) (*Server, error) {
 		subscribers: make(map[uint64]chan description.Server),
 	}
 	s.desc.Store(description.Server{Addr: addr})
+	s.updateTopologyCallback.Store(topo)
 
 	var maxConns uint64
 	if cfg.maxConns == 0 {
@@ -175,6 +178,8 @@ func (s *Server) Disconnect(ctx context.Context) error {
 		return ErrServerClosed
 	}
 
+	s.updateTopologyCallback.Store((func(description.Server))(nil))
+
 	// For every call to Connect there must be at least 1 goroutine that is
 	// waiting on the done channel.
 	s.done <- struct{}{}
@@ -373,6 +378,11 @@ func (s *Server) updateDescription(desc description.Server, initial bool) {
 	}()
 	s.desc.Store(desc)
 
+	topo := s.updateTopologyCallback.Load().(func(description.Server))
+	if topo != nil {
+		topo(desc)
+	}
+
 	s.subLock.Lock()
 	for _, c := range s.subscribers {
 		select {

x/mongo/driver/topology/server_test.go

@@ -74,7 +74,7 @@ func TestServer(t *testing.T) {
 
 	for _, tt := range serverTestTable {
 		t.Run(tt.name, func(t *testing.T) {
-			s, err := NewServer(address.Address("localhost"))
+			s, err := NewServer(address.Address("localhost"), nil)
 			require.NoError(t, err)
 
 			var desc *description.Server
@@ -103,7 +103,7 @@ func TestServer(t *testing.T) {
 		})
 	}
 	t.Run("WriteConcernError", func(t *testing.T) {
-		s, err := NewServer(address.Address("localhost"))
+		s, err := NewServer(address.Address("localhost"), nil)
 		require.NoError(t, err)
 
 		var desc *description.Server
@@ -127,7 +127,7 @@ func TestServer(t *testing.T) {
 		require.Equal(t, drained, true)
 	})
 	t.Run("no WriteConcernError", func(t *testing.T) {
-		s, err := NewServer(address.Address("localhost"))
+		s, err := NewServer(address.Address("localhost"), nil)
 		require.NoError(t, err)
 
 		var desc *description.Server
@@ -148,4 +148,12 @@ func TestServer(t *testing.T) {
 		drained := s.pool.(*pool).drainCalled.Load().(bool)
 		require.Equal(t, drained, false)
 	})
+	t.Run("update topology", func(t *testing.T) {
+		var updated bool
+		s, err := NewServer(address.Address("localhost"), func(description.Server) { updated = true })
+		require.NoError(t, err)
+		s.updateDescription(description.Server{Addr: s.address}, false)
+		require.True(t, updated)
+
+	})
 }

x/mongo/driver/topology/topology.go

@@ -63,9 +63,7 @@ type Topology struct {
 
 	done chan struct{}
 
-	fsm       *fsm
-	changes   chan description.Server
-	changeswg sync.WaitGroup
+	fsm *fsm
 
 	SessionPool *session.Pool
 
@@ -84,8 +82,6 @@ type Topology struct {
 	serversLock   sync.Mutex
 	serversClosed bool
 	servers       map[address.Address]*Server
-
-	wg sync.WaitGroup
 }
 
 // New creates a new topology.
@@ -99,7 +95,6 @@ func New(opts ...Option) (*Topology, error) {
 		cfg:         cfg,
 		done:        make(chan struct{}),
 		fsm:         newFSM(),
-		changes:     make(chan description.Server),
 		subscribers: make(map[uint64]chan description.Topology),
 		servers:     make(map[address.Address]*Server),
 	}
@@ -134,9 +129,6 @@ func (t *Topology) Connect(ctx context.Context) error {
 	}
 	t.serversLock.Unlock()
 
-	go t.update()
-	t.changeswg.Add(1)
-
 	t.subscriptionsClosed = false // explicitly set in case topology was disconnected and then reconnected
 
 	atomic.StoreInt32(&t.connectionstate, connected)
@@ -154,16 +146,25 @@ func (t *Topology) Disconnect(ctx context.Context) error {
 		return ErrTopologyClosed
 	}
 
+	servers := make(map[address.Address]*Server)
 	t.serversLock.Lock()
 	t.serversClosed = true
 	for addr, server := range t.servers {
-		t.removeServer(ctx, addr, server)
+		servers[addr] = server
 	}
 	t.serversLock.Unlock()
 
-	t.wg.Wait()
-	t.done <- struct{}{}
-	t.changeswg.Wait()
+	for _, server := range servers {
+		_ = server.Disconnect(ctx)
+	}
+
+	t.subLock.Lock()
+	for id, ch := range t.subscribers {
+		close(ch)
+		delete(t.subscribers, id)
+	}
+	t.subscriptionsClosed = true
+	t.subLock.Unlock()
 
 	t.desc.Store(description.Topology{})
 
@@ -328,110 +329,73 @@ func (t *Topology) selectServer(ctx context.Context, subscriptionCh <-chan descr
 	}
 }
 
-func (t *Topology) update() {
-	defer t.changeswg.Done()
-	defer func() {
-		//  ¯\_(ツ)_/¯
-		if r := recover(); r != nil {
-			<-t.done
-		}
-	}()
+func (t *Topology) apply(ctx context.Context, desc description.Server) {
+	var err error
 
-	for {
-		select {
-		case change := <-t.changes:
-			current, err := t.apply(context.TODO(), change)
-			if err != nil {
-				continue
-			}
+	t.serversLock.Lock()
+	defer t.serversLock.Unlock()
 
-			t.desc.Store(current)
-			t.subLock.Lock()
-			for _, ch := range t.subscribers {
-				// We drain the description if there's one in the channel
-				select {
-				case <-ch:
-				default:
-				}
-				ch <- current
-			}
-			t.subLock.Unlock()
-		case <-t.done:
-			t.subLock.Lock()
-			for id, ch := range t.subscribers {
-				close(ch)
-				delete(t.subscribers, id)
-			}
-			t.subscriptionsClosed = true
-			t.subLock.Unlock()
-			return
-		}
+	if _, ok := t.servers[desc.Addr]; t.serversClosed || !ok {
+		return
 	}
-}
 
-func (t *Topology) apply(ctx context.Context, desc description.Server) (description.Topology, error) {
-	var err error
 	prev := t.fsm.Topology
 
 	current, err := t.fsm.apply(desc)
 	if err != nil {
-		return description.Topology{}, err
+		return
 	}
 
 	diff := description.DiffTopology(prev, current)
-	t.serversLock.Lock()
-	if t.serversClosed {
-		t.serversLock.Unlock()
-		return description.Topology{}, nil
-	}
 
 	for _, removed := range diff.Removed {
 		if s, ok := t.servers[removed.Addr]; ok {
-			t.removeServer(ctx, removed.Addr, s)
+			go func() {
+				cancelCtx, cancel := context.WithCancel(ctx)
+				cancel()
+				_ = s.Disconnect(cancelCtx)
+			}()
+			delete(t.servers, removed.Addr)
 		}
 	}
 
 	for _, added := range diff.Added {
 		_ = t.addServer(ctx, added.Addr)
 	}
-	t.serversLock.Unlock()
-	return current, nil
+
+	t.desc.Store(current)
+
+	t.subLock.Lock()
+	for _, ch := range t.subscribers {
+		// We drain the description if there's one in the channel
+		select {
+		case <-ch:
+		default:
+		}
+		ch <- current
+	}
+	t.subLock.Unlock()
+
 }
 
 func (t *Topology) addServer(ctx context.Context, addr address.Address) error {
 	if _, ok := t.servers[addr]; ok {
 		return nil
 	}
 
-	svr, err := ConnectServer(ctx, addr, t.cfg.serverOpts...)
-	if err != nil {
-		return err
+	topoFunc := func(desc description.Server) {
+		t.apply(context.TODO(), desc)
 	}
-
-	t.servers[addr] = svr
-	var sub *ServerSubscription
-	sub, err = svr.Subscribe()
+	svr, err := ConnectServer(ctx, addr, topoFunc, t.cfg.serverOpts...)
 	if err != nil {
 		return err
 	}
 
-	t.wg.Add(1)
-	go func() {
-		for c := range sub.C {
-			t.changes <- c
-		}
-
-		t.wg.Done()
-	}()
+	t.servers[addr] = svr
 
 	return nil
 }
 
-func (t *Topology) removeServer(ctx context.Context, addr address.Address, server *Server) {
-	_ = server.Disconnect(ctx)
-	delete(t.servers, addr)
-}
-
 // String implements the Stringer interface
 func (t *Topology) String() string {
 	desc := t.Description()