Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/4525#discussion_r24634640
--- Diff:
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala ---
@@ -119,14 +138,79 @@ private[history] class FsHistoryProvider(conf:
SparkConf) extends ApplicationHis
if (!conf.contains("spark.testing")) {
logCheckingThread.setDaemon(true)
logCheckingThread.start()
+ logLazyReplayThread.setDaemon(true)
+ logLazyReplayThread.start()
+ } else {
+ logLazyReplay()
}
}
- override def getListing() = applications.values
+ /**
+ * Fetch and Parse the log files
+ */
+ private[history] def logLazyReplay() {
+ if(lazyApplications.isEmpty) return
+
+ logDebug("start doLazyReplay")
+ val mergeSize = 20
+ val bufferedApps = new ArrayBuffer[FsApplicationHistoryInfo](mergeSize)
+
+ def addIfAbsent(newApps: mutable.LinkedHashMap[String,
FsApplicationHistoryInfo],
+ info: FsApplicationHistoryInfo) {
+ if (!newApps.contains(info.id) ||
+
newApps(info.id).logPath.endsWith(EventLoggingListener.IN_PROGRESS) &&
+ !info.logPath.endsWith(EventLoggingListener.IN_PROGRESS)) {
+ newApps += (info.id -> info)
+ }
+ }
+
+ def mergeApps(): mutable.LinkedHashMap[String,
FsApplicationHistoryInfo] = {
+ val newApps = new mutable.LinkedHashMap[String,
FsApplicationHistoryInfo]()
+ bufferedApps.sortWith(compareAppInfo)
+
+ val newIterator = bufferedApps.iterator.buffered
+ val oldIterator = applications.values.iterator.buffered
+ while (newIterator.hasNext && oldIterator.hasNext) {
+ if (compareAppInfo(newIterator.head, oldIterator.head)) {
+ addIfAbsent(newApps, newIterator.next())
+ } else {
+ addIfAbsent(newApps, oldIterator.next())
+ }
+ }
+ newIterator.foreach(addIfAbsent(newApps, _))
+ oldIterator.foreach(addIfAbsent(newApps, _))
+
+ newApps
+ }
+
+ val bus = new ReplayListenerBus()
+ while(lazyApplications.nonEmpty){
--- End diff --
So, this feels a little racy, in that this thread might miss things added
by the log checking thread. I'd suggest the following:
- Create a single-threaded executor for running the replay tasks
- Create a list of app infos to parse in the log checking thread, break it
down into batches.
- Submit each batch to the executor
Basically, instead of having `logLazyReplay`, you'd have something like
`replay(apps: Seq[LazyAppInfo])`. You don't need `lazyApplications` because
that becomes part of the task being submitted to the executor, so you solve
another source of contention in the code. And since it's a single-threaded
executor, you know there's only a single thread touching `apps`, so it should
all be thread-safe.
For testing, you can use Guava's `sameThreadExecutor()` as I mentioned,
instead of the single-threaded executor.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
