After did these optimizations on windows (sorted by the their contributions to the optimized result):
1. build the *released version* from the server itself and all its dependent libraries. 2. using different threads and assigning each of the threads with a dedicated asynchronous client, each client with a distinguished channel. 3. separating the client and the server from being located on one single machine. And, I've compiled the released versions for all the dependencies and the > application itself. Results are basically the same . Note: this previously posted conclusion is wrong. I got a better result now: OS pool 1 thread 2 thread 4 thread 8 thread 16 thread 1 CQ win 100 42589 45289 43610 42194 37023 200 45998 48053 43840 41823 40048 400 43610 43497 39984 41084 41928 mac 100 42283 39416 25693 24497 24113 200 42408 39385 25354 24348 24691 400 42517 39261 25614 24491 24295 2 CQ win 100 42319 42918 49480 46533 41928 200 46189 47236 47824 48030 45310 400 47664 47778 42480 47058 36941 mac 100 42265 36456 36523 26695 24838 200 39463 36127 35842 24570 24467 400 42444 39292 36995 26702 24642 4 CQ win 100 40436 43610 41631 45850 40799 200 46882 44903 50813 41034 42354 400 45766 49504 46189 48590 45330 mac 100 41684 39619 34578 36805 26143 200 42753 36153 36140 36710 24844 400 42158 36710 36443 36927 26329 [image: image.png] Some conclusions based on the above: 1. the release version is a critical factor for benchmark on windows. 2. *scaling #CQ / #thread / #pool_size still doesn't help in improving the throughput.* 3. on mac, more threads even meaning worse result. There is still a big gap between my test and the official benchmark for C++ <https://performance-dot-grpc-testing.appspot.com/explore?dashboard=5685265389584384&widget=374364983&container=1597677437>, and my cpu usage still being low(50~60% on windows and 100~260% on mac), I don't know what's next. And I thought different channel objects on the client side may stands for different TCP connections, but my test proved it to be wrong: even I've created 4~8 channel objects on the client side, there was still *only one connection between the client and the server*, and I presumed that the #connections also being an important factor for throughput, although one single connection can be multiplexed. The code files are attached. On Thu, May 23, 2019 at 12:47 AM Lei Wang <[email protected]> wrote: > And one I thing I want to figure out is that CQ is just for event loop > processing. More cqs does not promise you more speed. You could run cq in > following modes: > > Sever > - 1 cq in a main loop thread : many worker threads pool to process tagged > events each of which releases itself once reading and writing finish. > - 1 cq in many worker threads because cqs are thread safe : many worker > threads pool to process tagged events each of which releases itself once > reading and writing finish. > - many cq in different loop threads : many worker threads to process > tagged events events each of which releases itself once reading and writing > finish. > > What ever the model you implement, only the number of worker threads > affect how much you can process when they arrive. > > On Thu, May 23, 2019 at 12:28 AM Lei Wang <[email protected]> wrote: > >> Are you implementing a server for streaming reply? I have provided you an >> implementation how to implement 'CallStreamDataReply' otherwise you won't >> be able to use asynchronous mechanism by grpc c++ server. Also, try to use >> c++ threads pool instead of grpc threads pool for cq. >> >> I have written an end to end test for pubsub services to test how >> asynchronous server works. There is no serious problems in my tests. >> >> >> On Wed, May 22, 2019 at 2:50 AM Lalit Kumar <[email protected]> wrote: >> >>> Just to add I'm trying server streaming using c++ bindings. Client opens >>> just one long lived RPC and Server starts streaming data to client. Both >>> running on same Debian machine having 4 cores. Server is implemented using >>> async api. I'm getting max 23-25K messages/second. Message payload is just >>> 4 bytes. Tried multiple threads to service completionQ but its not >>> changing anything. >>> >>> -Lalit >>> >>> On Tuesday, May 21, 2019 at 10:38:15 AM UTC-7, Arthur Wang wrote: >>>> >>>> And, I've compiled the released versions for all the dependencies and >>>> the application itself. Results are basically the same. >>>> >>>> On Sun, May 19, 2019 at 5:19 PM Arthur Wang <[email protected]> wrote: >>>> >>>>> And, both the client and the server on the same machine, communicating >>>>> with tcp, even though I don't think this could be a reason for poor >>>>> throughput. >>>>> >>>>> On Sun, May 19, 2019 at 5:12 PM Arthur Wang <[email protected]> wrote: >>>>> >>>>>> And, of course, the cpu are not fully running on both win(~30+%, >>>>>> 100% fully load) and mac(200+%,800% fully load ). >>>>>> >>>>>> On Sun, May 19, 2019 at 5:08 PM Arthur Wang <[email protected]> >>>>>> wrote: >>>>>> >>>>>>> Hi all: >>>>>>> >>>>>>> I've done a benchmark for the asynchronous version of the c++ >>>>>>> server, but the result is terribly poor compared to either the official >>>>>>> one >>>>>>> <https://performance-dot-grpc-testing.appspot.com/explore?dashboard=5685265389584384> >>>>>>> or >>>>>>> some other results like this >>>>>>> <https://github.com/grpc/grpc/issues/6504> done by other users, I'm >>>>>>> wondering what is wrong with my test, could someone do me a favor? >>>>>>> >>>>>>> There are the details for my test.: >>>>>>> >>>>>>> Enrivonment : >>>>>>> >>>>>>> - win 10pro 64bit : intel i7 4.0GHZ * 4 physical core * 2 logic >>>>>>> core RAM:16GB >>>>>>> - mac high serria 10.13.6: intel i7 2.2GHZ * 4 physical core * >>>>>>> 2 logic core RAM:16G >>>>>>> >>>>>>> *Note: I don't have any linux machines, so there are no >>>>>>> results for it.* >>>>>>> >>>>>>> Grpc version: >>>>>>> >>>>>>> - win : 77ec6c94ad58bedfdf8f28682914e9893e318be0, around 2018.1 >>>>>>> - mac: e3f37b7f4320f0e016a806796e2adaed03bf23f7, around 2019.2 >>>>>>> >>>>>>> Compiler: >>>>>>> >>>>>>> - win: vs2015 Microsoft (R) C/C++ Optimizing Compiler Version >>>>>>> 19.00.24215.1 for x86 >>>>>>> - mac: Apple LLVM version 10.0.0 (clang-1000.10.44.4) >>>>>>> >>>>>>> Parameters: >>>>>>> >>>>>>> - number of CQs an server instance has. >>>>>>> - number of threads totally exist, evenly distributed on each CQ. >>>>>>> - number of pool size(which is the *CallData* instances >>>>>>> pre-allocated for each CQ. >>>>>>> >>>>>>> Client & Server Behavior: >>>>>>> >>>>>>> - Client: Sending arond 30-50k requests to the async server in >>>>>>> parallel, and waiting for all of them to be responded. >>>>>>> - Server: Based on the *greeter_async_server.cc of the >>>>>>> helloworld example* - just do some basically ignorable logic and >>>>>>> response immediately after that. >>>>>>> >>>>>>> * The code are in the attachments.* >>>>>>> >>>>>>> The result (requests the server can deal per second in average) >>>>>>> is : >>>>>>> >>>>>>> OS pool 1 thread 2 thread 4 thread 8 thread 16 thread >>>>>>> 1 CQ win 100 3647 4016 3366 3012 3427 >>>>>>> 200 3963 5202 3308 3411 4507 >>>>>>> 400 3787 4494 3476 3203 3123 >>>>>>> mac 100 38880 35803 22629 22128 22321 >>>>>>> 200 37778 35285 24801 21805 22172 >>>>>>> 400 39200 36231 23446 22311 22553 >>>>>>> 2 CQ win 100 3086 4144 4575 3662 3959 >>>>>>> 200 3334 4354 3588 3507 3536 >>>>>>> 400 3069 4299 3565 3830 3852 >>>>>>> mac 100 39154 32362 31948 25227 23441 >>>>>>> 200 37792 33046 31908 25169 24142 >>>>>>> 400 40584 33909 32446 25284 24943 >>>>>>> 4 CQ win 100 3204 4199 4644 3988 3742 >>>>>>> 200 3125 4097 3954 3997 3638 >>>>>>> 400 3207 4196 3731 4008 3526 >>>>>>> mac 100 39793 33101 28851 32310 25214 >>>>>>> 200 38804 31545 32605 32268 24869 >>>>>>> 400 38819 32776 31786 32133 25207 >>>>>>> >>>>>>> [image: image.png] >>>>>>> >>>>>>> Recap of the result: >>>>>>> >>>>>>> - number of CQs basically has nothing to do with the throughput, >>>>>>> *which >>>>>>> is unbelievable, in my understanding, it is an critical factor after >>>>>>> learned how the polling-engine >>>>>>> >>>>>>> <https://github.com/grpc/grpc/blob/master/doc/core/epoll-polling-engine.md> >>>>>>> works.* >>>>>>> - number of pool size has nothing to do with the throughput,* >>>>>>> I'm not sure whether this is normal or not.* >>>>>>> - number of threads which are evenly distributed among the CQs *has >>>>>>> nothing to do with the throughput on windows, but decreasing it on >>>>>>> mac. >>>>>>> This is somewhere wired.* >>>>>>> >>>>>>> Besides, I tested the synchronous version(*greeter_server.cc*) >>>>>>> of c++ server, giving me the result of 18807 on mac and 3041 on win, >>>>>>> which >>>>>>> is also very poor. >>>>>>> >>>>>>> I've read the example of *grpc\test\cpp\qps\server_async.cc *which >>>>>>> said to be an good example of how to write a high throughput async c++ >>>>>>> server. I found its approach of enhancing the throughput is more of less >>>>>>> the same with mine: >>>>>>> >>>>>>> - scale #CQ. >>>>>>> - scale #thread. >>>>>>> >>>>>>> I haven't test the *qps example* on my machine, I haven't find a >>>>>>> easier way to build it and hard to believe to it can achieve a better >>>>>>> result with the same environment of mine. >>>>>>> >>>>>>> I know there are something must be wrong, but where are they? >>>>>>> >>>>>>> - Thanks a lot. >>>>>>> - Arthur. >>>>>>> >>>>>> -- >>> You received this message because you are subscribed to the Google >>> Groups "grpc.io" group. >>> To unsubscribe from this group and stop receiving emails from it, send >>> an email to [email protected]. >>> To view this discussion on the web visit >>> https://groups.google.com/d/msgid/grpc-io/0bb9c0de-e91e-42ce-b99f-9ace2b44d5b7%40googlegroups.com >>> <https://groups.google.com/d/msgid/grpc-io/0bb9c0de-e91e-42ce-b99f-9ace2b44d5b7%40googlegroups.com?utm_medium=email&utm_source=footer> >>> . >>> >> -- > You received this message because you are subscribed to the Google Groups " > grpc.io" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to [email protected]. > To view this discussion on the web visit > https://groups.google.com/d/msgid/grpc-io/CAF84%3DtevjXUAcuZ-e4dHQO1S-86BxMH%3DNT0Vcy7DPd%3D50%2BZ_Dg%40mail.gmail.com > <https://groups.google.com/d/msgid/grpc-io/CAF84%3DtevjXUAcuZ-e4dHQO1S-86BxMH%3DNT0Vcy7DPd%3D50%2BZ_Dg%40mail.gmail.com?utm_medium=email&utm_source=footer> > . > -- You received this message because you are subscribed to the Google Groups "grpc.io" group. To unsubscribe from this group and stop receiving emails from it, send an email to [email protected]. To view this discussion on the web visit https://groups.google.com/d/msgid/grpc-io/CAKymdqNnbH56NYGzLGuZNFxKGjQ%2BCa1YtOH1t7U%3DYehEVdCRYQ%40mail.gmail.com.
/* * * Copyright 2015 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 <memory> #include <iostream> #include <string> #include <thread> #include <vector> #include <random> #include <atomic> #include <grpc++/grpc++.h> #include <grpc/support/log.h> #include "helloworld.grpc.pb.h" using grpc::Server; using grpc::ServerAsyncResponseWriter; using grpc::ServerBuilder; using grpc::ServerContext; using grpc::ServerAsyncReaderWriter; using grpc::ServerCompletionQueue; using grpc::Status; using grpc::StatusCode; using helloworld::HelloRequest; using helloworld::HelloReply; using helloworld::Greeter; int g_thread_num = 1; int g_cq_num = 1; int g_pool = 1; int g_port = 50051; std::atomic<void*>** g_instance_pool = nullptr; class CallDataBase { public: CallDataBase(Greeter::AsyncService* service, ServerCompletionQueue* cq) : service_(service), cq_(cq){ } virtual void Proceed(bool ok) = 0; protected: // The means of communication with the gRPC runtime for an asynchronous // server. Greeter::AsyncService* service_; // The producer-consumer queue where for asynchronous server notifications. ServerCompletionQueue* cq_; // Context for the rpc, allowing to tweak aspects of it such as the use // of compression, authentication, as well as to send metadata back to the // client. ServerContext ctx_; // What we get from the client. HelloRequest request_; // What we send back to the client. HelloReply reply_; }; class CallDataUnary : CallDataBase { public: // Take in the "service" instance (in this case representing an asynchronous // server) and the completion queue "cq" used for asynchronous communication // with the gRPC runtime. CallDataUnary(Greeter::AsyncService* service, ServerCompletionQueue* cq) : CallDataBase(service,cq),responder_(&ctx_), status_(CREATE) { // Invoke the serving logic right away. // As part of the initial CREATE state, we *request* that the system // start processing SayHello requests. In this request, "this" acts are // the tag uniquely identifying the request (so that different CallDataUnary // instances can serve different requests concurrently), in this case // the memory address of this CallDataUnary instance. status_ = PROCESS; service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_, (void*)this); } void Proceed(bool ok) { if (status_ == PROCESS) { // Spawn a new CallDataUnary instance to serve new clients while we process // the one for this CallDataUnary. The instance will deallocate itself as // part of its FINISH state. new CallDataUnary(service_, cq_); // The actual processing. std::string prefix("Hello "); reply_.set_message(prefix + request_.name()); //std::this_thread::sleep_for(std::chrono::seconds(4)); // And we are done! Let the gRPC runtime know we've finished, using the // memory address of this instance as the uniquely identifying tag for // the event. status_ = FINISH; responder_.Finish(reply_, Status::OK, (void*)this); } else { if (status_ != FINISH) { std::cout << "wrong status:" << status_ << std::endl; assert(false); } // Once in the FINISH state, deallocate ourselves (CallDataUnary). delete this; } } private: // The means to get back to the client. ServerAsyncResponseWriter<HelloReply> responder_; // Let's implement a tiny state machine with the following states. enum CallStatus { CREATE, PROCESS, FINISH }; CallStatus status_; // The current serving state. }; class CallDataBidi : CallDataBase { public: // Take in the "service" instance (in this case representing an asynchronous // server) and the completion queue "cq" used for asynchronous communication // with the gRPC runtime. CallDataBidi(Greeter::AsyncService* service, ServerCompletionQueue* cq) : CallDataBase(service,cq),rw_(&ctx_){ // Invoke the serving logic right away. status_ = BidiStatus::CONNECT; ctx_.AsyncNotifyWhenDone((void*)this); service_->RequestSayHelloEx(&ctx_, &rw_, cq_, cq_, (void*)this); //std::thread* _t(new std::thread(&CallDataBidi::threadFunc,this)); //_t->detach(); } void Proceed(bool ok) { std::unique_lock<std::mutex> _wlock(this->m_mutex); switch (status_) { case BidiStatus::READ: //Meaning client said it wants to end the stream either by a 'writedone' or 'finish' call. if (!ok) { std::cout << "thread:" << std::this_thread::get_id() << " tag:" << this << " CQ returned false." << std::endl; Status _st(StatusCode::OUT_OF_RANGE,"test error msg"); rw_.Finish(_st,(void*)this); status_ = BidiStatus::DONE; std::cout << "thread:" << std::this_thread::get_id() << " tag:" << this << " after call Finish(), cancelled:" << this->ctx_.IsCancelled() << std::endl; break; } std::cout << "thread:" << std::this_thread::get_id() << " tag:" << this << " Read a new message:" << request_.name() << std::endl; reply_.set_message("arthur"); rw_.Write(reply_, (void*)this); status_ = BidiStatus::WRITE; break; case BidiStatus::WRITE: std::cout << "thread:" << std::this_thread::get_id() << " tag:" << this << " Written a message:" << reply_.message() << std::endl; rw_.Read(&request_, (void*)this); status_ = BidiStatus::READ; break; case BidiStatus::CONNECT: std::cout << "thread:" << std::this_thread::get_id() << " tag:" << this << " connected:" << std::endl; new CallDataBidi(service_, cq_); rw_.Read(&request_, (void*)this); status_ = BidiStatus::READ; break; case BidiStatus::DONE: std::cout << "thread:" << std::this_thread::get_id() << " tag:" << this << " Server done, cancelled:" << this->ctx_.IsCancelled() << std::endl; status_ = BidiStatus::FINISH; break; case BidiStatus::FINISH: std::cout << "thread:" << std::this_thread::get_id() << "tag:" << this << " Server finish, cancelled:" << this->ctx_.IsCancelled() << std::endl; _wlock.unlock(); delete this; break; default: std::cerr << "Unexpected tag " << int(status_) << std::endl; assert(false); } } private: // The means to get back to the client. ServerAsyncReaderWriter<HelloReply,HelloRequest> rw_; // Let's implement a tiny state machine with the following states. enum class BidiStatus { READ = 1, WRITE = 2, CONNECT = 3, DONE = 4, FINISH = 5 }; BidiStatus status_; std::mutex m_mutex; }; class ServerImpl final { public: ~ServerImpl() { server_->Shutdown(); // Always shutdown the completion queue after the server. for (const auto& _cq : m_cq) _cq->Shutdown(); } // There is no shutdown handling in this code. void Run() { std::string server_address("0.0.0.0:" + std::to_string(g_port)); ServerBuilder builder; // Listen on the given address without any authentication mechanism. builder.AddListeningPort(server_address, grpc::InsecureServerCredentials()); // Register "service_" as the instance through which we'll communicate with // clients. In this case it corresponds to an *asynchronous* service. builder.RegisterService(&service_); // Get hold of the completion queue used for the asynchronous communication // with the gRPC runtime. for (int i = 0; i < g_cq_num; ++i) { //cq_ = builder.AddCompletionQueue(); m_cq.emplace_back(builder.AddCompletionQueue()); } // Finally assemble the server. server_ = builder.BuildAndStart(); std::cout << "Server listening on " << server_address << std::endl; // Proceed to the server's main loop. std::vector<std::thread*> _vec_threads; for (int i = 0; i < g_thread_num; ++i) { int _cq_idx = i % g_cq_num; for (int j = 0; j < g_pool; ++j) { new CallDataUnary(&service_, m_cq[_cq_idx].get()); new CallDataBidi(&service_, m_cq[_cq_idx].get()); } _vec_threads.emplace_back(new std::thread(&ServerImpl::HandleRpcs, this, _cq_idx)); } std::cout << g_thread_num << " working aysnc threads spawned" << std::endl; for (const auto& _t : _vec_threads) _t->join(); } private: // Class encompasing the state and logic needed to serve a request. // This can be run in multiple threads if needed. void HandleRpcs(int cq_idx) { // Spawn a new CallDataUnary instance to serve new clients. void* tag; // uniquely identifies a request. bool ok; while (true) { // Block waiting to read the next event from the completion queue. The // event is uniquely identified by its tag, which in this case is the // memory address of a CallDataUnary instance. // The return value of Next should always be checked. This return value // tells us whether there is any kind of event or cq_ is shutting down. //GPR_ASSERT(cq_->Next(&tag, &ok)); GPR_ASSERT(m_cq[cq_idx]->Next(&tag, &ok)); CallDataBase* _p_ins = (CallDataBase*)tag; _p_ins->Proceed(ok); } } std::vector<std::unique_ptr<ServerCompletionQueue>> m_cq; Greeter::AsyncService service_; std::unique_ptr<Server> server_; }; const char* ParseCmdPara( char* argv,const char* para) { auto p_target = std::strstr(argv,para); if (p_target == nullptr) { printf("para error argv[%s] should be %s \n",argv,para); return nullptr; } p_target += std::strlen(para); return p_target; } int main(int argc, char** argv) { if (argc != 5) { std::cout << "Usage:./program --thread=xx --cq=xx --pool=xx --port=xx"; return 0; } g_thread_num = std::atoi(ParseCmdPara(argv[1],"--thread=")); g_cq_num = std::atoi(ParseCmdPara(argv[2],"--cq=")); g_pool = std::atoi(ParseCmdPara(argv[3],"--pool=")); g_port = std::atoi(ParseCmdPara(argv[4],"--port=")); ServerImpl server; server.Run(); return 0; }
/* * * Copyright 2015 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 <iostream> #include <memory> #include <string> #include <grpc++/grpc++.h> #include <grpc/support/log.h> #include <thread> #include "helloworld.grpc.pb.h" using grpc::Channel; using grpc::ClientAsyncResponseReader; using grpc::ClientContext; using grpc::CompletionQueue; using grpc::Status; using helloworld::HelloRequest; using helloworld::HelloReply; using helloworld::Greeter; class GreeterClient { public: explicit GreeterClient(const std::string &addr,CompletionQueue* in_cq,uint32_t count) : m_total(count) { auto shp_channel = grpc::CreateChannel(addr, grpc::InsecureChannelCredentials()); stub_ = Greeter::NewStub(shp_channel); this->cq_ = in_cq; //this->cq_ = new CompletionQueue(); } // Assembles the client's payload and sends it to the server. void SayHello(const std::string& user) { // Data we are sending to the server. HelloRequest request; request.set_name(user); // Call object to store rpc data AsyncClientCall* call = new AsyncClientCall; std::chrono::time_point<std::chrono::system_clock> _deadline = std::chrono::system_clock::now() + std::chrono::seconds(2); //call->context.set_deadline(_deadline); // stub_->PrepareAsyncSayHello() creates an RPC object, returning // an instance to store in "call" but does not actually start the RPC // Because we are using the asynchronous API, we need to hold on to // the "call" instance in order to get updates on the ongoing RPC. call->response_reader = stub_->PrepareAsyncSayHello(&call->context, request, cq_); // StartCall initiates the RPC call call->response_reader->StartCall(); // Request that, upon completion of the RPC, "reply" be updated with the // server's response; "status" with the indication of whether the operation // was successful. Tag the request with the memory address of the call object. call->response_reader->Finish(&call->reply, &call->status, (void*)call); } // Loop while listening for completed responses. // Prints out the response from the server. void AsyncCompleteRpc() { //std::cout << "thread " << std::this_thread::get_id() << " in.." << std::endl;; void* got_tag; bool ok = false; uint32_t _counter = 0; auto _start = std::chrono::steady_clock::now(); std::cout << "thread " << std::this_thread::get_id() << " start timer" << std::endl;; // Block until the next result is available in the completion queue "cq". while (cq_->Next(&got_tag, &ok)) { // The tag in this example is the memory location of the call object AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag); // Verify that the request was completed successfully. Note that "ok" // corresponds solely to the request for updates introduced by Finish(). GPR_ASSERT(ok); if (!call->status.ok()) { std::cout << call->status.error_code() << ",msg:" << call->status.error_message(); GPR_ASSERT(false); } /* if (call->status.ok()) std::cout << "Greeter received: " << call->reply.message() << std::endl; else std::cout << "RPC failed" << std::endl; */ // Once we're complete, deallocate the call object. //delete call; if (++_counter >= m_total) break; } auto _end = std::chrono::steady_clock::now(); auto _ms = std::chrono::duration_cast<std::chrono::milliseconds>(_end - _start); std::cout << "thread " << std::this_thread::get_id() << " inner time cost:" << _ms.count() << std::endl; uint32_t _throughput = m_total / float(_ms.count()) * 1000; std::cout << "thread " << std::this_thread::get_id() << " inner throughput : " << _throughput << std::endl; } private: // struct for keeping state and data information struct AsyncClientCall { // Container for the data we expect from the server. HelloReply reply; // Context for the client. It could be used to convey extra information to // the server and/or tweak certain RPC behaviors. ClientContext context; // Storage for the status of the RPC upon completion. Status status; std::unique_ptr<ClientAsyncResponseReader<HelloReply>> response_reader; }; // Out of the passed in Channel comes the stub, stored here, our view of the // server's exposed services. std::unique_ptr<Greeter::Stub> stub_; // The producer-consumer queue we use to communicate asynchronously with the // gRPC runtime. CompletionQueue* cq_; uint32_t m_total; }; int main(int argc, char** argv) { uint32_t count = 50000; if (argc != 4) { std::cout << "Usage:./program --count=xx --thread=xx --addr=xx"; return 0; } const char * target_str = "--count="; auto p_target = std::strstr(argv[1],target_str); if (p_target == nullptr) { printf("para error argv[1] should be --count=xx \n"); return 0; } p_target += std::strlen(target_str); count = std::atoi(p_target); uint32_t thread_num = 1; target_str = "--thread="; p_target = std::strstr(argv[2],target_str); if (p_target == nullptr) { printf("para error argv[2] should be --thread=xx \n"); return 0; } p_target += std::strlen(target_str); thread_num = std::atoi(p_target); std::string _addr = "localhost:50051"; target_str = "--addr="; p_target = std::strstr(argv[3],target_str); if (p_target == nullptr) { printf("para error argv[1] should be --addr=xx \n"); return 0; } p_target += std::strlen(target_str); _addr = p_target; std::cout << "req for each thread:" << count << std::endl; // Instantiate the client. It requires a channel, out of which the actual RPCs // are created. This channel models a connection to an endpoint (in this case, // localhost at port 50051). We indicate that the channel isn't authenticated // (use of InsecureChannelCredentials()). CompletionQueue one_cq; std::vector<GreeterClient*> _vec; std::vector<std::thread*> _vec_t; for (int i = 0; i < thread_num; ++i) { auto * _p_client = new GreeterClient(_addr,&one_cq,count); _vec.push_back(_p_client); std::string user("world " + std::to_string(i)); for (int j = 0; j < count; j++) _p_client->SayHello(user); // The actual RPC call! } auto _start = std::chrono::steady_clock::now(); for (uint32_t i = 0; i < thread_num; i++) _vec_t.push_back(new std::thread(&GreeterClient::AsyncCompleteRpc, _vec[i])); //std::this_thread::sleep_for(std::chrono::milliseconds(30)); for (uint32_t i = 0; i < thread_num; i++) _vec_t[i]->join(); int _total = thread_num * count; std::cout << "m_total:" << _total << std::endl; auto _end = std::chrono::steady_clock::now(); auto _ms = std::chrono::duration_cast<std::chrono::milliseconds>(_end - _start); std::cout << "time cost:" << _ms.count() << std::endl; uint32_t _throughput = _total / float(_ms.count()) * 1000; std::cout << "throughput : " << _throughput << std::endl; return 0; }
