Lazy connections

Author: avsej

core/bucket.cxx

@@ -19,6 +19,7 @@
 
 #include "collection_id_cache_entry.hxx"
 #include "core/app_telemetry_meter.hxx"
+#include "core/cluster_options.hxx"
 #include "core/config_listener.hxx"
 #include "core/document_id.hxx"
 #include "core/error_context/key_value_error_map_info.hxx"
@@ -367,6 +368,53 @@ class bucket_impl
     return { 0, std::nullopt };
   }
 
+  void connect_session(std::size_t index)
+  {
+    const std::scoped_lock lock(config_mutex_, sessions_mutex_);
+    if (!config_) {
+      return;
+    }
+
+    const auto& node = config_->nodes[index];
+
+    const auto& hostname = node.hostname_for(origin_.options().network);
+    auto port = node.port_or(
+      origin_.options().network, service_type::key_value, origin_.options().enable_tls, 0);
+    if (port == 0) {
+      return;
+    }
+
+    const couchbase::core::origin origin(origin_.credentials(), hostname, port, origin_.options());
+    io::mcbp_session session =
+      origin_.options().enable_tls
+        ? io::mcbp_session(
+            client_id_, node.node_uuid, ctx_, tls_, origin, state_listener_, name_, known_features_)
+        : io::mcbp_session(
+            client_id_, node.node_uuid, ctx_, origin, state_listener_, name_, known_features_);
+    CB_LOG_DEBUG(R"({} rev={}, connect idx={}, session="{}", address="{}:{}")",
+                 log_prefix_,
+                 config_->rev_str(),
+                 node.index,
+                 session.id(),
+                 hostname,
+                 port);
+    session.bootstrap(
+      [self = shared_from_this(), session](std::error_code err,
+                                           topology::configuration cfg) mutable -> void {
+        if (err) {
+          return self->remove_session(session.id());
+        }
+        self->update_config(std::move(cfg));
+        session.on_configuration_update(self);
+        session.on_stop([id = session.id(), self]() -> void {
+          self->remove_session(id);
+        });
+        self->drain_deferred_queue({});
+      },
+      true);
+    sessions_.insert_or_assign(index, std::move(session));
+  }
+
   void restart_sessions()
   {
     const std::scoped_lock lock(config_mutex_, sessions_mutex_);
@@ -812,50 +860,56 @@ class bucket_impl
           continue;
         }
 
-        const couchbase::core::origin origin(
-          origin_.credentials(), hostname, port, origin_.options());
-        io::mcbp_session session =
-          origin_.options().enable_tls
-            ? io::mcbp_session(client_id_,
-                               node.node_uuid,
-                               ctx_,
-                               tls_,
-                               origin,
-                               state_listener_,
-                               name_,
-                               known_features_)
-            : io::mcbp_session(
-                client_id_, node.node_uuid, ctx_, origin, state_listener_, name_, known_features_);
-        CB_LOG_DEBUG(R"({} rev={}, add session="{}", address="{}:{}", index={})",
-                     log_prefix_,
-                     config.rev_str(),
-                     session.id(),
-                     hostname,
-                     port,
-                     node.index);
-        session.bootstrap(
-          [self = shared_from_this(), session, idx = next_index](
-            std::error_code err, topology::configuration cfg) mutable {
-            if (err) {
-              CB_LOG_WARNING(
-                R"({} failed to bootstrap session="{}", address="{}:{}", index={}, ec={})",
-                session.log_prefix(),
-                session.id(),
-                session.bootstrap_hostname(),
-                session.bootstrap_port(),
-                idx,
-                err.message());
-              return self->remove_session(session.id());
-            }
-            self->update_config(std::move(cfg));
-            session.on_configuration_update(self);
-            session.on_stop([id = session.id(), self]() {
-              self->remove_session(id);
-            });
-            self->drain_deferred_queue({});
-          },
-          true);
-        new_sessions.insert_or_assign(next_index, std::move(session));
+        if (!origin_.options().enable_lazy_connections) {
+          const couchbase::core::origin origin(
+            origin_.credentials(), hostname, port, origin_.options());
+          io::mcbp_session session = origin_.options().enable_tls
+                                       ? io::mcbp_session(client_id_,
+                                                          node.node_uuid,
+                                                          ctx_,
+                                                          tls_,
+                                                          origin,
+                                                          state_listener_,
+                                                          name_,
+                                                          known_features_)
+                                       : io::mcbp_session(client_id_,
+                                                          node.node_uuid,
+                                                          ctx_,
+                                                          origin,
+                                                          state_listener_,
+                                                          name_,
+                                                          known_features_);
+          CB_LOG_DEBUG(R"({} rev={}, add session="{}", address="{}:{}", index={})",
+                       log_prefix_,
+                       config.rev_str(),
+                       session.id(),
+                       hostname,
+                       port,
+                       node.index);
+          session.bootstrap(
+            [self = shared_from_this(), session, idx = next_index](
+              std::error_code err, topology::configuration cfg) mutable {
+              if (err) {
+                CB_LOG_WARNING(
+                  R"({} failed to bootstrap session="{}", address="{}:{}", index={}, ec={})",
+                  session.log_prefix(),
+                  session.id(),
+                  session.bootstrap_hostname(),
+                  session.bootstrap_port(),
+                  idx,
+                  err.message());
+                return self->remove_session(session.id());
+              }
+              self->update_config(std::move(cfg));
+              session.on_configuration_update(self);
+              session.on_stop([id = session.id(), self]() {
+                self->remove_session(id);
+              });
+              self->drain_deferred_queue({});
+            },
+            true);
+          new_sessions.insert_or_assign(next_index, std::move(session));
+        }
         ++next_index;
       }
       std::swap(sessions_, new_sessions);
@@ -1206,4 +1260,10 @@ bucket::direct_re_queue(const std::shared_ptr<mcbp::queue_request>& req, bool is
 {
   return impl_->direct_re_queue(req, is_retry);
 }
+
+void
+bucket::connect_session(std::size_t index)
+{
+  return impl_->connect_session(index);
+}
 } // namespace couchbase::core

core/bucket.hxx

@@ -105,6 +105,8 @@ public:
     });
   }
 
+  void connect_session(std::size_t index);
+
   template<typename Request>
   void map_and_send(std::shared_ptr<operations::mcbp_command<bucket, Request>> cmd)
   {
@@ -142,6 +144,9 @@ public:
         session.has_value() ? session->bootstrap_address() : "",
         session.has_value() && session->has_config(),
         config_rev());
+      if (!session) {
+          connect_session(index);
+      }
       return defer_command([self = shared_from_this(), cmd](std::error_code ec) {
         if (ec == errc::common::request_canceled) {
           return cmd->cancel(retry_reason::do_not_retry);

core/cluster_options.hxx

@@ -113,6 +113,7 @@ public:
   };
   bool preserve_bootstrap_nodes_order{ false };
   bool allow_enterprise_analytics{ false };
+  bool enable_lazy_connections{ false };
 };
 
 } // namespace couchbase::core

core/utils/connection_string.cxx

@@ -590,6 +590,8 @@ extract_options(connection_string& connstr)
       parse_option(connstr.options.preserve_bootstrap_nodes_order, name, value, connstr.warnings);
     } else if (name == "allow_enterprise_analytics") {
       parse_option(connstr.options.allow_enterprise_analytics, name, value, connstr.warnings);
+    } else if (name == "enable_lazy_connections") {
+      parse_option(connstr.options.enable_lazy_connections, name, value, connstr.warnings);
     } else {
       connstr.warnings.push_back(
         fmt::format(R"(unknown parameter "{}" in connection string (value "{}"))", name, value));