// SparkContext.scala 核心初始化流程
class SparkContext(config: SparkConf) extends Logging {
  
  // 1. 创建SparkEnv - 核心运行环境
  private val env: SparkEnv = {
    SparkEnv.createDriverEnv(conf, isLocal, listenerBus, numCores, mockOutputCommitCoordinator)
  }
  
  // 2. 创建状态跟踪器
  private val statusTracker = new SparkStatusTracker(this, sparkUI)
  
  // 3. 创建任务调度器
  private val (sched, ts) = SparkContext.createTaskScheduler(this, master, deployMode)
  private val taskScheduler = ts
  
  // 4. 创建DAG调度器
  private val dagScheduler = new DAGScheduler(this)
  
  // 5. 启动任务调度器
  taskScheduler.start()
  
  // 6. 设置默认并行度
  private val defaultParallelism: Int = taskScheduler.defaultParallelism
  
  // 核心方法：创建RDD
  def parallelize[T: ClassTag](
      seq: Seq[T],
      numSlices: Int = defaultParallelism): RDD[T] = withScope {
    new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())
  }
  
  // 核心方法：提交作业
  def runJob[T, U: ClassTag](
      rdd: RDD[T],
      func: (TaskContext, Iterator[T]) => U,
      partitions: Seq[Int],
      resultHandler: (Int, U) => Unit): Unit = {
    dagScheduler.runJob(rdd, func, partitions, callSite, resultHandler, localProperties.get)
  }
}

// RDD.scala 核心抽象
abstract class RDD[T: ClassTag](
    @transient private var _sc: SparkContext,
    @transient private var deps: Seq[Dependency[_]]
  ) extends Serializable with Logging {

  // 五大特性的具体实现
  
  // 1. 分区列表
  protected def getPartitions: Array[Partition]
  
  // 2. 计算函数
  def compute(split: Partition, context: TaskContext): Iterator[T]
  
  // 3. 依赖关系
  protected def getDependencies: Seq[Dependency[_]] = deps
  
  // 4. 分区器（可选）
  @transient val partitioner: Option[Partitioner] = None
  
  // 5. 位置偏好（可选）
  protected def getPreferredLocations(split: Partition): Seq[String] = Nil
  
  // Transformation操作实现
  def map[U: ClassTag](f: T => U): RDD[U] = withScope {
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
  }
  
  def filter(f: T => Boolean): RDD[T] = withScope {
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[T, T](
      this,
      (context, pid, iter) => iter.filter(cleanF),
      preservesPartitioning = true)
  }
  
  def reduceByKey(func: (T, T) => T): RDD[T] = self.withScope {
    reduceByKey(defaultPartitioner(self), func)
  }
  
  // Action操作实现
  def collect(): Array[T] = withScope {
    val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
    Array.concat(results: _*)
  }
  
  def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum
  
  def foreach(f: T => Unit): Unit = withScope {
    val cleanF = sc.clean(f)
    sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
  }
}

