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* tag从未给出值），因此您基本上无法判断这些操作中的任何一个何时完成。更重要的是（这可能是崩溃发生的地方），Server::HandleRpcs() 中轮询您的完成队列的代码假设每个标签实际上是一个ServiceMethod 对象的地址，其Proceed() 方法将被调用.由于返回的标记是未初始化的指针，因此您基本上是在任意地址上调用方法，这会导致崩溃。

我还有几个关于这里更广泛的设计的 cmets，以防它们有用。

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

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

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