Current state: [One of "Under Discussion", "Accepted", "Rejected"]
Discussion thread:
JIRA or Github Issue:
Released: <Doris Version>
Design Doc in PDF：For convenience of Chinese developer, I also uploaded the Chinese version design doc in PDF（使用中文的开发者也可以参考PDF版的中文版设计文档）
The current execution engine of Doris is volcano's pull model, which has some problems in single multi-core scenarios as follows：
Each instance of a standalone query corresponds to one thread of the thread pool, which brings two additional problems.
Once the thread pool is full, Doris's query engine will enter a false deadlock and fail to respond to subsequent queries. There is also a certain probability of entering a logical deadlock situation: for example, all threads are executing an instance's probe task.
Blocking arithmetic relies on the thread scheduling mechanism of the operating system, and the thread switching overhead is high (especially in the scenario of mixed system distribution)
CPU resource management is difficult, and it is difficult to achieve finer-grained resource management, and the mixed concurrency of multiple queries achieves reasonable resource scheduling.
After a large number of instances are generated for a large query, the thread pool is filled up. Small queries hardly get the chance to be scheduled, resulting in higher latency for small queries under mixed load.
Blocking arithmetic poses two problems：
Thread switching : introduces additional context switching overhead
Thread occupation: blocking threads also crowd the thread resources of a single process
Physical blocking: Includes disk I/O, network I/O, locks, and other operations that cause thread blocking
Logical blocking : Sort, HashJoin, HashAgg and other arithmetic that require full materialization
CPU resource management
In mixed-load scenarios, there is CPU contention, where large CPU-intensive queries crowd CPU resources for a long time, and small queries are difficult to be scheduled by CPU.
From the above figure, it can be different from the traditional volcano model model, in Pipeline model, it is the data that drives the control flow execution. Specifically, the data between different operators are interacted by means of push data. Each operator in the query execution process is merged and split into different pipeline tasks, and the whole execution engine uses a fixed number of thread pools, and the logic of each pipeline being scheduled is data-driven.
As an example.
R join S join T, the execution plan is split into 3 pipelines, each pipeline can be scheduled by different threads. pipeline splitting point is the blocking logic of Join build, due to the special join arithmetic, need to wait for the Build data Ready in the Probe phase before execution. Whether or not each pipeline occupies thread resources for scheduling depends on whether or not the preceding data is Ready, so the core point of scheduling for each pipeline is data-driven, and pipelines that are not ready need to release thread resources actively.
To sum up, the core idea of PipeLine can be analogous to the time-sharing scheduling system of the operating system, and the core points are as follows.
Here CPU resources = thread resources of the execution engine, scheduling strategy between processes = scheduling strategy of Pipeline, priority of processes = resource management of the query, and by putting these logics together, we can better understand the essence of the PipeLine execution engine.
Pipeline splitting for blocking operations The blocking logic is split into different pipelines for scheduling, while the pipelines between the blocking logic are driven by data. This solves the problem of blocking arithmetic taking up thread resources on the pull model, and the extra overhead caused by thread switching. At the same time, concurrent computation can be achieved between different pipelines and between the same pipeline.
PipeLine resource management After pipeline scheduling, thread resources -> pipeline -> query. It is possible to schedule different queries fairly by the amount of resources occupied by different queries. It is reasonable to share thread resources for queries with mixed load, and pipelines that occupy thread resources excessively need to release thread resources actively.
Framework of the overall execution engine：
Two new structural modules have been added here to the original QueryEngine.
Main class: BE.main calls the constructor of BlockedTaskScheduler
Start Schedule Thread, end less for loop, poll Blocked Pipeline Queue, perform state initialization and query work
Start multiple Work Threads, end less for loop polling its own task queue and other cores' task queues to find the Runnable Pipeline Task
Pipeline execution module: responsible for the translation of PipeLine Task and the original QueryPlan, PipeLine execution plan and execution logic refactoring
Pipeline execution module: responsible for the translation of PipeLine Task and the original QueryPlan, PipeLine execution plan and execution logic refactoring
Pipeline disassembly of blocking operator. The blocking operator is designed as follows: - The original Fragment Instance is disassembled into multiple PipeLine, and the basic unit of the subsequent scheduling module is the PipeLineTask.
Join Build
Sort
Agg
Scan
Exchange
ExecNode pushing transformation. (This part is to be determined, depending on the progress of implementation. The core point is the refactoring of the interface of the operator, from the original pull mode to the push mode.
This part of the transformation only designs the way of data interaction between Operators inside the Pipeline: the core advantage is to be able to implement the Pipeline Task to interrupt between any Operators and cede the thread time slice.
Parallelization between single pipelines
Pipeline's scheduling module.
As shown in the figure above, a fixed number of Work Threads, usually the number of CPU cores, are started on the Pipeline to execute different PipeLineTask. different colors represent different queries of Pipetask, each PipeTask can be scheduled on any execution thread arbitrarily, making full use of the parallelism capability of multi-core CPUs.
A globally unique thread that polls for the readiness of the PipeLine Task between queries.
while (!_shutdown.load()) {
{
auto iter = local_blocked_tasks.begin();
DateTimeValue now = DateTimeValue::local_time();
while (iter != local_blocked_tasks.end()) {
auto* task = *iter;
auto state = task->get_state();
if (state == PENDING_FINISH || task->fragment_context()->is_canceled()) {
// should cancel or should finish
if (task->is_pending_finish()) {
iter++;
} else {
_make_task_run(local_blocked_tasks, iter, ready_tasks);
}
} else if (task->query_fragments_context()->is_timeout(now)) {
LOG(WARNING) << "Timeout, query_id="
<< print_id(task->query_fragments_context()->query_id)
<< ", instance_id="
<< print_id(task->fragment_context()->get_fragment_id());
task->fragment_context()->cancel(PPlanFragmentCancelReason::TIMEOUT);
if (task->is_pending_finish()) {
iter++;
} else {
_make_task_run(local_blocked_tasks, iter, ready_tasks);
}
} else if (state == BLOCKED) {
if (!task->is_blocking()) {
task->set_state(RUNNABLE);
_make_task_run(local_blocked_tasks, iter, ready_tasks);
} else {
iter++;
}
} else {
// TODO: DCHECK the state
_make_task_run(local_blocked_tasks, iter, ready_tasks);
}
}
The execution engine starts a fixed number of execution threads for query task execution, the number can be configured by the user, and the default is the number of CPU cores.
If a PipeLine task is blocked during execution, it will be put back into the blocking queue and delivered to the polling thread for subsequent processing.
while (*marker) {
// 执行队列里获取PipeLine Task
auto task = queue->try_take(index);
if (!task) {
task = queue->steal_take(index); // work steal
if (!task) {
// TODO: The take is a stock method, rethink the logic
task = queue->take(index);
if (!task) {
continue;
}
}
}
// 实际执行对应的task，绑定执行的线程的核心
auto status = task->execute(&eos); //
task->set_previous_core_id(index);
auto pipeline_state = task->get_state();
switch (pipeline_state) {
case BLOCKED:
_blocked_task_scheduler->add_blocked_task(task); // 放置回阻塞队列
break;
case RUNNABLE:
queue->push_back(task, index); // 放置回调度队列
break;
default:
DCHECK(false);
break;
}
}
LOG(INFO) << "Stop TaskScheduler worker " << index;
Multi-Level Feedback Queue: Multi-Level Feedback Queue applied to process scheduling for BSD UNIX systems, Windows NT and subsequent Windows system operating systems, proposed by Turing Award winner Corbato.
level 1 50% of cpu time slice
level 2 25% of cpu time slice
level 3 12% cpu time slice
..
When no task exists in the corresponding level queue, the execution thread automatically gets the PipeTask of the next level priority queue for execution
// Each thread have private muti level queue
class WorkTaskQueue {
public:
explicit WorkTaskQueue() : _closed(false) {
double factor = 1;
for (int i = SUB_QUEUE_LEVEL - 1; i >= 0; --i) {
_sub_queues[i].set_factor_for_normal(factor);
factor *= LEVEL_QUEUE_TIME_FACTOR;
}
} // 初始化多级反馈队列
// 提交任务
void push(PipelineTask* task) {
size_t level = _compute_level(task); // 计算任务应该处于的队列
std::unique_lock<std::mutex> lock(_work_size_mutex);
_sub_queues[level].push_back(task);
_total_task_size++;
_wait_task.notify_one();
}
// 获取任务
PipelineTask* try_take_unprotected() {
if (_total_task_size == 0 || _closed) {
return nullptr;
}
double min_consume_time = _sub_queues[0].total_consume_time();
int idx = 0;
for (int i = 1; i < SUB_QUEUE_LEVEL; ++i) {
if (!_sub_queues[i].empty()) {
double consume_time = _sub_queues[i].total_consume_time(); // 每个反馈队列有个自己的耗时情况
if (idx == -1 || consume_time < min_consume_time) {
idx = i;
min_consume_time = consume_time;
}
}
}
auto task = _sub_queues[idx].try_take();
if (task) {
_total_task_size--;
}
return task;
}
PipeLine's execution state definition.
enum PipelineTaskState : uint8_t {
NOT_READY = 0, // do not prepare
BLOCKED = 1, // have some dependencies not finished or some conditions not met
BLOCKED_FOR_DEPENDENCY = 2,
BLOCKED_FOR_SOURCE = 3,
BLOCKED_FOR_SINK = 4,
RUNNABLE = 5, // can execute
PENDING_FINISH = 6, // compute task is over, but still hold resource. like some scan and sink task
FINISHED = 7,
CANCELED = 8
};
PipelineTaskState is the state of the dispatched PipeLineTask
SourceOperator's data is not readable, data is not ready: BLOCKED_FOR_SOURCE
SinkOperator's data is not writable: BLOCKED_FOR_SINK
|-----------------------------------------------------|
|---| transfer 2 transfer 3 | transfer 4
|-------> BLOCKED ------------| |---------------------------------------> CANCELED
|------| | | transfer 5 transfer 6|
NOT_READY ---| transfer 0 |-----> RUNNABLE ---|---------> PENDING_FINISH ------|
| | ^ | transfer 7|
|------------------------------------| |--------|---------------------------------------> FINISHED
transfer 1 transfer 9 transfer 8
* transfer 0 (NOT_READY -> BLOCKED): this pipeline task has some incomplete dependencies
* transfer 1 (NOT_READY -> RUNNABLE): this pipeline task has no incomplete dependencies
* transfer 2 (BLOCKED -> RUNNABLE): runnable condition for this pipeline task is met (e.g. get a new block from rpc)
* transfer 3 (RUNNABLE -> BLOCKED): runnable condition for this pipeline task is not met (e.g. sink operator send a block by RPC and wait for a response)
* transfer 4 (RUNNABLE -> CANCELED): current fragment is cancelled
* transfer 5 (RUNNABLE -> PENDING_FINISH): this pipeline task completed but wait for releasing resources hold by itself
* transfer 6 (PENDING_FINISH -> CANCELED): current fragment is cancelled
* transfer 7 (PENDING_FINISH -> FINISHED): this pipeline task completed and resources hold by itself have been released already
* transfer 8 (RUNNABLE -> FINISHED): this pipeline task completed and no resource need to be released
* transfer 9 (RUNNABLE -> RUNNABLE): this pipeline task yields CPU and re-enters the runnable queue if it is runnable and has occupied CPU for a max time slice
Operator is the basic unit of Pipeline execution query, which can be understood as the role of ExecNode in the original execution framework.
This part is relatively simple after abstraction, and the core logic of Operator can follow the execution logic of the previous Exec Node. The author abstracted part of the interface on the ExecNode to achieve code reuse
Operator class
abstract class, different ExecNode abstraction for the corresponding Operator. such as UnionNode - "UnionOperator, the interface is basically the same as the ExecNode
class Operator {
public:
explicit Operator(OperatorTemplate* operator_template);
virtual ~Operator() = default;
// After both sink and source need to know the cancel state.
// do cancel work
bool is_sink() const;
bool is_source() const;
// Should be call after ExecNode is constructed
virtual Status init(ExecNode* exec_node, RuntimeState* state = nullptr);
// Only result sink and data stream sink need to impl the virtual function
virtual Status init(const TDataSink& tsink) { return Status::OK(); };
// Do prepare some state of Operator
virtual Status prepare(RuntimeState* state);
// Like ExecNode when pipeline task first time be scheduled? can't block
// the pipeline should be open after dependencies is finish
// Eg a -> c, b-> c, after a, b pipeline finish, c pipeline should call open
// Now the pipeline only have one task, so the there is no performance bottleneck for the mechanism??
// but if one pipeline have multi task to parallel work, need to rethink the logic
//
// Each operator should call open_self() to prepare resource to do data compute.
// if ExecNode split to sink and source operator, open_self() should be called in sink operator
virtual Status open(RuntimeState* state);
// Release the resource, should not block the thread
//
// Each operator should call close_self() to release resource
// if ExecNode split to sink and source operator, close_self() should be called in source operator
virtual Status close(RuntimeState* state);
virtual bool can_read() { return false; } // for source
virtual bool can_write() { return false; } // for sink
// for pipeline
virtual Status get_block(RuntimeState* state, vectorized::Block* block, bool* eos) {
std::stringstream error_msg;
error_msg << " has not implements get_block";
return Status::NotSupported(error_msg.str());
}
Implementation of the OperatorBuilder class
An abstract class: mainly implements the work of ExecNode -> Operator, and multiple Operators can reuse the logic and shared resources, in order to prepare for the subsequent Pipeline parallelism
class OperatorTemplate {
protected:
const int32_t _id;
const std::string _name;
ExecNode* _related_exec_node;
std::shared_ptr<RuntimeProfile> _runtime_profile;
RuntimeState* _state = nullptr;
bool _is_closed = false;
};
The structure of PipeLine is simpler than the original execution tree of the volcano model; it is not a multinomial structure. The overall structure is a chain: a data processing chain.
Several Operators in the middle of the chain are readable and writable, stream computing, and no blocking operation is involved
PipeLine Head dependency: for example, Join Probe needs to wait for the data to be prepared by the preceding Join Build
PipeLine Tail dependency: after the execution of Pipeline is completed, it has to wake up the PipeLine Task that is waiting downstream.
Prerequisites for PipeLine Task to be schedulable.
After the above conditions are met, the PipeLine Task can enter the Runnable state. When any Operator generates EOS, or when the query is cancelled, the PipeLineTask ends and the processing of the Tail dependency is executed.
The blocking operator involved in the disassembly of the original Fragment Instance architecture into multiple PipeLine, the basic unit of the subsequent scheduling module is the PipeLineTask.
Join Build
Sort
Agg
Scan
Exchange
These operators need to be disassembled into the following structure.
Two Operators:
is_sink: the data writing side. Writes blocks to the opertor, e.g. result sink, shuffle sink. can_write means that data can be written to it, e.g. for the exchange sink operator (created by the original VDataSteamWriter), when the block written by shuffle does not exceed the limit.
is_source: Data output side, such as olap scan and shuffle, etc. can_read means the sink is executable, e.g. for olap scan operator, can_read returns true when there is data, cancel, no scan task.
For example, AggNode, here it needs to be disassembled into AggSink and AggSource to achieve the effect of blocking logic and PipeLine isolation. This part of the work follows the Operator structure mentioned above for implementation.
The intermediate states of Sink and Source are connected in series, through the original PreAggNode.
Principle: Small steps and local validation.
POC validation results based on SSB-FLAT.
Configuration : Single 48 core
Test method: 4 threads, concurrently performing SSB-FALT queries.
Change the instance of a single query and observe the change in the number of threads and execution time.
Fixed the number of instances of single query to 48, limited the number of threads, and observed the change of execution time.
To solve the problem of fair scheduling of large and small queries, the problem of resource isolation between multiple users can be solved using the Pipeline framework. A more reasonable way is to store and calculate the separation of physical resources in isolation.
Compared with the original volcano model, the Pipeline model makes the entire control flow and data flow of the execution engine to be disassembled. This will bring about an increase in engineering complexity and problem diagnosis complexity, while the blocking between Pipelines is disassembled into logical dependencies, where careless design will also bring about probable logical deadlock problems, and the performance and deadlock problems derived from such problems are difficult to locate.
[1] Leis, Viktor and Boncz, Peter and Kemper, Alfons and Neumann, Thomas, Morsel-driven parallelism: A NUMA-aware query evaluation framework for the many-core age, SIGMOD 2014
[2] Shaikhha, Amir and Dashti, Mohammad and Koch, Christoph, Push versus pull-based loop fusion in query engines, Journal of Functional Programming, Cambridge University Press, 2018
[3] Push-Based Execution in DuckDB, by Mark Raasveldt: