Hadoop3.2.1 【 YARN 】源码分析 : ApplicationMasterService 源码浅析 [ 二 ]

我就是我 2023-01-10 10:17 181阅读 0赞

# 一. 前言 #

ApplicationMasterService在执行registerApplicationMaster , finishApplicationMaster , allocate这三个方法的时候,都调用了amsProcessingChain中对应的方法, 所以我们看来一下amsProcessingChain到底干了啥.

# 二. 数据结构 #

AMSProcessingChain 实现了ApplicationMasterServiceProcessor 接口 , 一共三个方法.

*  init 初始化
 *  registerApplicationMaster 注册ApplicationMaster
 *  allocate 分配container&心跳
 *  finishApplicationMaster 完成AM

## 2.1. 属性 ##

这里存放了就两个属性,

// 头结点
      private ApplicationMasterServiceProcessor head;
    
      // RM 上下文信息
      private RMContext rmContext;

## 2.2. 构造方法 ##

AMSProcessingChain 在构造的时候,会设置一个默认的rootProcessor . 这里会将 head指向 rootProcessor.

/** * This has to be initialized with at-least 1 Processor. * @param rootProcessor Root processor. */
      AMSProcessingChain(ApplicationMasterServiceProcessor rootProcessor) { 
        if (rootProcessor == null) { 
          throw new YarnRuntimeException("No root ApplicationMasterService" +
              "Processor specified for the processing chain..");
        }
        this.head = rootProcessor;
      }

## 2.3. init ##

ApplicationMasterService的initializeProcessingChain进行初始化操作.

在这里,会设置RM的上下文信息, 并且初始化root process .

@Override
      public void init(ApplicationMasterServiceContext amsContext,
          ApplicationMasterServiceProcessor nextProcessor) { 
        LOG.info("Initializing AMS Processing chain. Root Processor=["
            + this.head.getClass().getName() + "].");
        this.rmContext = (RMContext)amsContext;
        // The head is initialized with a null 'next' processor
        this.head.init(amsContext, null);
      }

## 2.4. addProcessor ##

这里就是根据传入的ApplicationMasterServiceProcessor 进行处理, 主要作用是构建责任链.

/** * Add an processor to the top of the chain. * @param processor ApplicationMasterServiceProcessor */
      public synchronized void addProcessor(
          ApplicationMasterServiceProcessor processor) { 
        LOG.info("Adding [" + processor.getClass().getName() + "] tp top of" +
            " AMS Processing chain. ");
        
        // 首先初始化 processor
        processor.init(this.rmContext, this.head);
        // 然后将当前的head设置为新传入的processor 
        this.head = processor;
      }

其实就是一个构建链表.在这里只是定义了一个接口, 具体如何实现都是根据实现类来控制的.

/** * Initialize with and ApplicationMasterService Context as well as the * next processor in the chain. * @param amsContext AMSContext. * @param nextProcessor next ApplicationMasterServiceProcessor */
      void init(ApplicationMasterServiceContext amsContext,  ApplicationMasterServiceProcessor nextProcessor);

## 2.5.registerApplicationMaster ##

直接调用head中的registerApplicationMaster 方法.

## 2.6.allocate ##

直接调用head中的allocate 方法 .

@Override
      public void allocate(ApplicationAttemptId appAttemptId,
          AllocateRequest request, AllocateResponse response) throws YarnException { 
        this.head.allocate(appAttemptId, request, response);
      }

## 2.7. ##

直接调用head中的finishApplicationMaster .

# 三. 构建 #

*  AMSProcessingChain 是 ApplicationMasterService中的一个属性.

// 存放AM的处理Chain
        private final AMSProcessingChain amsProcessingChain;

*  在ApplicationMasterService构造方法中进行创建

public ApplicationMasterService(String name, RMContext rmContext,
          YarnScheduler scheduler) { 
        super(name);
        this.amLivelinessMonitor = rmContext.getAMLivelinessMonitor();
        this.rScheduler = scheduler;
        this.rmContext = rmContext;
        // AMSProcessingChain通过责任链模式处理ApplicationMaster的注册
        // 责任链上的processor的头节点目前是DefaultAMSProcessor
        this.amsProcessingChain = new AMSProcessingChain(new DefaultAMSProcessor());
      }

*  在ApplicationMasterService\#serviceInit方法里面调用initializeProcessingChain进行初始化.  
    这里我们看到首先是处理的放置策略, 根据配置参数yarn.resourcemanager.placement-constraints.handler 指定.  
    然后加载配置中的其他继承AbstractPlacementProcessor的配置.

private void initializeProcessingChain(Configuration conf) { 
         // 执行root process 初始化操作
        amsProcessingChain.init(rmContext, null);
    
        // 处理放置策略, 默认 拒绝
        // yarn.resourcemanager.placement-constraints.handler : disabled
        addPlacementConstraintHandler(conf);
    
        // 从配置文件中获取 ApplicationMasterServiceProcessor 添加到amsProcessingChain 中.
        List<ApplicationMasterServiceProcessor> processors = getProcessorList(conf);
        if (processors != null) { 
          Collections.reverse(processors);
          for (ApplicationMasterServiceProcessor p : processors) { 
            // Ensure only single instance of PlacementProcessor is included
            if (p instanceof AbstractPlacementProcessor) { 
              LOG.warn("Found PlacementProcessor=" + p.getClass().getCanonicalName()
                  + " defined in "
                  + YarnConfiguration.RM_APPLICATION_MASTER_SERVICE_PROCESSORS
                  + ", however PlacementProcessor handler should be configured "
                  + "by using " + YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_HANDLER
                  + ", this processor will be ignored.");
              continue;
            }
            this.amsProcessingChain.addProcessor(p);
          }
        }
      }

