ENG-2709 Allow disabling GC for idle connection.

Summary:
Redis keeps its connections open indefinitely, even if there is no
activity. Or default of closing idle connections after 65s breaks
the expected default behavior.

Will be useful to be able to disable this, if desired.

if rpc_default_keepalive_time_ms is set to 0, we disable closing idle connections.

Test Plan:
set rpc_default_keepalive_time_ms to 0 for a local cluster.
Verify logs to ensure that the connections are not gc-ed.

Reviewers: kannan

Reviewed By: kannan

Subscribers: robert, ybase

Differential Revision: https://phabricator.dev.yugabyte.com/D3865

src/yb/rpc/messenger.cc

@@ -78,7 +78,7 @@ using strings::Substitute;
 DECLARE_int32(num_connections_to_server);
 DEFINE_int32(rpc_default_keepalive_time_ms, 65000,
              "If an RPC connection from a client is idle for this amount of time, the server "
-             "will disconnect the client.");
+             "will disconnect the client. Setting flag to 0 disables this clean up.");
 TAG_FLAG(rpc_default_keepalive_time_ms, advanced);
 DEFINE_uint64(io_thread_pool_size, 4, "Size of allocated IO Thread Pool.");
 

src/yb/rpc/reactor.cc

@@ -368,6 +368,11 @@ void Reactor::RegisterTimeout(ev::timer *watcher) {
 
 void Reactor::ScanIdleConnections() {
   DCHECK(IsCurrentThread());
+  if (connection_keepalive_time_ == CoarseMonoClock::Duration::zero()) {
+    VLOG(3) << "Skipping Idle connections check since connection_keepalive_time_ = 0";
+    return;
+  }
+
   // enforce TCP connection timeouts
   auto c = server_conns_.begin();
   auto c_end = server_conns_.end();

src/yb/server/rpc_server.cc

@@ -67,14 +67,16 @@ TAG_FLAG(rpc_server_allow_ephemeral_ports, unsafe);
 
 DEFINE_int32(rpc_queue_limit, 5000, "Queue limit for rpc server");
 DEFINE_int32(rpc_workers_limit, 128, "Workers limit for rpc server");
+DECLARE_int32(rpc_default_keepalive_time_ms);
 
 namespace yb {
 
 RpcServerOptions::RpcServerOptions()
   : rpc_bind_addresses(FLAGS_rpc_bind_addresses),
     default_port(0),
     queue_limit(FLAGS_rpc_queue_limit),
-    workers_limit(FLAGS_rpc_workers_limit) {
+    workers_limit(FLAGS_rpc_workers_limit),
+    connection_keepalive_time_ms(FLAGS_rpc_default_keepalive_time_ms) {
 }
 
 RpcServer::RpcServer(const std::string& name, RpcServerOptions opts)

src/yb/server/rpc_server.h

@@ -53,6 +53,7 @@ struct RpcServerOptions {
   uint16_t default_port;
   size_t queue_limit;
   size_t workers_limit;
+  int32_t connection_keepalive_time_ms;
 };
 
 class RpcServer {

src/yb/server/server_base.cc

@@ -91,6 +91,7 @@ TAG_FLAG(generic_svc_queue_length, advanced);
 DEFINE_string(yb_test_name, "",
               "Specifies test name this daemon is running as part of.");
 
+using namespace std::literals;
 using std::shared_ptr;
 using std::string;
 using std::stringstream;
@@ -187,6 +188,7 @@ Status RpcServerBase::Init() {
 
   builder.set_num_reactors(FLAGS_num_reactor_threads);
   builder.set_metric_entity(metric_entity());
+  builder.set_connection_keepalive_time(options_.rpc_opts.connection_keepalive_time_ms * 1ms);
   builder.use_connection_context_factory(connection_context_factory_);
   RETURN_NOT_OK(builder.Build().MoveTo(&messenger_));
 

src/yb/yql/cql/cqlserver/cql_server_options.cc

@@ -13,15 +13,22 @@
 
 #include "yb/yql/cql/cqlserver/cql_server_options.h"
 
+#include "yb/util/flag_tags.h"
 #include "yb/yql/cql/cqlserver/cql_rpc.h"
 #include "yb/yql/cql/cqlserver/cql_server.h"
 
+DEFINE_int32(cql_rpc_keepalive_time_ms, 120000,
+             "If an RPC connection from a client is idle for this amount of time, the server "
+             "will disconnect the client. Setting flag to 0 disables this clean up.");
+TAG_FLAG(cql_rpc_keepalive_time_ms, advanced);
+
 namespace yb {
 namespace cqlserver {
 
 CQLServerOptions::CQLServerOptions() {
   server_type = "tserver";
   rpc_opts.default_port = CQLServer::kDefaultPort;
+  rpc_opts.connection_keepalive_time_ms = FLAGS_cql_rpc_keepalive_time_ms;
   connection_context_factory = &std::make_unique<CQLConnectionContext>;
 }
 

src/yb/yql/redis/redisserver/redis_server_options.cc

@@ -13,15 +13,22 @@
 
 #include "yb/yql/redis/redisserver/redis_server_options.h"
 
+#include "yb/util/flag_tags.h"
 #include "yb/yql/redis/redisserver/redis_rpc.h"
 #include "yb/yql/redis/redisserver/redis_server.h"
 
+DEFINE_int32(redis_rpc_keepalive_time_ms, 0,
+             "If an RPC connection from a client is idle for this amount of time, the server "
+             "will disconnect the client. Setting flag to 0 disables this clean up.");
+TAG_FLAG(redis_rpc_keepalive_time_ms, advanced);
+
 namespace yb {
 namespace redisserver {
 
 RedisServerOptions::RedisServerOptions() {
   server_type = "tserver";
   rpc_opts.default_port = RedisServer::kDefaultPort;
+  rpc_opts.connection_keepalive_time_ms = FLAGS_redis_rpc_keepalive_time_ms;
   connection_context_factory = &std::make_unique<RedisConnectionContext>;
 }