[multistage] implement naive round robin operator chain scheduling #9753
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| /** | ||
| * 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.executor; | ||
|
|
||
| import org.apache.pinot.query.mailbox.MailboxIdentifier; | ||
| import org.apache.pinot.query.runtime.operator.OpChain; | ||
|
|
||
|
|
||
| /** | ||
| * An interface that defines different scheduling strategies to work with the | ||
| * {@link OpChainSchedulerService}. All methods are thread safe and can be guaranteed | ||
| * to never be called concurrently - therefore all implementations may use data | ||
| * structures that are not concurrent. | ||
| */ | ||
| public interface OpChainScheduler { | ||
|
|
||
| /** | ||
| * @param operatorChain the operator chain to register | ||
| */ | ||
| void register(OpChain operatorChain); | ||
|
|
||
| /** | ||
| * This method is called whenever {@code mailbox} has new data available to consume, | ||
| * this can be useful for advanced scheduling algorithms | ||
| * | ||
| * @param mailbox the mailbox ID | ||
| */ | ||
| void onDataAvailable(MailboxIdentifier mailbox); | ||
|
There was a problem hiding this comment. is this the only trigger? (other than register) There was a problem hiding this comment. there's three possible triggers:
Triggers are defined in the implementation instead of the interface. |
||
|
|
||
| /** | ||
| * @return whether or not there is any work for the scheduler to do | ||
| */ | ||
| boolean hasNext(); | ||
|
|
||
| /** | ||
| * @return the next operator chain to process | ||
| * @throws java.util.NoSuchElementException if {@link #hasNext()} returns false | ||
| * prior to this call | ||
| */ | ||
| OpChain next(); | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,147 @@ | ||
| /** | ||
| * 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.executor; | ||
|
|
||
| import com.google.common.util.concurrent.AbstractExecutionThreadService; | ||
| import com.google.common.util.concurrent.Monitor; | ||
| import java.util.concurrent.ExecutorService; | ||
| import org.apache.pinot.common.request.context.ThreadTimer; | ||
| import org.apache.pinot.core.util.trace.TraceRunnable; | ||
| import org.apache.pinot.query.mailbox.MailboxIdentifier; | ||
| import org.apache.pinot.query.runtime.blocks.TransferableBlock; | ||
| import org.apache.pinot.query.runtime.operator.OpChain; | ||
| import org.slf4j.Logger; | ||
| 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. | ||
| * | ||
| * <p>Note that a yielded operator chain will be re-registered with the underlying scheduler. | ||
| */ | ||
| @SuppressWarnings("UnstableApiUsage") | ||
| public class OpChainSchedulerService extends AbstractExecutionThreadService { | ||
|
|
||
| private static final Logger LOGGER = LoggerFactory.getLogger(OpChainSchedulerService.class); | ||
|
|
||
| private final OpChainScheduler _scheduler; | ||
|
There was a problem hiding this comment. Add a comment saying this is guarded by monitor below? |
||
| private final ExecutorService _workerPool; | ||
|
|
||
| // anything that is guarded by this monitor should be non-blocking | ||
| private final Monitor _monitor = new Monitor(); | ||
| protected final Monitor.Guard _hasNextOrClosing = new Monitor.Guard(_monitor) { | ||
|
There was a problem hiding this comment. this is more of a question --> why one of these is private and the other is protected? There was a problem hiding this comment. that is a very good question... I think it was an autocomplete typo |
||
| @Override | ||
| public boolean isSatisfied() { | ||
| return _scheduler.hasNext() || !isRunning(); | ||
| } | ||
| }; | ||
|
|
||
| public OpChainSchedulerService(OpChainScheduler scheduler, ExecutorService workerPool) { | ||
| _scheduler = scheduler; | ||
| _workerPool = workerPool; | ||
| } | ||
|
|
||
| @Override | ||
| protected void triggerShutdown() { | ||
| // this wil just notify all waiters that the scheduler is shutting down | ||
| _monitor.enter(); | ||
| _monitor.leave(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void run() | ||
| throws Exception { | ||
| while (isRunning()) { | ||
| _monitor.enterWhen(_hasNextOrClosing); | ||
| try { | ||
| if (!isRunning()) { | ||
| return; | ||
| } | ||
|
|
||
| OpChain operatorChain = _scheduler.next(); | ||
| _workerPool.submit(new TraceRunnable() { | ||
| @Override | ||
| public void runJob() { | ||
| try { | ||
| ThreadTimer timer = operatorChain.getAndStartTimer(); | ||
|
|
||
| // 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(); | ||
|
There was a problem hiding this comment. IIUC, for the current mechanism, this will return NO-OP everytime it reaches all the way down to mailbox receive and the buffer is empty. yes? There was a problem hiding this comment. 👍 yup that's correct |
||
| while (!result.isNoOpBlock() && !result.isEndOfStreamBlock()) { | ||
| LOGGER.debug("Got block with {} rows.", result.getNumRows()); | ||
| result = operatorChain.getRoot().nextBlock(); | ||
| } | ||
|
|
||
| if (!result.isEndOfStreamBlock()) { | ||
| // not complete, needs to re-register for scheduling | ||
| register(operatorChain); | ||
| } else { | ||
| LOGGER.info("Execution time: " + timer.getThreadTimeNs()); | ||
|
There was a problem hiding this comment. for more complex scheduling algorithms, we will add a callback here to There was a problem hiding this comment. +1. good splitting point There was a problem hiding this comment. Is this logging expensive? I feel there would be a lot of logs if we log the pause every time. Can we have some class such as OpChainStats to hold the data and we decide later where to report them? There was a problem hiding this comment. oops good callout. this could be a debug log |
||
| } | ||
| } catch (Exception e) { | ||
| LOGGER.error("Failed to execute query!", e); | ||
| } | ||
|
Comment on lines
+102
to
+104
There was a problem hiding this comment. i dont think this is returnable via data blocks. echo back to @61yao's comment on timeout. maybe we need other ways to indicate out-of-norm failures. let's follow up on other PR (i don't think we handle these correctly right now either) |
||
| } | ||
| }); | ||
| } finally { | ||
| _monitor.leave(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Register a new operator chain with the scheduler. | ||
| * | ||
| * @param operatorChain the chain to register | ||
| */ | ||
| public final void register(OpChain operatorChain) { | ||
| _monitor.enter(); | ||
| try { | ||
| _scheduler.register(operatorChain); | ||
| } finally { | ||
| _monitor.leave(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * This method should be called whenever data is available in a given mailbox. | ||
| * Implementations of this method should be idempotent, it may be called in the | ||
| * scenario that no mail is available. | ||
| * | ||
| * @param mailbox the identifier of the mailbox that now has data | ||
| */ | ||
| public final void onDataAvailable(MailboxIdentifier mailbox) { | ||
| _monitor.enter(); | ||
| try { | ||
| _scheduler.onDataAvailable(mailbox); | ||
| } finally { | ||
| _monitor.leave(); | ||
| } | ||
| } | ||
|
|
||
| // TODO: remove this method after we pipe down the proper executor pool to the v1 engine | ||
|
There was a problem hiding this comment. this is a bit unfortunate, but it's how the existing code works and refactoring it would be out of scope for this PR. While it's not particularly efficient, it also isn't dangerous - V1 queries are non-blocking, so using the same worker pool for executing V1 queries (that are issued as part of V2) and V2 intermediate queries does not threaten liveness. There was a problem hiding this comment. lol should've read this first. good call out. another way is to simple decoupled the executor service used by v1 from v2. i am not sure which is better. There was a problem hiding this comment. yeah, I think we should split v1/v2 executor pools - that's probably the safest option. Alternatively we may also want three pools: a v2-intermediate pool, a v1-via-v2 pool and a v1-vanilla pool. That will allow us to make sure that clusters that run an existing v1 vanilla workload have their exposure to the v2 engine entirely limited |
||
| public ExecutorService getWorkerPool() { | ||
| return _workerPool; | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /** | ||
| * 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.executor; | ||
|
|
||
| import java.util.LinkedList; | ||
| import java.util.Queue; | ||
| import org.apache.pinot.query.mailbox.MailboxIdentifier; | ||
| import org.apache.pinot.query.runtime.operator.OpChain; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
|
|
||
| public class RoundRobinScheduler implements OpChainScheduler { | ||
| private static final Logger LOGGER = LoggerFactory.getLogger(RoundRobinScheduler.class); | ||
|
|
||
| private final Queue<OpChain> _opChainQueue = new LinkedList<>(); | ||
|
|
||
| @Override | ||
| public void register(OpChain operatorChain) { | ||
| _opChainQueue.add(operatorChain); | ||
| } | ||
|
|
||
| @Override | ||
| public void onDataAvailable(MailboxIdentifier mailbox) { | ||
| // do nothing - this doesn't change order of execution | ||
| } | ||
|
|
||
| @Override | ||
| public boolean hasNext() { | ||
| // don't use _nextOpChain.hasNext() because that may potentially create | ||
| // a new iterator that gets tossed | ||
| return !_opChainQueue.isEmpty(); | ||
| } | ||
|
|
||
| @Override | ||
| public OpChain next() { | ||
| return _opChainQueue.poll(); | ||
| } | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
freaking awesome!!!