[feature](executor) Add time sharing executor framework.
diff --git a/be/src/common/status.h b/be/src/common/status.h index 841d509..0a6c35e 100644 --- a/be/src/common/status.h +++ b/be/src/common/status.h
@@ -737,6 +737,16 @@ std::forward<T>(res).value(); \ }); +#define TEST_TRY(stmt) \ + ({ \ + auto&& res = (stmt); \ + using T = std::decay_t<decltype(res)>; \ + if (!res.has_value()) [[unlikely]] { \ + ASSERT_TRUE(res.has_value()) << "Expected success, but got error: " << res.error(); \ + } \ + std::forward<T>(res).value(); \ + }) + } // namespace doris // specify formatter for Status
diff --git a/be/src/vec/exec/executor/batch_split_scheduler.h b/be/src/vec/exec/executor/batch_split_scheduler.h new file mode 100644 index 0000000..c614428 --- /dev/null +++ b/be/src/vec/exec/executor/batch_split_scheduler.h
@@ -0,0 +1,142 @@ +//#pragma once +// +//#include <atomic> +//#include <memory> +//#include <mutex> +//#include <condition_variable> +//#include <thread> +//#include <queue> +// +//#include "common/status.h" +//#include "vec/exec/scan/split_source_connector.h" +//#include "vec/exec/scan/vfile_scanner.h" +//#include "runtime/runtime_state.h" +// +//namespace doris::vectorized { +// +//// BatchSplitScheduler 负责在独立线程中获取文件分片并动态创建 VFileScanner +//class BatchSplitScheduler { +//public: +// BatchSplitScheduler(std::shared_ptr<SplitSourceConnector> split_source, +// RuntimeState* state, +// const TFileScanNode& scan_node) +// : _split_source(split_source) +// , _state(state) +// , _scan_node(scan_node) +// , _running(false) +// , _stop(false) {} +// +// ~BatchSplitScheduler() { +// stop(); +// } +// +// Status start() { +// if (_running) { +// return Status::OK(); +// } +// _running = true; +// _stop = false; +// _fetch_thread = std::thread(&BatchSplitScheduler::_fetch_loop, this); +// return Status::OK(); +// } +// +// void stop() { +// if (!_running) { +// return; +// } +// { +// std::lock_guard<std::mutex> l(_lock); +// _stop = true; +// _cv.notify_one(); +// } +// if (_fetch_thread.joinable()) { +// _fetch_thread.join(); +// } +// _running = false; +// } +// +// // 获取一个可用的 scanner +// Status get_next_scanner(std::shared_ptr<VFileScanner>* scanner) { +// std::unique_lock<std::mutex> l(_lock); +// while (_scanners.empty() && !_stop && _running) { +// _cv.wait(l); +// } +// +// if (_stop || !_running) { +// return Status::Cancelled("BatchSplitScheduler stopped"); +// } +// +// if (!_scanners.empty()) { +// *scanner = _scanners.front(); +// _scanners.pop(); +// return Status::OK(); +// } +// +// return Status::EndOfFile("No more scanners"); +// } +// +// // 通知当前 scanner 已完成，可以获取新的分片 +// void notify_scanner_finished() { +// std::lock_guard<std::mutex> l(_lock); +// _scanner_finished = true; +// _cv.notify_one(); +// } +// +//private: +// void _fetch_loop() { +// while (true) { +// { +// std::unique_lock<std::mutex> l(_lock); +// _cv.wait(l, [this]() { +// return _stop || _scanner_finished || _scanners.empty(); +// }); +// if (_stop) { +// break; +// } +// _scanner_finished = false; +// } +// +// // 获取新的分片并创建 scanner +// bool has_next = false; +// TFileScanRange range; +// auto status = _split_source->get_next(&has_next, &range); +// if (!status.ok()) { +// LOG(WARNING) << "Failed to get next split: " << status; +// continue; +// } +// +// if (!has_next) { +// stop(); +// break; +// } +// +// // 创建新的 scanner +// auto scanner = std::make_shared<VFileScanner>(_state, _scan_node); +// status = scanner->init(_state, _scan_node, {range}, nullptr, nullptr); +// if (!status.ok()) { +// LOG(WARNING) << "Failed to init scanner: " << status; +// continue; +// } +// +// { +// std::lock_guard<std::mutex> l(_lock); +// _scanners.push(scanner); +// _cv.notify_one(); +// } +// } +// } +// +//private: +// std::shared_ptr<SplitSourceConnector> _split_source; +// RuntimeState* _state; +// TFileScanNode _scan_node; +// std::atomic<bool> _running; +// std::atomic<bool> _stop; +// std::atomic<bool> _scanner_finished{true}; // 初始为 true 以获取第一批分片 +// std::mutex _lock; +// std::condition_variable _cv; +// std::thread _fetch_thread; +// std::queue<std::shared_ptr<VFileScanner>> _scanners; +//}; +// +//} // namespace doris::vectorized
diff --git a/be/src/vec/exec/executor/listenable_future.h b/be/src/vec/exec/executor/listenable_future.h new file mode 100644 index 0000000..420dc31 --- /dev/null +++ b/be/src/vec/exec/executor/listenable_future.h
@@ -0,0 +1,250 @@ +// 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. + +#pragma once + +#include <condition_variable> +#include <exception> +#include <functional> +#include <iostream> +#include <memory> +#include <mutex> +#include <vector> + +#include "common/status.h" +#include "glog/logging.h" + +namespace doris { +namespace vectorized { + +struct Void { + bool operator==(const Void&) const { return true; } + bool operator!=(const Void&) const { return false; } +}; + +template <typename T> +class SharedListenableFuture; + +template <typename T> +class ListenableFuture { +public: + using Callback = std::function<void(const T&, const doris::Status&)>; + + ListenableFuture(const ListenableFuture&) = delete; + ListenableFuture& operator=(const ListenableFuture&) = delete; + + ListenableFuture(ListenableFuture&& other) noexcept + : _ready(other._ready), + _value(std::move(other._value)), + _status(std::move(other._status)), + _callbacks(std::move(other._callbacks)) { + other._ready = false; + } + + ListenableFuture& operator=(ListenableFuture&& other) noexcept { + if (this != &other) { + std::lock_guard<std::mutex> lock(mutex_); + std::lock_guard<std::mutex> other_lock(other.mutex_); + _ready = other._ready; + _value = std::move(other._value); + _status = std::move(other._status); + _callbacks = std::move(other._callbacks); + other._ready = false; + } + return *this; + } + + ListenableFuture() : _ready(false) {} + + void set_value(const T& value) { _execute(value, doris::Status::OK()); } + + void set_error(const doris::Status& status) { _execute({}, status); } + + void add_callback(Callback cb) { + std::lock_guard<std::mutex> lock(mutex_); + if (_ready) { + cb(_value, _status); + } else { + _callbacks.emplace_back(std::move(cb)); + } + } + + static ListenableFuture<T> create_ready(T value) { + ListenableFuture<T> future; + future.set_value(std::move(value)); + return future; + } + + template <typename U = T> + static typename std::enable_if<std::is_same<U, Void>::value, ListenableFuture<U>>::type + create_ready() { + ListenableFuture<U> future; + future.set_value(Void {}); + return future; + } + + bool is_ready() const { + std::lock_guard<std::mutex> lock(mutex_); + return _ready; + } + + bool is_done() const { + std::lock_guard<std::mutex> lock(mutex_); + return _ready && _status.ok(); + } + + bool is_error() const { + std::lock_guard<std::mutex> lock(mutex_); + return _ready && !_status.ok(); + } + + const doris::Status& get_status() const { + std::lock_guard<std::mutex> lock(mutex_); + return _status; + } + + Result<T> get() { + std::unique_lock<std::mutex> lock(mutex_); + while (!_ready) { + cv_.wait(lock); + } + if (!_status.ok()) { + return unexpected(_status); + } + return _value; + } + + SharedListenableFuture<T> share() && { return SharedListenableFuture<T>(std::move(*this)); } + + friend class SharedListenableFuture<T>; + +private: + void _execute(const T& value, const doris::Status& status) { + std::vector<Callback> tmp_callbacks; + { + std::lock_guard<std::mutex> lock(mutex_); + if (_ready) { + return; + } + + _value = value; + _status = status; + _ready = true; + tmp_callbacks.swap(_callbacks); + } + + for (auto& cb : tmp_callbacks) { + cb(_value, _status); + } + cv_.notify_all(); + } + + mutable std::mutex mutex_; + std::condition_variable cv_; + bool _ready; + T _value; + doris::Status _status = doris::Status::OK(); + std::vector<Callback> _callbacks; +}; + +template <typename T> +class SharedListenableFuture { +public: + using Callback = typename ListenableFuture<T>::Callback; + + SharedListenableFuture(const SharedListenableFuture&) = default; + SharedListenableFuture& operator=(const SharedListenableFuture&) = default; + + SharedListenableFuture(SharedListenableFuture&&) = default; + SharedListenableFuture& operator=(SharedListenableFuture&&) = default; + + explicit SharedListenableFuture(ListenableFuture<T>&& future) + : impl_(std::make_shared<ListenableFuture<T>>(std::move(future))) {} + + explicit SharedListenableFuture(std::shared_ptr<ListenableFuture<T>> future_ptr) + : impl_(std::move(future_ptr)) {} + + void add_callback(Callback cb) { return impl_->add_callback(std::move(cb)); } + + bool is_ready() const { return impl_->is_ready(); } + + bool is_done() const { return impl_->is_done(); } + + bool is_error() const { return impl_->is_error(); } + + const doris::Status& get_status() const { return impl_->get_status(); } + + Result<T> get() { return impl_->get(); } + + static SharedListenableFuture<T> create_ready(T value) { + return SharedListenableFuture<T>(ListenableFuture<T>::create_ready(std::move(value))); + } + + SharedListenableFuture() : impl_(std::make_shared<ListenableFuture<T>>()) {} + + void set_value(const T& value) { impl_->set_value(value); } + + void set_error(const doris::Status& status) { impl_->set_error(status); } + + template <typename U = T> + static typename std::enable_if<std::is_same<U, Void>::value, SharedListenableFuture<U>>::type + create_ready() { + return SharedListenableFuture<U>(ListenableFuture<U>::create_ready()); + } + +private: + std::shared_ptr<ListenableFuture<T>> impl_; +}; + +namespace listenable_future { +inline SharedListenableFuture<Void> null_future = + SharedListenableFuture<Void>::create_ready(Void {}); +} // namespace listenable_future + +} // namespace vectorized +} // namespace doris + +// 使用示例 +//int main() { +// ListenableFuture<int> future; +// +// // 添加回调 +// future.add_callback( +// [](int value) { +// printf("Got value: %d\n", value); +// }, +// [](auto e) { +// try { +// if (e) std::rethrow_exception(e); +// } catch (const std::exception& ex) { +// printf("Exception: %s\n", ex.what()); +// } +// } +// ); +// +// // 链式调用 +// auto next = future.then([](int x) { +// return x * 2; +// }).then([](int x) { +// return std::to_string(x); +// }); +// +// // 设置结果 +// future.set_value(42); +// +// return 0; +//} \ No newline at end of file
diff --git a/be/src/vec/exec/executor/split_runner.h b/be/src/vec/exec/executor/split_runner.h new file mode 100644 index 0000000..49d8dd1 --- /dev/null +++ b/be/src/vec/exec/executor/split_runner.h
@@ -0,0 +1,40 @@ +// 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. + +#pragma once +#include <memory> +#include <string> + +#include "common/status.h" +#include "vec/exec/executor/listenable_future.h" + +namespace doris { +namespace vectorized { + +class SplitRunner { +public: + virtual ~SplitRunner() = default; + virtual Status init() = 0; + virtual Result<SharedListenableFuture<Void>> process_for(std::chrono::nanoseconds duration) = 0; + virtual void close(const Status& status) = 0; + virtual bool is_finished() = 0; + virtual Status finished_status() = 0; + virtual std::string get_info() const = 0; +}; + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/stats.h b/be/src/vec/exec/executor/stats.h new file mode 100644 index 0000000..c86961b --- /dev/null +++ b/be/src/vec/exec/executor/stats.h
@@ -0,0 +1,104 @@ +//// 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. +// +//#pragma once +//#include <array> +//#include <atomic> +//#include <chrono> +//#include <cmath> +//#include <vector> +// +//namespace doris { +//namespace vectorized { +// +//class TimeStats { +//public: +// void add(std::chrono::nanoseconds duration) { +// totalTime.fetch_add(duration.count()); +// count.fetch_add(1); +// } +// +// double getAverage() const { +// auto total = totalTime.load(); +// auto cnt = count.load(); +// return cnt > 0 ? static_cast<double>(total) / cnt : 0.0; +// } +// +//private: +// std::atomic<int64_t> totalTime {0}; +// std::atomic<int64_t> count {0}; +//}; +// +//class TimeDistribution { +//public: +// void add(std::chrono::microseconds duration) { +// auto bucket = getBucket(duration.count()); +// buckets[bucket].fetch_add(1); +// count.fetch_add(1); +// } +// +//private: +// static constexpr size_t BUCKET_COUNT = 32; +// std::array<std::atomic<int64_t>, BUCKET_COUNT> buckets {}; +// std::atomic<int64_t> count {0}; +// +// static size_t getBucket(int64_t value) { +// return std::min(static_cast<size_t>(std::log2(static_cast<double>(value) + 1)), +// BUCKET_COUNT - 1); +// } +//}; +// +//class CounterStats { +//public: +// void add(int64_t value) { total.fetch_add(value); } +// +// int64_t getTotal() const { return total.load(); } +// +//private: +// std::atomic<int64_t> total {0}; +//}; +// +//class DistributionStats { +//public: +// void add(double value) { +// count.fetch_add(1); +// sum.fetch_add(value); +// updateMin(value); +// updateMax(value); +// } +// +//private: +// std::atomic<int64_t> count {0}; +// std::atomic<double> sum {0.0}; +// std::atomic<double> min {std::numeric_limits<double>::max()}; +// std::atomic<double> max {std::numeric_limits<double>::lowest()}; +// +// void updateMin(double value) { +// double current = min.load(); +// while (value < current && !min.compare_exchange_weak(current, value)) { +// } +// } +// +// void updateMax(double value) { +// double current = max.load(); +// while (value > current && !max.compare_exchange_weak(current, value)) { +// } +// } +//}; +// +//} // namespace vectorized +//} // namespace doris
diff --git a/be/src/vec/exec/executor/task_executor.h b/be/src/vec/exec/executor/task_executor.h new file mode 100644 index 0000000..e6bf89c --- /dev/null +++ b/be/src/vec/exec/executor/task_executor.h
@@ -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. + +#pragma once +#include <future> +#include <memory> +#include <optional> +#include <vector> + +#include "vec/exec/executor/listenable_future.h" +#include "vec/exec/executor/split_runner.h" +#include "vec/exec/executor/task_id.h" + +namespace doris { +namespace vectorized { + +class TaskHandle; + +class TaskExecutor { +public: + virtual ~TaskExecutor() = default; + + virtual Status init() = 0; + virtual Status start() = 0; + virtual void stop() = 0; + + virtual Result<std::shared_ptr<TaskHandle>> create_task( + const TaskId& task_id, std::function<double()> utilization_supplier, + int initial_split_concurrency, + std::chrono::nanoseconds split_concurrency_adjust_frequency, + std::optional<int> max_drivers_per_task) = 0; + + virtual Status add_task(const TaskId& task_id, std::shared_ptr<TaskHandle> task_handle) = 0; + + virtual Status remove_task(std::shared_ptr<TaskHandle> task_handle) = 0; + + virtual Result<std::vector<SharedListenableFuture<Void>>> enqueue_splits( + std::shared_ptr<TaskHandle> task_handle, bool intermediate, + const std::vector<std::shared_ptr<SplitRunner>>& splits) = 0; +}; + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/task_handle.h b/be/src/vec/exec/executor/task_handle.h new file mode 100644 index 0000000..0193ab4 --- /dev/null +++ b/be/src/vec/exec/executor/task_handle.h
@@ -0,0 +1,35 @@ +// 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. + +#pragma once + +#include "common/status.h" +#include "vec/exec/executor/task_id.h" + +namespace doris { +namespace vectorized { + +class TaskHandle { +public: + virtual ~TaskHandle() = default; + virtual Status init() = 0; + virtual bool is_closed() const = 0; + virtual TaskId task_id() const = 0; +}; + +} // namespace vectorized +} // namespace doris \ No newline at end of file
diff --git a/be/src/vec/exec/executor/task_id.h b/be/src/vec/exec/executor/task_id.h new file mode 100644 index 0000000..347bafe --- /dev/null +++ b/be/src/vec/exec/executor/task_id.h
@@ -0,0 +1,45 @@ +// 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. + +#pragma once +#include <memory> +#include <string> + +namespace doris { +namespace vectorized { + +class TaskId { +public: + TaskId(const std::string& id) : _id(id) {} + std::string to_string() const { return _id; } + bool operator==(const TaskId& other) const { return _id == other._id; } + +private: + std::string _id; +}; + +} // namespace vectorized +} // namespace doris + +namespace std { +template <> +struct hash<doris::vectorized::TaskId> { + size_t operator()(const doris::vectorized::TaskId& task_id) const { + return std::hash<std::string> {}(task_id.to_string()); + } +}; +} // namespace std
diff --git a/be/src/vec/exec/executor/ticker.h b/be/src/vec/exec/executor/ticker.h new file mode 100644 index 0000000..5b36e26 --- /dev/null +++ b/be/src/vec/exec/executor/ticker.h
@@ -0,0 +1,49 @@ +// 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. + +#pragma once +#include <atomic> +#include <chrono> +#include <mutex> + +namespace doris { +namespace vectorized { + +class Ticker { +public: + virtual ~Ticker() = default; + + /** + * Returns the number of nanoseconds since a fixed reference point + */ + virtual int64_t read() const = 0; + +protected: + Ticker() = default; +}; + +class SystemTicker : public Ticker { +public: + int64_t read() const override { + return std::chrono::duration_cast<std::chrono::nanoseconds>( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + } +}; + +} // namespace vectorized +} // namespace doris \ No newline at end of file
diff --git a/be/src/vec/exec/executor/time_sharing/multilevel_split_queue.cpp b/be/src/vec/exec/executor/time_sharing/multilevel_split_queue.cpp new file mode 100644 index 0000000..9434b6c --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/multilevel_split_queue.cpp
@@ -0,0 +1,261 @@ +// 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. + +#include "vec/exec/executor/time_sharing/multilevel_split_queue.h" + +#include <algorithm> +#include <cmath> +#include <numeric> +#include <unordered_set> + +namespace doris { +namespace vectorized { + +MultilevelSplitQueue::MultilevelSplitQueue(double level_time_multiplier) + : _level_time_multiplier(level_time_multiplier), _level_waiting_splits() { + for (size_t i = 0; i < LEVEL_THRESHOLD_SECONDS.size(); ++i) { + _level_scheduled_time[i].store(0); + _level_min_priority[i].store(-1); + } +} + +int MultilevelSplitQueue::compute_level(int64_t scheduled_nanos) { + double scheduled_seconds = static_cast<double>(scheduled_nanos) / 1000000000.0; + + for (int level = 0; level < (LEVEL_THRESHOLD_SECONDS.size() - 1); level++) { + if (scheduled_seconds < LEVEL_THRESHOLD_SECONDS[level + 1]) { + return level; + } + } + + return LEVEL_THRESHOLD_SECONDS.size() - 1; +} + +/** + * 'Charges' the quanta run time to the task <i>and</i> the level it belongs to in + * an effort to maintain the target thread utilization ratios between levels and to + * maintain fairness within a level. + * <p> + * Consider an example split where a read hung for several minutes. This is either a bug + * or a failing dependency. In either case we do not want to charge the task too much, + * and we especially do not want to charge the level too much - i.e. cause other queries + * in this level to starve. + * + * @return the new priority for the task + */ +Priority MultilevelSplitQueue::update_priority(const Priority& old_priority, int64_t quanta_nanos, + int64_t scheduled_nanos) { + int old_level = old_priority.level(); + int new_level = compute_level(scheduled_nanos); + + int64_t level_contribution = std::min(quanta_nanos, LEVEL_CONTRIBUTION_CAP); + + if (old_level == new_level) { + _level_scheduled_time[old_level].fetch_add(level_contribution); + return Priority(old_level, old_priority.level_priority() + quanta_nanos); + } + + int64_t remaining_level_contribution = level_contribution; + int64_t remaining_task_time = quanta_nanos; + + for (int current_level = old_level; current_level < new_level; ++current_level) { + int64_t level_time_threshold = + static_cast<int64_t>((LEVEL_THRESHOLD_SECONDS[current_level + 1] - + LEVEL_THRESHOLD_SECONDS[current_level]) * + 1e9); + + int64_t time_accrued_to_level = + std::min(level_time_threshold, remaining_level_contribution); + _level_scheduled_time[current_level].fetch_add(time_accrued_to_level); + remaining_level_contribution -= time_accrued_to_level; + remaining_task_time -= time_accrued_to_level; + } + + _level_scheduled_time[new_level].fetch_add(remaining_level_contribution); + int64_t new_level_min_priority = get_level_min_priority(new_level, scheduled_nanos); + return Priority(new_level, new_level_min_priority + remaining_task_time); +} + +int64_t MultilevelSplitQueue::get_level_min_priority(int level, int64_t task_thread_usage_nanos) { + int64_t expected = -1; + _level_min_priority[level].compare_exchange_strong(expected, task_thread_usage_nanos); + return _level_min_priority[level].load(); +} + +void MultilevelSplitQueue::offer(std::shared_ptr<PrioritizedSplitRunner> split) { + split->set_ready(); + int level = split->priority().level(); + std::unique_lock<std::mutex> lock(_mutex); + _offer_locked(split, level, lock); +} + +void MultilevelSplitQueue::_offer_locked(std::shared_ptr<PrioritizedSplitRunner> split, int level, + std::unique_lock<std::mutex>& lock) { + if (_level_waiting_splits[level].empty()) { + // Accesses to _level_scheduled_time are not synchronized, so we have a data race + // here - our level time math will be off. However, the staleness is bounded by + // the fact that only running splits that complete during this computation + // can update the level time. Therefore, this is benign. + int64_t level0_time = _get_level0_target_time(lock); + int64_t level_expected_time = + static_cast<int64_t>(level0_time / std::pow(_level_time_multiplier, level)); + int64_t delta = level_expected_time - _level_scheduled_time[level].load(); + _level_scheduled_time[level].fetch_add(delta); + } + + _level_waiting_splits[level].push(split); + _not_empty.notify_all(); +} + +std::shared_ptr<PrioritizedSplitRunner> MultilevelSplitQueue::take() { + std::unique_lock<std::mutex> lock(_mutex); + + while (!_interrupted) { + auto split = _poll_split(lock); + if (split) { + if (split->update_level_priority()) { + _offer_locked(split, split->priority().level(), lock); + continue; + } + int selected_level = split->priority().level(); + _level_min_priority[selected_level].store(split->priority().level_priority()); + return split; + } + + _not_empty.wait(lock); + } + return nullptr; +} + +/** + * Attempts to give each level a target amount of scheduled time, which is configurable + * using levelTimeMultiplier. + * <p> + * This function selects the level that has the lowest ratio of actual to the target time + * with the objective of minimizing deviation from the target scheduled time. From this level, + * we pick the split with the lowest priority. + */ +std::shared_ptr<PrioritizedSplitRunner> MultilevelSplitQueue::_poll_split( + std::unique_lock<std::mutex>& lock) { + int64_t target_scheduled_time = _get_level0_target_time(lock); + double worst_ratio = 1.0; + int selected_level = -1; + + for (int level = 0; level < LEVEL_THRESHOLD_SECONDS.size(); ++level) { + if (!_level_waiting_splits[level].empty()) { + int64_t level_time = _level_scheduled_time[level].load(); + double ratio = (level_time == 0) ? 0 + : static_cast<double>(target_scheduled_time) / + (1.0 * level_time); + + if (selected_level == -1 || ratio > worst_ratio) { + worst_ratio = ratio; + selected_level = level; + } + } + target_scheduled_time = + static_cast<int64_t>(std::round(target_scheduled_time / _level_time_multiplier)); + } + + if (selected_level == -1) return nullptr; + + auto result = _level_waiting_splits[selected_level].top(); + _level_waiting_splits[selected_level].pop(); + return result; +} + +void MultilevelSplitQueue::remove(std::shared_ptr<PrioritizedSplitRunner> split) { + std::lock_guard<std::mutex> lock(_mutex); + + for (auto& level_queue : _level_waiting_splits) { + std::priority_queue<std::shared_ptr<PrioritizedSplitRunner>, + std::vector<std::shared_ptr<PrioritizedSplitRunner>>, + SplitRunnerComparator> + new_queue; + + while (!level_queue.empty()) { + auto current = level_queue.top(); + level_queue.pop(); + if (current != split) { + new_queue.emplace(std::move(current)); + } + } + level_queue.swap(new_queue); + } + + if (std::all_of(_level_waiting_splits.begin(), _level_waiting_splits.end(), + [](const auto& q) { return q.empty(); })) { + _not_empty.notify_all(); + } +} + +void MultilevelSplitQueue::remove_all( + const std::vector<std::shared_ptr<PrioritizedSplitRunner>>& splits) { + std::lock_guard<std::mutex> lock(_mutex); + + std::unordered_set<std::shared_ptr<PrioritizedSplitRunner>> to_remove(splits.begin(), + splits.end()); + + for (auto& level_queue : _level_waiting_splits) { + std::priority_queue<std::shared_ptr<PrioritizedSplitRunner>, + std::vector<std::shared_ptr<PrioritizedSplitRunner>>, + SplitRunnerComparator> + new_queue; + + while (!level_queue.empty()) { + auto current = level_queue.top(); + level_queue.pop(); + if (!to_remove.count(current)) { + new_queue.emplace(std::move(current)); + } + } + level_queue.swap(new_queue); + } + + if (std::all_of(_level_waiting_splits.begin(), _level_waiting_splits.end(), + [](const auto& q) { return q.empty(); })) { + _not_empty.notify_all(); + } +} + +size_t MultilevelSplitQueue::size() const { + std::lock_guard<std::mutex> lock(_mutex); + return std::accumulate(_level_waiting_splits.begin(), _level_waiting_splits.end(), size_t(0), + [](size_t sum, const auto& queue) { return sum + queue.size(); }); +} + +void MultilevelSplitQueue::interrupt() { + std::lock_guard<std::mutex> lock(_mutex); + _interrupted = true; + _not_empty.notify_all(); +} + +int64_t MultilevelSplitQueue::_get_level0_target_time(std::unique_lock<std::mutex>& lock) { + int64_t level0_target_time = _level_scheduled_time[0].load(); + double current_multiplier = _level_time_multiplier; + + for (int level = 0; level < LEVEL_THRESHOLD_SECONDS.size(); ++level) { + current_multiplier /= _level_time_multiplier; + int64_t level_time = _level_scheduled_time[level].load(); + level0_target_time = + std::max(level0_target_time, static_cast<int64_t>(level_time / current_multiplier)); + } + return level0_target_time; +} + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/time_sharing/multilevel_split_queue.h b/be/src/vec/exec/executor/time_sharing/multilevel_split_queue.h new file mode 100644 index 0000000..50eb6a3 --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/multilevel_split_queue.h
@@ -0,0 +1,94 @@ +// 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. + +#pragma once +#include <array> +#include <condition_variable> +#include <memory> +#include <mutex> +#include <queue> + +#include "common/factory_creator.h" +#include "vec/exec/executor/time_sharing/prioritized_split_runner.h" +#include "vec/exec/executor/time_sharing/priority.h" + +namespace doris { +namespace vectorized { + +struct SplitRunnerComparator { + bool operator()(const std::shared_ptr<PrioritizedSplitRunner>& a, + const std::shared_ptr<PrioritizedSplitRunner>& b) const { + const auto a_priority = a->priority().level_priority(); + const auto b_priority = b->priority().level_priority(); + if (a_priority != b_priority) { + return a_priority > b_priority; + } + + return a->worker_id() > b->worker_id(); + } +}; + +class MultilevelSplitQueue { + ENABLE_FACTORY_CREATOR(MultilevelSplitQueue); + +public: + static constexpr std::array<int, 5> LEVEL_THRESHOLD_SECONDS = {0, 1, 10, 60, 300}; + static constexpr int64_t LEVEL_CONTRIBUTION_CAP = 30 * 1000000000LL; // 30 seconds in nanos + + explicit MultilevelSplitQueue(double level_time_multiplier); + + static int compute_level(int64_t scheduled_nanos); + + Priority update_priority(const Priority& old_priority, int64_t quanta_nanos, + int64_t scheduled_nanos); + int64_t get_level_min_priority(int level, int64_t scheduled_nanos); + void offer(std::shared_ptr<PrioritizedSplitRunner> split); + std::shared_ptr<PrioritizedSplitRunner> take(); + size_t size() const; + void remove(std::shared_ptr<PrioritizedSplitRunner> split); + void remove_all(const std::vector<std::shared_ptr<PrioritizedSplitRunner>>& splits); + void interrupt(); + + int64_t level_scheduled_time(int level) const { return _level_scheduled_time[level].load(); } + +private: + int64_t _get_level0_target_time(std::unique_lock<std::mutex>& lock); + std::shared_ptr<PrioritizedSplitRunner> _poll_split(std::unique_lock<std::mutex>& lock); + void _offer_locked(std::shared_ptr<PrioritizedSplitRunner> split, int level, + std::unique_lock<std::mutex>& lock); + + const double _level_time_multiplier; + + std::array<std::priority_queue<std::shared_ptr<PrioritizedSplitRunner>, + std::vector<std::shared_ptr<PrioritizedSplitRunner>>, + SplitRunnerComparator>, + LEVEL_THRESHOLD_SECONDS.size()> + _level_waiting_splits; + + std::array<std::atomic<int64_t>, LEVEL_THRESHOLD_SECONDS.size()> _level_scheduled_time; + std::array<std::atomic<int64_t>, LEVEL_THRESHOLD_SECONDS.size()> _level_min_priority; + + std::atomic<bool> _interrupted {false}; + mutable std::mutex _mutex; + std::condition_variable _not_empty; + + // std::array<std::shared_ptr<CounterStats>, LEVEL_THRESHOLD_SECONDS.size()> + // _selected_level_counters; +}; + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/time_sharing/prioritized_split_runner.cpp b/be/src/vec/exec/executor/time_sharing/prioritized_split_runner.cpp new file mode 100644 index 0000000..a9e97ec --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/prioritized_split_runner.cpp
@@ -0,0 +1,207 @@ +// 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. + +#include "vec/exec/executor/time_sharing/prioritized_split_runner.h" + +#include <fmt/format.h> + +#include <chrono> +#include <functional> +#include <thread> + +#include "vec/exec/executor/time_sharing/time_sharing_task_handle.h" + +namespace doris { +namespace vectorized { + +std::atomic<int64_t> PrioritizedSplitRunner::_next_worker_id(0); + +PrioritizedSplitRunner::PrioritizedSplitRunner(std::shared_ptr<TimeSharingTaskHandle> task_handle, + int split_id, + std::shared_ptr<SplitRunner> split_runner, + std::shared_ptr<Ticker> ticker) + : _task_handle(std::move(task_handle)), + _split_id(split_id), + _worker_id(_next_worker_id.fetch_add(1, std::memory_order_relaxed)), + _split_runner(std::move(split_runner)), + _ticker(ticker) { + update_level_priority(); +} + +//PrioritizedSplitRunner::PrioritizedSplitRunner(CounterStats& globalCpuTimeMicros, +// CounterStats& globalScheduledTimeMicros, +// TimeStats& blockedQuantaWallTime, +// TimeStats& unblockedQuantaWallTime) +// : _created_nanos(std::chrono::steady_clock::now().time_since_epoch().count()), +// _split_id(0), +// _worker_id(_next_worker_id.fetch_add(1, std::memory_order_relaxed)), +// _ticker(std::make_shared<SystemTicker>()), +// _global_cpu_time_micros(globalCpuTimeMicros), +// _global_scheduled_time_micros(globalScheduledTimeMicros), +// _blocked_quanta_wall_time(blockedQuantaWallTime), +// _unblocked_quanta_wall_time(unblockedQuantaWallTime) { +//} + +Status PrioritizedSplitRunner::init() { + return _split_runner->init(); +} + +std::shared_ptr<TimeSharingTaskHandle> PrioritizedSplitRunner::task_handle() const { + return _task_handle; +} + +SharedListenableFuture<Void> PrioritizedSplitRunner::finished_future() { + return _finished_future; +} + +bool PrioritizedSplitRunner::is_closed() const { + return _closed.load(); +} + +void PrioritizedSplitRunner::close(const Status& status) { + if (!_closed.exchange(true)) { + _split_runner->close(status); + } +} + +int64_t PrioritizedSplitRunner::created_nanos() const { + return _created_nanos; +} + +bool PrioritizedSplitRunner::is_finished() { + bool finished = _split_runner->is_finished(); + if (finished) { + _finished_future.set_value({}); + } + return finished || _closed.load() || _task_handle->is_closed(); +} + +Status PrioritizedSplitRunner::finished_status() { + return _split_runner->finished_status(); +} + +int64_t PrioritizedSplitRunner::scheduled_nanos() const { + return _scheduled_nanos.load(); +} + +Result<SharedListenableFuture<Void>> PrioritizedSplitRunner::process() { + if (is_closed()) { + SharedListenableFuture<Void> future; + future.set_value({}); + return future; + } + + auto start_nanos = std::chrono::steady_clock::now().time_since_epoch().count(); + int64_t expected = 0; + _start.compare_exchange_strong(expected, start_nanos); + _last_ready.compare_exchange_strong(expected, start_nanos); + _process_calls.fetch_add(1); + + _wait_nanos.fetch_add(start_nanos - _last_ready.load()); + + auto process_start_time = _ticker->read(); + auto blocked = _split_runner->process_for(SPLIT_RUN_QUANTA); + if (!blocked.has_value()) { + return unexpected(std::move(blocked).error()); + } + auto process_end_time = _ticker->read(); + auto quanta_scheduled_nanos = process_end_time - process_start_time; + + _scheduled_nanos.fetch_add(quanta_scheduled_nanos); + + // _priority.store(_task_handle->add_scheduled_nanos(quanta_scheduled_nanos)); + + { + std::lock_guard<std::mutex> lock(_priority_mutex); + _priority = _task_handle->add_scheduled_nanos(quanta_scheduled_nanos); + } + + return blocked; +} + +void PrioritizedSplitRunner::set_ready() { + _last_ready.store(_ticker->read()); +} + +/** + * Updates the (potentially stale) priority value cached in this object. + * This should be called when this object is outside the queue. + * + * @return true if the level changed. + */ +//bool PrioritizedSplitRunner::update_level_priority() { +// Priority new_priority = _task_handle->priority(); +// Priority old_priority = _priority.exchange(new_priority); +// return new_priority.level() != old_priority.level(); +//} + +bool PrioritizedSplitRunner::update_level_priority() { + std::lock_guard<std::mutex> lock(_priority_mutex); + Priority new_priority = _task_handle->priority(); + Priority old_priority = _priority; + _priority = new_priority; + return new_priority.level() != old_priority.level(); +} + +// +//void PrioritizedSplitRunner::reset_level_priority() { +// _priority.store(_task_handle->reset_level_priority()); +//} + +void PrioritizedSplitRunner::reset_level_priority() { + std::lock_guard<std::mutex> lock(_priority_mutex); + _priority = _task_handle->reset_level_priority(); +} + +int64_t PrioritizedSplitRunner::worker_id() const { + return _worker_id; +} + +int PrioritizedSplitRunner::split_id() const { + return _split_id; +} + +//Priority PrioritizedSplitRunner::priority() const { +// return _priority.load(); +//} + +Priority PrioritizedSplitRunner::priority() const { + std::lock_guard<std::mutex> lock(_priority_mutex); + return _priority; +} + +std::string PrioritizedSplitRunner::get_info() const { + // const auto now = std::chrono::steady_clock::now(); + // const int64_t current_time = now.time_since_epoch().count(); + // const int64_t wall_time = (current_time - _start.load()) / 1'000'000; // 转换为毫秒 + + // return fmt::format( + // "Split {:<15}-{} {} (start = {:.1f}, wall = {} ms, cpu = {} ms, wait = {} ms, calls = {})", + // _task_handle->get_task_id(), // 假设返回字符串类型 + // _split_id, // 整数类型 + // _split_runner->get_info(), // 需要确保SplitRunner有get_info() + // _start.load() / 1'000'000.0, // 转换为毫秒并保留1位小数 + // wall_time, // 已计算的耗时 + // _global_cpu_time_micros.get() / 1000, // 微秒转毫秒 + // _wait_nanos.load() / 1'000'000, // 纳秒转毫秒 + // _process_calls.load() // 调用次数 + // ); + return ""; +} + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/time_sharing/prioritized_split_runner.h b/be/src/vec/exec/executor/time_sharing/prioritized_split_runner.h new file mode 100644 index 0000000..b7e396b --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/prioritized_split_runner.h
@@ -0,0 +1,93 @@ +// 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. + +#pragma once +#include <atomic> +#include <memory> + +#include "common/factory_creator.h" +#include "vec/exec/executor/listenable_future.h" +#include "vec/exec/executor/split_runner.h" +#include "vec/exec/executor/stats.h" +#include "vec/exec/executor/ticker.h" +#include "vec/exec/executor/time_sharing/priority.h" + +namespace doris { +namespace vectorized { + +class TimeSharingTaskHandle; + +class PrioritizedSplitRunner : public std::enable_shared_from_this<PrioritizedSplitRunner> { + ENABLE_FACTORY_CREATOR(PrioritizedSplitRunner); + +public: + static constexpr auto SPLIT_RUN_QUANTA = std::chrono::seconds(1); + + PrioritizedSplitRunner(std::shared_ptr<TimeSharingTaskHandle> task_handle, int split_id, + std::shared_ptr<SplitRunner> split_runner, + std::shared_ptr<Ticker> ticker); + + // PrioritizedSplitRunner(CounterStats& globalCpuTimeMicros, + // CounterStats& globalScheduledTimeMicros, + // TimeStats& blockedQuantaWallTime, TimeStats& unblockedQuantaWallTime); + + Status init(); + + virtual ~PrioritizedSplitRunner() = default; + + std::shared_ptr<TimeSharingTaskHandle> task_handle() const; + SharedListenableFuture<Void> finished_future(); + bool is_closed() const; + void close(const Status& status); + int64_t created_nanos() const; + bool is_finished(); + Status finished_status(); + int64_t scheduled_nanos() const; + Result<SharedListenableFuture<Void>> process(); + void set_ready(); + bool update_level_priority(); + void reset_level_priority(); + int64_t worker_id() const; + int split_id() const; + virtual Priority priority() const; + + std::string get_info() const; + +private: + static std::atomic<int64_t> _next_worker_id; + + const int64_t _created_nanos {std::chrono::steady_clock::now().time_since_epoch().count()}; + std::shared_ptr<TimeSharingTaskHandle> _task_handle; + const int _split_id; + const int64_t _worker_id; + std::shared_ptr<SplitRunner> _split_runner; + std::shared_ptr<Ticker> _ticker; + SharedListenableFuture<Void> _finished_future {}; + + std::atomic<bool> _closed {false}; + // std::atomic<Priority> _priority {Priority {0, 0}}; + Priority _priority {0, 0}; + mutable std::mutex _priority_mutex; + std::atomic<int64_t> _last_ready {0}; + std::atomic<int64_t> _start {0}; + std::atomic<int64_t> _scheduled_nanos {0}; + std::atomic<int64_t> _wait_nanos {0}; + std::atomic<int> _process_calls {0}; +}; + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/time_sharing/priority.h b/be/src/vec/exec/executor/time_sharing/priority.h new file mode 100644 index 0000000..f26e879 --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/priority.h
@@ -0,0 +1,67 @@ +// 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. + +#pragma once +#include <atomic> +#include <sstream> +#include <string> + +namespace doris { +namespace vectorized { + +/** + * Task (and split) priority is composed of a level and a within-level + * priority. Level decides which queue the split is placed in, while + * within-level priority decides which split is executed next in that level. + * <p> + * Tasks move from a lower to higher level as they exceed level thresholds + * of total scheduled time accrued to a task. + * <p> + * The priority within a level increases with the scheduled time accumulated + * in that level. This is necessary to achieve fairness when tasks acquire + * scheduled time at varying rates. + * <p> + * However, this priority is <b>not</b> equal to the task total accrued + * scheduled time. When a task graduates to a higher level, the level + * priority is set to the minimum current priority in the new level. This + * allows us to maintain instantaneous fairness in terms of scheduled time. + */ +class Priority { +public: + Priority(int level, int64_t level_priority) : _level(level), _level_priority(level_priority) {} + + int level() const { return _level; } + int64_t level_priority() const { return _level_priority; } + + bool operator<(const Priority& other) const { + if (_level != other._level) return _level < other._level; + return _level_priority < other._level_priority; + } + + std::string to_string() const { + std::ostringstream os; + os << "Priority(level=" << _level << ", priority=" << _level_priority << ")"; + return os.str(); + } + +private: + int _level; + int64_t _level_priority; +}; + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/time_sharing/split_concurrency_controller.h b/be/src/vec/exec/executor/time_sharing/split_concurrency_controller.h new file mode 100644 index 0000000..3c803cc --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/split_concurrency_controller.h
@@ -0,0 +1,88 @@ +// 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. +#pragma once + +#include <chrono> +#include <cmath> +#include <cstdint> +#include <stdexcept> + +#include "common/factory_creator.h" +#include "common/status.h" + +namespace doris::vectorized { + +class SplitConcurrencyController { + ENABLE_FACTORY_CREATOR(SplitConcurrencyController); + +public: + SplitConcurrencyController(int initial_concurrency, + std::chrono::nanoseconds adjustment_interval) + : _adjustment_interval_nanos(adjustment_interval), + _target_concurrency(initial_concurrency) {} + + void update(uint64_t nanos, double utilization, int current_concurrency) { + _validate_args(nanos, utilization, current_concurrency); + _thread_nanos_since_adjust += nanos; + if (_should_increase_concurrency(current_concurrency, utilization)) { + _reset_adjust_counter(); + ++_target_concurrency; + } + } + + int target_concurrency() const { return _target_concurrency; } + + void split_finished(uint64_t split_nanos, double utilization, int current_concurrency) { + _validate_args(split_nanos, utilization, current_concurrency); + if (_should_adjust(split_nanos)) { + if (utilization > TARGET_UTIL && _target_concurrency > 1) { + _reset_adjust_counter(); + --_target_concurrency; + } else if (utilization < TARGET_UTIL && current_concurrency >= _target_concurrency) { + _reset_adjust_counter(); + ++_target_concurrency; + } + } + } + +private: + static constexpr double TARGET_UTIL = 0.5; + const std::chrono::nanoseconds _adjustment_interval_nanos; + int _target_concurrency; + uint64_t _thread_nanos_since_adjust = 0; + + void _validate_args(uint64_t nanos, double util, int concurrency) const { + CHECK(nanos >= 0) << "Negative nanos"; + CHECK(std::isfinite(util)) << "Invalid utilization"; + CHECK(util >= 0) << "Negative utilization"; + CHECK(concurrency >= 0) << "Negative concurrency"; + } + + bool _should_increase_concurrency(int curr_concurrency, double util) const { + return _thread_nanos_since_adjust >= _adjustment_interval_nanos.count() && + util < TARGET_UTIL && curr_concurrency >= _target_concurrency; + } + + bool _should_adjust(uint64_t split_nanos) const { + return _thread_nanos_since_adjust >= _adjustment_interval_nanos.count() || + _thread_nanos_since_adjust >= split_nanos; + } + + void _reset_adjust_counter() { _thread_nanos_since_adjust = 0; } +}; + +} // namespace doris::vectorized
diff --git a/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.cpp b/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.cpp new file mode 100644 index 0000000..82907f1 --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.cpp
@@ -0,0 +1,408 @@ +// 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. + +#include "vec/exec/executor/time_sharing/time_sharing_task_executor.h" + +#include <functional> +#include <iostream> + +#include "util/defer_op.h" +#include "util/thread.h" +#include "util/threadpool.h" +#include "vec/exec/executor/time_sharing/time_sharing_task_handle.h" + +namespace doris { +namespace vectorized { + +TimeSharingTaskExecutor::TimeSharingTaskExecutor( + int runner_threads, int min_concurrency, int guaranteed_concurrency_per_task, + int max_concurrency_per_task, std::shared_ptr<Ticker> ticker, + std::chrono::milliseconds stuck_split_warning_threshold, + std::shared_ptr<MultilevelSplitQueue> split_queue) + : _runner_threads(runner_threads), + _min_concurrency(min_concurrency), + _guaranteed_concurrency_per_task(guaranteed_concurrency_per_task), + _max_concurrency_per_task(max_concurrency_per_task), + _ticker(ticker != nullptr ? ticker : std::make_shared<SystemTicker>()), + _stuck_split_warning_threshold(stuck_split_warning_threshold), + _waiting_splits(split_queue != nullptr ? std::move(split_queue) + : std::make_shared<MultilevelSplitQueue>(2)) {} + +Status TimeSharingTaskExecutor::init() { + ThreadPoolBuilder builder("SplitRunner"); + builder.set_min_threads(_runner_threads).set_max_threads(_runner_threads * 2); + RETURN_IF_ERROR(builder.build(&_thread_pool)); + return Status::OK(); +} + +TimeSharingTaskExecutor::~TimeSharingTaskExecutor() { + if (!_stopped.exchange(true)) { + stop(); + } + + std::vector<std::shared_ptr<PrioritizedSplitRunner>> splits_to_destroy; + { + std::lock_guard<std::mutex> guard(_mutex); + for (auto& [task_id, handle] : _tasks) { + if (handle->is_closed()) { + LOG(WARNING) << "task is already destroyed, task_id: " << task_id.to_string(); + continue; + } + auto task_splits = handle->close(); + splits_to_destroy.insert(splits_to_destroy.end(), + std::make_move_iterator(task_splits.begin()), + std::make_move_iterator(task_splits.end())); + _record_leaf_splits_size(guard); + } + //_tasks.clear(); + //_all_splits.clear(); + //_intermediate_splits.clear(); + //_blocked_splits.clear(); + _waiting_splits->remove_all(splits_to_destroy); + } + + if (splits_to_destroy.empty()) { + return; + } + + for (auto& split : splits_to_destroy) { + split->close(Status::OK()); + } + + if (_thread_pool) { + _thread_pool->shutdown(); + } +} + +Status TimeSharingTaskExecutor::start() { + std::lock_guard<std::mutex> guard(_mutex); + for (int i = 0; i < _runner_threads; ++i) { + RETURN_IF_ERROR(_add_runner_thread()); + } + return Status::OK(); +} + +void TimeSharingTaskExecutor::stop() { + _waiting_splits->interrupt(); + { + std::lock_guard<std::mutex> guard(_mutex); + _stopped = true; + } + if (_thread_pool) { + _thread_pool->shutdown(); + } +} + +Status TimeSharingTaskExecutor::_add_runner_thread() { + return _thread_pool->submit_func([this]() { + Thread::set_self_name("SplitRunner"); + + while (!_stopped) { + std::shared_ptr<PrioritizedSplitRunner> split; + split = _waiting_splits->take(); + if (!split) { + return; + } + + // ScopedRunnerTracker tracker(_mutex, _running_splits, split); + { + std::lock_guard<std::mutex> guard(_mutex); + _running_splits.insert(split); + } + Defer defer {[&]() { + std::lock_guard<std::mutex> guard(_mutex); + _running_splits.erase(split); + }}; + Result<SharedListenableFuture<Void>> blocked_future_result = split->process(); + if (!blocked_future_result.has_value()) { + return; + } + auto blocked_future = blocked_future_result.value(); + + if (split->is_finished()) { + _split_finished(split, split->finished_status()); + } else { + std::lock_guard<std::mutex> guard(_mutex); + if (blocked_future.is_done()) { + _waiting_splits->offer(split); + } else { + _blocked_splits[split] = blocked_future; + + _blocked_splits[split].add_callback( + [this, split](const Void& value, const Status& status) { + if (status.ok()) { + std::lock_guard<std::mutex> guard(_mutex); + _blocked_splits.erase(split); + split->reset_level_priority(); + _waiting_splits->offer(split); + } else { + LOG(WARNING) << "blocked split is failed, split_id: " + << split->split_id() << ", status: " << status; + _split_finished(split, status); + } + }); + } + } + } + }); +} + +Result<std::shared_ptr<TaskHandle>> TimeSharingTaskExecutor::create_task( + const TaskId& task_id, std::function<double()> utilization_supplier, + int initial_split_concurrency, std::chrono::nanoseconds split_concurrency_adjust_frequency, + std::optional<int> max_concurrency_per_task) { + auto task_handle = std::make_shared<TimeSharingTaskHandle>( + task_id, _waiting_splits, utilization_supplier, initial_split_concurrency, + split_concurrency_adjust_frequency, max_concurrency_per_task); + RETURN_IF_ERROR_RESULT(task_handle->init()); + + std::lock_guard<std::mutex> lock(_mutex); + + _tasks[task_id] = task_handle; + + return task_handle; +} + +Status TimeSharingTaskExecutor::add_task(const TaskId& task_id, + std::shared_ptr<TaskHandle> task_handle) { + std::lock_guard<std::mutex> lock(_mutex); + _tasks[task_id] = + std::dynamic_pointer_cast<doris::vectorized::TimeSharingTaskHandle>(task_handle); + return Status::OK(); +} + +Status TimeSharingTaskExecutor::remove_task(std::shared_ptr<TaskHandle> task_handle) { + auto handle = std::dynamic_pointer_cast<TimeSharingTaskHandle>(task_handle); + std::vector<std::shared_ptr<PrioritizedSplitRunner>> splits_to_destroy; + + { + std::lock_guard<std::mutex> guard(_mutex); + auto it = _tasks.find(handle->task_id()); + if (it == _tasks.end() || handle->is_closed()) { + return Status::OK(); + } + _tasks.erase(it); + + // Task is already closed + if (task_handle->is_closed()) { + return Status::OK(); + } + + splits_to_destroy = handle->close(); + + for (const auto& split : splits_to_destroy) { + _all_splits.erase(split); + _intermediate_splits.erase(split); + _blocked_splits.erase(split); + } + _waiting_splits->remove_all(splits_to_destroy); + _record_leaf_splits_size(guard); + } + + // call destroy outside of synchronized block as it is expensive and doesn't need a lock on the task executor + for (auto& split : splits_to_destroy) { + split->close(Status::OK()); + } + + // record completed stats + int64_t thread_usage_nanos = handle->scheduled_nanos(); + int level = MultilevelSplitQueue::compute_level(thread_usage_nanos); + _completed_tasks_per_level[level]++; + + if (splits_to_destroy.empty()) { + return Status::OK(); + } + + // replace blocked splits that were terminated + { + std::lock_guard<std::mutex> guard(_mutex); + _add_new_entrants(guard); + _record_leaf_splits_size(guard); + } + return Status::OK(); +} + +Result<std::vector<SharedListenableFuture<Void>>> TimeSharingTaskExecutor::enqueue_splits( + std::shared_ptr<TaskHandle> task_handle, bool intermediate, + const std::vector<std::shared_ptr<SplitRunner>>& splits) { + std::vector<std::shared_ptr<PrioritizedSplitRunner>> splits_to_destroy; + Defer defer {[&]() { + for (auto& split : splits_to_destroy) { + split->close(Status::OK()); + } + }}; + std::vector<SharedListenableFuture<Void>> finished_futures; + auto handle = std::dynamic_pointer_cast<TimeSharingTaskHandle>(task_handle); + { + std::lock_guard<std::mutex> guard(_mutex); + for (const auto& task_split : splits) { + TaskId task_id = handle->task_id(); + int split_id = handle->next_split_id(); + + auto prioritized_split = + PrioritizedSplitRunner::create_shared(handle, split_id, task_split, _ticker); + RETURN_IF_ERROR_RESULT(prioritized_split->init()); + if (intermediate) { + if (handle->record_intermediate_split(prioritized_split)) { + _start_intermediate_split(prioritized_split, guard); + } else { + splits_to_destroy.push_back(prioritized_split); + } + } else { + if (handle->enqueue_split(prioritized_split)) { + _schedule_task_if_necessary(handle, guard); + _add_new_entrants(guard); + } else { + splits_to_destroy.push_back(prioritized_split); + } + } + finished_futures.push_back(prioritized_split->finished_future()); + } + _record_leaf_splits_size(guard); + } + return finished_futures; +} + +void TimeSharingTaskExecutor::_split_finished(std::shared_ptr<PrioritizedSplitRunner> split, + const Status& status) { + _completed_splits_per_level[split->priority().level()]++; + { + std::lock_guard<std::mutex> guard(_mutex); + _all_splits.erase(split); + + auto task_handle = split->task_handle(); + task_handle->split_finished(split); + + _schedule_task_if_necessary(task_handle, guard); + + _add_new_entrants(guard); + _record_leaf_splits_size(guard); + } + // call close outside of synchronized block as it is expensive and doesn't need a lock on the task executor + split->close(status); +} + +void TimeSharingTaskExecutor::_schedule_task_if_necessary( + std::shared_ptr<TimeSharingTaskHandle> task_handle, std::lock_guard<std::mutex>& guard) { + int guaranteed_concurrency = std::min( + _guaranteed_concurrency_per_task, + task_handle->max_concurrency_per_task().value_or(std::numeric_limits<int>::max())); + int splits_to_schedule = guaranteed_concurrency - task_handle->running_leaf_splits(); + + for (int i = 0; i < splits_to_schedule; ++i) { + auto split = task_handle->poll_next_split(); + if (!split) return; + + _start_split(split, guard); + auto elapsed_nanos = std::chrono::nanoseconds( + std::chrono::steady_clock::now().time_since_epoch().count() - + split->created_nanos()); + _split_queued_time + << std::chrono::duration_cast<std::chrono::microseconds>(elapsed_nanos).count(); + } + _record_leaf_splits_size(guard); +} + +void TimeSharingTaskExecutor::_add_new_entrants(std::lock_guard<std::mutex>& guard) { + int running = _all_splits.size() - _intermediate_splits.size(); + for (int i = 0; i < _min_concurrency - running; i++) { + auto split = _poll_next_split_worker(guard); + if (!split) { + break; + } + + auto elapsed_nanos = std::chrono::nanoseconds( + std::chrono::steady_clock::now().time_since_epoch().count() - + split->created_nanos()); + _split_queued_time + << std::chrono::duration_cast<std::chrono::microseconds>(elapsed_nanos).count(); + _start_split(split, guard); + } +} + +void TimeSharingTaskExecutor::_start_intermediate_split( + std::shared_ptr<PrioritizedSplitRunner> split, std::lock_guard<std::mutex>& guard) { + _start_split(split, guard); + _intermediate_splits.insert(split); +} + +void TimeSharingTaskExecutor::_start_split(std::shared_ptr<PrioritizedSplitRunner> split, + std::lock_guard<std::mutex>& guard) { + _all_splits.insert(split); + _waiting_splits->offer(split); +} + +std::shared_ptr<PrioritizedSplitRunner> TimeSharingTaskExecutor::_poll_next_split_worker( + std::lock_guard<std::mutex>& guard) { + // 遍历任务列表，找到第一个能产生分片的任务 + // 然后将该任务移到列表末尾，实现轮询调度 + for (auto it = _tasks.begin(); it != _tasks.end();) { + auto task = it->second; + // 跳过已经运行配置的最大驱动数的任务 + if (task->running_leaf_splits() >= + task->max_concurrency_per_task().value_or(_max_concurrency_per_task)) { + ++it; + continue; + } + + auto split = task->poll_next_split(); + if (split) { + // 将任务移到列表末尾 + auto task_copy = task; + auto task_id = it->first; + it = _tasks.erase(it); + _tasks[task_id] = task_copy; + return split; + } + ++it; + } + return nullptr; +} + +void TimeSharingTaskExecutor::_record_leaf_splits_size(std::lock_guard<std::mutex>& guard) { + // auto now = _ticker->read(); + // auto time_difference = now - _last_leaf_splits_size_record_time; + // + // if (time_difference > 0) { + // // 记录当前叶子分片的数量和持续时间 + // // _leaf_splits_size.add(_last_leaf_splits_size, std::chrono::nanoseconds(time_difference)); + // _last_leaf_splits_size_record_time = now; + // } + // // always record new lastLeafSplitsSize as it might have changed + // // even if timeDifference is 0 + // _last_leaf_splits_size = _all_splits.size() - _intermediate_splits.size(); +} + +void TimeSharingTaskExecutor::_interrupt() { + std::lock_guard<std::mutex> guard(_mutex); + _condition.notify_all(); + _waiting_splits->interrupt(); +} + +int64_t TimeSharingTaskExecutor::_get_running_tasks_for_level(int level) const { + std::lock_guard<std::mutex> guard(_mutex); + int64_t count = 0; + for (const auto& [task_id, task] : _tasks) { + if (task->priority().level() == level) { + count++; + } + } + return count; +} + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.h b/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.h new file mode 100644 index 0000000..c3fe520 --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.h
@@ -0,0 +1,208 @@ +// 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. + +#pragma once + +#include <atomic> +#include <chrono> +#include <condition_variable> +#include <memory> +#include <mutex> +#include <optional> +#include <queue> +#include <thread> +#include <unordered_map> +#include <unordered_set> +#include <vector> + +#include "bvar/latency_recorder.h" +#include "util/threadpool.h" +#include "vec/exec/executor/listenable_future.h" +#include "vec/exec/executor/stats.h" +#include "vec/exec/executor/task_executor.h" +#include "vec/exec/executor/ticker.h" +#include "vec/exec/executor/time_sharing/multilevel_split_queue.h" +#include "vec/exec/executor/time_sharing/prioritized_split_runner.h" + +namespace doris { +namespace vectorized { + +/** + * ThreadSafe + */ +class TimeSharingTaskExecutor : public TaskExecutor { + ENABLE_FACTORY_CREATOR(TimeSharingTaskExecutor); + +public: + TimeSharingTaskExecutor(int threads, int min_concurrency, int guaranteed_concurrency_per_task, + int max_concurrency_per_task, std::shared_ptr<Ticker> ticker, + std::chrono::milliseconds stuck_split_warning_threshold = + std::chrono::milliseconds(60000), + std::shared_ptr<MultilevelSplitQueue> split_queue = nullptr); + + ~TimeSharingTaskExecutor() override; + + Status init() override; + + Status start() override; + void stop() override; + + Result<std::shared_ptr<TaskHandle>> create_task( + const TaskId& task_id, std::function<double()> utilization_supplier, + int initial_split_concurrency, + std::chrono::nanoseconds split_concurrency_adjust_frequency, + std::optional<int> max_concurrency_per_task) override; + + Status add_task(const TaskId& task_id, std::shared_ptr<TaskHandle> task_handle) override; + + Status remove_task(std::shared_ptr<TaskHandle> task_handle) override; + + Result<std::vector<SharedListenableFuture<Void>>> enqueue_splits( + std::shared_ptr<TaskHandle> task_handle, bool intermediate, + const std::vector<std::shared_ptr<SplitRunner>>& splits) override; + + size_t waiting_splits_size() const { return _waiting_splits->size(); } + + size_t intermediate_splits_size() const { + std::lock_guard<std::mutex> guard(_mutex); + return _intermediate_splits.size(); + } + + size_t running_splits_size() const { return _running_splits.size(); } + + size_t blocked_splits_size() const { return _blocked_splits.size(); } + + size_t total_splits_size() const { + std::lock_guard<std::mutex> guard(_mutex); + return _all_splits.size(); + } + + size_t tasks_size() const { + std::lock_guard<std::mutex> guard(_mutex); + return _tasks.size(); + } + + int64_t completed_tasks_level0() const { return _completed_tasks_per_level[0]; } + + int64_t completed_tasks_level1() const { return _completed_tasks_per_level[1]; } + + int64_t completed_tasks_level2() const { return _completed_tasks_per_level[2]; } + + int64_t completed_tasks_level3() const { return _completed_tasks_per_level[3]; } + + int64_t completed_tasks_level4() const { return _completed_tasks_per_level[4]; } + + int64_t completed_splits_level0() const { return _completed_splits_per_level[0]; } + + int64_t completed_splits_level1() const { return _completed_splits_per_level[1]; } + + int64_t completed_splits_level2() const { return _completed_splits_per_level[2]; } + + int64_t completed_splits_level3() const { return _completed_splits_per_level[3]; } + + int64_t completed_splits_level4() const { return _completed_splits_per_level[4]; } + + int64_t running_tasks_level0() const { return _get_running_tasks_for_level(0); } + + int64_t running_tasks_level1() const { return _get_running_tasks_for_level(1); } + + int64_t running_tasks_level2() const { return _get_running_tasks_for_level(2); } + + int64_t running_tasks_level3() const { return _get_running_tasks_for_level(3); } + + int64_t running_tasks_level4() const { return _get_running_tasks_for_level(4); } + +private: + class ScopedRunnerTracker { + public: + ScopedRunnerTracker( + std::mutex& mutex, + std::unordered_set<std::shared_ptr<PrioritizedSplitRunner>>& running_splits, + std::shared_ptr<PrioritizedSplitRunner> split) + : _mutex(mutex), _running_splits(running_splits), _split(std::move(split)) { + std::lock_guard<std::mutex> guard(_mutex); + _running_splits.insert(_split); + } + + ~ScopedRunnerTracker() { + std::lock_guard<std::mutex> guard(_mutex); + _running_splits.erase(_split); + } + + private: + std::mutex& _mutex; + std::unordered_set<std::shared_ptr<PrioritizedSplitRunner>>& _running_splits; + std::shared_ptr<PrioritizedSplitRunner> _split; + }; + + class TaskRunner { + public: + TaskRunner(TimeSharingTaskExecutor& executor); + void run(); + void stop(); + + private: + TimeSharingTaskExecutor& _executor; + std::atomic<bool> _running {true}; + }; + + Status _add_runner_thread(); + void _schedule_task_if_necessary(std::shared_ptr<TimeSharingTaskHandle> task_handle, + std::lock_guard<std::mutex>& guard); + void _add_new_entrants(std::lock_guard<std::mutex>& guard); + void _start_intermediate_split(std::shared_ptr<PrioritizedSplitRunner> split, + std::lock_guard<std::mutex>& guard); + void _start_split(std::shared_ptr<PrioritizedSplitRunner> split, + std::lock_guard<std::mutex>& guard); + std::shared_ptr<PrioritizedSplitRunner> _poll_next_split_worker( + std::lock_guard<std::mutex>& guard); + void _record_leaf_splits_size(std::lock_guard<std::mutex>& guard); + void _split_finished(std::shared_ptr<PrioritizedSplitRunner> split, const Status& status); + void _interrupt(); + + int64_t _get_running_tasks_for_level(int level) const; + + std::unique_ptr<ThreadPool> _thread_pool; + const int _runner_threads; + const int _min_concurrency; + const int _guaranteed_concurrency_per_task; + const int _max_concurrency_per_task; + std::shared_ptr<Ticker> _ticker; + const std::chrono::milliseconds _stuck_split_warning_threshold; + std::shared_ptr<MultilevelSplitQueue> _waiting_splits; + + mutable std::mutex _mutex; + std::condition_variable _condition; + std::atomic<bool> _stopped {false}; + + std::vector<std::unique_ptr<TaskRunner>> _task_runners; + + std::unordered_map<TaskId, std::shared_ptr<TimeSharingTaskHandle>> _tasks; + + std::unordered_set<std::shared_ptr<PrioritizedSplitRunner>> _all_splits; + std::unordered_set<std::shared_ptr<PrioritizedSplitRunner>> _intermediate_splits; + std::unordered_set<std::shared_ptr<PrioritizedSplitRunner>> _running_splits; + std::unordered_map<std::shared_ptr<PrioritizedSplitRunner>, SharedListenableFuture<Void>> + _blocked_splits; + std::array<std::atomic<int64_t>, 5> _completed_tasks_per_level = {0, 0, 0, 0, 0}; + std::array<std::atomic<int64_t>, 5> _completed_splits_per_level = {0, 0, 0, 0, 0}; + + bvar::LatencyRecorder _split_queued_time; +}; + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/time_sharing/time_sharing_task_handle.cpp b/be/src/vec/exec/executor/time_sharing/time_sharing_task_handle.cpp new file mode 100644 index 0000000..0979b72 --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/time_sharing_task_handle.cpp
@@ -0,0 +1,217 @@ +// 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. + +#include "vec/exec/executor/time_sharing/time_sharing_task_handle.h" + +#include "vec/exec/executor/time_sharing/multilevel_split_queue.h" + +namespace doris { +namespace vectorized { + +TimeSharingTaskHandle::TimeSharingTaskHandle( + const TaskId& task_id, std::shared_ptr<MultilevelSplitQueue> split_queue, + std::function<double()> utilization_supplier, int initial_split_concurrency, + std::chrono::nanoseconds split_concurrency_adjust_frequency, + std::optional<int> max_concurrency_per_task) + : _task_id(task_id), + _split_queue(std::move(split_queue)), + _utilization_supplier(std::move(utilization_supplier)), + _max_concurrency_per_task(max_concurrency_per_task), + _concurrency_controller(initial_split_concurrency, split_concurrency_adjust_frequency) {} + +Status TimeSharingTaskHandle::init() { + return Status::OK(); +} + +//Priority TimeSharingTaskHandle::add_scheduled_nanos(int64_t duration_nanos) { +// std::lock_guard<std::mutex> lock(_mutex); +// _concurrency_controller.update(duration_nanos, _utilization_supplier(), +// _running_leaf_splits.size()); +// _scheduled_nanos += duration_nanos; +// +// Priority new_priority = +// _split_queue->update_priority(_priority.load(), duration_nanos, _scheduled_nanos); +// +// _priority.store(new_priority); +// return new_priority; +//} + +Priority TimeSharingTaskHandle::add_scheduled_nanos(int64_t duration_nanos) { + std::lock_guard<std::mutex> lock(_mutex); + _concurrency_controller.update(duration_nanos, _utilization_supplier(), + _running_leaf_splits.size()); + _scheduled_nanos += duration_nanos; + + Priority new_priority = + _split_queue->update_priority(_priority, duration_nanos, _scheduled_nanos); + + _priority = new_priority; + return new_priority; +} + +//Priority TimeSharingTaskHandle::reset_level_priority() { +// std::lock_guard<std::mutex> lock(_mutex); +// +// Priority current_priority = _priority.load(); +// int64_t level_min_priority = +// _split_queue->get_level_min_priority(current_priority.level(), _scheduled_nanos); +// +// if (current_priority.level_priority() < level_min_priority) { +// Priority new_priority(current_priority.level(), level_min_priority); +// _priority.store(new_priority); +// return new_priority; +// } +// +// return current_priority; +//} + +Priority TimeSharingTaskHandle::reset_level_priority() { + std::lock_guard<std::mutex> lock(_mutex); + + Priority current_priority = _priority; + int64_t level_min_priority = + _split_queue->get_level_min_priority(current_priority.level(), _scheduled_nanos); + + if (current_priority.level_priority() < level_min_priority) { + Priority new_priority(current_priority.level(), level_min_priority); + _priority = new_priority; + return new_priority; + } + + return current_priority; +} + +bool TimeSharingTaskHandle::is_closed() const { + return _closed.load(); +} + +//Priority TimeSharingTaskHandle::priority() const { +// return _priority.load(); +//} + +Priority TimeSharingTaskHandle::priority() const { + std::lock_guard<std::mutex> lock(_mutex); + return _priority; +} + +TaskId TimeSharingTaskHandle::task_id() const { + return _task_id; +} + +std::optional<int> TimeSharingTaskHandle::max_concurrency_per_task() const { + return _max_concurrency_per_task; +} + +std::vector<std::shared_ptr<PrioritizedSplitRunner>> TimeSharingTaskHandle::close() { + std::lock_guard<std::mutex> lock(_mutex); + + if (_closed.exchange(true)) { + return {}; + } + + std::vector<std::shared_ptr<PrioritizedSplitRunner>> result; + result.reserve(_running_intermediate_splits.size() + _running_leaf_splits.size() + + _queued_leaf_splits.size()); + + result.insert(result.end(), _running_intermediate_splits.begin(), + _running_intermediate_splits.end()); + result.insert(result.end(), _running_leaf_splits.begin(), _running_leaf_splits.end()); + + while (!_queued_leaf_splits.empty()) { + result.push_back(_queued_leaf_splits.front()); + _queued_leaf_splits.pop(); + } + + _running_intermediate_splits.clear(); + _running_leaf_splits.clear(); + + return result; +} + +bool TimeSharingTaskHandle::enqueue_split(std::shared_ptr<PrioritizedSplitRunner> split) { + std::lock_guard<std::mutex> lock(_mutex); + if (_closed) { + return false; + } + _queued_leaf_splits.emplace(std::move(split)); + return true; +} + +bool TimeSharingTaskHandle::record_intermediate_split( + std::shared_ptr<PrioritizedSplitRunner> split) { + std::lock_guard<std::mutex> lock(_mutex); + if (_closed) { + return false; + } + _running_intermediate_splits.emplace_back(std::move(split)); + return true; +} + +int TimeSharingTaskHandle::running_leaf_splits() const { + std::lock_guard<std::mutex> lock(_mutex); + return _running_leaf_splits.size(); +} + +int64_t TimeSharingTaskHandle::scheduled_nanos() const { + std::lock_guard<std::mutex> lock(_mutex); + return _scheduled_nanos; +} + +std::shared_ptr<PrioritizedSplitRunner> TimeSharingTaskHandle::poll_next_split() { + std::lock_guard<std::mutex> lock(_mutex); + + if (_closed) { + return nullptr; + } + + if (_running_leaf_splits.size() >= _concurrency_controller.target_concurrency()) { + return nullptr; + } + + if (_queued_leaf_splits.empty()) { + return nullptr; + } + + auto split = _queued_leaf_splits.front(); + _queued_leaf_splits.pop(); + _running_leaf_splits.push_back(split); + return split; +} + +void TimeSharingTaskHandle::split_finished(std::shared_ptr<PrioritizedSplitRunner> split) { + std::lock_guard<std::mutex> lock(_mutex); + _concurrency_controller.split_finished(split->scheduled_nanos(), _utilization_supplier(), + _running_leaf_splits.size()); + + auto it = std::find(_running_intermediate_splits.begin(), _running_intermediate_splits.end(), + split); + if (it != _running_intermediate_splits.end()) { + _running_intermediate_splits.erase(it); + } + + it = std::find(_running_leaf_splits.begin(), _running_leaf_splits.end(), split); + if (it != _running_leaf_splits.end()) { + _running_leaf_splits.erase(it); + } +} + +int TimeSharingTaskHandle::next_split_id() { + return _next_split_id.fetch_add(1); +} + +} // namespace vectorized +} // namespace doris
diff --git a/be/src/vec/exec/executor/time_sharing/time_sharing_task_handle.h b/be/src/vec/exec/executor/time_sharing/time_sharing_task_handle.h new file mode 100644 index 0000000..3373a68 --- /dev/null +++ b/be/src/vec/exec/executor/time_sharing/time_sharing_task_handle.h
@@ -0,0 +1,84 @@ +// 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. + +#pragma once + +#include <atomic> +#include <memory> +#include <mutex> +#include <optional> +#include <queue> +#include <vector> + +#include "vec/exec/executor/task_handle.h" +#include "vec/exec/executor/task_id.h" +#include "vec/exec/executor/time_sharing/prioritized_split_runner.h" +#include "vec/exec/executor/time_sharing/priority.h" +#include "vec/exec/executor/time_sharing/split_concurrency_controller.h" + +namespace doris { +namespace vectorized { + +class MultilevelSplitQueue; + +class TimeSharingTaskHandle : public TaskHandle { + ENABLE_FACTORY_CREATOR(TimeSharingTaskHandle); + +public: + TimeSharingTaskHandle(const TaskId& task_id, std::shared_ptr<MultilevelSplitQueue> split_queue, + std::function<double()> utilization_supplier, + int initial_split_concurrency, + std::chrono::nanoseconds split_concurrency_adjust_frequency, + std::optional<int> max_concurrency_per_task); + + Status init() override; + + Priority add_scheduled_nanos(int64_t duration_nanos); + Priority reset_level_priority(); + bool is_closed() const override; + Priority priority() const; + TaskId task_id() const override; + std::optional<int> max_concurrency_per_task() const; + std::vector<std::shared_ptr<PrioritizedSplitRunner>> close(); + bool enqueue_split(std::shared_ptr<PrioritizedSplitRunner> split); + bool record_intermediate_split(std::shared_ptr<PrioritizedSplitRunner> split); + int running_leaf_splits() const; + int64_t scheduled_nanos() const; + std::shared_ptr<PrioritizedSplitRunner> poll_next_split(); + void split_finished(std::shared_ptr<PrioritizedSplitRunner> split); + int next_split_id(); + +private: + mutable std::mutex _mutex; + std::atomic<bool> _closed {false}; + const TaskId _task_id; + std::shared_ptr<MultilevelSplitQueue> _split_queue; + std::function<double()> _utilization_supplier; + std::optional<int> _max_concurrency_per_task; + SplitConcurrencyController _concurrency_controller; + + std::queue<std::shared_ptr<PrioritizedSplitRunner>> _queued_leaf_splits; + std::vector<std::shared_ptr<PrioritizedSplitRunner>> _running_leaf_splits; + std::vector<std::shared_ptr<PrioritizedSplitRunner>> _running_intermediate_splits; + int64_t _scheduled_nanos {0}; + std::atomic<int> _next_split_id {0}; + // std::atomic<Priority> _priority {Priority(0, 0)}; + Priority _priority {0, 0}; +}; + +} // namespace vectorized +} // namespace doris
diff --git a/be/test/vec/exec/executor/time_sharing/multilevel_split_queue_test.cpp b/be/test/vec/exec/executor/time_sharing/multilevel_split_queue_test.cpp new file mode 100644 index 0000000..cef23ff --- /dev/null +++ b/be/test/vec/exec/executor/time_sharing/multilevel_split_queue_test.cpp
@@ -0,0 +1,260 @@ +//// 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. +// +//#include <gtest/gtest.h> +//#include <atomic> +//#include <thread> +//#include "vec/exec/scan/MultilevelSplitQueue.h" +// +//namespace doris::vectorized { +// +//// 测试用SplitRunner实现 +//class TestSplitRunner : public PrioritizedSplitRunner { +//public: +// TestSplitRunner(Priority priority) +// : PrioritizedSplitRunner(_global_cpu_time_micros, _global_scheduled_time_micros, _blocked_quanta_wall_time, _unblocked_quanta_wall_time), _priority(priority) {} +// +// Priority getPriority() const override { return _priority; } +// void updatePriority(Priority new_pri) { _priority = new_pri; } +// +//private: +// Priority _priority; +// CounterStats _global_cpu_time_micros; +// CounterStats _global_scheduled_time_micros; +// +// TimeStats _blocked_quanta_wall_time; +// TimeStats _unblocked_quanta_wall_time; +//}; +// +//class MultilevelSplitQueueTest : public testing::Test { +//protected: +// void SetUp() override { +// queue = std::make_unique<MultilevelSplitQueue>(2.0); +// } +// +// std::shared_ptr<TestSplitRunner> make_split(int level, int64_t pri) { +// return std::make_shared<TestSplitRunner>(Priority(level, pri)); +// } +// +// std::unique_ptr<MultilevelSplitQueue> queue; +//}; +// +//TEST_F(MultilevelSplitQueueTest, BasicFunction) { +// // 验证空队列 +// EXPECT_EQ(queue->size(), 0); +// +// // 添加任务 +// auto split = make_split(0, 100); +// queue->addSplit(split); +// EXPECT_EQ(queue->size(), 1); +// +// // 取出任务 +// auto task = queue->pollSplit(); +// ASSERT_NE(task, nullptr); +// EXPECT_EQ(task->getPriority().getLevel(), 0); +// EXPECT_EQ(queue->size(), 0); +//} +// +//TEST_F(MultilevelSplitQueueTest, LevelContributionCap) { +// // 创建两个测试任务 +// auto split0 = make_split(0, 0); +// auto split1 = make_split(0, 0); +// +// // 获取级别阈值数组（假设实现中有这个静态成员） +// const auto& levelThresholds = MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS; +// +// int64_t scheduledNanos = 0; +// for (size_t i = 0; i < levelThresholds.size() - 1; ++i) { +// // 计算需要增加的时间量（秒转纳秒） +// const int64_t levelAdvance = +// (levelThresholds[i + 1] - levelThresholds[i]) * 1'000'000'000LL; +// // 模拟任务累积时间 +// scheduledNanos += levelAdvance; +// split0->updatePriority(queue->updatePriority( +// split0->getPriority(), levelAdvance, scheduledNanos)); +// split1->updatePriority(queue->updatePriority( +// split1->getPriority(), levelAdvance, scheduledNanos)); +// +// // 验证级别提升 +// EXPECT_EQ(split0->getPriority().getLevel(), i + 1); +// EXPECT_EQ(split1->getPriority().getLevel(), i + 1); +// +// // 验证级别累计时间（应用贡献上限） +// const int64_t expected = +// 2 * std::min(levelAdvance, MultilevelSplitQueue::LEVEL_CONTRIBUTION_CAP); +// EXPECT_EQ(queue->get_level_scheduled_time(i), expected); +// EXPECT_EQ(queue->get_level_scheduled_time(i + 1), 0); +// } +//} +// +//TEST_F(MultilevelSplitQueueTest, LevelContributionWithCap) { +// // 创建队列（levelTimeMultiplier=2） +// MultilevelSplitQueue queue(2.0); +// +// // 创建测试任务并初始化优先级 +// auto split = make_split(0, 0); +// queue.addSplit(split); +// +// // 模拟运行10分钟（转换为纳秒） +// constexpr int64_t quantaNanos = 10LL * 60 * 1'000'000'000; // 10分钟 +// int64_t scheduledNanos = 0; +// +// // 更新优先级触发时间分配 +// auto task = std::static_pointer_cast<TestSplitRunner>(queue.pollSplit()); +// scheduledNanos += quantaNanos; +// Priority new_pri = queue.updatePriority( +// task->getPriority(), +// quantaNanos, +// scheduledNanos +// ); +// task->updatePriority(new_pri); +// +// // 计算受限制的时间贡献 +// int64_t cappedNanos = std::min(quantaNanos, MultilevelSplitQueue::LEVEL_CONTRIBUTION_CAP); +// const auto& levels = MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS; +// +// // 验证每个级别的时间分配 +// for (size_t i = 0; i < levels.size() - 1; ++i) { +// // 计算当前级别时间阈值（转换为纳秒） +// int64_t levelThreshold = static_cast<int64_t>((levels[i + 1] - levels[i])) * 1'000'000'000L; +// int64_t expected = std::min(levelThreshold, cappedNanos); +// +// // 验证当前级别累计时间 +// EXPECT_EQ(queue.get_level_scheduled_time(i), expected); +// cappedNanos -= expected; +// } +// +// // 验证剩余时间分配到最后一个级别 +//// EXPECT_EQ(queue.get_level_scheduled_time(levels.size() - 1), cappedNanos); +//} +// +//TEST_F(MultilevelSplitQueueTest, TimeCompensation) { +// // 初始状态验证 +// EXPECT_EQ(queue->get_level_scheduled_time(2), 0); +// +// int64_t scheduledNanos = 0L; +// +// // 1. 创建并添加任务 +// auto split = make_split(2, 200); +// queue->addSplit(split); +// +// // 2. 取出任务（触发时间补偿） +// auto task = std::static_pointer_cast<TestSplitRunner>(queue->pollSplit()); +// ASSERT_NE(task, nullptr); +// +// // 3. 模拟任务运行5秒并更新优先级 +// const int64_t RUN_TIME = 5'000'000'000; // 5秒 +// scheduledNanos += RUN_TIME; +// Priority new_pri = queue->updatePriority( +// task->getPriority(), +// RUN_TIME, +// scheduledNanos +// ); +// task->updatePriority(new_pri); +// +// // 4. 重新加入队列（此时level2已有时间记录） +// queue->addSplit(task); +// +// // 5. 再次添加新任务触发补偿 +// auto new_split = make_split(2, 300); +// queue->addSplit(new_split); +// +// // 验证补偿逻辑 +// int64_t level2_time = queue->get_level_scheduled_time(2); +// int64_t level0_time = queue->get_level0_target_time(); +// +// // 预期补偿 = (level0_time / 4) + 5秒 +// int64_t expected = (level0_time / 4) + RUN_TIME; +// +// // 允许10%误差 +// EXPECT_GT(level2_time, expected - expected/10); +// EXPECT_LT(level2_time, expected + expected/10); +//} +// +//TEST_F(MultilevelSplitQueueTest, PriorityUpdate) { +// auto split = make_split(0, 0); +// queue->addSplit(split); +// +// // 模拟运行30秒 +// auto split_ptr = std::static_pointer_cast<TestSplitRunner>(queue->pollSplit()); +// ASSERT_NE(split_ptr, nullptr); +// Priority new_pri = queue->updatePriority( +// split_ptr->getPriority(), +// 30'000'000'000LL, // 30秒 +// 0 +// ); +// +// // 验证升级到level3 +// EXPECT_EQ(new_pri.getLevel(), 3); +// EXPECT_GT(new_pri.getLevelPriority(), 30'000'000'000LL); +//} +// +//TEST_F(MultilevelSplitQueueTest, LevelSelection) { +// // 准备不同级别的任务 +// queue->addSplit(make_split(0, 100)); // level0 +// queue->addSplit(make_split(3, 50)); // level3 +// +// // 多次轮询触发时间补偿 +// for (int i = 0; i < 3; ++i) { +// auto task = queue->pollSplit(); +// queue->addSplit(task); +// } +// +// // 应优先选择level3 +// auto task = queue->pollSplit(); +// EXPECT_EQ(task->getPriority().getLevel(), 3); +//} +// +//TEST_F(MultilevelSplitQueueTest, Concurrency) { +// constexpr int THREAD_NUM = 4; +// constexpr int OPS = 1000; +// std::vector<std::thread> workers; +// +// for (int i = 0; i < THREAD_NUM; ++i) { +// workers.emplace_back([&] { +// for (int j = 0; j < OPS; ++j) { +// auto split = make_split(j % 5, j); +// queue->addSplit(split); +// if (auto task = queue->pollSplit()) { +// // 模拟处理任务 +// std::this_thread::sleep_for(std::chrono::microseconds(10)); +// } +// } +// }); +// } +// +// for (auto& t : workers) t.join(); +// SUCCEED(); // 验证无崩溃 +//} +// +//TEST_F(MultilevelSplitQueueTest, Interrupt) { +// std::atomic<bool> stopped{false}; +// std::thread consumer([&] { +// while (!stopped) { +// if (auto task = queue->pollSplit()) { +// // process task +// } +// } +// }); +// +// std::this_thread::sleep_for(std::chrono::milliseconds(10)); +// queue->interrupt(); +// stopped = true; +// consumer.join(); +//} +// +//} // namespace doris::vectorized \ No newline at end of file
diff --git a/be/test/vec/exec/executor/time_sharing/split_concurrency_controller_test.cpp b/be/test/vec/exec/executor/time_sharing/split_concurrency_controller_test.cpp new file mode 100644 index 0000000..58868a9 --- /dev/null +++ b/be/test/vec/exec/executor/time_sharing/split_concurrency_controller_test.cpp
@@ -0,0 +1,71 @@ +// 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. + +#include "vec/exec/executor/time_sharing/split_concurrency_controller.h" + +#include <gtest/gtest.h> + +#include <chrono> + +namespace doris::vectorized { + +using namespace std::chrono_literals; + +class SplitConcurrencyControllerTest : public testing::Test {}; + +TEST_F(SplitConcurrencyControllerTest, test_ramp_up) { + SplitConcurrencyController controller(1, std::chrono::seconds(1)); + + for (int i = 0; i < 10; ++i) { + controller.update(2'000'000'000, 0.0, i + 1); + EXPECT_EQ(controller.target_concurrency(), i + 2) << "Rampup failed at iteration " << i; + } +} + +TEST_F(SplitConcurrencyControllerTest, test_ramp_down) { + SplitConcurrencyController controller(10, std::chrono::seconds(1)); + + for (int i = 0; i < 9; ++i) { + controller.update(2'000'000'000, 1.0, 10 - i); + controller.split_finished(30'000'000'000, 1.0, 10 - i); + EXPECT_EQ(controller.target_concurrency(), 10 - i - 1) + << "Rampdown failed at iteration " << i; + } +} + +TEST_F(SplitConcurrencyControllerTest, test_rapid_adjust_for_quick_splits) { + SplitConcurrencyController controller(10, std::chrono::seconds(1)); + + // 测试快速递减 + for (int i = 0; i < 9; ++i) { + controller.update(200'000'000, 1.0, 10 - i); + controller.split_finished(100'000'000, 1.0, 10 - i); + EXPECT_EQ(controller.target_concurrency(), 10 - i - 1) + << "Rapid decrease failed at iteration " << i; + } + + // 重置状态后测试递增 + controller.update(30'000'000'000, 0.0, 1); + for (int i = 0; i < 10; ++i) { + controller.update(200'000'000, 0.0, i + 1); + controller.split_finished(100'000'000, 0.0, i + 1); + EXPECT_EQ(controller.target_concurrency(), i + 2) + << "Rapid increase failed at iteration " << i; + } +} + +} // namespace doris::vectorized \ No newline at end of file
diff --git a/be/test/vec/exec/executor/time_sharing/time_sharing_task_executor_test.cpp b/be/test/vec/exec/executor/time_sharing/time_sharing_task_executor_test.cpp new file mode 100644 index 0000000..104c626 --- /dev/null +++ b/be/test/vec/exec/executor/time_sharing/time_sharing_task_executor_test.cpp
@@ -0,0 +1,1238 @@ +// 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. + +#include "vec/exec/executor/time_sharing/time_sharing_task_executor.h" + +#include <gtest/gtest.h> + +#include <atomic> +#include <chrono> +#include <condition_variable> +#include <functional> +#include <future> +#include <mutex> +#include <random> +#include <thread> + +#include "vec/exec/executor/ticker.h" +#include "vec/exec/executor/time_sharing/time_sharing_task_handle.h" + +namespace doris { +namespace vectorized { + +class PhaseController { +public: + explicit PhaseController(int parties) : _parties(parties), _count(parties), _generation(0) {} + + int arrive_and_await() { + std::unique_lock<std::mutex> lock(_mutex); + int current_generation = _generation; + const int arrived_count = --_count; + + if (arrived_count == 0) { + // 重置计数器并进入下一代 + _count = _parties; + _generation++; + _cv.notify_all(); + } else { + // 等待直到进入下一代 + _cv.wait(lock, [this, current_generation] { + return _generation > current_generation || _count == 0; + }); + } + return _generation; + } + + void arrive_and_deregister() { + std::unique_lock<std::mutex> lock(_mutex); + _parties--; // 永久减少参与方数量 + const int arrived_count = --_count; + + // fprintf(stderr, "[PhaseController] 注销参与者，剩余参与方=%d\n", _parties); + + if (arrived_count == 0) { + // 立即触发代际更新 + _count = _parties; + _generation++; + _cv.notify_all(); + } else { + // 仅记录当前代际，不阻塞等待 + _cv.notify_all(); // 唤醒可能存在的等待线程 + } + } + + void register_party() { + std::lock_guard<std::mutex> lock(_mutex); + _parties++; + _count++; // 立即生效新参与者 + _cv.notify_all(); // 唤醒可能存在的等待线程 + fprintf(stderr, "[PhaseController] 注册新参与者，当前总参与方=%d\n", _parties); + } + + int parties() const { + std::lock_guard<std::mutex> lock(_mutex); + return _parties; + } + + int count() const { + std::lock_guard<std::mutex> lock(_mutex); + return _count; + } + + int generation() const { + std::lock_guard<std::mutex> lock(_mutex); + return _generation; + } + +private: + mutable std::mutex _mutex; + std::condition_variable _cv; + int _parties; + int _count; + int _generation; // 新增代际计数器 +}; + +// 线程安全的测试用Ticker +class TestingTicker final : public Ticker { +public: + TestingTicker() : time_(0) {} + + int64_t read() const override { return time_.load(std::memory_order_relaxed); } + + auto now() const { return time_.load(); } + + void increment(int64_t delta, std::chrono::nanoseconds unit) { + if (delta < 0) { + throw std::invalid_argument("delta is negative"); + } + + std::lock_guard<std::mutex> lock(mutex_); + time_ += unit.count() * delta; + } + +private: + std::atomic<int64_t> time_; + mutable std::mutex mutex_; +}; + +class TestingSplitRunner : public SplitRunner { +public: + TestingSplitRunner(std::string name, std::shared_ptr<TestingTicker> ticker, + std::shared_ptr<PhaseController> global_controller, + std::shared_ptr<PhaseController> begin_controller, + std::shared_ptr<PhaseController> end_controller, int required_phases, + int quanta_time_ms) + : _name(std::move(name)), + _ticker(ticker), + _global_controller(global_controller), + _begin_controller(begin_controller), + _end_controller(end_controller), + _required_phases(required_phases), + _quanta_time(quanta_time_ms), + _completed_phases(0), + _first_phase(-1), + _last_phase(-1), + _started(false) { + // _completion_future(_completion_promise.get_future()) { + _begin_controller->register_party(); + _end_controller->register_party(); + + if (_global_controller->parties() == 0) { + _global_controller->register_party(); + } + } + + Status init() override { return Status::OK(); } + + Result<SharedListenableFuture<Void>> process_for(std::chrono::nanoseconds) override { + // fprintf(stderr, "[%zu] TestingSplitRunner::process_for()\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id())); + _started = true; + _ticker->increment(_quanta_time, std::chrono::milliseconds(1)); + // 创建新的promise/future对 + std::promise<void> phase_promise; + std::future<void> phase_future = phase_promise.get_future(); + + // Global phase synchronization + // fprintf(stderr, + // "[%zu] TestingSplitRunner::process_for() _global_controller.arrive_and_await()," + // " _global_controller.count(): %d, _global_controller.parties(): %d;\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id()), + // _global_controller->count(), _global_controller->parties()); + _global_controller->arrive_and_await(); + // fprintf(stderr, + // "[%zu] TestingSplitRunner::process_for() _global_controller.arrive_and_await() " + // "finished;\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id())); + + // Begin quanta phase + // static_cast<void>(_begin_controller); + int generation = _begin_controller->arrive_and_await(); + int expected = -1; + _first_phase.compare_exchange_weak(expected, generation - 1, std::memory_order_relaxed); + _last_phase.store(generation, std::memory_order_relaxed); + + // fprintf(stderr, + // "[%zu] TestingSplitRunner::process_for() _end_controller.arrive_and_await()," + // " _end_controller.count(): %d, _end_controller.parties(): %d;\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id()), _end_controller->count(), + // _end_controller->parties()); + + // End quanta phase + _end_controller->arrive_and_await(); + + // fprintf(stderr, + // "[%zu] TestingSplitRunner::process_for() _end_controller.arrive_and_await() " + // "finished;\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id())); + + if (++_completed_phases >= _required_phases) { + // fprintf(stderr, + // "[%zu] TestingSplitRunner::process_for() all completed: completed_phases=%d, " + // "required=%d, global_count=%d, global_parties=%d\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id()), + // _completed_phases.load(), _required_phases, _global_controller->count(), + // _global_controller->parties()); + _end_controller->arrive_and_deregister(); + _begin_controller->arrive_and_deregister(); + _global_controller->arrive_and_deregister(); + // 仅当任务完成时设置主promise + _completion_future.set_value(Void {}); + } + + // fprintf(stderr, + // "[%zu] TestingSplitRunner::process_for() _completed_phases: %d," + // " _required_phases: %d, _global_controller.count(): %d, " + // "_global_controller.parties(): %d;\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id()), _completed_phases.load(), + // _required_phases, _global_controller->count(), _global_controller->parties()); + + // 立即完成当前阶段的promise + // phase_promise.set_value(); + + return SharedListenableFuture<Void>::create_ready(); + } + + int completed_phases() const { return _completed_phases; } + + int first_phase() const { return _first_phase; } + + int last_phase() const { return _last_phase; } + + void close(const Status& status) override { + // Implementation needed + } + + std::string get_info() const override { + // Implementation needed + return ""; + } + + bool is_finished() override { + // return _completion_future.wait_for(std::chrono::seconds(0)) == std::future_status::ready; + return _completion_future.is_ready(); + } + + Status finished_status() override { return _completion_future.get_status(); } + + bool is_started() const { return _started.load(); } + +private: + std::string _name; + std::shared_ptr<TestingTicker> _ticker; + std::shared_ptr<PhaseController> _global_controller; + std::shared_ptr<PhaseController> _begin_controller; + std::shared_ptr<PhaseController> _end_controller; + int _required_phases; + int _quanta_time; + + std::atomic<int> _completed_phases; + std::atomic<int> _first_phase; + std::atomic<int> _last_phase; + std::atomic<bool> _started; + // std::promise<void> _completion_promise; // 用于整体完成通知 + // std::shared_future<void> _completion_future; // 改为共享future + ListenableFuture<Void> _completion_future {}; +}; + +class TimeSharingTaskExecutorTest : public testing::Test { +protected: + void SetUp() override {} + + void TearDown() override {} + +private: + // 修改为模板函数以支持不同容器类型 + template <typename Container> + void assert_split_states(int end_index, const Container& splits) { + // 打印所有split的启动状态 + fprintf(stderr, "[验证] 所有split启动状态总览（共%zu个）:\n", splits.size()); + for (int i = 0; i < splits.size(); ++i) { + fprintf(stderr, " → split[%d] 启动状态: %s\n", i, + splits[i]->is_started() ? "true" : "false"); + } + + // 验证end_index之前的split都已启动 + fprintf(stderr, "\n[验证] 详细检查0-%d号split应已启动:\n", end_index); + for (int i = 0; i <= end_index; ++i) { + bool actual = splits[i]->is_started(); + fprintf(stderr, " → split[%d] 预期启动=true，实际=%s\n", i, actual ? "true" : "false"); + EXPECT_TRUE(actual) << "Split " << i << " 应该已启动"; + } + + // 验证end_index之后的split未启动 + fprintf(stderr, "\n[验证] 详细检查%d-%zu号split应未启动:\n", end_index + 1, + splits.size() - 1); + for (int i = end_index + 1; i < splits.size(); ++i) { + bool actual = splits[i]->is_started(); + fprintf(stderr, " → split[%d] 预期启动=false，实际=%s\n", i, + actual ? "true" : "false"); + EXPECT_FALSE(actual) << "Split " << i << " 应该未启动"; + } + + fprintf(stderr, "[验证] 完整状态验证完成\n\n"); + } + + std::mt19937_64 _rng {std::random_device {}()}; +}; + +// 辅助函数创建测试任务 +//std::shared_ptr<TestingSplitRunner> create_testing_split( +// std::shared_ptr<TestingTicker> ticker, +// std::shared_ptr<PhaseController> begin, +// std::shared_ptr<PhaseController> end, +// int required_phases) { +// return std::make_shared<TestingSplitRunner>( +// "driver", ticker, +// std::make_shared<PhaseController>(0), // global_controller +// begin, +// end, +// required_phases, +// 0); +//} + +// 修改原有测试用例（示例） +TEST_F(TimeSharingTaskExecutorTest, TestTasksComplete) { + auto ticker = std::make_shared<TestingTicker>(); + // const auto stuck_threshold = std::chrono::minutes(10); + + TimeSharingTaskExecutor executor(4, 8, 3, 4, ticker); + ASSERT_TRUE(executor.init().ok()); + ASSERT_TRUE(executor.start().ok()); + + try { + ticker->increment(20, std::chrono::milliseconds(1)); + TaskId task_id("test_task"); + auto task_handle = TEST_TRY(executor.create_task( + task_id, []() { return 0.0; }, 10, std::chrono::milliseconds(1), std::nullopt)); + + auto begin_phase = std::make_shared<PhaseController>(1); + auto verification_complete = std::make_shared<PhaseController>(1); + + // 添加两个初始任务 + auto driver1 = std::make_shared<TestingSplitRunner>( + "driver1", ticker, + std::make_shared<PhaseController>(0), // global_controller + begin_phase, verification_complete, 10, 0); + auto driver2 = std::make_shared<TestingSplitRunner>( + "driver2", ticker, + std::make_shared<PhaseController>(0), // global_controller + begin_phase, verification_complete, 10, 0); + executor.enqueue_splits(task_handle, true, {driver1, driver2}); + + // 验证初始状态 + EXPECT_EQ(driver1->completed_phases(), 0); + EXPECT_EQ(driver2->completed_phases(), 0); + + // 触发第一阶段 + begin_phase->arrive_and_await(); + ticker->increment(60, std::chrono::seconds(1)); + + // 验证停滞任务检测 + // auto stuck_splits = executor.get_stuck_split_task_ids(stuck_threshold, [](auto&){ return true; }); + // EXPECT_TRUE(stuck_splits.empty()) << "应无停滞任务"; + // EXPECT_EQ(executor.get_runaway_split_count(), 0) << "失控任务计数应为0"; + + // 触发任务超时 + ticker->increment(600, std::chrono::seconds(1)); + // stuck_splits = executor.get_stuck_split_task_ids(stuck_threshold, [](auto&){ return true; }); + // EXPECT_EQ(stuck_splits.size(), 1) << "应检测到1个停滞任务"; + // EXPECT_EQ(*stuck_splits.begin(), task_id) << "停滞任务ID不匹配"; + // EXPECT_EQ(executor.get_runaway_split_count(), 2) << "应检测到2个失控任务"; + verification_complete->arrive_and_await(); + + // advance one phase and verify + begin_phase->arrive_and_await(); + EXPECT_EQ(driver1->completed_phases(), 1); + EXPECT_EQ(driver2->completed_phases(), 1); + verification_complete->arrive_and_await(); + + // add one more job + auto driver3 = std::make_shared<TestingSplitRunner>( + "driver3", ticker, + std::make_shared<PhaseController>(0), // global_controller + begin_phase, verification_complete, 10, 0); + executor.enqueue_splits(task_handle, false, {driver3}); + // advance one phase and verify + begin_phase->arrive_and_await(); + EXPECT_EQ(driver1->completed_phases(), 2); + EXPECT_EQ(driver2->completed_phases(), 2); + EXPECT_EQ(driver3->completed_phases(), 0); + // future1.get(1, SECONDS); + // future2.get(1, SECONDS); + verification_complete->arrive_and_await(); + + // advance to the end of the first two task and verify + begin_phase->arrive_and_await(); + for (int i = 0; i < 7; i++) { + verification_complete->arrive_and_await(); + begin_phase->arrive_and_await(); + EXPECT_EQ(begin_phase->generation(), verification_complete->generation() + 1); + } + EXPECT_EQ(driver1->completed_phases(), 10); + EXPECT_EQ(driver2->completed_phases(), 10); + EXPECT_EQ(driver3->completed_phases(), 8); + // future3.get(1, SECONDS); + verification_complete->arrive_and_await(); + + // advance two more times and verify + begin_phase->arrive_and_await(); + verification_complete->arrive_and_await(); + begin_phase->arrive_and_await(); + EXPECT_EQ(driver1->completed_phases(), 10); + EXPECT_EQ(driver2->completed_phases(), 10); + EXPECT_EQ(driver3->completed_phases(), 10); + // future3.get(1, SECONDS); + verification_complete->arrive_and_await(); + // + EXPECT_EQ(driver1->first_phase(), 0); + EXPECT_EQ(driver2->first_phase(), 0); + EXPECT_EQ(driver3->first_phase(), 2); + EXPECT_EQ(driver1->last_phase(), 10); + EXPECT_EQ(driver2->last_phase(), 10); + EXPECT_EQ(driver3->last_phase(), 12); + + // 清理后验证无滞留任务 + ticker->increment(610, std::chrono::seconds(1)); + // stuck_splits = executor.get_stuck_split_task_ids(stuck_threshold, [](auto&){ return true; }); + // EXPECT_TRUE(stuck_splits.empty()) << "最终应无停滞任务"; + // EXPECT_EQ(executor.get_runaway_split_count(), 0) << "最终失控任务计数应为0"; + } catch (...) { + executor.stop(); + throw; + } + executor.stop(); +} + +TEST_F(TimeSharingTaskExecutorTest, TestQuantaFairness) { + constexpr int TOTAL_PHASES = 11; + std::shared_ptr<TestingTicker> ticker = std::make_shared<TestingTicker>(); + + TimeSharingTaskExecutor executor(1, 2, 3, 4, ticker); + ASSERT_TRUE(executor.init().ok()); + ASSERT_TRUE(executor.start().ok()); + + ticker->increment(20, std::chrono::milliseconds(1)); + // 创建两个独立的任务句柄 + TaskId short_task_id("short_quanta"); + TaskId long_task_id("long_quanta"); + auto short_handle = TEST_TRY(executor.create_task( + short_task_id, []() { return 0.0; }, // 优先级函数 + 10, // 初始并发数 + std::chrono::milliseconds(1), std::nullopt)); + + auto long_handle = TEST_TRY(executor.create_task( + long_task_id, []() { return 0.0; }, 10, std::chrono::milliseconds(1), std::nullopt)); + + auto end_controller = std::make_shared<PhaseController>(0); + + auto short_split = std::make_shared<TestingSplitRunner>( + "short", ticker, std::make_shared<PhaseController>(0), + std::make_shared<PhaseController>(0), end_controller, TOTAL_PHASES, 10); + + auto long_split = std::make_shared<TestingSplitRunner>( + "long", ticker, std::make_shared<PhaseController>(0), + std::make_shared<PhaseController>(0), end_controller, TOTAL_PHASES, 20); + + // 添加任务到对应的句柄 + executor.enqueue_splits(short_handle, true, {short_split}); + executor.enqueue_splits(long_handle, true, {long_split}); + + // 修改主线程同步逻辑 + for (int i = 0; i < TOTAL_PHASES; ++i) { + end_controller->arrive_and_await(); + } + + // fprintf(stderr, "验证阶段数\n"); + + // 验证阶段数 + EXPECT_GE(short_split->completed_phases(), 7); + EXPECT_LE(short_split->completed_phases(), 8); + EXPECT_GE(long_split->completed_phases(), 3); + EXPECT_LE(long_split->completed_phases(), 4); + end_controller->arrive_and_deregister(); + executor.stop(); +} + +TEST_F(TimeSharingTaskExecutorTest, TestLevelMovement) { + // fprintf(stderr, "[%zu] TestLevelMovement\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id())); + + std::shared_ptr<TestingTicker> ticker = std::make_shared<TestingTicker>(); + TimeSharingTaskExecutor executor(2, 2, 3, 4, ticker); + ASSERT_TRUE(executor.init().ok()); + ASSERT_TRUE(executor.start().ok()); + + try { + // 初始化时间 + ticker->increment(20, std::chrono::milliseconds(1)); + + // 创建任务句柄 + TaskId task_id("test_task"); + auto task_handle = + std::static_pointer_cast<TimeSharingTaskHandle>(TEST_TRY(executor.create_task( + task_id, []() { return 0.0; }, // 优先级函数 + 10, // 初始并发数 + std::chrono::milliseconds(1), std::nullopt))); + + // 同步控制器 + auto global_controller = std::make_shared<PhaseController>(3); // 2个执行线程 + 主线程 + const int quanta_time_ms = 500; + const int phases_per_second = 1000 / quanta_time_ms; + const int total_phases = + MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS + [MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS.size() - 1] * + phases_per_second; // 假设最后一级阈值 + + // fprintf(stderr, "total_phases: %d\n", total_phases); + // 创建测试任务 + auto driver1 = std::make_shared<TestingSplitRunner>( + "driver1", ticker, global_controller, std::make_shared<PhaseController>(0), + std::make_shared<PhaseController>(0), total_phases, quanta_time_ms); + auto driver2 = std::make_shared<TestingSplitRunner>( + "driver2", ticker, global_controller, std::make_shared<PhaseController>(0), + std::make_shared<PhaseController>(0), total_phases, quanta_time_ms); + + // 添加任务 + executor.enqueue_splits(task_handle, false, {driver1, driver2}); + + int completed_phases = 0; + for (int i = 0; i < MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS.size() - 1; ++i) { + const int target_seconds = MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS[i + 1]; + while ((completed_phases / phases_per_second) < target_seconds) { + // fprintf(stderr, "completed_phases: %d, phases_per_second: %d, target_seconds: %d\n", + // completed_phases, phases_per_second, target_seconds); + // fprintf(stderr, + // "[%zu] TestLevelMovement _global_controller.arrive_and_await()" + // " _global_controller.count(): %d, _global_controller.parties(): %d;\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id()), + // global_controller->count(), global_controller->parties()); + global_controller->arrive_and_await(); + // fprintf(stderr, + // "[%zu] TestLevelMovement _global_controller.arrive_and_await() finished;\n", + // std::hash<std::thread::id> {}(std::this_thread::get_id())); + ++completed_phases; + } + + // 验证优先级级别 + EXPECT_EQ(task_handle->priority().level(), i + 1); + } + + global_controller->arrive_and_deregister(); + } catch (...) { + executor.stop(); + throw; + } + executor.stop(); +} + +TEST_F(TimeSharingTaskExecutorTest, TestLevelMultipliers) { + constexpr int TASK_THREADS = 6; + constexpr int CONCURRENCY = 3; + + std::shared_ptr<TestingTicker> ticker = std::make_shared<TestingTicker>(); + std::shared_ptr<MultilevelSplitQueue> split_queue = + std::make_shared<MultilevelSplitQueue>(2); // 初始化多级队列 + TimeSharingTaskExecutor executor(TASK_THREADS, CONCURRENCY, 3, 4, ticker, + std::chrono::milliseconds(60000), split_queue); + ASSERT_TRUE(executor.init().ok()); + ASSERT_TRUE(executor.start().ok()); + + try { + ticker->increment(20, std::chrono::milliseconds(1)); + + for (int i = 0; + i < (sizeof(MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS) / sizeof(int) - 1); ++i) { + // 创建三个任务句柄 + TaskId task_id1("test1"), task_id2("test2"), task_id3("test3"); + auto task_handle1 = TEST_TRY(executor.create_task( + task_id1, []() { return 0.0; }, 10, std::chrono::milliseconds(1), + std::nullopt)); + auto task_handle2 = TEST_TRY(executor.create_task( + task_id2, []() { return 0.0; }, 10, std::chrono::milliseconds(1), + std::nullopt)); + auto task_handle3 = TEST_TRY(executor.create_task( + task_id3, []() { return 0.0; }, 10, std::chrono::milliseconds(1), + std::nullopt)); + + // 移动任务0到下一级别 + auto task0_job = std::make_shared<TestingSplitRunner>( + "task0", ticker, std::make_shared<PhaseController>(0), + std::make_shared<PhaseController>(0), std::make_shared<PhaseController>(0), 1, + MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS[i + 1] * 1000); + executor.enqueue_splits(task_handle1, false, {task0_job}); + + // 移动任务1和2到当前级别 + auto task1_job = std::make_shared<TestingSplitRunner>( + "task1", ticker, std::make_shared<PhaseController>(0), + std::make_shared<PhaseController>(0), std::make_shared<PhaseController>(0), 1, + MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS[i] * 1000); + auto task2_job = std::make_shared<TestingSplitRunner>( + "task2", ticker, std::make_shared<PhaseController>(0), + std::make_shared<PhaseController>(0), std::make_shared<PhaseController>(0), 1, + MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS[i] * 1000); + executor.enqueue_splits(task_handle2, false, {task1_job}); + executor.enqueue_splits(task_handle3, false, {task2_job}); + + // 等待任务完成 + while (!task0_job->is_finished() || !task1_job->is_finished() || + !task2_job->is_finished()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + // fprintf(stderr, "start new task\n"); + // 启动新任务 + auto global_controller = std::make_shared<PhaseController>(7); // 6个执行线程 + 主线程 + const int phases_for_next_level = MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS[i + 1] - + MultilevelSplitQueue::LEVEL_THRESHOLD_SECONDS[i]; + std::array<std::shared_ptr<TestingSplitRunner>, 6> drivers; + + for (int j = 0; j < 6; ++j) { + drivers[j] = std::make_shared<TestingSplitRunner>( + "driver" + std::to_string(j), ticker, global_controller, + std::make_shared<PhaseController>(0), std::make_shared<PhaseController>(0), + phases_for_next_level, 1000); + } + + executor.enqueue_splits(task_handle1, false, {drivers[0], drivers[1]}); + executor.enqueue_splits(task_handle2, false, {drivers[2], drivers[3]}); + executor.enqueue_splits(task_handle3, false, {drivers[4], drivers[5]}); + + // 收集初始阶段数 + int lower_level_start = 0, higher_level_start = 0; + for (int j = 2; j < 6; ++j) lower_level_start += drivers[j]->completed_phases(); + for (int j = 0; j < 2; ++j) higher_level_start += drivers[j]->completed_phases(); + + // 运行所有任务 + while (std::any_of(drivers.begin(), drivers.end(), + [](auto& d) { return !d->is_finished(); })) { + global_controller->arrive_and_await(); + + int lower_level_end = 0, higher_level_end = 0; + for (int j = 2; j < 6; ++j) lower_level_end += drivers[j]->completed_phases(); + for (int j = 0; j < 2; ++j) higher_level_end += drivers[j]->completed_phases(); + + int lower_level_time = lower_level_end - lower_level_start; + int higher_level_time = higher_level_end - higher_level_start; + + if (higher_level_time > 20) { + EXPECT_GT(lower_level_time, higher_level_time * 2 - 10); + EXPECT_LT(higher_level_time, lower_level_time * 2 + 10); + } + } + + global_controller->arrive_and_deregister(); + static_cast<void>(executor.remove_task(task_handle1)); + static_cast<void>(executor.remove_task(task_handle2)); + static_cast<void>(executor.remove_task(task_handle3)); + } + } catch (...) { + executor.stop(); + throw; + } + executor.stop(); +} + +TEST_F(TimeSharingTaskExecutorTest, TestTaskHandle) { + std::shared_ptr<TestingTicker> ticker = std::make_shared<TestingTicker>(); + TimeSharingTaskExecutor executor(4, 8, 3, 4, ticker); + ASSERT_TRUE(executor.init().ok()); + ASSERT_TRUE(executor.start().ok()); + + try { + // 创建任务句柄 + TaskId task_id("test_task"); + auto task_handle = + std::static_pointer_cast<TimeSharingTaskHandle>(TEST_TRY(executor.create_task( + task_id, []() { return 0.0; }, 10, std::chrono::milliseconds(1), + std::nullopt))); + + // 准备同步控制器 + auto begin_phase = std::make_shared<PhaseController>(1); + auto verification_complete = std::make_shared<PhaseController>(1); + + // 创建测试任务 + auto driver1 = std::make_shared<TestingSplitRunner>( + "driver1", ticker, + std::make_shared<PhaseController>(0), // global_controller + begin_phase, verification_complete, + 10, // required_phases + 0); // quanta_time + + auto driver2 = std::make_shared<TestingSplitRunner>( + "driver2", ticker, + std::make_shared<PhaseController>(0), // global_controller + begin_phase, verification_complete, + 10, // required_phases + 0); // quanta_time + + // // 强制入队split（不立即执行） + // executor.enqueue_splits(task_handle, true, {driver1}); + // EXPECT_EQ(task_handle->get_running_leaf_splits(), 0) + // << "强制入队后应立即运行的任务数应为0"; + // + // // 正常入队split + // executor.enqueue_splits(task_handle, false, {driver2}); + // EXPECT_EQ(task_handle->get_running_leaf_splits(), 1) + // << "正常入队后应立即运行的任务数应为1"; + + // // 正常入队split + // executor.enqueue_splits(task_handle, false, {driver2}); + // EXPECT_EQ(task_handle->get_running_leaf_splits(), 2) + // << "正常入队后应立即运行的任务数应为2"; + + executor.enqueue_splits(task_handle, true, {driver1}); + EXPECT_EQ(task_handle->running_leaf_splits(), 0) << "强制入队后应立即运行的任务数应为0"; + + // 正常入队split + executor.enqueue_splits(task_handle, false, {driver2}); + EXPECT_EQ(task_handle->running_leaf_splits(), 1) << "正常入队后应立即运行的任务数应为1"; + + // 释放同步锁让任务继续执行 + begin_phase->arrive_and_deregister(); + verification_complete->arrive_and_deregister(); + + // 等待任务完成 + while (!driver1->is_finished() || !driver2->is_finished()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + } catch (...) { + executor.stop(); + throw; + } + executor.stop(); +} + +// 辅助函数实现（与之前测试保持一致） +void wait_until_splits_start(const std::vector<std::shared_ptr<TestingSplitRunner>>& splits) { + constexpr auto TIMEOUT = std::chrono::seconds(30); + auto start = std::chrono::steady_clock::now(); + + while (std::any_of(splits.begin(), splits.end(), + [](const auto& split) { return !split->is_started(); })) { + if (std::chrono::steady_clock::now() - start > TIMEOUT) { + throw std::runtime_error("等待split启动超时"); + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } +} + +TEST_F(TimeSharingTaskExecutorTest, TestMinMaxDriversPerTask) { + constexpr int MAX_DRIVERS_PER_TASK = 2; + constexpr int TASK_THREADS = 4; + constexpr int CONCURRENCY = 16; + constexpr int BATCH_COUNT = 4; + constexpr int SPLITS_PER_BATCH = 2; + + std::shared_ptr<TestingTicker> ticker = std::make_shared<TestingTicker>(); + std::shared_ptr<MultilevelSplitQueue> split_queue = + std::make_shared<MultilevelSplitQueue>(2); // 初始化多级队列 + TimeSharingTaskExecutor executor(TASK_THREADS, CONCURRENCY, 1, MAX_DRIVERS_PER_TASK, ticker, + std::chrono::milliseconds(60000), split_queue); + ASSERT_TRUE(executor.init().ok()); + ASSERT_TRUE(executor.start().ok()); + + try { + TaskId task_id("test_task"); + auto task_handle = TEST_TRY(executor.create_task( + task_id, []() { return 0.0; }, 10, std::chrono::milliseconds(1), std::nullopt)); + + std::array<std::shared_ptr<TestingSplitRunner>, BATCH_COUNT * SPLITS_PER_BATCH> splits; + std::array<std::shared_ptr<PhaseController>, BATCH_COUNT> batch_controllers; + + // 准备4个批次的split，每批2个 + for (int batch = 0; batch < BATCH_COUNT; ++batch) { + batch_controllers[batch] = std::make_shared<PhaseController>(0); + batch_controllers[batch]->register_party(); + auto& controller = batch_controllers[batch]; + + for (int i = 0; i < SPLITS_PER_BATCH; ++i) { + splits[batch * SPLITS_PER_BATCH + i] = std::make_shared<TestingSplitRunner>( + "batch" + std::to_string(batch) + "_split" + std::to_string(i), ticker, + std::make_shared<PhaseController>(0), // global_controller + std::make_shared<PhaseController>(0), // begin_controller + controller, // end_controller + 1, // required_phases + 0); // quanta_time + } + + executor.enqueue_splits( + task_handle, false, + {splits[batch * SPLITS_PER_BATCH], splits[batch * SPLITS_PER_BATCH + 1]}); + } + + // fprintf(stderr, "wait and validation.\n"); + // 验证分批执行逻辑 + for (int current_batch = 0; current_batch < BATCH_COUNT; ++current_batch) { + // 等待当前批次split启动 + // fprintf(stderr, "wait_until_splits_start batch: %d\n", current_batch); + wait_until_splits_start({splits[current_batch * SPLITS_PER_BATCH], + splits[current_batch * SPLITS_PER_BATCH + 1]}); + // fprintf(stderr, "wait_until_splits_start batch finished: %d\n", current_batch); + + // 替换原有的验证逻辑 + assert_split_states(current_batch * SPLITS_PER_BATCH + 1, splits); + + // 完成当前批次 + batch_controllers[current_batch]->arrive_and_deregister(); + } + + // 验证所有split最终完成 + for (auto& split : splits) { + while (!split->is_finished()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + } + } catch (...) { + executor.stop(); + throw; + } + executor.stop(); +} + +TEST_F(TimeSharingTaskExecutorTest, TestUserSpecifiedMaxDriversPerTask) { + constexpr int TASK_THREADS = 4; + constexpr int CONCURRENCY = 16; + constexpr int BATCH_COUNT = 4; + + std::shared_ptr<TestingTicker> ticker = std::make_shared<TestingTicker>(); + std::shared_ptr<MultilevelSplitQueue> split_queue = std::make_shared<MultilevelSplitQueue>(2); + TimeSharingTaskExecutor executor(TASK_THREADS, CONCURRENCY, 2, 4, ticker, + std::chrono::milliseconds(60000), split_queue); + ASSERT_TRUE(executor.init().ok()); + ASSERT_TRUE(executor.start().ok()); + + try { + // 创建任务并覆盖最大驱动器数为1 + TaskId task_id("user_specified_task"); + auto task_handle = TEST_TRY(executor.create_task( + task_id, []() { return 0.0; }, 10, std::chrono::milliseconds(1), + std::optional<int>(1))); // 显式指定最大驱动器数 + + std::array<std::shared_ptr<TestingSplitRunner>, BATCH_COUNT> splits; + std::array<std::shared_ptr<PhaseController>, BATCH_COUNT> batch_controllers; + + // 准备4个批次的split + for (int batch = 0; batch < BATCH_COUNT; ++batch) { + batch_controllers[batch] = std::make_shared<PhaseController>(0); + batch_controllers[batch]->register_party(); + + splits[batch] = std::make_shared<TestingSplitRunner>( + "batch" + std::to_string(batch), ticker, + std::make_shared<PhaseController>(0), // global_controller + std::make_shared<PhaseController>(0), // begin_controller + batch_controllers[batch], // end_controller + 1, // required_phases + 0); // quanta_time + + executor.enqueue_splits(task_handle, false, {splits[batch]}); + } + + // 验证分批执行逻辑 + for (int current_batch = 0; current_batch < BATCH_COUNT; ++current_batch) { + // 等待当前批次split启动 + wait_until_splits_start({splits[current_batch]}); + + // 验证当前批次状态 + assert_split_states(current_batch, splits); + + // 完成当前批次 + batch_controllers[current_batch]->arrive_and_deregister(); + } + + // 验证所有split最终完成 + for (auto& split : splits) { + while (!split->is_finished()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + } + } catch (...) { + executor.stop(); + throw; + } + executor.stop(); +} + +TEST_F(TimeSharingTaskExecutorTest, TestMinDriversPerTaskWhenTargetConcurrencyIncreases) { + constexpr int TASK_THREADS = 4; + constexpr int CONCURRENCY = 1; + constexpr int BATCH_COUNT = 3; + + std::shared_ptr<TestingTicker> ticker = std::make_shared<TestingTicker>(); + std::shared_ptr<MultilevelSplitQueue> split_queue = std::make_shared<MultilevelSplitQueue>(2); + TimeSharingTaskExecutor executor(TASK_THREADS, CONCURRENCY, 2, 2, ticker, + std::chrono::milliseconds(60000), split_queue); + ASSERT_TRUE(executor.init().ok()); + ASSERT_TRUE(executor.start().ok()); + + try { + // 创建任务并指定最小并发数 + TaskId task_id("test_query"); + auto task_handle = TEST_TRY(executor.create_task( + task_id, []() { return 0.0; }, // 确保缓冲区未充分利用 + 1, // 初始并发数 + std::chrono::milliseconds(1), + std::optional<int>(2))); // 显式指定最小并发数 + + std::array<std::shared_ptr<TestingSplitRunner>, BATCH_COUNT> splits; + std::array<std::shared_ptr<PhaseController>, BATCH_COUNT> batch_controllers; + + // 创建3个split + for (int batch = 0; batch < BATCH_COUNT; ++batch) { + batch_controllers[batch] = std::make_shared<PhaseController>(0); + batch_controllers[batch]->register_party(); + + splits[batch] = std::make_shared<TestingSplitRunner>( + "split" + std::to_string(batch), ticker, + std::make_shared<PhaseController>(0), // global_controller + std::make_shared<PhaseController>(0), // begin_controller + batch_controllers[batch], // end_controller + 1, // required_phases + 0); // quanta_time + } + + // 批量添加所有split + executor.enqueue_splits(task_handle, false, {splits[0], splits[1], splits[2]}); + + // 等待第一个split启动 + wait_until_splits_start({splits[0]}); + + // 验证只有第一个split启动 + assert_split_states(0, splits); + + // 完成第一个split（应触发并发数调整） + batch_controllers[0]->arrive_and_deregister(); + + // 等待剩余split启动 + wait_until_splits_start({splits[1], splits[2]}); + + // 验证所有split已启动 + assert_split_states(2, splits); + + // 完成剩余split + for (int batch = 1; batch < BATCH_COUNT; ++batch) { + batch_controllers[batch]->arrive_and_deregister(); + } + + // 验证所有split完成 + for (auto& split : splits) { + while (!split->is_finished()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + } + } catch (...) { + executor.stop(); + throw; + } + executor.stop(); +} + +TEST_F(TimeSharingTaskExecutorTest, TestLeafSplitsSize) {} + +//TEST_F(TimeSharingTaskExecutorTest, TaskRotation) { +// constexpr int TASK_THREADS = 4; +// constexpr int CONCURRENCY = 1; +// constexpr int BATCH_COUNT = 3; +// +// std::shared_ptr<TestingTicker> ticker = std::make_shared<TestingTicker>(); +// std::shared_ptr<MultilevelSplitQueue> split_queue = std::make_shared<MultilevelSplitQueue>(2); +// +// TimeSharingTaskExecutor executor(TASK_THREADS, CONCURRENCY, 2, 2, ticker, +// std::chrono::milliseconds(60000), split_queue); +// ASSERT_TRUE(executor.init().ok()); +// ASSERT_TRUE(executor.start().ok()); +// // 创建三个任务 +// auto task1 = std::make_shared<TimeSharingTaskHandle>( +// "task1", executor->_waiting_splits, []() { return 0.5; }, 1, +// std::chrono::milliseconds(100), 2); +// auto task2 = std::make_shared<TimeSharingTaskHandle>( +// "task2", executor->_waiting_splits, []() { return 0.5; }, 1, +// std::chrono::milliseconds(100), 2); +// auto task3 = std::make_shared<TimeSharingTaskHandle>( +// "task3", executor->_waiting_splits, []() { return 0.5; }, 1, +// std::chrono::milliseconds(100), 2); +// +// // 添加任务到执行器 +// executor->add_task("task1", task1); +// executor->add_task("task2", task2); +// executor->add_task("task3", task3); +// +// // 添加分片到任务 +// auto split1 = std::make_shared<MockSplitRunner>(); +// auto split2 = std::make_shared<MockSplitRunner>(); +// auto split3 = std::make_shared<MockSplitRunner>(); +// task1->enqueue_split(split1); +// task2->enqueue_split(split2); +// task3->enqueue_split(split3); +// +// // 测试任务轮转 +// { +// std::lock_guard<std::mutex> guard(executor->_mutex); +// +// // 第一次轮询，应该返回 task1 的分片 +// auto result1 = executor->_poll_next_split_worker(guard); +// EXPECT_EQ(result1, split1); +// +// // 第二次轮询，应该返回 task2 的分片 +// auto result2 = executor->_poll_next_split_worker(guard); +// EXPECT_EQ(result2, split2); +// +// // 第三次轮询，应该返回 task3 的分片 +// auto result3 = executor->_poll_next_split_worker(guard); +// EXPECT_EQ(result3, split3); +// +// // 第四次轮询，应该再次返回 task1 的分片（轮转） +// auto result4 = executor->_poll_next_split_worker(guard); +// EXPECT_EQ(result4, split1); +// } +// +// // 测试任务轮转时跳过没有分片的任务 +// { +// std::lock_guard<std::mutex> guard(executor->_mutex); +// +// // 清空 task2 的分片 +// task2->set_running_leaf_splits(2); +// +// // 第一次轮询，应该返回 task3 的分片（跳过 task2） +// auto result1 = executor->_poll_next_split_worker(guard); +// EXPECT_EQ(result1, split3); +// +// // 第二次轮询，应该返回 task1 的分片（跳过 task2） +// auto result2 = executor->_poll_next_split_worker(guard); +// EXPECT_EQ(result2, split1); +// } +//} + +class PhaseControllerTest : public testing::Test { +protected: + const int THREAD_COUNT = 4; + std::shared_ptr<PhaseController> controller; + + void SetUp() override { controller = std::make_shared<PhaseController>(THREAD_COUNT); } +}; + +// 基础同步测试 +TEST_F(PhaseControllerTest, BasicSynchronization) { + std::atomic<int> ready {0}; + std::vector<std::thread> threads; + const int GENERATIONS = 3; + + for (int g = 0; g < GENERATIONS; ++g) { + for (int i = 0; i < THREAD_COUNT; ++i) { + threads.emplace_back([&, g] { + // 同步线程启动 + if (ready.fetch_add(1) == THREAD_COUNT - 1) { + ready.store(0); + } else { + while (ready.load() != 0) + ; + } + + int gen = controller->arrive_and_await(); + EXPECT_EQ(gen, g + 1) << "代际计数错误"; + }); + } + + for (auto& t : threads) t.join(); + threads.clear(); + + EXPECT_EQ(controller->generation(), g + 1); + EXPECT_EQ(controller->count(), THREAD_COUNT) << "每代应重置计数器"; + } +} + +// 动态参与者管理测试 +TEST_F(PhaseControllerTest, DynamicRegistration) { + constexpr int NEW_PARTIES = 2; + std::atomic<int> completed {0}; + std::vector<std::thread> threads; + std::mutex cv_mtx; + std::condition_variable cv; + + // 修改初始参与方数（包含主线程） + controller = std::make_shared<PhaseController>(THREAD_COUNT + 1); + + // 初始参与者 (4 threads + 主线程) + for (int i = 0; i < THREAD_COUNT; ++i) { + threads.emplace_back([&] { + // 第一阶段同步 + controller->arrive_and_await(); + // fprintf(stderr, "[初始线程 %d] 第一阶段完成\n", idx); + + // 第二阶段同步 + controller->arrive_and_await(); + // fprintf(stderr, "[初始线程 %d] 第二阶段完成\n", idx); + + completed++; + }); + } + + // 主线程第一阶段同步 + // fprintf(stderr, "主线程开始同步第一阶段\n"); + int main_gen = controller->arrive_and_await(); + fprintf(stderr, "第一阶段完成，当前代际=%d\n", main_gen); + + // 新注册参与者 (2 threads) + std::vector<std::thread> new_threads; + for (int i = 0; i < NEW_PARTIES; ++i) { + new_threads.emplace_back([&] { + { + std::lock_guard<std::mutex> lk(cv_mtx); + controller->register_party(); + // fprintf(stderr, "[新线程 %d] 注册完成，当前参与方=%d\n", my_id, + // controller->parties()); + } + cv.notify_one(); + + // 只参与第二阶段同步 + controller->arrive_and_await(); + completed++; + // fprintf(stderr, "[新线程 %d] 完成\n", my_id); + }); + } + + // 等待新线程完成注册 + { + std::unique_lock<std::mutex> lk(cv_mtx); + cv.wait(lk, [&] { return controller->parties() == THREAD_COUNT + 1 + NEW_PARTIES; }); + } + + // 主线程触发第二阶段同步 + // fprintf(stderr, "主线程开始同步第二阶段\n"); + main_gen = controller->arrive_and_await(); + // fprintf(stderr, "第二阶段完成，当前代际=%d\n", main_gen); + + // 等待所有线程完成 + for (auto& t : threads) t.join(); + for (auto& t : new_threads) t.join(); + + // 验证结果 + // fprintf(stderr, "总完成数=%d (预期=%d)\n", completed.load(), THREAD_COUNT + NEW_PARTIES); + // fprintf(stderr, "当前参与方=%d (预期=%d)\n", controller->parties(), + // THREAD_COUNT + 1 + NEW_PARTIES); + + EXPECT_EQ(completed.load(), THREAD_COUNT + NEW_PARTIES); + EXPECT_EQ(controller->parties(), THREAD_COUNT + 1 + NEW_PARTIES); +} + +// // 注销参与者测试 +// TEST_F(PhaseControllerTest, DeregistrationTest) { +// std::vector<std::thread> threads; + +// // 第一个线程注销 +// threads.emplace_back([&] { +// controller->arrive_and_deregister(); +// EXPECT_EQ(controller->parties(), THREAD_COUNT - 1); +// }); + +// // 其他线程正常完成 +// for (int i = 1; i < THREAD_COUNT; ++i) { +// threads.emplace_back([&] { +// controller->arrive_and_await(); +// }); +// } + +// for (auto& t : threads) t.join(); + +// EXPECT_EQ(controller->parties(), THREAD_COUNT - 1); +// EXPECT_EQ(controller->count(), THREAD_COUNT - 1) << "注销后计数器应更新"; +// } + +// // 混合操作压力测试 +// TEST_F(PhaseControllerTest, ConcurrencyStressTest) { +// const int TOTAL_OPERATIONS = 1000; +// std::vector<std::thread> threads; +// std::atomic<int> ops {0}; + +// auto worker = [&] { +// std::random_device rd; +// std::mt19937 gen(rd()); +// std::uniform_int_distribution<> dist(0, 2); + +// while (ops.fetch_add(1) < TOTAL_OPERATIONS) { +// switch (dist(gen)) { +// case 0: +// controller->arrive_and_await(); +// break; +// case 1: +// controller->register_party(); +// break; +// case 2: +// if (controller->parties() > 0) { +// controller->arrive_and_deregister(); +// } +// break; +// } +// } +// }; + +// // 启动工作线程 +// for (int i = 0; i < 4; ++i) { +// threads.emplace_back(worker); +// } + +// // 验证最终状态一致性 +// for (auto& t : threads) t.join(); + +// EXPECT_GE(controller->parties(), 0) << "参与者数不应为负"; +// EXPECT_GE(controller->count(), 0) << "计数器不应为负"; +// EXPECT_GE(controller->generation(), 0) << "代际计数不应回退"; +// } + +// // 边界条件测试 +// TEST_F(PhaseControllerTest, EdgeCases) { +// // 测试空控制器 +// auto empty_controller = std::make_shared<PhaseController>(0); +// EXPECT_EQ(empty_controller->arrive_and_await(), 0); + +// // 测试单参与者 +// auto single_controller = std::make_shared<PhaseController>(1); +// std::thread t([&] { +// EXPECT_EQ(single_controller->arrive_and_await(), 1); +// }); +// t.join(); +// EXPECT_EQ(single_controller->generation(), 1); + +// // 测试多次注销 +// auto controller = std::make_shared<PhaseController>(2); +// controller->arrive_and_deregister(); +// controller->arrive_and_deregister(); +// EXPECT_EQ(controller->parties(), 0); +// } + +} // namespace vectorized +} // namespace doris