Read documents in parallel.

This makes compareDocsFromChannels() use an errgroup to read its
document-reader channels in parallel.

Author: FGasper

internal/verifier/compare.go

@@ -64,8 +64,8 @@ func (verifier *Verifier) compareDocsFromChannels(
 	error,
 ) {
 	results := []VerificationResult{}
-	var docCount types.DocumentCount
-	var byteCount types.ByteCount
+	var srcDocCount types.DocumentCount
+	var srcByteCount types.ByteCount
 
 	mapKeyFieldNames := make([]string, 1+len(task.QueryFilter.ShardKeys))
 	mapKeyFieldNames[0] = "_id"
@@ -89,6 +89,9 @@ func (verifier *Verifier) compareDocsFromChannels(
 		if isSrc {
 			ourMap = srcCache
 			theirMap = dstCache
+
+			srcDocCount++
+			srcByteCount += types.ByteCount(len(doc))
 		} else {
 			ourMap = dstCache
 			theirMap = srcCache
@@ -146,63 +149,86 @@ func (verifier *Verifier) compareDocsFromChannels(
 	// lags the other significantly, we won’t keep caching the faster side’s
 	// documents and thus consume more & more memory.
 	for !srcClosed || !dstClosed {
+		simpleTimerReset(readTimer, readTimeout)
+
+		var srcDoc, dstDoc bson.Raw
+
+		eg, egCtx := errgroup.WithContext(ctx)
+
 		if !srcClosed {
-			simpleTimerReset(readTimer, readTimeout)
-
-			select {
-			case <-ctx.Done():
-				return nil, 0, 0, ctx.Err()
-			case <-readTimer.C:
-				return nil, 0, 0, errors.Errorf(
-					"failed to read from source after %s",
-					readTimeout,
-				)
-			case doc, alive := <-srcChannel:
-				if !alive {
-					srcClosed = true
-					break
+			eg.Go(func() error {
+				var alive bool
+				select {
+				case <-egCtx.Done():
+					return egCtx.Err()
+				case <-readTimer.C:
+					return errors.Errorf(
+						"failed to read from source after %s",
+						readTimeout,
+					)
+				case srcDoc, alive = <-srcChannel:
+					if !alive {
+						srcClosed = true
+						break
+					}
 				}
-				docCount++
-				byteCount += types.ByteCount(len(doc))
 
-				err := handleNewDoc(doc, true)
+				return nil
+			})
+		}
 
-				if err != nil {
-					return nil, 0, 0, errors.Wrapf(
-						err,
-						"comparer thread failed to handle source doc with ID %v",
-						doc.Lookup("_id"),
+		if !dstClosed {
+			eg.Go(func() error {
+				var alive bool
+				select {
+				case <-egCtx.Done():
+					return egCtx.Err()
+				case <-readTimer.C:
+					return errors.Errorf(
+						"failed to read from destination after %s",
+						readTimeout,
 					)
+				case dstDoc, alive = <-dstChannel:
+					if !alive {
+						dstClosed = true
+						break
+					}
 				}
-			}
+
+				return nil
+			})
 		}
 
-		if !dstClosed {
-			simpleTimerReset(readTimer, readTimeout)
-
-			select {
-			case <-ctx.Done():
-				return nil, 0, 0, ctx.Err()
-			case <-readTimer.C:
-				return nil, 0, 0, errors.Errorf(
-					"failed to read from destination after %s",
-					readTimeout,
+		if err := eg.Wait(); err != nil {
+			return nil, 0, 0, errors.Wrap(
+				err,
+				"failed to read documents",
+			)
+		}
+
+		if srcDoc != nil {
+			err := handleNewDoc(srcDoc, true)
+
+			if err != nil {
+				return nil, 0, 0, errors.Wrapf(
+					err,
+					"comparer thread failed to handle task %s's source doc with ID %v",
+					task.PrimaryKey,
+					srcDoc.Lookup("_id"),
 				)
-			case doc, alive := <-dstChannel:
-				if !alive {
-					dstClosed = true
-					break
-				}
+			}
+		}
 
-				err := handleNewDoc(doc, false)
+		if dstDoc != nil {
+			err := handleNewDoc(dstDoc, false)
 
-				if err != nil {
-					return nil, 0, 0, errors.Wrapf(
-						err,
-						"comparer thread failed to handle destination doc with ID %v",
-						doc.Lookup("_id"),
-					)
-				}
+			if err != nil {
+				return nil, 0, 0, errors.Wrapf(
+					err,
+					"comparer thread failed to handle task %s's destination doc with ID %v",
+					task.PrimaryKey,
+					dstDoc.Lookup("_id"),
+				)
 			}
 		}
 	}
@@ -243,7 +269,7 @@ func (verifier *Verifier) compareDocsFromChannels(
 		)
 	}
 
-	return results, docCount, byteCount, nil
+	return results, srcDocCount, srcByteCount, nil
 }
 
 func simpleTimerReset(t *time.Timer, dur time.Duration) {

internal/verifier/migration_verifier_test.go

@@ -574,7 +574,7 @@ func TestVerifierCompareDocs(t *testing.T) {
 				{{"_id", id}, {"sharded", 123}},
 			},
 			compareFn: func(t *testing.T, mismatchedIds []VerificationResult) {
-				assert.Empty(t, mismatchedIds)
+				assert.Empty(t, mismatchedIds, "should be no problems")
 			},
 		},
 	}
@@ -1491,7 +1491,7 @@ func (suite *IntegrationTestSuite) TestVerifierWithFilter() {
 	status := waitForTasks()
 	suite.Require().Greater(status.CompletedTasks, 1)
 	suite.Require().Greater(status.TotalTasks, 1)
-	suite.Require().Equal(status.FailedTasks, 0)
+	suite.Require().Zero(status.FailedTasks, "there should be no failed tasks")
 
 	// Insert another document that is not in the filter.
 	// This should trigger a recheck despite being outside the filter.