// DAGScheduler.scala 核心调度逻辑
class DAGScheduler(
    private[scheduler] val sc: SparkContext,
    private[scheduler] val taskScheduler: TaskScheduler,
    listenerBus: LiveListenerBus,
    mapOutputTracker: MapOutputTrackerMaster,
    blockManagerMaster: BlockManagerMaster,
    env: SparkEnv,
    clock: Clock = new SystemClock())
  extends Logging {

  // 事件处理循环
  private val eventProcessLoop = new DAGSchedulerEventProcessLoop(this)
  
  // 提交作业的核心方法
  def runJob[T, U](
      rdd: RDD[T],
      func: (TaskContext, Iterator[T]) => U,
      partitions: Seq[Int],
      callSite: CallSite,
      resultHandler: (Int, U) => Unit,
      properties: Properties): Unit = {
    
    val start = System.nanoTime
    val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
    ThreadUtils.awaitReady(waiter, Duration.Inf)
    waiter.value.get match {
      case scala.util.Success(_) =>
        logInfo("Job %d finished: %s, took %f s".format
          (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
      case scala.util.Failure(exception) =>
        logInfo("Job %d failed: %s, took %f s".format
          (waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
        throw exception
    }
  }
  
  // Stage划分核心算法
  private def getOrCreateShuffleMapStage(
      shuffleDep: ShuffleDependency[_, _, _],
      firstJobId: Int): ShuffleMapStage = {
    
    shuffleIdToMapStage.get(shuffleDep.shuffleId) match {
      case Some(stage) =>
        stage
        
      case None =>
        // 递归创建父Stage
        getMissingAncestorShuffleDependencies(shuffleDep.rdd).foreach { dep =>
          if (!shuffleIdToMapStage.contains(dep.shuffleId)) {
            createShuffleMapStage(dep, firstJobId)
          }
        }
        createShuffleMapStage(shuffleDep, firstJobId)
    }
  }
  
  // 查找缺失的父依赖
  private def getMissingAncestorShuffleDependencies(
      rdd: RDD[_]): ArrayStack[ShuffleDependency[_, _, _]] = {
    val ancestors = new ArrayStack[ShuffleDependency[_, _, _]]
    val visited = new HashSet[RDD[_]]
    val waitingForVisit = new ArrayStack[RDD[_]]
    
    waitingForVisit.push(rdd)
    while (waitingForVisit.nonEmpty) {
      val toVisit = waitingForVisit.pop()
      if (!visited(toVisit)) {
        visited += toVisit
        toVisit.dependencies.foreach {
          case shuffleDep: ShuffleDependency[_, _, _] =>
            if (!shuffleIdToMapStage.contains(shuffleDep.shuffleId)) {
              ancestors.push(shuffleDep)
              waitingForVisit.push(shuffleDep.rdd)
            }
          case narrowDep: NarrowDependency[_] =>
            waitingForVisit.push(narrowDep.rdd)
        }
      }
    }
    ancestors
  }
  
  // 提交Stage
  private def submitStage(stage: Stage): Unit = {
    val jobId = activeJobForStage(stage)
    if (jobId.isDefined) {
      if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
        val missing = getMissingParentStages(stage).sortBy(_.id)
        if (missing.isEmpty) {
          submitMissingTasks(stage, jobId.get)
        } else {
          for (parent <- missing) {
            submitStage(parent)
          }
          waitingStages += stage
        }
      }
    }
  }
}

// 内存使用监控组件
class MemoryMonitor {
  // 监控Map的内存使用
  def monitorMapMemory(map: SizeTrackingAppendOnlyMap[_, _]): MemoryUsage = {
    val estimatedSize = map.estimateSize()
    val currentMemory = map.currentMemory
    val maxMemory = map.maxMemory
    
    MemoryUsage(
      estimatedSize = estimatedSize,
      currentMemory = currentMemory,
      maxMemory = maxMemory,
      utilization = currentMemory.toDouble / maxMemory
    )
  }
  
  // 监控Spill状态
  def monitorSpillStatus(externalMap: ExternalAppendOnlyMap[_, _, _]): SpillStatus = {
    val spillCount = externalMap.spills.size
    val totalSpillSize = externalMap.spills.map(_.size).sum
    
    SpillStatus(
      spillCount = spillCount,
      totalSpillSize = totalSpillSize,
      averageSpillSize = if (spillCount > 0) totalSpillSize / spillCount else 0
    )
  }
}

case class MemoryUsage(
  estimatedSize: Long,
  currentMemory: Long,
  maxMemory: Long,
  utilization: Double)

case class SpillStatus(
  spillCount: Int,
  totalSpillSize: Long,
  averageSpillSize: Long)

// 内存分配优化
class MemoryOptimizer {
  // 动态调整内存阈值
  def adjustMemoryThreshold(
      currentMemory: Long,
      maxMemory: Long,
      spillCount: Int): Long = {
    
    val utilization = currentMemory.toDouble / maxMemory
    
    if (utilization > 0.8 && spillCount > 0) {
      // 内存使用率高且有Spill，降低阈值
      (maxMemory * 0.6).toLong
    } else if (utilization < 0.5 && spillCount == 0) {
      // 内存使用率低且无Spill，提高阈值
      (maxMemory * 0.9).toLong
    } else {
      // 保持当前阈值
      (maxMemory * 0.8).toLong
    }
  }
  
  // 优化Map初始容量
  def optimizeInitialCapacity(dataSize: Long): Int = {
    val estimatedSize = (dataSize * 1.2).toInt
    math.max(64, math.min(estimatedSize, 1024 * 1024))
  }
}

// 用户代码触发Action
val result = rdd.collect()

// SparkContext.collect()
def collect(): Array[T] = withScope {
  val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
  Array.concat(results: _*)
}

// SparkContext.runJob()
def runJob[T, U: ClassTag](
    rdd: RDD[T],
    func: (TaskContext, Iterator[T]) => U,
    partitions: Seq[Int],
    resultHandler: (Int, U) => Unit): Unit = {
  dagScheduler.runJob(rdd, func, partitions, callSite, resultHandler, localProperties.get)
}

// DAGScheduler.scala
private[scheduler] def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
  val parents = new ArrayBuffer[Stage]()
  val visited = new HashSet[RDD[_]]
  
  def visit(r: RDD[_]): Unit = {
    if (!visited(r)) {
      visited += r
      for (dep <- r.dependencies) {
        dep match {
          case shufDep: ShuffleDependency[_, _, _] =>
            // 宽依赖，创建新的ShuffleMapStage
            parents += getOrCreateShuffleMapStage(shufDep, firstJobId)
          case _ =>
            // 窄依赖，递归访问父RDD
            visit(dep.rdd)
        }
      }
    }
  }
  
  visit(rdd)
  parents.toList
}

// TaskSchedulerImpl.resourceOffers()
def resourceOffers(offers: IndexedSeq[WorkerOffer]): Seq[Seq[TaskDescription]] = {
  // 1. 随机打乱offers避免热点
  val shuffledOffers = Random.shuffle(offers)
  val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores / CPUS_PER_TASK))
  val availableCpus = shuffledOffers.map(o => o.cores).toArray
  
  // 2. 按本地性级别分配任务
  val sortedTaskSets = rootPool.getSortedTaskSetQueue
  for (taskSet <- sortedTaskSets) {
    // PROCESS_LOCAL -> NODE_LOCAL -> NO_PREF -> RACK_LOCAL -> ANY
    for (currentMaxLocality <- taskSet.myLocalityLevels) {
      do {
        launchedAnyTask = resourceOfferSingleTaskSet(
          taskSet, currentMaxLocality, shuffledOffers, availableCpus, tasks)
      } while (launchedAnyTask)
    }
  }
  
  tasks
}

