Hadoop3.2.1 【 YARN 】源码分析 : ResourceLocalizationService解析

谁践踏了优雅 2023-01-08 13:27 95阅读 0赞

# 一 .前言 #

负责Container所需资源的本地化。 它能够按照描述从HDFS上下载Container所需的文件资源， 并尽量将它们分摊到各个磁盘上以防止出现访问热点。 此外， 它会为下载的文件添加访问控制限制， 并为之施加合适的磁盘空间使用份额。

YARN将资源分为PUBLIC、 PRIVATE和APPLICATION三类， 不同级别的资源对不同用户和应用程序的访问权限不同， 这也直接导致资源的本地化方式不同。 尽管它们的本地化由ResourceLocalizationService服务完成， 但内部由不同线程负载下载， 即所有应用程序的PUBLIC资源由ResourceLocalizationService内部专门的线程PublicLocalizer下载完成， 该线程内部维护了一个线程池以加快资源下载速度， 而每个应用程序的每种PRIVATE或APPLICATION资源分别由一个专门的线程LocalizerRunner负责下载。 该线程将启动一个资源下载客户端ContainerLocalizer， 该客户端通过RPC协议LocalizationProtocol访问ResourceLocalizationService服务， 以获取待下载的资源。

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

# 二 .LocalizationProtocol 协议 #

LocalizationProtocol 是RPC协议 , 里面只有一个heartbeat 方法.

public interface LocalizationProtocol { 
      public LocalizerHeartbeatResponse heartbeat(LocalizerStatus status)
          throws YarnException, IOException;
    }

# 三 .属性 #

