Coverage Report

Created: 2026-10-14 10:12

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
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 <chrono>
41
#include <cstring>
42
#include <functional>
43
#include <memory>
44
#include <ostream>
45
#include <string>
46
#include <string_view>
47
#include <thread>
48
#include <tuple>
49
#include <vector>
50
51
#include "cloud/cloud_backend_service.h"
52
#include "cloud/config.h"
53
#include "common/phdr_cache.h"
54
#include "common/stack_trace.h"
55
#if defined(__ELF__) && !defined(__FreeBSD__)
56
#include "common/symbol_index.h"
57
#endif
58
#include "runtime/memory/mem_tracker_limiter.h"
59
#include "storage/tablet/tablet_schema_cache.h"
60
#include "storage/utils.h"
61
#include "util/concurrency_stats.h"
62
63
#if defined(LEAK_SANITIZER)
64
#include <sanitizer/lsan_interface.h>
65
#endif
66
67
#include <curl/curl.h>
68
#include <thrift/TOutput.h>
69
70
#include "agent/heartbeat_server.h"
71
#include "common/config.h"
72
#include "common/daemon.h"
73
#include "common/logging.h"
74
#include "common/signal_handler.h"
75
#include "common/status.h"
76
#include "io/cache/block_file_cache_factory.h"
77
#include "load/stream_load/stream_load_recorder_manager.h"
78
#include "runtime/exec_env.h"
79
#include "runtime/user_function_cache.h"
80
#include "service/arrow_flight/flight_sql_service.h"
81
#include "service/backend_options.h"
82
#include "service/backend_service.h"
83
#include "service/http_service.h"
84
#include "service/server/be_server_starter_factory.h"
85
#include "storage/disk_health_check_watchdog.h"
86
#include "storage/options.h"
87
#include "storage/storage_engine.h"
88
#include "udf/python/python_env.h"
89
#include "util/debug_util.h"
90
#include "util/disk_info.h"
91
#include "util/jni_plugin_registry.h"
92
#include "util/mem_info.h"
93
#include "util/string_util.h"
94
#include "util/thread.h"
95
#include "util/thrift_rpc_helper.h"
96
#include "util/thrift_server.h"
97
#include "util/uid_util.h"
98
99
namespace doris {} // namespace doris
100
101
static void help(const char*);
102
103
extern "C" {
104
void __lsan_do_leak_check();
105
int __llvm_profile_write_file();
106
}
107
108
namespace doris {
109
110
7
void signal_handler(int signal) {
111
7
    if (signal == SIGINT || signal == SIGTERM) {
112
7
        k_doris_exit = true;
113
7
    }
114
    // SIGQUIT deliberately does nothing here; see init_signals().
115
7
}
116
117
21
int install_signal(int signo, void (*handler)(int)) {
118
21
    struct sigaction sa;
119
21
    memset(&sa, 0, sizeof(struct sigaction));
120
21
    sa.sa_handler = handler;
121
    // Restartable syscalls stay restartable. It matters most for SIGQUIT: unlike the two
122
    // shutdown signals, that one is sent to a HEALTHY BE - `kill -3` is the operator's habit
123
    // for asking a running process for a thread dump, and since the handler now produces
124
    // nothing they send it again. Without SA_RESTART each of those turns whatever syscall the
125
    // receiving thread happened to be in into EINTR, in the middle of normal serving.
126
21
    sa.sa_flags = SA_RESTART;
127
21
    sigemptyset(&sa.sa_mask);
128
21
    auto ret = sigaction(signo, &sa, nullptr);
129
21
    if (ret != 0) {
130
0
        char buf[64];
131
0
        LOG(ERROR) << "install signal failed, signo=" << signo << ", errno=" << errno
132
0
                   << ", errmsg=" << strerror_r(errno, buf, sizeof(buf));
133
0
    }
134
21
    return ret;
135
21
}
136
137
7
void init_signals() {
138
7
    auto ret = install_signal(SIGINT, signal_handler);
139
7
    if (ret < 0) {
140
0
        exit(-1);
141
0
    }
142
7
    ret = install_signal(SIGTERM, signal_handler);
143
7
    if (ret < 0) {
144
0
        exit(-1);
145
0
    }
146
    // SIGQUIT is taken over even though the BE does nothing with it, because its default
147
    // action is not "nothing": it terminates the process and dumps core. `kill -3 <pid>` is
148
    // what an operator reaches for to get a thread dump out of a process that looks stuck,
149
    // and until the JVM started running with -Xrs it got one - the JVM installed a handler
150
    // for this signal along with the shutdown ones. -Xrs stops it from doing that (see
151
    // JvmLauncher::_build_options), which would leave SIGQUIT at SIG_DFL and turn that
152
    // habitual command into a crash. Handling it and ignoring it is the pre-change
153
    // behaviour minus the thread dump; jcmd and jstack, which attach rather than signal,
154
    // are how to get one now.
155
    //
156
    // Installed with a handler rather than SIG_IGN so that the disposition is inheritable
157
    // by nothing and visible to JvmLauncher's BeOwnedSignalGuard, which saves and restores
158
    // it around any VM creation it does not control.
159
7
    ret = install_signal(SIGQUIT, signal_handler);
160
7
    if (ret < 0) {
161
0
        exit(-1);
162
0
    }
163
7
}
164
165
8
static void thrift_output(const char* x) {
166
8
    LOG(WARNING) << "thrift internal message: " << x;
167
8
}
168
169
} // namespace doris
170
171
// These code is referenced from clickhouse
172
// It is used to check the SIMD instructions
173
enum class InstructionFail {
174
    NONE = 0,
175
    SSE3 = 1,
176
    SSSE3 = 2,
177
    SSE4_1 = 3,
178
    SSE4_2 = 4,
179
    POPCNT = 5,
180
    AVX = 6,
181
    AVX2 = 7,
182
    AVX512 = 8,
183
    ARM_NEON = 9
184
};
185
186
0
auto instruction_fail_to_string(InstructionFail fail) {
187
0
    switch (fail) {
188
0
#define ret(x) return std::make_tuple(STDERR_FILENO, x, ARRAY_SIZE(x) - 1)
189
0
    case InstructionFail::NONE:
190
0
        ret("NONE");
191
0
    case InstructionFail::SSE3:
192
0
        ret("SSE3");
193
0
    case InstructionFail::SSSE3:
194
0
        ret("SSSE3");
195
0
    case InstructionFail::SSE4_1:
196
0
        ret("SSE4.1");
197
0
    case InstructionFail::SSE4_2:
198
0
        ret("SSE4.2");
199
0
    case InstructionFail::POPCNT:
200
0
        ret("POPCNT");
201
0
    case InstructionFail::AVX:
202
0
        ret("AVX");
203
0
    case InstructionFail::AVX2:
204
0
        ret("AVX2");
205
0
    case InstructionFail::AVX512:
206
0
        ret("AVX512");
207
0
    case InstructionFail::ARM_NEON:
208
0
        ret("ARM_NEON");
209
0
    }
210
211
0
    LOG(ERROR) << "Unrecognized instruction fail value." << std::endl;
212
0
    exit(-1);
213
0
}
214
215
sigjmp_buf jmpbuf;
216
217
0
void sig_ill_check_handler(int, siginfo_t*, void*) {
218
0
    siglongjmp(jmpbuf, 1);
219
0
}
220
221
/// Check if necessary SSE extensions are available by trying to execute some sse instructions.
222
/// If instruction is unavailable, SIGILL will be sent by kernel.
223
8
void check_required_instructions_impl(volatile InstructionFail& fail) {
224
8
#if defined(__SSE3__)
225
8
    fail = InstructionFail::SSE3;
226
8
    __asm__ volatile("addsubpd %%xmm0, %%xmm0" : : : "xmm0");
227
8
#endif
228
229
8
#if defined(__SSSE3__)
230
8
    fail = InstructionFail::SSSE3;
231
8
    __asm__ volatile("pabsw %%xmm0, %%xmm0" : : : "xmm0");
232
233
8
#endif
234
235
8
#if defined(__SSE4_1__)
236
8
    fail = InstructionFail::SSE4_1;
237
8
    __asm__ volatile("pmaxud %%xmm0, %%xmm0" : : : "xmm0");
238
8
#endif
239
240
8
#if defined(__SSE4_2__)
241
8
    fail = InstructionFail::SSE4_2;
242
8
    __asm__ volatile("pcmpgtq %%xmm0, %%xmm0" : : : "xmm0");
243
8
#endif
244
245
    /// Defined by -msse4.2
246
8
#if defined(__POPCNT__)
247
8
    fail = InstructionFail::POPCNT;
248
8
    {
249
8
        uint64_t a = 0;
250
8
        uint64_t b = 0;
251
8
        __asm__ volatile("popcnt %1, %0" : "=r"(a) : "r"(b) :);
252
8
    }
253
8
#endif
254
255
8
#if defined(__AVX__)
256
8
    fail = InstructionFail::AVX;
257
8
    __asm__ volatile("vaddpd %%ymm0, %%ymm0, %%ymm0" : : : "ymm0");
258
8
#endif
259
260
8
#if defined(__AVX2__)
261
8
    fail = InstructionFail::AVX2;
262
8
    __asm__ volatile("vpabsw %%ymm0, %%ymm0" : : : "ymm0");
263
8
#endif
264
265
#if defined(__AVX512__)
266
    fail = InstructionFail::AVX512;
267
    __asm__ volatile("vpabsw %%zmm0, %%zmm0" : : : "zmm0");
268
#endif
269
270
#if defined(__ARM_NEON__)
271
    fail = InstructionFail::ARM_NEON;
272
#ifndef __APPLE__
273
    __asm__ volatile("vadd.i32  q8, q8, q8" : : : "q8");
274
#endif
275
#endif
276
277
8
    fail = InstructionFail::NONE;
278
8
}
279
280
0
bool write_retry(int fd, const char* data, size_t size) {
281
0
    if (!size) size = strlen(data);
282
283
0
    while (size != 0) {
284
0
        ssize_t res = ::write(fd, data, size);
285
286
0
        if ((-1 == res || 0 == res) && errno != EINTR) return false;
287
288
0
        if (res > 0) {
289
0
            data += res;
290
0
            size -= res;
291
0
        }
292
0
    }
293
294
0
    return true;
295
0
}
296
297
/// Macros to avoid using strlen(), since it may fail if SSE is not supported.
298
#define WRITE_ERROR(data)                                                      \
299
0
    do {                                                                       \
300
0
        static_assert(__builtin_constant_p(data));                             \
301
0
        if (!write_retry(STDERR_FILENO, data, ARRAY_SIZE(data) - 1)) _Exit(1); \
302
0
    } while (false)
