/root/doris/be/src/util/thrift_client.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 "util/thrift_client.h" | 
| 19 |  |  | 
| 20 |  | #include <thrift/transport/TTransport.h> | 
| 21 |  | #include <thrift/transport/TTransportException.h> | 
| 22 |  | // IWYU pragma: no_include <bits/chrono.h> | 
| 23 |  | #include <chrono> // IWYU pragma: keep | 
| 24 |  | #include <string> | 
| 25 |  | #include <thread> | 
| 26 |  |  | 
| 27 |  | #include "absl/strings/substitute.h" | 
| 28 |  |  | 
| 29 |  | namespace doris { | 
| 30 |  |  | 
| 31 | 0 | Status ThriftClientImpl::open() { | 
| 32 | 0 |     try { | 
| 33 | 0 |         if (!_transport->isOpen()) { | 
| 34 | 0 |             _transport->open(); | 
| 35 | 0 |         } | 
| 36 | 0 |     } catch (const apache::thrift::transport::TTransportException& e) { | 
| 37 | 0 |         try { | 
| 38 | 0 |             _transport->close(); | 
| 39 | 0 |         } catch (const apache::thrift::transport::TTransportException& e) { | 
| 40 | 0 |             VLOG_CRITICAL << "Error closing socket to: " << ipaddress() << ":" << port() | 
| 41 | 0 |                           << ", ignoring (" << e.what() << ")"; | 
| 42 | 0 |         } | 
| 43 |  |         // In certain cases in which the remote host is overloaded, this failure can | 
| 44 |  |         // happen quite frequently. Let's print this error message without the stack | 
| 45 |  |         // trace as there aren't many callers of this function. | 
| 46 | 0 |         const std::string& err_msg = absl::Substitute("Couldn't open transport for $0:$1 ($2)", | 
| 47 | 0 |                                                       ipaddress(), port(), e.what()); | 
| 48 | 0 |         VLOG_CRITICAL << err_msg; | 
| 49 | 0 |         return Status::RpcError(err_msg); | 
| 50 | 0 |     } | 
| 51 | 0 |     return Status::OK(); | 
| 52 | 0 | } | 
| 53 |  |  | 
| 54 | 0 | Status ThriftClientImpl::open_with_retry(int num_tries, int wait_ms) { | 
| 55 | 0 |     DCHECK_GE(wait_ms, 0); | 
| 56 | 0 |     Status status; | 
| 57 | 0 |     int try_count = 0L; | 
| 58 |  | 
 | 
| 59 | 0 |     while (num_tries <= 0 || try_count < num_tries) { | 
| 60 | 0 |         ++try_count; | 
| 61 | 0 |         status = open(); | 
| 62 |  | 
 | 
| 63 | 0 |         if (status.ok()) { | 
| 64 | 0 |             return status; | 
| 65 | 0 |         } | 
| 66 |  |  | 
| 67 | 0 |         LOG(INFO) << "Unable to connect to " << _ipaddress << ":" << _port; | 
| 68 |  | 
 | 
| 69 | 0 |         if (num_tries < 0) { | 
| 70 | 0 |             LOG(INFO) << "(Attempt " << try_count << ", will retry indefinitely)"; | 
| 71 | 0 |         } else { | 
| 72 | 0 |             LOG(INFO) << "(Attempt " << try_count << " of " << num_tries << ")"; | 
| 73 | 0 |         } | 
| 74 |  | 
 | 
| 75 | 0 |         std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms)); | 
| 76 | 0 |     } | 
| 77 |  |  | 
| 78 | 0 |     return status; | 
| 79 | 0 | } | 
| 80 |  |  | 
| 81 | 0 | void ThriftClientImpl::close() { | 
| 82 | 0 |     try { | 
| 83 | 0 |         if (_transport != nullptr && _transport->isOpen()) { | 
| 84 | 0 |             _transport->close(); | 
| 85 | 0 |         } | 
| 86 | 0 |     } catch (const apache::thrift::transport::TTransportException& e) { | 
| 87 | 0 |         LOG(INFO) << "Error closing connection to: " << ipaddress() << ":" << port() | 
| 88 | 0 |                   << ", ignoring (" << e.what() << ")"; | 
| 89 |  |         // Forcibly close the socket (since the transport may have failed to get that far | 
| 90 |  |         // during close()) | 
| 91 | 0 |         try { | 
| 92 | 0 |             if (_socket != nullptr) { | 
| 93 | 0 |                 _socket->close(); | 
| 94 | 0 |             } | 
| 95 | 0 |         } catch (const apache::thrift::transport::TTransportException& e) { | 
| 96 |  |             LOG(INFO) << "Error closing socket to: " << ipaddress() << ":" << port() | 
| 97 | 0 |                       << ", ignoring (" << e.what() << ")"; | 
| 98 | 0 |         } | 
| 99 | 0 |     } | 
| 100 | 0 | } | 
| 101 |  |  | 
| 102 |  | } // namespace doris |