// 私有目录
      public static final String NM_PRIVATE_DIR = "nmPrivate";
    
      // 私有目录权限 700
      public static final FsPermission NM_PRIVATE_PERM = new FsPermission((short) 0700);
    
      // 共有目录权限 755
      private static final FsPermission PUBLIC_FILECACHE_FOLDER_PERMS = new FsPermission((short) 0755);
    
      // 0.0.0.0 : 8040
      private Server server;
      // BoYi-Pro.local/192.168.8.188:8040
      private InetSocketAddress localizationServerAddress;
    
      @VisibleForTesting
      // 缓存大小 : 10G
      long cacheTargetSize;
      // 缓存清理周期 10min
      private long cacheCleanupPeriod;
      // DefaultContainerExecutor or LinuxContainerExecutor
      private final ContainerExecutor exec;
      // AsyncDispatcher
      protected final Dispatcher dispatcher;
      // org.apache.hadoop.yarn.server.nodemanager.DeletionService
      private final DeletionService delService;
    
    
      // Service org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ResourceLocalizationService$LocalizerTracker
      private LocalizerTracker localizerTracker;
    
      private RecordFactory recordFactory;
    
      // org.apache.hadoop.util.concurrent.HadoopScheduledThreadPoolExecutor@24c4ddae[Running, pool size = 1, active threads = 0, queued tasks = 1, completed tasks = 0]
      private final ScheduledExecutorService cacheCleanup;
      // 权限相关
      private LocalizerTokenSecretManager secretManager;
      // 持久化存储
      private NMStateStoreService stateStore;
    
    
      @VisibleForTesting
      // 度量数据
      final NodeManagerMetrics metrics;
    
      @VisibleForTesting
      // LocalResourcesTrackerImpl
      LocalResourcesTracker publicRsrc;
    
      private LocalDirsHandlerService dirsHandler;
    
      private DirsChangeListener localDirsChangeListener;
    
      private DirsChangeListener logDirsChangeListener;
    
    
      // NodeManager#NMContext 信息
      //
      // nodeId = {NodeIdPBImpl@3904} "192.168.8.188:57344"
      // conf = {YarnConfiguration@3079} "Configuration: core-default.xml, core-site.xml, yarn-default.xml, yarn-site.xml, resource-types.xml"
      // metrics = {NodeManagerMetrics@2120}
      // applications = {ConcurrentHashMap@3905} size = 0
      // systemCredentials = {HashMap@3906} size = 0
      // containers = {ConcurrentSkipListMap@3907} size = 0
      // registeringCollectors = null
      // knownCollectors = null
      // increasedContainers = {ConcurrentHashMap@3908} size = 0
      // containerTokenSecretManager = {NMContainerTokenSecretManager@3909}
      // nmTokenSecretManager = {NMTokenSecretManagerInNM@3910}
      // containerManager = {ContainerManagerImpl@2592} "Service org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManagerImpl in state org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManagerImpl: STARTED"
      // nodeResourceMonitor = {NodeResourceMonitorImpl@3911} "Service org.apache.hadoop.yarn.server.nodemanager.NodeResourceMonitorImpl in state org.apache.hadoop.yarn.server.nodemanager.NodeResourceMonitorImpl: STARTED"
      // dirsHandler = {LocalDirsHandlerService@2118} "Service org.apache.hadoop.yarn.server.nodemanager.LocalDirsHandlerService in state org.apache.hadoop.yarn.server.nodemanager.LocalDirsHandlerService: STARTED"
      // aclsManager = {ApplicationACLsManager@3912}
      // webServer = {WebServer@3913} "Service org.apache.hadoop.yarn.server.nodemanager.webapp.WebServer in state org.apache.hadoop.yarn.server.nodemanager.webapp.WebServer: INITED"
      // nodeHealthStatus = {NodeHealthStatusPBImpl@3914} "is_node_healthy: true health_report: "Healthy" last_health_report_time: 1612342130099"
      // stateStore = {NMNullStateStoreService@2122} "Service org.apache.hadoop.yarn.server.nodemanager.recovery.NMNullStateStoreService in state org.apache.hadoop.yarn.server.nodemanager.recovery.NMNullStateStoreService: STARTED"
      // isDecommissioned = false
      // logAggregationReportForApps = {ConcurrentLinkedQueue@3915} size = 0
      // nodeStatusUpdater = {NodeStatusUpdaterImpl@3832} "Service org.apache.hadoop.yarn.server.nodemanager.NodeStatusUpdaterImpl in state org.apache.hadoop.yarn.server.nodemanager.NodeStatusUpdaterImpl: INITED"
      // isDistSchedulingEnabled = false
      // deletionService = {DeletionService@2117} "Service org.apache.hadoop.yarn.server.nodemanager.DeletionService in state org.apache.hadoop.yarn.server.nodemanager.DeletionService: STARTED"
      // containerAllocator = {OpportunisticContainerAllocator@3916}
      // executor = {DefaultContainerExecutor@2115}
      // nmTimelinePublisher = null
      // containerStateTransitionListener = {NodeManager$DefaultContainerStateListener@3917}
      // resourcePluginManager = {ResourcePluginManager@3918}
      // nmLogAggregationStatusTracker = {NMLogAggregationStatusTracker@3919} "Service org.apache.hadoop.yarn.server.nodemanager.logaggregation.tracker.NMLogAggregationStatusTracker in state org.apache.hadoop.yarn.server.nodemanager.logaggregation.tracker.NMLogAggregationStatusTracker: INITED"
      private Context nmContext;
    
      // BasicDiskValidator
      private DiskValidator diskValidator;
    
      /** * 每个用户一个 LocalResourcesTracker ?????? * 用户名 -> LocalResourcesTracker * Map of LocalResourceTrackers keyed by username, for private resources. */
      @VisibleForTesting
      final ConcurrentMap<String, LocalResourcesTracker> privateRsrc =  new ConcurrentHashMap<String,LocalResourcesTracker>();
    
      /** * appid -> LocalResourcesTracker ?????? * Map of LocalResourceTrackers keyed by appid, for application resources. */
      private final ConcurrentMap<String,LocalResourcesTracker> appRsrc = new ConcurrentHashMap<String,LocalResourcesTracker>();
    
      // FileContext 中的信息
      //
      // defaultFS = {LocalFs@3928}
      // workingDir = {Path@3929} "file:/opt/workspace/apache/hadoop-3.2.1-src/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager"
      // umask = {FsPermission@3930} "----w--w-"
      // conf = {YarnConfiguration@3079} "Configuration: core-default.xml, core-site.xml, yarn-default.xml, yarn-site.xml, resource-types.xml"
      // ugi = {UserGroupInformation@3931} "henghe (auth:SIMPLE)"
      // resolveSymlinks = true
      // tracer = {Tracer@3932} "Tracer(FSClient/192.168.8.188)"
      // util = {FileContext$Util@3933}
      FileContext lfs;

# 四 .构造方法 #

ResourceLocalizationService 的实例对象是由ContainerManagerImpl\#createResourceLocalizationService方法构建的.  
这里面除了赋值.操作之外,还构建了一个HadoopScheduledThreadPoolExecutor 用于清理无效的缓存.

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

