[CARMEL-3152] driver scheduling enhancement (#58)

Author: chrysanxia

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

@@ -17,6 +17,7 @@
 
 package org.apache.spark.scheduler
 
+import java.util
 import java.util.concurrent.LinkedBlockingQueue
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
 
@@ -98,18 +99,40 @@ private class AsyncEventQueue(
   }
 
   private def dispatch(): Unit = LiveListenerBus.withinListenerThread.withValue(true) {
-    var next: SparkListenerEvent = eventQueue.take()
-    while (next != POISON_PILL) {
-      val ctx = processingTime.time()
-      try {
-        super.postToAll(next)
-      } finally {
-        ctx.stop()
+    try {
+      val events = new util.ArrayList[SparkListenerEvent](100)
+      var isPoisonPill = false
+      while (!isPoisonPill) {
+        events.clear()
+        drainFromEventQueue(events, 99)
+        val it = events.iterator()
+        while (it.hasNext) {
+          val next = it.next()
+          if (next == POISON_PILL) {
+            isPoisonPill = true
+          } else {
+            val ctx = processingTime.time()
+            try {
+              super.postToAll(next)
+            } finally {
+              ctx.stop()
+            }
+            eventCount.decrementAndGet()
+          }
+        }
       }
       eventCount.decrementAndGet()
-      next = eventQueue.take()
+
+    } catch {
+      case ie: InterruptedException =>
+        logInfo(s"Stopping listener queue $name.", ie)
+    }
+  }
+
+  private def drainFromEventQueue(collection: util.Collection[SparkListenerEvent], size: Int) = {
+    if (eventQueue.drainTo(collection, size) == 0) {
+      collection.add(eventQueue.take())
     }
-    eventCount.decrementAndGet()
   }
 
   override protected def getTimer(listener: SparkListenerInterface): Option[Timer] = {