# 四. DefaultAMSProcessor #

ApplicationMasterService在初始化的方法中构建amsProcessingChain . 在初始化的时候,需要传入一个root processor . 这个processor就是 DefaultAMSProcessor . 我们来看一下这个都干了啥…

## 4.1. 属性 ##

// 空的Container 集合
      private final static List<Container> EMPTY_CONTAINER_LIST =  new ArrayList<Container>();
    
      // 构建一个 空的 Allocation ??
      protected static final Allocation EMPTY_ALLOCATION = new Allocation(  EMPTY_CONTAINER_LIST, Resources.createResource(0), null, null, null);
    
      private final RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null);
    
      // RM 上下文信息
      private RMContext rmContext;
    
      // resource profiles
      private ResourceProfilesManager resourceProfilesManager;
    
      // 是否启用timelineServiceV2
      private boolean timelineServiceV2Enabled;
      // 是否启用node label
      private boolean nodelabelsEnabled;

## 4.2. init 方法 ##

init 比较简单,就是根据入参&配置文件 初始化参数.

@Override
      public void init(ApplicationMasterServiceContext amsContext, ApplicationMasterServiceProcessor nextProcessor) { 
    
        this.rmContext = (RMContext)amsContext;
    
        this.resourceProfilesManager = rmContext.getResourceProfilesManager();
    
        this.timelineServiceV2Enabled = YarnConfiguration. timelineServiceV2Enabled(rmContext.getYarnConfiguration());
        this.nodelabelsEnabled = YarnConfiguration .areNodeLabelsEnabled(rmContext.getYarnConfiguration());
      }

## 4.3. registerApplicationMaster ##

执行注册操作. 注册 采用RM 的Dispatcher , 处理RMAppAttemptRegistrationEvent 类型任务.

@Override
      public void registerApplicationMaster(
          ApplicationAttemptId applicationAttemptId,
          RegisterApplicationMasterRequest request,
          RegisterApplicationMasterResponse response)
          throws IOException, YarnException { 
    
        // 获取 application
        RMApp app = getRmContext().getRMApps().get(  applicationAttemptId.getApplicationId());
    
        //向RMAppAttemptImpl发送一个RMAppAttemptEventType.registered事件
        LOG.info("AM registration " + applicationAttemptId);
    
        // !!!! 在这里通过Dispatcher获取对应事件的Handler处理任务.!!!!!
        getRmContext().getDispatcher().getEventHandler()
            .handle(
                new RMAppAttemptRegistrationEvent(applicationAttemptId, request
                    .getHost(), request.getRpcPort(), request.getTrackingUrl()));
    
        // 日志审计
        RMAuditLogger.logSuccess(app.getUser(),
            RMAuditLogger.AuditConstants.REGISTER_AM,
            "ApplicationMasterService", app.getApplicationId(),
            applicationAttemptId);
    
        // 返回资源的最大容量
        response.setMaximumResourceCapability(getScheduler()
            .getMaximumResourceCapability(app.getQueue()));
    
        // 设置权限
        response.setApplicationACLs(app.getRMAppAttempt(applicationAttemptId)
            .getSubmissionContext().getAMContainerSpec().getApplicationACLs());
    
        // 设置队列
        response.setQueue(app.getQueue());
        if (UserGroupInformation.isSecurityEnabled()) { 
          LOG.info("Setting client token master key");
          response.setClientToAMTokenMasterKey(java.nio.ByteBuffer.wrap(
              getRmContext().getClientToAMTokenSecretManager()
              .getMasterKey(applicationAttemptId).getEncoded()));
        }
    
    
        // For work-preserving AM restart, retrieve previous attempts' containers and corresponding NM tokens.
        if (app.getApplicationSubmissionContext()  .getKeepContainersAcrossApplicationAttempts()) { 
    
          // 获取需要处理的Container
          List<Container> transferredContainers = getScheduler().getTransferredContainers(applicationAttemptId);
    
          if (!transferredContainers.isEmpty()) { 
            response.setContainersFromPreviousAttempts(transferredContainers);
            // Clear the node set remembered by the secret manager. Necessary
            // for UAM restart because we use the same attemptId.
            rmContext.getNMTokenSecretManager()
                .clearNodeSetForAttempt(applicationAttemptId);
    
            List<NMToken> nmTokens = new ArrayList<NMToken>();
            for (Container container : transferredContainers) { 
              try { 
                // 更新token操作
                NMToken token = getRmContext().getNMTokenSecretManager()
                    .createAndGetNMToken(app.getUser(), applicationAttemptId,
                        container);
                if (null != token) { 
                  nmTokens.add(token);
                }
              } catch (IllegalArgumentException e) { 
                // if it's a DNS issue, throw UnknowHostException directly and
                // that
                // will be automatically retried by RMProxy in RPC layer.
                if (e.getCause() instanceof UnknownHostException) { 
                  throw (UnknownHostException) e.getCause();
                }
              }
            }
    
            response.setNMTokensFromPreviousAttempts(nmTokens);
            LOG.info("Application " + app.getApplicationId() + " retrieved "
                + transferredContainers.size() + " containers from previous"
                + " attempts and " + nmTokens.size() + " NM tokens.");
          }
        }
    
        // 设置资源类型
        response.setSchedulerResourceTypes(getScheduler()
            .getSchedulingResourceTypes());
    
        // 设置资源类型 返回响应信息
        response.setResourceTypes(ResourceUtils.getResourcesTypeInfo());
    
    
        // yarn.resourcemanager.resource-profiles.enabled : false
        if (getRmContext().getYarnConfiguration().getBoolean(
            YarnConfiguration.RM_RESOURCE_PROFILES_ENABLED,
            YarnConfiguration.DEFAULT_RM_RESOURCE_PROFILES_ENABLED)) { 
    
          // 如果启用了, 返回信息
          response.setResourceProfiles(
              resourceProfilesManager.getResourceProfiles());
        }
      }

