1.内存初始化

初始化代码在LocalMemoryManager中，启动时将内存分为2个内存池，分别是：

  • RESERVED_POOL：预留内存池，用于执行最耗费内存资源的查询。
  • GENERAL_POOL：普通内存池，用于执行除最耗费内存查询以外的查询。

image.png
LocalMemoryManager的configureMemoryPools方法实现内存的分配
1.首先扣除heapHeadroom的大小，占用堆空间大小的30%，剩下的内存留给GENERAL_POOL（普通内存池）和RESERVED_POOL（预留内存池）
2.判断是否启动预留内存池（默认是开启），分配的堆内存大小占堆空间大小的30%，主要用于query比较占用内存使用
3.剩下的内存分配给普通内存池

  1. private void configureMemoryPools(NodeMemoryConfig config, long availableMemory)
  2. {
  3.     validateHeapHeadroom(config, availableMemory);
  5.     // 扣除系统  jvm(xmX)的0.3 表示堆大小
  6.     maxMemory = new DataSize(availableMemory - config.getHeapHeadroom().toBytes(), BYTE);
  8.     checkArgument(
  9.         config.getMaxQueryMemoryPerNode().toBytes() <= config.getMaxQueryTotalMemoryPerNode().toBytes(),
  10.         "Max query memory per node (%s) cannot be greater than the max query total memory per node (%s).",
  11.         QUERY_MAX_MEMORY_PER_NODE_CONFIG,
  12.         QUERY_MAX_TOTAL_MEMORY_PER_NODE_CONFIG);
  14.     // 缓存不同内存类型 与 内存池
  15.     ImmutableMap.Builder<MemoryPoolId, MemoryPool> builder = ImmutableMap.builder();
  17.     // 普通内存池
  18.     long generalPoolSize = maxMemory.toBytes();
  20.     // 是否开启 预留内存池，用于执行最耗费内存资源的查询
  21.     if (config.isReservedPoolEnabled()) {
  22.         // AVAILABLE_HEAP_MEMORY * 0.3  每个query最大内存限制条件
  23.         builder.put(RESERVED_POOL, new MemoryPool(RESERVED_POOL, config.getMaxQueryTotalMemoryPerNode()));
  24.         // 重新计普通内存大小
  25.         generalPoolSize -= config.getMaxQueryTotalMemoryPerNode().toBytes();
  26.     }
  27.     verify(generalPoolSize > 0, "general memory pool size is 0");
  28.     // 剩余的普通内存池的大小
  29.     builder.put(GENERAL_POOL, new MemoryPool(GENERAL_POOL, new DataSize(generalPoolSize, BYTE)));
  30.     this.pools = builder.build();
  31. }

2.定时调整内存池分配策略

SqlQueryManager中的start方法是在创建该实例之后就开始调用，其中enforceMemoryLimits()方法用于内存调整

  1. @PostConstruct  // 实现类执行构造函数之后调用该方法
  2. public void start()
  3. {
  4.     queryTracker.start();
  5.     // 定时调度
  6.     queryManagementExecutor.scheduleWithFixedDelay(() -> {
  7.         try {
  8.             killBlockingQuery();
  9.             // 强制内存调整
  10.             enforceMemoryLimits();
  11.         }
  12.         catch (Throwable e) {
  13.             log.error(e, "Error enforcing memory limits");
  14.         }
  16.         try {
  17.             enforceCpuLimits();
  18.         }
  19.         catch (Throwable e) {
  20.             log.error(e, "Error enforcing query CPU time limits");
  21.         }
  22.     }, 1, 1, TimeUnit.SECONDS);  // 每一秒触发一次执行
  23. }