// TaskSchedulerImpl.scala - 核心任务调度器实现
class TaskSchedulerImpl(
    val sc: SparkContext,
    val maxTaskFailures: Int,
    isLocal: Boolean = false)
  extends TaskScheduler with Logging {

  // 调度后端，负责与集群管理器通信
  var backend: SchedulerBackend = null
  
  // 根调度池，管理所有TaskSetManager
  val rootPool: Pool = new Pool("", SchedulingMode.FIFO, 0, 0)
  
  // 调度构建器，负责构建调度树
  var schedulableBuilder: SchedulableBuilder = null
  
  // 正在运行的TaskSet管理器
  private val taskSetsByStageIdAndAttempt = new HashMap[Int, HashMap[Int, TaskSetManager]]
  
  // 1. 提交TaskSet的核心方法
  override def submitTasks(taskSet: TaskSet): Unit = synchronized {
    val tasks = taskSet.tasks
    logInfo(s"Adding task set ${taskSet.id} with ${tasks.length} tasks")
    
    // 创建TaskSetManager来管理这个TaskSet
    val manager = createTaskSetManager(taskSet, maxTaskFailures)
    val stage = taskSet.stageId
    val stageTaskSets = taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
    
    // 检查是否有重复的TaskSet
    stageTaskSets.foreach { case (_, ts) =>
      ts.isZombie = true
    }
    stageTaskSets(taskSet.stageAttemptId) = manager
    
    // 将TaskSetManager添加到调度池
    schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
    
    // 如果不是本地模式，触发资源分配
    if (!isLocal && manager.tasks.length > 0) {
      backend.reviveOffers()
    }
  }
  
  // 2. 资源分配的核心方法
  def resourceOffers(offers: IndexedSeq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
    // 标记所有Executor为活跃状态
    var newExecAvail = false
    for (o <- offers) {
      if (!hostToExecutors.contains(o.host)) {
        hostToExecutors(o.host) = new HashSet[String]()
      }
      if (!executorIdToRunningTaskIds.contains(o.executorId)) {
        hostToExecutors(o.host) += o.executorId
        executorAdded(o.executorId, o.host)
        executorIdToHost(o.executorId) = o.host
        executorIdToRunningTaskIds(o.executorId) = HashSet[Long]()
        newExecAvail = true
      }
    }
    
    // 随机打乱offers，避免总是在同一个Executor上分配任务
    val shuffledOffers = Random.shuffle(offers)
    
    // 为每个offer创建任务列表
    val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores / CPUS_PER_TASK))
    val availableCpus = shuffledOffers.map(o => o.cores).toArray
    
    // 获取排序后的TaskSet队列
    val sortedTaskSets = rootPool.getSortedTaskSetQueue
    
    // 按本地性级别分配任务
    for (taskSet <- sortedTaskSets) {
      var launchedAnyTask = false
      
      // 遍历所有本地性级别：PROCESS_LOCAL -> NODE_LOCAL -> NO_PREF -> RACK_LOCAL -> ANY
      for (currentMaxLocality <- taskSet.myLocalityLevels) {
        do {
          launchedAnyTask = resourceOfferSingleTaskSet(
            taskSet, currentMaxLocality, shuffledOffers, availableCpus, tasks)
        } while (launchedAnyTask)
      }
    }
    
    // 如果有新的Executor可用，检查是否有推测执行的任务
    if (newExecAvail) {
      for (taskSet <- sortedTaskSets) {
        taskSet.executorAdded()
      }
    }
    
    return tasks
  }
  
  // 3. 单个TaskSet的资源分配
  private def resourceOfferSingleTaskSet(
      taskSet: TaskSetManager,
      maxLocality: TaskLocality,
      shuffledOffers: Seq[WorkerOffer],
      availableCpus: Array[Int],
      tasks: IndexedSeq[ArrayBuffer[TaskDescription]]): Boolean = {
    
    var launchedTask = false
    
    // 遍历所有可用的Executor
    for (i <- 0 until shuffledOffers.length) {
      val execId = shuffledOffers(i).executorId
      val host = shuffledOffers(i).host
      
      // 检查是否有足够的CPU核心
      if (availableCpus(i) >= CPUS_PER_TASK) {
        try {
          // 尝试在当前Executor上启动任务
          for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {
            tasks(i) += task
            val tid = task.taskId
            taskIdToTaskSetManager(tid) = taskSet
            taskIdToExecutorId(tid) = execId
            executorIdToRunningTaskIds(execId) += tid
            availableCpus(i) -= CPUS_PER_TASK
            assert(availableCpus(i) >= 0)
            launchedTask = true
          }
        } catch {
          case e: TaskNotSerializableException =>
            logError(s"Resource offer failed, task set ${taskSet.name} was not serializable")
            taskSet.abort("TaskSet %s was not serializable".format(taskSet.name))
            return launchedTask
        }
      }
    }
    return launchedTask
  }
  
  // 4. 任务状态更新处理
  def statusUpdate(tid: Long, state: TaskState, serializedData: ByteBuffer): Unit = {
    var failedExecutor: Option[String] = None
    var reason: Option[TaskFailedReason] = None
    
    synchronized {
      try {
        taskIdToTaskSetManager.get(tid) match {
          case Some(taskSet) =>
            if (state == TaskState.LOST) {
              // 任务丢失，可能是Executor失败
              val execId = taskIdToExecutorId.getOrElse(tid, throw new IllegalStateException(
                "taskIdToTaskSetManager.contains(tid) <=> taskIdToExecutorId.contains(tid)"))
              if (executorIdToRunningTaskIds.contains(execId)) {
                reason = Some(ExecutorLostFailure(execId, exitCausedByApp = false,
                  Some("Task $tid was lost, so marking the executor as lost as well.")))
                removeExecutor(execId, reason.get)
                failedExecutor = Some(execId)
              }
            }
            
            if (TaskState.isFinished(state)) {
              cleanupTaskState(tid)
              taskSet.removeRunningTask(tid)
              if (state == TaskState.FINISHED) {
                taskResultGetter.enqueueSuccessfulTask(taskSet, tid, serializedData)
              } else if (Set(TaskState.FAILED, TaskState.KILLED, TaskState.LOST).contains(state)) {
                taskResultGetter.enqueueFailedTask(taskSet, tid, state, serializedData)
              }
            }
          case None =>
            logError(
              ("Ignoring update with state %s for TID %s because its task set is gone (this is " +
               "likely the result of receiving duplicate task finished status updates) or its " +
               "executor has been marked as failed.")
                .format(state, tid))
        }
      } catch {
        case e: Exception => logError("Exception in statusUpdate", e)
      }
    }
    
    // 如果有Executor失败，更新DAGScheduler
    if (failedExecutor.isDefined) {
      assert(reason.isDefined)
      dagScheduler.executorLost(failedExecutor.get, reason.get)
      backend.reviveOffers()
    }
  }
  
  // 5. 创建TaskSetManager
  private def createTaskSetManager(
      taskSet: TaskSet,
      maxTaskFailures: Int): TaskSetManager = {
    new TaskSetManager(this, taskSet, maxTaskFailures, blacklistTrackerOpt)
  }
  
  // 6. 清理任务状态
  private def cleanupTaskState(tid: Long): Unit = {
    taskIdToTaskSetManager.remove(tid)
    taskIdToExecutorId.remove(tid).foreach { execId =>
      executorIdToRunningTaskIds.get(execId).foreach { _.remove(tid) }
    }
  }
  
  // 7. 移除失败的Executor
  private def removeExecutor(executorId: String, reason: ExecutorLossReason): Unit = {
    executorIdToRunningTaskIds.remove(executorId).foreach { taskIds =>
      logDebug("Cleaning up TaskScheduler state for tasks " +
        s"${taskIds.mkString("[", ",", "]")} on failed executor $executorId")
      // 将运行在失败Executor上的任务标记为失败
      taskIds.foreach { tid =>
        val taskSetMgr = taskIdToTaskSetManager(tid)
        if (taskSetMgr != null) {
          taskSetMgr.executorLost(executorId, tid, reason)
        }
      }
    }
    
    val host = executorIdToHost(executorId)
    val execs = hostToExecutors.getOrElse(host, new HashSet)
    execs -= executorId
    if (execs.isEmpty) {
      hostToExecutors -= host
    }
    executorIdToHost -= executorId
    
    rootPool.executorLost(executorId, host, reason)
  }
}

