gortiz commented on code in PR #11205:
URL: https://github.com/apache/pinot/pull/11205#discussion_r1281493383
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java:
##########
@@ -31,151 +33,84 @@
import org.slf4j.LoggerFactory;
-/**
- * This class provides the implementation for scheduling multistage queries on
a single node based
- * on the {@link OpChainScheduler} logic that is passed in. Multistage queries
support partial execution
- * and will return a NOOP metadata block as a "yield" signal, indicating that
the next operator
- * chain ({@link OpChainScheduler#next} will be requested.
- */
-@SuppressWarnings("UnstableApiUsage")
-public class OpChainSchedulerService extends AbstractExecutionThreadService {
+public class OpChainSchedulerService implements SchedulerService {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpChainSchedulerService.class);
- /**
- * Default time scheduler is allowed to wait for a runnable OpChain to be
available.
- */
- private static final long DEFAULT_SCHEDULER_NEXT_WAIT_MS = 100;
- /**
- * Default cancel signal retention, this should be set to several times
larger than
- * {@link
org.apache.pinot.query.service.QueryConfig#DEFAULT_SCHEDULER_RELEASE_TIMEOUT_MS}.
- */
- private static final long SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS =
60_000L;
-
- private final OpChainScheduler _scheduler;
- private final ExecutorService _workerPool;
- private final Cache<Long, Long> _cancelledRequests =
CacheBuilder.newBuilder()
- .expireAfterWrite(SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS,
TimeUnit.MILLISECONDS).build();
- public OpChainSchedulerService(OpChainScheduler scheduler, ExecutorService
workerPool) {
- _scheduler = scheduler;
- _workerPool = workerPool;
- }
+ private final ExecutorService _executorService;
+ private final ConcurrentHashMap<OpChainId, Future<?>> _submittedOpChainMap;
- @Override
- protected void triggerShutdown() {
- // TODO: Figure out shutdown lifecycle with graceful shutdown in mind.
- LOGGER.info("Triggered shutdown on OpChainScheduler...");
+ public OpChainSchedulerService(ExecutorService executorService) {
+ _executorService = executorService;
+ _submittedOpChainMap = new ConcurrentHashMap<>();
}
@Override
- protected void run()
- throws Exception {
- while (isRunning()) {
- OpChain operatorChain = _scheduler.next(DEFAULT_SCHEDULER_NEXT_WAIT_MS,
TimeUnit.MILLISECONDS);
- if (operatorChain == null) {
- continue;
- }
- LOGGER.trace("({}): Scheduling", operatorChain);
- _workerPool.submit(new TraceRunnable() {
+ public void register(OpChain operatorChain) {
+ Future<?> scheduledFuture = _executorService.submit(new TraceRunnable() {
@Override
public void runJob() {
boolean isFinished = false;
- boolean returnedErrorBlock = false;
+ TransferableBlock returnedErrorBlock = null;
Throwable thrown = null;
try {
LOGGER.trace("({}): Executing", operatorChain);
- // throw if the operatorChain is cancelled.
- if
(_cancelledRequests.asMap().containsKey(operatorChain.getId().getRequestId())) {
- throw new InterruptedException("Query was cancelled!");
- }
operatorChain.getStats().executing();
- // so long as there's work to be done, keep getting the next block
- // when the operator chain returns a NOOP block, then yield the
execution
- // of this to another worker
TransferableBlock result = operatorChain.getRoot().nextBlock();
- while (!result.isNoOpBlock() && !result.isEndOfStreamBlock()) {
+ while (!result.isEndOfStreamBlock()) {
result = operatorChain.getRoot().nextBlock();
}
-
- if (result.isNoOpBlock()) {
- // TODO: There should be a waiting-for-data state in
OpChainStats.
- operatorChain.getStats().queued();
- _scheduler.yield(operatorChain);
+ isFinished = true;
+ if (result.isErrorBlock()) {
+ returnedErrorBlock = result;
+ LOGGER.error("({}): Completed erroneously {} {}", operatorChain,
operatorChain.getStats(),
+ result.getDataBlock().getExceptions());
} else {
- isFinished = true;
- if (result.isErrorBlock()) {
- returnedErrorBlock = true;
- LOGGER.error("({}): Completed erroneously {} {}",
operatorChain, operatorChain.getStats(),
- result.getDataBlock().getExceptions());
- } else {
- LOGGER.debug("({}): Completed {}", operatorChain,
operatorChain.getStats());
- }
+ LOGGER.debug("({}): Completed {}", operatorChain,
operatorChain.getStats());
}
} catch (Exception e) {
LOGGER.error("({}): Failed to execute operator chain! {}",
operatorChain, operatorChain.getStats(), e);
thrown = e;
} finally {
- if (returnedErrorBlock || thrown != null) {
+ if (returnedErrorBlock != null || thrown != null) {
+ if (thrown == null) {
+ String blockMsg;
+ if (returnedErrorBlock.getDataBlock() instanceof
MetadataBlock) {
+ MetadataBlock metadataBlock = (MetadataBlock)
returnedErrorBlock.getDataBlock();
+ blockMsg = String.join(", ",
metadataBlock.getExceptions().values());
+ } else {
+ blockMsg = "Unknown";
+ }
+ thrown = new RuntimeException("Error block " + blockMsg);
+ }
cancelOpChain(operatorChain, thrown);
} else if (isFinished) {
closeOpChain(operatorChain);
}
}
}
});
- }
+ _submittedOpChainMap.put(operatorChain.getId(), scheduledFuture);
}
- /**
- * Register a new operator chain with the scheduler.
- *
- * @param operatorChain the chain to register
- */
- public final void register(OpChain operatorChain) {
- operatorChain.getStats().queued();
- _scheduler.register(operatorChain);
- LOGGER.debug("({}): Scheduler is now handling operator chain listening to
mailboxes {}. "
- + "There are a total of {} chains awaiting execution.",
operatorChain,
- operatorChain.getReceivingMailboxIds(),
- _scheduler.size());
- }
-
- /**
- * Async cancel a request. Request will not be fully cancelled until the
next time opChain is being polled.
- *
- * @param requestId requestId to be cancelled.
- */
- public final void cancel(long requestId) {
- _cancelledRequests.put(requestId, requestId);
- }
-
- /**
- * This method should be called whenever data is available for an {@link
OpChain} to consume.
- * Implementations of this method should be idempotent, it may be called in
the scenario that no data is available.
- *
- * @param opChainId the identifier of the operator chain
- */
- public final void onDataAvailable(OpChainId opChainId) {
- _scheduler.onDataAvailable(opChainId);
- }
-
- // TODO: remove this method after we pipe down the proper executor pool to
the v1 engine
- public ExecutorService getWorkerPool() {
- return _workerPool;
+ @Override
+ public void cancel(long requestId) {
+ // simple cancellation. for leaf stage this cannot be a dangling opchain
b/c they will eventually be cleared up
+ // via query timeout.
+ List<OpChainId> opChainIdsToCancel = _submittedOpChainMap.keySet()
+ .stream().filter(opChainId -> opChainId.getRequestId() ==
requestId).collect(Collectors.toList());
+ for (OpChainId opChainId : opChainIdsToCancel) {
+ Future<?> future = _submittedOpChainMap.get(opChainId);
+ if (future != null) {
+ future.cancel(true);
+ }
+ }
}
private void closeOpChain(OpChain opChain) {
- try {
opChain.close();
- } finally {
- _scheduler.deregister(opChain);
- }
}
- private void cancelOpChain(OpChain opChain, Throwable e) {
- try {
- opChain.cancel(e);
- } finally {
- _scheduler.deregister(opChain);
- }
+ private void cancelOpChain(OpChain opChain, @Nullable Throwable t) {
Review Comment:
Fixed in fcf8bc4
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java:
##########
@@ -31,151 +33,84 @@
import org.slf4j.LoggerFactory;
-/**
- * This class provides the implementation for scheduling multistage queries on
a single node based
- * on the {@link OpChainScheduler} logic that is passed in. Multistage queries
support partial execution
- * and will return a NOOP metadata block as a "yield" signal, indicating that
the next operator
- * chain ({@link OpChainScheduler#next} will be requested.
- */
-@SuppressWarnings("UnstableApiUsage")
-public class OpChainSchedulerService extends AbstractExecutionThreadService {
+public class OpChainSchedulerService implements SchedulerService {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpChainSchedulerService.class);
- /**
- * Default time scheduler is allowed to wait for a runnable OpChain to be
available.
- */
- private static final long DEFAULT_SCHEDULER_NEXT_WAIT_MS = 100;
- /**
- * Default cancel signal retention, this should be set to several times
larger than
- * {@link
org.apache.pinot.query.service.QueryConfig#DEFAULT_SCHEDULER_RELEASE_TIMEOUT_MS}.
- */
- private static final long SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS =
60_000L;
-
- private final OpChainScheduler _scheduler;
- private final ExecutorService _workerPool;
- private final Cache<Long, Long> _cancelledRequests =
CacheBuilder.newBuilder()
- .expireAfterWrite(SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS,
TimeUnit.MILLISECONDS).build();
- public OpChainSchedulerService(OpChainScheduler scheduler, ExecutorService
workerPool) {
- _scheduler = scheduler;
- _workerPool = workerPool;
- }
+ private final ExecutorService _executorService;
+ private final ConcurrentHashMap<OpChainId, Future<?>> _submittedOpChainMap;
- @Override
- protected void triggerShutdown() {
- // TODO: Figure out shutdown lifecycle with graceful shutdown in mind.
- LOGGER.info("Triggered shutdown on OpChainScheduler...");
+ public OpChainSchedulerService(ExecutorService executorService) {
+ _executorService = executorService;
+ _submittedOpChainMap = new ConcurrentHashMap<>();
}
@Override
- protected void run()
- throws Exception {
- while (isRunning()) {
- OpChain operatorChain = _scheduler.next(DEFAULT_SCHEDULER_NEXT_WAIT_MS,
TimeUnit.MILLISECONDS);
- if (operatorChain == null) {
- continue;
- }
- LOGGER.trace("({}): Scheduling", operatorChain);
- _workerPool.submit(new TraceRunnable() {
+ public void register(OpChain operatorChain) {
+ Future<?> scheduledFuture = _executorService.submit(new TraceRunnable() {
@Override
public void runJob() {
boolean isFinished = false;
- boolean returnedErrorBlock = false;
+ TransferableBlock returnedErrorBlock = null;
Throwable thrown = null;
try {
LOGGER.trace("({}): Executing", operatorChain);
- // throw if the operatorChain is cancelled.
- if
(_cancelledRequests.asMap().containsKey(operatorChain.getId().getRequestId())) {
- throw new InterruptedException("Query was cancelled!");
- }
operatorChain.getStats().executing();
- // so long as there's work to be done, keep getting the next block
- // when the operator chain returns a NOOP block, then yield the
execution
- // of this to another worker
TransferableBlock result = operatorChain.getRoot().nextBlock();
- while (!result.isNoOpBlock() && !result.isEndOfStreamBlock()) {
+ while (!result.isEndOfStreamBlock()) {
result = operatorChain.getRoot().nextBlock();
}
-
- if (result.isNoOpBlock()) {
- // TODO: There should be a waiting-for-data state in
OpChainStats.
- operatorChain.getStats().queued();
- _scheduler.yield(operatorChain);
+ isFinished = true;
+ if (result.isErrorBlock()) {
+ returnedErrorBlock = result;
+ LOGGER.error("({}): Completed erroneously {} {}", operatorChain,
operatorChain.getStats(),
+ result.getDataBlock().getExceptions());
} else {
- isFinished = true;
- if (result.isErrorBlock()) {
- returnedErrorBlock = true;
- LOGGER.error("({}): Completed erroneously {} {}",
operatorChain, operatorChain.getStats(),
- result.getDataBlock().getExceptions());
- } else {
- LOGGER.debug("({}): Completed {}", operatorChain,
operatorChain.getStats());
- }
+ LOGGER.debug("({}): Completed {}", operatorChain,
operatorChain.getStats());
}
} catch (Exception e) {
LOGGER.error("({}): Failed to execute operator chain! {}",
operatorChain, operatorChain.getStats(), e);
thrown = e;
} finally {
- if (returnedErrorBlock || thrown != null) {
+ if (returnedErrorBlock != null || thrown != null) {
+ if (thrown == null) {
+ String blockMsg;
+ if (returnedErrorBlock.getDataBlock() instanceof
MetadataBlock) {
+ MetadataBlock metadataBlock = (MetadataBlock)
returnedErrorBlock.getDataBlock();
+ blockMsg = String.join(", ",
metadataBlock.getExceptions().values());
+ } else {
+ blockMsg = "Unknown";
+ }
+ thrown = new RuntimeException("Error block " + blockMsg);
+ }
cancelOpChain(operatorChain, thrown);
} else if (isFinished) {
closeOpChain(operatorChain);
}
}
}
});
- }
+ _submittedOpChainMap.put(operatorChain.getId(), scheduledFuture);
}
- /**
- * Register a new operator chain with the scheduler.
- *
- * @param operatorChain the chain to register
- */
- public final void register(OpChain operatorChain) {
- operatorChain.getStats().queued();
- _scheduler.register(operatorChain);
- LOGGER.debug("({}): Scheduler is now handling operator chain listening to
mailboxes {}. "
- + "There are a total of {} chains awaiting execution.",
operatorChain,
- operatorChain.getReceivingMailboxIds(),
- _scheduler.size());
- }
-
- /**
- * Async cancel a request. Request will not be fully cancelled until the
next time opChain is being polled.
- *
- * @param requestId requestId to be cancelled.
- */
- public final void cancel(long requestId) {
- _cancelledRequests.put(requestId, requestId);
- }
-
- /**
- * This method should be called whenever data is available for an {@link
OpChain} to consume.
- * Implementations of this method should be idempotent, it may be called in
the scenario that no data is available.
- *
- * @param opChainId the identifier of the operator chain
- */
- public final void onDataAvailable(OpChainId opChainId) {
- _scheduler.onDataAvailable(opChainId);
- }
-
- // TODO: remove this method after we pipe down the proper executor pool to
the v1 engine
- public ExecutorService getWorkerPool() {
- return _workerPool;
+ @Override
+ public void cancel(long requestId) {
+ // simple cancellation. for leaf stage this cannot be a dangling opchain
b/c they will eventually be cleared up
+ // via query timeout.
+ List<OpChainId> opChainIdsToCancel = _submittedOpChainMap.keySet()
Review Comment:
> We also need to remove the entry because there is no guarantee that the
future already starts executing
In fact the entry was never removed, even if the future was executed, right?
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java:
##########
@@ -31,151 +33,84 @@
import org.slf4j.LoggerFactory;
-/**
- * This class provides the implementation for scheduling multistage queries on
a single node based
- * on the {@link OpChainScheduler} logic that is passed in. Multistage queries
support partial execution
- * and will return a NOOP metadata block as a "yield" signal, indicating that
the next operator
- * chain ({@link OpChainScheduler#next} will be requested.
- */
-@SuppressWarnings("UnstableApiUsage")
-public class OpChainSchedulerService extends AbstractExecutionThreadService {
+public class OpChainSchedulerService implements SchedulerService {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpChainSchedulerService.class);
- /**
- * Default time scheduler is allowed to wait for a runnable OpChain to be
available.
- */
- private static final long DEFAULT_SCHEDULER_NEXT_WAIT_MS = 100;
- /**
- * Default cancel signal retention, this should be set to several times
larger than
- * {@link
org.apache.pinot.query.service.QueryConfig#DEFAULT_SCHEDULER_RELEASE_TIMEOUT_MS}.
- */
- private static final long SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS =
60_000L;
-
- private final OpChainScheduler _scheduler;
- private final ExecutorService _workerPool;
- private final Cache<Long, Long> _cancelledRequests =
CacheBuilder.newBuilder()
- .expireAfterWrite(SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS,
TimeUnit.MILLISECONDS).build();
- public OpChainSchedulerService(OpChainScheduler scheduler, ExecutorService
workerPool) {
- _scheduler = scheduler;
- _workerPool = workerPool;
- }
+ private final ExecutorService _executorService;
+ private final ConcurrentHashMap<OpChainId, Future<?>> _submittedOpChainMap;
- @Override
- protected void triggerShutdown() {
- // TODO: Figure out shutdown lifecycle with graceful shutdown in mind.
- LOGGER.info("Triggered shutdown on OpChainScheduler...");
+ public OpChainSchedulerService(ExecutorService executorService) {
+ _executorService = executorService;
+ _submittedOpChainMap = new ConcurrentHashMap<>();
}
@Override
- protected void run()
- throws Exception {
- while (isRunning()) {
- OpChain operatorChain = _scheduler.next(DEFAULT_SCHEDULER_NEXT_WAIT_MS,
TimeUnit.MILLISECONDS);
- if (operatorChain == null) {
- continue;
- }
- LOGGER.trace("({}): Scheduling", operatorChain);
- _workerPool.submit(new TraceRunnable() {
+ public void register(OpChain operatorChain) {
+ Future<?> scheduledFuture = _executorService.submit(new TraceRunnable() {
@Override
public void runJob() {
boolean isFinished = false;
- boolean returnedErrorBlock = false;
+ TransferableBlock returnedErrorBlock = null;
Throwable thrown = null;
try {
LOGGER.trace("({}): Executing", operatorChain);
- // throw if the operatorChain is cancelled.
- if
(_cancelledRequests.asMap().containsKey(operatorChain.getId().getRequestId())) {
- throw new InterruptedException("Query was cancelled!");
- }
operatorChain.getStats().executing();
- // so long as there's work to be done, keep getting the next block
- // when the operator chain returns a NOOP block, then yield the
execution
- // of this to another worker
TransferableBlock result = operatorChain.getRoot().nextBlock();
- while (!result.isNoOpBlock() && !result.isEndOfStreamBlock()) {
+ while (!result.isEndOfStreamBlock()) {
result = operatorChain.getRoot().nextBlock();
}
-
- if (result.isNoOpBlock()) {
- // TODO: There should be a waiting-for-data state in
OpChainStats.
- operatorChain.getStats().queued();
- _scheduler.yield(operatorChain);
+ isFinished = true;
+ if (result.isErrorBlock()) {
+ returnedErrorBlock = result;
+ LOGGER.error("({}): Completed erroneously {} {}", operatorChain,
operatorChain.getStats(),
+ result.getDataBlock().getExceptions());
} else {
- isFinished = true;
- if (result.isErrorBlock()) {
- returnedErrorBlock = true;
- LOGGER.error("({}): Completed erroneously {} {}",
operatorChain, operatorChain.getStats(),
- result.getDataBlock().getExceptions());
- } else {
- LOGGER.debug("({}): Completed {}", operatorChain,
operatorChain.getStats());
- }
+ LOGGER.debug("({}): Completed {}", operatorChain,
operatorChain.getStats());
}
} catch (Exception e) {
LOGGER.error("({}): Failed to execute operator chain! {}",
operatorChain, operatorChain.getStats(), e);
thrown = e;
} finally {
- if (returnedErrorBlock || thrown != null) {
+ if (returnedErrorBlock != null || thrown != null) {
+ if (thrown == null) {
+ String blockMsg;
+ if (returnedErrorBlock.getDataBlock() instanceof
MetadataBlock) {
+ MetadataBlock metadataBlock = (MetadataBlock)
returnedErrorBlock.getDataBlock();
+ blockMsg = String.join(", ",
metadataBlock.getExceptions().values());
+ } else {
+ blockMsg = "Unknown";
+ }
+ thrown = new RuntimeException("Error block " + blockMsg);
Review Comment:
Applied on 9031c6
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/OpChainExecutor.java:
##########
@@ -0,0 +1,129 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.query.runtime;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class OpChainExecutor implements ExecutorService, AutoCloseable {
Review Comment:
Removed on 306edff
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java:
##########
@@ -31,151 +33,84 @@
import org.slf4j.LoggerFactory;
-/**
- * This class provides the implementation for scheduling multistage queries on
a single node based
- * on the {@link OpChainScheduler} logic that is passed in. Multistage queries
support partial execution
- * and will return a NOOP metadata block as a "yield" signal, indicating that
the next operator
- * chain ({@link OpChainScheduler#next} will be requested.
- */
-@SuppressWarnings("UnstableApiUsage")
-public class OpChainSchedulerService extends AbstractExecutionThreadService {
+public class OpChainSchedulerService implements SchedulerService {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpChainSchedulerService.class);
- /**
- * Default time scheduler is allowed to wait for a runnable OpChain to be
available.
- */
- private static final long DEFAULT_SCHEDULER_NEXT_WAIT_MS = 100;
- /**
- * Default cancel signal retention, this should be set to several times
larger than
- * {@link
org.apache.pinot.query.service.QueryConfig#DEFAULT_SCHEDULER_RELEASE_TIMEOUT_MS}.
- */
- private static final long SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS =
60_000L;
-
- private final OpChainScheduler _scheduler;
- private final ExecutorService _workerPool;
- private final Cache<Long, Long> _cancelledRequests =
CacheBuilder.newBuilder()
- .expireAfterWrite(SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS,
TimeUnit.MILLISECONDS).build();
- public OpChainSchedulerService(OpChainScheduler scheduler, ExecutorService
workerPool) {
- _scheduler = scheduler;
- _workerPool = workerPool;
- }
+ private final ExecutorService _executorService;
+ private final ConcurrentHashMap<OpChainId, Future<?>> _submittedOpChainMap;
- @Override
- protected void triggerShutdown() {
- // TODO: Figure out shutdown lifecycle with graceful shutdown in mind.
- LOGGER.info("Triggered shutdown on OpChainScheduler...");
+ public OpChainSchedulerService(ExecutorService executorService) {
+ _executorService = executorService;
+ _submittedOpChainMap = new ConcurrentHashMap<>();
}
@Override
- protected void run()
- throws Exception {
- while (isRunning()) {
- OpChain operatorChain = _scheduler.next(DEFAULT_SCHEDULER_NEXT_WAIT_MS,
TimeUnit.MILLISECONDS);
- if (operatorChain == null) {
- continue;
- }
- LOGGER.trace("({}): Scheduling", operatorChain);
- _workerPool.submit(new TraceRunnable() {
+ public void register(OpChain operatorChain) {
+ Future<?> scheduledFuture = _executorService.submit(new TraceRunnable() {
Review Comment:
Should be better in d34f90
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java:
##########
@@ -31,151 +33,84 @@
import org.slf4j.LoggerFactory;
-/**
- * This class provides the implementation for scheduling multistage queries on
a single node based
- * on the {@link OpChainScheduler} logic that is passed in. Multistage queries
support partial execution
- * and will return a NOOP metadata block as a "yield" signal, indicating that
the next operator
- * chain ({@link OpChainScheduler#next} will be requested.
- */
-@SuppressWarnings("UnstableApiUsage")
-public class OpChainSchedulerService extends AbstractExecutionThreadService {
+public class OpChainSchedulerService implements SchedulerService {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpChainSchedulerService.class);
- /**
- * Default time scheduler is allowed to wait for a runnable OpChain to be
available.
- */
- private static final long DEFAULT_SCHEDULER_NEXT_WAIT_MS = 100;
- /**
- * Default cancel signal retention, this should be set to several times
larger than
- * {@link
org.apache.pinot.query.service.QueryConfig#DEFAULT_SCHEDULER_RELEASE_TIMEOUT_MS}.
- */
- private static final long SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS =
60_000L;
-
- private final OpChainScheduler _scheduler;
- private final ExecutorService _workerPool;
- private final Cache<Long, Long> _cancelledRequests =
CacheBuilder.newBuilder()
- .expireAfterWrite(SCHEDULER_CANCELLATION_SIGNAL_RETENTION_MS,
TimeUnit.MILLISECONDS).build();
- public OpChainSchedulerService(OpChainScheduler scheduler, ExecutorService
workerPool) {
- _scheduler = scheduler;
- _workerPool = workerPool;
- }
+ private final ExecutorService _executorService;
+ private final ConcurrentHashMap<OpChainId, Future<?>> _submittedOpChainMap;
- @Override
- protected void triggerShutdown() {
- // TODO: Figure out shutdown lifecycle with graceful shutdown in mind.
- LOGGER.info("Triggered shutdown on OpChainScheduler...");
+ public OpChainSchedulerService(ExecutorService executorService) {
+ _executorService = executorService;
+ _submittedOpChainMap = new ConcurrentHashMap<>();
}
@Override
- protected void run()
- throws Exception {
- while (isRunning()) {
- OpChain operatorChain = _scheduler.next(DEFAULT_SCHEDULER_NEXT_WAIT_MS,
TimeUnit.MILLISECONDS);
- if (operatorChain == null) {
- continue;
- }
- LOGGER.trace("({}): Scheduling", operatorChain);
- _workerPool.submit(new TraceRunnable() {
+ public void register(OpChain operatorChain) {
+ Future<?> scheduledFuture = _executorService.submit(new TraceRunnable() {
@Override
public void runJob() {
boolean isFinished = false;
- boolean returnedErrorBlock = false;
+ TransferableBlock returnedErrorBlock = null;
Throwable thrown = null;
try {
LOGGER.trace("({}): Executing", operatorChain);
- // throw if the operatorChain is cancelled.
- if
(_cancelledRequests.asMap().containsKey(operatorChain.getId().getRequestId())) {
- throw new InterruptedException("Query was cancelled!");
- }
operatorChain.getStats().executing();
- // so long as there's work to be done, keep getting the next block
- // when the operator chain returns a NOOP block, then yield the
execution
- // of this to another worker
TransferableBlock result = operatorChain.getRoot().nextBlock();
- while (!result.isNoOpBlock() && !result.isEndOfStreamBlock()) {
+ while (!result.isEndOfStreamBlock()) {
result = operatorChain.getRoot().nextBlock();
}
-
- if (result.isNoOpBlock()) {
- // TODO: There should be a waiting-for-data state in
OpChainStats.
- operatorChain.getStats().queued();
- _scheduler.yield(operatorChain);
+ isFinished = true;
+ if (result.isErrorBlock()) {
+ returnedErrorBlock = result;
+ LOGGER.error("({}): Completed erroneously {} {}", operatorChain,
operatorChain.getStats(),
+ result.getDataBlock().getExceptions());
} else {
- isFinished = true;
- if (result.isErrorBlock()) {
- returnedErrorBlock = true;
- LOGGER.error("({}): Completed erroneously {} {}",
operatorChain, operatorChain.getStats(),
- result.getDataBlock().getExceptions());
- } else {
- LOGGER.debug("({}): Completed {}", operatorChain,
operatorChain.getStats());
- }
+ LOGGER.debug("({}): Completed {}", operatorChain,
operatorChain.getStats());
}
} catch (Exception e) {
LOGGER.error("({}): Failed to execute operator chain! {}",
operatorChain, operatorChain.getStats(), e);
thrown = e;
} finally {
- if (returnedErrorBlock || thrown != null) {
+ if (returnedErrorBlock != null || thrown != null) {
+ if (thrown == null) {
+ String blockMsg;
+ if (returnedErrorBlock.getDataBlock() instanceof
MetadataBlock) {
+ MetadataBlock metadataBlock = (MetadataBlock)
returnedErrorBlock.getDataBlock();
+ blockMsg = String.join(", ",
metadataBlock.getExceptions().values());
+ } else {
+ blockMsg = "Unknown";
+ }
+ thrown = new RuntimeException("Error block " + blockMsg);
+ }
cancelOpChain(operatorChain, thrown);
} else if (isFinished) {
closeOpChain(operatorChain);
}
}
}
});
- }
+ _submittedOpChainMap.put(operatorChain.getId(), scheduledFuture);
}
- /**
- * Register a new operator chain with the scheduler.
- *
- * @param operatorChain the chain to register
- */
- public final void register(OpChain operatorChain) {
- operatorChain.getStats().queued();
- _scheduler.register(operatorChain);
- LOGGER.debug("({}): Scheduler is now handling operator chain listening to
mailboxes {}. "
- + "There are a total of {} chains awaiting execution.",
operatorChain,
- operatorChain.getReceivingMailboxIds(),
- _scheduler.size());
- }
-
- /**
- * Async cancel a request. Request will not be fully cancelled until the
next time opChain is being polled.
- *
- * @param requestId requestId to be cancelled.
- */
- public final void cancel(long requestId) {
- _cancelledRequests.put(requestId, requestId);
- }
-
- /**
- * This method should be called whenever data is available for an {@link
OpChain} to consume.
- * Implementations of this method should be idempotent, it may be called in
the scenario that no data is available.
- *
- * @param opChainId the identifier of the operator chain
- */
- public final void onDataAvailable(OpChainId opChainId) {
- _scheduler.onDataAvailable(opChainId);
- }
-
- // TODO: remove this method after we pipe down the proper executor pool to
the v1 engine
- public ExecutorService getWorkerPool() {
- return _workerPool;
+ @Override
+ public void cancel(long requestId) {
+ // simple cancellation. for leaf stage this cannot be a dangling opchain
b/c they will eventually be cleared up
+ // via query timeout.
+ List<OpChainId> opChainIdsToCancel = _submittedOpChainMap.keySet()
Review Comment:
I've changed the code, but it is not clear to me the requirements of this
method.
It is also not clear to me whether we can use `entrySet().iterator()`
because we are going to be modifying the map to remove entries from the
iterator while other threads will be adding elements. By the specification, the
behavior there is undefined. Specifically, javadoc says we can remove elements
from the entrySet, but doesn't specify what happens when other threads modify
the map while iterating.
Given that this method is not going to be executed that often and the list
shouldn't be that large, I don't think it is worth to enter in unspecified
territory in order to do not create a short living array list.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -115,12 +115,12 @@ public MultiStageBrokerRequestHandler(PinotConfiguration
config, String brokerId
long releaseMs =
config.getProperty(QueryConfig.KEY_OF_SCHEDULER_RELEASE_TIMEOUT_MS,
QueryConfig.DEFAULT_SCHEDULER_RELEASE_TIMEOUT_MS);
- _reducerScheduler = new OpChainSchedulerService(new
RoundRobinScheduler(releaseMs),
- Executors.newCachedThreadPool(new
NamedThreadFactory("query_broker_reducer_" + _reducerPort + "_port")));
- _mailboxService = new MailboxService(_reducerHostname, _reducerPort,
config, _reducerScheduler::onDataAvailable);
+ //TODO: make this configurable
+ _opChainExecutor = new OpChainExecutor(new
NamedThreadFactory("op_chain_worker_on_" + _reducerPort + "_port"));
Review Comment:
Fixed on 306edff
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
