Copilot commented on code in PR #7080:
URL: https://github.com/apache/ignite-3/pull/7080#discussion_r2566749364
##########
modules/platforms/cpp/ignite/client/detail/node_connection.cpp:
##########
@@ -203,7 +208,46 @@ 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);
+
+ std::vector<int64_t> keys_for_erasure;
+
+ for (auto& [id, req] : m_request_handlers) {
+ if (req.timeouts_at > now) {
Review Comment:
The comparison logic is inverted. This should be `req.timeouts_at < now` to
check if the request has timed out (i.e., the timeout time point is in the
past). Currently, it marks requests as timed out when they haven't expired yet.
```suggestion
if (req.timeouts_at < now) {
```
##########
modules/platforms/cpp/tests/fake_server/fake_server.h:
##########
@@ -0,0 +1,330 @@
+/*
+* 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.
+ */
+
+#pragma once
+
+#include "ignite/protocol/bitmask_feature.h"
+#include "ignite/protocol/client_operation.h"
+#include "ignite/protocol/protocol_version.h"
+#include "ignite/protocol/reader.h"
+#include "ignite/protocol/writer.h"
+#include "response_action.h"
+#include "tcp_client_channel.h"
+
+#include <atomic>
+#include <cstring>
+#include <ignite/common/ignite_error.h>
+#include <ignite/protocol/utils.h>
+#include <iostream>
+#include <queue>
+#include <thread>
+#include <unistd.h>
+
+using namespace ignite;
+
+using raw_msg = std::vector<std::byte>;
+
+class fake_server {
+public:
+ explicit fake_server(
+ int srv_port = 10800,
+
std::function<std::unique_ptr<response_action>(protocol::client_operation)>
op_type_handler = nullptr
+ )
+ : m_srv_port(srv_port)
+ , m_op_type_handler(op_type_handler)
+ {}
+
+ ~fake_server() {
+ m_stopped.store(true);
+
+ if (m_started)
+ m_client_channel->stop();
+
+ if (m_srv_fd > 0) {
+ ::close(m_srv_fd);
+ }
+
+ m_io_thread->join();
+ }
+
+ void start() {
+ start_socket();
+
+ bind_address_port();
+
+ start_socket_listen();
+
+ m_io_thread = std::make_unique<std::thread>([this] {
+ accept_client_connection();
+
+ m_started.store(true);
+
+ handle_client_handshake();
+
+ send_server_handshake();
+
+ handle_requests();
+ });
+
+ std::cout << "fake server started" << std::endl;
+ }
+
+private:
+ int m_srv_fd = -1;
+ std::atomic_bool m_started{false};
+ std::atomic_bool m_stopped{false};
+ const int m_srv_port;
+ std::unique_ptr<std::thread> m_io_thread{};
+ std::unique_ptr<tcp_client_channel> m_client_channel{};
+
+ /**
+ * Allows developer to define custom action on requests according to their
type.
+ */
+
std::function<std::unique_ptr<response_action>(protocol::client_operation)>
m_op_type_handler;
+
+ void start_socket() {
+ m_srv_fd = socket(AF_INET, SOCK_STREAM, 6);
+
+ if (m_srv_fd < 0) {
+ throw ignite_error("socket failed");
+ }
+ }
+
+ void bind_address_port() const {
+ sockaddr_in srv_addr{};
+
+ srv_addr.sin_family = AF_INET;
+ srv_addr.sin_addr.s_addr = INADDR_ANY;
+ srv_addr.sin_port = htons(m_srv_port);
+
+ int bind_res = bind(m_srv_fd, reinterpret_cast<sockaddr *>(&srv_addr),
sizeof(srv_addr));
+
+ if (bind_res < 0) {
+ str:
Review Comment:
Unused label 'str:'. This appears to be a leftover from debugging or an
incomplete refactoring and should be removed.
```suggestion
```
##########
modules/platforms/cpp/ignite/client/ignite_client_configuration.h:
##########
@@ -187,6 +192,28 @@ class ignite_client_configuration {
m_heartbeat_interval = heartbeat_interval;
}
+ /**
+ * Gets the operation timeout, in milliseconds. 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.
+ *
+ * @return Operation timeout, in milliseconds.
+ */
+ [[nodiscard]] std::chrono::milliseconds get_operation_timeout() const {
return m_operation_timeout; }
+
+ /**
+ * Sets the 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()) + " microseconds");
Review Comment:
Corrected unit from 'microseconds' to 'milliseconds' to match the parameter
type.
```suggestion
+ std::to_string(operation_timeout.count()) + "
milliseconds");
```
##########
modules/platforms/cpp/ignite/client/detail/node_connection.cpp:
##########
@@ -203,7 +208,46 @@ 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);
+
+ std::vector<int64_t> keys_for_erasure;
+
+ for (auto& [id, req] : m_request_handlers) {
+ if (req.timeouts_at > now) {
+
+ std::stringstream ss;
+ ss << "Operation timeout [req_id=" << id << "]";
+ auto res =
req.handler->set_error(ignite_error(error::code::CLIENT_OPERATION_TIMEOUT,ss.str()));
Review Comment:
Missing space after comma in function call. Should be `ss.str())` →
`ss.str())` with proper spacing: `CLIENT_OPERATION_TIMEOUT, ss.str()`.
```suggestion
auto res =
req.handler->set_error(ignite_error(error::code::CLIENT_OPERATION_TIMEOUT,
ss.str()));
```
##########
modules/platforms/cpp/tests/fake_server/tcp_client_channel.h:
##########
@@ -0,0 +1,54 @@
+/*
+* 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.
+ */
+
+#pragma once
+#include <atomic>
+#include <cstddef>
+#include <ignite/common/ignite_error.h>
+#include <ignite/protocol/utils.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+namespace ignite {
+
+/**
+ * Owning wrapper around server-side client socket.
+ */
+class tcp_client_channel {
+ int m_srv_fd;
+ int m_cl_fd = -1;
+ std::byte m_buf[1024];
+ size_t m_pos = 0;
+ size_t m_remaining = 0;
+ std::atomic_bool m_stopped{false};
+
+public:
+ explicit tcp_client_channel(int srv_socket_fd)
+ : m_srv_fd(srv_socket_fd) {}
+
+ void start();
+ void stop();
+ std::vector<std::byte> read_next_n_bytes(size_t n);
+ void send_message(std::vector<std::byte> msg);
+
+private:
+ void receive_next_packet();
+
+
Review Comment:
Extra blank line before the closing brace. Remove one of these blank lines
for consistency with codebase style.
```suggestion
```
##########
modules/platforms/cpp/tests/fake_server/response_action.h:
##########
@@ -0,0 +1,56 @@
+/*
+* 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.
+ */
+
+#pragma once
+#include <chrono>
+
+namespace ignite {
+
+enum action_type { DROP, DELAY };
+
+class response_action {
+public:
+ virtual action_type type() = 0;
+
+ virtual ~response_action() = default;
+};
+
+class drop_action: public response_action {
+public:
+ action_type type() override {
+ return DROP;
+ }
+};
+
+class delayed_action: public response_action {
+public:
+ explicit delayed_action(std::chrono::milliseconds delay)
+ : m_delay(delay) {}
+
+ action_type type() override {
+ return DELAY;
+ }
+
+ std::chrono::milliseconds delay() const {
+ return m_delay;
+ }
+
+private:
+ std::chrono::milliseconds m_delay;;
Review Comment:
Double semicolon at end of line. Remove the extra semicolon.
```suggestion
std::chrono::milliseconds m_delay;
```
##########
modules/platforms/cpp/tests/fake_server/response_action.h:
##########
@@ -0,0 +1,56 @@
+/*
+* 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.
+ */
+
+#pragma once
+#include <chrono>
+
+namespace ignite {
+
+enum action_type { DROP, DELAY };
+
+class response_action {
+public:
+ virtual action_type type() = 0;
+
+ virtual ~response_action() = default;
+};
+
+class drop_action: public response_action {
+public:
+ action_type type() override {
+ return DROP;
+ }
+};
+
+class delayed_action: public response_action {
+public:
+ explicit delayed_action(std::chrono::milliseconds delay)
+ : m_delay(delay) {}
+
+ action_type type() override {
+ return DELAY;
+ }
+
+ std::chrono::milliseconds delay() const {
+ return m_delay;
+ }
+
+private:
+ std::chrono::milliseconds m_delay;;
+};
+
+}
Review Comment:
Missing namespace closing comment. Add `// namespace ignite` for consistency
with other header files in the codebase.
```suggestion
} // namespace ignite
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