进一步细分enforceMemoryLimits方法实现细节

  1. /**
  2.     * Enforce memory limits at the query level  在查询级别强制执行内存限制
  3. */
  4. private void enforceMemoryLimits()
  5. {
  6.     List<QueryExecution> runningQueries;
  7.     Supplier<List<BasicQueryInfo>> allQueryInfoSupplier;
  9.     Map<String, SharedQueryState> queryStates = StateCacheStore.get().getCachedStates(StateStoreConstants.QUERY_STATE_COLLECTION_NAME);
  10.     if (isMultiCoordinatorEnabled() && queryStates != null) {
  11.         // Get all queries from state store
  12.         runningQueries = queryStates.values().stream()
  13.             .filter(state -> state.getBasicQueryInfo().getState() == RUNNING)
  14.             .map(state -> new SharedQueryExecution(state, sessionPropertyManager)).collect(Collectors.toList());
  15.         allQueryInfoSupplier = () -> queryStates.values().stream().map(state -> state.getBasicQueryInfo()).collect(Collectors.toList());
  16.     }
  17.     else {
  18.         runningQueries = queryTracker.getAllQueries().stream()
  19.             .filter(query -> query.getState() == RUNNING)
  20.             .collect(toImmutableList());
  21.         allQueryInfoSupplier = this::getQueries;
  22.     }
  24.     // 核心处理源码
  25.     memoryManager.process(runningQueries, allQueryInfoSupplier);
  27.     // Put the chosen query to kill into state store
  28.     // 将所选查询杀死以进入状态存储
  29.     if (isMultiCoordinatorEnabled() && queryStates != null) {
  30.         updateKillingQuery(runningQueries, queryStates);
  31.     }
  32. }