303
304
/// Check SSE and others instructions availability. Calls exit on fail.
305
/// This function must be called as early as possible, even before main, because static initializers may use unavailable instructions.
306
8
void check_required_instructions() {
307
8
    struct sigaction sa {};
308
8
    struct sigaction sa_old {};
309
8
    sa.sa_sigaction = sig_ill_check_handler;
310
8
    sa.sa_flags = SA_SIGINFO;
311
8
    auto signal = SIGILL;
312
8
    if (sigemptyset(&sa.sa_mask) != 0 || sigaddset(&sa.sa_mask, signal) != 0 ||
313
8
        sigaction(signal, &sa, &sa_old) != 0) {
314
        /// You may wonder about strlen.
315
        /// Typical implementation of strlen is using SSE4.2 or AVX2.
316
        /// But this is not the case because it's compiler builtin and is executed at compile time.
317
318
0
        WRITE_ERROR("Can not set signal handler\n");
319
0
        _Exit(1);
320
0
    }
321
322
8
    volatile InstructionFail fail = InstructionFail::NONE;
323
324
8
    if (sigsetjmp(jmpbuf, 1)) {
325
0
        WRITE_ERROR("Instruction check fail. The CPU does not support ");
326
0
        if (!std::apply(write_retry, instruction_fail_to_string(fail))) _Exit(1);
327
0
        WRITE_ERROR(" instruction set.\n");
328
0
        WRITE_ERROR(
329
0
                "For example, if your CPU does not support AVX2, you need to rebuild the Doris BE "
330
0
                "with: USE_AVX2=0 sh build.sh --be");
331
0
        _Exit(1);
332
0
    }
333
334
8
    check_required_instructions_impl(fail);
335
336
8
    if (sigaction(signal, &sa_old, nullptr)) {
337
0
        WRITE_ERROR("Can not set signal handler\n");
338
0
        _Exit(1);
339
0
    }
340
8
}
341
342
struct Checker {
343
8
    Checker() { check_required_instructions(); }
344
} checker
345
#ifndef __APPLE__
346
        __attribute__((init_priority(101))) /// Run before other static initializers.