// TaskSetManager.scala - 管理单个TaskSet的执行
private[spark] class TaskSetManager(
    sched: TaskSchedulerImpl,
    val taskSet: TaskSet,
    val maxTaskFailures: Int,
    blacklistTracker: Option[BlacklistTracker] = None)
  extends Schedulable with Logging {

  // TaskSet中的所有任务
  val tasks = taskSet.tasks
  val numTasks = tasks.length
  
  // 任务状态跟踪
  private val copiesRunning = new Array[Int](numTasks)
  private val successful = new Array[Boolean](numTasks)
  private val numFailures = new Array[Int](numTasks)
  
  // 本地性级别管理
  private val myLocalityLevels = computeValidLocalityLevels()
  private val localityWaits = myLocalityLevels.map(getLocalityWait)
  
  // 待调度任务队列（按本地性级别分组）
  private val pendingTasksForExecutor = new HashMap[String, ArrayBuffer[Int]]
  private val pendingTasksForHost = new HashMap[String, ArrayBuffer[Int]]
  private val pendingTasksForRack = new HashMap[String, ArrayBuffer[Int]]
  private val pendingTasksWithNoPrefs = new ArrayBuffer[Int]
  private val allPendingTasks = new ArrayBuffer[Int]
  
  // 1. 资源分配的核心方法
  def resourceOffer(
      execId: String,
      host: String,
      maxLocality: TaskLocality): Option[TaskDescription] = {
    
    val offerBlacklisted = blacklistTracker.exists(_.isExecutorBlacklisted(execId)) ||
      blacklistTracker.exists(_.isNodeBlacklisted(host))
    
    if (!isZombie && !offerBlacklisted) {
      val curTime = clock.getTimeMillis()
      
      var allowedLocality = maxLocality
      
      if (maxLocality != TaskLocality.NO_PREF) {
        allowedLocality = getAllowedLocalityLevel(curTime)
        if (allowedLocality > maxLocality) {
          // 如果允许的本地性级别比提供的级别更宽松，则使用提供的级别
          allowedLocality = maxLocality
        }
      }
      
      dequeueTask(execId, host, allowedLocality).map { case ((index, taskLocality, speculative)) =>
        // 找到了可以调度的任务
        val task = tasks(index)
        val taskId = sched.newTaskId()
        
        // 更新任务状态
        copiesRunning(index) += 1
        successful(index) = false
        
        val attemptNum = taskAttempts(index).size
        val info = new TaskInfo(taskId, index, attemptNum, curTime,
          execId, host, taskLocality, speculative)
        taskInfos(taskId) = info
        taskAttempts(index) = info :: taskAttempts(index)
        
        // 更新运行任务统计
        if (maxLocality == TaskLocality.NO_PREF) {
          stats.incNoPrefs(1)
        } else {
          stats.incLocality(taskLocality)
        }
        
        // 序列化任务
        val serializedTask: ByteBuffer = try {
          ser.serialize(task)
        } catch {
          case NonFatal(e) =>
            val msg = s"Failed to serialize task $taskId, not attempting to retry it."
            logError(msg, e)
            abort(s"$msg Exception during serialization: $e")
            throw new TaskNotSerializableException(e)
        }
        
        if (serializedTask.limit() > TaskSetManager.TASK_SIZE_TO_WARN_KB * 1024 &&
            !emittedTaskSizeWarning) {
          emittedTaskSizeWarning = true
          logWarning(s"Stage ${task.stageId} contains a task of very large size " +
            s"(${serializedTask.limit() / 1024} KB). The maximum recommended task size is " +
            s"${TaskSetManager.TASK_SIZE_TO_WARN_KB} KB.")
        }
        
        addRunningTask(taskId)
        
        // 创建TaskDescription
        new TaskDescription(
          taskId = taskId,
          attemptNumber = attemptNum,
          execId,
          task.name,
          index,
          task.partitionId,
          addedFiles,
          addedJars,
          task.localProperties,
          serializedTask)
      }
    } else {
      None
    }
  }
  
  // 2. 任务出队逻辑
  private def dequeueTask(execId: String, host: String, maxLocality: TaskLocality)
    : Option[(Int, TaskLocality, Boolean)] = {
    
    // 按本地性级别依次尝试获取任务
    for (locality <- Array(TaskLocality.PROCESS_LOCAL, TaskLocality.NODE_LOCAL,
                          TaskLocality.NO_PREF, TaskLocality.RACK_LOCAL, TaskLocality.ANY)) {
      if (locality <= maxLocality) {
        val taskSetIndex = locality match {
          case TaskLocality.PROCESS_LOCAL => dequeueTaskFromList(execId, host, 
            pendingTasksForExecutor.getOrElse(execId, ArrayBuffer()))
          case TaskLocality.NODE_LOCAL => dequeueTaskFromList(execId, host,
            pendingTasksForHost.getOrElse(host, ArrayBuffer()))
          case TaskLocality.NO_PREF => dequeueTaskFromList(execId, host, pendingTasksWithNoPrefs)
          case TaskLocality.RACK_LOCAL => dequeueTaskFromList(execId, host,
            pendingTasksForRack.getOrElse(sched.getRackForHost(host).orNull, ArrayBuffer()))
          case TaskLocality.ANY => dequeueTaskFromList(execId, host, allPendingTasks)
        }
        
        if (taskSetIndex.isDefined) {
          return Some((taskSetIndex.get, locality, false))
        }
      }
    }
    None
  }
  
  // 3. 从任务列表中出队
  private def dequeueTaskFromList(
      execId: String,
      host: String,
      list: ArrayBuffer[Int]): Option[Int] = {
    var indexOffset = list.size
    while (indexOffset > 0) {
      indexOffset -= 1
      val index = list(indexOffset)
      if (copiesRunning(index) == 0 && !successful(index)) {
        // 找到可运行的任务
        list.remove(indexOffset)
        if (pendingTasksForExecutor.contains(execId)) {
          pendingTasksForExecutor(execId) -= index
        }
        if (pendingTasksForHost.contains(host)) {
          pendingTasksForHost(host) -= index
        }
        val rack = sched.getRackForHost(host)
        if (rack.isDefined && pendingTasksForRack.contains(rack.get)) {
          pendingTasksForRack(rack.get) -= index
        }
        pendingTasksWithNoPrefs -= index
        allPendingTasks -= index
        return Some(index)
      }
    }
    None
  }
  
  // 4. 任务完成处理
  def handleSuccessfulTask(tid: Long, result: DirectTaskResult[_]): Unit = {
    val info = taskInfos(tid)
    val index = info.index
    
    // 标记任务成功
    successful(index) = true
    removalPendingTask(index)
    
    // 更新统计信息
    tasksSuccessful += 1
    logInfo(s"Finished task ${info.id} in stage ${taskSet.stageId} (TID $tid) in" +
      s" ${info.duration} ms on ${info.host} (executor ${info.executorId})" +
      s" ($tasksSuccessful/$numTasks)")
    
    // 检查TaskSet是否完成
    if (tasksSuccessful == numTasks) {
      isZombie = true
    }
    
    // 通知DAGScheduler任务完成
    sched.dagScheduler.taskEnded(tasks(index), Success, result.value(), result.accumUpdates, info)
  }
  
  // 5. 任务失败处理
  def handleFailedTask(tid: Long, state: TaskState, reason: TaskFailedReason): Unit = {
    val info = taskInfos(tid)
    if (info != null && !successful(info.index)) {
      val index = info.index
      copiesRunning(index) -= 1
      
      reason match {
        case fetchFailed: FetchFailed =>
          logWarning(s"Lost task ${info.id} in stage ${taskSet.stageId} (TID $tid, ${info.host}," +
            s" executor ${info.executorId}): ${reason.toErrorString}")
          if (!successful(index)) {
            successful(index) = true
            tasksSuccessful += 1
          }
          isZombie = true
          
        case ef: ExceptionFailure =>
          // 任务执行异常
          numFailures(index) += 1
          if (numFailures(index) >= maxTaskFailures) {
            logError(s"Task ${info.id} in stage ${taskSet.stageId} failed $maxTaskFailures times; aborting job")
            abort(s"Task $index in stage ${taskSet.stageId} failed $maxTaskFailures times, " +
              s"most recent failure: ${ef.description}")
            return
          } else {
            // 重新调度任务
            addPendingTask(index)
          }
          
        case TaskKilled(_) =>
          logWarning(s"Task ${info.id} in stage ${taskSet.stageId} was killed")
          
        case _ =>
          logWarning(s"Lost task ${info.id} in stage ${taskSet.stageId} (TID $tid) on " +
            s"executor ${info.executorId}: ${reason.toErrorString}")
          addPendingTask(index)
      }
    }
  }
  
  // 6. 计算本地性级别
  private def computeValidLocalityLevels(): Array[TaskLocality.TaskLocality] = {
    import TaskLocality._
    val levels = new ArrayBuffer[TaskLocality.TaskLocality]
    
    if (!pendingTasksForExecutor.isEmpty &&
        pendingTasksForExecutor.values.exists(_.nonEmpty)) {
      levels += PROCESS_LOCAL
    }
    if (!pendingTasksForHost.isEmpty &&
        pendingTasksForHost.values.exists(_.nonEmpty)) {
      levels += NODE_LOCAL
    }
    if (!pendingTasksWithNoPrefs.isEmpty) {
      levels += NO_PREF
    }
    if (!pendingTasksForRack.isEmpty &&
        pendingTasksForRack.values.exists(_.nonEmpty)) {
      levels += RACK_LOCAL
    }
    levels += ANY
    
    logDebug("Valid locality levels for " + taskSet + ": " + levels.mkString(", "))
    levels.toArray
  }
}

