Trino 查询调度和线程池

public static final int DISPATCHER_THREADPOOL_MAX_SIZE = max(50, getRuntime().availableProcessors() * 10);

public DispatchExecutor(QueryManagerConfig config, VersionEmbedder versionEmbedder)
{
    ExecutorService coreExecutor = newCachedThreadPool(daemonThreadsNamed("dispatcher-query-%s"));
    closer.register(coreExecutor::shutdownNow);
    executor = decorateWithVersion(coreExecutor, versionEmbedder);

    ScheduledExecutorService coreScheduledExecutor = newScheduledThreadPool(config.getQueryManagerExecutorPoolSize(), daemonThreadsNamed("dispatch-executor-%s"));
    closer.register(coreScheduledExecutor::shutdownNow);
    scheduledExecutor = listeningDecorator(coreScheduledExecutor);

    mbeans = new DispatchExecutorMBeans(coreExecutor, coreScheduledExecutor);
}

@Inject
public DispatchManager(
        QueryIdGenerator queryIdGenerator,
        QueryPreparer queryPreparer,
        ResourceGroupManager<?> resourceGroupManager,
        DispatchQueryFactory dispatchQueryFactory,
        FailedDispatchQueryFactory failedDispatchQueryFactory,
        AccessControl accessControl,
        SessionSupplier sessionSupplier,
        SessionPropertyDefaults sessionPropertyDefaults,
        SessionPropertyManager sessionPropertyManager,
        Tracer tracer,
        QueryManagerConfig queryManagerConfig,
        DispatchExecutor dispatchExecutor,
        QueryMonitor queryMonitor)
{
    this.queryIdGenerator = requireNonNull(queryIdGenerator, "queryIdGenerator is null");
    this.queryPreparer = requireNonNull(queryPreparer, "queryPreparer is null");
    this.resourceGroupManager = requireNonNull(resourceGroupManager, "resourceGroupManager is null");
    this.dispatchQueryFactory = requireNonNull(dispatchQueryFactory, "dispatchQueryFactory is null");
    this.failedDispatchQueryFactory = requireNonNull(failedDispatchQueryFactory, "failedDispatchQueryFactory is null");
    this.accessControl = requireNonNull(accessControl, "accessControl is null");
    this.sessionSupplier = requireNonNull(sessionSupplier, "sessionSupplier is null");
    this.sessionPropertyDefaults = requireNonNull(sessionPropertyDefaults, "sessionPropertyDefaults is null");
    this.sessionPropertyManager = sessionPropertyManager;
    this.tracer = requireNonNull(tracer, "tracer is null");

    this.maxQueryLength = queryManagerConfig.getMaxQueryLength();

    this.dispatchExecutor = new BoundedExecutor(dispatchExecutor.getExecutor(), queryManagerConfig.getDispatcherQueryPoolSize());

    this.queryTracker = new QueryTracker<>(queryManagerConfig, dispatchExecutor.getScheduledExecutor());
    this.queryMonitor = requireNonNull(queryMonitor, "queryMonitor is null");
    this.statsUpdaterExecutor = newSingleThreadScheduledExecutor(daemonThreadsNamed("dispatch-manager-stats-%s"));
}

private void asyncQueryResults(
        Query query,
        long token,
        Duration maxWait,
        DataSize targetResultSize,
        ExternalUriInfo externalUriInfo,
        DisconnectionAwareAsyncResponse asyncResponse)

private synchronized ListenableFuture<Void> getFutureStateChange()
{
    // if the exchange client is open, wait for data
    if (!exchangeDataSource.isFinished()) {
        if (exchangeDataSourceBlocked != null && !exchangeDataSourceBlocked.isDone()) {
            return exchangeDataSourceBlocked;
        }
        ListenableFuture<Void> blocked = exchangeDataSource.isBlocked();
        if (blocked.isDone()) {
            // not blocked
            return immediateVoidFuture();
        }
        // cache future to avoid accumulation of callbacks on the underlying future
        exchangeDataSourceBlocked = ignoreCancellation(blocked);
        return exchangeDataSourceBlocked;
    }
    exchangeDataSourceBlocked = null;

    if (!resultsConsumed) {
        return immediateVoidFuture();
    }

    // otherwise, wait for the query to finish
    queryManager.recordHeartbeat(queryId);
    try {
        return queryDoneFuture(queryManager.getQueryState(queryId));
    }
    catch (NoSuchElementException e) {
        return immediateVoidFuture();
    }
}