查询一下核心处理代码process

  1. public synchronized void process(Iterable<QueryExecution> runningQueries, Supplier<List<BasicQueryInfo>> allQueryInfoSupplier)
  2. {
  3.     // 判断是否启动集群
  4.     if (!enabled) {
  5.         return;
  6.     }
  8.     //make sure state store is loaded when multiple coordinator is enabled
  9.     // 确保在启用多个协调器时加载状态存储
  10.     if (!hetuConfig.isMultipleCoordinatorEnabled()
  11.         || (hetuConfig.isMultipleCoordinatorEnabled()
  12.             && stateStoreProvider.getStateStore() != null)) {
  13.         Lock lock = null;
  14.         boolean locked = false;
  15.         try {
  16.             if (hetuConfig.isMultipleCoordinatorEnabled()) {
  17.                 lock = stateStoreProvider.getStateStore().getLock(StateStoreConstants.HANDLE_OOM_QUERY_LOCK_NAME);
  18.             }
  19.             else {
  20.                 lock = new ReentrantLock();
  21.             }
  23.             // 加锁
  24.             locked = lock.tryLock(StateStoreConstants.DEFAULT_ACQUIRED_LOCK_TIME_MS, TimeUnit.MILLISECONDS);
  25.             if (locked) {
  26.                 // TODO revocable memory reservations can also leak and may need to be detected in the future
  27.                 // 可撤销的内存预留也可能会泄漏，将来可能需要检测到
  28.                 // We are only concerned about the leaks in general pool.
  29.                 // 我们只关心普通池中的泄漏
  31.                 // 检查内存泄漏，并保存起来
  32.                 memoryLeakDetector.checkForMemoryLeaks(allQueryInfoSupplier, pools.get(GENERAL_POOL).getQueryMemoryReservations());
  34.                 // 判断集群是否出现内存泄漏
  35.                 /**
  36.   * 以下两种情况出现：
  37.   * 1.当预留内存池空间为空， 普通内存池出现堵塞的节点，则表示集群出现OOM
  38.   * 2.当预留内存池不为空，并且普通内存池出现堵塞节点，则表示集群出现OOM
  39.   */
  40.                 boolean outOfMemory = isClusterOutOfMemory();
  41.                 if (!outOfMemory) {
  42.                     lastTimeNotOutOfMemory = System.nanoTime(); // 最后一次检查的时间
  43.                 }
  45.                 boolean queryKilled = false;
  46.                 // 所有用户内存字节数
  47.                 long totalUserMemoryBytes = 0L;
  48.                 // 所有字节数
  49.                 long totalMemoryBytes = 0L;
  52.                 // 遍历正在运行的查询语句
  53.                 for (QueryExecution query : runningQueries) {
  55.                     boolean resourceOvercommit = resourceOvercommit(query.getSession());
  57.                     // 用户剩余的内存大小
  58.                     long userMemoryReservation = query.getUserMemoryReservation().toBytes();
  60.                     // 所有剩余内存的大小
  61.                     long totalMemoryReservation = query.getTotalMemoryReservation().toBytes();
  63.                     // 当出现资源超支，并且 出现OOM， 出现执行kill 一些query
  64.                     if (resourceOvercommit && outOfMemory) {
  65.                         // If a query has requested resource overcommit, only kill it if the cluster has run out of memory
  66.                         // 如果查询请求了资源过量使用，则只有在群集内存不足时才将其杀死
  67.                         DataSize memory = succinctBytes(getQueryMemoryReservation(query));
  68.                         //
  69.                         query.fail(new PrestoException(CLUSTER_OUT_OF_MEMORY,
  70.                                                        format("The cluster is out of memory and %s=true, so this query was killed. It was using %s of memory", RESOURCE_OVERCOMMIT, memory)));
  71.                         queryKilled = true;
  72.                     }
  74.                     // 请求资料没有过量时
  75.                     if (!resourceOvercommit) {
  76.                         // 获取用户内存使用最小值
  77.                         long userMemoryLimit = min(maxQueryMemory.toBytes(), getQueryMaxMemory(query.getSession()).toBytes());
  78.                         // 用户请求内存大于限制条件时，则请求失败，并标记为可以kill
  79.                         if (userMemoryReservation > userMemoryLimit) {
  80.                             query.fail(exceededGlobalUserLimit(succinctBytes(userMemoryLimit)));
  81.                             queryKilled = true;
  82.                         }
  84.                         // 请求所有内存限制值
  85.                         long totalMemoryLimit = min(maxQueryTotalMemory.toBytes(), getQueryMaxTotalMemory(query.getSession()).toBytes());
  86.                         // 大于所有内存最大小值时，则标记kill
  87.                         if (totalMemoryReservation > totalMemoryLimit) {
  88.                             query.fail(exceededGlobalTotalLimit(succinctBytes(totalMemoryLimit)));
  89.                             queryKilled = true;
  90.                         }
  91.                     }
  93.                     // 并累加之前的内存值, 统计所有的用户使用的内存大小
  94.                     // 所有内存的总大小
  95.                     totalUserMemoryBytes += userMemoryReservation;
  96.                     totalMemoryBytes += totalMemoryReservation;
  97.                 }
  99.                 // 再次设置需要所有内存值
  100.                 clusterUserMemoryReservation.set(totalUserMemoryBytes);
  101.                 clusterTotalMemoryReservation.set(totalMemoryBytes);
  103.                 // 当lowMemoryKiller是TotalReservationLowMemoryKiller， TotalReservationOnBlockedNodesLowMemoryKiller时
  104.                 // 出现OOM
  105.                 // 非querykilled
  106.                 // 最后一次出现的OOM时间大于上次OOM的时间时
  107.                 //
  108.                 if (!(lowMemoryKiller instanceof NoneLowMemoryKiller) &&
  109.                     outOfMemory &&
  110.                     !queryKilled &&
  111.                     nanosSince(lastTimeNotOutOfMemory).compareTo(killOnOutOfMemoryDelay) > 0) {
  113.                     // 上一次killquery是否完成
  114.                     if (isLastKilledQueryGone()) {
  115.                         // 调用oom的killer
  116.                         callOomKiller(runningQueries);
  117.                     }
  118.                     else {
  119.                         log.debug("Last killed query is still not gone: %s", lastKilledQuery);
  120.                     }
  121.                 }
  122.             }
  123.         }
  124.         catch (Exception e) {
  125.             log.error("Error handleOOMQuery: " + e.getMessage());
  126.         }
  127.         finally {
  128.             if (lock != null && locked) {
  129.                 lock.unlock();
  130.             }
  131.         }
  132.     }
  135.     Map<MemoryPoolId, Integer> countByPool = new HashMap<>();
  136.     // 统计不同的query使用线程池
  137.     for (QueryExecution query : runningQueries) {
  138.         MemoryPoolId id = query.getMemoryPool().getId();
  139.         countByPool.put(id, countByPool.getOrDefault(id, 0) + 1);
  140.     }
  142.     // 更新线程池
  143.     updatePools(countByPool);
  145.     MemoryPoolAssignmentsRequest assignmentsRequest;
  146.     //
  147.     if (pools.containsKey(RESERVED_POOL)) {
  148.         // 如果预留线程池已使用，需要更新
  149.         assignmentsRequest = updateAssignments(runningQueries);
  150.     }
  151.     else {
  152.         // If reserved pool is not enabled, we don't create a MemoryPoolAssignmentsRequest that puts all the queries
  153.         // in the general pool (as they already are). In this case we create an effectively NOOP MemoryPoolAssignmentsRequest.
  154.         // Once the reserved pool is removed we should get rid of the logic of putting queries into reserved pool including
  155.         // this piece of code.
  157.         // 如果未启用保留池，则不会创建将所有查询放入常规池中的MemoryPoolAssignmentsRequest（因为它们已经存在）。
  158.         // 在这种情况下，我们将创建一个有效的NOOP MemoryPoolAssignmentsRequest。
  159.         // 除去保留池后，我们应该摆脱将查询放入保留池的逻辑，包括这段代码。
  161.         assignmentsRequest = new MemoryPoolAssignmentsRequest(coordinatorId, Long.MIN_VALUE, ImmutableList.of());
  162.     }
  163.     // 更新节点
  164.     updateNodes(assignmentsRequest);
  165. }

