Trino源码学习-TASK执行

//io.trino.execution.scheduler.PipelinedStageExecution
public synchronized Optional<RemoteTask> scheduleTask(
            InternalNode node,
            int partition,
            Multimap<PlanNodeId, Split> initialSplits)
{
    if (stateMachine.getState().isDone()) {
        return Optional.empty();
    }

    checkArgument(!tasks.containsKey(partition), "A task for partition %s already exists", partition);

    OutputBuffers outputBuffers = outputBufferManagers.get(stage.getFragment().getId()).getOutputBuffers();
    // 此处创建了Task，TaskID结构:stageId + "." + partitionId + "." + attemptId
    Optional<RemoteTask> optionalTask = stage.createTask(
            node,
            partition,
            attempt,
            bucketToPartition,
            outputBuffers,
            initialSplits,
            ImmutableSet.of(),
            Optional.empty());

    if (optionalTask.isEmpty()) {
        return Optional.empty();
    }

    RemoteTask task = optionalTask.get();

    tasks.put(partition, task);

    ImmutableMultimap.Builder<PlanNodeId, Split> exchangeSplits = ImmutableMultimap.builder();
    // sourceTasks中存放了上游Stage的SourceTask,即当前Stage的输入
    // 在DistributedStagesScheduler#create方法中
    // parentStageExecution会将自己提供的listener注册到Child上，以获取Child的Task作为SourceTask。
    sourceTasks.forEach((fragmentId, sourceTask) -> {
        TaskStatus status = sourceTask.getTaskStatus();
        if (status.getState() != TaskState.FINISHED) {
            // 在DistributedStagesScheduler#create方法中
            // exchangeSources 存放了当前Stage的所有remoteSourceFragmentId
            PlanNodeId planNodeId = exchangeSources.get(fragmentId).getId();
            exchangeSplits.put(planNodeId, createExchangeSplit(sourceTask, task));
        }
    });
    //记录创建的TASK
    allTasks.add(task.getTaskId());
    // task尝试添加Source Split
    task.addSplits(exchangeSplits.build());
    // 通知task完成的Source
    completeSources.forEach(task::noMoreSplits);
    // 添加状态变更listener
    task.addStateChangeListener(this::updateTaskStatus);
    // 启动task(此处会在worker上发起创建Task请求)
    task.start();
    // 通知listener(parent stage) task创建信息
    taskLifecycleListener.taskCreated(stage.getFragment().getId(), task);

    // update output buffers
    OutputBufferId outputBufferId = new OutputBufferId(task.getTaskId().getPartitionId());
    updateSourceTasksOutputBuffers(outputBufferManager -> outputBufferManager.addOutputBuffer(outputBufferId));
    return Optional.of(task);
}

//io.trino.server.remotetask.HttpRemoteTask
public HttpRemoteTask(... ...)
{
    // require param

    try (SetThreadName ignored = new SetThreadName("HttpRemoteTask-%s", taskId)) {
        ... ...
        // 将所有的initialSplits构建为ScheduledSplit，放入pendingSplits。
        for (Entry<PlanNodeId, Split> entry : initialSplits.entries()) {
            ScheduledSplit scheduledSplit = new ScheduledSplit(nextSplitId.getAndIncrement(), entry.getKey(), entry.getValue());
            pendingSplits.put(entry.getKey(), scheduledSplit);
        }
        ... ...
        int pendingSourceSplitCount = 0;
        long pendingSourceSplitsWeight = 0;
        // getPartitionedSources中存放着planFragment的TableScanNode Id
        for (PlanNodeId planNodeId : planFragment.getPartitionedSources()) {
            Collection<Split> tableScanSplits = initialSplits.get(planNodeId);
            if (!tableScanSplits.isEmpty()) {
                // 记录Split处理的数量
                pendingSourceSplitCount += tableScanSplits.size();
                pendingSourceSplitsWeight = addExact(pendingSourceSplitsWeight, SplitWeight.rawValueSum(tableScanSplits, Split::getSplitWeight));
            }
        }
        this.pendingSourceSplitCount = pendingSourceSplitCount;
        this.pendingSourceSplitsWeight = pendingSourceSplitsWeight;

        ... ...
        // 创建TaskInfo保存Task的内部信息
        TaskInfo initialTask = createInitialTask(taskId, location, nodeId, pipelinedBufferStates, new TaskStats(DateTime.now(), null));
        // 持续获取task状态
        this.taskStatusFetcher = new ContinuousTaskStatusFetcher();
        // 持续获取task信息
        this.taskInfoFetcher = new TaskInfoFetcher();
        ... ...
    }
}