## 4.4.allocate ##

@Override
      public void allocate(ApplicationAttemptId appAttemptId,
          AllocateRequest request, AllocateResponse response) throws YarnException { 
        // 处理 请求
        handleProgress(appAttemptId, request);
    
        // 获取资源请求
        List<ResourceRequest> ask = request.getAskList();
    
        // 获取需要释放的 ContainerId
        List<ContainerId> release = request.getReleaseList();
    
        // 获取请求的黑名单
        ResourceBlacklistRequest blacklistRequest =  request.getResourceBlacklistRequest();
        List<String> blacklistAdditions =
            (blacklistRequest != null) ?
                blacklistRequest.getBlacklistAdditions() : Collections.emptyList();
    
        // 获取移除黑名单的数据
        List<String> blacklistRemovals =
            (blacklistRequest != null) ?
                blacklistRequest.getBlacklistRemovals() : Collections.emptyList();
    
    
        // 获取 RM application
        RMApp app =
            getRmContext().getRMApps().get(appAttemptId.getApplicationId());
    
        // set label expression for Resource Requests if resourceName=ANY
        // 根据资源请求 设置 label表达式
        ApplicationSubmissionContext asc = app.getApplicationSubmissionContext();
        for (ResourceRequest req : ask) { 
          if (null == req.getNodeLabelExpression()
              && ResourceRequest.ANY.equals(req.getResourceName())) { 
            req.setNodeLabelExpression(asc.getNodeLabelExpression());
          }
        }
    
        // 获取队列中的最大资源能力
        Resource maximumCapacity =
            getScheduler().getMaximumResourceCapability(app.getQueue());
    
        // 健全性检查
        // sanity check
        try { 
          RMServerUtils.normalizeAndValidateRequests(ask,
              maximumCapacity, app.getQueue(),
              getScheduler(), getRmContext(), nodelabelsEnabled);
        } catch (InvalidResourceRequestException e) { 
          RMAppAttempt rmAppAttempt = app.getRMAppAttempt(appAttemptId);
          handleInvalidResourceException(e, rmAppAttempt);
        }
    
        try { 
          // 验证请求
          RMServerUtils.validateBlacklistRequest(blacklistRequest);
        }  catch (InvalidResourceBlacklistRequestException e) { 
          LOG.warn("Invalid blacklist request by application " + appAttemptId, e);
          throw e;
        }
        // 在工作保持AM重启的情况下，AM有可能从先前的尝试中释放容器。
        // In the case of work-preserving AM restart, it's possible for the
        // AM to release containers from the earlier attempt.
        if (!app.getApplicationSubmissionContext()
            .getKeepContainersAcrossApplicationAttempts()) { 
          try { 
            RMServerUtils.validateContainerReleaseRequest(release, appAttemptId);
          } catch (InvalidContainerReleaseException e) { 
            LOG.warn("Invalid container release by application " + appAttemptId,
                e);
            throw e;
          }
        }
    
        // 将更新资源请求拆分为增加和减少。
        // 这里没有抛出异常。所有更新错误都将聚合并返回给AM。
        // Split Update Resource Requests into increase and decrease.
        // No Exceptions are thrown here. All update errors are aggregated
        // and returned to the AM.
        List<UpdateContainerError> updateErrors = new ArrayList<>();
        ContainerUpdates containerUpdateRequests =
            RMServerUtils.validateAndSplitUpdateResourceRequests(
                getRmContext(), request, maximumCapacity, updateErrors);
    
        // 向appAttempt 发送新的requests
        // Send new requests to appAttempt.
        Allocation allocation;
        RMAppAttemptState state =
            app.getRMAppAttempt(appAttemptId).getAppAttemptState();
    
        if (state.equals(RMAppAttemptState.FINAL_SAVING) ||
            state.equals(RMAppAttemptState.FINISHING) ||
            app.isAppFinalStateStored()) { 
          // 如果任务正处于保存中或者保存完成 , 不做操作,修改allocation状态为EMPTY_ALLOCATION
          LOG.warn(appAttemptId + " is in " + state +
              " state, ignore container allocate request.");
          allocation = EMPTY_ALLOCATION;
        } else { 
          try { 
    
            // 重点!!! 获取 YarnScheduler 处理 allocate
            allocation = getScheduler().allocate(appAttemptId, ask,
                request.getSchedulingRequests(), release,
                blacklistAdditions, blacklistRemovals, containerUpdateRequests);
          } catch (SchedulerInvalidResoureRequestException e) { 
            LOG.warn("Exceptions caught when scheduler handling requests");
            throw new YarnException(e);
          }
        }
    
        if (!blacklistAdditions.isEmpty() || !blacklistRemovals.isEmpty()) { 
          LOG.info("blacklist are updated in Scheduler." +
              "blacklistAdditions: " + blacklistAdditions + ", " +
              "blacklistRemovals: " + blacklistRemovals);
        }
        // 获取RMAppAttempt
        RMAppAttempt appAttempt = app.getRMAppAttempt(appAttemptId);
    
        if (allocation.getNMTokens() != null &&
            !allocation.getNMTokens().isEmpty()) { 
          response.setNMTokens(allocation.getNMTokens());
        }
    
        // 通知AM的container更新错误信息
        // Notify the AM of container update errors
        ApplicationMasterServiceUtils.addToUpdateContainerErrors(
            response, updateErrors);
    
        // 使用节点状态更改的增量更新响应
        // update the response with the deltas of node status changes
        handleNodeUpdates(app, response);
    
        // 添加新分配的container...
        ApplicationMasterServiceUtils.addToAllocatedContainers(
            response, allocation.getContainers());
    
        // 设置已完成Container状态
        response.setCompletedContainersStatuses(appAttempt
            .pullJustFinishedContainers());
        // 设置可用资源
        response.setAvailableResources(allocation.getResourceLimit());
    
        // 设置Container变更
        addToContainerUpdates(response, allocation,
            ((AbstractYarnScheduler)getScheduler())
                .getApplicationAttempt(appAttemptId).pullUpdateContainerErrors());
    
        // 设置集群节点数量
        response.setNumClusterNodes(getScheduler().getNumClusterNodes());
    
        // add collector address for this application
        if (timelineServiceV2Enabled) { 
          CollectorInfo collectorInfo = app.getCollectorInfo();
          if (collectorInfo != null) { 
            response.setCollectorInfo(collectorInfo);
          }
        }
    
        // 向allocateResponse消息添加抢占（如果有）
        // add preemption to the allocateResponse message (if any)
        response.setPreemptionMessage(generatePreemptionMessage(allocation));
    
        // 设置application 优先级
        // Set application priority
        response.setApplicationPriority(app
            .getApplicationPriority());
    
        response.setContainersFromPreviousAttempts(
            allocation.getPreviousAttemptContainers());
      }

