be/src/service/doris_main.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 <arrow/flight/client.h> |
19 | | #include <arrow/flight/sql/client.h> |
20 | | #include <arrow/scalar.h> |
21 | | #include <arrow/status.h> |
22 | | #include <arrow/table.h> |
23 | | #include <butil/macros.h> |
24 | | // IWYU pragma: no_include <bthread/errno.h> |
25 | | #include <errno.h> // IWYU pragma: keep |
26 | | #include <fcntl.h> |
27 | | #include <fmt/core.h> |
28 | | #if !defined(__SANITIZE_ADDRESS__) && !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && \ |
29 | | !defined(THREAD_SANITIZER) && !defined(USE_JEMALLOC) |
30 | | #include <gperftools/malloc_extension.h> // IWYU pragma: keep |
31 | | #endif |
32 | | #include <libgen.h> |
33 | | #include <setjmp.h> |
34 | | #include <signal.h> |
35 | | #include <stdint.h> |
36 | | #include <stdio.h> |
37 | | #include <stdlib.h> |
38 | | #include <unistd.h> |
39 | | |
40 | | #include <cstring> |
41 | | #include <functional> |
42 | | #include <memory> |
43 | | #include <ostream> |
44 | | #include <string> |
45 | | #include <string_view> |
46 | | #include <tuple> |
47 | | #include <vector> |
48 | | |
49 | | #include "cloud/cloud_backend_service.h" |
50 | | #include "cloud/config.h" |
51 | | #include "common/phdr_cache.h" |
52 | | #include "common/stack_trace.h" |
53 | | #if defined(__ELF__) && !defined(__FreeBSD__) |
54 | | #include "common/symbol_index.h" |
55 | | #endif |
56 | | #include "runtime/memory/mem_tracker_limiter.h" |
57 | | #include "storage/tablet/tablet_schema_cache.h" |
58 | | #include "storage/utils.h" |
59 | | #include "util/concurrency_stats.h" |
60 | | #include "util/jni-util.h" |
61 | | |
62 | | #if defined(LEAK_SANITIZER) |
63 | | #include <sanitizer/lsan_interface.h> |
64 | | #endif |
65 | | |
66 | | #include <curl/curl.h> |
67 | | #include <thrift/TOutput.h> |
68 | | |
69 | | #include "agent/heartbeat_server.h" |
70 | | #include "common/config.h" |
71 | | #include "common/daemon.h" |
72 | | #include "common/logging.h" |
73 | | #include "common/signal_handler.h" |
74 | | #include "common/status.h" |
75 | | #include "io/cache/block_file_cache_factory.h" |
76 | | #include "load/stream_load/stream_load_recorder_manager.h" |
77 | | #include "runtime/exec_env.h" |
78 | | #include "runtime/user_function_cache.h" |
79 | | #include "service/arrow_flight/flight_sql_service.h" |
80 | | #include "service/backend_options.h" |
81 | | #include "service/backend_service.h" |
82 | | #include "service/http_service.h" |
83 | | #include "service/server/be_server_starter_factory.h" |
84 | | #include "storage/options.h" |
85 | | #include "storage/storage_engine.h" |
86 | | #include "udf/python/python_env.h" |
87 | | #include "util/debug_util.h" |
88 | | #include "util/disk_info.h" |
89 | | #include "util/mem_info.h" |
90 | | #include "util/string_util.h" |
91 | | #include "util/thrift_rpc_helper.h" |
92 | | #include "util/thrift_server.h" |
93 | | #include "util/uid_util.h" |
94 | | |
95 | | namespace doris {} // namespace doris |
96 | | |
97 | | static void help(const char*); |
98 | | |
99 | | extern "C" { |
100 | | void __lsan_do_leak_check(); |
101 | | int __llvm_profile_write_file(); |
102 | | } |
103 | | |
104 | | namespace doris { |
105 | | |
106 | 7 | void signal_handler(int signal) { |
107 | 7 | if (signal == SIGINT || signal == SIGTERM) { |
108 | 7 | k_doris_exit = true; |
109 | 7 | } |
110 | 7 | } |
111 | | |
112 | 14 | int install_signal(int signo, void (*handler)(int)) { |
113 | 14 | struct sigaction sa; |
114 | 14 | memset(&sa, 0, sizeof(struct sigaction)); |
115 | 14 | sa.sa_handler = handler; |
116 | 14 | sigemptyset(&sa.sa_mask); |
117 | 14 | auto ret = sigaction(signo, &sa, nullptr); |
118 | 14 | if (ret != 0) { |
119 | 0 | char buf[64]; |
120 | 0 | LOG(ERROR) << "install signal failed, signo=" << signo << ", errno=" << errno |
121 | 0 | << ", errmsg=" << strerror_r(errno, buf, sizeof(buf)); |
122 | 0 | } |
123 | 14 | return ret; |
124 | 14 | } |
125 | | |
126 | 7 | void init_signals() { |
127 | 7 | auto ret = install_signal(SIGINT, signal_handler); |
128 | 7 | if (ret < 0) { |
129 | 0 | exit(-1); |
130 | 0 | } |
131 | 7 | ret = install_signal(SIGTERM, signal_handler); |
132 | 7 | if (ret < 0) { |
133 | 0 | exit(-1); |
134 | 0 | } |
135 | 7 | } |
136 | | |
137 | 8 | static void thrift_output(const char* x) { |
138 | 8 | LOG(WARNING) << "thrift internal message: " << x; |
139 | 8 | } |
140 | | |
141 | | } // namespace doris |
142 | | |
143 | | // These code is referenced from clickhouse |
144 | | // It is used to check the SIMD instructions |
145 | | enum class InstructionFail { |
146 | | NONE = 0, |
147 | | SSE3 = 1, |
148 | | SSSE3 = 2, |
149 | | SSE4_1 = 3, |
150 | | SSE4_2 = 4, |
151 | | POPCNT = 5, |
152 | | AVX = 6, |
153 | | AVX2 = 7, |
154 | | AVX512 = 8, |
155 | | ARM_NEON = 9 |
156 | | }; |
157 | | |
158 | 0 | auto instruction_fail_to_string(InstructionFail fail) { |
159 | 0 | switch (fail) { |
160 | 0 | #define ret(x) return std::make_tuple(STDERR_FILENO, x, ARRAY_SIZE(x) - 1) |
161 | 0 | case InstructionFail::NONE: |
162 | 0 | ret("NONE"); |
163 | 0 | case InstructionFail::SSE3: |
164 | 0 | ret("SSE3"); |
165 | 0 | case InstructionFail::SSSE3: |
166 | 0 | ret("SSSE3"); |
167 | 0 | case InstructionFail::SSE4_1: |
168 | 0 | ret("SSE4.1"); |
169 | 0 | case InstructionFail::SSE4_2: |
170 | 0 | ret("SSE4.2"); |
171 | 0 | case InstructionFail::POPCNT: |
172 | 0 | ret("POPCNT"); |
173 | 0 | case InstructionFail::AVX: |
174 | 0 | ret("AVX"); |
175 | 0 | case InstructionFail::AVX2: |
176 | 0 | ret("AVX2"); |
177 | 0 | case InstructionFail::AVX512: |
178 | 0 | ret("AVX512"); |
179 | 0 | case InstructionFail::ARM_NEON: |
180 | 0 | ret("ARM_NEON"); |
181 | 0 | } |
182 | | |
183 | 0 | LOG(ERROR) << "Unrecognized instruction fail value." << std::endl; |
184 | 0 | exit(-1); |
185 | 0 | } |
186 | | |
187 | | sigjmp_buf jmpbuf; |
188 | | |
189 | 0 | void sig_ill_check_handler(int, siginfo_t*, void*) { |
190 | 0 | siglongjmp(jmpbuf, 1); |
191 | 0 | } |
192 | | |
193 | | /// Check if necessary SSE extensions are available by trying to execute some sse instructions. |
194 | | /// If instruction is unavailable, SIGILL will be sent by kernel. |
195 | 7 | void check_required_instructions_impl(volatile InstructionFail& fail) { |
196 | 7 | #if defined(__SSE3__) |
197 | 7 | fail = InstructionFail::SSE3; |
198 | 7 | __asm__ volatile("addsubpd %%xmm0, %%xmm0" : : : "xmm0"); |
199 | 7 | #endif |
200 | | |
201 | 7 | #if defined(__SSSE3__) |
202 | 7 | fail = InstructionFail::SSSE3; |
203 | 7 | __asm__ volatile("pabsw %%xmm0, %%xmm0" : : : "xmm0"); |
204 | | |
205 | 7 | #endif |
206 | | |
207 | 7 | #if defined(__SSE4_1__) |
208 | 7 | fail = InstructionFail::SSE4_1; |
209 | 7 | __asm__ volatile("pmaxud %%xmm0, %%xmm0" : : : "xmm0"); |
210 | 7 | #endif |
211 | | |
212 | 7 | #if defined(__SSE4_2__) |
213 | 7 | fail = InstructionFail::SSE4_2; |
214 | 7 | __asm__ volatile("pcmpgtq %%xmm0, %%xmm0" : : : "xmm0"); |
215 | 7 | #endif |
216 | | |
217 | | /// Defined by -msse4.2 |
218 | 7 | #if defined(__POPCNT__) |
219 | 7 | fail = InstructionFail::POPCNT; |
220 | 7 | { |
221 | 7 | uint64_t a = 0; |
222 | 7 | uint64_t b = 0; |
223 | 7 | __asm__ volatile("popcnt %1, %0" : "=r"(a) : "r"(b) :); |
224 | 7 | } |
225 | 7 | #endif |
226 | | |
227 | 7 | #if defined(__AVX__) |
228 | 7 | fail = InstructionFail::AVX; |
229 | 7 | __asm__ volatile("vaddpd %%ymm0, %%ymm0, %%ymm0" : : : "ymm0"); |
230 | 7 | #endif |
231 | | |
232 | 7 | #if defined(__AVX2__) |
233 | 7 | fail = InstructionFail::AVX2; |
234 | 7 | __asm__ volatile("vpabsw %%ymm0, %%ymm0" : : : "ymm0"); |
235 | 7 | #endif |
236 | | |
237 | | #if defined(__AVX512__) |
238 | | fail = InstructionFail::AVX512; |
239 | | __asm__ volatile("vpabsw %%zmm0, %%zmm0" : : : "zmm0"); |
240 | | #endif |
241 | | |
242 | | #if defined(__ARM_NEON__) |
243 | | fail = InstructionFail::ARM_NEON; |
244 | | #ifndef __APPLE__ |
245 | | __asm__ volatile("vadd.i32 q8, q8, q8" : : : "q8"); |
246 | | #endif |
247 | | #endif |
248 | | |
249 | 7 | fail = InstructionFail::NONE; |
250 | 7 | } |
251 | | |
252 | 0 | bool write_retry(int fd, const char* data, size_t size) { |
253 | 0 | if (!size) size = strlen(data); |
254 | |
|
255 | 0 | while (size != 0) { |
256 | 0 | ssize_t res = ::write(fd, data, size); |
257 | |
|
258 | 0 | if ((-1 == res || 0 == res) && errno != EINTR) return false; |
259 | | |
260 | 0 | if (res > 0) { |
261 | 0 | data += res; |
262 | 0 | size -= res; |
263 | 0 | } |
264 | 0 | } |
265 | | |
266 | 0 | return true; |
267 | 0 | } |
268 | | |
269 | | /// Macros to avoid using strlen(), since it may fail if SSE is not supported. |
270 | | #define WRITE_ERROR(data) \ |
271 | 0 | do { \ |
272 | 0 | static_assert(__builtin_constant_p(data)); \ |
273 | 0 | if (!write_retry(STDERR_FILENO, data, ARRAY_SIZE(data) - 1)) _Exit(1); \ |
274 | 0 | } while (false) |
275 | | |
276 | | /// Check SSE and others instructions availability. Calls exit on fail. |
277 | | /// This function must be called as early as possible, even before main, because static initializers may use unavailable instructions. |
278 | 7 | void check_required_instructions() { |
279 | 7 | struct sigaction sa {}; |
280 | 7 | struct sigaction sa_old {}; |
281 | 7 | sa.sa_sigaction = sig_ill_check_handler; |
282 | 7 | sa.sa_flags = SA_SIGINFO; |
283 | 7 | auto signal = SIGILL; |
284 | 7 | if (sigemptyset(&sa.sa_mask) != 0 || sigaddset(&sa.sa_mask, signal) != 0 || |
285 | 7 | sigaction(signal, &sa, &sa_old) != 0) { |
286 | | /// You may wonder about strlen. |
287 | | /// Typical implementation of strlen is using SSE4.2 or AVX2. |
288 | | /// But this is not the case because it's compiler builtin and is executed at compile time. |
289 | |
|
290 | 0 | WRITE_ERROR("Can not set signal handler\n"); |
291 | 0 | _Exit(1); |
292 | 0 | } |
293 | | |
294 | 7 | volatile InstructionFail fail = InstructionFail::NONE; |
295 | | |
296 | 7 | if (sigsetjmp(jmpbuf, 1)) { |
297 | 0 | WRITE_ERROR("Instruction check fail. The CPU does not support "); |
298 | 0 | if (!std::apply(write_retry, instruction_fail_to_string(fail))) _Exit(1); |
299 | 0 | WRITE_ERROR(" instruction set.\n"); |
300 | 0 | WRITE_ERROR( |
301 | 0 | "For example, if your CPU does not support AVX2, you need to rebuild the Doris BE " |
302 | 0 | "with: USE_AVX2=0 sh build.sh --be"); |
303 | 0 | _Exit(1); |
304 | 0 | } |
305 | | |
306 | 7 | check_required_instructions_impl(fail); |
307 | | |
308 | 7 | if (sigaction(signal, &sa_old, nullptr)) { |
309 | 0 | WRITE_ERROR("Can not set signal handler\n"); |
310 | 0 | _Exit(1); |
311 | 0 | } |
312 | 7 | } |
313 | | |
314 | | struct Checker { |
315 | 7 | Checker() { check_required_instructions(); } |
316 | | } checker |
317 | | #ifndef __APPLE__ |
318 | | __attribute__((init_priority(101))) /// Run before other static initializers. |
319 | | #endif |
320 | | ; |
321 | | |
322 | | // A startup failure that happens after ExecEnv::init() has run must terminate the |
323 | | // process the same way normal shutdown does (see the _exit(0) at the end of main): |
324 | | // in the default mode we _exit() immediately, skipping global destructors and the |
325 | | // LeakSanitizer atexit check. Init-time singletons (e.g. the internal workload |
326 | | // group's task scheduler) intentionally live for the whole process lifetime, so |
327 | | // running the leak check on this abnormal-exit path reports them as false-positive |
328 | | // leaks. enable_graceful_exit_check is honored so memleak-check mode still runs LSAN. |
329 | 0 | [[noreturn]] static void exit_on_startup_failure() { |
330 | 0 | google::FlushLogFiles(google::GLOG_INFO); |
331 | 0 | if (!doris::config::enable_graceful_exit_check) { |
332 | 0 | _exit(1); |
333 | 0 | } |
334 | 0 | exit(1); |
335 | 0 | } |
336 | | |
337 | 6 | int main(int argc, char** argv) { |
338 | 6 | doris::signal::InstallFailureSignalHandler(); |
339 | | // create StackTraceCache Instance, at the beginning, other static destructors may use. |
340 | 6 | StackTrace::createCache(); |
341 | | // extern doris::ErrorCode::ErrorCodeInitializer error_code_init; |
342 | | // Some developers will modify status.h and we use a very ticky logic to init error_states |
343 | | // and it maybe not inited. So add a check here. |
344 | 6 | doris::ErrorCode::error_code_init.check_init(); |
345 | | // check if print version or help |
346 | 6 | if (argc > 1) { |
347 | 0 | if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-v") == 0) { |
348 | 0 | puts(doris::get_build_version(false).c_str()); |
349 | 0 | exit(0); |
350 | 0 | } else if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0) { |
351 | 0 | help(basename(argv[0])); |
352 | 0 | exit(0); |
353 | 0 | } |
354 | 0 | } |
355 | | |
356 | 6 | if (getenv("DORIS_HOME") == nullptr) { |
357 | 0 | fprintf(stderr, "you need set DORIS_HOME environment variable.\n"); |
358 | 0 | exit(-1); |
359 | 0 | } |
360 | 6 | if (getenv("PID_DIR") == nullptr) { |
361 | 0 | fprintf(stderr, "you need set PID_DIR environment variable.\n"); |
362 | 0 | exit(-1); |
363 | 0 | } |
364 | | |
365 | 6 | SCOPED_INIT_THREAD_CONTEXT(); |
366 | | |
367 | 6 | using doris::Status; |
368 | 6 | using std::string; |
369 | | |
370 | | // open pid file, obtain file lock and save pid |
371 | 6 | string pid_file = string(getenv("PID_DIR")) + "/be.pid"; |
372 | 6 | int fd = open(pid_file.c_str(), O_RDWR | O_CREAT, |
373 | 6 | S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH); |
374 | 6 | if (fd < 0) { |
375 | 0 | fprintf(stderr, "fail to create pid file."); |
376 | 0 | exit(-1); |
377 | 0 | } |
378 | | |
379 | 6 | string pid = std::to_string((long)getpid()); |
380 | 6 | pid += "\n"; |
381 | 6 | size_t length = write(fd, pid.c_str(), pid.size()); |
382 | 6 | if (length != pid.size()) { |
383 | 0 | fprintf(stderr, "fail to save pid into pid file."); |
384 | 0 | exit(-1); |
385 | 0 | } |
386 | | |
387 | | // descriptor will be leaked when failing to close fd |
388 | 6 | if (::close(fd) < 0) { |
389 | 0 | fprintf(stderr, "failed to close fd of pidfile."); |
390 | 0 | exit(-1); |
391 | 0 | } |
392 | | |
393 | | // init config. |
394 | | // the config in be_custom.conf will overwrite the config in be.conf |
395 | | // Must init custom config after init config, separately. |
396 | | // Because the path of custom config file is defined in be.conf |
397 | 6 | string conffile = string(getenv("DORIS_HOME")) + "/conf/be.conf"; |
398 | 6 | if (!doris::config::init(conffile.c_str(), true, true, true)) { |
399 | 0 | fprintf(stderr, "error read config file. \n"); |
400 | 0 | return -1; |
401 | 0 | } |
402 | | |
403 | 6 | string custom_conffile = doris::config::custom_config_dir + "/be_custom.conf"; |
404 | 6 | if (!doris::config::init(custom_conffile.c_str(), true, false, false)) { |
405 | 0 | fprintf(stderr, "error read custom config file. \n"); |
406 | 0 | return -1; |
407 | 0 | } |
408 | | |
409 | | // ATTN: Callers that want to override default gflags variables should do so before calling this method |
410 | 6 | google::ParseCommandLineFlags(&argc, &argv, true); |
411 | | // ATTN: MUST init before LOG |
412 | 6 | doris::init_glog("be"); |
413 | | |
414 | 6 | LOG(INFO) << doris::get_version_string(false); |
415 | | |
416 | 6 | doris::init_thrift_logging(); |
417 | | |
418 | 6 | if (doris::config::enable_fuzzy_mode) { |
419 | 6 | Status status = doris::config::set_fuzzy_configs(); |
420 | 6 | if (!status.ok()) { |
421 | 0 | LOG(WARNING) << "Failed to initialize fuzzy config: " << status; |
422 | 0 | exit(1); |
423 | 0 | } |
424 | 6 | } |
425 | | |
426 | | #if !defined(__SANITIZE_ADDRESS__) && !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && \ |
427 | | !defined(THREAD_SANITIZER) && !defined(USE_JEMALLOC) |
428 | | // Change the total TCMalloc thread cache size if necessary. |
429 | | const size_t kDefaultTotalThreadCacheBytes = 1024 * 1024 * 1024; |
430 | | if (!MallocExtension::instance()->SetNumericProperty("tcmalloc.max_total_thread_cache_bytes", |
431 | | kDefaultTotalThreadCacheBytes)) { |
432 | | fprintf(stderr, "Failed to change TCMalloc total thread cache size.\n"); |
433 | | return -1; |
434 | | } |
435 | | #endif |
436 | | |
437 | 6 | std::vector<doris::StorePath> paths; |
438 | 6 | auto olap_res = doris::parse_conf_store_paths(doris::config::storage_root_path, &paths); |
439 | 6 | if (!olap_res) { |
440 | 0 | LOG(ERROR) << "parse config storage path failed, path=" << doris::config::storage_root_path; |
441 | 0 | exit(-1); |
442 | 0 | } |
443 | | |
444 | 6 | std::vector<doris::StorePath> spill_paths; |
445 | 6 | if (doris::config::spill_storage_root_path.empty()) { |
446 | 6 | doris::config::spill_storage_root_path = doris::config::storage_root_path; |
447 | 6 | } |
448 | 6 | olap_res = doris::parse_conf_store_paths(doris::config::spill_storage_root_path, &spill_paths); |
449 | 6 | if (!olap_res) { |
450 | 0 | LOG(ERROR) << "parse config spill storage path failed, path=" |
451 | 0 | << doris::config::spill_storage_root_path; |
452 | 0 | exit(-1); |
453 | 0 | } |
454 | 6 | std::set<std::string> broken_paths; |
455 | 6 | doris::parse_conf_broken_store_paths(doris::config::broken_storage_path, &broken_paths); |
456 | | |
457 | 6 | auto it = paths.begin(); |
458 | 16 | for (; it != paths.end();) { |
459 | 10 | if (broken_paths.count(it->path) > 0) { |
460 | 0 | if (doris::config::ignore_broken_disk) { |
461 | 0 | LOG(WARNING) << "ignore broken disk, path = " << it->path; |
462 | 0 | it = paths.erase(it); |
463 | 0 | } else { |
464 | 0 | LOG(ERROR) << "a broken disk is found " << it->path; |
465 | 0 | exit(-1); |
466 | 0 | } |
467 | 10 | } else if (!doris::check_datapath_rw(it->path)) { |
468 | 0 | if (doris::config::ignore_broken_disk) { |
469 | 0 | LOG(WARNING) << "read write test file failed, path=" << it->path; |
470 | 0 | it = paths.erase(it); |
471 | 0 | } else { |
472 | 0 | LOG(ERROR) << "read write test file failed, path=" << it->path; |
473 | | // if only one disk and the disk is full, also need exit because rocksdb will open failed |
474 | 0 | exit(-1); |
475 | 0 | } |
476 | 10 | } else { |
477 | 10 | ++it; |
478 | 10 | } |
479 | 10 | } |
480 | | |
481 | 6 | if (paths.empty()) { |
482 | 0 | LOG(ERROR) << "All disks are broken, exit."; |
483 | 0 | exit(-1); |
484 | 0 | } |
485 | | |
486 | 6 | it = spill_paths.begin(); |
487 | 16 | for (; it != spill_paths.end();) { |
488 | 10 | if (!doris::check_datapath_rw(it->path)) { |
489 | 0 | if (doris::config::ignore_broken_disk) { |
490 | 0 | LOG(WARNING) << "read write test file failed, path=" << it->path; |
491 | 0 | it = spill_paths.erase(it); |
492 | 0 | } else { |
493 | 0 | LOG(ERROR) << "read write test file failed, path=" << it->path; |
494 | 0 | exit(-1); |
495 | 0 | } |
496 | 10 | } else { |
497 | 10 | ++it; |
498 | 10 | } |
499 | 10 | } |
500 | 6 | if (spill_paths.empty()) { |
501 | 0 | LOG(ERROR) << "All spill disks are broken, exit."; |
502 | 0 | exit(-1); |
503 | 0 | } |
504 | | |
505 | | // initialize libcurl here to avoid concurrent initialization |
506 | 6 | auto curl_ret = curl_global_init(CURL_GLOBAL_ALL); |
507 | 6 | if (curl_ret != 0) { |
508 | 0 | LOG(ERROR) << "fail to initialize libcurl, curl_ret=" << curl_ret; |
509 | 0 | exit(-1); |
510 | 0 | } |
511 | | // add logger for thrift internal |
512 | 6 | apache::thrift::GlobalOutput.setOutputFunction(doris::thrift_output); |
513 | | |
514 | 6 | Status status = Status::OK(); |
515 | 6 | if (doris::config::enable_java_support) { |
516 | | // Init jni |
517 | 6 | status = doris::Jni::Util::Init(); |
518 | 6 | if (!status.ok()) { |
519 | 0 | LOG(WARNING) << "Failed to initialize JNI: " << status; |
520 | 0 | exit(1); |
521 | 6 | } else { |
522 | 6 | LOG(INFO) << "Doris backend JNI is initialized."; |
523 | 6 | } |
524 | 6 | } |
525 | | |
526 | 6 | if (doris::config::enable_python_udf_support) { |
527 | 6 | if (std::string python_udf_root_path = |
528 | 6 | fmt::format("{}/lib/udf/python", std::getenv("DORIS_HOME")); |
529 | 6 | !std::filesystem::exists(python_udf_root_path)) { |
530 | 0 | std::filesystem::create_directories(python_udf_root_path); |
531 | 0 | } |
532 | | |
533 | | // Normalize and trim all Python-related config parameters |
534 | 6 | std::string python_env_mode = |
535 | 6 | std::string(doris::trim(doris::to_lower(doris::config::python_env_mode))); |
536 | 6 | std::string python_conda_root_path = |
537 | 6 | std::string(doris::trim(doris::config::python_conda_root_path)); |
538 | 6 | std::string python_venv_root_path = |
539 | 6 | std::string(doris::trim(doris::config::python_venv_root_path)); |
540 | 6 | std::string python_venv_interpreter_paths = |
541 | 6 | std::string(doris::trim(doris::config::python_venv_interpreter_paths)); |
542 | | |
543 | 6 | if (python_env_mode == "conda") { |
544 | 0 | if (python_conda_root_path.empty()) { |
545 | 0 | LOG(ERROR) |
546 | 0 | << "Python conda root path is empty, please set `python_conda_root_path` " |
547 | 0 | "or set `enable_python_udf_support` to `false`"; |
548 | 0 | exit(1); |
549 | 0 | } |
550 | 0 | LOG(INFO) << "Doris backend python version manager is initialized. Python conda " |
551 | 0 | "root path: " |
552 | 0 | << python_conda_root_path; |
553 | 0 | status = doris::PythonVersionManager::instance().init(doris::PythonEnvType::CONDA, |
554 | 0 | python_conda_root_path, ""); |
555 | 6 | } else if (python_env_mode == "venv") { |
556 | 6 | if (python_venv_root_path.empty()) { |
557 | 0 | LOG(ERROR) |
558 | 0 | << "Python venv root path is empty, please set `python_venv_root_path` or " |
559 | 0 | "set `enable_python_udf_support` to `false`"; |
560 | 0 | exit(1); |
561 | 0 | } |
562 | 6 | if (python_venv_interpreter_paths.empty()) { |
563 | 0 | LOG(ERROR) |
564 | 0 | << "Python interpreter paths is empty, please set " |
565 | 0 | "`python_venv_interpreter_paths` or set `enable_python_udf_support` to " |
566 | 0 | "`false`"; |
567 | 0 | exit(1); |
568 | 0 | } |
569 | 6 | LOG(INFO) << "Doris backend python version manager is initialized. Python venv " |
570 | 6 | "root path: " |
571 | 6 | << python_venv_root_path |
572 | 6 | << ", python interpreter paths: " << python_venv_interpreter_paths; |
573 | 6 | status = doris::PythonVersionManager::instance().init(doris::PythonEnvType::VENV, |
574 | 6 | python_venv_root_path, |
575 | 6 | python_venv_interpreter_paths); |
576 | 6 | } else { |
577 | 0 | status = Status::InvalidArgument( |
578 | 0 | "Python env mode is invalid, should be `conda` or `venv`. If you don't want to " |
579 | 0 | "enable the Python UDF function, please set `enable_python_udf_support` to " |
580 | 0 | "`false`"); |
581 | 0 | } |
582 | | |
583 | 6 | if (!status.ok()) { |
584 | 0 | LOG(ERROR) << "Failed to initialize python version manager: " << status; |
585 | 0 | exit(1); |
586 | 0 | } |
587 | 6 | LOG(INFO) << doris::PythonVersionManager::instance().to_string(); |
588 | 6 | } |
589 | | |
590 | | // Doris own signal handler must be register after jvm is init. |
591 | | // Or our own sig-handler for SIGINT & SIGTERM will not be chained ... |
592 | | // https://www.oracle.com/java/technologies/javase/signals.html |
593 | 6 | doris::init_signals(); |
594 | | // ATTN: MUST init before `ExecEnv`, `StorageEngine` and other daemon services |
595 | | // |
596 | | // Daemon ───┬──► StorageEngine ──► ExecEnv ──► Disk/Mem/CpuInfo |
597 | | // │ |
598 | | // │ |
599 | | // BackendService ─┘ |
600 | 6 | doris::CpuInfo::init(); |
601 | 6 | doris::DiskInfo::init(); |
602 | 6 | doris::MemInfo::init(); |
603 | | |
604 | 6 | LOG(INFO) << doris::CpuInfo::debug_string(); |
605 | 6 | LOG(INFO) << doris::DiskInfo::debug_string(); |
606 | 6 | LOG(INFO) << doris::MemInfo::debug_string(); |
607 | | |
608 | | // Doris-patched GNU libunwind reads PHDR metadata from our lock-free snapshot instead of |
609 | | // entering glibc dl_iterate_phdr while jemalloc profiling or signal-context unwinding may |
610 | | // already be involved in loader-lock-sensitive code. Configure libunwind before daemon threads |
611 | | // start so all later heap-profile and stack-trace unwinds use the same lock-safe policy. |
612 | 6 | configureLibunwindPHDRCache(); |
613 | 6 | updatePHDRCache(); |
614 | 6 | LOG(INFO) << "PHDR cache enabled: " << hasPHDRCache(); |
615 | 6 | #if defined(__ELF__) && !defined(__FreeBSD__) |
616 | 6 | auto symbol_index = doris::SymbolIndex::instance(); |
617 | 6 | LOG(INFO) << "SymbolIndex preloaded: objects=" << symbol_index->objects().size() |
618 | 6 | << " symbols=" << symbol_index->symbols().size(); |
619 | 6 | #endif |
620 | 6 | if (!doris::BackendOptions::init()) { |
621 | 0 | exit(-1); |
622 | 0 | } |
623 | | |
624 | | // init exec env |
625 | 6 | auto* exec_env(doris::ExecEnv::GetInstance()); |
626 | 6 | status = doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths, spill_paths, broken_paths); |
627 | 6 | if (status != Status::OK()) { |
628 | 0 | std::cerr << "failed to init doris storage engine, res=" << status; |
629 | 0 | exit_on_startup_failure(); |
630 | 0 | } |
631 | | |
632 | | // Start concurrency stats manager |
633 | 6 | doris::ConcurrencyStatsManager::instance().start(); |
634 | | |
635 | | // begin to start services |
636 | 6 | doris::ThriftRpcHelper::setup(exec_env); |
637 | | // 1. thrift server with be_port |
638 | 6 | std::shared_ptr<doris::BaseBackendService> service; |
639 | 6 | std::function<void(Status&, std::string_view)> stop_work_if_error = [&](Status& status, |
640 | 42 | std::string_view msg) { |
641 | 42 | if (!status.ok()) { |
642 | 0 | std::cerr << msg << '\n'; |
643 | 0 | service->stop_works(); |
644 | 0 | exit_on_startup_failure(); |
645 | 0 | } |
646 | 42 | }; |
647 | | |
648 | 6 | if (doris::config::is_cloud_mode()) { |
649 | 0 | service = std::make_shared<doris::CloudBackendService>( |
650 | 0 | exec_env->storage_engine().to_cloud(), exec_env); |
651 | 6 | } else { |
652 | 6 | service = std::make_shared<doris::BackendService>(exec_env->storage_engine().to_local(), |
653 | 6 | exec_env); |
654 | 6 | } |
655 | | |
656 | 6 | std::unique_ptr<doris::server::IServerStarter> backend_thrift_starter; |
657 | 6 | EXIT_IF_ERROR(doris::server::create_backend_thrift_starter(exec_env, doris::config::be_port, |
658 | 6 | service, &backend_thrift_starter)); |
659 | 6 | status = backend_thrift_starter->start(); |
660 | 6 | stop_work_if_error(status, "Doris BE server did not start correctly, exiting"); |
661 | | |
662 | | // 2. brpc service |
663 | 6 | std::unique_ptr<doris::server::IServerStarter> brpc_starter; |
664 | 6 | EXIT_IF_ERROR(doris::server::create_brpc_starter( |
665 | 6 | exec_env, doris::config::brpc_port, doris::config::brpc_num_threads, &brpc_starter)); |
666 | 6 | status = brpc_starter->start(); |
667 | 6 | stop_work_if_error(status, "BRPC service did not start correctly, exiting"); |
668 | | |
669 | | // 3. http service |
670 | 6 | std::unique_ptr<doris::server::IServerStarter> http_starter; |
671 | 6 | EXIT_IF_ERROR(doris::server::create_http_starter(exec_env, doris::config::webserver_port, |
672 | 6 | doris::config::webserver_num_workers, |
673 | 6 | &http_starter)); |
674 | 6 | status = http_starter->start(); |
675 | 6 | stop_work_if_error(status, "Doris Be http service did not start correctly, exiting"); |
676 | | |
677 | | // 4. heart beat server |
678 | 6 | doris::ClusterInfo* cluster_info = exec_env->cluster_info(); |
679 | 6 | std::unique_ptr<doris::server::IServerStarter> heartbeat_thrift_starter; |
680 | 6 | status = doris::server::create_heartbeat_thrift_starter( |
681 | 6 | exec_env, doris::config::heartbeat_service_port, |
682 | 6 | doris::config::heartbeat_service_thread_count, cluster_info, &heartbeat_thrift_starter); |
683 | 6 | stop_work_if_error(status, "Heartbeat services did not start correctly, exiting"); |
684 | | |
685 | 6 | status = heartbeat_thrift_starter->start(); |
686 | 6 | stop_work_if_error(status, "Doris BE HeartBeat Service did not start correctly, exiting: " + |
687 | 6 | status.to_string()); |
688 | | |
689 | | // 5. arrow flight service |
690 | 6 | std::unique_ptr<doris::server::IServerStarter> flight_starter; |
691 | 6 | EXIT_IF_ERROR(doris::server::create_flight_starter(doris::config::arrow_flight_sql_port, |
692 | 6 | &flight_starter)); |
693 | 6 | status = flight_starter->start(); |
694 | 6 | stop_work_if_error( |
695 | 6 | status, "Arrow Flight Service did not start correctly, exiting, " + status.to_string()); |
696 | | |
697 | | // 6. start daemon thread to do clean or gc jobs |
698 | 6 | doris::Daemon daemon; |
699 | 6 | daemon.start(); |
700 | | |
701 | 6 | exec_env->storage_engine().notify_listeners(); |
702 | | |
703 | 6 | doris::k_is_server_ready = true; |
704 | | |
705 | 562 | while (!doris::k_doris_exit) { |
706 | | #if defined(LEAK_SANITIZER) |
707 | | __lsan_do_leak_check(); |
708 | | #endif |
709 | 556 | sleep(3); |
710 | 556 | } |
711 | 6 | doris::k_is_server_ready = false; |
712 | 6 | LOG(INFO) << "Doris main exiting."; |
713 | 6 | #if defined(LLVM_PROFILE) |
714 | 6 | __llvm_profile_write_file(); |
715 | 6 | LOG(INFO) << "Flush profile file."; |
716 | 6 | #endif |
717 | | // For graceful shutdown, need to wait for all running queries to stop |
718 | 6 | exec_env->wait_for_all_tasks_done(); |
719 | | |
720 | 6 | if (!doris::config::enable_graceful_exit_check) { |
721 | | // If not in memleak check mode, no need to wait all objects de-constructed normally, just exit. |
722 | | // It will make sure that graceful shutdown can be done definitely. |
723 | 0 | LOG(INFO) << "Doris main exited."; |
724 | 0 | google::FlushLogFiles(google::GLOG_INFO); |
725 | 0 | _exit(0); // Do not call exit(0), it will wait for all objects de-constructed normally |
726 | 0 | return 0; |
727 | 0 | } |
728 | 6 | daemon.stop(); |
729 | 6 | flight_starter->stop(); |
730 | 6 | flight_starter->join(); |
731 | 6 | LOG(INFO) << "Flight server stopped."; |
732 | 6 | heartbeat_thrift_starter->stop(); |
733 | 6 | heartbeat_thrift_starter->join(); |
734 | 6 | LOG(INFO) << "Heartbeat server stopped"; |
735 | | // The stream load recorder manager writes its audit records through this BE's own http |
736 | | // service, so it has to be stopped while that service is still up. Otherwise an audit |
737 | | // load that is in flight here can never be answered, and it blocks the join() done by |
738 | | // SAFE_STOP(_stream_load_recorder_manager) in ExecEnv::destroy() for up to the stream |
739 | | // load timeout, which is longer than the grace period of stop_be.sh --grace. |
740 | 6 | if (auto* recorder_manager = exec_env->stream_load_recorder_manager(); |
741 | 6 | recorder_manager != nullptr) { |
742 | 3 | recorder_manager->stop(); |
743 | 3 | } |
744 | | // TODO(zhiqiang): http_service |
745 | 6 | http_starter->stop(); |
746 | 6 | http_starter->join(); |
747 | 6 | LOG(INFO) << "Http service stopped"; |
748 | 6 | backend_thrift_starter->stop(); |
749 | 6 | backend_thrift_starter->join(); |
750 | 6 | LOG(INFO) << "Be server stopped"; |
751 | 6 | brpc_starter->stop(); |
752 | 6 | brpc_starter->join(); |
753 | 6 | LOG(INFO) << "Brpc service stopped"; |
754 | 6 | service.reset(); |
755 | 6 | LOG(INFO) << "Backend Service stopped"; |
756 | 6 | exec_env->destroy(); |
757 | 6 | LOG(INFO) << "All service stopped, doris main exited."; |
758 | 6 | return 0; |
759 | 6 | } |
760 | | |
761 | 0 | static void help(const char* progname) { |
762 | 0 | printf("%s is the Doris backend server.\n\n", progname); |
763 | 0 | printf("Usage:\n %s [OPTION]...\n\n", progname); |
764 | 0 | printf("Options:\n"); |
765 | 0 | printf(" -v, --version output version information, then exit\n"); |
766 | 0 | printf(" -?, --help show this help, then exit\n"); |
767 | 0 | } |