Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/22062#discussion_r209372979
--- Diff:
core/src/test/scala/org/apache/spark/shuffle/sort/ShuffleExternalSorterSuite.scala
---
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.sort
+
+import java.lang.{Long => JLong}
+
+import org.mockito.Mockito.when
+import org.scalatest.mockito.MockitoSugar
+
+import org.apache.spark._
+import org.apache.spark.executor.{ShuffleWriteMetrics, TaskMetrics}
+import org.apache.spark.memory._
+import org.apache.spark.unsafe.Platform
+
+class ShuffleExternalSorterSuite extends SparkFunSuite with
LocalSparkContext with MockitoSugar {
+
+ test("nested spill should be no-op") {
+ val conf = new SparkConf()
+ .setMaster("local[1]")
+ .setAppName("ShuffleExternalSorterSuite")
+ .set("spark.testing", "true")
+ .set("spark.testing.memory", "1600")
+ .set("spark.memory.fraction", "1")
+ sc = new SparkContext(conf)
+
+ val memoryManager = UnifiedMemoryManager(conf, 1)
+
+ var shouldAllocate = false
+
+ // Mock `TaskMemoryManager` to allocate free memory when
`shouldAllocate` is true.
+ // This will trigger a nested spill and expose issues if we don't
handle this case properly.
+ val taskMemoryManager = new TaskMemoryManager(memoryManager, 0) {
+ override def acquireExecutionMemory(required: Long, consumer:
MemoryConsumer): Long = {
+ // ExecutionMemoryPool.acquireMemory will wait until there are 400
bytes for a task to use.
+ // So we leave 400 bytes for the task.
+ if (shouldAllocate &&
+ memoryManager.maxHeapMemory - memoryManager.executionMemoryUsed
> 400) {
+ val acquireExecutionMemoryMethod =
+ memoryManager.getClass.getMethods.filter(_.getName ==
"acquireExecutionMemory").head
+ acquireExecutionMemoryMethod.invoke(
+ memoryManager,
+ JLong.valueOf(
+ memoryManager.maxHeapMemory -
memoryManager.executionMemoryUsed - 400),
+ JLong.valueOf(1L), // taskAttemptId
+ MemoryMode.ON_HEAP
+ ).asInstanceOf[java.lang.Long]
+ }
+ super.acquireExecutionMemory(required, consumer)
+ }
+ }
+ val taskContext = mock[TaskContext]
--- End diff --
lol
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