347
#endif
348
        ;
349
350
// A startup failure that happens after ExecEnv::init() has run must terminate the
351
// process the same way normal shutdown does (see the _exit(0) at the end of main):
352
// in the default mode we _exit() immediately, skipping global destructors and the
353
// LeakSanitizer atexit check. Init-time singletons (e.g. the internal workload
354
// group's task scheduler) intentionally live for the whole process lifetime, so
355
// running the leak check on this abnormal-exit path reports them as false-positive
356
// leaks. enable_graceful_exit_check is honored so memleak-check mode still runs LSAN.
357
0
[[noreturn]] static void exit_on_startup_failure() {
358
0
    if (!doris::config::enable_graceful_exit_check) {
359
0
        google::FlushLogFiles(google::GLOG_INFO);
360
0
        _exit(1);
361
0
    }
362
363
    // exit() starts destroying function-local statics while background threads are still
364
    // running. Tear down ExecEnv first so StorageEngine workers are stopped and joined before
365
    // they can race with those destructors.
366
0
    if (doris::ExecEnv::ready()) {
367
0
        doris::ExecEnv::GetInstance()->destroy();
368
0
    }
369
0
    google::FlushLogFiles(google::GLOG_INFO);
370
0
    exit(1);
371
0
}
372
373
8
int main(int argc, char** argv) {
374
8
    doris::signal::InstallFailureSignalHandler();
375
    // create StackTraceCache Instance, at the beginning, other static destructors may use.
376
8
    StackTrace::createCache();
377
    // extern doris::ErrorCode::ErrorCodeInitializer error_code_init;
378
    // Some developers will modify status.h and we use a very ticky logic to init error_states
379
    // and it maybe not inited. So add a check here.
380
8
    doris::ErrorCode::error_code_init.check_init();
381
    // check if print version or help
382
8
    if (argc > 1) {
383
1
        if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-v") == 0) {
384
1
            puts(doris::get_build_version(false).c_str());
385
1
            exit(0);
386
1
        } else if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0) {
387
0
            help(basename(argv[0]));
388
0
            exit(0);
389
0
        }
390
1
    }
391
392
7
    if (getenv("DORIS_HOME") == nullptr) {
393
0
        fprintf(stderr, "you need set DORIS_HOME environment variable.\n");
394
0
        exit(-1);
395
0
    }
