From 3fccb30174ee4bda07094271f88ab96a48dcecdd Mon Sep 17 00:00:00 2001 From: Ge Jun Date: Wed, 18 Jul 2018 02:52:27 -0700 Subject: [PATCH] fix the bug in SocketMapList that input is not cleared when socketmap is not created & add a multi-threaded client for thrift --- example/thrift_extension_c++/client2.cpp | 146 +++++++++++++++++++++++ src/brpc/socket_map.cpp | 2 + 2 files changed, 148 insertions(+) create mode 100644 example/thrift_extension_c++/client2.cpp diff --git a/example/thrift_extension_c++/client2.cpp b/example/thrift_extension_c++/client2.cpp new file mode 100644 index 00000000..fd3224d2 --- /dev/null +++ b/example/thrift_extension_c++/client2.cpp @@ -0,0 +1,146 @@ +// Copyright (c) 2014 Baidu, Inc. +// +// 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. + +// A client sending requests to server by multiple threads. + +#include "gen-cpp/echo_types.h" + +#include +#include +#include +#include +#include +#include +#include + +DEFINE_int32(thread_num, 50, "Number of threads to send requests"); +DEFINE_bool(use_bthread, false, "Use bthread to send requests"); +DEFINE_int32(request_size, 16, "Bytes of each request"); +DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short"); +DEFINE_string(server, "0.0.0.0:8019", "IP Address of server"); +DEFINE_string(load_balancer, "", "The algorithm for load balancing"); +DEFINE_int32(timeout_ms, 100, "RPC timeout in milliseconds"); +DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)"); +DEFINE_bool(dont_fail, false, "Print fatal when some call failed"); +DEFINE_bool(enable_ssl, false, "Use SSL connection"); +DEFINE_int32(dummy_port, -1, "Launch dummy server at this port"); + +std::string g_request; + +bvar::LatencyRecorder g_latency_recorder("client"); +bvar::Adder g_error_count("client_error_count"); + +static void* sender(void* arg) { + // Normally, you should not call a Channel directly, but instead construct + // a stub Service wrapping it. stub can be shared by all threads as well. + brpc::ThriftStub stub(static_cast(arg)); + + while (!brpc::IsAskedToQuit()) { + // We will receive response synchronously, safe to put variables + // on stack. + example::EchoRequest req; + example::EchoResponse res; + brpc::Controller cntl; + + req.__set_data(g_request); + req.__set_need_by_proxy(10); + + // Because `done'(last parameter) is NULL, this function waits until + // the response comes back or error occurs(including timedout). + stub.CallMethod("Echo", &cntl, &req, &res, NULL); + if (!cntl.Failed()) { + g_latency_recorder << cntl.latency_us(); + } else { + g_error_count << 1; + CHECK(brpc::IsAskedToQuit() || !FLAGS_dont_fail) + << "error=" << cntl.ErrorText() << " latency=" << cntl.latency_us(); + // We can't connect to the server, sleep a while. Notice that this + // is a specific sleeping to prevent this thread from spinning too + // fast. You should continue the business logic in a production + // server rather than sleeping. + bthread_usleep(50000); + } + } + return NULL; +} + +int main(int argc, char* argv[]) { + // Parse gflags. We recommend you to use gflags as well. + GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true); + + // A Channel represents a communication line to a Server. Notice that + // Channel is thread-safe and can be shared by all threads in your program. + brpc::Channel channel; + + // Initialize the channel, NULL means using default options. + brpc::ChannelOptions options; + options.ssl_options.enable = FLAGS_enable_ssl; + options.protocol = brpc::PROTOCOL_THRIFT; + options.connection_type = FLAGS_connection_type; + options.connect_timeout_ms = std::min(FLAGS_timeout_ms / 2, 100); + options.timeout_ms = FLAGS_timeout_ms; + options.max_retry = FLAGS_max_retry; + if (channel.Init(FLAGS_server.c_str(), FLAGS_load_balancer.c_str(), &options) != 0) { + LOG(ERROR) << "Fail to initialize channel"; + return -1; + } + + if (FLAGS_request_size <= 0) { + LOG(ERROR) << "Bad request_size=" << FLAGS_request_size; + return -1; + } + g_request.resize(FLAGS_request_size, 'r'); + + if (FLAGS_dummy_port >= 0) { + brpc::StartDummyServerAt(FLAGS_dummy_port); + } + + std::vector bids; + std::vector pids; + if (!FLAGS_use_bthread) { + pids.resize(FLAGS_thread_num); + for (int i = 0; i < FLAGS_thread_num; ++i) { + if (pthread_create(&pids[i], NULL, sender, &channel) != 0) { + LOG(ERROR) << "Fail to create pthread"; + return -1; + } + } + } else { + bids.resize(FLAGS_thread_num); + for (int i = 0; i < FLAGS_thread_num; ++i) { + if (bthread_start_background( + &bids[i], NULL, sender, &channel) != 0) { + LOG(ERROR) << "Fail to create bthread"; + return -1; + } + } + } + + while (!brpc::IsAskedToQuit()) { + sleep(1); + LOG(INFO) << "Sending EchoRequest at qps=" << g_latency_recorder.qps(1) + << " latency=" << g_latency_recorder.latency(1); + } + + LOG(INFO) << "EchoClient is going to quit"; + for (int i = 0; i < FLAGS_thread_num; ++i) { + if (!FLAGS_use_bthread) { + pthread_join(pids[i], NULL); + } else { + bthread_join(bids[i], NULL); + } + } + + return 0; +} diff --git a/src/brpc/socket_map.cpp b/src/brpc/socket_map.cpp index 68835ea9..331bc1a8 100644 --- a/src/brpc/socket_map.cpp +++ b/src/brpc/socket_map.cpp @@ -169,6 +169,8 @@ void SocketMapList(std::vector* ids) { SocketMap* m = get_client_side_socket_map(); if (m) { m->List(ids); + } else { + ids->clear(); } }