mirror of https://github.com/grpc/grpc.git
The C based gRPC (C++, Python, Ruby, Objective-C, PHP, C#)
https://grpc.io/
619 lines
23 KiB
619 lines
23 KiB
/* |
|
* |
|
* Copyright 2020 gRPC authors. |
|
* |
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
|
* you may not use this file except in compliance with the License. |
|
* You may obtain a copy of the License at |
|
* |
|
* http://www.apache.org/licenses/LICENSE-2.0 |
|
* |
|
* Unless required by applicable law or agreed to in writing, software |
|
* distributed under the License is distributed on an "AS IS" BASIS, |
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|
* See the License for the specific language governing permissions and |
|
* limitations under the License. |
|
* |
|
*/ |
|
|
|
#include <atomic> |
|
#include <chrono> |
|
#include <condition_variable> |
|
#include <deque> |
|
#include <map> |
|
#include <mutex> |
|
#include <set> |
|
#include <sstream> |
|
#include <string> |
|
#include <thread> |
|
#include <vector> |
|
|
|
#include "absl/algorithm/container.h" |
|
#include "absl/flags/flag.h" |
|
#include "absl/strings/str_split.h" |
|
|
|
#include <grpcpp/ext/admin_services.h> |
|
#include <grpcpp/ext/proto_server_reflection_plugin.h> |
|
#include <grpcpp/grpcpp.h> |
|
#include <grpcpp/server.h> |
|
#include <grpcpp/server_builder.h> |
|
#include <grpcpp/server_context.h> |
|
|
|
#include "src/core/lib/channel/status_util.h" |
|
#include "src/core/lib/gpr/env.h" |
|
#include "src/proto/grpc/testing/empty.pb.h" |
|
#include "src/proto/grpc/testing/messages.pb.h" |
|
#include "src/proto/grpc/testing/test.grpc.pb.h" |
|
#include "test/core/util/test_config.h" |
|
#include "test/cpp/util/test_config.h" |
|
|
|
ABSL_FLAG(bool, fail_on_failed_rpc, false, |
|
"Fail client if any RPCs fail after first successful RPC."); |
|
ABSL_FLAG(int32_t, num_channels, 1, "Number of channels."); |
|
ABSL_FLAG(bool, print_response, false, "Write RPC response to stdout."); |
|
ABSL_FLAG(int32_t, qps, 1, "Qps per channel."); |
|
// TODO(Capstan): Consider using absl::Duration |
|
ABSL_FLAG(int32_t, rpc_timeout_sec, 30, "Per RPC timeout seconds."); |
|
ABSL_FLAG(std::string, server, "localhost:50051", "Address of server."); |
|
ABSL_FLAG(int32_t, stats_port, 50052, |
|
"Port to expose peer distribution stats service."); |
|
ABSL_FLAG(std::string, rpc, "UnaryCall", |
|
"a comma separated list of rpc methods."); |
|
ABSL_FLAG(std::string, metadata, "", "metadata to send with the RPC."); |
|
ABSL_FLAG(std::string, expect_status, "OK", |
|
"RPC status for the test RPC to be considered successful"); |
|
ABSL_FLAG( |
|
bool, secure_mode, false, |
|
"If true, XdsCredentials are used, InsecureChannelCredentials otherwise"); |
|
|
|
using grpc::Channel; |
|
using grpc::ClientAsyncResponseReader; |
|
using grpc::ClientContext; |
|
using grpc::CompletionQueue; |
|
using grpc::Server; |
|
using grpc::ServerBuilder; |
|
using grpc::ServerContext; |
|
using grpc::Status; |
|
using grpc::testing::ClientConfigureRequest; |
|
using grpc::testing::ClientConfigureRequest_RpcType_Name; |
|
using grpc::testing::ClientConfigureResponse; |
|
using grpc::testing::Empty; |
|
using grpc::testing::LoadBalancerAccumulatedStatsRequest; |
|
using grpc::testing::LoadBalancerAccumulatedStatsResponse; |
|
using grpc::testing::LoadBalancerStatsRequest; |
|
using grpc::testing::LoadBalancerStatsResponse; |
|
using grpc::testing::LoadBalancerStatsService; |
|
using grpc::testing::SimpleRequest; |
|
using grpc::testing::SimpleResponse; |
|
using grpc::testing::TestService; |
|
using grpc::testing::XdsUpdateClientConfigureService; |
|
|
|
class XdsStatsWatcher; |
|
|
|
struct StatsWatchers { |
|
// Unique ID for each outgoing RPC |
|
int global_request_id = 0; |
|
// Unique ID for each outgoing RPC by RPC method type |
|
std::map<int, int> global_request_id_by_type; |
|
// Stores a set of watchers that should be notified upon outgoing RPC |
|
// completion |
|
std::set<XdsStatsWatcher*> watchers; |
|
// Global watcher for accumululated stats. |
|
XdsStatsWatcher* global_watcher; |
|
// Mutex for global_request_id and watchers |
|
std::mutex mu; |
|
}; |
|
// Whether at least one RPC has succeeded, indicating xDS resolution completed. |
|
std::atomic<bool> one_rpc_succeeded(false); |
|
// RPC configuration detailing how RPC should be sent. |
|
struct RpcConfig { |
|
ClientConfigureRequest::RpcType type; |
|
std::vector<std::pair<std::string, std::string>> metadata; |
|
int timeout_sec = 0; |
|
}; |
|
struct RpcConfigurationsQueue { |
|
// A queue of RPC configurations detailing how RPCs should be sent. |
|
std::deque<std::vector<RpcConfig>> rpc_configs_queue; |
|
// Mutex for rpc_configs_queue |
|
std::mutex mu_rpc_configs_queue; |
|
}; |
|
struct AsyncClientCall { |
|
Empty empty_response; |
|
SimpleResponse simple_response; |
|
ClientContext context; |
|
Status status; |
|
int saved_request_id; |
|
ClientConfigureRequest::RpcType rpc_type; |
|
std::unique_ptr<ClientAsyncResponseReader<Empty>> empty_response_reader; |
|
std::unique_ptr<ClientAsyncResponseReader<SimpleResponse>> |
|
simple_response_reader; |
|
}; |
|
|
|
/** Records the remote peer distribution for a given range of RPCs. */ |
|
class XdsStatsWatcher { |
|
public: |
|
XdsStatsWatcher(int start_id, int end_id) |
|
: start_id_(start_id), end_id_(end_id), rpcs_needed_(end_id - start_id) {} |
|
|
|
// Upon the completion of an RPC, we will look at the request_id, the |
|
// rpc_type, and the peer the RPC was sent to in order to count |
|
// this RPC into the right stats bin. |
|
void RpcCompleted(AsyncClientCall* call, const std::string& peer) { |
|
// We count RPCs for global watcher or if the request_id falls into the |
|
// watcher's interested range of request ids. |
|
if ((start_id_ == 0 && end_id_ == 0) || |
|
(start_id_ <= call->saved_request_id && |
|
call->saved_request_id < end_id_)) { |
|
{ |
|
std::lock_guard<std::mutex> lock(m_); |
|
if (peer.empty()) { |
|
no_remote_peer_++; |
|
++no_remote_peer_by_type_[call->rpc_type]; |
|
} else { |
|
// RPC is counted into both per-peer bin and per-method-per-peer bin. |
|
rpcs_by_peer_[peer]++; |
|
rpcs_by_type_[call->rpc_type][peer]++; |
|
} |
|
rpcs_needed_--; |
|
// Report accumulated stats. |
|
auto& stats_per_method = *accumulated_stats_.mutable_stats_per_method(); |
|
auto& method_stat = |
|
stats_per_method[ClientConfigureRequest_RpcType_Name( |
|
call->rpc_type)]; |
|
auto& result = *method_stat.mutable_result(); |
|
grpc_status_code code = |
|
static_cast<grpc_status_code>(call->status.error_code()); |
|
auto& num_rpcs = result[code]; |
|
++num_rpcs; |
|
auto rpcs_started = method_stat.rpcs_started(); |
|
method_stat.set_rpcs_started(++rpcs_started); |
|
} |
|
cv_.notify_one(); |
|
} |
|
} |
|
|
|
void WaitForRpcStatsResponse(LoadBalancerStatsResponse* response, |
|
int timeout_sec) { |
|
std::unique_lock<std::mutex> lock(m_); |
|
cv_.wait_for(lock, std::chrono::seconds(timeout_sec), |
|
[this] { return rpcs_needed_ == 0; }); |
|
response->mutable_rpcs_by_peer()->insert(rpcs_by_peer_.begin(), |
|
rpcs_by_peer_.end()); |
|
auto& response_rpcs_by_method = *response->mutable_rpcs_by_method(); |
|
for (const auto& rpc_by_type : rpcs_by_type_) { |
|
std::string method_name; |
|
if (rpc_by_type.first == ClientConfigureRequest::EMPTY_CALL) { |
|
method_name = "EmptyCall"; |
|
} else if (rpc_by_type.first == ClientConfigureRequest::UNARY_CALL) { |
|
method_name = "UnaryCall"; |
|
} else { |
|
GPR_ASSERT(0); |
|
} |
|
// TODO(@donnadionne): When the test runner changes to accept EMPTY_CALL |
|
// and UNARY_CALL we will just use the name of the enum instead of the |
|
// method_name variable. |
|
auto& response_rpc_by_method = response_rpcs_by_method[method_name]; |
|
auto& response_rpcs_by_peer = |
|
*response_rpc_by_method.mutable_rpcs_by_peer(); |
|
for (const auto& rpc_by_peer : rpc_by_type.second) { |
|
auto& response_rpc_by_peer = response_rpcs_by_peer[rpc_by_peer.first]; |
|
response_rpc_by_peer = rpc_by_peer.second; |
|
} |
|
} |
|
response->set_num_failures(no_remote_peer_ + rpcs_needed_); |
|
} |
|
|
|
void GetCurrentRpcStats(LoadBalancerAccumulatedStatsResponse* response, |
|
StatsWatchers* stats_watchers) { |
|
std::unique_lock<std::mutex> lock(m_); |
|
response->CopyFrom(accumulated_stats_); |
|
// TODO(@donnadionne): delete deprecated stats below when the test is no |
|
// longer using them. |
|
auto& response_rpcs_started_by_method = |
|
*response->mutable_num_rpcs_started_by_method(); |
|
auto& response_rpcs_succeeded_by_method = |
|
*response->mutable_num_rpcs_succeeded_by_method(); |
|
auto& response_rpcs_failed_by_method = |
|
*response->mutable_num_rpcs_failed_by_method(); |
|
for (const auto& rpc_by_type : rpcs_by_type_) { |
|
auto total_succeeded = 0; |
|
for (const auto& rpc_by_peer : rpc_by_type.second) { |
|
total_succeeded += rpc_by_peer.second; |
|
} |
|
response_rpcs_succeeded_by_method[ClientConfigureRequest_RpcType_Name( |
|
rpc_by_type.first)] = total_succeeded; |
|
response_rpcs_started_by_method[ClientConfigureRequest_RpcType_Name( |
|
rpc_by_type.first)] = |
|
stats_watchers->global_request_id_by_type[rpc_by_type.first]; |
|
response_rpcs_failed_by_method[ClientConfigureRequest_RpcType_Name( |
|
rpc_by_type.first)] = no_remote_peer_by_type_[rpc_by_type.first]; |
|
} |
|
} |
|
|
|
private: |
|
int start_id_; |
|
int end_id_; |
|
int rpcs_needed_; |
|
int no_remote_peer_ = 0; |
|
std::map<int, int> no_remote_peer_by_type_; |
|
// A map of stats keyed by peer name. |
|
std::map<std::string, int> rpcs_by_peer_; |
|
// A two-level map of stats keyed at top level by RPC method and second level |
|
// by peer name. |
|
std::map<int, std::map<std::string, int>> rpcs_by_type_; |
|
// Storing accumulated stats in the response proto format. |
|
LoadBalancerAccumulatedStatsResponse accumulated_stats_; |
|
std::mutex m_; |
|
std::condition_variable cv_; |
|
}; |
|
|
|
class TestClient { |
|
public: |
|
TestClient(const std::shared_ptr<Channel>& channel, |
|
StatsWatchers* stats_watchers) |
|
: stub_(TestService::NewStub(channel)), stats_watchers_(stats_watchers) {} |
|
|
|
void AsyncUnaryCall(const RpcConfig& config) { |
|
SimpleResponse response; |
|
int saved_request_id; |
|
{ |
|
std::lock_guard<std::mutex> lock(stats_watchers_->mu); |
|
saved_request_id = ++stats_watchers_->global_request_id; |
|
++stats_watchers_ |
|
->global_request_id_by_type[ClientConfigureRequest::UNARY_CALL]; |
|
} |
|
std::chrono::system_clock::time_point deadline = |
|
std::chrono::system_clock::now() + |
|
std::chrono::seconds(config.timeout_sec != 0 |
|
? config.timeout_sec |
|
: absl::GetFlag(FLAGS_rpc_timeout_sec)); |
|
AsyncClientCall* call = new AsyncClientCall; |
|
for (const auto& data : config.metadata) { |
|
call->context.AddMetadata(data.first, data.second); |
|
// TODO(@donnadionne): move deadline to separate proto. |
|
if (data.first == "rpc-behavior" && data.second == "keep-open") { |
|
deadline = |
|
std::chrono::system_clock::now() + std::chrono::seconds(INT_MAX); |
|
} |
|
} |
|
call->context.set_deadline(deadline); |
|
call->saved_request_id = saved_request_id; |
|
call->rpc_type = ClientConfigureRequest::UNARY_CALL; |
|
call->simple_response_reader = stub_->PrepareAsyncUnaryCall( |
|
&call->context, SimpleRequest::default_instance(), &cq_); |
|
call->simple_response_reader->StartCall(); |
|
call->simple_response_reader->Finish(&call->simple_response, &call->status, |
|
call); |
|
} |
|
|
|
void AsyncEmptyCall(const RpcConfig& config) { |
|
Empty response; |
|
int saved_request_id; |
|
{ |
|
std::lock_guard<std::mutex> lock(stats_watchers_->mu); |
|
saved_request_id = ++stats_watchers_->global_request_id; |
|
++stats_watchers_ |
|
->global_request_id_by_type[ClientConfigureRequest::EMPTY_CALL]; |
|
} |
|
std::chrono::system_clock::time_point deadline = |
|
std::chrono::system_clock::now() + |
|
std::chrono::seconds(config.timeout_sec != 0 |
|
? config.timeout_sec |
|
: absl::GetFlag(FLAGS_rpc_timeout_sec)); |
|
AsyncClientCall* call = new AsyncClientCall; |
|
for (const auto& data : config.metadata) { |
|
call->context.AddMetadata(data.first, data.second); |
|
// TODO(@donnadionne): move deadline to separate proto. |
|
if (data.first == "rpc-behavior" && data.second == "keep-open") { |
|
deadline = |
|
std::chrono::system_clock::now() + std::chrono::seconds(INT_MAX); |
|
} |
|
} |
|
call->context.set_deadline(deadline); |
|
call->saved_request_id = saved_request_id; |
|
call->rpc_type = ClientConfigureRequest::EMPTY_CALL; |
|
call->empty_response_reader = stub_->PrepareAsyncEmptyCall( |
|
&call->context, Empty::default_instance(), &cq_); |
|
call->empty_response_reader->StartCall(); |
|
call->empty_response_reader->Finish(&call->empty_response, &call->status, |
|
call); |
|
} |
|
|
|
void AsyncCompleteRpc() { |
|
void* got_tag; |
|
bool ok = false; |
|
while (cq_.Next(&got_tag, &ok)) { |
|
AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag); |
|
GPR_ASSERT(ok); |
|
{ |
|
std::lock_guard<std::mutex> lock(stats_watchers_->mu); |
|
auto server_initial_metadata = call->context.GetServerInitialMetadata(); |
|
auto metadata_hostname = |
|
call->context.GetServerInitialMetadata().find("hostname"); |
|
std::string hostname = |
|
metadata_hostname != call->context.GetServerInitialMetadata().end() |
|
? std::string(metadata_hostname->second.data(), |
|
metadata_hostname->second.length()) |
|
: call->simple_response.hostname(); |
|
for (auto watcher : stats_watchers_->watchers) { |
|
watcher->RpcCompleted(call, hostname); |
|
} |
|
} |
|
|
|
if (!RpcStatusCheckSuccess(call)) { |
|
if (absl::GetFlag(FLAGS_print_response) || |
|
absl::GetFlag(FLAGS_fail_on_failed_rpc)) { |
|
std::cout << "RPC failed: " << call->status.error_code() << ": " |
|
<< call->status.error_message() << std::endl; |
|
} |
|
if (absl::GetFlag(FLAGS_fail_on_failed_rpc) && |
|
one_rpc_succeeded.load()) { |
|
abort(); |
|
} |
|
} else { |
|
if (absl::GetFlag(FLAGS_print_response)) { |
|
auto metadata_hostname = |
|
call->context.GetServerInitialMetadata().find("hostname"); |
|
std::string hostname = |
|
metadata_hostname != |
|
call->context.GetServerInitialMetadata().end() |
|
? std::string(metadata_hostname->second.data(), |
|
metadata_hostname->second.length()) |
|
: call->simple_response.hostname(); |
|
std::cout << "Greeting: Hello world, this is " << hostname |
|
<< ", from " << call->context.peer() << std::endl; |
|
} |
|
one_rpc_succeeded = true; |
|
} |
|
|
|
delete call; |
|
} |
|
} |
|
|
|
private: |
|
static bool RpcStatusCheckSuccess(AsyncClientCall* call) { |
|
// Determine RPC success based on expected status. |
|
grpc_status_code code; |
|
GPR_ASSERT(grpc_status_code_from_string( |
|
absl::GetFlag(FLAGS_expect_status).c_str(), &code)); |
|
return code == static_cast<grpc_status_code>(call->status.error_code()); |
|
} |
|
|
|
std::unique_ptr<TestService::Stub> stub_; |
|
StatsWatchers* stats_watchers_; |
|
CompletionQueue cq_; |
|
}; |
|
|
|
class LoadBalancerStatsServiceImpl : public LoadBalancerStatsService::Service { |
|
public: |
|
explicit LoadBalancerStatsServiceImpl(StatsWatchers* stats_watchers) |
|
: stats_watchers_(stats_watchers) {} |
|
|
|
Status GetClientStats(ServerContext* /*context*/, |
|
const LoadBalancerStatsRequest* request, |
|
LoadBalancerStatsResponse* response) override { |
|
int start_id; |
|
int end_id; |
|
XdsStatsWatcher* watcher; |
|
{ |
|
std::lock_guard<std::mutex> lock(stats_watchers_->mu); |
|
start_id = stats_watchers_->global_request_id + 1; |
|
end_id = start_id + request->num_rpcs(); |
|
watcher = new XdsStatsWatcher(start_id, end_id); |
|
stats_watchers_->watchers.insert(watcher); |
|
} |
|
watcher->WaitForRpcStatsResponse(response, request->timeout_sec()); |
|
{ |
|
std::lock_guard<std::mutex> lock(stats_watchers_->mu); |
|
stats_watchers_->watchers.erase(watcher); |
|
} |
|
delete watcher; |
|
return Status::OK; |
|
} |
|
|
|
Status GetClientAccumulatedStats( |
|
ServerContext* /*context*/, |
|
const LoadBalancerAccumulatedStatsRequest* /*request*/, |
|
LoadBalancerAccumulatedStatsResponse* response) override { |
|
std::lock_guard<std::mutex> lock(stats_watchers_->mu); |
|
stats_watchers_->global_watcher->GetCurrentRpcStats(response, |
|
stats_watchers_); |
|
return Status::OK; |
|
} |
|
|
|
private: |
|
StatsWatchers* stats_watchers_; |
|
}; |
|
|
|
class XdsUpdateClientConfigureServiceImpl |
|
: public XdsUpdateClientConfigureService::Service { |
|
public: |
|
explicit XdsUpdateClientConfigureServiceImpl( |
|
RpcConfigurationsQueue* rpc_configs_queue) |
|
: rpc_configs_queue_(rpc_configs_queue) {} |
|
|
|
Status Configure(ServerContext* /*context*/, |
|
const ClientConfigureRequest* request, |
|
ClientConfigureResponse* /*response*/) override { |
|
std::map<int, std::vector<std::pair<std::string, std::string>>> |
|
metadata_map; |
|
for (const auto& data : request->metadata()) { |
|
metadata_map[data.type()].push_back({data.key(), data.value()}); |
|
} |
|
std::vector<RpcConfig> configs; |
|
for (const auto& rpc : request->types()) { |
|
RpcConfig config; |
|
config.timeout_sec = request->timeout_sec(); |
|
config.type = static_cast<ClientConfigureRequest::RpcType>(rpc); |
|
auto metadata_iter = metadata_map.find(rpc); |
|
if (metadata_iter != metadata_map.end()) { |
|
config.metadata = metadata_iter->second; |
|
} |
|
configs.push_back(std::move(config)); |
|
} |
|
{ |
|
std::lock_guard<std::mutex> lock( |
|
rpc_configs_queue_->mu_rpc_configs_queue); |
|
rpc_configs_queue_->rpc_configs_queue.emplace_back(std::move(configs)); |
|
} |
|
return Status::OK; |
|
} |
|
|
|
private: |
|
RpcConfigurationsQueue* rpc_configs_queue_; |
|
}; |
|
|
|
void RunTestLoop(std::chrono::duration<double> duration_per_query, |
|
StatsWatchers* stats_watchers, |
|
RpcConfigurationsQueue* rpc_configs_queue) { |
|
grpc::ChannelArguments channel_args; |
|
channel_args.SetInt(GRPC_ARG_ENABLE_RETRIES, 1); |
|
TestClient client( |
|
grpc::CreateCustomChannel(absl::GetFlag(FLAGS_server), |
|
absl::GetFlag(FLAGS_secure_mode) |
|
? grpc::experimental::XdsCredentials( |
|
grpc::InsecureChannelCredentials()) |
|
: grpc::InsecureChannelCredentials(), |
|
channel_args), |
|
stats_watchers); |
|
std::chrono::time_point<std::chrono::system_clock> start = |
|
std::chrono::system_clock::now(); |
|
std::chrono::duration<double> elapsed; |
|
|
|
std::thread thread = std::thread(&TestClient::AsyncCompleteRpc, &client); |
|
|
|
std::vector<RpcConfig> configs; |
|
while (true) { |
|
{ |
|
std::lock_guard<std::mutex> lockk( |
|
rpc_configs_queue->mu_rpc_configs_queue); |
|
if (!rpc_configs_queue->rpc_configs_queue.empty()) { |
|
configs = std::move(rpc_configs_queue->rpc_configs_queue.front()); |
|
rpc_configs_queue->rpc_configs_queue.pop_front(); |
|
} |
|
} |
|
|
|
elapsed = std::chrono::system_clock::now() - start; |
|
if (elapsed > duration_per_query) { |
|
start = std::chrono::system_clock::now(); |
|
for (const auto& config : configs) { |
|
if (config.type == ClientConfigureRequest::EMPTY_CALL) { |
|
client.AsyncEmptyCall(config); |
|
} else if (config.type == ClientConfigureRequest::UNARY_CALL) { |
|
client.AsyncUnaryCall(config); |
|
} else { |
|
GPR_ASSERT(0); |
|
} |
|
} |
|
} |
|
} |
|
GPR_UNREACHABLE_CODE(thread.join()); |
|
} |
|
|
|
void RunServer(const int port, StatsWatchers* stats_watchers, |
|
RpcConfigurationsQueue* rpc_configs_queue) { |
|
GPR_ASSERT(port != 0); |
|
std::ostringstream server_address; |
|
server_address << "0.0.0.0:" << port; |
|
|
|
LoadBalancerStatsServiceImpl stats_service(stats_watchers); |
|
XdsUpdateClientConfigureServiceImpl client_config_service(rpc_configs_queue); |
|
|
|
grpc::reflection::InitProtoReflectionServerBuilderPlugin(); |
|
ServerBuilder builder; |
|
builder.RegisterService(&stats_service); |
|
builder.RegisterService(&client_config_service); |
|
grpc::AddAdminServices(&builder); |
|
builder.AddListeningPort(server_address.str(), |
|
grpc::InsecureServerCredentials()); |
|
std::unique_ptr<Server> server(builder.BuildAndStart()); |
|
gpr_log(GPR_DEBUG, "Server listening on %s", server_address.str().c_str()); |
|
|
|
server->Wait(); |
|
} |
|
|
|
void BuildRpcConfigsFromFlags(RpcConfigurationsQueue* rpc_configs_queue) { |
|
// Store Metadata like |
|
// "EmptyCall:key1:value1,UnaryCall:key1:value1,UnaryCall:key2:value2" into a |
|
// map where the key is the RPC method and value is a vector of key:value |
|
// pairs. {EmptyCall, [{key1,value1}], |
|
// UnaryCall, [{key1,value1}, {key2,value2}]} |
|
std::vector<std::string> rpc_metadata = |
|
absl::StrSplit(absl::GetFlag(FLAGS_metadata), ',', absl::SkipEmpty()); |
|
std::map<int, std::vector<std::pair<std::string, std::string>>> metadata_map; |
|
for (auto& data : rpc_metadata) { |
|
std::vector<std::string> metadata = |
|
absl::StrSplit(data, ':', absl::SkipEmpty()); |
|
GPR_ASSERT(metadata.size() == 3); |
|
if (metadata[0] == "EmptyCall") { |
|
metadata_map[ClientConfigureRequest::EMPTY_CALL].push_back( |
|
{metadata[1], metadata[2]}); |
|
} else if (metadata[0] == "UnaryCall") { |
|
metadata_map[ClientConfigureRequest::UNARY_CALL].push_back( |
|
{metadata[1], metadata[2]}); |
|
} else { |
|
GPR_ASSERT(0); |
|
} |
|
} |
|
std::vector<RpcConfig> configs; |
|
std::vector<std::string> rpc_methods = |
|
absl::StrSplit(absl::GetFlag(FLAGS_rpc), ',', absl::SkipEmpty()); |
|
for (const std::string& rpc_method : rpc_methods) { |
|
RpcConfig config; |
|
if (rpc_method == "EmptyCall") { |
|
config.type = ClientConfigureRequest::EMPTY_CALL; |
|
} else if (rpc_method == "UnaryCall") { |
|
config.type = ClientConfigureRequest::UNARY_CALL; |
|
} else { |
|
GPR_ASSERT(0); |
|
} |
|
auto metadata_iter = metadata_map.find(config.type); |
|
if (metadata_iter != metadata_map.end()) { |
|
config.metadata = metadata_iter->second; |
|
} |
|
configs.push_back(std::move(config)); |
|
} |
|
{ |
|
std::lock_guard<std::mutex> lock(rpc_configs_queue->mu_rpc_configs_queue); |
|
rpc_configs_queue->rpc_configs_queue.emplace_back(std::move(configs)); |
|
} |
|
} |
|
|
|
int main(int argc, char** argv) { |
|
grpc::testing::TestEnvironment env(&argc, argv); |
|
grpc::testing::InitTest(&argc, &argv, true); |
|
// Validate the expect_status flag. |
|
grpc_status_code code; |
|
GPR_ASSERT(grpc_status_code_from_string( |
|
absl::GetFlag(FLAGS_expect_status).c_str(), &code)); |
|
StatsWatchers stats_watchers; |
|
RpcConfigurationsQueue rpc_config_queue; |
|
|
|
{ |
|
std::lock_guard<std::mutex> lock(stats_watchers.mu); |
|
stats_watchers.global_watcher = new XdsStatsWatcher(0, 0); |
|
stats_watchers.watchers.insert(stats_watchers.global_watcher); |
|
} |
|
|
|
BuildRpcConfigsFromFlags(&rpc_config_queue); |
|
|
|
std::chrono::duration<double> duration_per_query = |
|
std::chrono::nanoseconds(std::chrono::seconds(1)) / |
|
absl::GetFlag(FLAGS_qps); |
|
|
|
std::vector<std::thread> test_threads; |
|
test_threads.reserve(absl::GetFlag(FLAGS_num_channels)); |
|
for (int i = 0; i < absl::GetFlag(FLAGS_num_channels); i++) { |
|
test_threads.emplace_back(std::thread(&RunTestLoop, duration_per_query, |
|
&stats_watchers, &rpc_config_queue)); |
|
} |
|
|
|
RunServer(absl::GetFlag(FLAGS_stats_port), &stats_watchers, |
|
&rpc_config_queue); |
|
|
|
for (auto it = test_threads.begin(); it != test_threads.end(); it++) { |
|
it->join(); |
|
} |
|
|
|
return 0; |
|
}
|
|
|