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