updateNodes方法更新所有节点

  1. private synchronized void updateNodes(MemoryPoolAssignmentsRequest assignments)
  2. {
  3.     ImmutableSet.Builder<InternalNode> builder = ImmutableSet.builder();
  4.     // 统计所有的内部节点
  5.     Set<InternalNode> aliveNodes = builder
  6.         .addAll(nodeManager.getNodes(ACTIVE))
  7.         .addAll(nodeManager.getNodes(ISOLATING))
  8.         .addAll(nodeManager.getNodes(ISOLATED))
  9.         .addAll(nodeManager.getNodes(SHUTTING_DOWN))
  10.         .build();
  12.     ImmutableSet<String> aliveNodeIds = aliveNodes.stream()
  13.         .map(InternalNode::getNodeIdentifier)
  14.         .collect(toImmutableSet());
  16.     // Remove nodes that don't exist anymore
  17.     // Make a copy to materialize the set difference
  18.     Set<String> deadNodes = ImmutableSet.copyOf(difference(nodes.keySet(), aliveNodeIds));
  20.     // 移除已经dead的node
  21.     nodes.keySet().removeAll(deadNodes);
  23.     // Add new nodes
  24.     for (InternalNode node : aliveNodes) {
  25.         if (!nodes.containsKey(node.getNodeIdentifier()) && shouldIncludeNode(node)) {
  26.             //                nodes.put(node.getNodeIdentifier(), new RemoteNodeMemory(node, httpClient, memoryInfoCodec, assignmentsRequestCodec, locationFactory.createMemoryInfoLocation(node), isBinaryEncoding));
  27.             nodes.put(node.getInternalUri().toString(), new RemoteNodeMemory(node, httpClient, memoryInfoCodec, assignmentsRequestCodec, locationFactory.createMemoryInfoLocation(node), isBinaryEncoding));
  28.             allNodes.put(node.getInternalUri().toString(), new RemoteNodeMemory(node, httpClient, memoryInfoCodec, assignmentsRequestCodec, locationFactory.createMemoryInfoLocation(node), isBinaryEncoding));
  29.         }
  30.     }
  32.     // Schedule refresh  刷新集群信息
  33.     for (RemoteNodeMemory node : nodes.values()) {
  34.         node.asyncRefresh(assignments);
  35.     }
  37.     // Schedule All refresh
  38.     //
  39.     for (RemoteNodeMemory node : allNodes.values()) {
  40.         node.asyncRefresh(assignments);
  41.     }
  42. }

