be/src/udf/python/python_udf_runtime.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 "udf/python/python_udf_runtime.h" |
19 | | |
20 | | #include <butil/fd_utility.h> |
21 | | #include <sys/wait.h> |
22 | | #include <unistd.h> |
23 | | |
24 | | #include <boost/process.hpp> |
25 | | |
26 | | #include "common/logging.h" |
27 | | |
28 | | namespace doris { |
29 | | |
30 | 28 | void PythonUDFProcess::remove_unix_socket() { |
31 | 28 | if (_uri.empty() || _unix_socket_file_path.empty()) return; |
32 | | |
33 | 28 | if (unlink(_unix_socket_file_path.c_str()) == 0) { |
34 | 16 | LOG(INFO) << "Successfully removed unix socket: " << _unix_socket_file_path; |
35 | 16 | return; |
36 | 16 | } |
37 | | |
38 | 12 | if (errno == ENOENT) { |
39 | | // File does not exist, this is fine, no need to warn |
40 | 11 | LOG(INFO) << "Unix socket not found (already removed): " << _uri; |
41 | 11 | } else { |
42 | 1 | LOG(WARNING) << "Failed to remove unix socket " << _uri << ": " << std::strerror(errno) |
43 | 1 | << " (errno=" << errno << ")"; |
44 | 1 | } |
45 | 12 | } |
46 | | |
47 | 57 | void PythonUDFProcess::shutdown() { |
48 | 57 | if (!_child.valid() || _is_shutdown) return; |
49 | | |
50 | 28 | _child.terminate(); |
51 | 28 | bool graceful = false; |
52 | 28 | constexpr std::chrono::milliseconds retry_interval(100); // 100ms |
53 | | |
54 | 28 | for (int i = 0; i < TERMINATE_RETRY_TIMES; ++i) { |
55 | 28 | if (!_child.running()) { |
56 | 28 | graceful = true; |
57 | 28 | break; |
58 | 28 | } |
59 | 0 | std::this_thread::sleep_for(retry_interval); |
60 | 0 | } |
61 | | |
62 | 28 | if (!graceful) { |
63 | 0 | LOG(WARNING) << "Python process did not terminate gracefully, sending SIGKILL"; |
64 | 0 | ::kill(_child_pid, SIGKILL); |
65 | 0 | _child.wait(); |
66 | 0 | } |
67 | | |
68 | 28 | if (int exit_code = _child.exit_code(); exit_code > 128 && exit_code <= 255) { |
69 | 0 | int signal = exit_code - 128; |
70 | 0 | LOG(INFO) << "Python process was killed by signal " << signal; |
71 | 28 | } else { |
72 | 28 | LOG(INFO) << "Python process exited normally with code: " << exit_code; |
73 | 28 | } |
74 | | |
75 | 28 | _output_stream.close(); |
76 | 28 | remove_unix_socket(); |
77 | 28 | _is_shutdown = true; |
78 | 28 | } |
79 | | |
80 | 1.31k | std::string PythonUDFProcess::to_string() const { |
81 | 1.31k | return fmt::format( |
82 | 1.31k | "PythonUDFProcess(child_pid={}, uri={}, " |
83 | 1.31k | "unix_socket_file_path={}, is_shutdown={})", |
84 | 1.31k | _child_pid, _uri, _unix_socket_file_path, _is_shutdown); |
85 | 1.31k | } |
86 | | |
87 | | } // namespace doris |