Presto内存调优及原理（基础篇）

public final class LocalMemoryManager
{
    public static final MemoryPoolId GENERAL_POOL = new MemoryPoolId("general");
    public static final MemoryPoolId RESERVED_POOL = new MemoryPoolId("reserved");
    public static final MemoryPoolId SYSTEM_POOL = new MemoryPoolId("system");

    private final DataSize maxMemory;
    private final Map<MemoryPoolId, MemoryPool> pools;

    @Inject
    public LocalMemoryManager(NodeMemoryConfig config, ReservedSystemMemoryConfig systemMemoryConfig)
    {
        requireNonNull(config, "config is null");
        requireNonNull(systemMemoryConfig, "systemMemoryConfig is null");
        long maxHeap = Runtime.getRuntime().maxMemory();
        checkArgument(systemMemoryConfig.getReservedSystemMemory().toBytes() < maxHeap, "Reserved memory %s is greater than available heap %s", systemMemoryConfig.getReservedSystemMemory(), new DataSize(maxHeap, BYTE));
        maxMemory = new DataSize(maxHeap - systemMemoryConfig.getReservedSystemMemory().toBytes(), BYTE);

        ImmutableMap.Builder<MemoryPoolId, MemoryPool> builder = ImmutableMap.builder();
        checkArgument(config.getMaxQueryMemoryPerNode().toBytes() <= maxMemory.toBytes(), format("%s set to %s, but only %s of useable heap available", QUERY_MAX_MEMORY_PER_NODE_CONFIG, config.getMaxQueryMemoryPerNode(), maxMemory));
        builder.put(RESERVED_POOL, new MemoryPool(RESERVED_POOL, config.getMaxQueryMemoryPerNode()));
        DataSize generalPoolSize = new DataSize(Math.max(0, maxMemory.toBytes() - config.getMaxQueryMemoryPerNode().toBytes()), BYTE);
        builder.put(GENERAL_POOL, new MemoryPool(GENERAL_POOL, generalPoolSize));
        builder.put(SYSTEM_POOL, new MemoryPool(SYSTEM_POOL, systemMemoryConfig.getReservedSystemMemory()));
        this.pools = builder.build();
    }

public class NodeMemoryConfig
{
    public static final String QUERY_MAX_MEMORY_PER_NODE_CONFIG = "query.max-memory-per-node";

    private DataSize maxQueryMemoryPerNode = new DataSize(Runtime.getRuntime().maxMemory() * 0.1, BYTE);

    @NotNull
    public DataSize getMaxQueryMemoryPerNode()
    {
        return maxQueryMemoryPerNode;
    }

    @Config(QUERY_MAX_MEMORY_PER_NODE_CONFIG)
    public NodeMemoryConfig setMaxQueryMemoryPerNode(DataSize maxQueryMemoryPerNode)
    {
        this.maxQueryMemoryPerNode = maxQueryMemoryPerNode;
        return this;
    }
}

public class ReservedSystemMemoryConfig
{
    private DataSize reservedSystemMemory = new DataSize(Runtime.getRuntime().maxMemory() * 0.4, BYTE);

    @NotNull
    public DataSize getReservedSystemMemory()
    {
        return reservedSystemMemory;
    }

    @Config("resources.reserved-system-memory")
    public ReservedSystemMemoryConfig setReservedSystemMemory(DataSize reservedSystemMemory)
    {
        this.reservedSystemMemory = reservedSystemMemory;
        return this;
    }
}

public SqlTaskManager(
            LocalExecutionPlanner planner,
            LocationFactory locationFactory,
            TaskExecutor taskExecutor,
            QueryMonitor queryMonitor,
            NodeInfo nodeInfo,
            LocalMemoryManager localMemoryManager,
            TaskManagementExecutor taskManagementExecutor,
            TaskManagerConfig config,
            NodeMemoryConfig nodeMemoryConfig,
            LocalSpillManager localSpillManager,
            NodeSpillConfig nodeSpillConfig)
    {     

...      

  queryContexts = CacheBuilder.newBuilder().weakValues().build(CacheLoader.from(

                queryId -> new QueryContext(
                        queryId,
                        maxQueryMemoryPerNode,
                        localMemoryManager.getPool(LocalMemoryManager.GENERAL_POOL),
                        localMemoryManager.getPool(LocalMemoryManager.SYSTEM_POOL),
                        taskNotificationExecutor,
                        driverYieldExecutor,
                        maxQuerySpillPerNode,
                        localSpillManager.getSpillSpaceTracker())));

queryManagementExecutor.scheduleWithFixedDelay(() -> {
    ...
    enforceMemoryLimits();
    ...
}, 1, 1, TimeUnit.SECONDS);

if (reservedPool.getAssignedQueries() == 0 && generalPool.getBlockedNodes() > 0) {
    QueryExecution biggestQuery = null;
    long maxMemory = -1;
    for (QueryExecution queryExecution : queries) {

        if (resourceOvercommit(queryExecution.getSession())) {     
            continue;
        }

        long bytesUsed = queryExecution.getTotalMemoryReservation();

        if (bytesUsed > maxMemory) {
            biggestQuery = queryExecution;
            maxMemory = bytesUsed;
        }
    }

    if (biggestQuery != null) {
        biggestQuery.setMemoryPool(new VersionedMemoryPoolId(RESERVED_POOL, version));
    }

}

for (RemoteNodeMemory node : nodes.values()) {
    node.asyncRefresh(assignments);
}