Coverage Report

Created: 2024-11-21 14:00

/root/doris/be/src/util/proto_util.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 <brpc/http_method.h>
21
#include <gen_cpp/internal_service.pb.h>
22
23
#include "common/config.h"
24
#include "common/status.h"
25
#include "network_util.h"
26
#include "runtime/exec_env.h"
27
#include "runtime/runtime_state.h"
28
#include "util/brpc_client_cache.h"
29
30
namespace doris {
31
32
// When the tuple/block data is greater than 2G, embed the tuple/block data
33
// and the request serialization string in the attachment, and use "http" brpc.
34
// "http"brpc requires that only one of request and attachment be non-null.
35
//
36
// 2G: In the default "baidu_std" brpcd, upper limit of the request and attachment length is 2G.
37
constexpr size_t MIN_HTTP_BRPC_SIZE = (1ULL << 31);
38
39
// Embed column_values and brpc request serialization string in controller attachment.
40
template <typename Params, typename Closure>
41
Status request_embed_attachment_contain_blockv2(Params* brpc_request,
42
0
                                                std::unique_ptr<Closure>& closure) {
43
0
    std::string column_values = std::move(*brpc_request->mutable_block()->mutable_column_values());
44
0
    brpc_request->mutable_block()->mutable_column_values()->clear();
45
0
    return request_embed_attachmentv2(brpc_request, column_values, closure);
46
0
}
Unexecuted instantiation: _ZN5doris40request_embed_attachment_contain_blockv2INS_28PTabletWriterAddBlockRequestENS_18AutoReleaseClosureIS1_NS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEEEEENS_6StatusEPT_RSt10unique_ptrIT0_St14default_deleteISC_EE
Unexecuted instantiation: _ZN5doris40request_embed_attachment_contain_blockv2INS_19PTransmitDataParamsENS_18AutoReleaseClosureIS1_NS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEEEEENS_6StatusEPT_RSt10unique_ptrIT0_St14default_deleteISC_EE
47
48
0
inline bool enable_http_send_block(const PTransmitDataParams& request) {
49
0
    if (!config::transfer_large_data_by_brpc) {
50
0
        return false;
51
0
    }
52
0
    if (!request.has_block() || !request.block().has_column_values()) {
53
0
        return false;
54
0
    }
55
0
    if (request.ByteSizeLong() < MIN_HTTP_BRPC_SIZE) {
56
0
        return false;
57
0
    }
58
0
    return true;
59
0
}
60
61
template <typename Closure>
62
0
void transmit_blockv2(PBackendService_Stub& stub, std::unique_ptr<Closure> closure) {
63
0
    closure->cntl_->http_request().Clear();
64
0
    stub.transmit_block(closure->cntl_.get(), closure->request_.get(), closure->response_.get(),
65
0
                        closure.get());
66
0
    closure.release();
67
0
}
68
69
template <typename Closure>
70
Status transmit_block_httpv2(ExecEnv* exec_env, std::unique_ptr<Closure> closure,
71
0
                             TNetworkAddress brpc_dest_addr) {
72
0
    RETURN_IF_ERROR(request_embed_attachment_contain_blockv2(closure->request_.get(), closure));
73
74
0
    std::string host = brpc_dest_addr.hostname;
75
0
    auto dns_cache = ExecEnv::GetInstance()->dns_cache();
76
0
    if (dns_cache == nullptr) {
77
0
        LOG(WARNING) << "DNS cache is not initialized, skipping hostname resolve";
78
0
    } else if (!is_valid_ip(brpc_dest_addr.hostname)) {
79
0
        Status status = dns_cache->get(brpc_dest_addr.hostname, &host);
80
0
        if (!status.ok()) {
81
0
            LOG(WARNING) << "failed to get ip from host " << brpc_dest_addr.hostname << ": "
82
0
                         << status.to_string();
83
0
            return Status::InternalError("failed to get ip from host {}", brpc_dest_addr.hostname);
84
0
        }
85
0
    }
86
    //format an ipv6 address
87
0
    std::string brpc_url = get_brpc_http_url(brpc_dest_addr.hostname, brpc_dest_addr.port);
88
89
0
    std::shared_ptr<PBackendService_Stub> brpc_http_stub =
90
0
            exec_env->brpc_internal_client_cache()->get_new_client_no_cache(brpc_url, "http");
91
0
    if (brpc_http_stub == nullptr) {
92
0
        return Status::InternalError("failed to open brpc http client to {}", brpc_url);
93
0
    }
94
0
    closure->cntl_->http_request().uri() =
95
0
            brpc_url + "/PInternalServiceImpl/transmit_block_by_http";
96
0
    closure->cntl_->http_request().set_method(brpc::HTTP_METHOD_POST);
97
0
    closure->cntl_->http_request().set_content_type("application/json");
98
0
    brpc_http_stub->transmit_block_by_http(closure->cntl_.get(), nullptr, closure->response_.get(),
99
0
                                           closure.get());
100
0
    closure.release();
101
102
0
    return Status::OK();
103
0
}
104
105
template <typename Params, typename Closure>
106
Status request_embed_attachmentv2(Params* brpc_request, const std::string& data,
107
0
                                  std::unique_ptr<Closure>& closure) {
108
0
    butil::IOBuf attachment;
109
110
    // step1: serialize brpc_request to string, and append to attachment.
111
0
    std::string req_str;
112
0
    if (!brpc_request->SerializeToString(&req_str)) {
113
0
        return Status::InternalError("failed to serialize the request");
114
0
    }
115
0
    int64_t req_str_size = req_str.size();
116
0
    attachment.append(&req_str_size, sizeof(req_str_size));
117
0
    attachment.append(req_str);
118
119
    // step2: append data to attachment and put it in the closure.
120
0
    int64_t data_size = data.size();
121
0
    attachment.append(&data_size, sizeof(data_size));
122
0
    try {
123
0
        attachment.append(data);
124
0
    } catch (...) {
125
0
        LOG(WARNING) << "Try to alloc " << data_size
126
0
                     << " bytes for append data to attachment failed. ";
127
0
        return Status::MemoryAllocFailed("request embed attachment failed to memcpy {} bytes",
128
0
                                         data_size);
129
0
    }
130
    // step3: attachment add to closure.
131
0
    closure->cntl_->request_attachment().swap(attachment);
132
0
    return Status::OK();
133
0
}
Unexecuted instantiation: _ZN5doris26request_embed_attachmentv2INS_28PTabletWriterAddBlockRequestENS_18AutoReleaseClosureIS1_NS_10vectorized18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEEEEENS_6StatusEPT_RKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERSt10unique_ptrIT0_St14default_deleteISK_EE
Unexecuted instantiation: _ZN5doris26request_embed_attachmentv2INS_19PTransmitDataParamsENS_18AutoReleaseClosureIS1_NS_8pipeline20ExchangeSendCallbackINS_19PTransmitDataResultEEEEEEENS_6StatusEPT_RKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERSt10unique_ptrIT0_St14default_deleteISK_EE
134
135
// Extract the brpc request and block from the controller attachment,
136
// and put the block into the request.
137
template <typename Params>
138
Status attachment_extract_request_contain_block(const Params* brpc_request,
139
0
                                                brpc::Controller* cntl) {
140
0
    Params* req = const_cast<Params*>(brpc_request);
141
0
    auto block = req->mutable_block();
142
0
    return attachment_extract_request(req, cntl, block->mutable_column_values());
143
0
}
Unexecuted instantiation: _ZN5doris40attachment_extract_request_contain_blockINS_28PTabletWriterAddBlockRequestEEENS_6StatusEPKT_PN4brpc10ControllerE
Unexecuted instantiation: _ZN5doris40attachment_extract_request_contain_blockINS_19PTransmitDataParamsEEENS_6StatusEPKT_PN4brpc10ControllerE
144
145
template <typename Params>
146
Status attachment_extract_request(const Params* brpc_request, brpc::Controller* cntl,
147
0
                                  std::string* data) {
148
0
    const butil::IOBuf& io_buf = cntl->request_attachment();
149
150
    // step1: deserialize request string to brpc_request from attachment.
151
0
    int64_t req_str_size;
152
0
    io_buf.copy_to(&req_str_size, sizeof(req_str_size), 0);
153
0
    std::string req_str;
154
0
    io_buf.copy_to(&req_str, req_str_size, sizeof(req_str_size));
155
0
    Params* req = const_cast<Params*>(brpc_request);
156
0
    req->ParseFromString(req_str);
157
158
    // step2: extract data from attachment.
159
0
    int64_t data_size;
160
0
    io_buf.copy_to(&data_size, sizeof(data_size), sizeof(req_str_size) + req_str_size);
161
0
    try {
162
0
        io_buf.copy_to(data, data_size, sizeof(data_size) + sizeof(req_str_size) + req_str_size);
163
0
    } catch (...) {
164
0
        LOG(WARNING) << "Try to alloc " << data_size
165
0
                     << " bytes for extract data from attachment failed. ";
166
0
        return Status::MemoryAllocFailed("attachment extract request failed to memcpy {} bytes",
167
0
                                         data_size);
168
0
    }
169
0
    return Status::OK();
170
0
}
Unexecuted instantiation: _ZN5doris26attachment_extract_requestINS_28PTabletWriterAddBlockRequestEEENS_6StatusEPKT_PN4brpc10ControllerEPNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEE
Unexecuted instantiation: _ZN5doris26attachment_extract_requestINS_19PTransmitDataParamsEEENS_6StatusEPKT_PN4brpc10ControllerEPNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEE
171
172
} // namespace doris