// CoarseGrainedSchedulerBackend.scala - 粗粒度调度后端
private[spark] class CoarseGrainedSchedulerBackend(
    scheduler: TaskSchedulerImpl,
    val rpcEnv: RpcEnv)
  extends ExecutorAllocationClient with SchedulerBackend with Logging {

  // Executor信息管理
  private val executorDataMap = new HashMap[String, ExecutorData]
  private val addressToExecutorId = new HashMap[RpcAddress, String]
  
  // 资源管理
  private val totalCoreCount = new AtomicInteger(0)
  private val totalRegisteredExecutors = new AtomicInteger(0)
  private val maxRpcMessageSize = RpcUtils.maxMessageSizeBytes(conf)
  
  // RPC端点引用
  var driverEndpoint: RpcEndpointRef = null
  
  // 1. 启动调度后端
  override def start(): Unit = {
    val properties = Seq[(String, String)](
      ("spark.scheduler.mode", scheduler.schedulingMode.toString),
      ("spark.starvation.timeout", starvationTimer.toString),
      ("spark.rpc.askTimeout", askTimeout.toString)
    ) ++ scheduler.applicationAttemptId().map(id => ("spark.app.attempt.id", id.toString))
    
    driverEndpoint = rpcEnv.setupEndpoint(ENDPOINT_NAME, createDriverEndpoint(properties))
  }
  
  // 2. 创建Driver端点
  protected def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = {
    new DriverEndpoint(rpcEnv, properties)
  }
  
  // 3. Driver端点实现
  class DriverEndpoint(override val rpcEnv: RpcEnv, sparkProperties: Seq[(String, String)])
    extends ThreadSafeRpcEndpoint with Logging {
    
    // 处理Executor注册
    override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
      case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls) =>
        if (executorDataMap.contains(executorId)) {
          executorRef.send(RegisterExecutorFailed("Duplicate executor ID: " + executorId))
          context.reply(true)
        } else if (scheduler.nodeBlacklist.contains(hostname)) {
          logInfo(s"Rejecting $executorId as it has been blacklisted.")
          executorRef.send(RegisterExecutorFailed(s"Executor is blacklisted: $executorId"))
          context.reply(true)
        } else {
          // 注册新的Executor
          val executorAddress = if (executorRef.address != null) {
            executorRef.address
          } else {
            context.senderAddress
          }
          
          logInfo(s"Registered executor $executorRef ($executorAddress) with ID $executorId")
          addressToExecutorId(executorAddress) = executorId
          totalCoreCount.addAndGet(cores)
          totalRegisteredExecutors.addAndGet(1)
          
          val data = new ExecutorData(executorRef, executorAddress, hostname,
            cores, cores, logUrls)
          
          // 将Executor信息存储到映射中
          CoarseGrainedSchedulerBackend.this.synchronized {
            executorDataMap.put(executorId, data)
            if (currentExecutorIdCounter < executorId.toInt) {
              currentExecutorIdCounter = executorId.toInt
            }
            if (numPendingExecutors > 0) {
              numPendingExecutors -= 1
              logDebug(s"Decremented number of pending executors ($numPendingExecutors left)")
            }
          }
          
          executorRef.send(RegisteredExecutor)
          context.reply(true)
          listenerBus.post(
            SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data))
          makeOffers()
        }
        
      case StopDriver =>
        context.reply(true)
        stop()
        
      case StopExecutors =>
        logInfo("Asking each executor to shut down")
        for ((_, executorData) <- executorDataMap) {
          executorData.executorEndpoint.send(StopExecutor)
        }
        context.reply(true)
    }
    
    // 处理Executor状态更新
    override def receive: PartialFunction[Any, Unit] = {
      case StatusUpdate(executorId, taskId, state, data) =>
        scheduler.statusUpdate(taskId, state, data.value)
        if (TaskState.isFinished(state)) {
          executorDataMap.get(executorId) match {
            case Some(executorInfo) =>
              executorInfo.freeCores += scheduler.CPUS_PER_TASK
              makeOffers(executorId)
            case None =>
              logWarning(s"Ignored task status update ($taskId state $state) " +
                s"from unknown executor with ID $executorId")
          }
        }
        
      case ReviveOffers =>
        makeOffers()
        
      case KillTask(taskId, executorId, interruptThread, reason) =>
        executorDataMap.get(executorId) match {
          case Some(executorInfo) =>
            executorInfo.executorEndpoint.send(
              KillTask(taskId, executorId, interruptThread, reason))
          case None =>
            logWarning(s"Attempted to kill task $taskId for unknown executor $executorId.")
        }
    }
    
    // 4. 资源分配核心方法
    private def makeOffers(): Unit = {
      // 过滤出活跃的Executor
      val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
      val workOffers = activeExecutors.map {
        case (id, executorData) =>
          new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
      }.toIndexedSeq
      
      launchTasks(scheduler.resourceOffers(workOffers))
    }
    
    private def makeOffers(executorId: String): Unit = {
      // 为特定Executor创建资源offer
      if (executorIsAlive(executorId)) {
        val executorData = executorDataMap(executorId)
        val workOffers = IndexedSeq(
          new WorkerOffer(executorId, executorData.executorHost, executorData.freeCores))
        launchTasks(scheduler.resourceOffers(workOffers))
      }
    }
    
    // 5. 启动任务
    private def launchTasks(tasks: Seq[Seq[TaskDescription]]): Unit = {
      for (i <- tasks.indices) {
        val execId = tasks(i).head.executorId
        val executorData = executorDataMap(execId)
        val executorOffers = tasks(i)
        
        if (executorOffers.nonEmpty) {
          // 减少可用核心数
          executorData.freeCores -= executorOffers.size * scheduler.CPUS_PER_TASK
          
          logDebug(s"Launching ${executorOffers.size} tasks on executor $execId")
          
          // 序列化任务并发送给Executor
          val serializedTasks = executorOffers.map { task =>
            ser.serialize(task)
          }
          
          if (serializedTasks.nonEmpty) {
            executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(
              Utils.serialize(serializedTasks))))
          }
        }
      }
    }
  }
  
  // 6. 停止调度后端
  override def stop(): Unit = {
    reviveThread.shutdownNow()
    try {
      if (driverEndpoint != null) {
        driverEndpoint.askSync[Boolean](StopDriver)
      }
    } catch {
      case e: Exception =>
        logWarning("Exception during shutdown", e)
    }
  }
}

