c++ - GRPC/C++ - 服务器只读取双向流中的第一条消息

Question

我正在使用带有 C++ 的 gRPC。我有一个异步服务器和一个同步客户端。 rpc是双向流类型。

以下是我与客户端发送消息的方式:

class ConnectionService {
public:
    ConnectionService(std::shared_ptr<Channel> channel)
            : stub_(Connection::NewStub(channel)) {}

    void HearthBeat() {
        ClientContext context;

        std::shared_ptr<grpc::ClientReaderWriter<Pulse, Pulse> > stream(
                stub_->HearthBeat(&context));

        std::thread writer([stream]() {
            for (int i = 0; i < 100; ++i) {
                Pulse p;
                p.set_rate(50);
                stream->Write(p);
            }
            stream->WritesDone();
        });

        Pulse server_pulse;
        while (stream->Read(&server_pulse)) {
            std::cout << "Got message " << server_pulse.rate()<< std::endl;
        }
        writer.join();
        Status status = stream->Finish();
        if (!status.ok()) {
            std::cout << "RouteChat rpc failed." << std::endl;
        }
    }

private:
    std::unique_ptr<Connection::Stub> stub_;
};

这是我在服务器上阅读和回复的方式:

void Vibranium::ConnectionManager::HearthBeatMethod::Create() {
    connectionService_->RequestHearthBeat(&ctx_, &stream_, cq_, cq_,this);
    status_ = PROCESS;
}

void Vibranium::ConnectionManager::HearthBeatMethod::Process() {
    new HearthBeatMethod(connectionService_, cq_);
    stream_.Read(&request_, this);
    status_ = READ_CALLED;
}

bool Vibranium::ConnectionManager::HearthBeatMethod::CheckForClientMetadata() {
    return false;
}

void Vibranium::ConnectionManager::HearthBeatMethod::ReadStream() {
    std::cout << "Received: " << request_.rate() << std::endl;
    reply_.set_rate(65);
    std::cout << "Rate replied: " << reply_.rate() << std::endl;
    stream_.Write(reply_, this);
    status_ = WRITE_CALLED;
}

void Vibranium::ConnectionManager::HearthBeatMethod::WriteToStream() {
    stream_.Finish(grpc::Status::OK, this);
    status_ = FINISH;
}

这是我启动服务器的方式:

void Vibranium::Server::Run() {
    std::string server_address(serverIp_+":"+serverPort_);

    grpc::ServerBuilder builder;
    // Listen on the given address without any authentication mechanism.
    builder.AddChannelArgument(GRPC_ARG_KEEPALIVE_TIME_MS, 3000);
    builder.AddChannelArgument(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 3000);
    builder.AddChannelArgument(GRPC_ARG_HTTP2_BDP_PROBE, 1);
    builder.AddChannelArgument(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
    builder.AddChannelArgument(GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS, 1000);
    builder.AddChannelArgument(GRPC_ARG_HTTP2_MIN_SENT_PING_INTERVAL_WITHOUT_DATA_MS, 3000);
    builder.AddChannelArgument(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
    RegisterServices(builder);
    cq_ = 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();
}

void Vibranium::Server::HandleRpcs() {
    RegisterMethods();
    void* tag;  // uniquely identifies a request.
    bool ok;
    while (true) {
        GPR_ASSERT(cq_->Next(&tag, &ok));
        GPR_ASSERT(ok);
        static_cast<ServiceMethod*>(tag)->Proceed();
    }
}

这是 Proceed(); 正在做的事情:

void ServiceMethod::Proceed() {
    if (status_ == CREATE) {
        Create();
    } else if (status_ == PROCESS) {
        CheckClient();
        Process();
    } else if(status_ == READ_CALLED){
        ReadStream();
    } else if(status_ == WRITE_CALLED){
        WriteToStream();
    } else {
        Finish();
    }
}

void ServiceMethod::Finish() {
    GPR_ASSERT(status_ == FINISH);
    // Once in the FINISH state, deallocate ourselves (ServiceMethod).
    delete this;
}

因此，当我触发客户端时，它会发送 1 条消息，而不是 for 循环中描述的 100 条消息。在服务器上我可以看到输出:

Received: 50
Rate replied: 65

客户端的输出是:

Got message 65

因此我可以看到客户端和服务器之间存在通信，但服务器似乎只接收并发回第一条消息。为什么会这样？我该如何解决？

Answer 1

我认为您看到的崩溃的直接原因是在 HearthBeatMethod::Process() 中，您使用相同的标签开始读取和写入，而该标签不是甚至初始化(它是一个从未被赋予值的 void* 标签)，所以基本上没有办法知道这些操作中的任何一个何时完成。更重要的是(这可能是崩溃发生的地方)，Server::HandleRpcs() 中轮询您的完成队列的代码假设每个标记实际上是 的地址ServiceMethod 对象，其 Proceed() 方法将被调用。由于返回的标记是未初始化的指针，您基本上是在任意地址上调用方法，这会导致崩溃。

我在这里也有一些关于更广泛设计的评论，以防它们有用。

首先，对于双向流媒体服务，您可能需要比简单的 CREATE、PROCESS 和 FINISH 更多的状态，因为您一次只能进行一个读取或写入。您需要知道每次读取何时完成，以便您可以等到前一个读取完成后再开始下一个读取。写也是一样。请注意，读取和写入根本不同步，因此它们可以在完全不同的时间开始和完成。

其次，我不清楚您为什么要创建一个单独的流式 RPC 来处理连接管理。 gRPC channel 本身应该为您处理连接管理；在您的应用程序中，您不必担心这一点。原则上，应用程序应该只发送它想要的单个 RPC，让 channel 为您处理连接管理。如果您关心这个的原因是您正在做某种 session 亲和性的事情，那么请考虑只为您的应用程序本身使用流式 RPC。

希望这些信息对您有所帮助。