Merge branch 'master' into SPARK-30018

Author: yaooqinn

.github/workflows/stale.yml

@@ -11,12 +11,13 @@ jobs:
     - uses: actions/stale@v1.1.0
       with:
         repo-token: ${{ secrets.GITHUB_TOKEN }}
-        stale-pr-message: |
+        stale-pr-message: >
           We're closing this PR because it hasn't been updated in a while.
           This isn't a judgement on the merit of the PR in any way. It's just
           a way of keeping the PR queue manageable.
 
-          If you'd like to revive this PR, please reopen it!
+          If you'd like to revive this PR, please reopen it and ask a
+          committer to remove the Stale tag!
         days-before-stale: 100
         # Setting this to 0 is the same as setting it to 1.
         # See: https://github.com/actions/stale/issues/28

R/pkg/R/functions.R

@@ -879,8 +879,8 @@ setMethod("factorial",
 #'
 #' The function by default returns the first values it sees. It will return the first non-missing
 #' value it sees when na.rm is set to true. If all values are missing, then NA is returned.
-#' Note: the function is non-deterministic because its results depends on order of rows which
-#' may be non-deterministic after a shuffle.
+#' Note: the function is non-deterministic because its results depends on the order of the rows
+#' which may be non-deterministic after a shuffle.
 #'
 #' @param na.rm a logical value indicating whether NA values should be stripped
 #'        before the computation proceeds.
@@ -1024,8 +1024,8 @@ setMethod("kurtosis",
 #'
 #' The function by default returns the last values it sees. It will return the last non-missing
 #' value it sees when na.rm is set to true. If all values are missing, then NA is returned.
-#' Note: the function is non-deterministic because its results depends on order of rows which
-#' may be non-deterministic after a shuffle.
+#' Note: the function is non-deterministic because its results depends on the order of the rows
+#' which may be non-deterministic after a shuffle.
 #'
 #' @param x column to compute on.
 #' @param na.rm a logical value indicating whether NA values should be stripped
@@ -3706,7 +3706,7 @@ setMethod("create_map",
 #' @details
 #' \code{collect_list}: Creates a list of objects with duplicates.
 #' Note: the function is non-deterministic because the order of collected results depends
-#' on order of rows which may be non-deterministic after a shuffle.
+#' on the order of the rows which may be non-deterministic after a shuffle.
 #'
 #' @rdname column_aggregate_functions
 #' @aliases collect_list collect_list,Column-method
@@ -3727,7 +3727,7 @@ setMethod("collect_list",
 #' @details
 #' \code{collect_set}: Creates a list of objects with duplicate elements eliminated.
 #' Note: the function is non-deterministic because the order of collected results depends
-#' on order of rows which may be non-deterministic after a shuffle.
+#' on the order of the rows which may be non-deterministic after a shuffle.
 #'
 #' @rdname column_aggregate_functions
 #' @aliases collect_set collect_set,Column-method

common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java

@@ -57,7 +57,7 @@ private static class StreamState {
     int curChunk = 0;
 
     // Used to keep track of the number of chunks being transferred and not finished yet.
-    volatile long chunksBeingTransferred = 0L;
+    final AtomicLong chunksBeingTransferred = new AtomicLong(0L);
 
     StreamState(String appId, Iterator<ManagedBuffer> buffers, Channel channel) {
       this.appId = appId;
@@ -153,7 +153,7 @@ public void checkAuthorization(TransportClient client, long streamId) {
   public void chunkBeingSent(long streamId) {
     StreamState streamState = streams.get(streamId);
     if (streamState != null) {
-      streamState.chunksBeingTransferred++;
+      streamState.chunksBeingTransferred.incrementAndGet();
     }
 
   }
@@ -167,7 +167,7 @@ public void streamBeingSent(String streamId) {
   public void chunkSent(long streamId) {
     StreamState streamState = streams.get(streamId);
     if (streamState != null) {
-      streamState.chunksBeingTransferred--;
+      streamState.chunksBeingTransferred.decrementAndGet();
     }
   }
 
@@ -180,7 +180,7 @@ public void streamSent(String streamId) {
   public long chunksBeingTransferred() {
     long sum = 0L;
     for (StreamState streamState: streams.values()) {
-      sum += streamState.chunksBeingTransferred;
+      sum += streamState.chunksBeingTransferred.get();
     }
     return sum;
   }

core/src/main/java/org/apache/spark/io/NioBufferedFileInputStream.java

@@ -59,10 +59,10 @@ private boolean refill() throws IOException {
       while (nRead == 0) {
         nRead = fileChannel.read(byteBuffer);
       }
+      byteBuffer.flip();
       if (nRead < 0) {
         return false;
       }
-      byteBuffer.flip();
     }
     return true;
   }

core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala

@@ -611,7 +611,7 @@ private[spark] class PythonRunner(funcs: Seq[ChainedPythonFunctions])
               val obj = new Array[Byte](length)
               stream.readFully(obj)
               obj
-            case 0 => Array.empty[Byte]
+            case 0 => Array.emptyByteArray
             case SpecialLengths.TIMING_DATA =>
               handleTimingData()
               read()

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -1507,7 +1507,8 @@ package object config {
         "longer time than the threshold. This config helps speculate stage with very few " +
         "tasks. Regular speculation configs may also apply if the executor slots are " +
         "large enough. E.g. tasks might be re-launched if there are enough successful runs " +
-        "even though the threshold hasn't been reached.")
+        "even though the threshold hasn't been reached. The number of slots is computed based " +
+        "on the conf values of spark.executor.cores and spark.task.cpus minimum 1.")
       .timeConf(TimeUnit.MILLISECONDS)
       .createOptional
 

core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala

@@ -119,18 +119,22 @@ private[spark] object ResourceUtils extends Logging {
   def parseAllResourceRequests(
       sparkConf: SparkConf,
       componentName: String): Seq[ResourceRequest] = {
-    listResourceIds(sparkConf, componentName).map { id =>
-      parseResourceRequest(sparkConf, id)
-    }
+    listResourceIds(sparkConf, componentName)
+      .map(id => parseResourceRequest(sparkConf, id))
+      .filter(_.amount > 0)
   }
 
   def parseResourceRequirements(sparkConf: SparkConf, componentName: String)
     : Seq[ResourceRequirement] = {
-    listResourceIds(sparkConf, componentName).map { resourceId =>
+    val resourceIds = listResourceIds(sparkConf, componentName)
+    val rnamesAndAmounts = resourceIds.map { resourceId =>
       val settings = sparkConf.getAllWithPrefix(resourceId.confPrefix).toMap
       val amountDouble = settings.getOrElse(AMOUNT,
         throw new SparkException(s"You must specify an amount for ${resourceId.resourceName}")
       ).toDouble
+      (resourceId.resourceName, amountDouble)
+    }
+    rnamesAndAmounts.filter { case (_, amount) => amount > 0 }.map { case (rName, amountDouble) =>
       val (amount, parts) = if (componentName.equalsIgnoreCase(SPARK_TASK_PREFIX)) {
         val parts = if (amountDouble <= 0.5) {
           Math.floor(1.0 / amountDouble).toInt
@@ -147,7 +151,7 @@ private[spark] object ResourceUtils extends Logging {
       } else {
         (amountDouble.toInt, 1)
       }
-      ResourceRequirement(resourceId.resourceName, amount, parts)
+      ResourceRequirement(rName, amount, parts)
     }
   }
 
@@ -184,10 +188,15 @@ private[spark] object ResourceUtils extends Logging {
     val allocated = resourcesFileOpt.toSeq.flatMap(parseAllocatedFromJsonFile)
       .filter(_.id.componentName == componentName)
     val otherResourceIds = listResourceIds(sparkConf, componentName).diff(allocated.map(_.id))
-    allocated ++ otherResourceIds.map { id =>
+    val otherResources = otherResourceIds.flatMap { id =>
       val request = parseResourceRequest(sparkConf, id)
-      ResourceAllocation(id, discoverResource(request).addresses)
+      if (request.amount > 0) {
+        Some(ResourceAllocation(id, discoverResource(request).addresses))
+      } else {
+        None
+      }
     }
+    allocated ++ otherResources
   }
 
   private def assertResourceAllocationMeetsRequest(

core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala

@@ -22,6 +22,7 @@ import javax.annotation.concurrent.GuardedBy
 
 import scala.collection.JavaConverters._
 import scala.concurrent.Promise
+import scala.util.control.NonFatal
 
 import org.apache.spark.SparkException
 import org.apache.spark.internal.Logging
@@ -44,16 +45,6 @@ private[netty] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) exte
   private val shutdownLatch = new CountDownLatch(1)
   private lazy val sharedLoop = new SharedMessageLoop(nettyEnv.conf, this, numUsableCores)
 
-  private def getMessageLoop(name: String, endpoint: RpcEndpoint): MessageLoop = {
-    endpoint match {
-      case e: IsolatedRpcEndpoint =>
-        new DedicatedMessageLoop(name, e, this)
-      case _ =>
-        sharedLoop.register(name, endpoint)
-        sharedLoop
-    }
-  }
-
   /**
    * True if the dispatcher has been stopped. Once stopped, all messages posted will be bounced
    * immediately.
@@ -68,11 +59,31 @@ private[netty] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) exte
       if (stopped) {
         throw new IllegalStateException("RpcEnv has been stopped")
       }
-      if (endpoints.putIfAbsent(name, getMessageLoop(name, endpoint)) != null) {
+      if (endpoints.containsKey(name)) {
         throw new IllegalArgumentException(s"There is already an RpcEndpoint called $name")
       }
+
+      // This must be done before assigning RpcEndpoint to MessageLoop, as MessageLoop sets Inbox be
+      // active when registering, and endpointRef must be put into endpointRefs before onStart is
+      // called.
+      endpointRefs.put(endpoint, endpointRef)
+
+      var messageLoop: MessageLoop = null
+      try {
+        messageLoop = endpoint match {
+          case e: IsolatedRpcEndpoint =>
+            new DedicatedMessageLoop(name, e, this)
+          case _ =>
+            sharedLoop.register(name, endpoint)
+            sharedLoop
+        }
+        endpoints.put(name, messageLoop)
+      } catch {
+        case NonFatal(e) =>
+          endpointRefs.remove(endpoint)
+          throw e
+      }
     }
-    endpointRefs.put(endpoint, endpointRef)
     endpointRef
   }
 

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

@@ -87,7 +87,13 @@ private[spark] class TaskSetManager(
   // number of slots on a single executor, would the task manager speculative run the tasks if
   // their duration is longer than the given threshold. In this way, we wouldn't speculate too
   // aggressively but still handle basic cases.
-  val speculationTasksLessEqToSlots = numTasks <= (conf.get(EXECUTOR_CORES) / sched.CPUS_PER_TASK)
+  // SPARK-30417: #cores per executor might not be set in spark conf for standalone mode, then
+  // the value of the conf would 1 by default. However, the executor would use all the cores on
+  // the worker. Therefore, CPUS_PER_TASK is okay to be greater than 1 without setting #cores.
+  // To handle this case, we assume the minimum number of slots is 1.
+  // TODO: use the actual number of slots for standalone mode.
+  val speculationTasksLessEqToSlots =
+    numTasks <= Math.max(conf.get(EXECUTOR_CORES) / sched.CPUS_PER_TASK, 1)
 
   // For each task, tracks whether a copy of the task has succeeded. A task will also be
   // marked as "succeeded" if it failed with a fetch failure, in which case it should not

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -95,9 +95,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
   // Executors we have requested the cluster manager to kill that have not died yet; maps
   // the executor ID to whether it was explicitly killed by the driver (and thus shouldn't
-  // be considered an app-related failure).
+  // be considered an app-related failure). Visible for testing only.
   @GuardedBy("CoarseGrainedSchedulerBackend.this")
-  private val executorsPendingToRemove = new HashMap[String, Boolean]
+  private[scheduler] val executorsPendingToRemove = new HashMap[String, Boolean]
 
   // Executors that have been lost, but for which we don't yet know the real exit reason.
   private val executorsPendingLossReason = new HashSet[String]
@@ -487,12 +487,12 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   /**
    * Reset the state of CoarseGrainedSchedulerBackend to the initial state. Currently it will only
    * be called in the yarn-client mode when AM re-registers after a failure.
+   * Visible for testing only.
    * */
-  protected def reset(): Unit = {
+  protected[scheduler] def reset(): Unit = {
     val executors: Set[String] = synchronized {
       requestedTotalExecutors = 0
       numPendingExecutors = 0
-      executorsPendingToRemove.clear()
       executorDataMap.keys.toSet
     }
 

core/src/main/scala/org/apache/spark/ui/ConsoleProgressBar.scala

@@ -94,7 +94,7 @@ private[spark] class ConsoleProgressBar(sc: SparkContext) extends Logging {
     // only refresh if it's changed OR after 1 minute (or the ssh connection will be closed
     // after idle some time)
     if (bar != lastProgressBar || now - lastUpdateTime > 60 * 1000L) {
-      System.err.print(CR + bar)
+      System.err.print(CR + bar + CR)
       lastUpdateTime = now
     }
     lastProgressBar = bar

core/src/main/scala/org/apache/spark/util/SizeEstimator.scala

@@ -326,7 +326,7 @@ object SizeEstimator extends Logging {
     val parent = getClassInfo(cls.getSuperclass)
     var shellSize = parent.shellSize
     var pointerFields = parent.pointerFields
-    val sizeCount = Array.fill(fieldSizes.max + 1)(0)
+    val sizeCount = Array.ofDim[Int](fieldSizes.max + 1)
 
     // iterate through the fields of this class and gather information.
     for (field <- cls.getDeclaredFields) {

core/src/test/java/org/apache/spark/io/GenericFileInputStreamSuite.java

@@ -48,8 +48,12 @@ public void setUp() throws IOException {
   }
 
   @After
-  public void tearDown() {
+  public void tearDown() throws IOException {
     inputFile.delete();
+
+    for (InputStream is : inputStreams) {
+      is.close();
+    }
   }
 
   @Test
@@ -141,4 +145,15 @@ public void testBytesSkippedAfterEOF() throws IOException {
       assertEquals(-1, inputStream.read());
     }
   }
+
+  @Test
+  public void testReadPastEOF() throws IOException {
+    InputStream is = inputStreams[0];
+    byte[] buf = new byte[1024];
+    int read;
+    while ((read = is.read(buf, 0, buf.length)) != -1);
+
+    int readAfterEOF = is.read(buf, 0, buf.length);
+    assertEquals(-1, readAfterEOF);
+  }
 }

core/src/test/scala/org/apache/spark/SparkConfSuite.scala

@@ -457,6 +457,19 @@ class SparkConfSuite extends SparkFunSuite with LocalSparkContext with ResetSyst
     assert(taskResourceRequirement.get(FPGA).isEmpty)
   }
 
+  test("test task resource requirement with 0 amount") {
+    val conf = new SparkConf()
+    conf.set(TASK_GPU_ID.amountConf, "2")
+    conf.set(TASK_FPGA_ID.amountConf, "0")
+    var taskResourceRequirement =
+      parseResourceRequirements(conf, SPARK_TASK_PREFIX)
+        .map(req => (req.resourceName, req.amount)).toMap
+
+    assert(taskResourceRequirement.size == 1)
+    assert(taskResourceRequirement(GPU) == 2)
+  }
+
+
   test("Ensure that we can configure fractional resources for a task") {
     val ratioSlots = Seq(
       (0.10, 10), (0.11, 9), (0.125, 8), (0.14, 7), (0.16, 6),

core/src/test/scala/org/apache/spark/resource/ResourceUtilsSuite.scala

@@ -60,6 +60,20 @@ class ResourceUtilsSuite extends SparkFunSuite
     }
   }
 
+  test("Resource discoverer amount 0") {
+    val conf = new SparkConf
+    assume(!(Utils.isWindows))
+    withTempDir { dir =>
+      val scriptPath = createTempScriptWithExpectedOutput(dir, "gpuDiscoverScript",
+        """{"name": "gpu"}""")
+      conf.set(EXECUTOR_GPU_ID.amountConf, "0")
+      conf.set(EXECUTOR_GPU_ID.discoveryScriptConf, scriptPath)
+
+      val res = getOrDiscoverAllResources(conf, SPARK_EXECUTOR_PREFIX, None)
+      assert(res.isEmpty)
+    }
+  }
+
   test("Resource discoverer multiple resource types") {
     val conf = new SparkConf
     assume(!(Utils.isWindows))
@@ -74,6 +88,13 @@ class ResourceUtilsSuite extends SparkFunSuite
       conf.set(EXECUTOR_FPGA_ID.amountConf, "2")
       conf.set(EXECUTOR_FPGA_ID.discoveryScriptConf, fpgaDiscovery)
 
+      // test one with amount 0 to make sure ignored
+      val fooDiscovery = createTempScriptWithExpectedOutput(dir, "fooDiscoverScript",
+        """{"name": "foo", "addresses": ["f1", "f2", "f3"]}""")
+      val fooId = ResourceID(SPARK_EXECUTOR_PREFIX, "foo")
+      conf.set(fooId.amountConf, "0")
+      conf.set(fooId.discoveryScriptConf, fooDiscovery)
+
       val resources = getOrDiscoverAllResources(conf, SPARK_EXECUTOR_PREFIX, None)
       assert(resources.size === 2)
       val gpuValue = resources.get(GPU)