// DAGSchedulerEventProcessLoop事件处理
private def doOnReceive(event: DAGSchedulerEvent): Unit = event match {
  case TaskFailed(taskId, taskType, reason, exception) =>
    reason match {
      case _: FetchFailed =>
        // Shuffle数据获取失败，需要重新计算父Stage
        val shuffleMapStage = shuffleIdToMapStage(reason.shuffleId)
        markStageAsFinished(shuffleMapStage, Some(reason.toString))
        submitStage(shuffleMapStage)
        
      case _: ExecutorLostFailure =>
        // Executor丢失，需要重新调度Task
        removeExecutorAndUnregisterOutputs(reason.execId, filesLost = true)
        
      case _: TaskKilled =>
        // Task被杀死，通常是推测执行
        logInfo(s"Task $taskId was killed")
        
      case _ =>
        // 其他异常，Task级别重试
        if (task.attempt < maxTaskFailures) {
          taskScheduler.submitTasks(createTaskSet(Array(task)))
        } else {
          abortStage(currentStage, s"Task $taskId failed $maxTaskFailures times")
        }
    }
    
  case StageCompleted(stage) =>
    // Stage完成，检查并提交依赖的Stage
    markStageAsFinished(stage)
    submitWaitingChildStages(stage)
}

// Executor.launchTask()
def launchTask(context: ExecutorBackend, taskDescription: TaskDescription): Unit = {
  val tr = new TaskRunner(context, taskDescription)
  runningTasks.put(taskDescription.taskId, tr)
  threadPool.execute(tr)
}

// TaskRunner.run()
class TaskRunner(
    execBackend: ExecutorBackend,
    private val taskDescription: TaskDescription)
  extends Runnable {
  
  override def run(): Unit = {
    try {
      // 1. 反序列化Task
      val task = ser.deserialize[Task[Any]](
        taskDescription.serializedTask, 
        Thread.currentThread.getContextClassLoader)
      
      // 2. 设置TaskContext
      val taskContext = new TaskContextImpl(
        stageId = taskDescription.stageId,
        taskAttemptId = taskDescription.taskId,
        attemptNumber = taskDescription.attemptNumber,
        partitionId = task.partitionId,
        localProperties = taskDescription.properties,
        taskMemoryManager = taskMemoryManager,
        metricsSystem = env.metricsSystem)
      
      // 3. 执行Task
      val value = task.run(
        taskAttemptId = taskDescription.taskId,
        attemptNumber = taskDescription.attemptNumber,
        metricsSystem = env.metricsSystem)
      
      // 4. 序列化结果并返回
      val serializedResult = ser.serialize(value)
      execBackend.statusUpdate(
        taskDescription.taskId, 
        TaskState.FINISHED, 
        serializedResult)
        
    } catch {
      case e: Exception =>
        // 异常处理
        val reason = new ExceptionFailure(e, taskContext.taskMetrics())
        execBackend.statusUpdate(
          taskDescription.taskId, 
          TaskState.FAILED, 
          ser.serialize(TaskFailedReason(reason)))
    } finally {
      // 清理资源
      runningTasks.remove(taskDescription.taskId)
    }
  }
}

// RDD.iterator() 递归调用流程
final def iterator(split: Partition, context: TaskContext): Iterator[T] = {
  if (storageLevel != StorageLevel.NONE) {
    // 1. 尝试从缓存读取
    getOrCompute(split, context)
  } else {
    // 2. 直接计算
    computeOrReadCheckpoint(split, context)
  }
}

