gRPC generic async server crashes

[TataG]

Hi.

I'm trying to run a M producer - N consumer model.

when I use predefined examples all is fine.

But when I run the greeter_async_server and > 1 greeter_async_client, it segfaults.

I identified that if I use multiple threads, then server can handle number of clients <= to number of threads as in this multithread example:

https://github.com/LesnyRumcajs/grpc_bench/blob/master/cpp_grpc_mt_bench/main.cpp

It segfaults in cq_->Next() on destroying the byte buffer.

Is it my logical misunderstanding how to use the library or it is a bug? Sorry, I'm completely new to gRPC.

I'm using gRPC for c++ from master branch on Linux machine. Same problem was on MacOS

Thank you in advance

server side:

/*

*

* Copyright 2015 gRPC authors.

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*

*/

#include <iostream>

#include <memory>

#include <sstream>

#include <string>

#include <thread>

#include <google/protobuf/message_lite.h>

#include "absl/memory/memory.h"

#include <grpcpp/support/slice.h>

#include <grpcpp/generic/async_generic_service.h>

#include <grpcpp/generic/generic_stub.h>

#include <grpcpp/impl/codegen/proto_utils.h>

#include <grpcpp/server.h>

#include <grpcpp/grpcpp.h>

#include <grpc/support/log.h>

#include <grpcpp/server_context.h>

#include <grpcpp/support/proto_buffer_writer.h>

#include <grpcpp/server_builder.h>

#include <test/cpp/util/byte_buffer_proto_helper.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;

using namespace grpc;

void* tag(int i) { return reinterpret_cast<void*>(i); }

bool ParseFromByteBuffer(ByteBuffer* buffer, grpc::protobuf::MessageLite* message) {

std::vector<Slice> slices;

(void)buffer->Dump(&slices);

std::string buf;

buf.reserve(buffer->Length());

for (auto s = slices.begin(); s != slices.end(); s++) {

buf.append(reinterpret_cast<const char*>(s->begin()), s->size());

}

return message->ParseFromString(buf);

}

class ServerImpl final {

public:

~ServerImpl() {

server_->Shutdown();

// Always shutdown the completion queue after the server.

cq_->Shutdown();

}

// There is no shutdown handling in this code.

void Run() {

std::string server_address("0.0.0.0:50051");

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.RegisterAsyncGenericService(&service_);

// Get hold of the completion queue used for the asynchronous communication

// with the gRPC runtime.

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();

}

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(AsyncGenericService* service, ServerCompletionQueue* cq)

: service_(service), cq_(cq), status_(CREATE), stream(&ctx_) {

// 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_->RequestCall(&ctx_, &stream, 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 DESTROY state.

new CallData(service_, cq_);

ByteBuffer srv_recv_buffer;

stream.Read(&srv_recv_buffer, this);

cq_->Next(&got_tag, &ok);

ParseFromByteBuffer(&srv_recv_buffer, &recv_request);

const char* test_data = "test_data";

Slice slice(test_data, strlen(test_data));

std::unique_ptr<ByteBuffer> srv_send_buffer = absl::make_unique<ByteBuffer>(&slice, 1);

stream.Write(*srv_send_buffer, this);

// 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;

} else if (status_ == FINISH) {

stream.Finish(Status::OK, this);

GPR_ASSERT(status_ == FINISH);

status_ = DESTROY;

} else {

// Once in the DESTROY state, deallocate ourselves (CallData).

delete this;

}

}

private:

// The means of communication with the gRPC runtime for an asynchronous

// server.

AsyncGenericService* 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.

GenericServerContext ctx_;

bool ok;

void* got_tag;

// What we get from the client.

HelloRequest request_;

// What we send back to the client.

HelloReply reply_;

HelloRequest recv_request;

// The means to get back to the client.

// ServerAsyncResponseWriter<HelloReply> responder_;

GenericServerAsyncReaderWriter stream;

// Let's implement a tiny state machine with the following states.

enum CallStatus { CREATE, PROCESS, FINISH, DESTROY };

