Merge branch 'v1' into dont-waste-id

Author: adhocore

bson/bsoncodec/registry.go

@@ -388,6 +388,9 @@ func (r *Registry) RegisterTypeMapEntry(bt bsontype.Type, rt reflect.Type) {
 // If no encoder is found, an error of type ErrNoEncoder is returned. LookupEncoder is safe for
 // concurrent use by multiple goroutines after all codecs and encoders are registered.
 func (r *Registry) LookupEncoder(valueType reflect.Type) (ValueEncoder, error) {
+	if valueType == nil {
+		return nil, ErrNoEncoder{Type: valueType}
+	}
 	enc, found := r.lookupTypeEncoder(valueType)
 	if found {
 		if enc == nil {
@@ -400,15 +403,10 @@ func (r *Registry) LookupEncoder(valueType reflect.Type) (ValueEncoder, error) {
 	if found {
 		return r.typeEncoders.LoadOrStore(valueType, enc), nil
 	}
-	if valueType == nil {
-		r.storeTypeEncoder(valueType, nil)
-		return nil, ErrNoEncoder{Type: valueType}
-	}
 
 	if v, ok := r.kindEncoders.Load(valueType.Kind()); ok {
 		return r.storeTypeEncoder(valueType, v), nil
 	}
-	r.storeTypeEncoder(valueType, nil)
 	return nil, ErrNoEncoder{Type: valueType}
 }
 
@@ -474,7 +472,6 @@ func (r *Registry) LookupDecoder(valueType reflect.Type) (ValueDecoder, error) {
 	if v, ok := r.kindDecoders.Load(valueType.Kind()); ok {
 		return r.storeTypeDecoder(valueType, v), nil
 	}
-	r.storeTypeDecoder(valueType, nil)
 	return nil, ErrNoDecoder{Type: valueType}
 }
 

bson/bsoncodec/registry_test.go

@@ -792,6 +792,36 @@ func TestRegistry(t *testing.T) {
 					})
 				})
 			}
+			t.Run("nil type", func(t *testing.T) {
+				t.Parallel()
+
+				t.Run("Encoder", func(t *testing.T) {
+					t.Parallel()
+
+					wanterr := ErrNoEncoder{Type: reflect.TypeOf(nil)}
+
+					gotcodec, goterr := reg.LookupEncoder(nil)
+					if !cmp.Equal(goterr, wanterr, cmp.Comparer(compareErrors)) {
+						t.Errorf("errors did not match: got %#v, want %#v", goterr, wanterr)
+					}
+					if !cmp.Equal(gotcodec, nil, allowunexported, cmp.Comparer(comparepc)) {
+						t.Errorf("codecs did not match: got %#v, want nil", gotcodec)
+					}
+				})
+				t.Run("Decoder", func(t *testing.T) {
+					t.Parallel()
+
+					wanterr := ErrNilType
+
+					gotcodec, goterr := reg.LookupDecoder(nil)
+					if !cmp.Equal(goterr, wanterr, cmp.Comparer(compareErrors)) {
+						t.Errorf("errors did not match: got %#v, want %#v", goterr, wanterr)
+					}
+					if !cmp.Equal(gotcodec, nil, allowunexported, cmp.Comparer(comparepc)) {
+						t.Errorf("codecs did not match: got %v: want nil", gotcodec)
+					}
+				})
+			})
 			// lookup a type whose pointer implements an interface and expect that the registered hook is
 			// returned
 			t.Run("interface implementation with hook (pointer)", func(t *testing.T) {

bson/marshal_test.go

@@ -11,6 +11,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"sync"
 	"testing"
 	"time"
 
@@ -380,3 +381,19 @@ func TestMarshalExtJSONIndent(t *testing.T) {
 		})
 	}
 }
+
+func TestMarshalConcurrently(t *testing.T) {
+	t.Parallel()
+
+	const size = 10_000
+
+	wg := sync.WaitGroup{}
+	wg.Add(size)
+	for i := 0; i < size; i++ {
+		go func() {
+			defer wg.Done()
+			_, _ = Marshal(struct{ LastError error }{})
+		}()
+	}
+	wg.Wait()
+}

bson/unmarshal_test.go