396
7
    if (getenv("PID_DIR") == nullptr) {
397
0
        fprintf(stderr, "you need set PID_DIR environment variable.\n");
398
0
        exit(-1);
399
0
    }
400
401
7
    SCOPED_INIT_THREAD_CONTEXT();
402
403
7
    using doris::Status;
404
7
    using std::string;
405
406
    // open pid file, obtain file lock and save pid
407
7
    string pid_file = string(getenv("PID_DIR")) + "/be.pid";
408
7
    int fd = open(pid_file.c_str(), O_RDWR | O_CREAT,
409
7
                  S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH);
410
7
    if (fd < 0) {
411
0
        fprintf(stderr, "fail to create pid file.");
412
0
        exit(-1);
413
0
    }
414
415
7
    string pid = std::to_string((long)getpid());
416
7
    pid += "\n";
417
7
    size_t length = write(fd, pid.c_str(), pid.size());
418
7
    if (length != pid.size()) {
419
0
        fprintf(stderr, "fail to save pid into pid file.");
420
0
        exit(-1);
421
0
    }
422
423
    // descriptor will be leaked when failing to close fd
424
7
    if (::close(fd) < 0) {
425
0
        fprintf(stderr, "failed to close fd of pidfile.");
426
0
        exit(-1);
427
0
    }
428
429
    // init config.
430
    // the config in be_custom.conf will overwrite the config in be.conf
431
    // Must init custom config after init config, separately.
432
    // Because the path of custom config file is defined in be.conf
433
7
    string conffile = string(getenv("DORIS_HOME")) + "/conf/be.conf";
434
7
    if (!doris::config::init(conffile.c_str(), true, true, true)) {
435
0
        fprintf(stderr, "error read config file. \n");
436
0
        return -1;
437
0
    }
438
439
7
    string custom_conffile = doris::config::custom_config_dir + "/be_custom.conf";
440
7
    if (!doris::config::init(custom_conffile.c_str(), true, false, false)) {
441
0
        fprintf(stderr, "error read custom config file. \n");
442
0
        return -1;
443
0
    }
444
445
7
    doris::init_be_version_metrics();
446
447
    // ATTN: Callers that want to override default gflags variables should do so before calling this method
448
7
    google::ParseCommandLineFlags(&argc, &argv, true);
449
    // ATTN: MUST init before LOG
450
7
    doris::init_glog("be");
451
452
7
    LOG(INFO) << doris::get_version_string(false);
453
454
7
    doris::init_thrift_logging();
455
456
7
    if (doris::config::enable_fuzzy_mode) {
457
7
        Status status = doris::config::set_fuzzy_configs();
458
7
        if (!status.ok()) {
459
0
            LOG(WARNING) << "Failed to initialize fuzzy config: " << status;
460
0
            exit(1);
461
0
        }
462
7
    }
463
464
#if !defined(__SANITIZE_ADDRESS__) && !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && \
465
        !defined(THREAD_SANITIZER) && !defined(USE_JEMALLOC)
466
    // Change the total TCMalloc thread cache size if necessary.
467
    const size_t kDefaultTotalThreadCacheBytes = 1024 * 1024 * 1024;
468
    if (!MallocExtension::instance()->SetNumericProperty("tcmalloc.max_total_thread_cache_bytes",
469
                                                         kDefaultTotalThreadCacheBytes)) {
470
        fprintf(stderr, "Failed to change TCMalloc total thread cache size.\n");
471
        return -1;
472
    }
473
#endif
474
475
7
    std::vector<doris::StorePath> paths;
476
7
    auto olap_res = doris::parse_conf_store_paths(doris::config::storage_root_path, &paths);
477
7
    if (!olap_res) {
478
0
        LOG(ERROR) << "parse config storage path failed, path=" << doris::config::storage_root_path;
479
0
        exit(-1);
480
0
    }
481
482
7
    std::vector<doris::StorePath> spill_paths;
483
7
    if (doris::config::spill_storage_root_path.empty()) {
484
7
        doris::config::spill_storage_root_path = doris::config::storage_root_path;
485
7
    }
486
7
    olap_res = doris::parse_conf_store_paths(doris::config::spill_storage_root_path, &spill_paths);
487
7
    if (!olap_res) {
488
0
        LOG(ERROR) << "parse config spill storage path failed, path="
489
0
                   << doris::config::spill_storage_root_path;
490
0
        exit(-1);
491
0
    }
492
7
    std::set<std::string> broken_paths;
493
7
    doris::parse_conf_broken_store_paths(doris::config::broken_storage_path, &broken_paths);
494
495
7
    doris::DiskHealthCheckWatchdog startup_disk_watchdog;
496
7
    auto watchdog_status = startup_disk_watchdog.start(
497
7
            std::chrono::seconds(doris::config::disk_health_check_timeout_seconds));
498
7
    if (!watchdog_status.ok()) {
499
0
        LOG(ERROR) << "Cannot supervise startup disk checks: " << watchdog_status;
500
0
        return -1;
501
0
    }
502
7
    auto it = paths.begin();