try {
    long bytes = 0;
    while (bytes < targetResultBytes) {
        Slice serializedPage = exchangeDataSource.pollPage();
        if (serializedPage == null) {
            break;
        }

        Page page = deserializer.deserialize(serializedPage);
        bytes += page.getLogicalSizeInBytes();
        resultBuilder.addPage(page);
    }
    if (exchangeDataSource.isFinished()) {
        exchangeDataSource.close();
        deserializer = null; // null to reclaim memory of PagesSerde which does not expose explicit lifecycle
    }
}
catch (Throwable cause) {
    queryManager.failQuery(queryId, cause);
}

@Override
public void taskCreated(PlanFragmentId fragmentId, RemoteTask task)
{
    URI taskUri = uriBuilderFrom(task.getTaskStatus().getSelf())
            .appendPath("results")
            .appendPath("0").build();
    DirectExchangeInput input = new DirectExchangeInput(task.getTaskId(), taskUri.toString());
    queryStateMachine.updateInputsForQueryResults(ImmutableList.of(input), false);
}

private void fireStateChangedIfReady(Optional<QueryOutputInfo> info, Optional<Consumer<QueryOutputInfo>> listener)
{
    if (info.isEmpty() || listener.isEmpty()) {
        return;
    }
    executor.execute(() -> listener.get().accept(info.get()));
}

private synchronized void setQueryOutputInfo(QueryExecution.QueryOutputInfo outputInfo)
{
    // if first callback, set column names
    if (columns == null) {
        List<String> columnNames = outputInfo.getColumnNames();
        List<Type> columnTypes = outputInfo.getColumnTypes();
        checkArgument(columnNames.size() == columnTypes.size(), "Column names and types size mismatch");

        ImmutableList.Builder<Column> list = ImmutableList.builder();
        for (int i = 0; i < columnNames.size(); i++) {
            list.add(createColumn(columnNames.get(i), columnTypes.get(i), supportsParametricDateTime));
        }
        columns = list.build();
        types = outputInfo.getColumnTypes();
    }

    outputInfo.drainInputs(exchangeDataSource::addInput);
    if (outputInfo.isNoMoreInputs()) {
        exchangeDataSource.noMoreInputs();
    }
}

synchronized int scheduleRequestIfNecessary()
{
    if ((buffer.isFinished() || buffer.isFailed()) && completedClients.size() == allClients.size()) {
        return 0;
    }

    long neededBytes = buffer.getRemainingCapacityInBytes();
    if (neededBytes <= 0) {
        return 0;
    }

    long reservedBytesForScheduledClients = allClients.values().stream()
            .filter(client -> !queuedClients.contains(client) && !completedClients.contains(client))
            .mapToLong(HttpPageBufferClient::getAverageRequestSizeInBytes)
            .sum();
    long projectedBytesToBeRequested = 0;
    int clientCount = 0;

    Iterator<HttpPageBufferClient> clientIterator = queuedClients.iterator();
    while (clientIterator.hasNext()) {
        HttpPageBufferClient client = clientIterator.next();
        if (projectedBytesToBeRequested >= neededBytes * concurrentRequestMultiplier - reservedBytesForScheduledClients) {
            break;
        }
        projectedBytesToBeRequested += client.getAverageRequestSizeInBytes();

        client.scheduleRequest();

        // Remove the client from the queuedClient's set.
        clientIterator.remove();

        clientCount++;
    }

    return clientCount;
}

@Provides
@Singleton
@ForStatementResource
public static ScheduledExecutorService createStatementTimeoutExecutor(TaskManagerConfig config)
{
    return newScheduledThreadPool(config.getHttpTimeoutThreads(), daemonThreadsNamed("statement-timeout-%s"));
}