Enable ClientStressTest and fix pool test.

Author: matthewdale

mongo/integration/client_test.go

@@ -19,6 +19,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson/bsoncodec"
 	"go.mongodb.org/mongo-driver/bson/bsonrw"
 	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/event"
 	"go.mongodb.org/mongo-driver/internal"
 	"go.mongodb.org/mongo-driver/internal/testutil"
 	"go.mongodb.org/mongo-driver/internal/testutil/assert"
@@ -469,15 +470,13 @@ func TestClient(t *testing.T) {
 }
 
 func TestClientStress(t *testing.T) {
-	// TODO: Enable with GODRIVER-2038.
-	t.Skip("TODO: Enable with GODRIVER-2038")
-
 	if testing.Short() {
 		t.Skip("skipping integration test in short mode")
 	}
 
 	mtOpts := mtest.NewOptions().CreateClient(false)
 	mt := mtest.New(t, mtOpts)
+	defer mt.Close()
 
 	// Test that a Client can recover from a massive traffic spike after the traffic spike is over.
 	mt.Run("Client recovers from traffic spike", func(mt *mtest.T) {
@@ -505,6 +504,7 @@ func TestClientStress(t *testing.T) {
 				if err != nil {
 					errs = append(errs, err)
 				}
+				time.Sleep(10 * time.Microsecond)
 			}
 			return errs
 		}
@@ -523,34 +523,52 @@ func TestClientStress(t *testing.T) {
 		}
 		assert.True(mt, maxRTT > 0, "RTT must be greater than 0")
 
-		// Run tests with various "maxPoolSize" values, including 1-connection pools and unlimited
-		// size pools, to test how the client handles traffic spikes using different connection pool
-		// configurations.
+		// Run tests with various "maxPoolSize" values, including 1-connection pools and the default
+		// size of 100, to test how the client handles traffic spikes using different connection
+		// pool configurations.
 		maxPoolSizes := []uint64{0, 1, 10, 100}
 		for _, maxPoolSize := range maxPoolSizes {
-			maxPoolSizeOpt := mtest.NewOptions().ClientOptions(options.Client().SetMaxPoolSize(maxPoolSize))
+			tpm := newTestPoolMonitor()
+			maxPoolSizeOpt := mtest.NewOptions().ClientOptions(
+				options.Client().
+					SetPoolMonitor(tpm.PoolMonitor).
+					SetMaxPoolSize(maxPoolSize))
 			mt.RunOpts(fmt.Sprintf("maxPoolSize %d", maxPoolSize), maxPoolSizeOpt, func(mt *mtest.T) {
+				// Print the count of connection created, connection closed, and pool clear events
+				// collected during the test to help with debugging.
+				defer func() {
+					created := len(tpm.Events(func(e *event.PoolEvent) bool { return e.Type == event.ConnectionCreated }))
+					closed := len(tpm.Events(func(e *event.PoolEvent) bool { return e.Type == event.ConnectionClosed }))
+					poolCleared := len(tpm.Events(func(e *event.PoolEvent) bool { return e.Type == event.PoolCleared }))
+					mt.Logf("Connections created: %d, connections closed: %d, pool clears: %d", created, closed, poolCleared)
+				}()
+
 				doc := bson.D{{Key: "_id", Value: oid}, {Key: "key", Value: "value"}}
 				_, err := mt.Coll.InsertOne(context.Background(), doc)
 				assert.Nil(mt, err, "InsertOne error: %v", err)
 
-				// Set the timeout to be 10x the maximum observed RTT. Use a minimum 10ms timeout to
+				// Set the timeout to be 100x the maximum observed RTT. Use a minimum 100ms timeout to
 				// prevent spurious test failures due to extremely low timeouts.
-				timeout := maxRTT * 10
-				if timeout < 10*time.Millisecond {
-					timeout = 10 * time.Millisecond
+				timeout := maxRTT * 100
+				minTimeout := 100 * time.Millisecond
+				if timeout < minTimeout {
+					timeout = minTimeout
 				}
-				t.Logf("Max RTT %v; using a timeout of %v", maxRTT, timeout)
+				mt.Logf("Max RTT %v; using a timeout of %v", maxRTT, timeout)
+
+				// Warm up the client for 1 second to allow connections to be established. Ignore
+				// any errors.
+				_ = findOneFor(mt.Coll, timeout, 1*time.Second)
 
 				// Simulate normal traffic by running one FindOne loop for 1 second and assert that there
 				// are no errors.
 				errs := findOneFor(mt.Coll, timeout, 1*time.Second)
 				assert.True(mt, len(errs) == 0, "expected no errors, but got %d (%v)", len(errs), errs)
 
-				// Simulate an extreme traffic spike by running 1,000 FindOne loops in parallel for 10
+				// Simulate an extreme traffic spike by running 2,000 FindOne loops in parallel for 10
 				// seconds and expect at least some errors to occur.
 				g := new(errgroup.Group)
-				for i := 0; i < 1000; i++ {
+				for i := 0; i < 2000; i++ {
 					g.Go(func() error {
 						errs := findOneFor(mt.Coll, timeout, 10*time.Second)
 						if len(errs) == 0 {
@@ -560,11 +578,11 @@ func TestClientStress(t *testing.T) {
 					})
 				}
 				err = g.Wait()
-				assert.NotNil(mt, err, "expected at least one error, got nil")
+				mt.Logf("Error from extreme traffic spike (errors are expected): %v", err)
 
-				// Simulate normal traffic again for 1 second. Ignore any errors to allow any outstanding
+				// Simulate normal traffic again for 5 second. Ignore any errors to allow any outstanding
 				// connection errors to stop.
-				_ = findOneFor(mt.Coll, timeout, 1*time.Second)
+				_ = findOneFor(mt.Coll, timeout, 5*time.Second)
 
 				// Simulate normal traffic again for 1 second and assert that there are no errors.
 				errs = findOneFor(mt.Coll, timeout, 1*time.Second)

x/mongo/driver/topology/pool_test.go

@@ -371,82 +371,6 @@ func TestPool(t *testing.T) {
 	t.Run("checkOut", func(t *testing.T) {
 		t.Parallel()
 
-		t.Run("checkOut on a full pool should return a new connection as soon as the pool isn't full", func(t *testing.T) {
-			t.Parallel()
-
-			cleanup := make(chan struct{})
-			defer close(cleanup)
-			addr := bootstrapConnections(t, 3, func(nc net.Conn) {
-				<-cleanup
-				_ = nc.Close()
-			})
-
-			d := newdialer(&net.Dialer{})
-			p := newPool(
-				poolConfig{
-					Address:     address.Address(addr.String()),
-					MaxPoolSize: 2,
-				},
-				WithDialer(func(Dialer) Dialer { return d }),
-			)
-			err := p.connect()
-			noerr(t, err)
-
-			// Check out two connections (MaxPoolSize) so that subsequent checkOut() calls should
-			// block until a connection is checked back in or removed from the pool.
-			c, err := p.checkOut(context.Background())
-			noerr(t, err)
-			_, err = p.checkOut(context.Background())
-			noerr(t, err)
-			assert.Equalf(t, 2, d.lenopened(), "should have opened 2 connection")
-			assert.Equalf(t, 2, p.totalConnectionCount(), "pool should have 2 total connection")
-			assert.Equalf(t, 0, p.availableConnectionCount(), "pool should have 0 idle connection")
-
-			// Run a checkOut() with timeout and expect it to time out because the pool is at
-			// MaxPoolSize and no connections are checked in or removed from the pool.
-			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond)
-			defer cancel()
-			_, err = p.checkOut(ctx)
-			assert.Equalf(t, context.DeadlineExceeded, err.(WaitQueueTimeoutError).Wrapped, "TODO")
-
-			// Start a goroutine that calls checkOut() without a timeout after recording the start
-			// time of the checkOut().
-			var start time.Time
-			waiting := make(chan struct{})
-			var wg sync.WaitGroup
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-
-				close(waiting)
-				start = time.Now()
-				_, err := p.checkOut(context.Background())
-				noerr(t, err)
-			}()
-
-			// Close a connection, wait for the checkOut() goroutine to signal it is waiting for a
-			// checkOut(), then check in the closed connection. Expect that the checkOut() goroutine
-			// exits within 5ms of checking in the closed connection.
-			c.close()
-			<-waiting
-			err = p.checkIn(c)
-			noerr(t, err)
-			wg.Wait()
-			assert.WithinDurationf(
-				t,
-				start,
-				time.Now(),
-				5*time.Millisecond,
-				"expected checkOut to complete within 5ms of checking in a closed connection")
-
-			assert.Equalf(t, 1, d.lenclosed(), "should have closed 1 connection")
-			assert.Equalf(t, 3, d.lenopened(), "should have opened 3 connection")
-			assert.Equalf(t, 2, p.totalConnectionCount(), "pool should have 2 total connection")
-			assert.Equalf(t, 0, p.availableConnectionCount(), "pool should have 0 idle connection")
-
-			err = p.disconnect(context.Background())
-			noerr(t, err)
-		})
 		t.Run("return error when attempting to create new connection", func(t *testing.T) {
 			t.Parallel()
 
@@ -686,6 +610,78 @@ func TestPool(t *testing.T) {
 			noerr(t, err)
 			wg.Wait()
 		})
+		// Test that checkOut() on a full connection pool creates and returns a new connection
+		// immediately as soon as the pool is no longer full.
+		t.Run("should return a new connection as soon as the pool isn't full", func(t *testing.T) {
+			t.Parallel()
+
+			cleanup := make(chan struct{})
+			defer close(cleanup)
+			addr := bootstrapConnections(t, 3, func(nc net.Conn) {
+				<-cleanup
+				_ = nc.Close()
+			})
+
+			d := newdialer(&net.Dialer{})
+			p := newPool(
+				poolConfig{
+					Address:     address.Address(addr.String()),
+					MaxPoolSize: 2,
+				},
+				WithDialer(func(Dialer) Dialer { return d }),
+			)
+			err := p.connect()
+			noerr(t, err)
+
+			// Check out two connections (MaxPoolSize) so that subsequent checkOut() calls should
+			// block until a connection is checked back in or removed from the pool.
+			c, err := p.checkOut(context.Background())
+			noerr(t, err)
+			_, err = p.checkOut(context.Background())
+			noerr(t, err)
+			assert.Equalf(t, 2, d.lenopened(), "should have opened 2 connection")
+			assert.Equalf(t, 2, p.totalConnectionCount(), "pool should have 2 total connection")
+			assert.Equalf(t, 0, p.availableConnectionCount(), "pool should have 0 idle connection")
+
+			// Run a checkOut() with timeout and expect it to time out because the pool is at
+			// MaxPoolSize and no connections are checked in or removed from the pool.
+			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond)
+			defer cancel()
+			_, err = p.checkOut(ctx)
+			assert.Equalf(
+				t,
+				context.DeadlineExceeded,
+				err.(WaitQueueTimeoutError).Wrapped,
+				"expected wrapped error to be a context.DeadlineExceeded")
+
+			// Start a goroutine that closes one of the checked-out conections and checks it in.
+			// Expect that the checked-in connection is closed and allows blocked checkOut() to
+			// complete. Assert that the time between checking in the closed connection and when the
+			// checkOut() completes is within 50ms.
+			var start time.Time
+			go func() {
+				c.close()
+				start = time.Now()
+				err := p.checkIn(c)
+				noerr(t, err)
+			}()
+			_, err = p.checkOut(context.Background())
+			noerr(t, err)
+			assert.WithinDurationf(
+				t,
+				time.Now(),
+				start,
+				50*time.Millisecond,
+				"expected checkOut to complete within 50ms of checking in a closed connection")
+
+			assert.Equalf(t, 1, d.lenclosed(), "should have closed 1 connection")
+			assert.Equalf(t, 3, d.lenopened(), "should have opened 3 connection")
+			assert.Equalf(t, 2, p.totalConnectionCount(), "pool should have 2 total connection")
+			assert.Equalf(t, 0, p.availableConnectionCount(), "pool should have 0 idle connection")
+
+			err = p.disconnect(context.Background())
+			noerr(t, err)
+		})
 	})
 	t.Run("checkIn", func(t *testing.T) {
 		t.Parallel()