503
18
    for (; it != paths.end();) {
504
11
        doris::DiskHealthCheckWatchdog::ScopedCheck check(startup_disk_watchdog);
505
11
        if (broken_paths.count(it->path) > 0) {
506
0
            if (doris::config::ignore_broken_disk) {
507
0
                LOG(WARNING) << "ignore broken disk, path = " << it->path;
508
0
                it = paths.erase(it);
509
0
            } else {
510
0
                LOG(ERROR) << "a broken disk is found " << it->path;
511
0
                exit(-1);
512
0
            }
513
11
        } else if (!doris::check_datapath_rw(it->path)) {
514
0
            if (doris::config::ignore_broken_disk) {
515
0
                LOG(WARNING) << "read write test file failed, path=" << it->path;
516
0
                it = paths.erase(it);
517
0
            } else {
518
0
                LOG(ERROR) << "read write test file failed, path=" << it->path;
519
                // if only one disk and the disk is full, also need exit because rocksdb will open failed
520
0
                exit(-1);
521
0
            }
522
11
        } else {
523
11
            ++it;
524
11
        }
525
11
    }
526
527
7
    if (paths.empty()) {
528
        // exit() does not unwind the scope: supervise logging and teardown as well.
529
0
        doris::DiskHealthCheckWatchdog::ScopedCheck check(startup_disk_watchdog);
530
0
        LOG(ERROR) << "All disks are broken, exit.";
531
0
        exit(-1);
532
0
    }
533
534
7
    it = spill_paths.begin();
535
18
    for (; it != spill_paths.end();) {
536
11
        doris::DiskHealthCheckWatchdog::ScopedCheck check(startup_disk_watchdog);
537
11
        if (!doris::check_datapath_rw(it->path)) {
538
0
            if (doris::config::ignore_broken_disk) {
539
0
                LOG(WARNING) << "read write test file failed, path=" << it->path;
540
0
                it = spill_paths.erase(it);
541
0
            } else {
542
0
                LOG(ERROR) << "read write test file failed, path=" << it->path;
543
0
                exit(-1);
544
0
            }
545
11
        } else {
546
11
            ++it;
547
11
        }
548
11
    }
549
7
    if (spill_paths.empty()) {
550
        // Keep the deadline armed through logging, atexit handlers and teardown.
551
0
        doris::DiskHealthCheckWatchdog::ScopedCheck check(startup_disk_watchdog);
552
0
        LOG(ERROR) << "All spill disks are broken, exit.";
553
0
        exit(-1);
554
0
    }
555
7
    startup_disk_watchdog.stop();
556
557
    // initialize libcurl here to avoid concurrent initialization
558
7
    auto curl_ret = curl_global_init(CURL_GLOBAL_ALL);
559
7
    if (curl_ret != 0) {
560
0
        LOG(ERROR) << "fail to initialize libcurl, curl_ret=" << curl_ret;
561
0
        exit(-1);
562
0
    }
563
    // add logger for thrift internal
564
7
    apache::thrift::TOutput::instance().setOutputFunction(doris::thrift_output);
565
566
7
    Status status = Status::OK();
567
    // No JVM is started here on purpose. It is created by the first Java feature that asks
568
    // for it - a JNI table format, a Java UDF, an hdfs access - and a BE that uses none of
569
    // them runs without one. See Jni::JvmLauncher.
570
571
7
    if (doris::config::enable_python_udf_support) {
572
7
        if (std::string python_udf_root_path =
573
7
                    fmt::format("{}/lib/udf/python", std::getenv("DORIS_HOME"));
574
7
            !std::filesystem::exists(python_udf_root_path)) {
575
1
            std::filesystem::create_directories(python_udf_root_path);
576
1
        }
577
578
        // Normalize and trim all Python-related config parameters
579
7
        std::string python_env_mode =
580
7
                std::string(doris::trim(doris::to_lower(doris::config::python_env_mode)));
581
7
        std::string python_conda_root_path =
582
7
                std::string(doris::trim(doris::config::python_conda_root_path));
583
7
        std::string python_venv_root_path =
584
7
                std::string(doris::trim(doris::config::python_venv_root_path));
585
7
        std::string python_venv_interpreter_paths =
586
7
                std::string(doris::trim(doris::config::python_venv_interpreter_paths));
587
588
7
        if (python_env_mode == "conda") {
589
1
            if (python_conda_root_path.empty()) {
590
0
                LOG(ERROR)
591
0
                        << "Python conda root path is empty, please set `python_conda_root_path` "
592
0
                           "or set `enable_python_udf_support` to `false`";
593
0
                exit(1);
594
0
            }
595
1
            LOG(INFO) << "Doris backend python version manager is initialized. Python conda "
596
1
                         "root path: "
597
1
                      << python_conda_root_path;
598
1
            status = doris::PythonVersionManager::instance().init(doris::PythonEnvType::CONDA,
599
1
                                                                  python_conda_root_path, "");
600
6
        } else if (python_env_mode == "venv") {
601
6
            if (python_venv_root_path.empty()) {
602
0
                LOG(ERROR)
603
0
                        << "Python venv root path is empty, please set `python_venv_root_path` or "
604
0
                           "set `enable_python_udf_support` to `false`";
605
0
                exit(1);
606
0
            }
607
6
            if (python_venv_interpreter_paths.empty()) {
608
0
                LOG(ERROR)
609
0
                        << "Python interpreter paths is empty, please set "
610
0
                           "`python_venv_interpreter_paths` or set `enable_python_udf_support` to "
611
0
                           "`false`";
612
0
                exit(1);
613
0
            }
614
6
            LOG(INFO) << "Doris backend python version manager is initialized. Python venv "
615
6
                         "root path: "
616
6
                      << python_venv_root_path
617
6
                      << ", python interpreter paths: " << python_venv_interpreter_paths;
618
6
            status = doris::PythonVersionManager::instance().init(doris::PythonEnvType::VENV,
619
6
                                                                  python_venv_root_path,
620
6
                                                                  python_venv_interpreter_paths);
621
6
        } else {
622
0
            status = Status::InvalidArgument(
623
0
                    "Python env mode is invalid, should be `conda` or `venv`. If you don't want to "
624
0
                    "enable the Python UDF function, please set `enable_python_udf_support` to "
625
0
                    "`false`");
626
0
        }
627
628
7
        if (!status.ok()) {
629
0
            LOG(ERROR) << "Failed to initialize python version manager: " << status;
630
0
            exit(1);
631
0
        }
632
7
        LOG(INFO) << doris::PythonVersionManager::instance().to_string();
633
7
    }
