SHS-NG M4.7: Remove JobProgressListener.

core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala

@@ -539,6 +539,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
     val logInput = EventLoggingListener.openEventLog(logPath, fs)
     try {
       bus.replay(logInput, logPath.toString, !appCompleted)
+      store.close()
     } catch {
       case e: Exception =>
         store.close()
@@ -555,6 +556,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
     }
     Utils.deleteRecursively(tempUiPath)
     listing.write(new LogInfo(logPath.toString(), fileStatus.getLen()))
+    logInfo(s"Finished parsing $logPath")
   }
 
   /**

core/src/main/scala/org/apache/spark/status/AppStateStore.scala

@@ -32,7 +32,10 @@ import org.apache.spark.util.{Distribution, Utils}
 /**
  * A wrapper around a KVStore that provides methods for accessing the API data stored within.
  */
-private[spark] class AppStateStore private (val store: KVStore, tempStorePath: Option[File]) {
+private[spark] class AppStateStore private (
+    val store: KVStore,
+    tempStorePath: Option[File],
+    listener: Option[AppStateListener]) {
 
   def appInfo(): v1.ApplicationInfo = {
     // There should be a single application info for a UIStore instance, so do no checks here.
@@ -326,6 +329,7 @@ private[spark] class AppStateStore private (val store: KVStore, tempStorePath: O
   }
 
   def close(): Unit = {
+    listener.foreach(_.stop())
     store.close()
     tempStorePath.foreach(Utils.deleteRecursively)
   }
@@ -334,11 +338,13 @@ private[spark] class AppStateStore private (val store: KVStore, tempStorePath: O
 
 private[spark] object AppStateStore {
 
+  import config._
+
   val CURRENT_VERSION = 1L
 
   /** Loads a UI store from the given path, creating an empty store if it doesn't exist. */
   def loadStore(conf: SparkConf, path: File): AppStateStore = {
-    new AppStateStore(loadStore(path), None)
+    new AppStateStore(loadStore(path), None, None)
   }
 
   /**
@@ -347,25 +353,30 @@ private[spark] object AppStateStore {
    */
   def createTempStore(conf: SparkConf, bus: SparkListenerBus): AppStateStore = {
     val temp = Utils.createTempDir(namePrefix = "appstate")
-    initializeStore(loadStore(temp), Some(temp), bus)
+    initializeStore(conf, loadStore(temp), Some(temp), bus)
   }
 
   /**
    * Create a store in the given path, attaching a listener to the given bus to populate the
    * store. The path will not be deleted after the store is closed.
    */
   def createStore(path: File, conf: SparkConf, bus: SparkListenerBus): AppStateStore = {
-    initializeStore(loadStore(path), None, bus)
+    initializeStore(conf, loadStore(path), None, bus)
   }
 
   private def initializeStore(
+      conf: SparkConf,
       store: KVStore,
       tempPath: Option[File],
       bus: SparkListenerBus): AppStateStore = {
-    val stateStore = new AppStateStore(store, tempPath)
-    val listener = new AppStateListener(store)
+    val cachingStore = if (conf.get(MAX_CACHED_ELEMENTS) > 0) {
+      new AsyncCachingStore(store, conf)
+    } else {
+      store
+    }
+    val listener = new AppStateListener(cachingStore, conf)
     bus.addListener(listener)
-    stateStore
+    new AppStateStore(cachingStore, tempPath, Some(listener))
   }
 
   private def loadStore(path: File): KVStore = {

core/src/main/scala/org/apache/spark/status/AsyncCachingStore.scala

@@ -0,0 +1,282 @@
+/*
+ * 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.status
+
+import java.util.{Arrays, HashMap, LinkedHashMap, Map => JMap}
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.locks.ReentrantLock
+
+import com.codahale.metrics.{MetricRegistry, Slf4jReporter}
+import com.google.common.base.Objects
+import com.google.common.cache.CacheBuilder
+
+import org.apache.spark.SparkConf
+import org.apache.spark.internal.Logging
+import org.apache.spark.kvstore._
+import org.apache.spark.util.{ThreadUtils, Utils}
+
+/**
+ * A KVStore implementation that provides an LRU cache of elements to speed up access, and writes
+ * to the underlying store asynchronously.
+ *
+ * Caching is only available for elements read using the [[#read(Class, Object)]] method. Iterators
+ * are not cached and in fact can return stale data when compared to what the
+ * [[#read(Class, Object)]] method might return.
+ *
+ * Writes are performed on a separate thread, which does de-duplication to avoid redundant writes.
+ * Because of those features, writers should always update entries based on what the
+ * [[#read(Class, Object)]] method returns, otherwise they may be operating on stale data.
+ */
+private class AsyncCachingStore(store: KVStore, conf: SparkConf) extends KVStore with Logging {
+
+  import config._
+  import KVUtils._
+
+  private val cache = CacheBuilder.newBuilder()
+    .maximumSize(conf.get(MAX_CACHED_ELEMENTS))
+    .build[CacheKey, AnyRef]()
+
+  private val writeThread = new Thread(new Runnable() {
+    override def run(): Unit = Utils.tryLog {
+      writeThreadImpl()
+    }
+  })
+
+  private val types = new HashMap[Class[_], KVTypeInfo]()
+  private val writeQueue = new LinkedHashMap[CacheKey, AsyncStoreOp]()
+  private val writeLock = new ReentrantLock()
+  private val writeSignal = writeLock.newCondition()
+  private val maxWriteQueueSize = conf.get(MAX_WRITE_QUEUE_SIZE)
+
+  private val metrics = new MetricRegistry()
+  private val hits = metrics.counter("hits")
+  private val misses = metrics.counter("misses")
+  private val writes = metrics.counter("writes")
+  private val writeTimer = metrics.timer("writeTimer")
+  private val maxQueued = new AtomicLong()
+
+  @volatile private var active = true
+
+  writeThread.setName("caching-store-writer")
+  writeThread.setDaemon(true)
+  writeThread.start()
+
+  override def read[T](klass: Class[T], naturalKey: AnyRef): T = {
+    val cacheKey = new CacheKey(klass, naturalKey)
+    val cached = cache.getIfPresent(cacheKey)
+    if (cached != null) {
+      hits.inc()
+      return cached.asInstanceOf[T]
+    }
+
+    // There is a possibility that an entry evicted from the cache has pending writes, so check
+    // the last write op too.
+    locked(writeLock) {
+      writeQueue.get(cacheKey) match {
+        case op: WriteOp =>
+          val value = op.value
+          if (value != null) {
+            hits.inc()
+            return value.asInstanceOf[T]
+          }
+
+        case _: DeleteOp =>
+          throw new NoSuchElementException()
+
+        case _ => // continue.
+      }
+    }
+
+    val stored = store.read(klass, naturalKey)
+    misses.inc()
+    cache.put(cacheKey, stored.asInstanceOf[AnyRef])
+    stored
+  }
+
+  override def write(value: AnyRef): Unit = {
+    var ti = types.get(value.getClass())
+    if (ti == null) {
+      ti = new KVTypeInfo(value.getClass())
+      types.put(value.getClass(), ti)
+    }
+
+    val key = ti.getIndexValue(KVIndex.NATURAL_INDEX_NAME, value)
+    val cacheKey = new CacheKey(value.getClass(), key)
+    cache.put(cacheKey, value)
+
+    enqueue(WriteOp(value, cacheKey))
+  }
+
+  override def delete(klass: Class[_], naturalKey: AnyRef): Unit = {
+    val cacheKey = new CacheKey(klass, naturalKey)
+    cache.invalidate(cacheKey)
+    enqueue(DeleteOp(cacheKey))
+  }
+
+  override def getMetadata[T](klass: Class[T]): T = store.getMetadata(klass)
+
+  override def setMetadata(value: AnyRef): Unit = store.setMetadata(value)
+
+  override def view[T](klass: Class[T]): KVStoreView[T] = store.view(klass)
+
+  override def count(klass: Class[_]): Long = store.count(klass)
+
+  override def close(): Unit = {
+    try {
+      val remaining = metrics.counter("writeBacklog")
+      remaining.inc(writeQueue.size())
+
+      val timer = metrics.timer("writeQueueDrain")
+      timeIt(timer) {
+        active = false
+        locked(writeLock) {
+          writeSignal.signal()
+        }
+        writeThread.join(TimeUnit.SECONDS.toMillis(30))
+        if (writeThread.isAlive()) {
+          logWarning(s"Write queue hasn't drained in 30 seconds, queue size = ${writeQueue.size()}")
+          writeThread.interrupt()
+        } else {
+          assert(writeQueue.size == 0, "Unwritten items left over in write queue.")
+        }
+      }
+
+      metrics.counter("maxQueueSize").inc(maxQueued.get())
+      Slf4jReporter.forRegistry(metrics)
+        .outputTo(log)
+        .withLoggingLevel(Slf4jReporter.LoggingLevel.DEBUG)
+        .build()
+        .report()
+
+      cache.invalidateAll()
+    } finally {
+      store.close()
+    }
+  }
+
+  private def enqueue(op: AsyncStoreOp): Unit = {
+    locked(writeLock) {
+      // If not replacing an existing item in the queue, wait until there's room before queueing
+      // the write.
+      if (writeQueue.remove(op.cacheKey) == null) {
+        while (writeQueue.size() >= maxWriteQueueSize) {
+          writeSignal.await()
+        }
+      }
+      writeQueue.put(op.cacheKey, op)
+      maxQueued.set(math.max(writeQueue.size(), maxQueued.get()))
+      writeSignal.signal()
+    }
+    writes.inc()
+  }
+
+  private def writeThreadImpl(): Unit = {
+    while (active || !writeQueue.isEmpty()) {
+      Utils.tryLog {
+        val next = locked(writeLock) {
+          while (active && writeQueue.isEmpty()) {
+            writeSignal.await()
+          }
+
+          if (!writeQueue.isEmpty()) {
+            val iter = writeQueue.entrySet().iterator()
+            val _next = iter.next()
+            iter.remove()
+            writeSignal.signal()
+            _next.getValue()
+          } else {
+            null
+          }
+        }
+
+        if (next != null) {
+          timeIt(writeTimer) {
+            next.perform()
+          }
+        }
+      }
+    }
+  }
+
+  // For testing. Halts the write thread.
+  private[status] def haltWrites(): Unit = {
+    writeLock.lock()
+  }
+
+  // For testing. Resumes the write thread.
+  private[status] def resumeWrites(): Unit = {
+    writeLock.unlock()
+  }
+
+  private sealed abstract class AsyncStoreOp(val cacheKey: CacheKey) {
+
+    def perform(): Unit
+
+  }
+
+  private case class WriteOp(value: AnyRef, key: CacheKey) extends AsyncStoreOp(key) {
+
+    override def perform(): Unit = store.write(value)
+
+  }
+
+  private case class DeleteOp(key: CacheKey) extends AsyncStoreOp(key) {
+
+    override def perform(): Unit = store.delete(cacheKey.klass, cacheKey.key)
+
+  }
+
+}
+
+private class CacheKey(val klass: Class[_], val key: AnyRef) {
+
+  private val isArray = key != null && key.getClass().isArray()
+
+  override def equals(o: Any): Boolean = {
+    if (o == null || !o.isInstanceOf[CacheKey]) {
+      false
+    }
+
+    val other = o.asInstanceOf[CacheKey]
+    if (!Objects.equal(klass, other.klass)) {
+      false
+    } else if (Objects.equal(key, other.key)) {
+      true
+    } else if (isArray && other.key.getClass().isArray()) {
+      val a1 = key.asInstanceOf[Array[_]]
+      val a2 = other.key.asInstanceOf[Array[_]]
+      a1.length == a2.length &&
+        a1.getClass().getComponentType() == a2.getClass().getComponentType() &&
+        a1.zip(a2).forall { case (v1, v2) => v1 == v2 }
+    } else {
+      false
+    }
+  }
+
+  override def hashCode(): Int = {
+    if (isArray) {
+      31 * key.asInstanceOf[Array[_]].map(_.hashCode()).sum
+    } else if (key != null) {
+      key.hashCode()
+    } else {
+      0
+    }
+  }
+
+}