def computeOrReadCheckpoint(split: Partition, context: TaskContext): Iterator[T] = {
  if (isCheckpointed) {
    // 从Checkpoint读取
    firstParent[T].iterator(split, context)
  } else {
    // 调用具体RDD的compute方法
    compute(split, context)
  }
}

// 以MapPartitionsRDD为例
override def compute(split: Partition, context: TaskContext): Iterator[U] = {
  // 递归调用父RDD的iterator
  f(context, split.index, firstParent[T].iterator(split, context))
}

// ResultTask - 产生最终结果
class ResultTask[T, U](
    stageId: Int,
    rdd: RDD[T],
    func: (TaskContext, Iterator[T]) => U,
    _partitionId: Int,
    locs: Seq[TaskLocation])
  extends Task[U](stageId, _partitionId) {
  
  override def runTask(context: TaskContext): U = {
    // 直接调用用户函数处理数据
    func(context, rdd.iterator(partition, context))
  }
}

// ShuffleMapTask - 产生中间Shuffle数据
class ShuffleMapTask(
    stageId: Int,
    rdd: RDD[_],
    dep: ShuffleDependency[_, _, _],
    _partitionId: Int,
    locs: Seq[TaskLocation])
  extends Task[MapStatus](stageId, _partitionId) {
  
  override def runTask(context: TaskContext): MapStatus = {
    // 获取ShuffleWriter
    val manager = SparkEnv.get.shuffleManager
    val writer = manager.getWriter[Any, Any](
      dep.shuffleHandle, partitionId, context)
    
    try {
      // 写入Shuffle数据
      writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
      writer.stop(success = true).get
    } catch {
      case e: Exception =>
        writer.stop(success = false)
        throw e
    }
  }
}