## 4.5. finishApplicationMaster ##

直接使用Dispatcher处理RMAppAttemptUnregistrationEvent事件 .

@Override
      public void finishApplicationMaster(
          ApplicationAttemptId applicationAttemptId,
          FinishApplicationMasterRequest request,
          FinishApplicationMasterResponse response) { 
    
        // 获取application
        RMApp app =
            getRmContext().getRMApps().get(applicationAttemptId.getApplicationId());
        // For UnmanagedAMs, return true so they don't retry
        response.setIsUnregistered(
                app.getApplicationSubmissionContext().getUnmanagedAM());
    
        // 通过Dispatcher处理对应的时间
        getRmContext().getDispatcher().getEventHandler().handle(
            new RMAppAttemptUnregistrationEvent(applicationAttemptId, request
                .getTrackingUrl(), request.getFinalApplicationStatus(), request
                .getDiagnostics()));
      }

# 五. ApplicationMasterServiceProcessor 实现类 #

amsProcessingChain 是一个链表, 所以有在构建的时候,可以通过 addProcessor 构建责任链. 我们看一下ApplicationMasterServiceProcessor的七个 实现.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

## 5.1. AbstractPlacementProcessor ##

AbstractPlacementProcessor 是一个抽象类. 他有三个实现类.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

*  属性  
    主要有三个属性. 因为是一个链表 , 核心的其实就nextAMSProcessor .

// 下一节点.
      protected ApplicationMasterServiceProcessor nextAMSProcessor;
      // yarn 调度器
      protected AbstractYarnScheduler scheduler;
    
      // 用于存储和检索放置约束的管理器 ??
      private PlacementConstraintManager constraintManager;

*  registerApplicationMaster  
    会向下传递,调用next的registerApplicationMaster 方法

@Override
      public void registerApplicationMaster(
          ApplicationAttemptId applicationAttemptId,
          RegisterApplicationMasterRequest request,
          RegisterApplicationMasterResponse response)
          throws IOException, YarnException { 
        Map<Set<String>, PlacementConstraint> appPlacementConstraints =
            request.getPlacementConstraints();
        processPlacementConstraints(applicationAttemptId.getApplicationId(),
            appPlacementConstraints);
        //重要 调用next的registerApplicationMaster 方法
        nextAMSProcessor.registerApplicationMaster(applicationAttemptId, request,
            response);
      }

*  finishApplicationMaster  
    会向下传递,调用next的finishApplicationMaster 方法

@Override
      public void finishApplicationMaster(ApplicationAttemptId applicationAttemptId,
          FinishApplicationMasterRequest request,
          FinishApplicationMasterResponse response) { 
        constraintManager.unregisterApplication(
            applicationAttemptId.getApplicationId());
        this.nextAMSProcessor.finishApplicationMaster(applicationAttemptId, request,
            response);
      }

## 5.2. AbstractPlacementProcessor 的实现类 ##

AbstractPlacementProcessor有三个实现类DisabledPlacementProcessor , PlacementConstraintProceessor , SchedulerPlacementProcessor . 分别对应三个策略 , 这三个策略都是由参数yarn.resourcemanager.placement-constraints.handler 控制.

*  placement-processor:\[PlacementConstraintProcessor\]如果使用放置处理器，则被认为是调度器被调用前的一个前置操作。一旦放置被决定后，调度器被调起去执行真正的分配。这个处理器的好处是支持各种约束（affinity, anti-affinity, cardinality）。因为独立于调度器，所以它可以被多个调度器调用。
 *  scheduler: \[SchedulerPlacementProcessor \]直接使用调度器作为处理器，目前，只有容量调度器支持SchedulingRequests。支持反亲和性约束。is that it follows the same ordering rules for queues (sorted by utilization, priority), apps (sorted by FIFO/fairness/priority) and tasks within the same app (priority) that are enforced by the existing main scheduler。
 *  disabled: \[DisabledPlacementProcessor\]如果使用了SchedulingRequest，则会被拒绝

## 5.3. DisabledPlacementProcessor ##

如果使用了SchedulingRequest，则会被拒绝

### 5.3.1. registerApplicationMaster ###

