added note about parallelism capped by Scala collection thread pool, adjusted comments

Author: BryanCutler

docs/ml-tuning.md

@@ -56,7 +56,7 @@ method in each of these evaluators.
 
 To help construct the parameter grid, users can use the [`ParamGridBuilder`](api/scala/index.html#org.apache.spark.ml.tuning.ParamGridBuilder) utility.
 Sets of parameters from the parameter grid can be evaluated in parallel by setting `numParallelEval` with a value of 2 or more (a value of 1 will evaluate in serial) before running model selection with `CrossValidator` or `TrainValidationSplit`.
-The value of `numParallelEval` should be chosen carefully to maximize concurrency without exceeding cluster resources.  Generally speaking, a value up to 10 should be sufficient for most clusters.
+The value of `numParallelEval` should be chosen carefully to maximize concurrency without exceeding cluster resources, and will be capped at the number of cores in the driver system.  Generally speaking, a value up to 10 should be sufficient for most clusters.
 
 
 # Cross-Validation

mllib/src/main/scala/org/apache/spark/ml/tuning/CrossValidator.scala

@@ -105,6 +105,9 @@ class CrossValidator @Since("1.2.0") (@Since("1.4.0") override val uid: String)
     val eval = $(evaluator)
     val epm = $(estimatorParamMaps)
     val numModels = epm.length
+    // Barrier to limit parallelism during model fit/evaluation
+    // NOTE: will be capped by size of thread pool used in Scala parallel collections, which is
+    // number of cores in the system by default
     val numParBarrier = new Semaphore($(numParallelEval))
 
     val instr = Instrumentation.create(this, dataset)
@@ -118,7 +121,6 @@ class CrossValidator @Since("1.2.0") (@Since("1.4.0") override val uid: String)
     val metrics = splits.zipWithIndex.map { case ((training, validation), splitIndex) =>
       val trainingDataset = sparkSession.createDataFrame(training, schema).cache()
       val validationDataset = sparkSession.createDataFrame(validation, schema).cache()
-      // multi-model training
       logDebug(s"Train split $splitIndex with multiple sets of parameters.")
 
       // Fit models concurrently, limited by a barrier with '$numParallelEval' permits

mllib/src/main/scala/org/apache/spark/ml/tuning/TrainValidationSplit.scala

@@ -99,6 +99,9 @@ class TrainValidationSplit @Since("1.5.0") (@Since("1.5.0") override val uid: St
     val est = $(estimator)
     val eval = $(evaluator)
     val epm = $(estimatorParamMaps)
+    // Barrier to limit parallelism during model fit/evaluation
+    // NOTE: will be capped by size of thread pool used in Scala parallel collections, which is
+    // number of cores in the system by default
     val numParBarrier = new Semaphore($(numParallelEval))
     logDebug(s"Running validation with level of parallelism: ${numParBarrier.availablePermits()}.")
 
@@ -111,15 +114,16 @@ class TrainValidationSplit @Since("1.5.0") (@Since("1.5.0") override val uid: St
     trainingDataset.cache()
     validationDataset.cache()
 
-    logDebug(s"Train split with multiple sets of parameters.")
     // Fit models concurrently, limited by a barrier with '$numParallelEval' permits
+    logDebug(s"Train split with multiple sets of parameters.")
     val models = epm.par.map { paramMap =>
       numParBarrier.acquire()
       val model = est.fit(trainingDataset, paramMap)
       numParBarrier.release()
       model.asInstanceOf[Model[_]]
     }.seq
     trainingDataset.unpersist()
+
     // Evaluate models concurrently, limited by a barrier with '$numParallelEval' permits
     val metrics = models.zip(epm).par.map { case (model, paramMap) =>
       numParBarrier.acquire()