be/src/runtime/workload_management/task_controller.h
Line | Count | Source |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #pragma once |
19 | | |
20 | | #include <gen_cpp/PaloInternalService_types.h> |
21 | | #include <gen_cpp/Types_types.h> |
22 | | |
23 | | #include "common/factory_creator.h" |
24 | | #include "common/status.h" |
25 | | #include "util/time.h" |
26 | | |
27 | | namespace doris { |
28 | | #include "common/compile_check_begin.h" |
29 | | class PipelineTask; |
30 | | |
31 | | class ResourceContext; |
32 | | class TaskController { |
33 | | ENABLE_FACTORY_CREATOR(TaskController); |
34 | | |
35 | | public: |
36 | 244k | TaskController() { task_id_ = TUniqueId(); }; |
37 | 244k | virtual ~TaskController() = default; |
38 | | |
39 | | /* common action |
40 | | */ |
41 | 502 | const TUniqueId& task_id() const { return task_id_; } |
42 | 121k | void set_task_id(TUniqueId task_id) { |
43 | 121k | task_id_ = task_id; |
44 | 121k | start_time_ = MonotonicMillis(); |
45 | 121k | } |
46 | 51 | TQueryType::type query_type() { return query_type_; } |
47 | 121k | void set_query_type(TQueryType::type query_type) { query_type_ = query_type; } |
48 | 13 | TNetworkAddress fe_addr() { return fe_addr_; } |
49 | 121k | void set_fe_addr(TNetworkAddress fe_addr) { fe_addr_ = fe_addr; } |
50 | | std::string debug_string(); |
51 | | |
52 | | /* finish action |
53 | | */ |
54 | 1 | bool is_finished() const { return is_finished_; } |
55 | 121k | void finish() { |
56 | 121k | if (!is_finished_) { |
57 | 121k | is_finished_ = true; |
58 | 121k | finish_time_ = MonotonicMillis(); |
59 | 121k | } |
60 | 121k | finish_impl(); |
61 | 121k | } |
62 | 121k | virtual void finish_impl() {} |
63 | 21 | int64_t start_time() const { return start_time_; } |
64 | 7 | int64_t finish_time() const { return finish_time_; } |
65 | 10 | int64_t running_time() const { |
66 | 10 | if (start_time() == 0) { |
67 | 1 | return 0; |
68 | 1 | } |
69 | 9 | if (is_finished_) { |
70 | 7 | return finish_time() - start_time(); |
71 | 7 | } else { |
72 | 2 | return MonotonicMillis() - start_time(); |
73 | 2 | } |
74 | 9 | } |
75 | | |
76 | | /* cancel action |
77 | | */ |
78 | 0 | virtual bool is_cancelled() const { return false; } |
79 | | |
80 | 16 | bool cancel(const Status& reason) { |
81 | 16 | if (cancelled_time_ == 0) { |
82 | 16 | cancelled_time_ = MonotonicMillis(); |
83 | 16 | } |
84 | 16 | return cancel_impl(reason); |
85 | 16 | } |
86 | | |
87 | 1 | int64_t cancel_elapsed_millis() const { return MonotonicMillis() - cancelled_time_; } |
88 | | |
89 | 0 | virtual bool cancel_impl(const Status& reason) { return false; } |
90 | | |
91 | 27 | int64_t cancelled_time() const { return cancelled_time_; } |
92 | | |
93 | | /* pause action & property |
94 | | */ |
95 | | void update_paused_reason(const Status& st); |
96 | 8 | void reset_paused_reason() { paused_reason_.reset(); } |
97 | 26 | Status paused_reason() { return paused_reason_.status(); } |
98 | 10 | void add_paused_count() { paused_count_.fetch_add(1); } |
99 | | |
100 | | /* memory status action |
101 | | */ |
102 | 0 | virtual int32_t get_slot_count() const { return 1; } |
103 | 0 | virtual bool is_pure_load_task() const { return false; } |
104 | 12 | void set_low_memory_mode(bool low_memory_mode) { low_memory_mode_ = low_memory_mode; } |
105 | 902k | bool low_memory_mode() { return low_memory_mode_; } |
106 | 3 | virtual void disable_reserve_memory() { enable_reserve_memory_ = false; } |
107 | 0 | virtual bool is_enable_reserve_memory() const { return enable_reserve_memory_; } |
108 | 0 | virtual void set_memory_sufficient(bool sufficient) {}; |
109 | 0 | virtual int64_t memory_sufficient_time() { return 0; }; |
110 | | |
111 | | /* memory revoke action |
112 | | */ |
113 | | virtual void get_revocable_info(size_t* revocable_size, size_t* memory_usage, |
114 | 0 | bool* has_running_task) {}; |
115 | 0 | virtual size_t get_revocable_size() { return 0; }; |
116 | 0 | virtual Status revoke_memory() { return Status::OK(); }; |
117 | 0 | virtual std::vector<PipelineTask*> get_revocable_tasks() { return {}; }; |
118 | 14 | void increase_revoking_tasks_count() { revoking_tasks_count_.fetch_add(1); } |
119 | 14 | void decrease_revoking_tasks_count() { revoking_tasks_count_.fetch_sub(1); } |
120 | 0 | int get_revoking_tasks_count() const { return revoking_tasks_count_.load(); } |
121 | | |
122 | | protected: |
123 | | friend class ResourceContext; |
124 | | |
125 | 244k | void set_resource_ctx(ResourceContext* resource_ctx) { resource_ctx_ = resource_ctx; } |
126 | | ResourceContext* resource_ctx_ {nullptr}; |
127 | | |
128 | | /* common property |
129 | | */ |
130 | | TUniqueId task_id_; |
131 | | TNetworkAddress fe_addr_; |
132 | | TQueryType::type query_type_; |
133 | | |
134 | | /* cancel property |
135 | | */ |
136 | | std::atomic<int64_t> cancelled_time_ = 0; |
137 | | |
138 | | /* finish property |
139 | | */ |
140 | | std::atomic<bool> is_finished_ = false; |
141 | | int64_t start_time_ = 0; |
142 | | std::atomic<int64_t> finish_time_ = 0; |
143 | | |
144 | | /* pause property |
145 | | */ |
146 | | AtomicStatus paused_reason_; |
147 | | std::atomic<int64_t> paused_count_ = 0; |
148 | | |
149 | | /* memory status property |
150 | | */ |
151 | | std::atomic<bool> low_memory_mode_ = false; |
152 | | std::atomic<bool> enable_reserve_memory_ = true; |
153 | | |
154 | | /* memory revoke property |
155 | | */ |
156 | | std::atomic<int> revoking_tasks_count_ = 0; |
157 | | }; |
158 | | |
159 | | #include "common/compile_check_end.h" |
160 | | } // namespace doris |