问题描述
我使用 http2
gRPC
的方式与 example 类似。
在此示例中(发布在 async client
官方 gRPC
)中,客户端为要发送的消息分配内存,使用地址作为 github
的 tag
,并且当消息在侦听器线程中被应答时,内存(由 completion queue
- 地址知道)是空闲的。
我担心服务器不响应消息并且内存永远不会空闲的情况。
-
tag
会保护我免受这种情况的影响吗? - 我应该以不同的方式实现它吗? (使用智能指针/将指针保存在数据结构中/等等...)
异步客户端发送功能
gRPC
线程的异步客户端接收函数
void SayHello(const std::string& user) {
// Data we are sending to the server.
HelloRequest request;
request.set_name(user);
// Call object to store rpc data
AsyncclientCall* call = new AsyncclientCall;
// Because we are using the asynchronous API,we need to hold on to
// the "call" instance in order to get updates on the ongoing RPC.
call->response_reader =
stub_->PrepareAsyncSayHello(&call->context,request,&cq_);
// StartCall initiates the RPC call
call->response_reader->StartCall();
call->response_reader->Finish(&call->reply,&call->status,(void*)call);
}
主要
void AsyncCompleteRpc() {
void* got_tag;
bool ok = false;
// Block until the next result is available in the completion queue "cq".
while (cq_.Next(&got_tag,&ok)) {
// The tag in this example is the memory location of the call object
AsyncclientCall* call = static_cast<AsyncclientCall*>(got_tag);
// Verify that the request was completed successfully. Note that "ok"
// corresponds solely to the request for updates introduced by Finish().
GPR_ASSERT(ok);
if (call->status.ok())
std::cout << "Greeter received: " << call->reply.message() << std::endl;
else
std::cout << "RPC Failed" << std::endl;
// Once we're complete,deallocate the call object.
delete call;
}
}
解决方法
gRPC 会保护我免受这种情况的影响吗?
有点。 gRPC 保证所有排队的操作迟早会在它们匹配的完成队列中结束。所以你的代码没问题,只要:
- 没有在不幸的时候抛出异常。
- 您不会对创建代码路径的代码进行更改,该代码路径不包括将操作排队或删除调用。
换句话说:没关系,但很脆弱。
选项 A:
如果您想变得真正健壮,要走的路是std::shared_ptr<>
。但是,它们可能会以意想不到的方式干扰多线程性能。因此,值得与否取决于您的应用在性能与稳健性范围内的哪个方面。
这样的重构看起来像:
- 让
AsyncClientCall
继承std::enable_shared_from_this
- 将
call
的构造更改为std::make_shared<AsyncClientCall>()
- 在完成队列处理程序中,增加引用计数:
while (cq_.Next(&got_tag,&ok)) {
auto call = static_cast<AsyncClientCall*>(got_tag)->shared_from_this();
显然,去掉delete
。
选项 B:
您还可以通过 unique_ptr<>
获得不错的中途测量:
auto call = std::make_unique<AsyncClientCall>();
...
call->response_reader->Finish(&call->reply,&call->status,(void*)call.release());
和
std::unique_ptr<AsyncClientCall> call{static_cast<AsyncClientCall*>(got_tag)};
这可以在维护其他一切的同时防止重构和异常。但是,这仅适用于产生单个完成事件的一元 rpc。流式 rpc 或交换元数据的 rpc 将需要完全不同的处理。