/**
 * Submit a job to the job scheduler and get a JobWaiter object back. The JobWaiter object
 * can be used to block until the the job finishes executing or can be used to cancel the job.
 */
def submitJob[T, U](
    rdd: RDD[T],
    func: (TaskContext, Iterator[T]) => U,
    partitions: Seq[Int],
    callSite: CallSite,
    allowLocal: Boolean,
    resultHandler: (Int, U) => Unit,
    properties: Properties = null): JobWaiter[U] =
{
  // Check to make sure we are not launching a task on a partition that does not exist.
  val maxPartitions = rdd.partitions.length
  partitions.find(p => p >= maxPartitions || p < 0).foreach { p =>
    throw new IllegalArgumentException(
      "Attempting to access a non-existent partition: " + p + ". " +
        "Total number of partitions: " + maxPartitions)
  }

  val jobId = nextJobId.getAndIncrement()
  if (partitions.size == 0) {
    return new JobWaiter[U](this, jobId, 0, resultHandler)
  }

  assert(partitions.size > 0)
  val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
  val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
  eventProcessActor ! JobSubmitted(
    jobId, rdd, func2, partitions.toArray, allowLocal, callSite, waiter, properties)
  waiter
}

/**
 * Run a job on an RDD locally, assuming it has only a single partition and no dependencies.
 * We run the operation in a separate thread just in case it takes a bunch of time, so that we
 * don‘t block the DAGScheduler event loop or other concurrent jobs.
 */
protected def runLocally(job: ActiveJob) {
  logInfo("Computing the requested partition locally")
  new Thread("Local computation of job " + job.jobId) {
    override def run() {
      runLocallyWithinThread(job)
    }
  }.start()
}

private[scheduler] def handleJobSubmitted(jobId: Int,
    finalRDD: RDD[_],
    func: (TaskContext, Iterator[_]) => _,
    partitions: Array[Int],
    allowLocal: Boolean,
    callSite: CallSite,
    listener: JobListener,
    properties: Properties = null)
{

/** Submits stage, but first recursively submits any missing parents. */
private def submitStage(stage: Stage) {

/** Called when stage‘s parents are available and we can now do its task. */
private def submitMissingTasks(stage: Stage, jobId: Int) {

/** Finds the earliest-created active job that needs the stage */
// TODO: Probably should actually find among the active jobs that need this
// stage the one with the highest priority (highest-priority pool, earliest created).
// That should take care of at least part of the priority inversion problem with
// cross-job dependencies.
private def activeJobForStage(stage: Stage): Option[Int] = {
  val jobsThatUseStage: Array[Int] = stage.jobIds.toArray.sorted
  jobsThatUseStage.find(jobIdToActiveJob.contains)
}

/**
 * Asynchronously passes SparkListenerEvents to registered SparkListeners.
 *
 * Until start() is called, all posted events are only buffered. Only after this listener bus
 * has started will events be actually propagated to all attached listeners. This listener bus
 * is stopped when it receives a SparkListenerShutdown event, which is posted using stop().
 */
private[spark] class LiveListenerBus extends SparkListenerBus with Logging {

/**
 * A SparkListenerEvent bus that relays events to its listeners
 */
private[spark] trait SparkListenerBus extends Logging {

  // SparkListeners attached to this event bus
  protected val sparkListeners = new ArrayBuffer[SparkListener]
    with mutable.SynchronizedBuffer[SparkListener]

  def addListener(listener: SparkListener) {
    sparkListeners += listener
  }

  /**
   * Post an event to all attached listeners.
   * This does nothing if the event is SparkListenerShutdown.
   */
  def postToAll(event: SparkListenerEvent) {

  /**
   * Apply the given function to all attached listeners, catching and logging any exception.
   */
  private def foreachListener(f: SparkListener => Unit): Unit = {
    sparkListeners.foreach { listener =>
      try {
        f(listener)
      } catch {
        case e: Exception =>
          logError(s"Listener ${Utils.getFormattedClassName(listener)} threw an exception", e)
      }
    }
  }

}