IGNITE-27056 Add request timeouts to C++ client

modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java

@@ -282,6 +282,9 @@ public static class Client {
 
         /** Resource not found. */
         public static final int RESOURCE_NOT_FOUND_ERR = CLIENT_ERR_GROUP.registerErrorCode((short) 10);
+
+        /** Client operation timeout. */
+        public static final int OPERATION_TIMEOUT_ERR = CLIENT_ERR_GROUP.registerErrorCode((short) 11);
     }
 
     /** SQL error group. */

modules/platforms/cpp/CMakeLists.txt

@@ -151,6 +151,10 @@ if (${ENABLE_TESTS})
 
     if (${ENABLE_CLIENT})
         add_subdirectory(tests/client-test)
+
+        if (NOT MSVC) # TODO enable when IGNITE-27213 implemented
+            add_subdirectory(tests/fake_server)
+        endif ()
     endif()
 
     if (${ENABLE_ODBC})

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

@@ -16,7 +16,6 @@
  */
 
 #include "ignite/client/detail/node_connection.h"
-#include "ignite/common/detail/duration_min_max.h"
 #include "ignite/protocol/heartbeat_timeout.h"
 
 #include <ignite/protocol/messages.h>
@@ -35,9 +34,10 @@ node_connection::node_connection(std::uint64_t id, std::shared_ptr<network::asyn
     , m_timer_thread(std::move(timer_thread)){}
 
 node_connection::~node_connection() {
-    for (auto &handler : m_request_handlers) {
+    for (auto &req : m_request_handlers) {
         auto handling_res = result_of_operation<void>([&]() {
-            auto res = handler.second->set_error(ignite_error("Connection closed before response was received"));
+            auto handler = req.second.handler;
+            auto res = handler->set_error(ignite_error("Connection closed before response was received"));
             if (res.has_error())
                 m_logger->log_error(
                     "Uncaught user callback exception while handling operation error: " + res.error().what_str());
@@ -182,6 +182,10 @@ ignite_result<void> node_connection::process_handshake_rsp(bytes_view msg) {
         plan_heartbeat(m_heartbeat_interval);
     }
 
+    if (m_configuration.get_operation_timeout().count() > 0) {
+        handle_timeouts();
+    }
+
     return {};
 }
 
@@ -192,7 +196,7 @@ std::shared_ptr<response_handler> node_connection::get_and_remove_handler(std::i
     if (it == m_request_handlers.end())
         return {};
 
-    auto res = std::move(it->second);
+    auto res = std::move(it->second.handler);
     m_request_handlers.erase(it);
 
     return res;
@@ -203,7 +207,45 @@ std::shared_ptr<response_handler> node_connection::find_handler_unsafe(std::int6
     if (it == m_request_handlers.end())
         return {};
 
-    return it->second;
+    return it->second.handler;
+}
+
+void node_connection::handle_timeouts() {
+    auto now = std::chrono::steady_clock::now();
+
+    {
+        std::lock_guard lock(m_request_handlers_mutex);
+
+        for (auto it = m_request_handlers.begin(); it != m_request_handlers.end();) {
+            auto &id = it->first;
+            auto &req = it->second;
+
+            if (req.timeouts_at && *req.timeouts_at < now) {
+                std::string err_msg = "Operation timeout [req_id=" + std::to_string(id) + "]";
+                auto res = req.handler->set_error(ignite_error(error::code::OPERATION_TIMEOUT, err_msg));
+
+                this->m_logger->log_warning(err_msg);
+
+                if (res.has_error())
+                    this->m_logger->log_error(
+                        "Uncaught user callback exception while handling operation error: " + res.error().what_str());
+
+                it = m_request_handlers.erase(it);
+            } else {
+                ++it;
+            }
+        }
+    }
+
+    if (auto timeout = m_configuration.get_operation_timeout(); timeout.count() > 0) {
+        if (auto timer_thread = m_timer_thread.lock()) {
+            timer_thread->add(timeout, [self_weak = weak_from_this()] {
+                if (auto self = self_weak.lock()) {
+                    self->handle_timeouts();
+                }
+            });
+        }
+    }
 }
 
 } // namespace ignite::detail

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

@@ -49,6 +49,23 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
 public:
     typedef std::function<void(protocol::writer&, const protocol::protocol_context&)> writer_function_type;
 
+    struct pending_request {
+        /** Handler function for request */
+        std::shared_ptr<response_handler> handler{};
+
+        /**
+         * Optional request timeout.
+         * When provided contains time point after which request would be considered as timed out.
+         */
+        std::optional<std::chrono::time_point<std::chrono::steady_clock>> timeouts_at{};
+
+        explicit pending_request(
+            std::shared_ptr<response_handler> handler,
+            std::optional<std::chrono::time_point<std::chrono::steady_clock>> timeouts_at)
+            : handler(std::move(handler))
+            , timeouts_at(timeouts_at) {}
+    };
+
     // Deleted
     node_connection() = delete;
     node_connection(node_connection &&) = delete;
@@ -103,8 +120,11 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
      * @param handler response handler.
      * @return A request ID on success and @c std::nullopt otherwise.
      */
-    [[nodiscard]] std::optional<std::int64_t> perform_request(protocol::client_operation op, const writer_function_type &wr,
-        std::shared_ptr<response_handler> handler) {
+    [[nodiscard]] std::optional<std::int64_t> perform_request(
+        protocol::client_operation op,
+        const writer_function_type &wr,
+        std::shared_ptr<response_handler> handler)
+    {
         auto req_id = generate_request_id();
         std::vector<std::byte> message;
         {
@@ -119,9 +139,14 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
             buffer.write_length_header();
         }
 
+        std::optional<std::chrono::time_point<std::chrono::steady_clock>> timeout{};
+        if (m_configuration.get_operation_timeout().count() > 0) {
+            timeout = std::chrono::steady_clock::now() + m_configuration.get_operation_timeout();
+        }
+
         {
             std::lock_guard<std::recursive_mutex> lock(m_request_handlers_mutex);
-            m_request_handlers[req_id] = std::move(handler);
+            m_request_handlers.emplace(req_id, pending_request(std::move(handler), timeout));
         }
 
         if (m_logger->is_debug_enabled()) {
@@ -134,6 +159,7 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
             get_and_remove_handler(req_id);
             return {};
         }
+
         return {req_id};
     }
 
@@ -148,8 +174,11 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
      * @return Channel used for the request.
      */
     template<typename T>
-    [[nodiscard]] std::optional<std::int64_t> perform_request(protocol::client_operation op, const writer_function_type &wr,
-        std::function<T(protocol::reader &)> rd, ignite_callback<T> callback) {
+    [[nodiscard]] std::optional<std::int64_t> perform_request(
+        protocol::client_operation op,
+        const writer_function_type &wr,
+        std::function<T(protocol::reader &)> rd,
+        ignite_callback<T> callback) {
         auto handler = std::make_shared<response_handler_reader<T>>(std::move(rd), std::move(callback));
         return perform_request(op, wr, std::move(handler));
     }
@@ -203,6 +232,8 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
      */
     std::shared_ptr<ignite_logger> get_logger() const { return m_logger; }
 
+    void handle_timeouts();
+
 private:
     /**
      * Constructor.
@@ -290,7 +321,7 @@ class node_connection : public std::enable_shared_from_this<node_connection> {
     std::atomic_int64_t m_req_id_gen{0};
 
     /** Pending request handlers. */
-    std::unordered_map<std::int64_t, std::shared_ptr<response_handler>> m_request_handlers;
+    std::unordered_map<std::int64_t, pending_request> m_request_handlers;
 
     /** Handlers map mutex. */
     std::recursive_mutex m_request_handlers_mutex;

modules/platforms/cpp/ignite/client/ignite_client_configuration.h

@@ -47,6 +47,11 @@ class ignite_client_configuration {
      */
     static constexpr std::chrono::milliseconds DEFAULT_HEARTBEAT_INTERVAL = std::chrono::seconds(30);
 
+    /**
+     * Default operation
+     */
+    static constexpr std::chrono::milliseconds DEFAULT_OPERATION_TIMEOUT{0};
+
     // Default
     ignite_client_configuration() = default;
 
@@ -187,6 +192,39 @@ class ignite_client_configuration {
         m_heartbeat_interval = heartbeat_interval;
     }
 
+    /**
+     * Gets the operation timeout. Default is 0 (no timeout).
+     *
+     * An "operation" is a single client request to the server. Some public API calls may involve multiple operations,
+     * in which case the operation timeout is applied to each individual network call.
+     *
+     * Notably compute job execution consists of two calls. First request: submit, when server responds with success
+     * that means that job has been stored in the executor's queue and will be executed at some point.
+     * Second request: get_result, client requests job status if it was executed, failed or canceled.
+     * This configuration only applies to each of those requests separately but not to its combination therefore real
+     * execution times of compute jobs could be greater than operation timeout.
+     *
+     * @return Operation timeout
+     * .
+     */
+    [[nodiscard]] std::chrono::milliseconds get_operation_timeout() const { return m_operation_timeout; }
+
+    /**
+     * Sets the operation timeout.
+     *
+     * See @ref get_operation_timeout() for details.
+     *
+     * @param operation_timeout Operation timeout.
+     */
+    void set_operation_timeout(std::chrono::milliseconds operation_timeout) {
+        if (operation_timeout.count() < 0) {
+            throw ignite_error(error::code::ILLEGAL_ARGUMENT, "Operation timeout can't be negative: "
+                + std::to_string(operation_timeout.count()) + " milliseconds");
+        }
+
+        m_operation_timeout = operation_timeout;
+    }
+
     /**
      * Gets the authenticator.
      *
@@ -306,6 +344,9 @@ class ignite_client_configuration {
     /** Heartbeat interval. */
     std::chrono::milliseconds m_heartbeat_interval{DEFAULT_HEARTBEAT_INTERVAL};
 
+    /** Operation timeout. */
+    std::chrono::milliseconds m_operation_timeout{DEFAULT_OPERATION_TIMEOUT};
+
     /** SSL Mode. */
     ssl_mode m_ssl_mode{ssl_mode::DISABLE};
 

modules/platforms/cpp/ignite/common/error_codes.h

@@ -95,6 +95,7 @@ enum class code : underlying_t {
     HANDSHAKE_HEADER = 0x30008,
     SERVER_TO_CLIENT_REQUEST = 0x30009,
     RESOURCE_NOT_FOUND = 0x3000a,
+    OPERATION_TIMEOUT = 0x3000b,
 
     // Sql group. Group code: 4
     QUERY_NO_RESULT_SET = 0x40001,

modules/platforms/cpp/ignite/odbc/common_types.cpp

@@ -153,6 +153,7 @@ sql_state error_code_to_sql_state(error::code code) {
         case error::code::HANDSHAKE_HEADER:
             return sql_state::S08004_CONNECTION_REJECTED;
         case error::code::RESOURCE_NOT_FOUND:
+        case error::code::OPERATION_TIMEOUT:
             return sql_state::SHY000_GENERAL_ERROR;
 
         // Sql group. Group code: 4

modules/platforms/cpp/ignite/protocol/buffer_adapter.h

@@ -61,7 +61,7 @@ class buffer_adapter {
      */
     void reserve_length_header() {
         m_length_pos = m_buffer.size();
-        m_buffer.insert(m_buffer.end(), 4, std::byte{0});
+        m_buffer.insert(m_buffer.end(), LENGTH_HEADER_SIZE, std::byte{0});
     }
 
     /**

modules/platforms/cpp/tests/client-test/CMakeLists.txt

@@ -23,7 +23,6 @@ set(SOURCES
     basic_authenticator_test.cpp
     column_order_test.cpp
     compute_test.cpp
-    gtest_logger.h
     ignite_client_test.cpp
     ignite_runner_suite.h
     key_value_binary_view_test.cpp

modules/platforms/cpp/tests/client-test/ignite_client_test.cpp

@@ -62,6 +62,23 @@ TEST_F(client_test, configuration_set_invalid_heartbeat) {
         ignite_error);
 }
 
+TEST_F(client_test, configuration_set_invalid_operation_timeout) {
+    using namespace std::chrono_literals;
+
+    auto cfg = create_default_client_config();
+
+    EXPECT_THROW(
+        {
+            try {
+                cfg.set_operation_timeout(-1s);
+            } catch (const ignite_error &e) {
+                EXPECT_THAT(e.what_str(), testing::HasSubstr("Operation timeout can't be negative"));
+                throw;
+            }
+        },
+        ignite_error);
+}
+
 TEST_F(client_test, configuration_set_empty_address_constructor) {
     EXPECT_THROW(
         {
@@ -172,5 +189,3 @@ TEST_F(client_test, heartbeat_disable_connection_is_closed) {
         },
         ignite_error);
 }
-
-

modules/platforms/cpp/tests/compatibility-tests/main.cpp

@@ -58,7 +58,7 @@ void set_process_abort_handler(std::function<void(int)> handler) {
 
 using namespace ignite;
 
-const std::vector<std::string> DEFAULT_VERSIONS = {"9.1.0", "9.1.1", "9.1.2", "9.1.3", "9.1.4"};
+const std::vector<std::string> DEFAULT_VERSIONS = {"3.0.0", "3.1.0"};
 
 /**
  * Structure to store argument values for automatic memory management.

modules/platforms/cpp/tests/fake_server/CMakeLists.txt

@@ -0,0 +1,29 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+project(ignite-fake-server)
+
+set(TARGET ${PROJECT_NAME})
+
+set(SOURCES
+    main.cpp
+    fake_server.cpp
+    tcp_client_channel.cpp
+        connection_test.cpp
+)
+
+ignite_test(${TARGET} SOURCES ${SOURCES} LIBS ignite-test-common ignite3-client)