public ResourceLocalizationService(Dispatcher dispatcher,
          ContainerExecutor exec, DeletionService delService,
          LocalDirsHandlerService dirsHandler, Context context,
          NodeManagerMetrics metrics) { 
    
        super(ResourceLocalizationService.class.getName());
        // DefaultContainerExecutor or LinuxContainerExecutor
        this.exec = exec;
        // AsyncDispatcher
        this.dispatcher = dispatcher;
        // org.apache.hadoop.yarn.server.nodemanager.DeletionService
        this.delService = delService;
        this.dirsHandler = dirsHandler;
    
        // org.apache.hadoop.util.concurrent.HadoopScheduledThreadPoolExecutor@24c4ddae[Running, pool size = 1, active threads = 0, queued tasks = 1, completed tasks = 0]
        this.cacheCleanup = new HadoopScheduledThreadPoolExecutor(1,
            new ThreadFactoryBuilder()
              .setNameFormat("ResourceLocalizationService Cache Cleanup")
              .build());
    
        this.stateStore = context.getNMStateStore();
    
    
        // NodeManager#NMContext 信息
        //
        // nodeId = {NodeIdPBImpl@3904} "192.168.8.188:57344"
        // conf = {YarnConfiguration@3079} "Configuration: core-default.xml, core-site.xml, yarn-default.xml, yarn-site.xml, resource-types.xml"
        // metrics = {NodeManagerMetrics@2120}
        // applications = {ConcurrentHashMap@3905} size = 0
        // systemCredentials = {HashMap@3906} size = 0
        // containers = {ConcurrentSkipListMap@3907} size = 0
        // registeringCollectors = null
        // knownCollectors = null
        // increasedContainers = {ConcurrentHashMap@3908} size = 0
        // containerTokenSecretManager = {NMContainerTokenSecretManager@3909}
        // nmTokenSecretManager = {NMTokenSecretManagerInNM@3910}
        // containerManager = {ContainerManagerImpl@2592} "Service org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManagerImpl in state org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManagerImpl: STARTED"
        // nodeResourceMonitor = {NodeResourceMonitorImpl@3911} "Service org.apache.hadoop.yarn.server.nodemanager.NodeResourceMonitorImpl in state org.apache.hadoop.yarn.server.nodemanager.NodeResourceMonitorImpl: STARTED"
        // dirsHandler = {LocalDirsHandlerService@2118} "Service org.apache.hadoop.yarn.server.nodemanager.LocalDirsHandlerService in state org.apache.hadoop.yarn.server.nodemanager.LocalDirsHandlerService: STARTED"
        // aclsManager = {ApplicationACLsManager@3912}
        // webServer = {WebServer@3913} "Service org.apache.hadoop.yarn.server.nodemanager.webapp.WebServer in state org.apache.hadoop.yarn.server.nodemanager.webapp.WebServer: INITED"
        // nodeHealthStatus = {NodeHealthStatusPBImpl@3914} "is_node_healthy: true health_report: "Healthy" last_health_report_time: 1612342130099"
        // stateStore = {NMNullStateStoreService@2122} "Service org.apache.hadoop.yarn.server.nodemanager.recovery.NMNullStateStoreService in state org.apache.hadoop.yarn.server.nodemanager.recovery.NMNullStateStoreService: STARTED"
        // isDecommissioned = false
        // logAggregationReportForApps = {ConcurrentLinkedQueue@3915} size = 0
        // nodeStatusUpdater = {NodeStatusUpdaterImpl@3832} "Service org.apache.hadoop.yarn.server.nodemanager.NodeStatusUpdaterImpl in state org.apache.hadoop.yarn.server.nodemanager.NodeStatusUpdaterImpl: INITED"
        // isDistSchedulingEnabled = false
        // deletionService = {DeletionService@2117} "Service org.apache.hadoop.yarn.server.nodemanager.DeletionService in state org.apache.hadoop.yarn.server.nodemanager.DeletionService: STARTED"
        // containerAllocator = {OpportunisticContainerAllocator@3916}
        // executor = {DefaultContainerExecutor@2115}
        // nmTimelinePublisher = null
        // containerStateTransitionListener = {NodeManager$DefaultContainerStateListener@3917}
        // resourcePluginManager = {ResourcePluginManager@3918}
        // nmLogAggregationStatusTracker = {NMLogAggregationStatusTracker@3919} "Service org.apache.hadoop.yarn.server.nodemanager.logaggregation.tracker.NMLogAggregationStatusTracker in state org.apache.hadoop.yarn.server.nodemanager.logaggregation.tracker.NMLogAggregationStatusTracker: INITED"
        this.nmContext = context;
    
        // 度量数据
        this.metrics = metrics;
      }

# 五 .serviceInit #

