forked from apache/spark
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[SPARK-18838][CORE] Add separate listener queues to LiveListenerBus.
This change modifies the live listener bus so that all listeners are added to queues; each queue has its own thread to dispatch events, making it possible to separate slow listeners from other more performance-sensitive ones. The public API has not changed - all listeners added with the existing "addListener" method, which after this change mostly means all user-defined listeners, end up in a default queue. Internally, there's an API allowing listeners to be added to specific queues, and that API is used to separate the internal Spark listeners into 3 categories: application status listeners (e.g. UI), executor management (e.g. dynamic allocation), and the event log. The queueing logic, while abstracted away in a separate class, is kept as much as possible hidden away from consumers. Aside from choosing their queue, there's no code change needed to take advantage of queues. Test coverage relies on existing tests; a few tests had to be tweaked because they relied on `LiveListenerBus.postToAll` being synchronous, and the change makes that method asynchronous. Other tests were simplified not to use the asynchronous LiveListenerBus. Author: Marcelo Vanzin <vanzin@cloudera.com> Closes apache#19211 from vanzin/SPARK-18838.
- Loading branch information
Showing
16 changed files
with
473 additions
and
290 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
196 changes: 196 additions & 0 deletions
196
core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,196 @@ | ||
| /* | ||
| * 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.scheduler | ||
|
|
||
| import java.util.concurrent.LinkedBlockingQueue | ||
| import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong} | ||
|
|
||
| import com.codahale.metrics.{Gauge, Timer} | ||
|
|
||
| import org.apache.spark.{SparkConf, SparkContext} | ||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.internal.config._ | ||
| import org.apache.spark.util.Utils | ||
|
|
||
| /** | ||
| * An asynchronous queue for events. All events posted to this queue will be delivered to the child | ||
| * listeners in a separate thread. | ||
| * | ||
| * Delivery will only begin when the `start()` method is called. The `stop()` method should be | ||
| * called when no more events need to be delivered. | ||
| */ | ||
| private class AsyncEventQueue(val name: String, conf: SparkConf, metrics: LiveListenerBusMetrics) | ||
| extends SparkListenerBus | ||
| with Logging { | ||
|
|
||
| import AsyncEventQueue._ | ||
|
|
||
| // Cap the capacity of the queue so we get an explicit error (rather than an OOM exception) if | ||
| // it's perpetually being added to more quickly than it's being drained. | ||
| private val eventQueue = new LinkedBlockingQueue[SparkListenerEvent]( | ||
| conf.get(LISTENER_BUS_EVENT_QUEUE_CAPACITY)) | ||
|
|
||
| // Keep the event count separately, so that waitUntilEmpty() can be implemented properly; | ||
| // this allows that method to return only when the events in the queue have been fully | ||
| // processed (instead of just dequeued). | ||
| private val eventCount = new AtomicLong() | ||
|
|
||
| /** A counter for dropped events. It will be reset every time we log it. */ | ||
| private val droppedEventsCounter = new AtomicLong(0L) | ||
|
|
||
| /** When `droppedEventsCounter` was logged last time in milliseconds. */ | ||
| @volatile private var lastReportTimestamp = 0L | ||
|
|
||
| private val logDroppedEvent = new AtomicBoolean(false) | ||
|
|
||
| private var sc: SparkContext = null | ||
|
|
||
| private val started = new AtomicBoolean(false) | ||
| private val stopped = new AtomicBoolean(false) | ||
|
|
||
| private val droppedEvents = metrics.metricRegistry.counter(s"queue.$name.numDroppedEvents") | ||
| private val processingTime = metrics.metricRegistry.timer(s"queue.$name.listenerProcessingTime") | ||
|
|
||
| // Remove the queue size gauge first, in case it was created by a previous incarnation of | ||
| // this queue that was removed from the listener bus. | ||
| metrics.metricRegistry.remove(s"queue.$name.size") | ||
| metrics.metricRegistry.register(s"queue.$name.size", new Gauge[Int] { | ||
| override def getValue: Int = eventQueue.size() | ||
| }) | ||
|
|
||
| private val dispatchThread = new Thread(s"spark-listener-group-$name") { | ||
| setDaemon(true) | ||
| override def run(): Unit = Utils.tryOrStopSparkContext(sc) { | ||
| dispatch() | ||
| } | ||
| } | ||
|
|
||
| private def dispatch(): Unit = LiveListenerBus.withinListenerThread.withValue(true) { | ||
| try { | ||
| var next: SparkListenerEvent = eventQueue.take() | ||
| while (next != POISON_PILL) { | ||
| val ctx = processingTime.time() | ||
| try { | ||
| super.postToAll(next) | ||
| } finally { | ||
| ctx.stop() | ||
| } | ||
| eventCount.decrementAndGet() | ||
| next = eventQueue.take() | ||
| } | ||
| eventCount.decrementAndGet() | ||
| } catch { | ||
| case ie: InterruptedException => | ||
| logInfo(s"Stopping listener queue $name.", ie) | ||
| } | ||
| } | ||
|
|
||
| override protected def getTimer(listener: SparkListenerInterface): Option[Timer] = { | ||
| metrics.getTimerForListenerClass(listener.getClass.asSubclass(classOf[SparkListenerInterface])) | ||
| } | ||
|
|
||
| /** | ||
| * Start an asynchronous thread to dispatch events to the underlying listeners. | ||
| * | ||
| * @param sc Used to stop the SparkContext in case the async dispatcher fails. | ||
| */ | ||
| private[scheduler] def start(sc: SparkContext): Unit = { | ||
| if (started.compareAndSet(false, true)) { | ||
| this.sc = sc | ||
| dispatchThread.start() | ||
| } else { | ||
| throw new IllegalStateException(s"$name already started!") | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Stop the listener bus. It will wait until the queued events have been processed, but new | ||
| * events will be dropped. | ||
| */ | ||
| private[scheduler] def stop(): Unit = { | ||
| if (!started.get()) { | ||
| throw new IllegalStateException(s"Attempted to stop $name that has not yet started!") | ||
| } | ||
| if (stopped.compareAndSet(false, true)) { | ||
| eventQueue.put(POISON_PILL) | ||
| eventCount.incrementAndGet() | ||
| } | ||
| dispatchThread.join() | ||
| } | ||
|
|
||
| def post(event: SparkListenerEvent): Unit = { | ||
| if (stopped.get()) { | ||
| return | ||
| } | ||
|
|
||
| eventCount.incrementAndGet() | ||
| if (eventQueue.offer(event)) { | ||
| return | ||
| } | ||
|
|
||
| eventCount.decrementAndGet() | ||
| droppedEvents.inc() | ||
| droppedEventsCounter.incrementAndGet() | ||
| if (logDroppedEvent.compareAndSet(false, true)) { | ||
| // Only log the following message once to avoid duplicated annoying logs. | ||
| logError(s"Dropping event from queue $name. " + | ||
| "This likely means one of the listeners is too slow and cannot keep up with " + | ||
| "the rate at which tasks are being started by the scheduler.") | ||
| } | ||
| logTrace(s"Dropping event $event") | ||
|
|
||
| val droppedCount = droppedEventsCounter.get | ||
| if (droppedCount > 0) { | ||
| // Don't log too frequently | ||
| if (System.currentTimeMillis() - lastReportTimestamp >= 60 * 1000) { | ||
| // There may be multiple threads trying to decrease droppedEventsCounter. | ||
| // Use "compareAndSet" to make sure only one thread can win. | ||
| // And if another thread is increasing droppedEventsCounter, "compareAndSet" will fail and | ||
| // then that thread will update it. | ||
| if (droppedEventsCounter.compareAndSet(droppedCount, 0)) { | ||
| val prevLastReportTimestamp = lastReportTimestamp | ||
| lastReportTimestamp = System.currentTimeMillis() | ||
| val previous = new java.util.Date(prevLastReportTimestamp) | ||
| logWarning(s"Dropped $droppedEvents events from $name since $previous.") | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * For testing only. Wait until there are no more events in the queue. | ||
| * | ||
| * @return true if the queue is empty. | ||
| */ | ||
| def waitUntilEmpty(deadline: Long): Boolean = { | ||
| while (eventCount.get() != 0) { | ||
| if (System.currentTimeMillis > deadline) { | ||
| return false | ||
| } | ||
| Thread.sleep(10) | ||
| } | ||
| true | ||
| } | ||
|
|
||
| } | ||
|
|
||
| private object AsyncEventQueue { | ||
|
|
||
| val POISON_PILL = new SparkListenerEvent() { } | ||
|
|
||
| } |
Oops, something went wrong.