GODRIVER-1589 consolidate pool.drain and pool.clear (#397)

Author: iwysiu

x/mongo/driver/topology/pool.go

@@ -179,9 +179,6 @@ func newPool(config poolConfig, connOpts ...ConnectionOption) (*pool, error) {
 	return pool, nil
 }
 
-// drain drains the pool by increasing the generation ID.
-func (p *pool) drain() { atomic.AddUint64(&p.generation, 1) }
-
 // stale checks if a given connection's generation is below the generation of the pool
 func (p *pool) stale(c *connection) bool {
 	return c == nil || c.generation < atomic.LoadUint64(&p.generation)
@@ -473,15 +470,13 @@ func (p *pool) put(c *connection) error {
 	return nil
 }
 
-// clear clears the pool by incrementing the generation and then maintaining the pool
+// clear clears the pool by incrementing the generation
 func (p *pool) clear() {
 	if p.monitor != nil {
 		p.monitor.Event(&event.PoolEvent{
 			Type:    event.PoolCleared,
 			Address: p.address.String(),
 		})
 	}
-
-	p.drain()
-	p.conns.Maintain()
+	atomic.AddUint64(&p.generation, 1)
 }

x/mongo/driver/topology/resource_pool_test.go

@@ -182,6 +182,9 @@ func TestResourcePool(t *testing.T) {
 			assert.Equal(t, int32(5), expiredCalled, "expected expire to be called 5 times, got %v", expiredCalled)
 			closeCalled := ec.getCloseCalled()
 			assert.Equal(t, int32(3), closeCalled, "expected close to be called 3 times, got %v", closeCalled)
+			// rp.maintainTimer should be reset after Maintain runs. Stop() returns true if the
+			// timer was set, ensuring that it was correctly reset.
+			assert.True(t, rp.maintainTimer.Stop(), "expected timer to be reset")
 		})
 	})
 }

x/mongo/driver/topology/server.go

@@ -148,7 +148,7 @@ func NewServer(addr address.Address, opts ...ServerOption) (*Server, error) {
 	}
 	s.desc.Store(description.Server{Addr: addr})
 
-	callback := func(desc description.Server) { s.updateDescription(desc, false) }
+	callback := func(desc description.Server) { s.updateDescription(desc) }
 	pc := poolConfig{
 		Address:     addr,
 		MinPoolSize: cfg.minConns,
@@ -254,7 +254,8 @@ func (s *Server) Connection(ctx context.Context) (driver.Connection, error) {
 			Kind:      description.Unknown,
 			LastError: wrappedConnErr,
 		}
-		s.updateDescription(desc, false)
+		s.updateDescription(desc)
+		s.pool.clear()
 
 		return nil, err
 	}
@@ -324,7 +325,7 @@ func (s *Server) ProcessError(err error) {
 		desc.Kind = description.Unknown
 		desc.LastError = err
 		// updates description to unknown
-		s.updateDescription(desc, false)
+		s.updateDescription(desc)
 		// If the node is shutting down or is older than 4.2, we synchronously clear the pool
 		if cerr.NodeIsShuttingDown() || desc.WireVersion == nil || desc.WireVersion.Max < 8 {
 			s.RequestImmediateCheck()
@@ -337,7 +338,7 @@ func (s *Server) ProcessError(err error) {
 		desc.Kind = description.Unknown
 		desc.LastError = err
 		// updates description to unknown
-		s.updateDescription(desc, false)
+		s.updateDescription(desc)
 		// If the node is shutting down or is older than 4.2, we synchronously clear the pool
 		if wcerr.NodeIsShuttingDown() || desc.WireVersion == nil || desc.WireVersion.Max < 8 {
 			s.RequestImmediateCheck()
@@ -363,7 +364,7 @@ func (s *Server) ProcessError(err error) {
 	desc.Kind = description.Unknown
 	desc.LastError = err
 	// updates description to unknown
-	s.updateDescription(desc, false)
+	s.updateDescription(desc)
 	s.pool.clear()
 }
 
@@ -393,7 +394,7 @@ func (s *Server) update() {
 	var desc description.Server
 
 	desc, conn = s.heartbeat(nil)
-	s.updateDescription(desc, true)
+	s.updateDescription(desc)
 
 	closeServer := func() {
 		doneOnce = true
@@ -433,15 +434,15 @@ func (s *Server) update() {
 		}
 
 		desc, conn = s.heartbeat(conn)
-		s.updateDescription(desc, false)
+		s.updateDescription(desc)
 	}
 }
 
 // updateDescription handles updating the description on the Server, notifying
 // subscribers, and potentially draining the connection pool. The initial
 // parameter is used to determine if this is the first description from the
 // server.
-func (s *Server) updateDescription(desc description.Server, initial bool) {
+func (s *Server) updateDescription(desc description.Server) {
 	defer func() {
 		//  ¯\_(ツ)_/¯
 		_ = recover()
@@ -463,16 +464,6 @@ func (s *Server) updateDescription(desc description.Server, initial bool) {
 		c <- desc
 	}
 	s.subLock.Unlock()
-
-	if initial {
-		// We don't clear the pool on the first update on the description.
-		return
-	}
-
-	switch desc.Kind {
-	case description.Unknown:
-		s.pool.drain()
-	}
 }
 
 // heartbeat sends a heartbeat to the server using the given connection. The connection can be nil.
@@ -555,7 +546,7 @@ func (s *Server) heartbeat(conn *connection) (description.Server, *connection) {
 			saved = err
 			conn = nil
 			if wrappedConnErr := unwrapConnectionError(err); wrappedConnErr != nil {
-				s.pool.drain()
+				s.pool.clear()
 				// If the server is not connected, give up and exit loop
 				if s.Description().Kind == description.Unknown {
 					break

x/mongo/driver/topology/server_test.go

@@ -122,7 +122,6 @@ func TestServer(t *testing.T) {
 				desc = &descript
 				require.Nil(t, desc.LastError)
 			}
-			//err = s.Connect(nil)
 			err = s.pool.connect()
 			require.NoError(t, err, "unable to connect to pool")
 			s.connectionstate = connected
@@ -245,7 +244,7 @@ func TestServer(t *testing.T) {
 		updated.Store(false)
 		s, err := ConnectServer(address.Address("localhost"), func(description.Server) { updated.Store(true) })
 		require.NoError(t, err)
-		s.updateDescription(description.Server{Addr: s.address}, false)
+		s.updateDescription(description.Server{Addr: s.address})
 		require.True(t, updated.Load().(bool))
 	})
 	t.Run("heartbeat", func(t *testing.T) {