634
635
    // SIGINT and SIGTERM are how the BE is asked to shut down, and the handler installed
636
    // here does nothing but raise the flag the loop at the end of main() waits on, so the
637
    // shutdown stays orderly. SIGQUIT is claimed here as well, so that it does nothing at
638
    // all rather than killing the BE with a core dump. A JVM would rather turn the first
639
    // two into a Java Shutdown.exit() and answer the third with a thread dump, and it
640
    // installs handlers of its own for all three when it starts. The JVM used to be created
641
    // a few lines above this call, which is what left these handlers on top; now that it is
642
    // created on demand, Jni::JvmLauncher::_bootstrap() is what puts them back once the JVM
643
    // has had its way with them.
644
    // https://www.oracle.com/java/technologies/javase/signals.html
645
7
    doris::init_signals();
646
    // ATTN: MUST init before `ExecEnv`, `StorageEngine` and other daemon services
647
    //
648
    //       Daemon ───┬──► StorageEngine ──► ExecEnv ──► Disk/Mem/CpuInfo
649
    //                 │
650
    //                 │
651
    // BackendService ─┘
652
7
    doris::CpuInfo::init();
653
7
    doris::DiskInfo::init();
654
7
    doris::MemInfo::init();
655
656
7
    LOG(INFO) << doris::CpuInfo::debug_string();
657
7
    LOG(INFO) << doris::DiskInfo::debug_string();
658
7
    LOG(INFO) << doris::MemInfo::debug_string();
659
660
    // Doris-patched GNU libunwind reads PHDR metadata from our lock-free snapshot instead of
661
    // entering glibc dl_iterate_phdr while jemalloc profiling or signal-context unwinding may
662
    // already be involved in loader-lock-sensitive code. Configure libunwind before daemon threads
663
    // start so all later heap-profile and stack-trace unwinds use the same lock-safe policy.
664
7
    configureLibunwindPHDRCache();
665
7
    updatePHDRCache();
666
7
    LOG(INFO) << "PHDR cache enabled: " << hasPHDRCache();
667
7
#if defined(__ELF__) && !defined(__FreeBSD__)
668
7
    auto symbol_index = doris::SymbolIndex::instance();
669
7
    LOG(INFO) << "SymbolIndex preloaded: objects=" << symbol_index->objects().size()
670
7
              << " symbols=" << symbol_index->symbols().size();
671
7
#endif
672
7
    if (!doris::BackendOptions::init()) {
673
0
        exit(-1);
674
0
    }
675
676
    // init exec env
677
7
    auto* exec_env(doris::ExecEnv::GetInstance());
678
7
    status = doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths, spill_paths, broken_paths);
679
7
    if (status != Status::OK()) {
680
0
        std::cerr << "failed to init doris storage engine, res=" << status;
681
0
        exit_on_startup_failure();
682
0
    }
683
684
    // Start concurrency stats manager
685
7
    doris::ConcurrencyStatsManager::instance().start();
686
687
    // begin to start services
688
7
    doris::ThriftRpcHelper::setup(exec_env);
689
    // 1. thrift server with be_port
690
7
    std::shared_ptr<doris::BaseBackendService> service;
691
7
    std::unique_ptr<doris::server::IServerStarter> backend_thrift_starter;
692
7
    std::unique_ptr<doris::server::IServerStarter> brpc_starter;
693
7
    std::unique_ptr<doris::server::IServerStarter> http_starter;
694
7
    std::unique_ptr<doris::server::IServerStarter> heartbeat_thrift_starter;
695
7
    std::unique_ptr<doris::server::IServerStarter> flight_starter;
