Inbound RPC calls are not destroyed until timeout or service shutdown

[ttyusupov]

Confirmed with following updated test:

TEST_F(TestRpc, TestCall) {
  // Set up server.
  TestServerOptions options;
  options.messenger_options.keep_alive_timeout = 60s;
  HostPort server_addr;
  StartTestServer(&server_addr, options);

  // Set up client.
  LOG(INFO) << "Connecting to " << server_addr;
  auto client_messenger = CreateAutoShutdownMessengerHolder("Client");
  Proxy p(client_messenger.get(), server_addr);

  for (int i = 0; i < 10; i++) {
    ASSERT_OK(DoTestSyncCall(&p, CalculatorServiceMethods::AddMethod()));
  }
  LOG(INFO) << "Calls OK";

  SleepFor(kSleep);

  LOG(INFO) << "After sleep";
}

RPC timeout is set by DoTestSyncCall to 10 seconds.
With kSleep = 5s:

I1108 10:09:22.890683 23572 rpc-test.cc:174] Connecting to 127.0.0.1:36683
I1108 10:09:22.903359 23572 rpc-test.cc:181] Calls OK
I1108 10:09:27.903471 23572 rpc-test.cc:185] After sleep
I1108 10:09:28.060669 23577 yb_rpc.cc:227]     @     0x7f8abc4768a6  yb::rpc::YBInboundCall::~YBInboundCall() (/Users/timur/code/yugabyte5/src/yb/rpc/yb_rpc.cc:227)
    @     0x7f8abc4783f1  void __gnu_cxx::new_allocator<yb::rpc::YBInboundCall>::destroy<yb::rpc::YBInboundCall>(yb::rpc::YBInboundCall*) (gcc/5.5.0_4/include/c++/5.5.0/ext/new_allocator.h:124)
    @     0x7f8abc4783f1  void std::allocator_traits<std::allocator<yb::rpc::YBInboundCall> >::destroy<yb::rpc::YBInboundCall>(std::allocator<yb::rpc::YBInboundCall>&, yb::rpc::YBInboundCall*) (gcc/5.5.0_4/include/c++/5.5.0/bits/alloc_traits.h:542)
    @     0x7f8abc4783f1  std::_Sp_counted_ptr_inplace<yb::rpc::YBInboundCall, std::allocator<yb::rpc::YBInboundCall>, (__gnu_cxx::_Lock_policy)2>::_M_dispose() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:531)
    @     0x7f8abc3fb36d  std::_Sp_counted_base<(__gnu_cxx::_Lock_policy)2>::_M_release() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:150)
    @     0x7f8abc3fb36d  std::__shared_count<(__gnu_cxx::_Lock_policy)2>::~__shared_count() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:659)
    @     0x7f8abc3fb36d  std::__shared_ptr<yb::rpc::InboundCall, (__gnu_cxx::_Lock_policy)2>::~__shared_ptr() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:925)
    @     0x7f8abc3fb36d  std::__shared_ptr<yb::rpc::InboundCall, (__gnu_cxx::_Lock_policy)2>::reset() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:1022)
    @     0x7f8abc3fb36d  yb::rpc::InboundCall::UnretainSelf() (/Users/timur/code/yugabyte5/src/yb/rpc/inbound_call.cc:226)
    @     0x7f8abc460dcc  yb::rpc::ServicePoolImpl::StartShutdown()::{lambda()#1}::operator()() const (/Users/timur/code/yugabyte5/src/yb/rpc/service_pool.cc:161)

With kSleep=15s:

I1108 09:47:30.447520 20764 rpc-test.cc:174] Connecting to 127.0.0.1:40423
I1108 09:47:30.460654 20764 rpc-test.cc:181] Calls OK
I1108 09:47:40.779435 20769 yb_rpc.cc:227]     @     0x7f336be6d8a6  yb::rpc::YBInboundCall::~YBInboundCall() (/Users/timur/code/yugabyte5/src/yb/rpc/yb_rpc.cc:227)
    @     0x7f336be6f3f1  void __gnu_cxx::new_allocator<yb::rpc::YBInboundCall>::destroy<yb::rpc::YBInboundCall>(yb::rpc::YBInboundCall*) (gcc/5.5.0_4/include/c++/5.5.0/ext/new_allocator.h:124)
    @     0x7f336be6f3f1  void std::allocator_traits<std::allocator<yb::rpc::YBInboundCall> >::destroy<yb::rpc::YBInboundCall>(std::allocator<yb::rpc::YBInboundCall>&, yb::rpc::YBInboundCall*) (gcc/5.5.0_4/include/c++/5.5.0/bits/alloc_traits.h:542)
    @     0x7f336be6f3f1  std::_Sp_counted_ptr_inplace<yb::rpc::YBInboundCall, std::allocator<yb::rpc::YBInboundCall>, (__gnu_cxx::_Lock_policy)2>::_M_dispose() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:531)
    @     0x7f336bdf236d  std::_Sp_counted_base<(__gnu_cxx::_Lock_policy)2>::_M_release() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:150)
    @     0x7f336bdf236d  std::__shared_count<(__gnu_cxx::_Lock_policy)2>::~__shared_count() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:659)
    @     0x7f336bdf236d  std::__shared_ptr<yb::rpc::InboundCall, (__gnu_cxx::_Lock_policy)2>::~__shared_ptr() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:925)
    @     0x7f336bdf236d  std::__shared_ptr<yb::rpc::InboundCall, (__gnu_cxx::_Lock_policy)2>::reset() (gcc/5.5.0_4/include/c++/5.5.0/bits/shared_ptr_base.h:1022)
    @     0x7f336bdf236d  yb::rpc::InboundCall::UnretainSelf() (/Users/timur/code/yugabyte5/src/yb/rpc/inbound_call.cc:226)
    @     0x7f336be5ac42  yb::rpc::ServicePoolImpl::CheckTimeout(long, std::chrono::time_point<yb::CoarseMonoClock, std::chrono::duration<long, std::ratio<1l, 1000000000l> > >, yb::Status const&) (/Users/timur/code/yugabyte5/src/yb/rpc/service_pool.cc:313)

[bmatican]

Adding a screenshot of one of the experiments. I had set the raft timeout to 30s and it seems as if this made the InboundCall cleanup only kick in every 30s.