Unable to Perform Server-Side Stream Initialization Checks in gRPC Client-Side Streaming RPC to return error on stream creation

[mostafafarzaneh]

I'm currently working on a gRPC implementation where the client is in Go, and the server is in C++. I have a client-side streaming RPC, and I want to perform some checks on the server when the client creates a new stream. Specifically, I want the server to return an error if it is busy.

However, I've observed that the server code only executes when the client starts sending requests, not during the creation of the stream itself. To overcome this, I had to send a dummy message to the server to check if it is busy or not.

Is this the expected behavior in gRPC, or am I missing something in my implementation?

[zasweq]

What does "is busy" semantically mean here? If you want to check if the server just doesn't create the stream, you can use the timeout in the context, and it will return an error. Once a stream is created, your application handler can use it to signal back to the client based off properties about the server whether it is busy or not.

[mostafafarzaneh]

I did more testing and figured out that I needed to read from the stream on the client side to get the error. I was expecting to get the error on stream creation, which I suppose was naive.

Here is the server code:

grpc::Status TestV1ServiceImpl::run_session(::grpc::ServerContext *context,
                                            ::grpc::ServerReader<::test::v1::Request> *stream,
                                            ::google::protobuf::Empty *) {
  {
    auto cpu = utils::get_cpu_usage();
    if (cpu > 0) {
      printf("1111CPU usage %f exeeds 80\n", cpu);
      return grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED, "CPU resource exhausted");
    }
  }


  proto::Request req;
  while(true) {
    auto res = stream->Read(&req);
    if (!res) {
      printf("Cancel connection. Read error\f");
      return grpc::Status::CANCELLED;
    } else {
      printf("RECEIVED!\n");
    }
  }
}

And here is my client:

func main() {
        conn, err := createConnection()
        if err != nil {
                fmt.Println("1: ", err)
                panic(err)
        }

        ctx, _ := context.WithCancel(context.Background())

        c := test.NewTestClient(conn)
        stream, err := c.RunSession(ctx)

        if err != nil {
                fmt.Println("2: ", err)
                panic(err)
        }

        m := new(emptypb.Empty)
        for err == nil {
                if err := stream.RecvMsg(m); err != nil {
                        fmt.Println("4: ", err)
                        panic(err)
                }

        }

        fmt.Println("####EXITING####")
        fmt.Println("5: ", err)
        return
}

[arvindbr8]

Great. Hope that solves your issue :)