696
7
    bool backend_thrift_started = false;
697
7
    bool brpc_started = false;
698
7
    bool http_started = false;
699
7
    bool heartbeat_thrift_started = false;
700
7
    bool flight_started = false;
701
7
    auto stop_and_join_server = [](std::unique_ptr<doris::server::IServerStarter>& starter,
702
7
                                   bool started) {
703
0
        if (starter != nullptr) {
704
0
            if (started) {
705
0
                starter->stop();
706
0
            }
707
0
            starter->join();
708
0
        }
709
0
    };
710
7
    std::function<void(Status&, std::string_view)> stop_work_if_error = [&](Status& status,
711
70
                                                                            std::string_view msg) {
712
70
        if (!status.ok()) {
713
0
            std::cerr << msg << '\n';
714
0
            service->stop_works();
715
0
            if (doris::config::enable_graceful_exit_check) {
716
0
                stop_and_join_server(flight_starter, flight_started);
717
0
                stop_and_join_server(heartbeat_thrift_starter, heartbeat_thrift_started);
718
0
                stop_and_join_server(http_starter, http_started);
719
0
                stop_and_join_server(backend_thrift_starter, backend_thrift_started);
720
0
                stop_and_join_server(brpc_starter, brpc_started);
721
0
                service.reset();
722
0
            }
723
0
            exit_on_startup_failure();
724
0
        }
725
70
    };
726
727
7
    if (doris::config::is_cloud_mode()) {
728
1
        service = std::make_shared<doris::CloudBackendService>(
729
1
                exec_env->storage_engine().to_cloud(), exec_env);
730
6
    } else {
731
6
        service = std::make_shared<doris::BackendService>(exec_env->storage_engine().to_local(),
732
6
                                                          exec_env);
733
6
    }
734
735
7
    status = doris::server::create_backend_thrift_starter(exec_env, doris::config::be_port, service,
736
7
                                                          &backend_thrift_starter);
737
7
    stop_work_if_error(status, "Failed to create BE server, exiting");
738
7
    status = backend_thrift_starter->start();
739
7
    backend_thrift_started = status.ok();
740
7
    stop_work_if_error(status, "Doris BE server did not start correctly, exiting");
741
742
    // 2. brpc service
743
7
    status = doris::server::create_brpc_starter(exec_env, doris::config::brpc_port,
744
7
                                                doris::config::brpc_num_threads, &brpc_starter);
745
7
    stop_work_if_error(status, "Failed to create BRPC service, exiting");
746
7
    status = brpc_starter->start();
747
7
    brpc_started = status.ok();
748
7
    stop_work_if_error(status, "BRPC service did not start correctly, exiting");
749
750
    // 3. http service
751
7
    status =
752
7
            doris::server::create_http_starter(exec_env, doris::config::webserver_port,
753
7
                                               doris::config::webserver_num_workers, &http_starter);
754
7
    stop_work_if_error(status, "Failed to create BE HTTP service, exiting");
755
7
    status = http_starter->start();
756
7
    http_started = status.ok();
757
7
    stop_work_if_error(status, "Doris Be http service did not start correctly, exiting");
758
759
    // 4. heart beat server
760
7
    doris::ClusterInfo* cluster_info = exec_env->cluster_info();
761
7
    status = doris::server::create_heartbeat_thrift_starter(
762
7
            exec_env, doris::config::heartbeat_service_port,
763
7
            doris::config::heartbeat_service_thread_count, cluster_info, &heartbeat_thrift_starter);
764
7
    stop_work_if_error(status, "Heartbeat services did not start correctly, exiting");
765
766
7
    status = heartbeat_thrift_starter->start();
767
7
    heartbeat_thrift_started = status.ok();
768
7
    stop_work_if_error(status, "Doris BE HeartBeat Service did not start correctly, exiting: " +
769
7
                                       status.to_string());
770
771
    // 5. arrow flight service
772
7
    status = doris::server::create_flight_starter(doris::config::arrow_flight_sql_port,
773
7
                                                  &flight_starter);
774
7
    stop_work_if_error(status, "Failed to create Arrow Flight service, exiting");
775
7
    status = flight_starter->start();
776
7
    flight_started = status.ok();
777
7
    stop_work_if_error(
778
7
            status, "Arrow Flight Service did not start correctly, exiting, " + status.to_string());
779
780
    // 6. start daemon thread to do clean or gc jobs
781
7
    doris::Daemon daemon;
782
7
    daemon.start();
783
784
7
    exec_env->storage_engine().notify_listeners();
785
786
7
    doris::k_is_server_ready = true;
787
788
    // 7. load the deployed Java plugins, once the BE is otherwise serving.
789
    //
790
    // On its own thread and non-fatal on purpose: the point is that a plugin broken by a bad
791
    // deployment shows up in the log now instead of inside the first user query that needs
792
    // it, and a plugin that cannot load must not hold up or take down everything else. When
793
    // no plugin is deployed this starts no JVM and returns immediately.
794
    //
795
    // Joined on the way out rather than detached, so that a stop arriving while plugins are
