mirror of https://github.com/grpc/grpc.git
Merge branch 'master' of https://github.com/grpc/grpc into more-eager-free
ncteisen
7 years ago
commit
50cfbe358e
22 changed files with 1213 additions and 277 deletions
@ -0,0 +1,103 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016 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 <arpa/inet.h> |
||||
#include <openssl/err.h> |
||||
#include <openssl/ssl.h> |
||||
#include <string.h> |
||||
#include <sys/socket.h> |
||||
#include <unistd.h> |
||||
|
||||
#include <grpc/grpc.h> |
||||
#include <grpc/grpc_security.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
#include <grpc/support/sync.h> |
||||
#include <grpc/support/thd.h> |
||||
#include "src/core/lib/iomgr/load_file.h" |
||||
#include "test/core/util/port.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
#include "src/core/lib/channel/handshaker_factory.h" |
||||
#include "src/core/lib/channel/handshaker_registry.h" |
||||
#include "src/core/lib/security/transport/security_handshaker.h" |
||||
|
||||
#include "test/core/handshake/server_ssl_common.h" |
||||
|
||||
/* The purpose of this test is to exercise the case when a
|
||||
* grpc *security_handshaker* begins its handshake with data already |
||||
* in the read buffer of the handshaker arg. This scenario is created by |
||||
* adding a fake "readahead" handshaker at the beginning of the server's |
||||
* handshaker list, which just reads from the connection and then places |
||||
* read bytes into the read buffer of the handshake arg (to be passed down |
||||
* to the security_handshaker). This test is meant to protect code relying on |
||||
* this functionality that lives outside of this repo. */ |
||||
|
||||
static void readahead_handshaker_destroy(grpc_exec_ctx* ctx, |
||||
grpc_handshaker* handshaker) { |
||||
gpr_free(handshaker); |
||||
} |
||||
|
||||
static void readahead_handshaker_shutdown(grpc_exec_ctx* ctx, |
||||
grpc_handshaker* handshaker, |
||||
grpc_error* error) {} |
||||
|
||||
static void readahead_handshaker_do_handshake( |
||||
grpc_exec_ctx* ctx, grpc_handshaker* handshaker, |
||||
grpc_tcp_server_acceptor* acceptor, grpc_closure* on_handshake_done, |
||||
grpc_handshaker_args* args) { |
||||
grpc_endpoint_read(ctx, args->endpoint, args->read_buffer, on_handshake_done); |
||||
} |
||||
|
||||
const grpc_handshaker_vtable readahead_handshaker_vtable = { |
||||
readahead_handshaker_destroy, readahead_handshaker_shutdown, |
||||
readahead_handshaker_do_handshake}; |
||||
|
||||
static grpc_handshaker* readahead_handshaker_create(grpc_exec_ctx* ctx) { |
||||
grpc_handshaker* h = (grpc_handshaker*)gpr_zalloc(sizeof(grpc_handshaker)); |
||||
grpc_handshaker_init(&readahead_handshaker_vtable, h); |
||||
return h; |
||||
} |
||||
|
||||
static void readahead_handshaker_factory_add_handshakers( |
||||
grpc_exec_ctx* exec_ctx, grpc_handshaker_factory* hf, |
||||
const grpc_channel_args* args, grpc_handshake_manager* handshake_mgr) { |
||||
grpc_handshake_manager_add(handshake_mgr, |
||||
readahead_handshaker_create(exec_ctx)); |
||||
} |
||||
|
||||
static void readahead_handshaker_factory_destroy( |
||||
grpc_exec_ctx* exec_ctx, grpc_handshaker_factory* handshaker_factory) {} |
||||
|
||||
static const grpc_handshaker_factory_vtable |
||||
readahead_handshaker_factory_vtable = { |
||||
readahead_handshaker_factory_add_handshakers, |
||||
readahead_handshaker_factory_destroy}; |
||||
|
||||
int main(int argc, char* argv[]) { |
||||
grpc_handshaker_factory readahead_handshaker_factory = { |
||||
&readahead_handshaker_factory_vtable}; |
||||
grpc_init(); |
||||
grpc_handshaker_factory_register(true /* at_start */, HANDSHAKER_SERVER, |
||||
&readahead_handshaker_factory); |
||||
const char* full_alpn_list[] = {"grpc-exp", "h2"}; |
||||
GPR_ASSERT(server_ssl_test(full_alpn_list, 2, "grpc-exp")); |
||||
grpc_shutdown(); |
||||
return 0; |
||||
} |
||||
@ -0,0 +1,238 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016 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 "test/core/handshake/server_ssl_common.h" |
||||
|
||||
#include <arpa/inet.h> |
||||
#include <openssl/err.h> |
||||
#include <openssl/ssl.h> |
||||
#include <string.h> |
||||
#include <sys/socket.h> |
||||
#include <unistd.h> |
||||
|
||||
#include <grpc/grpc.h> |
||||
#include <grpc/grpc_security.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
#include <grpc/support/sync.h> |
||||
#include <grpc/support/thd.h> |
||||
#include "src/core/lib/iomgr/load_file.h" |
||||
#include "test/core/util/port.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
#define SSL_CERT_PATH "src/core/tsi/test_creds/server1.pem" |
||||
#define SSL_KEY_PATH "src/core/tsi/test_creds/server1.key" |
||||
#define SSL_CA_PATH "src/core/tsi/test_creds/ca.pem" |
||||
|
||||
// Handshake completed signal to server thread.
|
||||
static gpr_event client_handshake_complete; |
||||
|
||||
static int create_socket(int port) { |
||||
int s; |
||||
struct sockaddr_in addr; |
||||
|
||||
addr.sin_family = AF_INET; |
||||
addr.sin_port = htons((uint16_t)port); |
||||
addr.sin_addr.s_addr = htonl(INADDR_ANY); |
||||
|
||||
s = socket(AF_INET, SOCK_STREAM, 0); |
||||
if (s < 0) { |
||||
perror("Unable to create socket"); |
||||
return -1; |
||||
} |
||||
|
||||
if (connect(s, (struct sockaddr*)&addr, sizeof(addr)) < 0) { |
||||
perror("Unable to connect"); |
||||
return -1; |
||||
} |
||||
|
||||
return s; |
||||
} |
||||
|
||||
// Simple gRPC server. This listens until client_handshake_complete occurs.
|
||||
static void server_thread(void* arg) { |
||||
const int port = *(int*)arg; |
||||
|
||||
// Load key pair and establish server SSL credentials.
|
||||
grpc_ssl_pem_key_cert_pair pem_key_cert_pair; |
||||
grpc_slice ca_slice, cert_slice, key_slice; |
||||
GPR_ASSERT(GRPC_LOG_IF_ERROR("load_file", |
||||
grpc_load_file(SSL_CA_PATH, 1, &ca_slice))); |
||||
GPR_ASSERT(GRPC_LOG_IF_ERROR("load_file", |
||||
grpc_load_file(SSL_CERT_PATH, 1, &cert_slice))); |
||||
GPR_ASSERT(GRPC_LOG_IF_ERROR("load_file", |
||||
grpc_load_file(SSL_KEY_PATH, 1, &key_slice))); |
||||
const char* ca_cert = (const char*)GRPC_SLICE_START_PTR(ca_slice); |
||||
pem_key_cert_pair.private_key = (const char*)GRPC_SLICE_START_PTR(key_slice); |
||||
pem_key_cert_pair.cert_chain = (const char*)GRPC_SLICE_START_PTR(cert_slice); |
||||
grpc_server_credentials* ssl_creds = grpc_ssl_server_credentials_create( |
||||
ca_cert, &pem_key_cert_pair, 1, 0, nullptr); |
||||
|
||||
// Start server listening on local port.
|
||||
char* addr; |
||||
gpr_asprintf(&addr, "127.0.0.1:%d", port); |
||||
grpc_server* server = grpc_server_create(nullptr, nullptr); |
||||
GPR_ASSERT(grpc_server_add_secure_http2_port(server, addr, ssl_creds)); |
||||
free(addr); |
||||
|
||||
grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr); |
||||
|
||||
grpc_server_register_completion_queue(server, cq, nullptr); |
||||
grpc_server_start(server); |
||||
|
||||
// Wait a bounded number of time until client_handshake_complete is set,
|
||||
// sleeping between polls.
|
||||
int retries = 10; |
||||
while (!gpr_event_get(&client_handshake_complete) && retries-- > 0) { |
||||
const gpr_timespec cq_deadline = grpc_timeout_seconds_to_deadline(1); |
||||
grpc_event ev = grpc_completion_queue_next(cq, cq_deadline, nullptr); |
||||
GPR_ASSERT(ev.type == GRPC_QUEUE_TIMEOUT); |
||||
} |
||||
|
||||
gpr_log(GPR_INFO, "Shutting down server"); |
||||
grpc_server_shutdown_and_notify(server, cq, nullptr); |
||||
grpc_completion_queue_shutdown(cq); |
||||
|
||||
const gpr_timespec cq_deadline = grpc_timeout_seconds_to_deadline(5); |
||||
grpc_event ev = grpc_completion_queue_next(cq, cq_deadline, nullptr); |
||||
GPR_ASSERT(ev.type == GRPC_OP_COMPLETE); |
||||
|
||||
grpc_server_destroy(server); |
||||
grpc_completion_queue_destroy(cq); |
||||
grpc_server_credentials_release(ssl_creds); |
||||
grpc_slice_unref(cert_slice); |
||||
grpc_slice_unref(key_slice); |
||||
grpc_slice_unref(ca_slice); |
||||
} |
||||
|
||||
// This test launches a gRPC server on a separate thread and then establishes a
|
||||
// TLS handshake via a minimal TLS client. The TLS client has configurable (via
|
||||
// alpn_list) ALPN settings and can probe at the supported ALPN preferences
|
||||
// using this (via alpn_expected).
|
||||
bool server_ssl_test(const char* alpn_list[], unsigned int alpn_list_len, |
||||
const char* alpn_expected) { |
||||
bool success = true; |
||||
|
||||
grpc_init(); |
||||
int port = grpc_pick_unused_port_or_die(); |
||||
gpr_event_init(&client_handshake_complete); |
||||
|
||||
// Launch the gRPC server thread.
|
||||
gpr_thd_options thdopt = gpr_thd_options_default(); |
||||
gpr_thd_id thdid; |
||||
gpr_thd_options_set_joinable(&thdopt); |
||||
GPR_ASSERT(gpr_thd_new(&thdid, server_thread, &port, &thdopt)); |
||||
|
||||
SSL_load_error_strings(); |
||||
OpenSSL_add_ssl_algorithms(); |
||||
|
||||
const SSL_METHOD* method = TLSv1_2_client_method(); |
||||
SSL_CTX* ctx = SSL_CTX_new(method); |
||||
if (!ctx) { |
||||
perror("Unable to create SSL context"); |
||||
ERR_print_errors_fp(stderr); |
||||
abort(); |
||||
} |
||||
|
||||
// Load key pair.
|
||||
if (SSL_CTX_use_certificate_file(ctx, SSL_CERT_PATH, SSL_FILETYPE_PEM) < 0) { |
||||
ERR_print_errors_fp(stderr); |
||||
abort(); |
||||
} |
||||
if (SSL_CTX_use_PrivateKey_file(ctx, SSL_KEY_PATH, SSL_FILETYPE_PEM) < 0) { |
||||
ERR_print_errors_fp(stderr); |
||||
abort(); |
||||
} |
||||
|
||||
// Set the cipher list to match the one expressed in
|
||||
// src/core/tsi/ssl_transport_security.c.
|
||||
const char* cipher_list = |
||||
"ECDHE-RSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-SHA256:ECDHE-RSA-AES256-" |
||||
"SHA384:ECDHE-RSA-AES256-GCM-SHA384"; |
||||
if (!SSL_CTX_set_cipher_list(ctx, cipher_list)) { |
||||
ERR_print_errors_fp(stderr); |
||||
gpr_log(GPR_ERROR, "Couldn't set server cipher list."); |
||||
abort(); |
||||
} |
||||
|
||||
// Configure ALPN list the client will send to the server. This must match the
|
||||
// wire format, see documentation for SSL_CTX_set_alpn_protos.
|
||||
unsigned int alpn_protos_len = alpn_list_len; |
||||
for (unsigned int i = 0; i < alpn_list_len; ++i) { |
||||
alpn_protos_len += (unsigned int)strlen(alpn_list[i]); |
||||
} |
||||
unsigned char* alpn_protos = |
||||
static_cast<unsigned char*>(gpr_malloc(alpn_protos_len)); |
||||
unsigned char* p = alpn_protos; |
||||
for (unsigned int i = 0; i < alpn_list_len; ++i) { |
||||
const uint8_t len = (uint8_t)strlen(alpn_list[i]); |
||||
*p++ = len; |
||||
memcpy(p, alpn_list[i], len); |
||||
p += len; |
||||
} |
||||
GPR_ASSERT(SSL_CTX_set_alpn_protos(ctx, alpn_protos, alpn_protos_len) == 0); |
||||
|
||||
// Try and connect to server. We allow a bounded number of retries as we might
|
||||
// be racing with the server setup on its separate thread.
|
||||
int retries = 10; |
||||
int sock = -1; |
||||
while (sock == -1 && retries-- > 0) { |
||||
sock = create_socket(port); |
||||
if (sock < 0) { |
||||
sleep(1); |
||||
} |
||||
} |
||||
GPR_ASSERT(sock > 0); |
||||
gpr_log(GPR_INFO, "Connected to server on port %d", port); |
||||
|
||||
// Establish a SSL* and connect at SSL layer.
|
||||
SSL* ssl = SSL_new(ctx); |
||||
GPR_ASSERT(ssl); |
||||
SSL_set_fd(ssl, sock); |
||||
if (SSL_connect(ssl) <= 0) { |
||||
ERR_print_errors_fp(stderr); |
||||
gpr_log(GPR_ERROR, "Handshake failed."); |
||||
success = false; |
||||
} else { |
||||
gpr_log(GPR_INFO, "Handshake successful."); |
||||
// Validate ALPN preferred by server matches alpn_expected.
|
||||
const unsigned char* alpn_selected; |
||||
unsigned int alpn_selected_len; |
||||
SSL_get0_alpn_selected(ssl, &alpn_selected, &alpn_selected_len); |
||||
if (strlen(alpn_expected) != alpn_selected_len || |
||||
strncmp((const char*)alpn_selected, alpn_expected, alpn_selected_len) != |
||||
0) { |
||||
gpr_log(GPR_ERROR, "Unexpected ALPN protocol preference"); |
||||
success = false; |
||||
} |
||||
} |
||||
gpr_event_set(&client_handshake_complete, &client_handshake_complete); |
||||
|
||||
SSL_free(ssl); |
||||
gpr_free(alpn_protos); |
||||
SSL_CTX_free(ctx); |
||||
EVP_cleanup(); |
||||
close(sock); |
||||
|
||||
gpr_thd_join(thdid); |
||||
|
||||
grpc_shutdown(); |
||||
|
||||
return success; |
||||
} |
||||
@ -0,0 +1,36 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016 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. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_SERVER_SSL_COMMON_H |
||||
#define GRPC_SERVER_SSL_COMMON_H |
||||
|
||||
#include <grpc/grpc.h> |
||||
#include <grpc/grpc_security.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
#include <grpc/support/sync.h> |
||||
#include <grpc/support/thd.h> |
||||
#include "src/core/lib/iomgr/load_file.h" |
||||
#include "test/core/util/port.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
bool server_ssl_test(const char* alpn_list[], unsigned int alpn_list_len, |
||||
const char* alpn_expected); |
||||
|
||||
#endif // GRPC_SERVER_SSL_COMMON_H
|
||||
@ -0,0 +1,51 @@ |
||||
# Copyright 2017 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. |
||||
|
||||
licenses(["notice"]) # Apache v2 |
||||
|
||||
load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_cc_test", "grpc_package") |
||||
|
||||
grpc_package(name = "test/cpp/client") |
||||
|
||||
grpc_cc_test( |
||||
name = "credentials_test", |
||||
srcs = ["credentials_test.cc"], |
||||
external_deps = [ |
||||
"gtest", |
||||
], |
||||
deps = [ |
||||
"//:gpr", |
||||
"//:grpc", |
||||
"//:grpc++", |
||||
], |
||||
) |
||||
|
||||
grpc_cc_test( |
||||
name = "client_channel_stress_test", |
||||
srcs = ["client_channel_stress_test.cc"], |
||||
deps = [ |
||||
"//:gpr", |
||||
"//:grpc", |
||||
"//:grpc++", |
||||
"//:grpc_resolver_fake", |
||||
"//src/proto/grpc/lb/v1:load_balancer_proto", |
||||
"//src/proto/grpc/testing:echo_messages_proto", |
||||
"//src/proto/grpc/testing:echo_proto", |
||||
"//src/proto/grpc/testing/duplicate:echo_duplicate_proto", |
||||
"//test/core/util:gpr_test_util", |
||||
"//test/core/util:grpc_test_util", |
||||
"//test/cpp/end2end:test_service_impl", |
||||
"//test/cpp/util:test_util", |
||||
], |
||||
) |
||||
@ -0,0 +1,329 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2017 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 <memory> |
||||
#include <mutex> |
||||
#include <sstream> |
||||
#include <thread> |
||||
|
||||
#include <grpc++/channel.h> |
||||
#include <grpc++/client_context.h> |
||||
#include <grpc++/create_channel.h> |
||||
#include <grpc++/server.h> |
||||
#include <grpc++/server_builder.h> |
||||
#include <grpc/grpc.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
#include <grpc/support/thd.h> |
||||
#include <grpc/support/time.h> |
||||
|
||||
extern "C" { |
||||
#include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h" |
||||
#include "src/core/lib/iomgr/sockaddr.h" |
||||
} |
||||
|
||||
#include "test/core/util/port.h" |
||||
#include "test/core/util/test_config.h" |
||||
#include "test/cpp/end2end/test_service_impl.h" |
||||
|
||||
#include "src/proto/grpc/lb/v1/load_balancer.grpc.pb.h" |
||||
#include "src/proto/grpc/testing/echo.grpc.pb.h" |
||||
|
||||
using grpc::lb::v1::LoadBalanceRequest; |
||||
using grpc::lb::v1::LoadBalanceResponse; |
||||
using grpc::lb::v1::LoadBalancer; |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
namespace { |
||||
|
||||
const size_t kNumBackends = 10; |
||||
const size_t kNumBalancers = 5; |
||||
const size_t kNumClientThreads = 100; |
||||
const int kResolutionUpdateIntervalMs = 50; |
||||
const int kServerlistUpdateIntervalMs = 10; |
||||
const int kTestDurationSec = 30; |
||||
|
||||
using BackendServiceImpl = TestServiceImpl; |
||||
|
||||
class BalancerServiceImpl : public LoadBalancer::Service { |
||||
public: |
||||
using Stream = ServerReaderWriter<LoadBalanceResponse, LoadBalanceRequest>; |
||||
|
||||
explicit BalancerServiceImpl(const std::vector<int>& all_backend_ports) |
||||
: all_backend_ports_(all_backend_ports) {} |
||||
|
||||
Status BalanceLoad(ServerContext* context, Stream* stream) override { |
||||
gpr_log(GPR_INFO, "LB[%p]: Start BalanceLoad.", this); |
||||
LoadBalanceRequest request; |
||||
stream->Read(&request); |
||||
while (!shutdown_) { |
||||
stream->Write(BuildRandomResponseForBackends()); |
||||
std::this_thread::sleep_for( |
||||
std::chrono::milliseconds(kServerlistUpdateIntervalMs)); |
||||
} |
||||
gpr_log(GPR_INFO, "LB[%p]: Finish BalanceLoad.", this); |
||||
return Status::OK; |
||||
} |
||||
|
||||
void Shutdown() { shutdown_ = true; } |
||||
|
||||
private: |
||||
grpc::string Ip4ToPackedString(const char* ip_str) { |
||||
struct in_addr ip4; |
||||
GPR_ASSERT(inet_pton(AF_INET, ip_str, &ip4) == 1); |
||||
return grpc::string(reinterpret_cast<const char*>(&ip4), sizeof(ip4)); |
||||
} |
||||
|
||||
LoadBalanceResponse BuildRandomResponseForBackends() { |
||||
// Generate a random serverlist with varying size (if N =
|
||||
// all_backend_ports_.size(), num_non_drop_entry is in [0, 2N],
|
||||
// num_drop_entry is in [0, N]), order, duplicate, and drop rate.
|
||||
size_t num_non_drop_entry = |
||||
std::rand() % (all_backend_ports_.size() * 2 + 1); |
||||
size_t num_drop_entry = std::rand() % (all_backend_ports_.size() + 1); |
||||
std::vector<int> random_backend_indices; |
||||
for (size_t i = 0; i < num_non_drop_entry; ++i) { |
||||
random_backend_indices.push_back(std::rand() % all_backend_ports_.size()); |
||||
} |
||||
for (size_t i = 0; i < num_drop_entry; ++i) { |
||||
random_backend_indices.push_back(-1); |
||||
} |
||||
std::random_shuffle(random_backend_indices.begin(), |
||||
random_backend_indices.end()); |
||||
// Build the response according to the random list generated above.
|
||||
LoadBalanceResponse response; |
||||
for (int index : random_backend_indices) { |
||||
auto* server = response.mutable_server_list()->add_servers(); |
||||
if (index < 0) { |
||||
server->set_drop(true); |
||||
server->set_load_balance_token("load_balancing"); |
||||
} else { |
||||
server->set_ip_address(Ip4ToPackedString("127.0.0.1")); |
||||
server->set_port(all_backend_ports_[index]); |
||||
} |
||||
} |
||||
return response; |
||||
} |
||||
|
||||
std::atomic_bool shutdown_{false}; |
||||
const std::vector<int> all_backend_ports_; |
||||
}; |
||||
|
||||
class ClientChannelStressTest { |
||||
public: |
||||
void Run() { |
||||
Start(); |
||||
// Keep updating resolution for the test duration.
|
||||
gpr_log(GPR_INFO, "Start updating resolution."); |
||||
const auto wait_duration = |
||||
std::chrono::milliseconds(kResolutionUpdateIntervalMs); |
||||
std::vector<AddressData> addresses; |
||||
auto start_time = std::chrono::steady_clock::now(); |
||||
while (true) { |
||||
if (std::chrono::duration_cast<std::chrono::seconds>( |
||||
std::chrono::steady_clock::now() - start_time) |
||||
.count() > kTestDurationSec) { |
||||
break; |
||||
} |
||||
// Generate a random subset of balancers.
|
||||
addresses.clear(); |
||||
for (const auto& balancer_server : balancer_servers_) { |
||||
// Select each address with probability of 0.8.
|
||||
if (std::rand() % 10 < 8) { |
||||
addresses.emplace_back(AddressData{balancer_server.port_, true, ""}); |
||||
} |
||||
} |
||||
std::random_shuffle(addresses.begin(), addresses.end()); |
||||
SetNextResolution(addresses); |
||||
std::this_thread::sleep_for(wait_duration); |
||||
} |
||||
gpr_log(GPR_INFO, "Finish updating resolution."); |
||||
Shutdown(); |
||||
} |
||||
|
||||
private: |
||||
template <typename T> |
||||
struct ServerThread { |
||||
explicit ServerThread(const grpc::string& type, |
||||
const grpc::string& server_host, T* service) |
||||
: type_(type), service_(service) { |
||||
std::mutex mu; |
||||
// We need to acquire the lock here in order to prevent the notify_one
|
||||
// by ServerThread::Start from firing before the wait below is hit.
|
||||
std::unique_lock<std::mutex> lock(mu); |
||||
port_ = grpc_pick_unused_port_or_die(); |
||||
gpr_log(GPR_INFO, "starting %s server on port %d", type_.c_str(), port_); |
||||
std::condition_variable cond; |
||||
thread_.reset(new std::thread( |
||||
std::bind(&ServerThread::Start, this, server_host, &mu, &cond))); |
||||
cond.wait(lock); |
||||
gpr_log(GPR_INFO, "%s server startup complete", type_.c_str()); |
||||
} |
||||
|
||||
void Start(const grpc::string& server_host, std::mutex* mu, |
||||
std::condition_variable* cond) { |
||||
// We need to acquire the lock here in order to prevent the notify_one
|
||||
// below from firing before its corresponding wait is executed.
|
||||
std::lock_guard<std::mutex> lock(*mu); |
||||
std::ostringstream server_address; |
||||
server_address << server_host << ":" << port_; |
||||
ServerBuilder builder; |
||||
builder.AddListeningPort(server_address.str(), |
||||
InsecureServerCredentials()); |
||||
builder.RegisterService(service_); |
||||
server_ = builder.BuildAndStart(); |
||||
cond->notify_one(); |
||||
} |
||||
|
||||
void Shutdown() { |
||||
gpr_log(GPR_INFO, "%s about to shutdown", type_.c_str()); |
||||
server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0)); |
||||
thread_->join(); |
||||
gpr_log(GPR_INFO, "%s shutdown completed", type_.c_str()); |
||||
} |
||||
|
||||
int port_; |
||||
grpc::string type_; |
||||
std::unique_ptr<Server> server_; |
||||
T* service_; |
||||
std::unique_ptr<std::thread> thread_; |
||||
}; |
||||
|
||||
struct AddressData { |
||||
int port; |
||||
bool is_balancer; |
||||
grpc::string balancer_name; |
||||
}; |
||||
|
||||
void SetNextResolution(const std::vector<AddressData>& address_data) { |
||||
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; |
||||
grpc_lb_addresses* addresses = |
||||
grpc_lb_addresses_create(address_data.size(), nullptr); |
||||
for (size_t i = 0; i < address_data.size(); ++i) { |
||||
char* lb_uri_str; |
||||
gpr_asprintf(&lb_uri_str, "ipv4:127.0.0.1:%d", address_data[i].port); |
||||
grpc_uri* lb_uri = grpc_uri_parse(&exec_ctx, lb_uri_str, true); |
||||
GPR_ASSERT(lb_uri != nullptr); |
||||
grpc_lb_addresses_set_address_from_uri( |
||||
addresses, i, lb_uri, address_data[i].is_balancer, |
||||
address_data[i].balancer_name.c_str(), nullptr); |
||||
grpc_uri_destroy(lb_uri); |
||||
gpr_free(lb_uri_str); |
||||
} |
||||
grpc_arg fake_addresses = grpc_lb_addresses_create_channel_arg(addresses); |
||||
grpc_channel_args fake_result = {1, &fake_addresses}; |
||||
grpc_fake_resolver_response_generator_set_response( |
||||
&exec_ctx, response_generator_, &fake_result); |
||||
grpc_lb_addresses_destroy(&exec_ctx, addresses); |
||||
grpc_exec_ctx_finish(&exec_ctx); |
||||
} |
||||
|
||||
void KeepSendingRequests() { |
||||
gpr_log(GPR_INFO, "Start sending requests."); |
||||
while (!shutdown_) { |
||||
ClientContext context; |
||||
context.set_deadline(grpc_timeout_milliseconds_to_deadline(1000)); |
||||
EchoRequest request; |
||||
request.set_message("test"); |
||||
EchoResponse response; |
||||
{ |
||||
std::lock_guard<std::mutex> lock(stub_mutex_); |
||||
stub_->Echo(&context, request, &response); |
||||
} |
||||
} |
||||
gpr_log(GPR_INFO, "Finish sending requests."); |
||||
} |
||||
|
||||
void CreateStub() { |
||||
ChannelArguments args; |
||||
response_generator_ = grpc_fake_resolver_response_generator_create(); |
||||
args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR, |
||||
response_generator_); |
||||
std::ostringstream uri; |
||||
uri << "fake:///servername_not_used"; |
||||
channel_ = |
||||
CreateCustomChannel(uri.str(), InsecureChannelCredentials(), args); |
||||
stub_ = grpc::testing::EchoTestService::NewStub(channel_); |
||||
} |
||||
|
||||
void Start() { |
||||
// Start the backends.
|
||||
std::vector<int> backend_ports; |
||||
for (size_t i = 0; i < kNumBackends; ++i) { |
||||
backends_.emplace_back(new BackendServiceImpl()); |
||||
backend_servers_.emplace_back(ServerThread<BackendServiceImpl>( |
||||
"backend", server_host_, backends_.back().get())); |
||||
backend_ports.push_back(backend_servers_.back().port_); |
||||
} |
||||
// Start the load balancers.
|
||||
for (size_t i = 0; i < kNumBalancers; ++i) { |
||||
balancers_.emplace_back(new BalancerServiceImpl(backend_ports)); |
||||
balancer_servers_.emplace_back(ServerThread<BalancerServiceImpl>( |
||||
"balancer", server_host_, balancers_.back().get())); |
||||
} |
||||
// Start sending RPCs in multiple threads.
|
||||
CreateStub(); |
||||
for (size_t i = 0; i < kNumClientThreads; ++i) { |
||||
client_threads_.emplace_back( |
||||
std::thread(&ClientChannelStressTest::KeepSendingRequests, this)); |
||||
} |
||||
} |
||||
|
||||
void Shutdown() { |
||||
shutdown_ = true; |
||||
for (size_t i = 0; i < client_threads_.size(); ++i) { |
||||
client_threads_[i].join(); |
||||
} |
||||
for (size_t i = 0; i < balancers_.size(); ++i) { |
||||
balancers_[i]->Shutdown(); |
||||
balancer_servers_[i].Shutdown(); |
||||
} |
||||
for (size_t i = 0; i < backends_.size(); ++i) { |
||||
backend_servers_[i].Shutdown(); |
||||
} |
||||
grpc_fake_resolver_response_generator_unref(response_generator_); |
||||
} |
||||
|
||||
std::atomic_bool shutdown_{false}; |
||||
const grpc::string server_host_ = "localhost"; |
||||
std::shared_ptr<Channel> channel_; |
||||
std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_; |
||||
std::mutex stub_mutex_; |
||||
std::vector<std::unique_ptr<BackendServiceImpl>> backends_; |
||||
std::vector<std::unique_ptr<BalancerServiceImpl>> balancers_; |
||||
std::vector<ServerThread<BackendServiceImpl>> backend_servers_; |
||||
std::vector<ServerThread<BalancerServiceImpl>> balancer_servers_; |
||||
grpc_fake_resolver_response_generator* response_generator_; |
||||
std::vector<std::thread> client_threads_; |
||||
}; |
||||
|
||||
} // namespace
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
int main(int argc, char** argv) { |
||||
grpc_init(); |
||||
grpc_test_init(argc, argv); |
||||
grpc::testing::ClientChannelStressTest test; |
||||
test.Run(); |
||||
grpc_shutdown(); |
||||
return 0; |
||||
} |
||||
Loading…
Reference in new issue