IGNITE-22141 C++ Client: Implement Partition Awareness (#7966)

Author: Erixonich

.teamcity/test/platform_tests/PlatformCppOdbcTestsDebLinux.kt

@@ -20,7 +20,6 @@ object PlatformCppOdbcTestsDebLinux : BuildType({
     """.trimIndent()
 
     params {
-        param("env.IGNITE_CPP_TESTS_USE_SINGLE_NODE", "")
         param("PATH__CMAKE_BUILD_DIRECTORY", "%PATH__WORKING_DIR%/cmake-build-debug")
         param("PATH__ODBC_TEST_RESULTS", "%PATH__WORKING_DIR%/odbc_tests_results.xml")
         text("PATH__WORKING_DIR", "%teamcity.build.checkoutDir%/%VCSROOT__IGNITE3%/modules/platforms/cpp", display = ParameterDisplay.HIDDEN, allowEmpty = true)

.teamcity/test/platform_tests/PlatformCppOdbcTestsRpmLinux.kt

@@ -20,7 +20,6 @@ object PlatformCppOdbcTestsRpmLinux : BuildType({
     """.trimIndent()
 
     params {
-        param("env.IGNITE_CPP_TESTS_USE_SINGLE_NODE", "")
         param("PATH__CMAKE_BUILD_DIRECTORY", "%PATH__WORKING_DIR%/cmake-build-debug")
         param("PATH__ODBC_TEST_RESULTS", "%PATH__WORKING_DIR%/odbc_tests_results.xml")
         text("PATH__WORKING_DIR", "%teamcity.build.checkoutDir%/%VCSROOT__IGNITE3%/modules/platforms/cpp", display = ParameterDisplay.HIDDEN, allowEmpty = true)

.teamcity/test/platform_tests/PlatformCppOdbcTestsTgzLinux.kt

@@ -20,7 +20,6 @@ object PlatformCppOdbcTestsTgzLinux : BuildType({
     """.trimIndent()
 
     params {
-        param("env.IGNITE_CPP_TESTS_USE_SINGLE_NODE", "")
         param("PATH__CMAKE_BUILD_DIRECTORY", "%PATH__WORKING_DIR%/cmake-build-debug")
         param("PATH__ODBC_TEST_RESULTS", "%PATH__WORKING_DIR%/odbc_tests_results.xml")
         text("PATH__WORKING_DIR", "%teamcity.build.checkoutDir%/%VCSROOT__IGNITE3%/modules/platforms/cpp", display = ParameterDisplay.HIDDEN, allowEmpty = true)

.teamcity/test/platform_tests/PlatformCppTestsLinux.kt

@@ -24,7 +24,6 @@ object PlatformCppTestsLinux : BuildType({
     """.trimIndent()
 
     params {
-        param("env.IGNITE_CPP_TESTS_USE_SINGLE_NODE", "")
         param("PATH__CMAKE_BUILD_DIRECTORY", "%PATH__WORKING_DIR%/cmake-build-debug")
         param("PATH__CLIENT_TEST_RESULTS", "%PATH__WORKING_DIR%/cpp_client_tests_results.xml")
         param("PATH__UNIT_TESTS_RESULT", "%PATH__WORKING_DIR%/cpp_unit_test_results.xml")

.teamcity/test/platform_tests/PlatformCppTestsWindows.kt

@@ -27,7 +27,6 @@ object PlatformCppTestsWindows : BuildType({
     """.trimIndent()
 
     params {
-        hiddenText("env.IGNITE_CPP_TESTS_USE_SINGLE_NODE", "")
         hiddenText("PATH__CMAKE_BUILD_DIRECTORY", """%PATH__WORKING_DIR%\cmake-build-debug""")
         hiddenText("PATH__CLIENT_TEST_RESULTS", """%PATH__CMAKE_BUILD_DIRECTORY%\cpp_client_tests_results.xml""")
         hiddenText("PATH__ODBC_TEST_RESULTS", """%PATH__CMAKE_BUILD_DIRECTORY%\odbc_tests_results.xml""")

modules/platforms/cpp/ignite/client/detail/cluster_connection.cpp

@@ -92,7 +92,9 @@ void cluster_connection::stop() {
 
 void cluster_connection::on_connection_success(const end_point &addr, uint64_t id) {
     m_logger->log_info("Established connection with remote host " + addr.to_string());
-    m_logger->log_debug("Connection ID: " + std::to_string(id));
+
+    if (m_logger->is_debug_enabled())
+        m_logger->log_debug("Connection ID: " + std::to_string(id));
 
     auto connection = node_connection::make_new(
         id, m_pool, weak_from_this(), m_logger, m_configuration, m_timer_thread);
@@ -214,6 +216,16 @@ void cluster_connection::on_observable_timestamp_changed(std::int64_t timestamp)
     }
 }
 
+void cluster_connection::on_partition_assignment_changed(std::int64_t timestamp) {
+    auto expected = m_assignment_timestamp.load();
+    while (expected < timestamp) {
+        auto success = m_assignment_timestamp.compare_exchange_weak(expected, timestamp);
+        if (success)
+            return;
+        expected = m_assignment_timestamp.load();
+    }
+}
+
 void cluster_connection::remove_client(uint64_t id) {
     [[maybe_unused]] std::unique_lock<std::recursive_mutex> lock(m_connections_mutex);
 
@@ -262,9 +274,28 @@ std::shared_ptr<node_connection> cluster_connection::get_random_connected_channe
     return std::next(m_connections.begin(), idx)->second;
 }
 
+std::shared_ptr<node_connection> cluster_connection::get_channel(
+    const std::optional<std::string>& preferred_node_name) {
+
+    if (preferred_node_name) {
+        std::unique_lock lock(m_connections_mutex);
+        for (auto& [id, conn] : m_connections) {
+            if (conn->get_node_name() == *preferred_node_name) {
+                return conn;
+            }
+        }
+    }
+
+    return get_random_connected_channel();
+}
+
 std::pair<std::shared_ptr<node_connection>, std::int64_t> cluster_connection::perform_request_handler(
-    const operation_function_type &op_func, transaction_impl *tx, const writer_function_type &wr,
-    const std::shared_ptr<response_handler> &handler) {
+    const operation_function_type &op_func,
+    transaction_impl *tx,
+    const writer_function_type &wr,
+    const std::shared_ptr<response_handler> &handler,
+    const std::optional<std::string>& preferred_node_name) {
+
     if (tx) {
         auto channel = tx->get_connection();
         if (!channel)
@@ -279,7 +310,7 @@ std::pair<std::shared_ptr<node_connection>, std::int64_t> cluster_connection::pe
     }
 
     while (true) {
-        auto channel = get_random_connected_channel();
+        auto channel = get_channel(preferred_node_name);
         if (!channel)
             throw ignite_error(error::code::CONNECTION, "No nodes connected");
 
@@ -290,10 +321,15 @@ std::pair<std::shared_ptr<node_connection>, std::int64_t> cluster_connection::pe
     }
 }
 
-void cluster_connection::perform_request_raw(protocol::client_operation op, transaction_impl *tx,
-    const writer_function_type &wr, ignite_callback<bytes_view> callback) {
+void cluster_connection::perform_request_raw(
+    protocol::client_operation op,
+    transaction_impl *tx,
+    const writer_function_type &wr,
+    ignite_callback<bytes_view> callback,
+    const std::optional<std::string>& preferred_node_name) {
     auto handler = std::make_shared<response_handler_raw>(std::move(callback));
-    perform_request_handler(static_op(op), tx, wr, std::move(handler));
+
+    perform_request_handler(static_op(op), tx, wr, std::move(handler), preferred_node_name);
 }
 
 } // namespace ignite::detail

modules/platforms/cpp/ignite/client/detail/cluster_connection.h

@@ -106,10 +106,15 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
      * @param tx Transaction.
      * @param wr Request writer function.
      * @param handler Request handler.
+     * @param preferred_node_name Name of preferred node.
      * @return A connection used to perform request and the request ID.
      */
-    std::pair<std::shared_ptr<node_connection>, std::int64_t> perform_request_handler(const operation_function_type &op_func,
-        transaction_impl *tx, const writer_function_type &wr, const std::shared_ptr<response_handler> &handler);
+    std::pair<std::shared_ptr<node_connection>, std::int64_t> perform_request_handler(
+        const operation_function_type &op_func,
+        transaction_impl *tx,
+        const writer_function_type &wr,
+        const std::shared_ptr<response_handler> &handler,
+        const std::optional<std::string>& preferred_node_name = std::nullopt);
 
     /**
      * Perform request raw.
@@ -119,9 +124,14 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
      * @param tx Transaction.
      * @param wr Request writer function.
      * @param callback Callback to call on a result.
+     * @param preferred_node_name Name of preferred node.
      */
-    void perform_request_raw(protocol::client_operation op, transaction_impl *tx,
-        const writer_function_type &wr, ignite_callback<bytes_view> callback);
+    void perform_request_raw(
+        protocol::client_operation op,
+        transaction_impl *tx,
+        const writer_function_type &wr,
+        ignite_callback<bytes_view> callback,
+        const std::optional<std::string>& preferred_node_name = std::nullopt);
 
     /**
      * Perform request raw.
@@ -151,13 +161,15 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
      * @param wr Request writer function.
      * @param rd Response reader function.
      * @param callback Callback to call on a result.
+     * @param preferred_node_name Name of preferred node.
      */
     template<typename T>
     std::pair<std::shared_ptr<node_connection>, std::int64_t> perform_request(protocol::client_operation op,
         transaction_impl *tx, const writer_function_type &wr,
-        reader_function_type<T> rd, ignite_callback<T> callback) {
+        reader_function_type<T> rd, ignite_callback<T> callback,
+        const std::optional<std::string>& preferred_node_name = std::nullopt) {
         auto handler = std::make_shared<response_handler_reader<T>>(std::move(rd), std::move(callback));
-        return perform_request_handler(static_op(op), tx, wr, std::move(handler));
+        return perform_request_handler(static_op(op), tx, wr, std::move(handler), preferred_node_name);
     }
 
     /**
@@ -189,7 +201,7 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
     void perform_request(protocol::client_operation op, const writer_function_type &wr,
         std::function<T(protocol::reader &, std::shared_ptr<node_connection>)> rd, ignite_callback<T> callback) {
         auto handler = std::make_shared<response_handler_reader_connection<T>>(std::move(rd), std::move(callback));
-        perform_request_handler(static_op(op), nullptr, wr, std::move(handler));
+        perform_request_handler(static_op(op), nullptr, wr, std::move(handler), std::nullopt);
     }
 
     /**
@@ -261,9 +273,13 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
      * @return A connection used to perform request and the request ID.
      */
     template<typename T>
-    std::pair<std::shared_ptr<node_connection>, std::int64_t> perform_request_wr(protocol::client_operation op,
-        transaction_impl *tx, const writer_function_type &wr, ignite_callback<T> callback) {
-        return perform_request<T>(op, tx, wr, [](protocol::reader &) {}, std::move(callback));
+    std::pair<std::shared_ptr<node_connection>, std::int64_t> perform_request_wr(
+        protocol::client_operation op,
+        transaction_impl *tx,
+        const writer_function_type &wr,
+        ignite_callback<T> callback,
+        const std::optional<std::string>& preferred_node_name = std::nullopt) {
+        return perform_request<T>(op, tx, wr, [](protocol::reader &) {}, std::move(callback), preferred_node_name);
     }
 
     /**
@@ -273,6 +289,20 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
      */
     std::int64_t get_observable_timestamp() const { return m_observable_timestamp.load(); }
 
+    /**
+     * Get assignment timestamp.
+     *
+     * @return Assignment timestamp.
+     */
+    std::int64_t get_assignment_timestamp() const { return m_assignment_timestamp.load(); }
+
+    /**
+     * Get logger.
+     *
+     * @return Logger.
+     */
+    [[nodiscard]] std::shared_ptr<ignite_logger> get_logger() const { return m_logger; }
+
     /**
      * @param op Operation code to return.
      * @return A function that always returns the same operation.
@@ -289,6 +319,14 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
      */
     std::shared_ptr<node_connection> get_random_connected_channel();
 
+    /**
+     * Get connection according to provided preference otherwise returns random node connection.
+     *
+     * @param preferred_node_name Name of preferred node.
+     * @return Node connection.
+     */
+    std::shared_ptr<node_connection> get_channel(const std::optional<std::string> &preferred_node_name);
+
     /**
      * Constructor.
      *
@@ -342,6 +380,13 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
      */
     void on_observable_timestamp_changed(std::int64_t timestamp) override;
 
+    /**
+     * Handle partition assignment change.
+     *
+     * @param timestamp Assignment timestamp.
+     */
+    void on_partition_assignment_changed(std::int64_t timestamp) override;
+
     /**
      * Remove client.
      *
@@ -404,6 +449,9 @@ class cluster_connection : public std::enable_shared_from_this<cluster_connectio
     /** Observable timestamp. */
     std::atomic_int64_t m_observable_timestamp{0};
 
+    /** Partition assignment timestamp. */
+    std::atomic_int64_t m_assignment_timestamp{0};
+
     /** Timer thread. */
     std::shared_ptr<thread_timer> m_timer_thread;
 };

modules/platforms/cpp/ignite/client/detail/connection_event_handler.h

@@ -44,6 +44,13 @@ class connection_event_handler {
      * @param timestamp Timestamp.
      */
     virtual void on_observable_timestamp_changed(std::int64_t timestamp) = 0;
+
+    /**
+     * Handle partition assignment change.
+     *
+     * @param timestamp Assignment timestamp.
+     */
+    virtual void on_partition_assignment_changed(std::int64_t timestamp) = 0;
 };
 
 } // namespace ignite::detail

modules/platforms/cpp/ignite/client/detail/node_connection.cpp

@@ -69,9 +69,12 @@ void node_connection::process_message(bytes_view msg) {
 
     auto req_id = reader.read_int64();
     auto flags = reader.read_int32();
+    auto event_handler = m_event_handler.lock();
     if (test_flag(flags, protocol::response_flag::PARTITION_ASSIGNMENT_CHANGED)) {
         auto assignment_ts = reader.read_int64();
-        UNUSED_VALUE assignment_ts;
+        if (event_handler) {
+            event_handler->on_partition_assignment_changed(assignment_ts);
+        }
     }
 
     auto observable_timestamp = reader.read_int64();
@@ -180,6 +183,8 @@ ignite_result<void> node_connection::process_handshake_rsp(bytes_view msg) {
         std::chrono::milliseconds(response.idle_timeout_ms));
 
     m_protocol_context = response.context;
+    m_node_id = response.node_id;
+    m_node_name = response.node_name;
     m_handshake_complete = true;
 
     if (m_heartbeat_interval.count()) {

modules/platforms/cpp/ignite/client/detail/node_connection.h

@@ -232,8 +232,20 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
      */
     std::shared_ptr<ignite_logger> get_logger() const { return m_logger; }
 
+    /**
+     * Cancels waiting for over-due responses.
+     */
     void handle_timeouts();
 
+    /**
+     * Name of the node this connection is tethered to.
+     *
+     * @return Name of the node.
+     */
+    const std::string& get_node_name() const {
+        return m_node_name;
+    }
+
 private:
     /**
      * Constructor.
@@ -334,6 +346,12 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
 
     /** Timer thread. */
     std::weak_ptr<thread_timer> m_timer_thread;
+
+    /** Node id. */
+    uuid m_node_id{};
+
+    /** Name of the node this connection is tethered to. */
+    std::string m_node_name{};
 };
 
 } // namespace ignite::detail