796
    // still loading waits for them instead of running the global destructors underneath a
797
    // thread that is inside the JVM. Warming up is bounded - one JVM start plus one pass over
798
    // the plugin directory - and the Java side of it is a single call, so there is nothing to
799
    // interrupt halfway.
800
7
    std::shared_ptr<doris::Thread> plugin_warmup_thread;
801
7
    if (doris::config::enable_java_support && doris::config::java_plugin_warmup) {
802
0
        EXIT_IF_ERROR(doris::Thread::create(
803
0
                "Jni", "java_plugin_warmup",
804
0
                []() {
805
                    // Named background thread with a thread context of its own: everything it
806
                    // allocates would otherwise be orphan memory, and the try/catch is what
807
                    // keeps a directory that becomes unreadable mid-iteration from reaching
808
                    // std::terminate (directory_iterator::operator++ throws).
809
0
                    SCOPED_INIT_THREAD_CONTEXT();
810
0
                    try {
811
0
                        if (Status status = doris::Jni::PluginRegistry::warmup(); !status.ok()) {
812
0
                            LOG(WARNING) << "failed to warm up Java plugins: " << status;
813
0
                        }
814
0
                    } catch (const std::exception& e) {
815
0
                        LOG(WARNING) << "failed to warm up Java plugins: " << e.what();
816
0
                    } catch (...) {
817
0
                        LOG(WARNING) << "failed to warm up Java plugins: unknown exception";
818
0
                    }
819
0
                },
820
0
                &plugin_warmup_thread));
821
0
    }
822
823
1.51k
    while (!doris::k_doris_exit) {
824
#if defined(LEAK_SANITIZER)
825
        __lsan_do_leak_check();
826
#endif
827
1.51k
        sleep(3);
828
1.51k
    }
829
7
    doris::k_is_server_ready = false;
830
7
    LOG(INFO) << "Doris main exiting.";
831
7
#if defined(LLVM_PROFILE)
832
7
    __llvm_profile_write_file();
833
7
    LOG(INFO) << "Flush profile file.";
834
7
#endif
835
    // For graceful shutdown, need to wait for all running queries to stop
836
7
    exec_env->wait_for_all_tasks_done();
837
838
7
    if (!doris::config::enable_graceful_exit_check) {
839
        // If not in memleak check mode, no need to wait all objects de-constructed normally, just exit.
840
        // It will make sure that graceful shutdown can be done definitely.
841
0
        LOG(INFO) << "Doris main exited.";
842
0
        google::FlushLogFiles(google::GLOG_INFO);
843
0
        _exit(0); // Do not call exit(0), it will wait for all objects de-constructed normally
844
0
        return 0;
845
0
    }
846
    // Before anything is torn down: the warmup thread may still be inside the JVM, and it
847
    // reaches BE state that the destructors below free. The fast path above does not need
848
    // this - _exit() runs no destructor at all.
849
7
    if (plugin_warmup_thread != nullptr) {
850
0
        plugin_warmup_thread->join();
851
0
        LOG(INFO) << "Java plugin warmup stopped";
852
0
    }
853
7
    daemon.stop();
854
7
    flight_starter->stop();
855
7
    flight_starter->join();
856
7
    LOG(INFO) << "Flight server stopped.";
857
7
    heartbeat_thrift_starter->stop();
858
7
    heartbeat_thrift_starter->join();
859
7
    LOG(INFO) << "Heartbeat server stopped";
860
    // The stream load recorder manager writes its audit records through this BE's own http
861
    // service, so it has to be stopped while that service is still up. Otherwise an audit
862
    // load that is in flight here can never be answered, and it blocks the join() done by
863
    // SAFE_STOP(_stream_load_recorder_manager) in ExecEnv::destroy() for up to the stream
864
    // load timeout, which is longer than the grace period of stop_be.sh --grace.
865
7
    if (auto* recorder_manager = exec_env->stream_load_recorder_manager();
866
7
        recorder_manager != nullptr) {
867
3
        recorder_manager->stop();
868
3
    }
869
    // TODO(zhiqiang): http_service
870
7
    http_starter->stop();
871
7
    http_starter->join();
872
7
    LOG(INFO) << "Http service stopped";
873
7
    backend_thrift_starter->stop();
874
7
    backend_thrift_starter->join();
875
7
    LOG(INFO) << "Be server stopped";
876
7
    brpc_starter->stop();
877
7
    brpc_starter->join();
878
7
    LOG(INFO) << "Brpc service stopped";
879
7
    service.reset();
880
7
    LOG(INFO) << "Backend Service stopped";
881
7
    exec_env->destroy();
882
7
    LOG(INFO) << "All service stopped, doris main exited.";
883
7
    return 0;
884
7
}
885
886
0
static void help(const char* progname) {
887
0
    printf("%s is the Doris backend server.\n\n", progname);
888
0
    printf("Usage:\n  %s [OPTION]...\n\n", progname);
889
0
    printf("Options:\n");
890
0
    printf("  -v, --version      output version information, then exit\n");
891
0
    printf("  -?, --help         show this help, then exit\n");
892
0
}