@@ -9,6 +9,7 @@ package bson
 import (
 	"math/rand"
 	"reflect"
+	"sync"
 	"testing"
 
 	"go.mongodb.org/mongo-driver/bson/bsoncodec"
@@ -773,3 +774,21 @@ func TestUnmarshalByteSlicesUseDistinctArrays(t *testing.T) {
 		})
 	}
 }
+
+func TestUnmarshalConcurrently(t *testing.T) {
+	t.Parallel()
+
+	const size = 10_000
+
+	data := []byte{16, 0, 0, 0, 10, 108, 97, 115, 116, 101, 114, 114, 111, 114, 0, 0}
+	wg := sync.WaitGroup{}
+	wg.Add(size)
+	for i := 0; i < size; i++ {
+		go func() {
+			defer wg.Done()
+			var res struct{ LastError error }
+			_ = Unmarshal(data, &res)
+		}()
+	}
+	wg.Wait()
+}

mongo/integration/unified/client_entity.go

@@ -107,7 +107,10 @@ func newClientEntity(ctx context.Context, em *EntityMap, entityOptions *entityOp
 	}
 
 	if olm := entityOptions.ObserveLogMessages; olm != nil {
-		clientLogger := newLogger(olm, expectedLogMessageCount(ctx))
+		expectedLogMessagesCount := expectedLogMessagesCount(ctx, entityOptions.ID)
+		ignoreLogMessages := ignoreLogMessages(ctx, entityOptions.ID)
+
+		clientLogger := newLogger(olm, expectedLogMessagesCount, ignoreLogMessages)
 
 		wrap := func(str string) options.LogLevel {
 			return options.LogLevel(logger.ParseLevel(str))

mongo/integration/unified/context.go

@@ -28,23 +28,23 @@ const (
 	failPointsKey ctxKey = "test-failpoints"
 	// targetedFailPointsKey is used to store a map from a fail point name to the host on which the fail point is set.
 	targetedFailPointsKey ctxKey = "test-targeted-failpoints"
-	// expectedLogMessageCountKey is used to store the number of log messages expected to be received by the test runner.
-	expectedLogMessageCountKey ctxKey = "test-expected-log-message-count"
+	clientLogMessagesKey  ctxKey = "test-expected-log-message-count"
+	ignoreLogMessagesKey  ctxKey = "test-ignore-log-message-count"
 )
 
 // newTestContext creates a new Context derived from ctx with values initialized to store the state required for test
 // execution.
 func newTestContext(
 	ctx context.Context,
 	entityMap *EntityMap,
-	expectedLogMessageCount int,
+	clientLogMessages []*clientLogMessages,
 	hasOperationalFailPoint bool,
 ) context.Context {
 	ctx = context.WithValue(ctx, operationalFailPointKey, hasOperationalFailPoint)
 	ctx = context.WithValue(ctx, entitiesKey, entityMap)
 	ctx = context.WithValue(ctx, failPointsKey, make(map[string]*mongo.Client))
 	ctx = context.WithValue(ctx, targetedFailPointsKey, make(map[string]string))
-	ctx = context.WithValue(ctx, expectedLogMessageCountKey, expectedLogMessageCount)
+	ctx = context.WithValue(ctx, clientLogMessagesKey, clientLogMessages)
 	return ctx
 }
 
@@ -84,6 +84,28 @@ func entities(ctx context.Context) *EntityMap {
 	return ctx.Value(entitiesKey).(*EntityMap)
 }
 
-func expectedLogMessageCount(ctx context.Context) int {
-	return ctx.Value(expectedLogMessageCountKey).(int)
+func expectedLogMessagesCount(ctx context.Context, clientID string) int {
+	messages := ctx.Value(clientLogMessagesKey).([]*clientLogMessages)
+
+	count := 0
+	for _, message := range messages {
+		if message.Client == clientID {
+			count += len(message.LogMessages)
+		}
+	}
+
+	return count
+}
+
+func ignoreLogMessages(ctx context.Context, clientID string) []*logMessage {
+	messages := ctx.Value(clientLogMessagesKey).([]*clientLogMessages)
+
+	ignoreMessages := []*logMessage{}
+	for _, message := range messages {
+		if message.Client == clientID {
+			ignoreMessages = append(ignoreMessages, message.IgnoreMessages...)
+		}
+	}
+
+	return ignoreMessages
 }

mongo/integration/unified/logger.go

@@ -7,6 +7,7 @@
 package unified
 
 import (
+	"context"
 	"sync"
 
 	"go.mongodb.org/mongo-driver/internal/logger"
@@ -33,19 +34,21 @@ type Logger struct {
 	// orderMu guards the order value, which increments each time the "Info"
 	// method is called. This is necessary since "Info" could be called from
 	// multiple go routines, e.g. SDAM logs.
-	orderMu  sync.RWMutex
-	logQueue chan orderedLogMessage
+	orderMu        sync.RWMutex
+	logQueue       chan orderedLogMessage
+	ignoreMessages []*logMessage
 }
 
-func newLogger(olm *observeLogMessages, bufSize int) *Logger {
+func newLogger(olm *observeLogMessages, bufSize int, ignoreMessages []*logMessage) *Logger {
 	if olm == nil {
 		return nil
 	}
 
 	return &Logger{
-		lastOrder: 1,
-		logQueue:  make(chan orderedLogMessage, bufSize),
-		bufSize:   bufSize,
+		lastOrder:      1,
+		logQueue:       make(chan orderedLogMessage, bufSize),
+		bufSize:        bufSize,
+		ignoreMessages: ignoreMessages,
 	}
 }
 
@@ -65,8 +68,6 @@ func (log *Logger) Info(level int, msg string, args ...interface{}) {
 		return
 	}
 
-	defer func() { log.lastOrder++ }()
-
 	// Add the Diff back to the level, as there is no need to create a
 	// logging offset.
 	level = level + logger.DiffToInfo
@@ -76,12 +77,19 @@ func (log *Logger) Info(level int, msg string, args ...interface{}) {
 		panic(err)
 	}
 
+	for _, ignoreMessage := range log.ignoreMessages {
+		if err := verifyLogMatch(context.Background(), ignoreMessage, logMessage); err == nil {
+			return
+		}
+	}
+
+	defer func() { log.lastOrder++ }()
+
 	// Send the log message to the "orderedLogMessage" channel for
 	// validation.
 	log.logQueue <- orderedLogMessage{
 		order:      log.lastOrder + 1,
-		logMessage: logMessage,
-	}
+		logMessage: logMessage}
 
 	// If the order has reached the buffer size, then close the channel.
 	if log.lastOrder == log.bufSize {

mongo/integration/unified/logger_verification.go

@@ -79,17 +79,6 @@ type clientLogMessages struct {
 	LogMessages    []*logMessage `bson:"messages"`
 }
 
-// ignore checks to see if the message is in the "IgnoreMessages" slice.
-func (clm clientLogMessages) ignore(ctx context.Context, msg *logMessage) bool {
-	for _, ignoreMessage := range clm.IgnoreMessages {
-		if err := verifyLogMatch(ctx, ignoreMessage, msg); err == nil {
-			return true
-		}
-	}
-
-	return false
-}
-
 // logMessageValidator defines the expectation for log messages across all
 // clients.
 type logMessageValidator struct {
@@ -191,8 +180,7 @@ type logQueues struct {
 }
 
 // partitionLogQueue will partition the expected logs into "unordered" and
-// "ordered" log channels. This function will also remove any logs in the
-// "ignoreMessages" list for a client.
+// "ordered" log channels.
 func partitionLogQueue(ctx context.Context, exp *clientLogMessages) logQueues {
 	orderedLogCh := make(chan *logMessage, len(exp.LogMessages))
 	unorderedLogCh := make(chan *logMessage, len(exp.LogMessages))
@@ -241,12 +229,6 @@ func matchOrderedLogs(ctx context.Context, logs logQueues) <-chan error {
 		defer close(errs)
 
 		for actual := range logs.ordered {
-			// Ignore logs that are in the "IngoreMessages" slice of
-			// the expected results.
-			if logs.expected.ignore(ctx, actual) {
-				continue
-			}
-
 			expected := expLogMessages[0]
 			if expected == nil {
 				continue