@Override
      public void registerApplicationMaster(
          ApplicationAttemptId applicationAttemptId,
          RegisterApplicationMasterRequest request,
          RegisterApplicationMasterResponse response)
          throws IOException, YarnException { 
    
        // 如果放置策略不为null 直接直接拒绝...
        if (request.getPlacementConstraints() != null && !request
            .getPlacementConstraints().isEmpty()) { 
          String message = "Found non empty placement constraints map in "
              + "RegisterApplicationMasterRequest for application="
              + applicationAttemptId.toString() + ", but the configured "
              + YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_HANDLER
              + " cannot handle placement constraints. Rejecting this "
              + "registerApplicationMaster operation";
          LOG.warn(message);
          throw new YarnException(message);
        }
        // 向下传递
        nextAMSProcessor.registerApplicationMaster(applicationAttemptId, request,
            response);
      }

### 5.3.2. allocate ###

@Override
      public void allocate(ApplicationAttemptId appAttemptId,
          AllocateRequest request, AllocateResponse response) throws YarnException { 
    
        // 如果放置策略不为null 直接直接拒绝...
        if (request.getSchedulingRequests() != null && !request
            .getSchedulingRequests().isEmpty()) { 
          String message = "Found non empty SchedulingRequest in "
              + "AllocateRequest for application="
              + appAttemptId.toString() + ", but the configured "
              + YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_HANDLER
              + " cannot handle placement constraints. Rejecting this "
              + "allocate operation";
          LOG.warn(message);
          throw new YarnException(message);
        }
        // 向下传递
        nextAMSProcessor.allocate(appAttemptId, request, response);
      }

## 5.4. SchedulerPlacementProcessor ##

*  scheduler: \[SchedulerPlacementProcessor \]直接使用调度器作为处理器，目前，只有容量调度器支持SchedulingRequests。支持反亲和性约束。is that it follows the same ordering rules for queues (sorted by utilization, priority), apps (sorted by FIFO/fairness/priority) and tasks within the same app (priority) that are enforced by the existing main scheduler。

@Override
      public void allocate(ApplicationAttemptId appAttemptId,
          AllocateRequest request, AllocateResponse response) throws YarnException { 
        if (request.getSchedulingRequests() != null
            && !request.getSchedulingRequests().isEmpty()) { 
    
          // 仅仅容量调度器支持该策略
          if (!(scheduler instanceof CapacityScheduler)) { 
            String message = "Found non empty SchedulingRequest of "
                + "AllocateRequest for application=" + appAttemptId.toString()
                + ", however the configured scheduler="
                + scheduler.getClass().getCanonicalName()
                + " cannot handle placement constraints, rejecting this "
                + "allocate operation";
            LOG.warn(message);
            throw new YarnException(message);
          }
        }
        //向下传递
        nextAMSProcessor.allocate(appAttemptId, request, response);
      }

## 5.5. PlacementConstraintProcessor ##

如果使用放置处理器，则被认为是调度器被调用前的一个前置操作。一旦放置被决定后，调度器被调起去执行真正的分配。这个处理器的好处是支持各种约束（affinity, anti-affinity, cardinality）。因为独立于调度器，所以它可以被多个调度器调用。

> 一种ApplicationMasterServiceProcessor，它执行Constrained 放置请求。  
> 它执行以下操作：
> 
> 1.  所有初始化。  
>     2.从寄存器调用截取放置约束并将其添加到放置约束管理器。  
>     3.将调度请求分派给计划器。

### 5.5.1. 属性 ###

// 调度线程池
      private ExecutorService schedulingThreadPool;
    
      // 重试次数,默认 3
      private int retryAttempts;
    
      // 重试请求.
      private Map<ApplicationId, List<BatchedRequests>> requestsToRetry = new ConcurrentHashMap<>();
    
      // 拒绝请求.
      private Map<ApplicationId, List<SchedulingRequest>> requestsToReject =   new ConcurrentHashMap<>();
    
      // 类型 串行? or 热门标签 ??
      private BatchedRequests.IteratorType iteratorType;
    
      // Dispatcher
      private PlacementDispatcher placementDispatcher;

### 5.5.2. init ###

初始化方法 , 创建自己的分配算法实例/类型/调度器器Dispatcher/线程池/重试次数.