// UnifiedMemoryManager.scala - 统一内存管理器
class UnifiedMemoryManager(
    conf: SparkConf,
    val maxHeapMemory: Long,
    onHeapStorageRegionSize: Long,
    numCores: Int)
  extends MemoryManager(conf, numCores, onHeapStorageRegionSize, maxHeapMemory) {

  // 系统保留内存（300MB）
  private val reservedMemory = conf.getLong("spark.testing.reservedMemory", 300 * 1024 * 1024)
  
  // 可用于Spark的总内存
  private val maxPoolSize = maxHeapMemory - reservedMemory
  
  // 内存分配给Spark的比例
  private val memoryFraction = conf.getDouble("spark.sql.adaptive.memoryFraction", 0.6)
  private val poolSize = (maxPoolSize * memoryFraction).toLong
  
  // 存储内存的初始比例
  private val storageFraction = conf.getDouble("spark.sql.adaptive.storageFraction", 0.5)
  private val initialStorageMemory = (poolSize * storageFraction).toLong
  
  // 创建内存池
  override val onHeapStorageMemoryPool = new StorageMemoryPool(this, MemoryMode.ON_HEAP)
  override val offHeapStorageMemoryPool = new StorageMemoryPool(this, MemoryMode.OFF_HEAP)
  override val onHeapExecutionMemoryPool = new ExecutionMemoryPool(this, MemoryMode.ON_HEAP)
  override val offHeapExecutionMemoryPool = new ExecutionMemoryPool(this, MemoryMode.OFF_HEAP)

  // 设置初始内存池大小
  onHeapStorageMemoryPool.incrementPoolSize(initialStorageMemory)
  onHeapExecutionMemoryPool.incrementPoolSize(poolSize - initialStorageMemory)

  // 获取存储内存
  override def acquireStorageMemory(
      blockId: BlockId,
      numBytes: Long,
      memoryMode: MemoryMode): Boolean = synchronized {
    
    assertInvariants()
    assert(numBytes >= 0)
    
    val (executionPool, storagePool, maxMemory) = memoryMode match {
      case MemoryMode.ON_HEAP => (
        onHeapExecutionMemoryPool,
        onHeapStorageMemoryPool,
        maxOnHeapStorageMemory)
      case MemoryMode.OFF_HEAP => (
        offHeapExecutionMemoryPool,
        offHeapStorageMemoryPool,
        maxOffHeapStorageMemory)
    }
    
    // 检查请求的内存是否超过最大限制
    if (numBytes > maxMemory) {
      logInfo(s"Will not store $blockId as the required space ($numBytes bytes) exceeds our " +
        s"memory limit ($maxMemory bytes)")
      return false
    }
    
    // 如果存储池内存不足，尝试从执行池借用
    if (numBytes > storagePool.memoryFree) {
      val memoryBorrowedFromExecution = Math.min(
        executionPool.memoryFree, 
        numBytes - storagePool.memoryFree)
      if (memoryBorrowedFromExecution > 0) {
        executionPool.decrementPoolSize(memoryBorrowedFromExecution)
        storagePool.incrementPoolSize(memoryBorrowedFromExecution)
        logInfo(s"Borrowed $memoryBorrowedFromExecution bytes from execution memory " +
          s"for storage of $blockId")
      }
    }
    
    // 在存储池中分配内存
    val success = storagePool.acquireMemory(blockId, numBytes)
    if (success) {
      logDebug(s"Successfully acquired $numBytes bytes for $blockId")
    } else {
      logDebug(s"Failed to acquire $numBytes bytes for $blockId")
    }
    
    assertInvariants()
    success
  }

  // 获取执行内存
  override def acquireExecutionMemory(
      numBytes: Long,
      taskAttemptId: Long,
      memoryMode: MemoryMode): Long = synchronized {
    
    assertInvariants()
    assert(numBytes >= 0)
    
    val (executionPool, storagePool, maxPoolSize, maybeGrowPool, maxMemory) = memoryMode match {
      case MemoryMode.ON_HEAP => (
        onHeapExecutionMemoryPool,
        onHeapStorageMemoryPool,
        maxOnHeapStorageMemory,
        () => expandOnHeapPool(numBytes),
        maxHeapMemory)
      case MemoryMode.OFF_HEAP => (
        offHeapExecutionMemoryPool,
        offHeapStorageMemoryPool,
        maxOffHeapStorageMemory,
        () => expandOffHeapPool(numBytes),
        maxOffHeapMemory)
    }

    /**
     * 执行内存分配策略：
     * 1. 首先尝试从执行池直接分配
     * 2. 如果执行池内存不足，尝试从存储池回收内存
     * 3. 如果存储池可以回收内存，将其转移到执行池
     * 4. 如果还是不足，考虑扩展内存池
     */

    // 1. 首先尝试从执行池直接分配
    val toGrant = Math.min(numBytes, executionPool.memoryFree)
    var memoryAcquired = 0L
    
    if (toGrant > 0) {
      memoryAcquired = executionPool.acquireMemory(toGrant, taskAttemptId)
    }
    
    // 2. 如果直接分配不够，尝试从存储池回收内存
    if (memoryAcquired < numBytes) {
      val memoryToReclaim = numBytes - memoryAcquired
      val memoryReclaimedFromStorage = storagePool.freeSpaceToShrinkPool(memoryToReclaim)
      
      if (memoryReclaimedFromStorage > 0) {
        // 将回收的内存转移到执行池
        storagePool.decrementPoolSize(memoryReclaimedFromStorage)
        executionPool.incrementPoolSize(memoryReclaimedFromStorage)
        
        logInfo(s"Reclaimed $memoryReclaimedFromStorage bytes from storage memory " +
          s"for execution task $taskAttemptId")
        
        // 尝试在执行池中分配剩余的内存
        val additionalMemory = Math.min(
          memoryToReclaim, 
          executionPool.memoryFree)
        if (additionalMemory > 0) {
          memoryAcquired += executionPool.acquireMemory(additionalMemory, taskAttemptId)
        }
      }
    }
    
    // 3. 如果还是不足，考虑扩展内存池
    if (memoryAcquired < numBytes) {
      val remainingNeeded = numBytes - memoryAcquired
      try {
        maybeGrowPool()
        val additionalMemory = Math.min(remainingNeeded, executionPool.memoryFree)
        if (additionalMemory > 0) {
          memoryAcquired += executionPool.acquireMemory(additionalMemory, taskAttemptId)
        }
      } catch {
        case _: Exception =>
          logWarning(s"Failed to expand memory pool for task $taskAttemptId")
      }
    }
    
    logDebug(s"Task $taskAttemptId acquired $memoryAcquired / $numBytes bytes of execution memory")
    
    assertInvariants()
    memoryAcquired
  }

  // 释放存储内存
  override def releaseStorageMemory(numBytes: Long, memoryMode: MemoryMode): Unit = synchronized {
    val storagePool = memoryMode match {
      case MemoryMode.ON_HEAP => onHeapStorageMemoryPool
      case MemoryMode.OFF_HEAP => offHeapStorageMemoryPool
    }
    storagePool.releaseMemory(numBytes)
    logDebug(s"Released $numBytes bytes of storage memory")
    assertInvariants()
  }

  // 释放执行内存
  override def releaseExecutionMemory(
      numBytes: Long,
      taskAttemptId: Long,
      memoryMode: MemoryMode): Unit = synchronized {
    val executionPool = memoryMode match {
      case MemoryMode.ON_HEAP => onHeapExecutionMemoryPool
      case MemoryMode.OFF_HEAP => offHeapExecutionMemoryPool
    }
    executionPool.releaseMemory(numBytes, taskAttemptId)
    logDebug(s"Task $taskAttemptId released $numBytes bytes of execution memory")
    assertInvariants()
  }

  // 扩展堆内内存池
  private def expandOnHeapPool(requestedMemory: Long): Unit = {
    val available = maxHeapMemory - onHeapExecutionMemoryPool.poolSize - onHeapStorageMemoryPool.poolSize
    val toExpand = Math.min(available, requestedMemory)
    if (toExpand > 0) {
      onHeapExecutionMemoryPool.incrementPoolSize(toExpand)
      logInfo(s"Expanded on-heap execution pool by ${Utils.bytesToString(toExpand)}")
    }
  }

  // 扩展堆外内存池  
  private def expandOffHeapPool(requestedMemory: Long): Unit = {
    val available = maxOffHeapMemory - offHeapExecutionMemoryPool.poolSize - offHeapStorageMemoryPool.poolSize
    val toExpand = Math.min(available, requestedMemory)
    if (toExpand > 0) {
      offHeapExecutionMemoryPool.incrementPoolSize(toExpand)
      logInfo(s"Expanded off-heap execution pool by ${Utils.bytesToString(toExpand)}")
    }
  }

  // 验证内存池状态的一致性
  private def assertInvariants(): Unit = {
    if (conf.getBoolean("spark.testing.memory.enableChecks", false)) {
      val totalOnHeap = onHeapExecutionMemoryPool.poolSize + onHeapStorageMemoryPool.poolSize
      val totalOffHeap = offHeapExecutionMemoryPool.poolSize + offHeapStorageMemoryPool.poolSize
      
      assert(totalOnHeap <= maxHeapMemory, 
        s"Total on-heap memory $totalOnHeap exceeds maximum $maxHeapMemory")
      assert(totalOffHeap <= maxOffHeapMemory, 
        s"Total off-heap memory $totalOffHeap exceeds maximum $maxOffHeapMemory")
      
      assert(onHeapStorageMemoryPool.memoryUsed <= onHeapStorageMemoryPool.poolSize,
        "Storage memory pool usage exceeds pool size")
      assert(onHeapExecutionMemoryPool.memoryUsed <= onHeapExecutionMemoryPool.poolSize,
        "Execution memory pool usage exceeds pool size")
    }
  }

  // 获取内存状态信息
  def getMemoryStatus: (Long, Long, Long, Long) = synchronized {
    (
      onHeapExecutionMemoryPool.memoryUsed,
      onHeapStorageMemoryPool.memoryUsed,
      offHeapExecutionMemoryPool.memoryUsed,
      offHeapStorageMemoryPool.memoryUsed
    )
  }

  // 显示内存使用详情
  override def toString: String = {
    synchronized {
      s"""UnifiedMemoryManager:
         |  On-heap Execution: ${Utils.bytesToString(onHeapExecutionMemoryPool.memoryUsed)} / ${Utils.bytesToString(onHeapExecutionMemoryPool.poolSize)}
         |  On-heap Storage: ${Utils.bytesToString(onHeapStorageMemoryPool.memoryUsed)} / ${Utils.bytesToString(onHeapStorageMemoryPool.poolSize)}
         |  Off-heap Execution: ${Utils.bytesToString(offHeapExecutionMemoryPool.memoryUsed)} / ${Utils.bytesToString(offHeapExecutionMemoryPool.poolSize)}
         |  Off-heap Storage: ${Utils.bytesToString(offHeapStorageMemoryPool.memoryUsed)} / ${Utils.bytesToString(offHeapStorageMemoryPool.poolSize)}
         |""".stripMargin
    }
  }
}