I * add the thread *according the code of grpc/examples/helloworld/greeter_async_server <https://github.com/grpc/grpc/blob/master/examples/cpp/helloworld/greeter_async_server.cc> as following.
however, I use the ghz(*https://ghz.sh/docs/usage* <https://ghz.sh/docs/usage>) as the performance tool test the multi-thread async server, the CPU useage between 100%~210%.* why? **Is there any configure?* I also wonder the sentence "register enough server requests for the desired level of concurrency"(whole as below) - For the async completion-queue API, make sure to register enough server requests for the desired level of concurrency to avoid the server continuously getting stuck in a slow path that results in essentially serial request processing.(from Performance Best Practices <https://grpc.io/docs/guides/performance/>) ```cpp #include <iostream> #include <memory> #include <string> #include <thread> #include <grpc/support/log.h> #include <grpcpp/grpcpp.h> #ifdef BAZEL_BUILD #include "examples/protos/helloworld.grpc.pb.h" #else #include "helloworld.grpc.pb.h" #endif using grpc::Server; using grpc::ServerAsyncResponseWriter; using grpc::ServerBuilder; using grpc::ServerCompletionQueue; using grpc::ServerContext; using grpc::Status; using helloworld::Greeter; using helloworld::HelloReply; using helloworld::HelloRequest; class ServerImpl final { public: ~ServerImpl() { server_->Shutdown(); // Always shutdown the completion queue after the server. // cq_->Shutdown(); for (const auto& cq : compl_queues_) { cq->Shutdown(); } } // There is no shutdown handling in this code. void Run() { std::string server_address("0.0.0.0:50060"); 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. #define THREAD_COUNT 30 // cq_ = builder.AddCompletionQueue(); for (auto i = 0; i < THREAD_COUNT; i++) { * compl_queues_.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. // HandleRpcs(); * std::vector<std::thread> threads;* for (auto i{0}; i < THREAD_COUNT; i++) { * threads.emplace_back( std::thread(&ServerImpl::HandleRpcs, this, compl_queues_[i].get()));* } std::cout << "compl_queues_.size() = " << compl_queues_.size() << "threads.size() = " << threads.size() << std::endl; for (auto i = 0; i < threads.size(); i++) { threads[i].join(); } } private: // Class encompasing the state and logic needed to serve a request. class CallData { 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. CallData(Greeter::AsyncService* service, ServerCompletionQueue* cq) : service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) { // Invoke the serving logic right away. Proceed(); } void Proceed() { if (status_ == CREATE) { // Make this instance progress to the PROCESS state. status_ = PROCESS; // 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 CallData // instances can serve different requests concurrently), in this case // the memory address of this CallData instance. service_->RequestSayHello(&ctx_, &request_, &responder_, cq_, cq_, this); } else if (status_ == PROCESS) { // Spawn a new CallData instance to serve new clients while we process // the one for this CallData. The instance will deallocate itself as // part of its FINISH state. new CallData(service_, cq_); // std::cout << "new connected from" << std::endl; // The actual processing. std::string prefix("(server:)Hello "); reply_.set_message(prefix + request_.name()); // 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, this); } else { GPR_ASSERT(status_ == FINISH); // Once in the FINISH state, deallocate ourselves (CallData). delete this; } } private: // 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_; // 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. }; // This can be run in multiple threads if needed. void HandleRpcs(grpc::ServerCompletionQueue* cq) { // Spawn a new CallData instance to serve new clients. new CallData(&service_, cq); 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 CallData 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(ok); static_cast<CallData*>(tag)->Proceed(); } } // std::unique_ptr<ServerCompletionQueue> cq_; * std::vector<std::unique_ptr<ServerCompletionQueue>> compl_queues_;* Greeter::AsyncService service_; std::unique_ptr<Server> server_; }; int main(int argc, char** argv) { ServerImpl server; server.Run(); return 0; } ``` -- 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/83a327d7-09cb-496e-8a07-bff0dc33d6ddn%40googlegroups.com.