asyncRefresh方法，将目前集群的所有资源信息发送到每个worker上，并获取对应的本地资源信息

  1. // 异步刷新
  2. public void asyncRefresh(MemoryPoolAssignmentsRequest assignments)
  3. {
  4.     Duration sinceUpdate = nanosSince(lastUpdateNanos.get());
  6.     //
  7.     if (nanosSince(lastWarningLogged.get()).toMillis() > 1_000 &&
  8.         sinceUpdate.toMillis() > 10_000 &&
  9.         future.get() != null) {
  10.         log.warn("Memory info update request to %s has not returned in %s", memoryInfoUri, sinceUpdate.toString(SECONDS));
  11.         lastWarningLogged.set(System.nanoTime());
  12.     }
  13.     //
  14.     if (sinceUpdate.toMillis() > 1_000 && future.get() == null) {
  16.         Request request = setContentTypeHeaders(isBinaryEncoding, preparePost())
  17.             .setUri(memoryInfoUri)
  18.             .setBodyGenerator(createBodyGenerator(assignments))
  19.             .build();
  21.         ResponseHandler responseHandler;
  22.         if (isBinaryEncoding) {
  23.             responseHandler = createFullSmileResponseHandler((SmileCodec<MemoryInfo>) memoryInfoCodec);
  24.         }
  25.         else {
  26.             responseHandler = createAdaptingJsonResponseHandler(unwrapJsonCodec(memoryInfoCodec));
  27.         }
  28.         HttpResponseFuture<BaseResponse<MemoryInfo>> responseFuture = httpClient.executeAsync(request, responseHandler);
  29.         future.compareAndSet(null, responseFuture);
  31.         Futures.addCallback(responseFuture, new FutureCallback<BaseResponse<MemoryInfo>>()
  32.                             {
  33.                                 @Override
  34.                                 public void onSuccess(@Nullable BaseResponse<MemoryInfo> result)
  35.                                 {
  36.                                     lastUpdateNanos.set(System.nanoTime());
  37.                                     future.compareAndSet(responseFuture, null);
  38.                                     long version = currentAssignmentVersion.get();
  39.                                     if (result != null) {
  40.                                         if (result.hasValue()) {
  41.                                             memoryInfo.set(Optional.ofNullable(result.getValue()));
  42.                                         }
  43.                                         if (result.getStatusCode() != OK.code()) {
  44.                                             log.warn("Error fetching memory info from %s returned status %d: %s", memoryInfoUri, result.getStatusCode(), result.getStatusCode());
  45.                                             return;
  46.                                         }
  47.                                     }
  48.                                     currentAssignmentVersion.compareAndSet(version, assignments.getVersion());
  49.                                 }
  51.                                 @Override
  52.                                 public void onFailure(Throwable t)
  53.                                 {
  54.                                     log.warn("Error fetching memory info from %s: %s", memoryInfoUri, t.getMessage());
  55.                                     lastUpdateNanos.set(System.nanoTime());
  56.                                     future.compareAndSet(responseFuture, null);
  57.                                 }
  58.                             }, directExecutor());
  59.     }
  60. }

MemoryResource在该类中，用户接收协调器发送的资源信息，POST方法，将本地的内存信息返回到协调器

  1. @POST
  2. @Produces({MediaType.APPLICATION_JSON, APPLICATION_JACKSON_SMILE})
  3. @Consumes({MediaType.APPLICATION_JSON, APPLICATION_JACKSON_SMILE})
  4. public MemoryInfo getMemoryInfo(MemoryPoolAssignmentsRequest request)
  5. {
  6.     taskManager.updateMemoryPoolAssignments(request);
  7.     return memoryManager.getInfo();
  8. }