gRPC异步服务器接收客户端流：断言失败及循环读取问题求助

断言失败（GPR_ASSERT(ok)为false）的修复 #

• 流正常终止或异常中断：客户端发送完所有消息后会主动关闭流，或网络异常导致流断开，此时Read返回的ok为false是正常状态，直接断言会触发失败。应先判断ok状态，再处理：

  void OnReadDone(bool ok) {
      if (!ok) {
          // 流结束，完成响应并清理
          responder_.Finish(grpc::Status::OK, this);
          return;
      }
      // 处理接收到的颜色数据
      handle_color(request_.color());
      // 继续读取下一条消息
      reader_.Read(&request_, this);
  }
  

• 请求对象初始化问题：确保request_是CallData类的成员变量，避免使用栈上临时变量导致的内存问题。
• 上下文生命周期错误：CallData实例必须在所有异步操作完成前保持有效，不能提前销毁。

持续接收流消息的实现方式 #

ServerAsyncReader::Read是单次异步调用，要接收客户端的所有流消息，需在每次Read完成的回调中再次发起Read请求，形成循环读取逻辑：

1. 初始化时发起第一次Read：在CallData的构造或初始化流程中，注册服务方法后立即发起第一次Read：

   CallDataSendColors(Greeter::AsyncService* service, ServerCompletionQueue* cq)
       : service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) {
       Proceed();
   }
   
   void Proceed() {
       if (status_ == CREATE) {
           status_ = READING;
           // 注册SendColors服务方法，等待客户端调用
           service_->RequestSendColors(&ctx_, &reader_, cq_, cq_, this);
           // 发起第一次Read请求
           reader_.Read(&request_, this);
       }
       // 其他状态处理...
   }
   

2. 回调中循环发起Read：在Read完成的回调逻辑中，每次成功读取消息后，再次调用reader_.Read，直到ok为false（流结束）：

   void Proceed() {
       switch (status_) {
           case CREATE: {
               status_ = READING;
               service_->RequestSendColors(&ctx_, &reader_, cq_, cq_, this);
               reader_.Read(&request_, this);
               break;
           }
           case READING: {
               bool ok = true;
               // 检查Read操作是否成功（实际ok是gRPC传入的回调参数）
               if (ok) {
                   // 处理当前颜色消息
                   std::cout << "Received color: " << request_.color() << std::endl;
                   // 发起下一次Read，持续接收流数据
                   reader_.Read(&request_, this);
               } else {
                   // 流结束，返回响应并标记为完成状态
                   responder_.Finish(grpc::Status::OK, this);
                   status_ = FINISH;
                   Proceed();
               }
               break;
           }
           case FINISH: {
               // 清理当前CallData实例
               delete this;
               break;
           }
       }
   }
   

完整CallDataSendColors示例 #

#include <iostream>
#include <grpcpp/grpcpp.h>
#include "greeter.grpc.pb.h"

using grpc::Server;
using grpc::ServerAsyncReader;
using grpc::ServerCompletionQueue;
using grpc::ServerContext;
using grpc::ServerAsyncResponseWriter;
using grpc::Status;
using greeter::ColorRequest;
using greeter::ColorResponse;
using greeter::Greeter;

class CallDataSendColors {
public:
    enum CallStatus { CREATE, READING, FINISH };

    CallDataSendColors(Greeter::AsyncService* service, ServerCompletionQueue* cq)
        : service_(service), cq_(cq), responder_(&ctx_), status_(CREATE) {
        Proceed();
    }

    void Proceed() {
        switch (status_) {
            case CREATE: {
                status_ = READING;
                // 注册服务方法，等待客户端发起调用
                service_->RequestSendColors(&ctx_, &reader_, cq_, cq_, this);
                // 发起第一次异步Read请求
                reader_.Read(&request_, this);
                break;
            }
            case READING: {
                bool ok = true;
                if (ok) {
                    std::cout << "Got color: " << request_.color() << std::endl;
                    // 继续发起下一次Read，循环接收流
                    reader_.Read(&request_, this);
                } else {
                    // 流正常结束或异常中断，返回响应
                    responder_.Finish(Status::OK, this);
                    status_ = FINISH;
                    Proceed();
                }
                break;
            }
            case FINISH: {
                // 销毁当前CallData实例，释放资源
                delete this;
                break;
            }
        }
    }

private:
    Greeter::AsyncService* service_;
    ServerCompletionQueue* cq_;
    ServerContext ctx_;
    ColorRequest request_;
    ColorResponse response_;
    ServerAsyncReader<ColorResponse, ColorRequest> reader_;
    ServerAsyncResponseWriter<ColorResponse> responder_;
    CallStatus status_;
};

关键注意事项 #

• 禁止直接断言ok：ok为false是流结束的正常场景，只有当ctx_.IsCancelled()为true时才是异常终止。
• CallData生命周期管理：必须确保CallData实例在所有异步操作完成前不被销毁，通常在FINISH状态下自行delete。
• CompletionQueue线程处理：需单独启动线程循环调用cq_->Next()，处理所有异步事件回调。

内容的提问来源于stack exchange，提问作者snooker9