@Override
      public void serviceInit(Configuration conf) throws Exception { 
        this.validateConf(conf);
    
        this.publicRsrc = new LocalResourcesTrackerImpl(null, null, dispatcher, true, conf, stateStore, dirsHandler);
    
        this.recordFactory = RecordFactoryProvider.getRecordFactory(conf);
    
        try { 
          // 构建本地文件系统
          lfs = getLocalFileContext(conf);
          // 设置权限 : 755
          // Umask 为022表示默bai认创建新文du件权限为755
          lfs.setUMask(new FsPermission((short) FsPermission.DEFAULT_UMASK));
    
          if (!stateStore.canRecover()|| stateStore.isNewlyCreated()) { 
            cleanUpLocalDirs(lfs, delService);
            cleanupLogDirs(lfs, delService);
    
            // ${yarn.nodemanager.local-dirs} : /opt/tools/hadoop-3.2.1/local-dirs
            initializeLocalDirs(lfs);
            // ${yarn.nodemanager.log-dirs} : /opt/tools/hadoop-3.2.1/logs/userlogs
            initializeLogDirs(lfs);
          }
        } catch (Exception e) { 
          throw new YarnRuntimeException(
            "Failed to initialize LocalizationService", e);
        }
    
        diskValidator = DiskValidatorFactory.getInstance(YarnConfiguration.DEFAULT_DISK_VALIDATOR);
    
        // 缓存大小 : 10G
        cacheTargetSize = conf.getLong(YarnConfiguration.NM_LOCALIZER_CACHE_TARGET_SIZE_MB, YarnConfiguration.DEFAULT_NM_LOCALIZER_CACHE_TARGET_SIZE_MB) << 20;
    
        // 清理周期 10min
        cacheCleanupPeriod =  conf.getLong(YarnConfiguration.NM_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, YarnConfiguration.DEFAULT_NM_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS);
    
        // yarn.nodemanager.bind-host
        // yarn.nodemanager.localizer.address : 0.0.0.0:8040
        localizationServerAddress = conf.getSocketAddr(
            YarnConfiguration.NM_BIND_HOST,
            YarnConfiguration.NM_LOCALIZER_ADDRESS,
            YarnConfiguration.DEFAULT_NM_LOCALIZER_ADDRESS,
            YarnConfiguration.DEFAULT_NM_LOCALIZER_PORT);
    
        // ResourceLocalizationService$LocalizerTracker
        localizerTracker = createLocalizerTracker(conf);
    
    
        addService(localizerTracker);
    
    
        dispatcher.register(LocalizerEventType.class, localizerTracker);
    
    
        localDirsChangeListener = new DirsChangeListener() { 
          @Override
          public void onDirsChanged() { 
            checkAndInitializeLocalDirs();
          }
        };
    
    
        logDirsChangeListener = new DirsChangeListener() { 
          @Override
          public void onDirsChanged() { 
            initializeLogDirs(lfs);
          }
        };
        super.serviceInit(conf);
      }

# 六 .serviceStart #

开启定时清理,默认使用 CacheCleanup 每10min执行一次…

@Override
      public void serviceStart() throws Exception { 
    
        // 开启定时清理 : 每10min执行一次...
        cacheCleanup.scheduleWithFixedDelay(new CacheCleanup(dispatcher), cacheCleanupPeriod, cacheCleanupPeriod, TimeUnit.MILLISECONDS);
    
        // 构建RPC 服务
        server = createServer();
        // 启动RPC服务.
        server.start();
    
        // 更新localizationServer地址...
        // BoYi-Pro.local/192.168.8.188:8040
        localizationServerAddress =
            getConfig().updateConnectAddr(YarnConfiguration.NM_BIND_HOST,
                                          YarnConfiguration.NM_LOCALIZER_ADDRESS,
                                          YarnConfiguration.DEFAULT_NM_LOCALIZER_ADDRESS,
                                          server.getListenerAddress());
        LOG.info("Localizer started on port " + server.getPort());
    
    
        super.serviceStart();
    
        // 注册监听程序
        dirsHandler.registerLocalDirsChangeListener(localDirsChangeListener);
    
        // 注册监听程序
        dirsHandler.registerLogDirsChangeListener(logDirsChangeListener);
      }

# 七 .清理缓存 #

## 7.1. CacheCleanup ##

CacheCleanup 是ResourceLocalizationService 的内部类,用于定时清理缓存 .  
注意两个默认值:

// 缓存大小 : 10G
        cacheTargetSize = conf.getLong(YarnConfiguration.NM_LOCALIZER_CACHE_TARGET_SIZE_MB, YarnConfiguration.DEFAULT_NM_LOCALIZER_CACHE_TARGET_SIZE_MB) << 20;
    
        // 清理周期 10min
        cacheCleanupPeriod =  conf.getLong(YarnConfiguration.NM_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS, YarnConfiguration.DEFAULT_NM_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS);

代码:

static class CacheCleanup extends Thread { 
    
        private final Dispatcher dispatcher;
    
        public CacheCleanup(Dispatcher dispatcher) { 
          super("CacheCleanup");
          this.dispatcher = dispatcher;
        }
    
        @Override
        @SuppressWarnings("unchecked") // dispatcher not typed
        public void run() { 
          dispatcher.getEventHandler().handle(
              new LocalizationEvent(LocalizationEventType.CACHE_CLEANUP));
        }
    
      }

核心的方法是run方法, 其实就是调用ResourceLocalizationService中的handle方法中的CACHE\_CLEANUP分支…

@Override
      public void handle(LocalizationEvent event) { 
        // TODO: create log dir as $logdir/$user/$appId
        switch (event.getType()) { 
        
        //省略N行代码.....
    
        case CACHE_CLEANUP:
          // 执行清理操作...
          handleCacheCleanup();
          break;
       	//省略N行代码.....
       	
        default:
          throw new YarnRuntimeException("Unknown localization event: " + event);
        }
      }