@Override
      public void init(ApplicationMasterServiceContext amsContext,   ApplicationMasterServiceProcessor nextProcessor) { 
        LOG.info("Initializing Constraint Placement Processor:");
        super.init(amsContext, nextProcessor);
        //考虑第一个类-即使提供了逗号分隔的列表。
        // （这是为了简单起见，因为getInstances通过正确处理事情做了很多好事）
    
        // Only the first class is considered - even if a comma separated list is provided.
        // (This is for simplicity, since getInstances does a lot of good things by handling things correctly)
    
        // 根据配置参数,获取 ConstraintPlacementAlgorithm
        // yarn.resourcemanager.placement-constraints.algorithm.class
        List<ConstraintPlacementAlgorithm> instances =
            ((RMContextImpl) amsContext).getYarnConfiguration().getInstances(
                YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_ALGORITHM_CLASS,
                ConstraintPlacementAlgorithm.class);
    
        ConstraintPlacementAlgorithm algorithm = null;
        if (instances != null && !instances.isEmpty()) { 
          algorithm = instances.get(0);
        } else { 
          algorithm = new DefaultPlacementAlgorithm();
        }
        LOG.info("Placement Algorithm [{}]", algorithm.getClass().getName());
    
        // 获取 yarn.resourcemanager.placement-constraints.algorithm.iterator
        String iteratorName = ((RMContextImpl) amsContext).getYarnConfiguration()
            .get(YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_ALGORITHM_ITERATOR,
                BatchedRequests.IteratorType.SERIAL.name());
        LOG.info("Placement Algorithm Iterator[{}]", iteratorName);
        try { 
          // 获取类型
          iteratorType = BatchedRequests.IteratorType.valueOf(iteratorName);
        } catch (IllegalArgumentException e) { 
          throw new YarnRuntimeException(
              "Could not instantiate Placement Algorithm Iterator: ", e);
        }
    
        // 算法池 大小 : 1
        // yarn.resourcemanager.placement-constraints.algorithm.pool-size : 1
        int algoPSize = ((RMContextImpl) amsContext).getYarnConfiguration().getInt(
            YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_ALGORITHM_POOL_SIZE,
            YarnConfiguration.DEFAULT_RM_PLACEMENT_CONSTRAINTS_ALGORITHM_POOL_SIZE);
    
        // 构建Dispatcher
        this.placementDispatcher = new PlacementDispatcher();
    
        // 初始化Dispatcher
        this.placementDispatcher.init(
            ((RMContextImpl)amsContext), algorithm, algoPSize);
        LOG.info("Planning Algorithm pool size [{}]", algoPSize);
    
    
        // yarn.resourcemanager.placement-constraints.scheduler.pool-size : 1
        int schedPSize = ((RMContextImpl) amsContext).getYarnConfiguration().getInt(
            YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_SCHEDULER_POOL_SIZE,
            YarnConfiguration.DEFAULT_RM_PLACEMENT_CONSTRAINTS_SCHEDULER_POOL_SIZE);
    
        // 创建线程池
        this.schedulingThreadPool = Executors.newFixedThreadPool(schedPSize);
        LOG.info("Scheduler pool size [{}]", schedPSize);
    
        // yarn.resourcemanager.设置重试次数
        // placement-constraints.retry-attempts: 3
        // Number of times a request that is not satisfied by the scheduler can be retried.
        this.retryAttempts =
            ((RMContextImpl) amsContext).getYarnConfiguration().getInt(
                YarnConfiguration.RM_PLACEMENT_CONSTRAINTS_RETRY_ATTEMPTS,
                YarnConfiguration.DEFAULT_RM_PLACEMENT_CONSTRAINTS_RETRY_ATTEMPTS);
        LOG.info("Num retry attempts [{}]", this.retryAttempts);
      }

### 5.5.3. allocate ###

@Override
      public void allocate(ApplicationAttemptId appAttemptId,
          AllocateRequest request, AllocateResponse response) throws YarnException { 
        // Copy the scheduling request since we will clear it later after sending to dispatcher
        // 复制调度请求，因为我们稍后将其发送给调度程序后将其清除
        List<SchedulingRequest> schedulingRequests = new ArrayList<>(request.getSchedulingRequests());
    
        // 派送 放置请求.
        dispatchRequestsForPlacement(appAttemptId, schedulingRequests);
    
        // 分配重试请求.
        reDispatchRetryableRequests(appAttemptId);
    
        // 处理已经处理的请求.
        schedulePlacedRequests(appAttemptId);
    
        // Remove SchedulingRequest from AllocateRequest to avoid SchedulingRequest added to scheduler.
        // 移除调度请求.
        request.setSchedulingRequests(Collections.emptyList());
    
        // 执行 下一请求.
        nextAMSProcessor.allocate(appAttemptId, request, response);
    
        // 处理拒绝请求.
        handleRejectedRequests(appAttemptId, response);
      }

这里方法比较多. 一条条的看.

*  派送 放置请求.

private void dispatchRequestsForPlacement(ApplicationAttemptId appAttemptId,  List<SchedulingRequest> schedulingRequests) { 
    
        if (schedulingRequests != null && !schedulingRequests.isEmpty()) { 
    
          // 获取 调度请求.
          SchedulerApplicationAttempt appAttempt =  scheduler.getApplicationAttempt(appAttemptId);
          String queueName = null;
    
          if(appAttempt != null) { 
            queueName = appAttempt.getQueueName();
          }
    
          // 获取队列最大资源
          Resource maxAllocation =  scheduler.getMaximumResourceCapability(queueName);
          // Normalize the Requests before dispatching
    
          // 处理请求信息? 干啥这是???
          schedulingRequests.forEach(req -> { 
            // 获取请求的资源
            Resource reqResource = req.getResourceSizing().getResources();
            // 重新设置资源
            req.getResourceSizing().setResources(
                this.scheduler.getNormalizedResource(reqResource, maxAllocation));
          });
    
          // 处理分发请求.
          this.placementDispatcher.dispatch(new BatchedRequests(iteratorType,
              appAttemptId.getApplicationId(), schedulingRequests, 1));
        }
      }

*  reDispatchRetryableRequests 分配重试请求.

private void reDispatchRetryableRequests(ApplicationAttemptId appAttId) { 
        // 获取重试
        List<BatchedRequests> reqsToRetry = this.requestsToRetry.get(appAttId.getApplicationId());
        if (reqsToRetry != null && !reqsToRetry.isEmpty()) { 
          synchronized (reqsToRetry) { 
            for (BatchedRequests bReq: reqsToRetry) { 
              // 重新分发
              this.placementDispatcher.dispatch(bReq);
            }
            reqsToRetry.clear();
          }
        }
      }

*  schedulePlacedRequests 调度放置请求.

