Merge pull request apache#76 from fishcus/kyspark-ke-15774

[kap] apache#15774, clean broadcast after query

Author: fishcus

assembly/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.2.1-kylin-r7</version>
+    <version>2.2.1-kylin-r8</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

common/network-common/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.2.1-kylin-r7</version>
+    <version>2.2.1-kylin-r8</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

common/network-shuffle/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.2.1-kylin-r7</version>
+    <version>2.2.1-kylin-r8</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

common/network-yarn/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.2.1-kylin-r7</version>
+    <version>2.2.1-kylin-r8</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

common/sketch/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.2.1-kylin-r7</version>
+    <version>2.2.1-kylin-r8</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

common/tags/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.2.1-kylin-r7</version>
+    <version>2.2.1-kylin-r8</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

common/unsafe/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.2.1-kylin-r7</version>
+    <version>2.2.1-kylin-r8</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

core/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.11</artifactId>
-    <version>2.2.1-kylin-r7</version>
+    <version>2.2.1-kylin-r8</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

core/src/main/scala/org/apache/spark/ContextCleaner.scala

@@ -19,7 +19,7 @@ package org.apache.spark
 
 import java.lang.ref.{ReferenceQueue, WeakReference}
 import java.util.Collections
-import java.util.concurrent.{ConcurrentHashMap, ConcurrentLinkedQueue, ScheduledExecutorService, TimeUnit}
+import java.util.concurrent._
 
 import scala.collection.JavaConverters._
 
@@ -112,6 +112,12 @@ private[spark] class ContextCleaner(sc: SparkContext) extends Logging {
   private val blockOnShuffleCleanupTasks = sc.conf.getBoolean(
     "spark.cleaner.referenceTracking.blocking.shuffle", false)
 
+  private val cleanupTaskThreads = sc.conf.getInt(
+    "spark.cleaner.referenceTracking.cleanupThreadNumber", 100)
+
+  private val cleanupExecutorPool: ExecutorService =
+    ThreadUtils.newDaemonFixedThreadPool(cleanupTaskThreads, "cleanup")
+
   @volatile private var stopped = false
 
   /** Attach a listener object to get information of when objects are cleaned. */
@@ -178,34 +184,43 @@ private[spark] class ContextCleaner(sc: SparkContext) extends Logging {
   private def keepCleaning(): Unit = Utils.tryOrStopSparkContext(sc) {
     while (!stopped) {
       try {
-        val reference = Option(referenceQueue.remove(ContextCleaner.REF_QUEUE_POLL_TIMEOUT))
-          .map(_.asInstanceOf[CleanupTaskWeakReference])
-        // Synchronize here to avoid being interrupted on stop()
-        synchronized {
-          reference.foreach { ref =>
-            logDebug("Got cleaning task " + ref.task)
-            referenceBuffer.remove(ref)
-            ref.task match {
-              case CleanRDD(rddId) =>
-                doCleanupRDD(rddId, blocking = blockOnCleanupTasks)
-              case CleanShuffle(shuffleId) =>
-                doCleanupShuffle(shuffleId, blocking = blockOnShuffleCleanupTasks)
-              case CleanBroadcast(broadcastId) =>
-                doCleanupBroadcast(broadcastId, blocking = blockOnCleanupTasks)
-              case CleanAccum(accId) =>
-                doCleanupAccum(accId, blocking = blockOnCleanupTasks)
-              case CleanCheckpoint(rddId) =>
-                doCleanCheckpoint(rddId)
-            }
-          }
+        Option(referenceQueue.remove(ContextCleaner.REF_QUEUE_POLL_TIMEOUT))
+          .map(_.asInstanceOf[CleanupTaskWeakReference]).foreach {
+          r =>
+            referenceBuffer.remove(r)
+            runtCleanTask(r)
         }
       } catch {
         case ie: InterruptedException if stopped => // ignore
-        case e: Exception => logError("Error in cleaning thread", e)
+        case e: Exception => logError("Error in cleaning main thread", e)
       }
     }
   }
 
+  private def runtCleanTask(ref: CleanupTaskWeakReference) = {
+    cleanupExecutorPool.submit(new Runnable {
+      override def run(): Unit = {
+        try {
+          ref.task match {
+            case CleanRDD(rddId) =>
+              doCleanupRDD(rddId, blocking = blockOnCleanupTasks)
+            case CleanShuffle(shuffleId) =>
+              doCleanupShuffle(shuffleId, blocking = blockOnShuffleCleanupTasks)
+            case CleanBroadcast(broadcastId) =>
+              doCleanupBroadcast(broadcastId, blocking = blockOnCleanupTasks)
+            case CleanAccum(accId) =>
+              doCleanupAccum(accId, blocking = blockOnCleanupTasks)
+            case CleanCheckpoint(rddId) =>
+              doCleanCheckpoint(rddId)
+          }
+        } catch {
+          case ie: InterruptedException if stopped => // ignore
+          case e: Exception => logError("Error in cleaning thread", e)
+        }
+      }
+    })
+  }
+
   /** Perform RDD cleanup. */
   def doCleanupRDD(rddId: Int, blocking: Boolean): Unit = {
     try {

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

@@ -623,7 +623,7 @@ private[spark] object MapOutputTracker extends Logging {
     if (arr.length >= minBroadcastSize) {
       // Use broadcast instead.
       // Important arr(0) is the tag == DIRECT, ignore that while deserializing !
-      val bcast = broadcastManager.newBroadcast(arr, isLocal)
+      val bcast = broadcastManager.newBroadcast(arr, isLocal, null)
       // toByteArray creates copy, so we can reuse out
       out.reset()
       out.write(BROADCAST)