## 7.2. LocalCacheCleaner 类 ##

在这里面使用了LocalCacheCleaner 类来处理本地的缓存…

### 7.2.1. 属性 ###

// 当前大小
      private long currentSize;
    
      // 目标大小
      private final long targetSize;
    
      // 删除服务...
      private final DeletionService delService;
    
      // 资源映射  LocalizedResource --> LocalResourcesTracker
      private final SortedMap<LocalizedResource, LocalResourcesTracker> resourceMap;

### 7.2.2.构造方法 ###

默认使用了LRU缓存 .  
其实就是用SortedMap类, 重写了compare方法.

LocalCacheCleaner(DeletionService delService, long targetSize) {
        
        // 使用LRU 缓存...
        this(delService, targetSize, new LRUComparator());
      }
    
      LocalCacheCleaner(DeletionService delService, long targetSize,
          Comparator<? super LocalizedResource> cmp) {
        this(delService, targetSize,
            new TreeMap<LocalizedResource, LocalResourcesTracker>(cmp));
      }
    
      LocalCacheCleaner(DeletionService delService, long targetSize,
          SortedMap<LocalizedResource, LocalResourcesTracker> resourceMap) {
        this.resourceMap = resourceMap;
        this.delService = delService;
        this.targetSize = targetSize;
      }

### 7.2.3. LRUComparator ###

首先对比文件的时间,然后再对比hashcode

private static class LRUComparator implements Comparator<LocalizedResource>,
          Serializable {
    
        private static final long serialVersionUID = 7034380228434701685L;
    
        public int compare(LocalizedResource r1, LocalizedResource r2) {
          long ret = r1.getTimestamp() - r2.getTimestamp();
          if (0 == ret) {
            return System.identityHashCode(r1) - System.identityHashCode(r2);
          }
          return ret > 0 ? 1 : -1;
        }
      }

### 7.2.4. 添加资源监控 ###

从传递的LocalResourcesTracker中添加要从缓存中删除的资源.

/**
       * 从传递的LocalResourcesTracker中添加要从缓存中删除的资源.
       *
       * Adds resources from the passed LocalResourceTracker
       *  that are candidates for deletion from the cache.
       *
       * @param newTracker add all resources being tracked by the passed
       *          LocalResourcesTracker to the LocalCacheCleaner.
       */
      public void addResources(LocalResourcesTracker newTracker) {
    
        for (LocalizedResource resource : newTracker) {
          // 累加新增资源,获取当前的资源大小.
          currentSize += resource.getSize();
    
          if (resource.getRefCount() > 0) {
            // 资源如果被引用, 忽略
            // Do not delete resources that are still in use
            continue;
          }
          // 将资源添加到对应的缓存中...
          resourceMap.put(resource, newTracker);
        }
      }

### 7.2.5. 清理缓存 \[ cleanCache \] ###

/**
       * Delete resources from the cache in the sorted order generated by the
       * Comparator used to construct this class.
       *
       * @return stats about what was cleaned up during this call of cleanCache
       */
      public LocalCacheCleanerStats cleanCache() {
    
        //根据当前的资源大小,构建 状态
        LocalCacheCleanerStats stats = new LocalCacheCleanerStats(currentSize);
    
        // 持续的清理, 直到达到目标缓存的大小.
        for (Iterator<Map.Entry<LocalizedResource, LocalResourcesTracker>> i = resourceMap.entrySet().iterator(); currentSize - stats.totalDelSize > targetSize && i.hasNext();) {
          Map.Entry<LocalizedResource, LocalResourcesTracker> rsrc = i.next();
          LocalizedResource resource = rsrc.getKey();
          LocalResourcesTracker tracker = rsrc.getValue();
          // 使用delService服务清理资源
          if (tracker.remove(resource, delService)) {
            stats.incDelSize(tracker.getUser(), resource.getSize());
          }
        }
        this.resourceMap.clear();
        return stats;
    
      }

### 7.2.6. LocalCacheCleanerStats ###

用于记录本地缓存清理的数据状态, 我们直接看常量即可.

// 用户删除文件的大小. 用户id ---> delSize
        private final Map<String, Long> userDelSizes = new TreeMap<String, Long>();
    
        // 清理缓存之前文件的大小.
        private final long cacheSizeBeforeClean;
        // 总清理资源大小.
        private long totalDelSize;
    
        // public 清理资源大小.
        private long publicDelSize;
    
        // private 清理资源大小
        private long privateDelSize;

*  唯一的一个核心方法 incDelSize .

void incDelSize(String user, long delSize) { 
    
          // 总资源减少..
          totalDelSize += delSize;
          if (user == null) { 
            // 如果用户为null , 删除public的资源
            publicDelSize += delSize;
          } else { 
            // 用户不为空, 删除private 资源
            privateDelSize += delSize;
            // 获取用户信息
            Long userDel = userDelSizes.get(user);
            if (userDel != null) { 
              // 删除用户带下的数据..
              userDel += delSize;
              // 更新缓存
              userDelSizes.put(user, userDel);
            } else { 
              // 新增用户缓存缓存.
              userDelSizes.put(user, delSize);
            }
          }
        }