private void schedulePlacedRequests(ApplicationAttemptId appAttemptId) { 
        // 获取applicationId
        ApplicationId applicationId = appAttemptId.getApplicationId();
        // 拉取请求
        List<PlacedSchedulingRequest> placedSchedulingRequests = this.placementDispatcher.pullPlacedRequests(applicationId);
    
        for (PlacedSchedulingRequest placedReq : placedSchedulingRequests) { 
          // 获取请求信息
          SchedulingRequest sReq = placedReq.getSchedulingRequest();
          // 获取请求的节点
          for (SchedulerNode node : placedReq.getNodes()) { 
    
            // 获取调度请求
            final SchedulingRequest sReqClone =
                SchedulingRequest.newInstance(sReq.getAllocationRequestId(),
                    sReq.getPriority(), sReq.getExecutionType(),
                    sReq.getAllocationTags(),
                    ResourceSizing.newInstance(
                        sReq.getResourceSizing().getResources()),
                    sReq.getPlacementConstraint());
    
            // 获取SchedulerApplicationAttempt
            SchedulerApplicationAttempt applicationAttempt =
                this.scheduler.getApplicationAttempt(appAttemptId);
    
            Runnable task = () -> { 
              // 尝试分配
              boolean success =
                  scheduler.attemptAllocationOnNode(applicationAttempt, sReqClone, node);
              if (!success) { 
                LOG.warn("Unsuccessful allocation attempt [{}] for [{}]",
                    placedReq.getPlacementAttempt(), sReqClone);
              }
              // 处理响应
              handleSchedulingResponse(
                  new Response(success, applicationId, sReqClone,
                  placedReq.getPlacementAttempt(), node));
            };
            // 提交任务...交由线程池执行...
            this.schedulingThreadPool.submit(task);
          }
        }
      }

### 5.5.4. finishApplicationMaster ###

其实就是做各种清理操作.

@Override
      public void finishApplicationMaster(ApplicationAttemptId appAttemptId,
          FinishApplicationMasterRequest request,
          FinishApplicationMasterResponse response) { 
    
        // 清理各种遗留信息
        placementDispatcher.clearApplicationState(appAttemptId.getApplicationId());
        requestsToReject.remove(appAttemptId.getApplicationId());
        requestsToRetry.remove(appAttemptId.getApplicationId());
    
        // 调用父类的请求.
        super.finishApplicationMaster(appAttemptId, request, response);
      }

## 5.6. PlacementDispatcher ##

此类初始化约束放置算法。它向算法分配输入并从中收集输出。

### 5.6.1. 常量 ###

// 分配算法
      private ConstraintPlacementAlgorithm algorithm;
      // 线程池
      private ExecutorService algorithmThreadPool;
    
      // 放置请求
      private Map<ApplicationId, List<PlacedSchedulingRequest>>  placedRequests = new ConcurrentHashMap<>();
      
      // 拒绝请求.
      private Map<ApplicationId, List<SchedulingRequestWithPlacementAttempt>> rejectedRequests = new ConcurrentHashMap<>();

### 5.6.2. init ###

初始化操作.

public void init(RMContext rmContext,
          ConstraintPlacementAlgorithm placementAlgorithm, int poolSize) { 
        LOG.info("Initializing Constraint Placement Planner:");
        this.algorithm = placementAlgorithm;
        this.algorithm.init(rmContext);
        this.algorithmThreadPool = Executors.newFixedThreadPool(poolSize);
      }

### 5.6.3. dispatch ###

根据请求,构建任务, 然后交由线程池执行.

void dispatch(final BatchedRequests batchedRequests) { 
        final ConstraintPlacementAlgorithmOutputCollector collector = this;
    
        //构建任务 , 使用分配算法处理请求数据
        Runnable placingTask = () -> { 
          LOG.debug("Got [{}] requests to place from application [{}].. " +
                  "Attempt count [{}]",
              batchedRequests.getSchedulingRequests().size(),
              batchedRequests.getApplicationId(),
              batchedRequests.getPlacementAttempt());
          algorithm.place(batchedRequests, collector);
        };
        // 交由线程池执行分配任务.
        this.algorithmThreadPool.submit(placingTask);
      }

### 5.6.4. pullPlacedRequests ###

根据获取请求applicationId获取调度请求. 将缓存中的数据clone , 然后清理掉.

// 根据获取请求applicationId获取调度请求. 将缓存中的数据clone , 然后清理掉.
      public List<PlacedSchedulingRequest> pullPlacedRequests(
          ApplicationId applicationId) { 
        List<PlacedSchedulingRequest> placedReqs = this.placedRequests.get(applicationId);
        
        if (placedReqs != null && !placedReqs.isEmpty()) { 
          List<PlacedSchedulingRequest> retList = new ArrayList<>();
          synchronized (placedReqs) { 
            if (placedReqs.size() > 0) { 
              retList.addAll(placedReqs);
              placedReqs.clear();
            }
          }
          return retList;
        }
        return Collections.emptyList();
      }

### 5.6.5. pullRejectedRequests ###

public List<SchedulingRequestWithPlacementAttempt> pullRejectedRequests(
          ApplicationId applicationId) { 
    
        // 获取拒绝的数据.
        List<SchedulingRequestWithPlacementAttempt> rejectedReqs =
            this.rejectedRequests.get(applicationId);
        if (rejectedReqs != null && !rejectedReqs.isEmpty()) { 
          List<SchedulingRequestWithPlacementAttempt> retList = new ArrayList<>();
          // 加锁,返回新的retList
          synchronized (rejectedReqs) { 
            if (rejectedReqs.size() > 0) { 
              retList.addAll(rejectedReqs);
              rejectedReqs.clear();
            }
          }
          return retList;
        }
        return Collections.emptyList();
      }

## 5.7 DefaultPlacementAlgorithm ##

### 5.7.1. 属性 ###

// 标签管理
      private LocalAllocationTagsManager tagsManager;
    
      // 放置
      private PlacementConstraintManager constraintManager;
    
      // node 选择器
      private NodeCandidateSelector nodeSelector;
    
      // 资源计算器
      private ResourceCalculator resourceCalculator;

### 5.7.2. init ###

执行初始化操作.

@Override
      public void init(RMContext rmContext) { 
        // 初始化
        this.tagsManager = new LocalAllocationTagsManager(
            rmContext.getAllocationTagsManager());
        
        this.constraintManager = rmContext.getPlacementConstraintManager();
        
        this.resourceCalculator = rmContext.getScheduler().getResourceCalculator();
        
        this.nodeSelector =
            filter -> ((AbstractYarnScheduler) (rmContext).getScheduler())
                .getNodes(filter);
      }