public void start()
{
    try (SetThreadName ignored = new SetThreadName("HttpRemoteTask-%s", taskId)) {
        // to start we just need to trigger an update
        started.set(true);
        triggerUpdate(); // 触发创建或更新

        dynamicFiltersFetcher.start();
        taskStatusFetcher.start();
        taskInfoFetcher.start();
    }
}
private void triggerUpdate()
{
    if (!started.get()) {
        // task has not started yet
        return;
    }
    if (pendingRequestsCounter.getAndIncrement() == 0) {
        // schedule update if this is the first update requested
        scheduleUpdate();
    }
}
private void scheduleUpdate()
{
    executor.execute(this::sendUpdate);
}
private void sendUpdate()
{
    ... ...
    HttpUriBuilder uriBuilder = getHttpUriBuilder(taskStatus);
    Request request = preparePost()
            .setUri(uriBuilder.build())
            .setHeader(HttpHeaders.CONTENT_TYPE, MediaType.JSON_UTF_8.toString())
            .setBodyGenerator(createStaticBodyGenerator(taskUpdateRequestJson))
            .build();
    ListenableFuture<JsonResponse<TaskInfo>> future = httpClient.executeAsync(request, 
    Futures.addCallback(
            future,
            new SimpleHttpResponseHandler<>(new UpdateResponseHandler(splitAssignments, dynamicFilterDomains.getVersion(), System.nanoTime(), currentPendingRequestsCounter), request.getUri(), stats),
            executor);
}

//io.trino.server.TaskResource
@Path("/v1/task")
public class TaskResource
{
  @ResourceSecurity(INTERNAL_ONLY)
  @POST
  @Path("{taskId}")
  @Consumes(MediaType.APPLICATION_JSON)
  @Produces(MediaType.APPLICATION_JSON)
  public void createOrUpdateTask(
          @PathParam("taskId") TaskId taskId,
          TaskUpdateRequest taskUpdateRequest,
          @Context UriInfo uriInfo,
          @Suspended AsyncResponse asyncResponse)
  {
      requireNonNull(taskUpdateRequest, "taskUpdateRequest is null");

      Session session = taskUpdateRequest.getSession().toSession(sessionPropertyManager, taskUpdateRequest.getExtraCredentials(), taskUpdateRequest.getExchangeEncryptionKey());

      if (injectFailure(session.getTraceToken(), taskId, RequestType.CREATE_OR_UPDATE_TASK, asyncResponse)) {
          return;
      }
      // 使用Taskmanager创建/更新Task，并返回task信息
      TaskInfo taskInfo = taskManager.updateTask(session,
              taskId,
              taskUpdateRequest.getFragment(),
              taskUpdateRequest.getSplitAssignments(),
              taskUpdateRequest.getOutputIds(),
              taskUpdateRequest.getDynamicFilterDomains());

      if (shouldSummarize(uriInfo)) {
          taskInfo = taskInfo.summarize();
      }

      asyncResponse.resume(Response.ok().entity(taskInfo).build());
  }
}

//io.trino.execution.SqlTaskExecution
public SqlTaskExecution(
            TaskStateMachine taskStateMachine,
            TaskContext taskContext,
            OutputBuffer outputBuffer,
            LocalExecutionPlan localExecutionPlan,
            TaskExecutor taskExecutor,
            SplitMonitor splitMonitor,
            Executor notificationExecutor)
{
    ... ...

    try (SetThreadName ignored = new SetThreadName("Task-%s", taskId)) {
        // 根据是否是本地Source节点，将driverFactory放到不同的管理器中
        // index driver factories
        for (DriverFactory driverFactory : localExecutionPlan.getDriverFactories()) {
            Optional<PlanNodeId> sourceId = driverFactory.getSourceId();
            if (sourceId.isPresent() && partitionedSources.contains(sourceId.get())) {
                driverRunnerFactoriesWithSplitLifeCycle.put(sourceId.get(), new DriverSplitRunnerFactory(driverFactory, true));
            }
            else {
                //DriverSplitRunnerFactory中会初始化PipelineContext
                DriverSplitRunnerFactory runnerFactory = new DriverSplitRunnerFactory(driverFactory, false);
                sourceId.ifPresent(planNodeId -> driverRunnerFactoriesWithRemoteSource.put(planNodeId, runnerFactory));
                driverRunnerFactoriesWithTaskLifeCycle.add(runnerFactory);
            }
        }
        ... ...
        //  向TaskExecutor中添加Task
        if (!taskStateMachine.getState().isDone()) {
                taskHandle = createTaskHandle(taskStateMachine, taskContext, outputBuffer, localExecutionPlan, taskExecutor);
            }
        else {
                taskHandle = null;
        }
        ... ...
    }
}
//io.trino.execution.SqlTaskExecution.TaskHandle
private static TaskHandle createTaskHandle(
            TaskStateMachine taskStateMachine,
            TaskContext taskContext,
            OutputBuffer outputBuffer,
            LocalExecutionPlan localExecutionPlan,
            TaskExecutor taskExecutor)
{
    TaskHandle taskHandle = taskExecutor.addTask(
            taskStateMachine.getTaskId(),
            outputBuffer::getUtilization,
            getInitialSplitsPerNode(taskContext.getSession()),
            getSplitConcurrencyAdjustmentInterval(taskContext.getSession()),
            getMaxDriversPerTask(taskContext.getSession()));
    taskStateMachine.addStateChangeListener(state -> {
        if (state.isDone()) {
            taskExecutor.removeTask(taskHandle);
            for (DriverFactory factory : localExecutionPlan.getDriverFactories()) {
                factory.noMoreDrivers();
            }
        }
    });
    return taskHandle;
}

//triggerUpdate -> sendUpdate
// io.trino.server.remotetask.HttpRemoteTask#sendUpdate
List<SplitAssignment> splitAssignments = getSplitAssignments();
VersionedDynamicFilterDomains dynamicFilterDomains = outboundDynamicFiltersCollector.acknowledgeAndGetNewDomains(sentDynamicFiltersVersion.get());

// Workers don't need the embedded JSON representation when the fragment is sent
Optional<PlanFragment> fragment = sendPlan.get() ? Optional.of(planFragment.withoutEmbeddedJsonRepresentation()) : Optional.empty();
TaskUpdateRequest updateRequest = new TaskUpdateRequest(
        session.toSessionRepresentation(),
        session.getIdentity().getExtraCredentials(),
        fragment,
        splitAssignments,
        outputBuffers.get(),
        dynamicFilterDomains.getDynamicFilterDomains(),
        session.getExchangeEncryptionKey());

private synchronized List<SplitAssignment> getSplitAssignments()
{
    return Stream.concat(planFragment.getPartitionedSourceNodes().stream(), planFragment.getRemoteSourceNodes().stream())
            .filter(Objects::nonNull)
            .map(PlanNode::getId)
            .map(this::getSplitAssignment)
            .filter(Objects::nonNull)
            .collect(toImmutableList());
}

private synchronized SplitAssignment getSplitAssignment(PlanNodeId planNodeId)
{   // addSplits会将Split记录到pendingSplits中
    Set<ScheduledSplit> splits = pendingSplits.get(planNodeId);
    boolean pendingNoMoreSplits = Boolean.TRUE.equals(this.noMoreSplits.get(planNodeId));
    boolean noMoreSplits = this.noMoreSplits.containsKey(planNodeId);

    SplitAssignment assignment = null;
    if (!splits.isEmpty() || pendingNoMoreSplits) {
        assignment = new SplitAssignment(planNodeId, splits, noMoreSplits);
    }
    return assignment;
}

public static class LocalExecutionPlan
{
    private final List<DriverFactory> driverFactories;
    private final List<PlanNodeId> partitionedSourceOrder; // stage中TableScan节点ID

    public LocalExecutionPlan(List<DriverFactory> driverFactories, List<PlanNodeId> partitionedSourceOrder)
    {
        this.driverFactories = ImmutableList.copyOf(requireNonNull(driverFactories, "driverFactories is null"));
        this.partitionedSourceOrder = ImmutableList.copyOf(requireNonNull(partitionedSourceOrder, "partitionedSourceOrder is null"));
    }

    public List<DriverFactory> getDriverFactories()
    {
        return driverFactories;
    }

    public List<PlanNodeId> getPartitionedSourceOrder()
    {
        return partitionedSourceOrder;
    }
}

public LocalExecutionPlan plan(
            TaskContext taskContext,
            PlanNode plan,
            TypeProvider types,
            PartitioningScheme partitioningScheme,
            List<PlanNodeId> partitionedSourceOrder,
            OutputBuffer outputBuffer)
{
    List<Symbol> outputLayout = partitioningScheme.getOutputLayout();

    if (partitioningScheme.getPartitioning().getHandle().equals(FIXED_BROADCAST_DISTRIBUTION) ||
            partitioningScheme.getPartitioning().getHandle().equals(FIXED_ARBITRARY_DISTRIBUTION) ||
            partitioningScheme.getPartitioning().getHandle().equals(SCALED_WRITER_ROUND_ROBIN_DISTRIBUTION) ||
            partitioningScheme.getPartitioning().getHandle().equals(SINGLE_DISTRIBUTION) ||
            partitioningScheme.getPartitioning().getHandle().equals(COORDINATOR_DISTRIBUTION)) {
        return plan(taskContext, plan, outputLayout, types, partitionedSourceOrder, new TaskOutputFactory(outputBuffer));
    }

    // We can convert the symbols directly into channels, because the root must be a sink and therefore the layout is fixed
    ... ...

    return plan(
            taskContext,
            plan,
            outputLayout,
            types,
            partitionedSourceOrder,
            new PartitionedOutputFactory(
                    partitionFunction,
                    partitionChannels,
                    partitionConstants,
                    partitioningScheme.isReplicateNullsAndAny(),
                    nullChannel,
                    outputBuffer,
                    maxPagePartitioningBufferSize,
                    positionsAppenderFactory));
}

public LocalExecutionPlan plan(
            TaskContext taskContext,
            PlanNode plan,
            List<Symbol> outputLayout,
            TypeProvider types,
            List<PlanNodeId> partitionedSourceOrder,
            OutputFactory outputOperatorFactory)
{
    Session session = taskContext.getSession();
    LocalExecutionPlanContext context = new LocalExecutionPlanContext(taskContext, types);

    PhysicalOperation physicalOperation = plan.accept(new Visitor(session), context);

    Function<Page, Page> pagePreprocessor = enforceLoadedLayoutProcessor(outputLayout, physicalOperation.getLayout());

    List<Type> outputTypes = outputLayout.stream()
            .map(types::get)
            .collect(toImmutableList());

    context.addDriverFactory(
            context.isInputDriver(),
            true,
            new PhysicalOperation(
                    outputOperatorFactory.createOutputOperator(
                            context.getNextOperatorId(),
                            plan.getId(),
                            outputTypes,
                            pagePreprocessor,
                            new PagesSerdeFactory(plannerContext.getBlockEncodingSerde(), isExchangeCompressionEnabled(session))),
                    physicalOperation),
            context.getDriverInstanceCount());

    // notify operator factories that planning has completed
    context.getDriverFactories().stream()
            .map(DriverFactory::getOperatorFactories)
            .flatMap(List::stream)
            .filter(LocalPlannerAware.class::isInstance)
            .map(LocalPlannerAware.class::cast)
            .forEach(LocalPlannerAware::localPlannerComplete);

    return new LocalExecutionPlan(context.getDriverFactories(), partitionedSourceOrder);
}

private static class PhysicalOperation
{
    private final List<OperatorFactoryWithTypes> operatorFactoriesWithTypes;
    private final Map<Symbol, Integer> layout; 
    private final List<Type> types;
    ... ... 

    private PhysicalOperation(
            OperatorFactory operatorFactory,
            Map<Symbol, Integer> layout,
            TypeProvider typeProvider,
            Optional<PhysicalOperation> source)
    {
        requireNonNull(operatorFactory, "operatorFactory is null");
        requireNonNull(layout, "layout is null");
        requireNonNull(typeProvider, "typeProvider is null");
        requireNonNull(source, "source is null");

        this.types = toTypes(layout, typeProvider);
        this.operatorFactoriesWithTypes = ImmutableList.<OperatorFactoryWithTypes>builder()
                // operatorFactoriesWithTypes的顺序是先上游节点，然后是自己
                .addAll(source.map(PhysicalOperation::getOperatorFactoriesWithTypes).orElse(ImmutableList.of()))
                .add(new OperatorFactoryWithTypes(operatorFactory, types))
                .build();
        this.layout = ImmutableMap.copyOf(layout);
    }
  ... ...
}
// private static class LocalExecutionPlanContext
public void addDriverFactory(boolean inputDriver, boolean outputDriver, PhysicalOperation physicalOperation, OptionalInt driverInstances)
{
    List<OperatorFactoryWithTypes> operatorFactoriesWithTypes = physicalOperation.getOperatorFactoriesWithTypes();
    // addLookupOuterDrivers 是对joinNode中 outer部分返回的特殊处理。
    addLookupOuterDrivers(outputDriver, toOperatorFactories(operatorFactoriesWithTypes));
    List<OperatorFactory> operatorFactories;
    if (isLateMaterializationEnabled(taskContext.getSession())) {
        operatorFactories = handleLateMaterialization(operatorFactoriesWithTypes);
    }
    else {
        operatorFactories = toOperatorFactories(operatorFactoriesWithTypes);
    }
    // 注意，在此次新建了一个pipeline
    driverFactories.add(new DriverFactory(getNextPipelineId(), inputDriver, outputDriver, operatorFactories, driverInstances));
}

//io.trino.execution.SqlTaskExecution
private synchronized Set<PlanNodeId> updateSplitAssignments(List<SplitAssignment> splitAssignments)
{
  ImmutableSet.Builder<PlanNodeId> updatedUnpartitionedSources = ImmutableSet.builder();

  // first remove any split that was already acknowledged
  long currentMaxAcknowledgedSplit = this.maxAcknowledgedSplit;
  splitAssignments = splitAssignments.stream()
          .map(assignment -> new SplitAssignment(
                  assignment.getPlanNodeId(),
                  assignment.getSplits().stream()
                          .filter(scheduledSplit -> scheduledSplit.getSequenceId() > currentMaxAcknowledgedSplit)
                          .collect(toImmutableSet()),
                  assignment.isNoMoreSplits()))
          // drop assignments containing no unacknowledged splits
          // the noMoreSplits signal acknowledgement is not tracked but it is okay to deliver it more than once
          .filter(assignment -> !assignment.getSplits().isEmpty() || assignment.isNoMoreSplits())
          .collect(toList());

  // update task with new assignments
  for (SplitAssignment assignment : splitAssignments) {
      if (driverRunnerFactoriesWithSplitLifeCycle.containsKey(assignment.getPlanNodeId())) {
          // 本地Source
          schedulePartitionedSource(assignment);
      }
      else {
          // 远程Source，才会加到updatedUnpartitionedSources中
          // tell existing drivers about the new splits
          DriverSplitRunnerFactory factory = driverRunnerFactoriesWithRemoteSource.get(assignment.getPlanNodeId());
          factory.enqueueSplits(assignment.getSplits(), assignment.isNoMoreSplits());
          updatedUnpartitionedSources.add(assignment.getPlanNodeId());
      }
  }

  // update maxAcknowledgedSplit
  maxAcknowledgedSplit = splitAssignments.stream()
          .flatMap(source -> source.getSplits().stream())
          .mapToLong(ScheduledSplit::getSequenceId)
          .max()
          .orElse(maxAcknowledgedSplit);
  return updatedUnpartitionedSources.build();
}

//io.trino.execution.SqlTaskExecution
private synchronized void schedulePartitionedSource(SplitAssignment splitAssignmentUpdate)
{
    mergeIntoPendingSplits(splitAssignmentUpdate.getPlanNodeId(), splitAssignmentUpdate.getSplits(), splitAssignmentUpdate.isNoMoreSplits());

    while (schedulingPlanNodeOrdinal < sourceStartOrder.size()) {
        // 注意这个循环只会执行一次，后面的SplitAssign都会加入pendingSplitsForPlanNode
        PlanNodeId schedulingPlanNode = sourceStartOrder.get(schedulingPlanNodeOrdinal);

        DriverSplitRunnerFactory partitionedDriverRunnerFactory = driverRunnerFactoriesWithSplitLifeCycle.get(schedulingPlanNode);

        PendingSplitsForPlanNode pendingSplits = pendingSplitsByPlanNode.get(schedulingPlanNode);

        // Enqueue driver runners with split lifecycle for this plan node and driver life cycle combination.
        ImmutableList.Builder<DriverSplitRunner> runners = ImmutableList.builder();
        for (ScheduledSplit scheduledSplit : pendingSplits.removeAllSplits()) {
            // create a new driver for the split
            // 创建用于执行的DriverRunner
            runners.add(partitionedDriverRunnerFactory.createDriverRunner(scheduledSplit));
        }
        // 将DriverRunner放入TaskExecutor
        enqueueDriverSplitRunner(false, runners.build());

        // If all driver runners have been enqueued for this plan node and driver life cycle combination,
        // move on to the next plan node.
        if (pendingSplits.getState() != NO_MORE_SPLITS) {
            break;
        }
        partitionedDriverRunnerFactory.noMoreDriverRunner();
        pendingSplits.markAsCleanedUp();

        schedulingPlanNodeOrdinal++;
    }
}
// 将Split添加到pendingSplitsForPlanNode中
private void mergeIntoPendingSplits(PlanNodeId planNodeId, Set<ScheduledSplit> scheduledSplits, boolean noMoreSplits)
{
    checkHoldsLock();

    DriverSplitRunnerFactory partitionedDriverFactory = driverRunnerFactoriesWithSplitLifeCycle.get(planNodeId);
    PendingSplitsForPlanNode pendingSplitsForPlanNode = pendingSplitsByPlanNode.get(planNodeId);

    partitionedDriverFactory.splitsAdded(scheduledSplits.size(), SplitWeight.rawValueSum(scheduledSplits, scheduledSplit -> scheduledSplit.getSplit().getSplitWeight()));
    for (ScheduledSplit scheduledSplit : scheduledSplits) {
        pendingSplitsForPlanNode.addSplit(scheduledSplit);
    }
    if (noMoreSplits) {
        pendingSplitsForPlanNode.setNoMoreSplits();
    }
}

public List<ListenableFuture<Void>> enqueueSplits(TaskHandle taskHandle, boolean intermediate, List<? extends SplitRunner> taskSplits)
{
    List<PrioritizedSplitRunner> splitsToDestroy = new ArrayList<>();
    List<ListenableFuture<Void>> finishedFutures = new ArrayList<>(taskSplits.size());
    synchronized (this) {
        for (SplitRunner taskSplit : taskSplits) {
            // 将原有的SplitRunner包装为PrioritizedSplitRunner
            PrioritizedSplitRunner prioritizedSplitRunner = new PrioritizedSplitRunner(
                    taskHandle,
                    taskSplit,
                    ticker,
                    globalCpuTimeMicros,
                    globalScheduledTimeMicros,
                    blockedQuantaWallTime,
                    unblockedQuantaWallTime);

            if (taskHandle.isDestroyed()) {
                // If the handle is destroyed, we destroy the task splits to complete the future
                splitsToDestroy.add(prioritizedSplitRunner);
            }
            else if (intermediate) {
                // 立刻启动
                // Note: we do not record queued time for intermediate splits
                startIntermediateSplit(prioritizedSplitRunner);
                // add the runner to the handle so it can be destroyed if the task is canceled
                taskHandle.recordIntermediateSplit(prioritizedSplitRunner);
            }
            else {
                // 将Split 入队
                taskHandle.enqueueSplit(prioritizedSplitRunner);
                // 如果Task启动的Split低于保证(guaranteedNumberOfDriversPerTask)，尝试启动Split
                scheduleTaskIfNecessary(taskHandle);
                // 如果有更多的全局资源(低于minimumNumberOfDrivers)，TaskExecutor会全局启动更多Split
                addNewEntrants();
            }

            finishedFutures.add(prioritizedSplitRunner.getFinishedFuture());
        }
        recordLeafSplitsSize();
    }
    // 销毁split
    for (PrioritizedSplitRunner split : splitsToDestroy) {
        split.destroy();
    }
    return finishedFutures;
}

public class MultilevelSplitQueue
{
    //每个level的阈值，比如LEVEL_THRESHOLD_SECONDS[1] = 1s，也就是说累计运行时间超过1s的任务会被放到level1。
    static final int[] LEVEL_THRESHOLD_SECONDS = {0, 1, 10, 60, 300};
    //一个保护性的值，一次process被计算的上限时间，避免某些任务因为特殊情况被计算了太多值。
    static final long LEVEL_CONTRIBUTION_CAP = SECONDS.toNanos(30);
    // 一个size为5的优先队列的2维数组。共5个level，
    // 每次take时，通过比对每一level已经消耗的时间，选取一个level，
    // 从该level里选取优先级最高的PrioritizedSplitRunner。
    @GuardedBy("lock")
    private final PriorityQueue<PrioritizedSplitRunner>[] levelWaitingSplits;
    //一个size为5的调度时间累加器，表示当前level的已用掉的调度时间。
    // PrioritizedSplitRunner在执行完任务的process之后会增加当前level的调度时间。
    // 这部分的数据主要用于在poll取PrioritizedSplitRunner数据的时候计算具体从哪个level的队列里面拿。
    private final AtomicLong[] levelScheduledTime;
    //是一个size为5的优先级分数数组，表明当前level的最小的优先级分数，
    // 每次从队里成功take出数据之后会更新这个当前level的优先级分数。
    private final AtomicLong[] levelMinPriority;
    private final CounterStat[] selectedLevelCounters;

    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    //这个字段用于设置不同level之间的cpu时间分配。默认用的是2，也就是level0-4时间占比为16:8:4:2:1
    private final double levelTimeMultiplier;
}

public void offer(PrioritizedSplitRunner split)
{
    checkArgument(split != null, "split is null");

    split.setReady();
    int level = split.getPriority().getLevel();
    lock.lock();
    try {
        if (levelWaitingSplits[level].isEmpty()) {
            // Accesses to levelScheduledTime are not synchronized, so we have a data race
            // here - our level time math will be off. However, the staleness is bounded by
            // the fact that only running splits that complete during this computation
            // can update the level time. Therefore, this is benign.
            long level0Time = getLevel0TargetTime();
            long levelExpectedTime = (long) (level0Time / Math.pow(levelTimeMultiplier, level));
            long delta = levelExpectedTime - levelScheduledTime[level].get();
            levelScheduledTime[level].addAndGet(delta);
        }

        levelWaitingSplits[level].offer(split);
        notEmpty.signal();
    }
    finally {
        lock.unlock();
    }
}

public PrioritizedSplitRunner take()
            throws InterruptedException
{
    while (true) {
        lock.lockInterruptibly();
        try {
            PrioritizedSplitRunner result;
            while ((result = pollSplit()) == null) {
                notEmpty.await();
            }

            if (result.updateLevelPriority()) {
                offer(result);
                continue;
            }

            int selectedLevel = result.getPriority().getLevel();
            levelMinPriority[selectedLevel].set(result.getPriority().getLevelPriority());
            selectedLevelCounters[selectedLevel].update(1);

            return result;
        }
        finally {
            lock.unlock();
        }
    }
}
// io.trino.execution.executor.PrioritizedSplitRunner
public boolean updateLevelPriority()
{
    Priority newPriority = taskHandle.getPriority();
    Priority oldPriority = priority.getAndSet(newPriority);
    return newPriority.getLevel() != oldPriority.getLevel();
}

private PrioritizedSplitRunner pollSplit()
{
    long targetScheduledTime = getLevel0TargetTime();
    double worstRatio = 1;
    int selectedLevel = -1;
    for (int level = 0; level < LEVEL_THRESHOLD_SECONDS.length; level++) {
        if (!levelWaitingSplits[level].isEmpty()) {
            long levelTime = levelScheduledTime[level].get();
            double ratio = levelTime == 0 ? 0 : targetScheduledTime / (1.0 * levelTime);
            if (selectedLevel == -1 || ratio > worstRatio) {
                worstRatio = ratio;
                selectedLevel = level;
            }
        }

        targetScheduledTime /= levelTimeMultiplier;
    }

    if (selectedLevel == -1) {
        return null;
    }

    PrioritizedSplitRunner result = levelWaitingSplits[selectedLevel].poll();
    checkState(result != null, "pollSplit cannot return null");

    return result;
}

private long getLevel0TargetTime()
{
    long level0TargetTime = levelScheduledTime[0].get();
    double currentMultiplier = levelTimeMultiplier;
 
    for (int level = 0; level < LEVEL_THRESHOLD_SECONDS.length; level++) {
        currentMultiplier /= levelTimeMultiplier;
        long levelTime = levelScheduledTime[level].get();
        // 选择的是所有level经过比例换算之后最大的那个时间。
        level0TargetTime = Math.max(level0TargetTime, (long) (levelTime / currentMultiplier));
    }
 
    return level0TargetTime;
}

public Priority updatePriority(Priority oldPriority, long quantaNanos, long scheduledNanos)
{
    int oldLevel = oldPriority.getLevel();
    // 计算新的level
    int newLevel = computeLevel(scheduledNanos); 
    //上限保护
    long levelContribution = Math.min(quantaNanos, LEVEL_CONTRIBUTION_CAP);

    if (oldLevel == newLevel) {
        // 增加调度耗时
        addLevelTime(oldLevel, levelContribution);
        return new Priority(oldLevel, oldPriority.getLevelPriority() + quantaNanos);
    }

    long remainingLevelContribution = levelContribution;
    long remainingTaskTime = quantaNanos;

    // a task normally slowly accrues scheduled time in a level and then moves to the next, but
    // if the split had a particularly long quanta, accrue time to each level as if it had run
    // in that level up to the level limit.
    for (int currentLevel = oldLevel; currentLevel < newLevel; currentLevel++) {
        // 对不同level的时间贡献
        long timeAccruedToLevel = Math.min(SECONDS.toNanos(LEVEL_THRESHOLD_SECONDS[currentLevel + 1] - LEVEL_THRESHOLD_SECONDS[currentLevel]), remainingLevelContribution);
        addLevelTime(currentLevel, timeAccruedToLevel);
        remainingLevelContribution -= timeAccruedToLevel;
        remainingTaskTime -= timeAccruedToLevel;
    }

    addLevelTime(newLevel, remainingLevelContribution);
    // 获取level中最低的优先级
    long newLevelMinPriority = getLevelMinPriority(newLevel, scheduledNanos);
    // 耗时越高，levelPriority越高，越不易从优先级队列中取出。
    return new Priority(newLevel, newLevelMinPriority + remainingTaskTime);
}

 public ListenableFuture<Void> process()
{
    try {
        // 本次调度启动时间
        long startNanos = ticker.read();
        // 启动时间
        start.compareAndSet(0, startNanos);
        // 上次退出时间
        lastReady.compareAndSet(0, startNanos);
        // 调度次数+1
        processCalls.incrementAndGet();
        // 等待时间
        waitNanos.getAndAdd(startNanos - lastReady.get());

        CpuTimer timer = new CpuTimer();
        // 调用内部Split，最多只能执行1s，否则返回ListenableFuture
        ListenableFuture<Void> blocked = split.processFor(SPLIT_RUN_QUANTA);
        CpuTimer.CpuDuration elapsed = timer.elapsedTime();

        long quantaScheduledNanos = ticker.read() - startNanos;
        scheduledNanos.addAndGet(quantaScheduledNanos);
        // task更新优先级和耗时
        priority.set(taskHandle.addScheduledNanos(quantaScheduledNanos));
        // 更新上次运行时间
        lastRun.set(ticker.read());

        if (blocked == NOT_BLOCKED) {
            unblockedQuantaWallTime.add(elapsed.getWall());
        }
        else {
            blockedQuantaWallTime.add(elapsed.getWall());
        }

        long quantaCpuNanos = elapsed.getCpu().roundTo(NANOSECONDS);
        cpuTimeNanos.addAndGet(quantaCpuNanos);

        globalCpuTimeMicros.update(quantaCpuNanos / 1000);
        globalScheduledTimeMicros.update(quantaScheduledNanos / 1000);

        return blocked;
    }
    catch (Throwable e) {
        finishedFuture.setException(e);
        throw e;
    }
}