# 八. LocalizerRunner #

## 8.1. 属性 ##

// 上下文信息
        final LocalizerContext context;
    
        //
        final String localizerId;
    
        // 等待处理的任务.
        final Map<LocalResourceRequest,LocalizerResourceRequestEvent> scheduled;
    
        // 它是私有本地化程序和调度程序线程之间的共享列表。
        // Its a shared list between Private Localizer and dispatcher thread.
        final List<LocalizerResourceRequestEvent> pending;
    
        
        private AtomicBoolean killContainerLocalizer = new AtomicBoolean(false);

## 8.2. 构建 ##

LocalizerRunner\#handle方法中,APPLICATION类型的事件处理.

> localizer = new LocalizerRunner(req.getContext(), locId);

LocalizerRunner(LocalizerContext context, String localizerId) { 
          super("LocalizerRunner for " + localizerId);
          this.context = context;
          this.localizerId = localizerId;
    
          // 同步方法...
          this.pending =
              Collections
                .synchronizedList(new ArrayList<LocalizerResourceRequestEvent>());
    
          // 待调度程序..
          this.scheduled =
              new HashMap<LocalResourceRequest, LocalizerResourceRequestEvent>();
        }

## 8.3. addResource ##

将请求加入pending缓存.

// 将请求加入缓存.
        public void addResource(LocalizerResourceRequestEvent request) { 
          pending.add(request);
        }

## 8.4. run ##

其实就是重queue队列中,获取任务的路径, 然后开始处理.

@Override
        @SuppressWarnings("unchecked") // dispatcher not typed
        public void run() { 
          Path nmPrivateCTokensPath = null;
          Throwable exception = null;
          try { 
            // Get nmPrivateDir
            nmPrivateCTokensPath =
              dirsHandler.getLocalPathForWrite(
                    NM_PRIVATE_DIR + Path.SEPARATOR
                        + String.format(ContainerLocalizer.TOKEN_FILE_NAME_FMT,
                            localizerId));
    
            // 0) init queue, etc.
            // 1) write credentials to private dir
            writeCredentials(nmPrivateCTokensPath);
            // 2) exec initApplication and wait
            if (dirsHandler.areDisksHealthy()) { 
    
              exec.startLocalizer(new LocalizerStartContext.Builder()
                      .setNmPrivateContainerTokens(nmPrivateCTokensPath)
                      .setNmAddr(localizationServerAddress)
                      .setUser(context.getUser())
                      .setAppId(context.getContainerId()
                              .getApplicationAttemptId().getApplicationId().toString())
                      .setLocId(localizerId)
                      .setDirsHandler(dirsHandler)
                      .build());
    
    
            } else { 
              throw new IOException("All disks failed. "
                  + dirsHandler.getDisksHealthReport(false));
            }
          // TODO handle ExitCodeException separately?
          } catch (FSError fe) { 
            exception = fe;
          } catch (Exception e) { 
            exception = e;
          } finally { 
            if (exception != null) { 
              LOG.info("Localizer failed for "+localizerId, exception);
              // On error, report failure to Container and signal ABORT
              // Notify resource of failed localization
              ContainerId cId = context.getContainerId();
              dispatcher.getEventHandler().handle(new ContainerResourceFailedEvent(
                  cId, null, exception.getMessage()));
            }
            List<Path> paths = new ArrayList<Path>();
            for (LocalizerResourceRequestEvent event : scheduled.values()) { 
              // This means some resources were in downloading state. Schedule
              // deletion task for localization dir and tmp dir used for downloading
              Path locRsrcPath = event.getResource().getLocalPath();
              if (locRsrcPath != null) { 
                Path locRsrcDirPath = locRsrcPath.getParent();
                paths.add(locRsrcDirPath);
                paths.add(new Path(locRsrcDirPath + "_tmp"));
              }
              event.getResource().unlock();
            }
            if (!paths.isEmpty()) { 
              FileDeletionTask deletionTask = new FileDeletionTask(delService,
                  context.getUser(), null, paths);
              delService.delete(deletionTask);
            }
            FileDeletionTask deletionTask = new FileDeletionTask(delService, null,
                nmPrivateCTokensPath, null);
            delService.delete(deletionTask);
          }
        }

# 九. PublicLocalizer #

## 9.1. 属性 ##

// 线程池
        // yarn.nodemanager.localizer.fetch.thread-count : 4
        final ExecutorService threadPool;
    
        // 队列
        final CompletionService<Path> queue;
    
        // Its shared between public localizer and dispatcher thread.
        final Map<Future<Path>,LocalizerResourceRequestEvent> pending;

