SqlQueryExecution::start方法

    1.   @Override
    2.     public void start()
    3.     {
    4.         try (SetThreadName ignored = new SetThreadName("Query-%s", stateMachine.getQueryId())) {
    5.             try {
    6.                 // transition to planning   过渡到计划
    7.                 if (!stateMachine.transitionToPlanning()) {
    8.                     // query already started or finished
    9.                     // 查询已经开始或完成
    10.                     return;
    11.                 }
    13.                 // analyze query   根据query生成执行计划
    14.                 PlanRoot plan = analyzeQuery();
    16.                 try {
    17.                     // 处理跨区域动态过滤器
    18.                     handleCrossRegionDynamicFilter(plan);
    19.                 }
    20.                 catch (Throwable e) {
    21.                     // ignore any exception
    22.                     log.warn("something unexpected happened.. cause: %s", e.getMessage());
    23.                 }
    25.                 // plan distribution of query  计划查询分布式
    26.                 planDistribution(plan);
    28.                 // transition to starting
    29.                 if (!stateMachine.transitionToStarting()) {
    30.                     // query already started or finished
    31.                     return;
    32.                 }
    33.                 stateMachine.addStateChangeListener(state -> {
    34.                     if (state == QueryState.RESUMING) {
    35.                         // Snapshot: old stages/tasks have finished. Ready to resume.
    36.                         try {
    37.                             // 当完成分布式计划后，需要进行
    38.                             resumeQuery(plan);
    39.                         }
    40.                         catch (Throwable e) {
    41.                             fail(e);
    42.                             throwIfInstanceOf(e, Error.class);
    43.                             log.warn(e, "Encountered error while rescheduling query");
    44.                         }
    45.                     }
    46.                 });
    47.                 // if query is not finished, start the scheduler, otherwise cancel it
    48.                 // 如果查询未完成，请启动调度程序，否则取消调度程序
    49.                 SqlQueryScheduler scheduler = queryScheduler.get();
    51.                 if (!stateMachine.isDone()) {
    52.                     scheduler.start();
    53.                 }
    54.             }
    55.             catch (Throwable e) {
    56.                 fail(e);
    57.                 throwIfInstanceOf(e, Error.class);
    58.                 log.warn(e, "Encountered error while scheduling query");
    59.             }
    60.         }
    61.     }

    image.png

    SqlQueryScheduler::start方法开始进行调度

    1. public void start()
    2. {
    3.     if (started.compareAndSet(false, true)) {
    4.         executor.submit(this::schedule);
    5.     }
    6. }

    其实内部调用内部的schedule方法

    1.     private void schedule()
    2.     {
    3.         try (SetThreadName ignored = new SetThreadName("Query-%s", queryStateMachine.getQueryId())) {
    4.             // 保存stageId
    5.             Set<StageId> completedStages = new HashSet<>();
    7.             // 根据SqlStageExecution 属于哪个阶段的执行，执行策略创建对应的执行调度器
    8.             ExecutionSchedule executionSchedule = executionPolicy.createExecutionSchedule(stages.values());
    9.             // 如果
    10.             while (!executionSchedule.isFinished()) {
    11.                 // 保存堵塞的stage
    12.                 List<ListenableFuture<?>> blockedStages = new ArrayList<>();
    13.                 //
    14.                 for (SqlStageExecution stage : executionSchedule.getStagesToSchedule()) {
    15.                     if (isReuseTableScanEnabled(session) && !SqlStageExecution.getReuseTableScanMappingIdStatus(stage.getStateMachine())) {
    16.                         continue;
    17.                     }
    19.                     stage.beginScheduling();
    21.                     // Resource group: Check if memory usage within the current resource grooup has exceeded
    22.                     // configured limit. If yes throttle further split scheduling.
    23.                     // Throttle Logic: Wait for x seconds (Wait time will increase till max as per THROTTLE_SLEEP_TIMER)
    24.                     // and then let it schedule 10% of splits.
    25.                     if (!canScheduleMoreSplits()) {
    26.                         try {
    27.                             SECONDS.sleep(THROTTLE_SLEEP_TIMER[currentTimerLevel]);
    28.                         }
    29.                         catch (InterruptedException e) {
    30.                             throw new PrestoException(GENERIC_INTERNAL_ERROR, "interrupted while sleeping");
    31.                         }
    32.                         currentTimerLevel = Math.min(currentTimerLevel + 1, THROTTLE_SLEEP_TIMER.length - 1);
    33.                         stage.setThrottledSchedule(true);
    34.                     }
    35.                     else {
    36.                         stage.setThrottledSchedule(false);
    37.                         currentTimerLevel = 0;
    38.                     }
    40.                     // perform some scheduling work  执行一些调度工作
    41.                     // 根据stageId获取对应的调度器进行执行，并返回调度结果
    42.                     ScheduleResult result = stageSchedulers.get(stage.getStageId())
    43.                             .schedule();
    45.                     // modify parent and children based on the results of the scheduling
    46.                     if (result.isFinished()) {
    47.                         stage.schedulingComplete();
    48.                     }
    49.                     else if (!result.getBlocked().isDone()) {
    50.                         blockedStages.add(result.getBlocked());
    51.                     }
    53.                     stageLinkages.get(stage.getStageId())
    54.                             .processScheduleResults(stage.getState(), result.getNewTasks());
    55.                     schedulerStats.getSplitsScheduledPerIteration().add(result.getSplitsScheduled());
    56.                     if (result.getBlockedReason().isPresent()) {
    57.                         switch (result.getBlockedReason().get()) {
    58.                             case WRITER_SCALING:
    59.                                 // no-op
    60.                                 break;
    61.                             case WAITING_FOR_SOURCE:
    62.                                 schedulerStats.getWaitingForSource().update(1);
    63.                                 break;
    64.                             case SPLIT_QUEUES_FULL:
    65.                                 schedulerStats.getSplitQueuesFull().update(1);
    66.                                 break;
    67.                             case MIXED_SPLIT_QUEUES_FULL_AND_WAITING_FOR_SOURCE:
    68.                             case NO_ACTIVE_DRIVER_GROUP:
    69.                                 break;
    70.                             default:
    71.                                 throw new UnsupportedOperationException("Unknown blocked reason: " + result.getBlockedReason().get());
    72.                         }
    73.                     }
    74.                 }
    76.                 // make sure to update stage linkage at least once per loop to catch async state changes (e.g., partial cancel)
    77.                 for (SqlStageExecution stage : stages.values()) {
    78.                     if (!completedStages.contains(stage.getStageId()) && stage.getState().isDone()) {
    79.                         stageLinkages.get(stage.getStageId())
    80.                                 .processScheduleResults(stage.getState(), ImmutableSet.of());
    81.                         completedStages.add(stage.getStageId());
    82.                     }
    83.                 }
    85.                 // wait for a state change and then schedule again
    86.                 if (!blockedStages.isEmpty()) {
    87.                     try (TimeStat.BlockTimer timer = schedulerStats.getSleepTime().time()) {
    88.                         tryGetFutureValue(whenAnyComplete(blockedStages), 1, SECONDS);
    89.                     }
    90.                     for (ListenableFuture<?> blockedStage : blockedStages) {
    91.                         blockedStage.cancel(true);
    92.                     }
    93.                 }
    94.             }
    96.             for (SqlStageExecution stage : stages.values()) {
    97.                 StageState state = stage.getState();
    98.                 // Snapshot: if state is resumable_failure, then state of stage and query will change soon again. Don't treat as an error.
    99.                 if (state != SCHEDULED && state != RUNNING && !state.isDone() && state != RESUMABLE_FAILURE) {
    100.                     throw new PrestoException(GENERIC_INTERNAL_ERROR, format("Scheduling is complete, but stage %s is in state %s", stage.getStageId(), state));
    101.                 }
    102.             }
    103.         }
    104.         catch (Throwable t) {
    105.             queryStateMachine.transitionToFailed(t);
    106.             throw t;
    107.         }
    108.         finally {
    109.             RuntimeException closeError = new RuntimeException();
    110.             for (StageScheduler scheduler : stageSchedulers.values()) {
    111.                 try {
    112.                     // Snapshot: when trying to reschedule, then don't close the scheduler (and more importantly, split sources in it)
    113.                     QueryState state = queryStateMachine.getQueryState();
    114.                     if (state != QueryState.RESCHEDULING && state != QueryState.RESUMING) {
    115.                         scheduler.close();
    116.                     }
    117.                 }
    118.                 catch (Throwable t) {
    119.                     queryStateMachine.transitionToFailed(t);
    120.                     // Self-suppression not permitted
    121.                     if (closeError != t) {
    122.                         closeError.addSuppressed(t);
    123.                     }
    124.                 }
    125.             }
    127.             // Snpashot: if resuming, notify the new scheduler so it can start scheduling new stages
    128.             schedulingFuture.set(null);
    129.             if (closeError.getSuppressed().length > 0) {
    130.                 throw closeError;
    131.             }
    132.         }
    133.     }

    image.png
    查看StageScheduler，是一个接口，实现不同阶段调度器的作用，我们以SourcePartitionedScheduler为例，schedule实现该方法
    image.png

    1. public PlanNodeId getPlanNodeId()
    2.     {
    3.         return partitionedNode;
    4.     }
    6.     /**
    7.      * Obtains an instance of {@code SourcePartitionedScheduler} suitable for use as a
    8.      * stage scheduler. 获取一个sourcepartitionedscheduler实例，作为一个阶段的调度器
    9.      * <p>
    10.      * This returns an ungrouped {@code SourcePartitionedScheduler} that requires
    11.      * minimal management from the caller, which is ideal for use as a stage scheduler.
    12.      * 返回一个为分组的source调度器
    13.      *
    14.      */
    15.     public static StageScheduler newSourcePartitionedSchedulerAsStageScheduler(
    16.             SqlStageExecution stage,
    17.             PlanNodeId partitionedNode,
    18.             SplitSource splitSource,
    19.             SplitPlacementPolicy splitPlacementPolicy,
    20.             int splitBatchSize,
    21.             Session session,
    22.             HeuristicIndexerManager heuristicIndexerManager,
    23.             Map<Integer, Integer> parallelSources)
    24.     {
    26.         // 创建源调度器，
    27.         SourcePartitionedScheduler sourcePartitionedScheduler = new SourcePartitionedScheduler(stage, partitionedNode, splitSource,
    28.                 splitPlacementPolicy, splitBatchSize, false, session, heuristicIndexerManager,
    29.                 parallelSources == null ? 1 : parallelSources.getOrDefault(stage.getStageId().getId(), 1));
    32.         sourcePartitionedScheduler.startLifespan(Lifespan.taskWide(), NOT_PARTITIONED);
    33.         sourcePartitionedScheduler.noMoreLifespans();
    35.         // 重载源schedule调度方法
    36.         return new StageScheduler()
    37.         {
    38.             @Override
    39.             public ScheduleResult schedule()
    40.             {
    41.                 ScheduleResult scheduleResult = sourcePartitionedScheduler.schedule();
    42.                 sourcePartitionedScheduler.drainCompletedLifespans();
    43.                 return scheduleResult;
    44.             }
    46.             @Override
    47.             public void close()
    48.             {
    49.                 sourcePartitionedScheduler.close();
    50.             }
    51.         };
    52.     }

    部分代码入口，source调度器进行调度使用，
    image.png
    进入sourcePartitionedScheduler::schedule方法内

    1.     @Override
    2.     public synchronized ScheduleResult schedule()
    3.     {
    5.         // 遍历分组的调度器
    6.         for (Entry<Lifespan, ScheduleGroup> entry : scheduleGroups.entrySet()) {
    7.             // .....
    9.             // 分配好split进行task的创建
    10.             // assign the splits with successful placements
    11.             overallNewTasks.addAll(assignSplits(splitAssignment, noMoreSplitsNotification));
    13.             // ......
    15.         }
    18.     }

    overallNewTasks 保存了所有的task，其中调用了assignSplits的方法，具体实现如下所示：

    1. private Set<RemoteTask> assignSplits(Multimap<InternalNode, Split> splitAssignment, Multimap<InternalNode, Lifespan> noMoreSplitsNotification)
    2.     {
    3.         ImmutableSet.Builder<RemoteTask> newTasks = ImmutableSet.builder();
    5.         // 收集所有的节点
    6.         ImmutableSet<InternalNode> nodes = ImmutableSet.<InternalNode>builder()
    7.                 .addAll(splitAssignment.keySet())
    8.                 .addAll(noMoreSplitsNotification.keySet())
    9.                 .build();
    11.         // 遍历每个node
    12.         for (InternalNode node : nodes) {
    13.             // source partitioned tasks can only receive broadcast data; otherwise it would have a different distribution
    14.             // 源分区任务只能接受广播数据，否则它有一个不同的分布式
    15.             ImmutableMultimap<PlanNodeId, Split> splits = ImmutableMultimap.<PlanNodeId, Split>builder()
    16.                     .putAll(partitionedNode, splitAssignment.get(node))  // 每个节点对应的split
    17.                     .build();
    18.             // 每个节点对应的生命周阶段
    19.             ImmutableMultimap.Builder<PlanNodeId, Lifespan> noMoreSplits = ImmutableMultimap.builder();
    20.             if (noMoreSplitsNotification.containsKey(node)) {
    21.                 noMoreSplits.putAll(partitionedNode, noMoreSplitsNotification.get(node));
    22.             }
    23.             // node, split, lifespan 共同创建task
    24.             newTasks.addAll(stage.scheduleSplits(
    25.                     node,
    26.                     splits,
    27.                     noMoreSplits.build()));
    28.         }
    29.         return newTasks.build();
    30.     }

    进一步分析SqlStageExecution::scheduleSplits的方法

    1.    public synchronized Set<RemoteTask> scheduleSplits(InternalNode node, Multimap<PlanNodeId, Split> splits, Multimap<PlanNodeId, Lifespan> noMoreSplitsNotification)
    2.     {
    3.         requireNonNull(node, "node is null");
    4.         requireNonNull(splits, "splits is null");
    6.         // 如果检测到stage的状态已完成，则返回空任务， 并将split调度表示置为true
    7.         if (stateMachine.getState().isDone()) {
    8.             return ImmutableSet.of();
    9.         }
    10.         splitsScheduled.set(true);
    12.         checkArgument(stateMachine.getFragment().getPartitionedSources().containsAll(splits.keySet()), "Invalid splits");
    14.         // 创建新任务集合
    15.         ImmutableSet.Builder<RemoteTask> newTasks = ImmutableSet.builder();
    16.         Collection<RemoteTask> tasks = this.tasks.get(node);
    17.         RemoteTask task;
    19.         /**
    20.          * 1。如果tasks集合中没有对应的远程任务是需要进行创建
    21.          * 2。否则需要将splits添加到对应的任务中去
    22.          */
    23.         if (tasks == null) {
    24.             // The output buffer depends on the task id starting from 0 and being sequential, since each
    25.             // task is assigned a private buffer based on task id.
    26.             TaskId taskId = new TaskId(stateMachine.getStageId(), nextTaskId.getAndIncrement());
    27.             task = scheduleTask(node, taskId, splits, OptionalInt.empty());  // TODO 主要用于创建任务
    28.             newTasks.add(task);
    29.         }
    30.         else {
    31.             task = tasks.iterator().next();
    32.             task.addSplits(splits);
    33.         }
    36.         if (noMoreSplitsNotification.size() > 1) {
    37.             // The assumption that `noMoreSplitsNotification.size() <= 1` currently holds.
    38.             // If this assumption no longer holds, we should consider calling task.noMoreSplits with multiple entries in one shot.
    39.             // These kind of methods can be expensive since they are grabbing locks and/or sending HTTP requests on change.
    40.             throw new UnsupportedOperationException("This assumption no longer holds: noMoreSplitsNotification.size() < 1");
    41.         }
    42.         for (Entry<PlanNodeId, Lifespan> entry : noMoreSplitsNotification.entries()) {
    43.             task.noMoreSplits(entry.getKey(), entry.getValue());
    44.         }
    45.         return newTasks.build();
    46.     }

    所有任务的创建都会调用到该方法：SqlStageExecution::scheduleTask

    1.     private synchronized RemoteTask scheduleTask(InternalNode node, TaskId taskId, Multimap<PlanNodeId, Split> sourceSplits, OptionalInt totalPartitions)
    2.     {
    3.         checkArgument(!allTasks.contains(taskId), "A task with id %s already exists", taskId);
    5.         // 是否开启快照功能
    6.         if (SystemSessionProperties.isSnapshotEnabled(stateMachine.getSession())) {
    7.             // Snapshot: inform snapshot manager so it knows about all tasks,
    8.             // and can determine if a snapshot is complete for all tasks.
    9.             snapshotManager.addNewTask(taskId);
    10.         }
    12.         ImmutableMultimap.Builder<PlanNodeId, Split> initialSplits = ImmutableMultimap.builder();
    13.         initialSplits.putAll(sourceSplits);
    15.         /**
    16.          * 1。遍历所有源任务，查看任务状态是否为完成，如果没有完成，则需要将远程任务，并根据任务ID创建split
    17.          */
    18.         sourceTasks.forEach((planNodeId, task) -> {
    19.             TaskStatus status = task.getTaskStatus();
    20.             if (status.getState() != TaskState.FINISHED) {
    21.                 initialSplits.put(planNodeId, createRemoteSplitFor(taskId, status.getSelf()));
    22.             }
    23.         });
    25.         OutputBuffers outputBuffers = this.outputBuffers.get();
    26.         checkState(outputBuffers != null, "Initial output buffers must be set before a task can be scheduled");
    28.         // 
    29.         RemoteTask task = remoteTaskFactory.createRemoteTask(
    30.                 stateMachine.getSession(),
    31.                 taskId,
    32.                 node,
    33.                 stateMachine.getFragment(),
    34.                 initialSplits.build(),
    35.                 totalPartitions,
    36.                 outputBuffers,
    37.                 nodeTaskMap.createPartitionedSplitCountTracker(node, taskId),
    38.                 summarizeTaskInfo,
    39.                 Optional.ofNullable(parentId),
    40.                 snapshotManager);
    42.         completeSources.forEach(task::noMoreSplits);
    44.         allTasks.add(taskId);
    45.         tasks.computeIfAbsent(node, key -> newConcurrentHashSet()).add(task);
    46.         nodeTaskMap.addTask(node, task);
    48.         task.addStateChangeListener(new StageTaskListener());
    49.         task.addFinalTaskInfoListener(this::updateFinalTaskInfo);
    51.         if (!stateMachine.getState().isDone()) {
    52.             task.start();
    53.         }
    54.         else {
    55.             // stage finished while we were scheduling this task
    56.             // 如果stage完成了，然而我们还是调用的话，需要丢弃该任务
    57.             task.abort();
    58.         }
    60.         return task;
    61.     }

    最终会调用RemoteTaskFactory::createRemoteTask的方法来执行任务的创建，HttpRemoteTaskFactory重载该方法用于创建HttpRemoteTask，具体代码实现如下：

    1.    @Override
    2.     public RemoteTask createRemoteTask(Session session,
    3.             TaskId taskId,
    4.             InternalNode node,
    5.             PlanFragment fragment,
    6.             Multimap<PlanNodeId, Split> initialSplits,
    7.             OptionalInt totalPartitions,
    8.             OutputBuffers outputBuffers,
    9.             PartitionedSplitCountTracker partitionedSplitCountTracker,
    10.             boolean summarizeTaskInfo,
    11.             Optional<PlanNodeId> parent,
    12.             QuerySnapshotManager snapshotManager)
    13.     {
    14.         return new HttpRemoteTask(session,
    15.                 taskId,
    16.                 node.getNodeIdentifier(),
    17.                 locationFactory.createTaskLocation(node, taskId),
    18.                 fragment,
    19.                 initialSplits,
    20.                 totalPartitions,
    21.                 outputBuffers,
    22.                 httpClient,
    23.                 executor,
    24.                 updateScheduledExecutor,
    25.                 errorScheduledExecutor,
    26.                 maxErrorDuration,
    27.                 taskStatusRefreshMaxWait,
    28.                 taskInfoUpdateInterval,
    29.                 summarizeTaskInfo,
    30.                 taskStatusCodec,
    31.                 taskInfoCodec,
    32.                 taskUpdateRequestCodec,
    33.                 partitionedSplitCountTracker,
    34.                 stats,
    35.                 isBinaryEncoding,
    36.                 parent,
    37.                 snapshotManager);
    38.     }

    �HttpRemoteTask构造函数中的具体实现方式

    1.     public HttpRemoteTask(Session session,
    2.             TaskId taskId,
    3.             String nodeId,
    4.             URI location,
    5.             PlanFragment planFragment,
    6.             Multimap<PlanNodeId, Split> initialSplits,
    7.             OptionalInt totalPartitions,
    8.             OutputBuffers outputBuffers,
    9.             HttpClient httpClient,
    10.             Executor executor,
    11.             ScheduledExecutorService updateScheduledExecutor,
    12.             ScheduledExecutorService errorScheduledExecutor,
    13.             Duration maxErrorDuration,
    14.             Duration taskStatusRefreshMaxWait,
    15.             Duration taskInfoUpdateInterval,
    16.             boolean summarizeTaskInfo,
    17.             Codec<TaskStatus> taskStatusCodec,
    18.             Codec<TaskInfo> taskInfoCodec,
    19.             Codec<TaskUpdateRequest> taskUpdateRequestCodec,
    20.             PartitionedSplitCountTracker partitionedSplitCountTracker,
    21.             RemoteTaskStats stats,
    22.             boolean isBinaryEncoding,
    23.             Optional<PlanNodeId> parent,
    24.             QuerySnapshotManager snapshotManager)
    25.     {
    26.         requireNonNull(session, "session is null");
    27.         requireNonNull(taskId, "taskId is null");
    28.         requireNonNull(nodeId, "nodeId is null");
    29.         requireNonNull(location, "location is null");
    30.         requireNonNull(planFragment, "planFragment is null");
    31.         requireNonNull(totalPartitions, "totalPartitions is null");
    32.         requireNonNull(outputBuffers, "outputBuffers is null");
    33.         requireNonNull(httpClient, "httpClient is null");
    34.         requireNonNull(executor, "executor is null");
    35.         requireNonNull(taskStatusCodec, "taskStatusCodec is null");
    36.         requireNonNull(taskInfoCodec, "taskInfoCodec is null");
    37.         requireNonNull(taskUpdateRequestCodec, "taskUpdateRequestCodec is null");
    38.         requireNonNull(partitionedSplitCountTracker, "partitionedSplitCountTracker is null");
    39.         requireNonNull(stats, "stats is null");
    40.         requireNonNull(parent, "parent is null");
    42.         try (SetThreadName ignored = new SetThreadName("HttpRemoteTask-%s", taskId)) {
    43.             this.taskId = taskId;
    44.             this.session = session;
    45.             this.nodeId = nodeId;
    46.             this.planFragment = planFragment;
    47.             this.totalPartitions = totalPartitions;
    48.             this.outputBuffers.set(outputBuffers);
    49.             this.httpClient = httpClient;
    50.             this.executor = executor;
    51.             this.errorScheduledExecutor = errorScheduledExecutor;
    52.             this.summarizeTaskInfo = summarizeTaskInfo;
    53.             this.taskInfoCodec = taskInfoCodec;
    54.             this.taskUpdateRequestCodec = taskUpdateRequestCodec;
    55.             this.updateErrorTracker = new RequestErrorTracker(taskId, location, maxErrorDuration, errorScheduledExecutor, "updating task");
    56.             this.partitionedSplitCountTracker = requireNonNull(partitionedSplitCountTracker, "partitionedSplitCountTracker is null");
    57.             this.stats = stats;
    58.             this.isBinaryEncoding = isBinaryEncoding;
    59.             this.parent = parent;
    61.             for (Entry<PlanNodeId, Split> entry : requireNonNull(initialSplits, "initialSplits is null").entries()) {
    62.                 ScheduledSplit scheduledSplit = new ScheduledSplit(nextSplitId.getAndIncrement(), entry.getKey(), entry.getValue());
    63.                 pendingSplits.put(entry.getKey(), scheduledSplit);  // planNodeId --> split(splitId, plannodeId, Split)
    64.             }
    66.             // 统计所有split的总数
    67.             pendingSourceSplitCount = planFragment.getPartitionedSources().stream()
    68.                     .filter(initialSplits::containsKey)
    69.                     .mapToInt(partitionedSource -> initialSplits.get(partitionedSource).size())
    70.                     .sum();
    72.             //
    73.             List<BufferInfo> bufferStates = outputBuffers.getBuffers()
    74.                     .keySet().stream()
    75.                     .map(outputId -> new BufferInfo(outputId, false, 0, 0, PageBufferInfo.empty()))
    76.                     .collect(toImmutableList());
    78.             TaskInfo initialTask = createInitialTask(taskId, location, nodeId, bufferStates, new TaskStats(DateTime.now(), null));
    80.             this.taskStatusFetcher = new ContinuousTaskStatusFetcher(
    81.                     this::failTask,
    82.                     initialTask.getTaskStatus(),
    83.                     taskStatusRefreshMaxWait,
    84.                     taskStatusCodec,
    85.                     executor,
    86.                     httpClient,
    87.                     maxErrorDuration,
    88.                     errorScheduledExecutor,
    89.                     stats,
    90.                     isBinaryEncoding,
    91.                     snapshotManager);
    93.             this.taskInfoFetcher = new TaskInfoFetcher(
    94.                     this::failTask,
    95.                     initialTask,
    96.                     httpClient,
    97.                     taskInfoUpdateInterval,
    98.                     taskInfoCodec,
    99.                     maxErrorDuration,
    100.                     summarizeTaskInfo,
    101.                     executor,
    102.                     updateScheduledExecutor,
    103.                     errorScheduledExecutor,
    104.                     stats,
    105.                     isBinaryEncoding);
    107.             // 添加监听器， 当状态发生变化时，需要进行回调
    108.             taskStatusFetcher.addStateChangeListener(newStatus -> {
    109.                 TaskState state = newStatus.getState();
    110.                 if (state.isDone()) {
    111.                     cleanUpTask(state);
    112.                 }
    113.                 else {
    114.                     partitionedSplitCountTracker.setPartitionedSplitCount(getPartitionedSplitCount());
    115.                     updateSplitQueueSpace();
    116.                 }
    117.             });
    119.             partitionedSplitCountTracker.setPartitionedSplitCount(getPartitionedSplitCount());
    120.             updateSplitQueueSpace();
    121.         }
    122.     }

    创建完remote task之后，继续看代码，可以看到任务执行start方法启动任务，
    image.png

    最终调用HttpRemoteTask::start方法，具体代码实现方式

    1. @Override
    2. public void start()
    3. {
    4.     try (SetThreadName ignored = new SetThreadName("HttpRemoteTask-%s", taskId)) {
    5.         // to start we just need to trigger an update
    6.         // 启动调度任务
    7.         scheduleUpdate();
    9.         taskStatusFetcher.start();  // 任务状态收集器
    10.         taskInfoFetcher.start();   // 任务信息收集器
    11.     }
    12. }

    最终会调用HttpRemoteTask::�scheduleUpdate的方法进行开始任务的调度，具体代码：

    1. private void scheduleUpdate()
    2. {
    3.     executor.execute(this::sendUpdate);
    4. }

    �

    1.     private synchronized void sendUpdate()
    2.     {
    3.         if (abandoned.get()) {
    4.             // Snapshot: Corresponding task has been canceled to resume. Stop any communication with it.
    5.             // 相应任务已经取消需要重启的，需停止与他进行通信
    6.             return;
    7.         }
    8.         TaskStatus taskStatus = getTaskStatus();
    9.         // don't update if the task hasn't been started yet or if it is already finished
    10.         // 如果任务未开始，或者已经结束了，不需要更新
    11.         if (!needsUpdate.get() || taskStatus.getState().isDone()) {
    12.             return;
    13.         }
    15.         // if there is a request already running, wait for it to complete
    16.         // 有一个请求已经正在运行等待完成，不需要更新
    17.         if (this.currentRequest != null && !this.currentRequest.isDone()) {
    18.             return;
    19.         }
    21.         // if throttled due to error, asynchronously wait for timeout and try again
    22.         // 如果误杀任务导致错误的，异步的等待超时或者重试
    23.         ListenableFuture<?> errorRateLimit = updateErrorTracker.acquireRequestPermit();
    24.         if (!errorRateLimit.isDone()) {
    25.             errorRateLimit.addListener(this::sendUpdate, executor);
    26.             return;
    27.         }
    29.         List<TaskSource> sources = getSources();
    31.         Optional<PlanFragment> fragment = sendPlan.get() ? Optional.of(planFragment) : Optional.empty();
    32.         TaskUpdateRequest updateRequest = new TaskUpdateRequest(
    33.                 // Snapshot: Add task instance id to all task related requests,
    34.                 // so receiver can verify if the instance id matches
    35.                 taskStatus.getTaskInstanceId(),
    36.                 session.toSessionRepresentation(),
    37.                 session.getIdentity().getExtraCredentials(),
    38.                 fragment,
    39.                 sources,
    40.                 outputBuffers.get(),
    41.                 totalPartitions,
    42.                 parent);
    43.         byte[] taskUpdateRequestJson = taskUpdateRequestCodec.toBytes(updateRequest);
    44.         if (fragment.isPresent()) {
    45.             stats.updateWithPlanBytes(taskUpdateRequestJson.length);
    46.         }
    48.         HttpUriBuilder uriBuilder = getHttpUriBuilder(taskStatus);
    49.         Request request = setContentTypeHeaders(isBinaryEncoding, preparePost())
    50.                 .setUri(uriBuilder.build())
    51.                 .setBodyGenerator(StaticBodyGenerator.createStaticBodyGenerator(taskUpdateRequestJson))
    52.                 // .setBodyGenerator(createBodyGenerator(updateRequest))
    53.                 .build();
    55.         // 是否需要进行二进制加密发送
    56.         ResponseHandler responseHandler;
    57.         if (isBinaryEncoding) {
    58.             responseHandler = createFullSmileResponseHandler((SmileCodec<TaskInfo>) taskInfoCodec);
    59.         }
    60.         else {
    61.             responseHandler = createAdaptingJsonResponseHandler(unwrapJsonCodec(taskInfoCodec));
    62.         }
    63.         updateErrorTracker.startRequest();
    65.         // 开发发送
    66.         ListenableFuture<BaseResponse<TaskInfo>> future = httpClient.executeAsync(request, responseHandler);
    67.         currentRequest = future;
    68.         currentRequestStartNanos = System.nanoTime();
    70.         // The needsUpdate flag needs to be set to false BEFORE adding the Future callback since callback might change the flag value
    71.         // and does so without grabbing the instance lock.
    72.         needsUpdate.set(false);
    74.         // 当成功结果返回后，会执行SimpleHttpResponseHandler的onSuccess， 失败会执行onFailure
    75.         Futures.addCallback(future, new SimpleHttpResponseHandler<>(new UpdateResponseHandler(sources), request.getUri(), stats), executor);
    76.     }

    image.png