GODRIVER-2038 Make all connections in the background using connectTimeoutMS.

Author: matthewdale

data/connection-monitoring-and-pooling/wait-queue-timeout.json

@@ -4,7 +4,7 @@
   "description": "must aggressively timeout threads enqueued longer than waitQueueTimeoutMS",
   "poolOptions": {
     "maxPoolSize": 1,
-    "waitQueueTimeoutMS": 20
+    "waitQueueTimeoutMS": 50
   },
   "operations": [
     {

data/connection-monitoring-and-pooling/wait-queue-timeout.yml

@@ -3,7 +3,7 @@ style: unit
 description: must aggressively timeout threads enqueued longer than waitQueueTimeoutMS
 poolOptions:
   maxPoolSize: 1
-  waitQueueTimeoutMS: 20
+  waitQueueTimeoutMS: 50
 operations:
   # Check out only possible connection
   - name: checkOut

mongo/integration/client_test.go

@@ -464,3 +464,137 @@ type proxyMessage struct {
 	sent          wiremessage.WireMessage
 	received      wiremessage.WireMessage
 }
+
+// func TestClientStress(t *testing.T) {
+// 	mtOpts := mtest.NewOptions().
+// 		MinServerVersion("3.6").
+// 		Topologies(mtest.ReplicaSet, mtest.Sharded, mtest.Single). // TODO: What topologies?
+// 		CreateClient(false)
+// 	mt := mtest.New(t, mtOpts)
+
+// 	maxPoolOpts := mtest.NewOptions().ClientOptions(options.Client().SetMaxPoolSize(100))
+// 	mt.RunOpts("", maxPoolOpts, func(mt *mtest.T) {
+// 		_, err := mt.Coll.InsertOne(context.Background(), bson.D{{"key", "value"}})
+// 		assert.Nil(mt, err, "unexpected error inserting document: %v", err)
+
+// 		findOneFor := func(d time.Duration) []error {
+// 			timeout := time.After(d)
+// 			errs := make([]error, 0)
+
+// 			for {
+// 				select {
+// 				case <-timeout:
+// 					return errs
+// 				default:
+// 				}
+
+// 				ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+// 				defer cancel()
+// 				var res map[string]interface{}
+// 				err := mt.Coll.FindOne(ctx, bson.D{{"key", "value"}}).Decode(&res)
+// 				if err != nil {
+// 					errs = append(errs, err)
+// 				}
+// 			}
+// 		}
+
+// 		g := new(errgroup.Group)
+// 		for i := 0; i < 1000; i++ {
+// 			g.Go(func() error {
+// 				errs := findOneFor(5 * time.Second)
+
+// 				if len(errs) == 0 {
+// 					return nil
+// 				}
+
+// 				return errs[len(errs)-1]
+// 			})
+// 		}
+// 		err = g.Wait()
+// 		assert.NotNil(mt, err, "expected at least one error, got nil")
+
+// 		findOneFor(1 * time.Second)
+
+// 		errs := findOneFor(5 * time.Second)
+// 		assert.True(mt, len(errs) == 0, "expected no errors, but got %d (%v)", len(errs), errs)
+// 	})
+// }
+
+// func TestBlah(t *testing.T) {
+// 	uri := os.Getenv("MONGODB_URI")
+// 	client, err := mongo.Connect(context.Background(), options.Client().ApplyURI(uri).SetMaxPoolSize(100))
+// 	if err != nil {
+// 		t.Log(err)
+// 		t.FailNow()
+// 	}
+// 	coll := client.Database("sample_mflix").Collection("movies")
+// 	oid, err := primitive.ObjectIDFromHex("573a1398f29313caabce9682")
+// 	if err != nil {
+// 		t.Log(err)
+// 		t.FailNow()
+// 	}
+
+// 	findOneFor := func(d time.Duration) []error {
+// 		timeout := time.After(d)
+// 		errs := make([]error, 0)
+
+// 		for {
+// 			select {
+// 			case <-timeout:
+// 				return errs
+// 			default:
+// 			}
+
+// 			ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+// 			defer cancel()
+// 			var res bson.D
+// 			err := coll.FindOne(
+// 				ctx,
+// 				bson.D{{"_id", oid}},
+// 				options.FindOne().SetProjection(bson.D{{"title", 1}})).Decode(&res)
+// 			if err != nil {
+// 				errs = append(errs, err)
+// 			}
+// 		}
+// 	}
+
+// 	g := new(errgroup.Group)
+// 	for i := 0; i < 1000; i++ {
+// 		g.Go(func() error {
+// 			errs := findOneFor(5 * time.Second)
+
+// 			if len(errs) == 0 {
+// 				return nil
+// 			}
+
+// 			return errs[len(errs)-1]
+// 		})
+// 	}
+// 	err = g.Wait()
+// 	assert.NotNil(t, err, "expected at least one error, got nil")
+
+// 	// findOneFor(1 * time.Second)
+
+// 	errs := findOneFor(5 * time.Second)
+// 	assert.True(t, len(errs) == 0, "expected no errors, but got %d (%v)", len(errs), errs)
+// }
+
+// func TestClientOnce(t *testing.T) {
+// 	mtOpts := mtest.NewOptions().
+// 		MinServerVersion("3.6").
+// 		Topologies(mtest.ReplicaSet, mtest.Sharded, mtest.Single). // TODO: What topologies?
+// 		CreateClient(false)
+// 	mt := mtest.New(t, mtOpts)
+
+// 	maxPoolOpts := mtest.NewOptions().ClientOptions(options.Client().SetMaxPoolSize(100))
+// 	mt.RunOpts("", maxPoolOpts, func(mt *mtest.T) {
+// 		_, err := mt.Coll.InsertOne(context.Background(), bson.D{{"key", "value"}})
+// 		assert.Nil(mt, err, "unexpected error inserting document: %v", err)
+
+// 		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+// 		defer cancel()
+// 		var res map[string]interface{}
+// 		err = mt.Coll.FindOne(ctx, bson.D{{"key", "value"}}).Decode(&res)
+// 		assert.Nil(mt, err, "unexpected error calling FindOne")
+// 	})
+// }

mongo/integration/mtest/mongotest.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"go.mongodb.org/mongo-driver/bson"
@@ -103,7 +104,7 @@ type T struct {
 	dataLake          *bool
 	ssl               *bool
 	collCreateOpts    bson.D
-	connsCheckedOut   int // net number of connections checked out during test execution
+	connsCheckedOut   int64 // net number of connections checked out during test execution
 	requireAPIVersion *bool
 
 	// options copied to sub-tests
@@ -231,7 +232,7 @@ func (t *T) RunOpts(name string, opts *Options, callback func(*T)) {
 			// store number of sessions and connections checked out here but assert that they're equal to 0 after
 			// cleaning up test resources to make sure resources are always cleared
 			sessions := sub.Client.NumberSessionsInProgress()
-			conns := sub.connsCheckedOut
+			conns := sub.NumberConnectionsCheckedOut()
 
 			if sub.clientType != Mock {
 				sub.ClearFailPoints()
@@ -369,7 +370,7 @@ func (t *T) GetProxiedMessages() []*ProxyMessage {
 
 // NumberConnectionsCheckedOut returns the number of connections checked out from the test Client.
 func (t *T) NumberConnectionsCheckedOut() int {
-	return t.connsCheckedOut
+	return int(atomic.LoadInt64(&t.connsCheckedOut))
 }
 
 // ClearEvents clears the existing command monitoring events.
@@ -594,9 +595,9 @@ func (t *T) createTestClient() {
 
 				switch evt.Type {
 				case event.GetSucceeded:
-					t.connsCheckedOut++
+					atomic.AddInt64(&t.connsCheckedOut, 1)
 				case event.ConnectionReturned:
-					t.connsCheckedOut--
+					atomic.AddInt64(&t.connsCheckedOut, -1)
 				}
 			},
 		})

x/mongo/driver/topology/cmap_prose_test.go

@@ -52,13 +52,13 @@ func TestCMAPProse(t *testing.T) {
 			assert.Equal(t, numClosed, len(closed), "expected %d closed events, got %d", numClosed, len(closed))
 
 			netCount := numCreated - numClosed
-			assert.Equal(t, netCount, len(p.opened), "expected %d connections in opened map, got %d", netCount,
-				len(p.opened))
+			assert.Equal(t, netCount, p.totalConnectionCount(), "expected %d connections in opened map, got %d", netCount,
+				p.totalConnectionCount())
 		}
 
-		t.Run("get", func(t *testing.T) {
+		t.Run("checkOut", func(t *testing.T) {
 			t.Run("errored connection exists in pool", func(t *testing.T) {
-				// If a connection is created as part of minPoolSize maintenance and errors while connecting, get()
+				// If a connection is created as part of minPoolSize maintenance and errors while connecting, checkOut()
 				// should report that error and publish an event.
 				clearEvents()
 
@@ -76,20 +76,19 @@ func TestCMAPProse(t *testing.T) {
 				}
 				pool := createTestPool(t, cfg, connOpts...)
 
-				_, err := pool.get(context.Background())
-				assert.NotNil(t, err, "expected get() error, got nil")
+				_, err := pool.checkOut(context.Background())
+				assert.NotNil(t, err, "expected checkOut() error, got nil")
 
-				// If the connection doesn't finish connecting before resourcePool gives it back, the error will be
-				// detected by pool.get and result in a created/closed count of 1. If it does finish connecting, the
-				// error will be detected by resourcePool, which will return nil. Then, pool will try to create a new
-				// connection, which will also error. This process will result in a created/closed count of 2.
 				assert.True(t, len(created) == 1 || len(created) == 2, "expected 1 or 2 opened events, got %d", len(created))
 				assert.True(t, len(closed) == 1 || len(closed) == 2, "expected 1 or 2 closed events, got %d", len(closed))
+
+				_ = pool.disconnect(context.Background())
 				netCount := len(created) - len(closed)
 				assert.Equal(t, 0, netCount, "expected net connection count to be 0, got %d", netCount)
 			})
 			t.Run("pool is empty", func(t *testing.T) {
-				// If a new connection is created during get(), get() should report that error and publish an event.
+				// If a checkOut() has to create a new connection and that connection encounters an
+				// error while connecting, checkOut() should return that error and publish an event.
 				clearEvents()
 
 				var dialer DialerFunc = func(context.Context, string, string) (net.Conn, error) {
@@ -103,13 +102,16 @@ func TestCMAPProse(t *testing.T) {
 					}),
 				}
 				pool := createTestPool(t, getConfig(), connOpts...)
+				defer func() {
+					_ = pool.disconnect(context.Background())
+				}()
 
-				_, err := pool.get(context.Background())
-				assert.NotNil(t, err, "expected get() error, got nil")
+				_, err := pool.checkOut(context.Background())
+				assert.NotNil(t, err, "expected checkOut() error, got nil")
 				assertConnectionCounts(t, pool, 1, 1)
 			})
 		})
-		t.Run("put", func(t *testing.T) {
+		t.Run("checkIn", func(t *testing.T) {
 			t.Run("errored connection", func(t *testing.T) {
 				// If the connection being returned to the pool encountered a network error, it should be removed from
 				// the pool and an event should be published.
@@ -124,16 +126,19 @@ func TestCMAPProse(t *testing.T) {
 					WithDialer(func(Dialer) Dialer { return dialer }),
 				}
 				pool := createTestPool(t, getConfig(), connOpts...)
+				defer func() {
+					_ = pool.disconnect(context.Background())
+				}()
 
-				conn, err := pool.get(context.Background())
-				assert.Nil(t, err, "get error: %v", err)
+				conn, err := pool.checkOut(context.Background())
+				assert.Nil(t, err, "checkOut() error: %v", err)
 
 				// Force a network error by writing to the connection.
 				err = conn.writeWireMessage(context.Background(), nil)
 				assert.NotNil(t, err, "expected writeWireMessage error, got nil")
 
-				err = pool.put(conn)
-				assert.Nil(t, err, "put error: %v", err)
+				err = pool.checkIn(conn)
+				assert.Nil(t, err, "checkIn() error: %v", err)
 
 				assertConnectionCounts(t, pool, 1, 1)
 				evt := <-closed
@@ -157,16 +162,19 @@ func TestCMAPProse(t *testing.T) {
 					WithIdleTimeout(func(time.Duration) time.Duration { return 1 * time.Second }),
 				}
 				pool := createTestPool(t, getConfig(), connOpts...)
+				defer func() {
+					_ = pool.disconnect(context.Background())
+				}()
 
-				conn, err := pool.get(context.Background())
-				assert.Nil(t, err, "get error: %v", err)
+				conn, err := pool.checkOut(context.Background())
+				assert.Nil(t, err, "checkOut() error: %v", err)
 
 				// Set the idleDeadline to a time in the past to simulate expiration.
 				pastTime := time.Now().Add(-10 * time.Second)
 				conn.idleDeadline.Store(pastTime)
 
-				err = pool.put(conn)
-				assert.Nil(t, err, "put error: %v", err)
+				err = pool.checkIn(conn)
+				assert.Nil(t, err, "checkIn() error: %v", err)
 
 				assertConnectionCounts(t, pool, 1, 1)
 				evt := <-closed
@@ -189,10 +197,10 @@ func TestCMAPProse(t *testing.T) {
 				conns := checkoutConnections(t, pool, numConns)
 				assertConnectionCounts(t, pool, numConns, 0)
 
-				// Return all connections to the pool and assert that none were closed by put().
+				// Return all connections to the pool and assert that none were closed by checkIn().
 				for i, c := range conns {
-					err := pool.put(c)
-					assert.Nil(t, err, "put error at index %d: %v", i, err)
+					err := pool.checkIn(c)
+					assert.Nil(t, err, "checkIn() error at index %d: %v", i, err)
 				}
 				assertConnectionCounts(t, pool, numConns, 0)
 
@@ -223,8 +231,8 @@ func TestCMAPProse(t *testing.T) {
 
 				// Only return 2 of the connection.
 				for i := 0; i < 2; i++ {
-					err := pool.put(conns[i])
-					assert.Nil(t, err, "put error at index %d: %v", i, err)
+					err := pool.checkIn(conns[i])
+					assert.Nil(t, err, "checkIn() error at index %d: %v", i, err)
 				}
 				conns = conns[2:]
 				assertConnectionCounts(t, pool, numConns, 0)
@@ -237,8 +245,8 @@ func TestCMAPProse(t *testing.T) {
 				// Return the remaining connections and assert that the closed event count does not increase because
 				// these connections have already been closed.
 				for i, c := range conns {
-					err := pool.put(c)
-					assert.Nil(t, err, "put error at index %d: %v", i, err)
+					err := pool.checkIn(c)
+					assert.Nil(t, err, "checkIn() error at index %d: %v", i, err)
 				}
 				assertConnectionCounts(t, pool, numConns, numConns)
 
@@ -269,8 +277,8 @@ func checkoutConnections(t *testing.T, p *pool, numConns int) []*connection {
 	conns := make([]*connection, 0, numConns)
 
 	for i := 0; i < numConns; i++ {
-		conn, err := p.get(context.Background())
-		assert.Nil(t, err, "get error at index %d: %v", i, err)
+		conn, err := p.checkOut(context.Background())
+		assert.Nil(t, err, "checkOut() error at index %d: %v", i, err)
 		conns = append(conns, conn)
 	}