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