### 5.7.3.place 放置方法 ###

@Override
      public void place(ConstraintPlacementAlgorithmInput input, ConstraintPlacementAlgorithmOutputCollector collector) { 
        BatchedRequests requests = (BatchedRequests) input;
        
        // 请求次数
        int placementAttempt = requests.getPlacementAttempt();
        
        
        // 构建输出信息
        ConstraintPlacementAlgorithmOutput resp =  new ConstraintPlacementAlgorithmOutput(requests.getApplicationId());
       
        // 获取SchedulerNode
        List<SchedulerNode> allNodes = nodeSelector.selectNodes(null);
    
        // 构建拒绝请求
        List<SchedulingRequest> rejectedRequests = new ArrayList<>();
        
        // 可用资源清单
        Map<NodeId, Resource> availResources = new HashMap<>();
        
        // 重试次数: 2
        int rePlacementCount = RE_ATTEMPT_COUNT;
        
        while (rePlacementCount > 0) { 
          // 执行分配.
          doPlacement(requests, resp, allNodes, rejectedRequests, availResources);
          // Double check if placement constraints are really satisfied
          // 双重检查 , 放置位置是否满意.
          validatePlacement(requests.getApplicationId(), resp, rejectedRequests, availResources);
    
          // 如果拒绝 1, 2 次 可以直接重试..
          if (rejectedRequests.size() == 0 || rePlacementCount == 1) { 
            break;
          }
    
          // 分配成功, 反馈请求.
          requests = new BatchedRequests(requests.getIteratorType(),
              requests.getApplicationId(), rejectedRequests,
              requests.getPlacementAttempt());
          rejectedRequests = new ArrayList<>();
          
          // 数量 -1 
          rePlacementCount--;
        }
    
        // 处理拒绝请求.
        resp.getRejectedRequests().addAll(
            rejectedRequests.stream().map(
                x -> new SchedulingRequestWithPlacementAttempt(
                    placementAttempt, x)).collect(Collectors.toList()));
        
        collector.collect(resp);
        // 请求当前container tags
        // Clean current temp-container tags
        this.tagsManager.cleanTempContainers(requests.getApplicationId());
      }

### 5.7.4.doPlacement ###

执行分配…

private void doPlacement(BatchedRequests requests,
          ConstraintPlacementAlgorithmOutput resp,
          List<SchedulerNode> allNodes,
          List<SchedulingRequest> rejectedRequests,
          Map<NodeId, Resource> availableResources) { 
    
        // 获取调度请求.
        Iterator<SchedulingRequest> requestIterator = requests.iterator();
    
        // 获取所有的可用node
        Iterator<SchedulerNode> nIter = allNodes.iterator();
        SchedulerNode lastSatisfiedNode = null;
    
        // 循环分配
        while (requestIterator.hasNext()) { 
    
          // 验证资源是否可用...
          if (allNodes.isEmpty()) { 
            LOG.warn("No nodes available for placement at the moment !!");
            break;
          }
          // 获取分配请求.
          SchedulingRequest schedulingRequest = requestIterator.next();
    
          // 构建放置请求...
          PlacedSchedulingRequest placedReq =  new PlacedSchedulingRequest(schedulingRequest);
          placedReq.setPlacementAttempt(requests.getPlacementAttempt());
          resp.getPlacedRequests().add(placedReq);
    
          // 迭代检查其..
          CircularIterator<SchedulerNode> nodeIter = new CircularIterator(lastSatisfiedNode, nIter, allNodes);
          // 获取分配数量
          int numAllocs =  schedulingRequest.getResourceSizing().getNumAllocations();
    
          // 循环node
          while (nodeIter.hasNext() && numAllocs > 0) { 
            SchedulerNode node = nodeIter.next();
            try { 
              // 获取tag
              String tag = schedulingRequest.getAllocationTags() == null ? "" :
                  schedulingRequest.getAllocationTags().iterator().next();
    
              // 获取资源
              Resource unallocatedResource =
                  availableResources.computeIfAbsent(node.getNodeID(),
                      x -> Resource.newInstance(node.getUnallocatedResource()));
    
    
              if (!requests.getBlacklist(tag).contains(node.getNodeID()) &&
                  attemptPlacementOnNode(
                      requests.getApplicationId(), unallocatedResource,
                      schedulingRequest, node, false)) { 
    
                // 开始分配资源
                schedulingRequest.getResourceSizing()
                    .setNumAllocations(--numAllocs);
    
                Resources.addTo(unallocatedResource,
                    schedulingRequest.getResourceSizing().getResources());
                placedReq.getNodes().add(node);
                numAllocs =
                    schedulingRequest.getResourceSizing().getNumAllocations();
                // Add temp-container tags for current placement cycle
                // 为当前的放置循环添加临时的 tag
                this.tagsManager.addTempTags(node.getNodeID(),
                    requests.getApplicationId(),
                    schedulingRequest.getAllocationTags());
    
                lastSatisfiedNode = node;
              }
            } catch (InvalidAllocationTagsQueryException e) { 
              LOG.warn("Got exception from TagManager !", e);
            }
          }
        }
        // 将numAllocations仍大于0的所有请求添加到拒绝列表中。
        // Add all requests whose numAllocations still > 0 to rejected list.
        requests.getSchedulingRequests().stream()
            .filter(sReq -> sReq.getResourceSizing().getNumAllocations() > 0)
            .forEach(rejReq -> rejectedRequests.add(cloneReq(rejReq)));
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221119/6bd5bacf1bf4419aafa272b2d35d5fff.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221119/36e21ea14a9a407e9e6d568b1812cf57.png