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.

Reply via email to