Coverage Report

Created: 2026-09-15 16:32

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/root/doris/be/src/runtime/thread_context.cpp
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
#include "runtime/thread_context.h"
19
20
#include <mutex>
21
22
#include "common/signal_handler.h"
23
#include "runtime/query_context.h"
24
#include "runtime/runtime_state.h"
25
26
namespace doris {
27
class MemTracker;
28
29
bthread_key_t btls_key;
30
31
namespace {
32
33
56
void thread_context_deleter(void* data) {
34
56
    delete static_cast<ThreadContext*>(data);
35
56
}
36
37
} // namespace
38
39
2.07M
void init_thread_context_btls_key() {
40
2.07M
    static std::once_flag btls_key_once;
41
2.07M
    std::call_once(btls_key_once,
42
2.07M
                   []() { CHECK_EQ(0, bthread_key_create(&btls_key, thread_context_deleter)); });
43
2.07M
}
44
45
611
void AttachTask::init(const std::shared_ptr<ResourceContext>& rc) {
46
    // Validate the ResourceContext chain before mutating any thread-local
47
    // or signal state. If any link is null we throw immediately, so the
48
    // caller's stack-unwind sees a clean state (no thread-local handle is
49
    // acquired and no signal task id is set). Without these the previous
50
    // code would silently propagate a null mem_tracker into
51
    // ThreadMemTrackerMgr and crash much later inside the allocator.
52
611
    if (UNLIKELY(rc == nullptr)) {
53
0
        throw Exception(
54
0
                Status::FatalError("AttachTask::init: rc is null. signal_query_id={:x}-{:x}",
55
0
                                   signal::query_id_hi, signal::query_id_lo));
56
0
    }
57
611
    if (UNLIKELY(rc->memory_context() == nullptr)) {
58
0
        throw Exception(Status::FatalError(
59
0
                "AttachTask::init: rc->memory_context() is null. signal_query_id={:x}-{:x}",
60
0
                signal::query_id_hi, signal::query_id_lo));
61
0
    }
62
611
    if (UNLIKELY(rc->memory_context()->mem_tracker() == nullptr)) {
63
0
        throw Exception(Status::FatalError(
64
0
                "AttachTask::init: rc->memory_context()->mem_tracker() is null. "
65
0
                "ResourceContext was created but set_mem_tracker has not been called yet "
66
0
                "(likely a half-initialized QueryContext used before _init_query_mem_tracker). "
67
0
                "signal_query_id={:x}-{:x}",
68
0
                signal::query_id_hi, signal::query_id_lo));
69
0
    }
70
611
    if (UNLIKELY(rc->task_controller() == nullptr)) {
71
0
        throw Exception(Status::FatalError(
72
0
                "AttachTask::init: rc->task_controller() is null. signal_query_id={:x}-{:x}",
73
0
                signal::query_id_hi, signal::query_id_lo));
74
0
    }
75
611
    ThreadLocalHandle::create_thread_local_if_not_exits();
76
611
    signal::set_signal_task_id(rc->task_controller()->task_id());
77
611
    thread_context()->attach_task(rc);
78
611
}
79
80
216
AttachTask::AttachTask(const std::shared_ptr<ResourceContext>& rc) {
81
216
    init(rc);
82
216
}
83
84
321
AttachTask::AttachTask(const std::shared_ptr<MemTrackerLimiter>& mem_tracker) {
85
    // if parameter is `orphan_mem_tracker`, if you do not switch thraed mem tracker afterwards,
86
    // alloc or free memory from Allocator will fail DCHECK. unless you know for sure that
87
    // the thread will not alloc or free memory from Allocator later.
88
    //
89
    // Validate before constructing the ResourceContext: MemoryContext::set_mem_tracker()
90
    // immediately dereferences mem_tracker->limit(), so a null shared_ptr would
91
    // crash there before reaching init()'s diagnostics.
92
321
    if (UNLIKELY(mem_tracker == nullptr)) {
93
0
        throw Exception(Status::FatalError(
94
0
                "AttachTask(MemTrackerLimiter): mem_tracker is null. signal_query_id={:x}-{:x}",
95
0
                signal::query_id_hi, signal::query_id_lo));
96
0
    }
97
321
    std::shared_ptr<ResourceContext> rc = ResourceContext::create_shared();
98
321
    rc->memory_context()->set_mem_tracker(mem_tracker);
99
321
    init(rc);
100
321
}
101
102
74
AttachTask::AttachTask(RuntimeState* runtime_state) {
103
    // Walk the chain `runtime_state -> get_query_ctx() -> resource_ctx()`
104
    // step by step so that an unexpected null pinpoints exactly which link
105
    // failed instead of crashing with a generic NPE inside attach_task() or
106
    // even later inside the allocator.
107
74
    if (UNLIKELY(runtime_state == nullptr)) {
108
0
        throw Exception(Status::FatalError(
109
0
                "AttachTask(RuntimeState*): runtime_state is null. signal_query_id={:x}-{:x}",
110
0
                signal::query_id_hi, signal::query_id_lo));
111
0
    }
112
74
    if (UNLIKELY(runtime_state->get_query_ctx() == nullptr)) {
113
0
        throw Exception(Status::FatalError(
114
0
                "AttachTask(RuntimeState*): runtime_state->get_query_ctx() is null. "
115
0
                "signal_query_id={:x}-{:x}",
116
0
                signal::query_id_hi, signal::query_id_lo));
117
0
    }
118
74
    if (UNLIKELY(runtime_state->get_query_ctx()->resource_ctx() == nullptr)) {
119
0
        throw Exception(
120
0
                Status::FatalError("AttachTask(RuntimeState*): query_ctx->resource_ctx() is null. "
121
0
                                   "signal_query_id={:x}-{:x}",
122
0
                                   signal::query_id_hi, signal::query_id_lo));
123
0
    }
124
74
    signal::set_signal_is_nereids(runtime_state->is_nereids());
125
74
    init(runtime_state->get_query_ctx()->resource_ctx());
126
74
}
127
128
0
AttachTask::AttachTask(QueryContext* query_ctx) {
129
0
    if (UNLIKELY(query_ctx == nullptr)) {
130
0
        throw Exception(Status::FatalError(
131
0
                "AttachTask(QueryContext*): query_ctx is null. signal_query_id={:x}-{:x}",
132
0
                signal::query_id_hi, signal::query_id_lo));
133
0
    }
134
0
    init(query_ctx->resource_ctx());
135
0
}
136
137
611
AttachTask::~AttachTask() {
138
611
    signal::set_signal_task_id(TUniqueId());
139
611
    thread_context()->detach_task();
140
611
    ThreadLocalHandle::del_thread_local_if_count_is_zero();
141
611
}
142
143
5
SwitchResourceContext::SwitchResourceContext(const std::shared_ptr<ResourceContext>& rc) {
144
5
    DCHECK(rc != nullptr);
145
    // Validate the chain before mutating any thread-local or signal state,
146
    // symmetric to AttachTask::init(). Throwing after the thread-local
147
    // handle was acquired or the signal task id was set would skip this
148
    // object's destructor (because construction failed) and leak the
149
    // handle / leave a stale signal task id behind.
150
5
    if (UNLIKELY(rc == nullptr)) {
151
0
        throw Exception(
152
0
                Status::FatalError("SwitchResourceContext: rc is null. signal_query_id={:x}-{:x}",
153
0
                                   signal::query_id_hi, signal::query_id_lo));
154
0
    }
155
5
    if (UNLIKELY(rc->memory_context() == nullptr)) {
156
0
        throw Exception(Status::FatalError(
157
0
                "SwitchResourceContext: rc->memory_context() is null. signal_query_id={:x}-{:x}",
158
0
                signal::query_id_hi, signal::query_id_lo));
159
0
    }
160
5
    if (UNLIKELY(rc->memory_context()->mem_tracker() == nullptr)) {
161
0
        throw Exception(Status::FatalError(
162
0
                "SwitchResourceContext: rc->memory_context()->mem_tracker() is null. "
163
0
                "ResourceContext was switched in before _init_query_mem_tracker ran. "
164
0
                "signal_query_id={:x}-{:x}",
165
0
                signal::query_id_hi, signal::query_id_lo));
166
0
    }
167
5
    if (UNLIKELY(rc->task_controller() == nullptr)) {
168
0
        throw Exception(Status::FatalError(
169
0
                "SwitchResourceContext: rc->task_controller() is null. signal_query_id={:x}-{:x}",
170
0
                signal::query_id_hi, signal::query_id_lo));
171
0
    }
172
5
    doris::ThreadLocalHandle::create_thread_local_if_not_exits();
173
5
    DCHECK(thread_context()->is_attach_task());
174
5
    old_resource_ctx_ = thread_context()->resource_ctx();
175
5
    if (rc != old_resource_ctx_) {
176
4
        signal::set_signal_task_id(rc->task_controller()->task_id());
177
4
        thread_context()->resource_ctx_ = rc;
178
4
        thread_context()->thread_mem_tracker_mgr->attach_limiter_tracker(
179
4
                rc->memory_context()->mem_tracker(), rc->workload_group());
180
4
    }
181
5
}
182
183
5
SwitchResourceContext::~SwitchResourceContext() {
184
5
    if (old_resource_ctx_ != thread_context()->resource_ctx()) {
185
4
        DCHECK(old_resource_ctx_ != nullptr);
186
4
        signal::set_signal_task_id(old_resource_ctx_->task_controller()->task_id());
187
4
        thread_context()->resource_ctx_ = old_resource_ctx_;
188
4
        thread_context()->thread_mem_tracker_mgr->detach_limiter_tracker();
189
4
    }
190
5
    doris::ThreadLocalHandle::del_thread_local_if_count_is_zero();
191
5
}
192
193
SwitchThreadMemTrackerLimiter::SwitchThreadMemTrackerLimiter(
194
1.07M
        const std::shared_ptr<doris::MemTrackerLimiter>& mem_tracker) {
195
1.07M
    DCHECK(mem_tracker);
196
    // Third entry point that calls attach_limiter_tracker(). Without this
197
    // null guard a null mem_tracker silently propagates and the next
198
    // allocation on this thread would NPE deep inside the allocator. Throw
199
    // before acquiring the thread-local handle / doing any side effect so
200
    // the destructor (which is noexcept) never runs in a dirty state.
201
1.07M
    if (UNLIKELY(mem_tracker == nullptr)) {
202
0
        throw Exception(Status::FatalError(
203
0
                "SwitchThreadMemTrackerLimiter: mem_tracker is null. signal_query_id={:x}-{:x}",
204
0
                signal::query_id_hi, signal::query_id_lo));
205
0
    }
206
1.07M
    doris::ThreadLocalHandle::create_thread_local_if_not_exits();
207
1.07M
    if (mem_tracker != thread_context()->thread_mem_tracker_mgr->limiter_mem_tracker_sptr()) {
208
992k
        thread_context()->thread_mem_tracker_mgr->attach_limiter_tracker(mem_tracker);
209
992k
        is_switched_ = true;
210
992k
    }
211
1.07M
}
212
213
1.07M
SwitchThreadMemTrackerLimiter::~SwitchThreadMemTrackerLimiter() {
214
1.07M
    if (is_switched_) {
215
992k
        thread_context()->thread_mem_tracker_mgr->detach_limiter_tracker();
216
992k
    }
217
1.07M
    doris::ThreadLocalHandle::del_thread_local_if_count_is_zero();
218
1.07M
}
219
220
96
AddThreadMemTrackerConsumer::AddThreadMemTrackerConsumer(MemTracker* mem_tracker) {
221
96
    ThreadLocalHandle::create_thread_local_if_not_exits();
222
96
    if (mem_tracker) {
223
96
        _need_pop = thread_context()->thread_mem_tracker_mgr->push_consumer_tracker(mem_tracker);
224
96
    }
225
96
}
226
227
AddThreadMemTrackerConsumer::AddThreadMemTrackerConsumer(
228
        const std::shared_ptr<MemTracker>& mem_tracker)
229
76
        : _mem_tracker(mem_tracker) {
230
76
    ThreadLocalHandle::create_thread_local_if_not_exits();
231
76
    if (_mem_tracker) {
232
76
        _need_pop =
233
76
                thread_context()->thread_mem_tracker_mgr->push_consumer_tracker(_mem_tracker.get());
234
76
    }
235
76
}
236
237
172
AddThreadMemTrackerConsumer::~AddThreadMemTrackerConsumer() {
238
172
    if (_need_pop) {
239
171
        thread_context()->thread_mem_tracker_mgr->pop_consumer_tracker();
240
171
    }
241
172
    ThreadLocalHandle::del_thread_local_if_count_is_zero();
242
172
}
243
244
} // namespace doris