## 9.2. 构造方法 ##

PublicLocalizer(Configuration conf) { 
          super("Public Localizer");
          this.lfs = getLocalFileContext(conf);
          this.conf = conf;
          this.pending = Collections.synchronizedMap(
              new HashMap<Future<Path>, LocalizerResourceRequestEvent>());
    
          // yarn.nodemanager.localizer.fetch.thread-count : 4
          this.threadPool = createLocalizerExecutor(conf);
    
          // 创建队列
          this.queue = new ExecutorCompletionService<Path>(threadPool);
        }

## 9.3. addResource ##

public void addResource(LocalizerResourceRequestEvent request) { 
          // TODO handle failures, cancellation, requests by other containers
          LocalizedResource rsrc = request.getResource();
          LocalResourceRequest key = rsrc.getRequest();
          LOG.info("Downloading public resource: " + key);
          /* 多个containers或许会请求相同的资源. 所以我们在以下情况下启动下载: 1. ResourceState 为DOWNLOADING 2. 我们能够获得非阻塞信号量锁。 否则，我们将跳过此资源，因为它正在下载或已失败/本地化。 * Here multiple containers may request the same resource. So we need * to start downloading only when * 1) ResourceState == DOWNLOADING * 2) We are able to acquire non blocking semaphore lock. * If not we will skip this resource as either it is getting downloaded * or it FAILED / LOCALIZED. */
    
          // 尝试加锁
          if (rsrc.tryAcquire()) { 
            // 状态是否为 DOWNLOADING
            if (rsrc.getState() == ResourceState.DOWNLOADING) { 
              // 获取LocalResource
              LocalResource resource = request.getResource().getRequest();
              try { 
    
                // 获取public 的根目录
                Path publicRootPath =
                    dirsHandler.getLocalPathForWrite("." + Path.SEPARATOR
                        + ContainerLocalizer.FILECACHE,
                      ContainerLocalizer.getEstimatedSize(resource), true);
    
                // 获取目标 目录
                Path publicDirDestPath =
                    publicRsrc.getPathForLocalization(key, publicRootPath,
                        delService);
    
    
                if (publicDirDestPath == null) { 
                  return;
                }
    
                // 验证 & 构架目录
                if (!publicDirDestPath.getParent().equals(publicRootPath)) { 
                  createParentDirs(publicDirDestPath, publicRootPath);
                  if (diskValidator != null) { 
                    diskValidator.checkStatus(
                        new File(publicDirDestPath.toUri().getPath()));
                  } else { 
                    throw new DiskChecker.DiskErrorException(
                        "Disk Validator is null!");
                  }
                }
    
                // 在此处显式同步挂起，以避免将来的任务在挂起更新之前完成并退出队列
                // explicitly synchronize pending here to avoid future task
                // completing and being dequeued before pending updated
                synchronized (pending) { 
                  // 构建FSDownload ?
                  pending.put(queue.submit(new FSDownload(lfs, null, conf,
                      publicDirDestPath, resource, request.getContext().getStatCache())),
                      request);
                }
              } catch (IOException e) { 
                rsrc.unlock();
                publicRsrc.handle(new ResourceFailedLocalizationEvent(request
                  .getResource().getRequest(), e.getMessage()));
                LOG.error("Local path for public localization is not found. "
                    + " May be disks failed.", e);
              } catch (IllegalArgumentException ie) { 
                rsrc.unlock();
                publicRsrc.handle(new ResourceFailedLocalizationEvent(request
                    .getResource().getRequest(), ie.getMessage()));
                LOG.error("Local path for public localization is not found. "
                    + " Incorrect path. " + request.getResource().getRequest()
                    .getPath(), ie);
              } catch (RejectedExecutionException re) { 
                rsrc.unlock();
                publicRsrc.handle(new ResourceFailedLocalizationEvent(request
                  .getResource().getRequest(), re.getMessage()));
                LOG.error("Failed to submit rsrc " + rsrc + " for download."
                    + " Either queue is full or threadpool is shutdown.", re);
              }
            } else { 
              if (LOG.isDebugEnabled()) { 
                LOG.debug("Skip downloading resource: " + key + " since it's in"
                    + " state: " + rsrc.getState());
              }
              rsrc.unlock();
            }
          } else { 
            if (LOG.isDebugEnabled()) { 
              LOG.debug("Skip downloading resource: " + key + " since it is locked"
                  + " by other threads");
            }
          }
        }

## 9.4. run ##

