/root/doris/be/src/util/brpc_closure.h
Line | Count | Source (jump to first uncovered line) |
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 | | #pragma once |
19 | | |
20 | | #include <google/protobuf/stubs/common.h> |
21 | | |
22 | | #include <atomic> |
23 | | #include <utility> |
24 | | |
25 | | #include "runtime/query_context.h" |
26 | | #include "runtime/thread_context.h" |
27 | | #include "service/brpc.h" |
28 | | #include "util/brpc_closure.h" |
29 | | |
30 | | namespace doris { |
31 | | |
32 | | template <typename Response> |
33 | | class DummyBrpcCallback { |
34 | | ENABLE_FACTORY_CREATOR(DummyBrpcCallback); |
35 | | |
36 | | public: |
37 | | using ResponseType = Response; |
38 | 22 | DummyBrpcCallback() { |
39 | 22 | cntl_ = std::make_shared<brpc::Controller>(); |
40 | 22 | response_ = std::make_shared<Response>(); |
41 | 22 | } Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PSendFilterSizeResponseEEC2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_20PMergeFilterResponseEEC2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEEC2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEEC2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_22PPublishFilterResponseEEC2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEEC2Ev _ZN5doris17DummyBrpcCallbackINS_19PTransmitDataResultEEC2Ev Line | Count | Source | 38 | 22 | DummyBrpcCallback() { | 39 | 22 | cntl_ = std::make_shared<brpc::Controller>(); | 40 | 22 | response_ = std::make_shared<Response>(); | 41 | 22 | } |
Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_27PTabletWriterAddBlockResultEEC2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PTabletWriterOpenResultEEC2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_25PTabletWriterCancelResultEEC2Ev |
42 | | |
43 | 22 | virtual ~DummyBrpcCallback() = default; Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PSendFilterSizeResponseEED2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_20PMergeFilterResponseEED2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEED2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEED2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_22PPublishFilterResponseEED2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEED2Ev _ZN5doris17DummyBrpcCallbackINS_19PTransmitDataResultEED2Ev Line | Count | Source | 43 | 22 | virtual ~DummyBrpcCallback() = default; |
Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_27PTabletWriterAddBlockResultEED2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PTabletWriterOpenResultEED2Ev Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_25PTabletWriterCancelResultEED2Ev |
44 | | |
45 | 0 | virtual void call() {} Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_19PTransmitDataResultEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PSendFilterSizeResponseEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_20PMergeFilterResponseEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_22PPublishFilterResponseEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_27PTabletWriterAddBlockResultEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PTabletWriterOpenResultEE4callEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_25PTabletWriterCancelResultEE4callEv |
46 | | |
47 | 0 | virtual void join() { brpc::Join(cntl_->call_id()); } Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_19PTransmitDataResultEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PSendFilterSizeResponseEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_20PMergeFilterResponseEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_22PPublishFilterResponseEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_27PTabletWriterAddBlockResultEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_23PTabletWriterOpenResultEE4joinEv Unexecuted instantiation: _ZN5doris17DummyBrpcCallbackINS_25PTabletWriterCancelResultEE4joinEv |
48 | | |
49 | | // controller has to be the same lifecycle with the closure, because brpc may use |
50 | | // it in any stage of the rpc. |
51 | | std::shared_ptr<brpc::Controller> cntl_; |
52 | | // We do not know if brpc will use request or response after brpc method returns. |
53 | | // So that we need keep a shared ptr here to ensure that brpc could use req/rep |
54 | | // at any stage. |
55 | | std::shared_ptr<Response> response_; |
56 | | }; |
57 | | |
58 | | // The closure will be deleted after callback. |
59 | | // It could only be created by using shared ptr or unique ptr. |
60 | | // It will hold a weak ptr of T and call run of T |
61 | | // Callback() { |
62 | | // xxxx; |
63 | | // public |
64 | | // void run() { |
65 | | // logxxx |
66 | | // } |
67 | | // } |
68 | | // |
69 | | // std::shared_ptr<Callback> b; |
70 | | // |
71 | | // std::unique_ptr<AutoReleaseClosure> a(b); |
72 | | // brpc_call(a.release()); |
73 | | |
74 | | template <typename T> |
75 | | concept HasStatus = requires(T* response) { response->status(); }; |
76 | | |
77 | | template <typename Request, typename Callback> |
78 | | class AutoReleaseClosure : public google::protobuf::Closure { |
79 | | using Weak = typename std::shared_ptr<Callback>::weak_type; |
80 | | using ResponseType = typename Callback::ResponseType; |
81 | | ENABLE_FACTORY_CREATOR(AutoReleaseClosure); |
82 | | |
83 | | public: |
84 | | AutoReleaseClosure(std::shared_ptr<Request> req, std::shared_ptr<Callback> callback, |
85 | | std::weak_ptr<QueryContext> context = {}) |
86 | 28 | : request_(req), callback_(callback), context_(std::move(context)) { |
87 | 28 | this->cntl_ = callback->cntl_; |
88 | 28 | this->response_ = callback->response_; |
89 | 28 | } Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSendFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSendFilterSizeResponseEEEEC2ESt10shared_ptrIS1_ES6_IS4_ESt8weak_ptrINS_12QueryContextEE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_19PMergeFilterRequestENS_17DummyBrpcCallbackINS_20PMergeFilterResponseEEEEC2ESt10shared_ptrIS1_ES6_IS4_ESt8weak_ptrINS_12QueryContextEE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriteSlaveRequestENS_17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEEEEC2ESt10shared_ptrIS1_ES6_IS4_ESt8weak_ptrINS_12QueryContextEE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSyncFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEEEEC2ESt10shared_ptrIS1_ES6_IS4_ESt8weak_ptrINS_12QueryContextEE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_23PPublishFilterRequestV2ENS_17DummyBrpcCallbackINS_22PPublishFilterResponseEEEEC2ESt10shared_ptrIS1_ES6_IS4_ESt8weak_ptrINS_12QueryContextEE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriteSlaveDoneRequestENS_17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEEEEC2ESt10shared_ptrIS1_ES6_IS4_ESt8weak_ptrINS_12QueryContextEE _ZN5doris18AutoReleaseClosureINS_19PTransmitDataParamsENS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEEC2ESt10shared_ptrIS1_ES7_IS5_ESt8weak_ptrINS_12QueryContextEE Line | Count | Source | 86 | 28 | : request_(req), callback_(callback), context_(std::move(context)) { | 87 | 28 | this->cntl_ = callback->cntl_; | 88 | 28 | this->response_ = callback->response_; | 89 | 28 | } |
Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriterOpenRequestENS_17DummyBrpcCallbackINS_23PTabletWriterOpenResultEEEEC2ESt10shared_ptrIS1_ES6_IS4_ESt8weak_ptrINS_12QueryContextEE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriterAddBlockRequestENS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEEC2ESt10shared_ptrIS1_ES7_IS5_ESt8weak_ptrINS_12QueryContextEE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_26PTabletWriterCancelRequestENS_17DummyBrpcCallbackINS_25PTabletWriterCancelResultEEEEC2ESt10shared_ptrIS1_ES6_IS4_ESt8weak_ptrINS_12QueryContextEE |
90 | | |
91 | 28 | ~AutoReleaseClosure() override = default; Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSendFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSendFilterSizeResponseEEEED2Ev Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_19PMergeFilterRequestENS_17DummyBrpcCallbackINS_20PMergeFilterResponseEEEED2Ev Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriteSlaveRequestENS_17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEEEED2Ev Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSyncFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEEEED2Ev Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_23PPublishFilterRequestV2ENS_17DummyBrpcCallbackINS_22PPublishFilterResponseEEEED2Ev Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriteSlaveDoneRequestENS_17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEEEED2Ev _ZN5doris18AutoReleaseClosureINS_19PTransmitDataParamsENS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEED2Ev Line | Count | Source | 91 | 28 | ~AutoReleaseClosure() override = default; |
Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriterOpenRequestENS_17DummyBrpcCallbackINS_23PTabletWriterOpenResultEEEED2Ev Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriterAddBlockRequestENS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEED2Ev Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_26PTabletWriterCancelRequestENS_17DummyBrpcCallbackINS_25PTabletWriterCancelResultEEEED2Ev |
92 | | |
93 | | // Will delete itself |
94 | 28 | void Run() override { |
95 | 28 | Defer defer {[&]() { delete this; }}; Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_22PSendFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSendFilterSizeResponseEEEE3RunEvENKUlvE_clEv Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_19PMergeFilterRequestENS_17DummyBrpcCallbackINS_20PMergeFilterResponseEEEE3RunEvENKUlvE_clEv Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_24PTabletWriteSlaveRequestENS_17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEEEE3RunEvENKUlvE_clEv Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_22PSyncFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEEEE3RunEvENKUlvE_clEv Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_23PPublishFilterRequestV2ENS_17DummyBrpcCallbackINS_22PPublishFilterResponseEEEE3RunEvENKUlvE_clEv Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_28PTabletWriteSlaveDoneRequestENS_17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEEEE3RunEvENKUlvE_clEv _ZZN5doris18AutoReleaseClosureINS_19PTransmitDataParamsENS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEE3RunEvENKUlvE_clEv Line | Count | Source | 95 | 28 | Defer defer {[&]() { delete this; }}; |
Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_24PTabletWriterOpenRequestENS_17DummyBrpcCallbackINS_23PTabletWriterOpenResultEEEE3RunEvENKUlvE_clEv Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_28PTabletWriterAddBlockRequestENS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEE3RunEvENKUlvE_clEv Unexecuted instantiation: _ZZN5doris18AutoReleaseClosureINS_26PTabletWriterCancelRequestENS_17DummyBrpcCallbackINS_25PTabletWriterCancelResultEEEE3RunEvENKUlvE_clEv |
96 | | // If lock failed, it means the callback object is deconstructed, then no need |
97 | | // to deal with the callback any more. |
98 | 28 | if (auto tmp = callback_.lock()) { |
99 | 28 | tmp->call(); |
100 | 28 | } |
101 | 28 | if (cntl_->Failed()) { |
102 | 1 | _process_if_rpc_failed(); |
103 | 27 | } else { |
104 | 27 | _process_status<ResponseType>(response_.get()); |
105 | 27 | } |
106 | 28 | } _ZN5doris18AutoReleaseClosureINS_19PTransmitDataParamsENS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEE3RunEv Line | Count | Source | 94 | 28 | void Run() override { | 95 | 28 | Defer defer {[&]() { delete this; }}; | 96 | | // If lock failed, it means the callback object is deconstructed, then no need | 97 | | // to deal with the callback any more. | 98 | 28 | if (auto tmp = callback_.lock()) { | 99 | 28 | tmp->call(); | 100 | 28 | } | 101 | 28 | if (cntl_->Failed()) { | 102 | 1 | _process_if_rpc_failed(); | 103 | 27 | } else { | 104 | 27 | _process_status<ResponseType>(response_.get()); | 105 | 27 | } | 106 | 28 | } |
Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSendFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSendFilterSizeResponseEEEE3RunEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_19PMergeFilterRequestENS_17DummyBrpcCallbackINS_20PMergeFilterResponseEEEE3RunEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriteSlaveRequestENS_17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEEEE3RunEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSyncFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEEEE3RunEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_23PPublishFilterRequestV2ENS_17DummyBrpcCallbackINS_22PPublishFilterResponseEEEE3RunEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriteSlaveDoneRequestENS_17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEEEE3RunEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriterOpenRequestENS_17DummyBrpcCallbackINS_23PTabletWriterOpenResultEEEE3RunEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriterAddBlockRequestENS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEE3RunEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_26PTabletWriterCancelRequestENS_17DummyBrpcCallbackINS_25PTabletWriterCancelResultEEEE3RunEv |
107 | | |
108 | | // controller has to be the same lifecycle with the closure, because brpc may use |
109 | | // it in any stage of the rpc. |
110 | | std::shared_ptr<brpc::Controller> cntl_; |
111 | | // We do not know if brpc will use request or response after brpc method returns. |
112 | | // So that we need keep a shared ptr here to ensure that brpc could use req/rep |
113 | | // at any stage. |
114 | | std::shared_ptr<Request> request_; |
115 | | std::shared_ptr<ResponseType> response_; |
116 | | |
117 | | protected: |
118 | 1 | virtual void _process_if_rpc_failed() { |
119 | 1 | std::string error_msg = "RPC meet failed: " + cntl_->ErrorText(); |
120 | 1 | if (auto ctx = context_.lock(); ctx) { |
121 | 0 | ctx->cancel(Status::NetworkError(error_msg)); |
122 | 1 | } else { |
123 | 1 | LOG(WARNING) << error_msg; |
124 | 1 | } |
125 | 1 | } _ZN5doris18AutoReleaseClosureINS_19PTransmitDataParamsENS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEE22_process_if_rpc_failedEv Line | Count | Source | 118 | 1 | virtual void _process_if_rpc_failed() { | 119 | 1 | std::string error_msg = "RPC meet failed: " + cntl_->ErrorText(); | 120 | 1 | if (auto ctx = context_.lock(); ctx) { | 121 | 0 | ctx->cancel(Status::NetworkError(error_msg)); | 122 | 1 | } else { | 123 | 1 | LOG(WARNING) << error_msg; | 124 | 1 | } | 125 | 1 | } |
Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSendFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSendFilterSizeResponseEEEE22_process_if_rpc_failedEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_19PMergeFilterRequestENS_17DummyBrpcCallbackINS_20PMergeFilterResponseEEEE22_process_if_rpc_failedEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriteSlaveRequestENS_17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEEEE22_process_if_rpc_failedEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSyncFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEEEE22_process_if_rpc_failedEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_23PPublishFilterRequestV2ENS_17DummyBrpcCallbackINS_22PPublishFilterResponseEEEE22_process_if_rpc_failedEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriteSlaveDoneRequestENS_17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEEEE22_process_if_rpc_failedEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriterOpenRequestENS_17DummyBrpcCallbackINS_23PTabletWriterOpenResultEEEE22_process_if_rpc_failedEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriterAddBlockRequestENS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEE22_process_if_rpc_failedEv Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_26PTabletWriterCancelRequestENS_17DummyBrpcCallbackINS_25PTabletWriterCancelResultEEEE22_process_if_rpc_failedEv |
126 | | |
127 | 2 | virtual void _process_if_meet_error_status(const Status& status) { |
128 | 2 | if (status.is<ErrorCode::END_OF_FILE>()) { |
129 | | // no need to log END_OF_FILE, reduce the unlessful log |
130 | 2 | return; |
131 | 2 | } |
132 | 0 | if (auto ctx = context_.lock(); ctx) { |
133 | 0 | ctx->cancel(status); |
134 | 0 | } else { |
135 | 0 | LOG(WARNING) << "RPC meet error status: " << status; |
136 | 0 | } |
137 | 0 | } _ZN5doris18AutoReleaseClosureINS_19PTransmitDataParamsENS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEE29_process_if_meet_error_statusERKNS_6StatusE Line | Count | Source | 127 | 2 | virtual void _process_if_meet_error_status(const Status& status) { | 128 | 2 | if (status.is<ErrorCode::END_OF_FILE>()) { | 129 | | // no need to log END_OF_FILE, reduce the unlessful log | 130 | 2 | return; | 131 | 2 | } | 132 | 0 | if (auto ctx = context_.lock(); ctx) { | 133 | 0 | ctx->cancel(status); | 134 | 0 | } else { | 135 | 0 | LOG(WARNING) << "RPC meet error status: " << status; | 136 | 0 | } | 137 | 0 | } |
Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSendFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSendFilterSizeResponseEEEE29_process_if_meet_error_statusERKNS_6StatusE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_19PMergeFilterRequestENS_17DummyBrpcCallbackINS_20PMergeFilterResponseEEEE29_process_if_meet_error_statusERKNS_6StatusE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriteSlaveRequestENS_17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEEEE29_process_if_meet_error_statusERKNS_6StatusE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSyncFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEEEE29_process_if_meet_error_statusERKNS_6StatusE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_23PPublishFilterRequestV2ENS_17DummyBrpcCallbackINS_22PPublishFilterResponseEEEE29_process_if_meet_error_statusERKNS_6StatusE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriteSlaveDoneRequestENS_17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEEEE29_process_if_meet_error_statusERKNS_6StatusE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriterOpenRequestENS_17DummyBrpcCallbackINS_23PTabletWriterOpenResultEEEE29_process_if_meet_error_statusERKNS_6StatusE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriterAddBlockRequestENS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEE29_process_if_meet_error_statusERKNS_6StatusE Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_26PTabletWriterCancelRequestENS_17DummyBrpcCallbackINS_25PTabletWriterCancelResultEEEE29_process_if_meet_error_statusERKNS_6StatusE |
138 | | |
139 | | private: |
140 | | template <typename Response> |
141 | 0 | void _process_status(Response* response) {} |
142 | | |
143 | | template <HasStatus Response> |
144 | 27 | void _process_status(Response* response) { |
145 | 27 | if (Status status = Status::create(response->status()); !status.ok()) { |
146 | 2 | _process_if_meet_error_status(status); |
147 | 2 | } |
148 | 27 | } _ZN5doris18AutoReleaseClosureINS_19PTransmitDataParamsENS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEE15_process_statusIS4_EEvPT_ Line | Count | Source | 144 | 27 | void _process_status(Response* response) { | 145 | 27 | if (Status status = Status::create(response->status()); !status.ok()) { | 146 | 2 | _process_if_meet_error_status(status); | 147 | 2 | } | 148 | 27 | } |
Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSendFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSendFilterSizeResponseEEEE15_process_statusIS3_EEvPT_ Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_19PMergeFilterRequestENS_17DummyBrpcCallbackINS_20PMergeFilterResponseEEEE15_process_statusIS3_EEvPT_ Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriteSlaveRequestENS_17DummyBrpcCallbackINS_23PTabletWriteSlaveResultEEEE15_process_statusIS3_EEvPT_ Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_22PSyncFilterSizeRequestENS_17DummyBrpcCallbackINS_23PSyncFilterSizeResponseEEEE15_process_statusIS3_EEvPT_ Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_23PPublishFilterRequestV2ENS_17DummyBrpcCallbackINS_22PPublishFilterResponseEEEE15_process_statusIS3_EEvPT_ Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriteSlaveDoneRequestENS_17DummyBrpcCallbackINS_27PTabletWriteSlaveDoneResultEEEE15_process_statusIS3_EEvPT_ Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_24PTabletWriterOpenRequestENS_17DummyBrpcCallbackINS_23PTabletWriterOpenResultEEEE15_process_statusIS3_EEvPT_ Unexecuted instantiation: _ZN5doris18AutoReleaseClosureINS_28PTabletWriterAddBlockRequestENS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEE15_process_statusIS4_EEvPT_ |
149 | | // Use a weak ptr to keep the callback, so that the callback can be deleted if the main |
150 | | // thread is freed. |
151 | | Weak callback_; |
152 | | std::weak_ptr<QueryContext> context_; |
153 | | }; |
154 | | |
155 | | } // namespace doris |