gRPC C++ Callback API下安全销毁Stub与Channel的时机及优雅关闭方案咨询

我最近在用gRPC C++的Callback API开发客户端，逻辑很简单：批量发送异步RPC，等所有回调都执行完成后就立即退出程序。但遇到了一个头疼的问题——如果在所有回调完成后马上退出，偶尔会触发gRPC内部线程未正常关闭的错误，比如prior > 0的断言失败，尤其是重复运行程序的时候。目前加个短暂sleep能暂时避开这个问题，但这种靠“玄学延迟”的方式实在不优雅，我想要一个确定性的、干净的关闭方案。

我主要有这几个核心疑问，希望能得到解答：

• 当使用Callback API时，什么时候才是真正安全销毁或释放Stub和Channel的时机？
• 所有回调执行完成后，gRPC后台还在做内部线程的清理工作吗？
• 有没有官方推荐的gRPC C++ Callback API优雅关闭模式？

任何相关的建议、实践模式或者文档参考我都非常感激！

最小复现示例 #

依赖 #

• gRPC版本：1.50.1
• 测试框架：Googletest

测试时用--gtest_repeat=1000参数运行可稳定复现，手动重复运行程序也能触发错误。使用的proto文件是官方的helloworld.proto。

代码示例 #

#include "helloworld.grpc.pb.h"
#include <grpcpp/grpcpp.h>
#include <gtest/gtest.h>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <memory>
#include <mutex>
#include <thread>

namespace {

class PendingRpcCounter {
 public:
  void increment() {
    {
      std::lock_guard<std::mutex> lock{m_mutex};
      ++m_pendingCalls;
    }
    m_conditionVariable.notify_all();
  }

  void decrement() {
    {
      std::lock_guard<std::mutex> lock{m_mutex};
      --m_pendingCalls;
    }
    m_conditionVariable.notify_all();
  }

  void waitForZeroPendingCalls() {
    std::unique_lock<std::mutex> lock{m_mutex};
    m_conditionVariable.wait(lock, [this]() { return m_pendingCalls == 0; });
  }

 private:
  std::mutex m_mutex{};
  std::condition_variable m_conditionVariable{};
  std::size_t m_pendingCalls{0};
};

struct RpcData {
  grpc::ClientContext context{};
  helloworld::HelloRequest request{};
  helloworld::HelloReply response{};
};

TEST(CallbackApiShutdownTest, BasicTest) {
  auto channel = grpc::CreateChannel("localhost:51000", grpc::InsecureChannelCredentials());
  auto stub = helloworld::Greeter::NewStub(channel);
  auto pendingRpcCounter = PendingRpcCounter{};

  static constexpr std::size_t NUMBER_OF_CALLS = 1000;
  for (std::size_t i = 0; i < NUMBER_OF_CALLS; ++i) {
    auto rpcData = std::make_shared<RpcData>();
    pendingRpcCounter.increment();
    stub->async()->SayHello(
        &rpcData->context, &rpcData->request, &rpcData->response,
        [rpcData, &pendingRpcCounter](grpc::Status) {
          pendingRpcCounter.decrement();
        });
  }

  pendingRpcCounter.waitForZeroPendingCalls();
  // Adding a sleep here seems to help avoid repeated-run issues:
  // std::this_thread::sleep_for(std::chrono::seconds(1));
}

} // namespace

错误及栈追踪 #

E0714 20:48:21.357772481 1162770 ref_counted.h:183] assertion failed: prior > 0
PID 1162689 - core
TID 1162770:
#0 0x0000776e4ade4b2c pthread_kill@@GLIBC_2.34
#1 0x0000776e4ad8b27e raise
#2 0x0000776e4ad6e8ff abort
#3 0x000057106583cf9d grpc_core::RefCount::Unref(grpc_core::DebugLocation const&, char const*)
#4 0x0000571065b2dbca grpc_cq_internal_unref(grpc_completion_queue*, char const*, char const*, int)
#5 0x0000571065b2fc89 cq_next(grpc_completion_queue*, gpr_timespec, void*)
#6 0x0000571065b3005e grpc_completion_queue_next
#7 0x000057106567ac57 grpc::(anonymous namespace)::CallbackAlternativeCQ::Ref()::{lambda(void*)#1}::operator()(void*) const
#8 0x000057106567ad64 grpc::(anonymous namespace)::CallbackAlternativeCQ::Ref()::{lambda(void*)#1}::_FUN(void*)
#9 0x0000571065e3688c grpc_core::(anonymous namespace)::ThreadInternalsPosix::ThreadInternalsPosix(char const*, void (*)(void*), void*, bool*, grpc_core::Thread::Options const&)::{lambda(void*)#1}::operator()(void*) const
#10 0x0000571065e368d9 grpc_core::(anonymous namespace)::ThreadInternalsPosix::ThreadInternalsPosix(char const*, void (*)(void*), void*, bool*, grpc_core::Thread::Options const&)::{lambda(void*)#1}::_FUN(void*)
#11 0x0000776e4ade2aa4 start_thread
#12 0x0000776e4ae6fc3c __clone3
TID 1162764:
#0 0x0000776e4ae6d25d syscall
#1 0x0000571065e8f84d absl::lts_20230802::synchronization_internal::FutexImpl::WaitAbsoluteTimeout(std::atomic<int>*, int, timespec const*)
#2 0x0000571065e8f4bd absl::lts_20230802::synchronization_internal::FutexWaiter::WaitUntil(std::atomic<int>*, int, absl::lts_20230802::synchronization_internal::KernelTimeout)
#3 0x0000571065e8f605 absl::lts_20230802::synchronization_internal::FutexWaiter::Wait(absl::lts_20230802::synchronization_internal::KernelTimeout)
#4 0x0000571065e8f2c0 AbslInternalPerThreadSemWait_lts_20230802
#5 0x0000571065e8d98a absl::lts_20230802::synchronization_internal::PerThreadSem::Wait(absl::lts_20230802::synchronization_internal::KernelTimeout)
#6 0x0000571065e850f3 absl::...

内容来源于stack exchange