@Override
        public void run() { 
          try { 
            // TODO shutdown, better error handling esp. DU
            while (!Thread.currentThread().isInterrupted()) { 
              try { 
                // 获取任务路径
                Future<Path> completed = queue.take();
                // 从缓存中移除
                LocalizerResourceRequestEvent assoc = pending.remove(completed);
                try { 
                  if (null == assoc) { 
                    LOG.error("Localized unknown resource to " + completed);
                    // TODO delete
                    return;
                  }
                  // 获取路径
                  Path local = completed.get();
    
                  // 获取请求..
                  LocalResourceRequest key = assoc.getResource().getRequest();
    
                  // 处理缓存任务...
                  publicRsrc.handle(new ResourceLocalizedEvent(key, local, FileUtil.getDU(new File(local.toUri()))));
    
                  assoc.getResource().unlock();
    
                } catch (ExecutionException e) { 
                  String user = assoc.getContext().getUser();
                  ApplicationId applicationId = assoc.getContext().getContainerId().getApplicationAttemptId().getApplicationId();
                  LocalResourcesTracker tracker =
                    getLocalResourcesTracker(LocalResourceVisibility.APPLICATION, user, applicationId);
                  final String diagnostics = "Failed to download resource " +
                      assoc.getResource() + " " + e.getCause();
                  tracker.handle(new ResourceFailedLocalizationEvent(
                      assoc.getResource().getRequest(), diagnostics));
                  publicRsrc.handle(new ResourceFailedLocalizationEvent(
                      assoc.getResource().getRequest(), diagnostics));
                  LOG.error(diagnostics);
                  assoc.getResource().unlock();
                } catch (CancellationException e) { 
                  // ignore; shutting down
                }
              } catch (InterruptedException e) { 
                return;
              }
            }
          } catch(Throwable t) { 
            LOG.error("Error: Shutting down", t);
          } finally { 
            LOG.info("Public cache exiting");
            threadPool.shutdownNow();
          }
        }

# 十. LocalizerTracker #

## 10.1. 属性 ##

// 共有
        private final PublicLocalizer publicLocalizer;
    
    
        // 私有
        private final Map<String,LocalizerRunner> privLocalizers;

## 10.2. 构造方法 ##

ResourceLocalizationService\#serviceInit 的时候进行调用

LocalizerTracker(Configuration conf) { 
          this(conf, new HashMap<String,LocalizerRunner>());
        }
    
        LocalizerTracker(Configuration conf,
            Map<String,LocalizerRunner> privLocalizers) { 
          super(LocalizerTracker.class.getName());
          this.publicLocalizer = new PublicLocalizer(conf);
          this.privLocalizers = privLocalizers;
        }

## 10.4. serviceStart ##

启动publicLocalizer服务… 开始处理公共缓存文件…

@Override
        public synchronized void serviceStart() throws Exception { 
          publicLocalizer.start();
          super.serviceStart();
        }

## 10.5. handle ##

根据请求的 类型 进行不同的处理…

@Override
        public void handle(LocalizerEvent event) { 
          String locId = event.getLocalizerId();
          switch (event.getType()) { 
          case REQUEST_RESOURCE_LOCALIZATION:
            // 0) find running localizer or start new thread
            LocalizerResourceRequestEvent req = (LocalizerResourceRequestEvent)event;
    
    
            // 根据可见性进行处理
            switch (req.getVisibility()) { 
            case PUBLIC:
              publicLocalizer.addResource(req);
              break;
            case PRIVATE:
            case APPLICATION:
              synchronized (privLocalizers) { 
                // 获取 LocalizerRunner
                LocalizerRunner localizer = privLocalizers.get(locId);
    
                if (localizer != null && localizer.killContainerLocalizer.get()) { 
                  // Old localizer thread has been stopped, remove it and creates a new localizer thread.
                  LOG.info("New " + event.getType() + " localize request for "
                      + locId + ", remove old private localizer.");
                  cleanupPrivLocalizers(locId);
                  localizer = null;
                }
                if (null == localizer) { 
                  LOG.info("Created localizer for " + locId);
                  localizer = new LocalizerRunner(req.getContext(), locId);
                  privLocalizers.put(locId, localizer);
                  localizer.start();
                }
                // 1) propagate event
                localizer.addResource(req);
              }
              break;
            }
            break;
          }
        }

## 10.6. processHeartbeat ##

获取对应的请求数据对应的localizer, 交由具体的LocalizerRunner执行…

public LocalizerHeartbeatResponse processHeartbeat(LocalizerStatus status) { 
          String locId = status.getLocalizerId();
          synchronized (privLocalizers) { 
            // 获取LocalizerRunner
            LocalizerRunner localizer = privLocalizers.get(locId);
            if (null == localizer) { 
              // TODO process resources anyway
              LOG.info("Unknown localizer with localizerId " + locId
                  + " is sending heartbeat. Ordering it to DIE");
              LocalizerHeartbeatResponse response =
                recordFactory.newRecordInstance(LocalizerHeartbeatResponse.class);
              response.setLocalizerAction(LocalizerAction.DIE);
              return response;
            }
            // 处理心跳数据
            return localizer.processHeartbeat(status.getResources());
          }
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221119/98db36728ff64c7383f5bd76a558d204.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221119/e9cbe5152fb14cd49be0997ed415183e.png