CallStatus status_; // The current serving state.

};

// This can be run in multiple threads if needed.

void HandleRpcs() {

// Spawn a new CallData instance to serve new clients.

new CallData(&service_, cq_.get());

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.

cq_->Next(&tag, &ok);

GPR_ASSERT(ok);

static_cast<CallData*>(tag)->Proceed();

}

}

std::unique_ptr<ServerCompletionQueue> cq_;

AsyncGenericService service_;

std::unique_ptr<Server> server_;

std::ostringstream server_address_;

bool shutting_down_;

bool shut_down_;

};

int main(int argc, char** argv) {

ServerImpl server;

server.Run();

return 0;

}


client side:

void SendRpc(int num_rpcs) {

for (int i = 0; i < num_rpcs; i++) {

HelloRequest send_request;

HelloRequest recv_request;

HelloReply send_response;

HelloReply recv_response;

Status recv_status;

ClientContext cli_ctx;

GenericServerContext srv_ctx;

Slice slice(test_data, strlen(test_data));

ByteBuffer cli_send_buffer(&slice, 1);

std::unique_ptr<GenericClientAsyncResponseReader> call =

generic_stub_->PrepareUnaryCall(&cli_ctx, kMethodName, cli_send_buffer, &cli_cq_);

call->StartCall();

ByteBuffer cli_recv_buffer;

call->Finish(&cli_recv_buffer, &recv_status, tag(1));

ParseFromByteBuffer(&cli_recv_buffer, &recv_request);

std::thread client_check([this] { client_ok(1); });

client_check.join();

[Stan Pavlovsky]

> {

> ...

> std::unique_ptr<ByteBuffer> srv_send_buffer = absl::make_unique<ByteBuffer>(&slice, 1);

> stream.Write(*srv_send_buffer, this);

> ...

> } else if (status_ == FINISH) {

The buffer will be deallocated at the end of the scope here.

[TataG]

Thank you for suggestion. I think that was one of the problems. After making that buffer global variable, it is not crashing anymore.

Now it is calling the pure virtual method:

Thread 1 "generic_async_s" received signal SIGABRT, Aborted.

0x00007ffff6130387 in raise () from /lib64/libc.so.6

(gdb) bt

#0  0x00007ffff6130387 in raise () from /lib64/libc.so.6

#1  0x00007ffff6131a78 in abort () from /lib64/libc.so.6

#2  0x00007ffff673ea95 in __gnu_cxx::__verbose_terminate_handler() () from /lib64/libstdc++.so.6

#3  0x00007ffff673ca06 in ?? () from /lib64/libstdc++.so.6

#4  0x00007ffff673ca33 in std::terminate() () from /lib64/libstdc++.so.6

#5  0x00007ffff673d59f in __cxa_pure_virtual () from /lib64/libstdc++.so.6

#6  0x00007ffff72a3f42 in grpc::CompletionQueue::AsyncNextInternal(void**, bool*, gpr_timespec) () from /lib64/libgrpc++.so.1.42

#7  0x000000000041883b in grpc::CompletionQueue::Next (ok=0x7fffffffdac7, tag=0x7fffffffdac8, this=0x630fa0)

    at grpc/include/grpcpp/impl/codegen/completion_queue.h:187

#8  ServerImpl::HandleRpcs (this=this@entry=0x7fffffffdc80) at generic_async_server.cc:242

#9  0x0000000000418a09 in ServerImpl::Run (this=this@entry=0x7fffffffdc80) at generic_async_server.cc:113

#10 0x0000000000409389 in main (argc=<optimized out>, argv=<optimized out>) at generic_async_server.cc:264

I had similar issue before, that is when I found in this conversation:
https://groups.google.com/g/grpc-io/c/uHv_2ZyT_Q0/m/ZQCIctFQBwAJ
 recommendation to add a separate processing status for the FINISH state and add a separate DESTROY state. 

Now not sure what it could be.

воскресенье, 26 сентября 2021 г. в 19:44:31 UTC-4, dv....@gmail.com: