From 3c6e0124742c00112e9c48fa6f96d34d4dcd901f Mon Sep 17 00:00:00 2001 From: Sungsik Date: Mon, 2 Jun 2025 22:43:41 +0200 Subject: [PATCH 1/6] Handle multiple events returned by recursive watch --- etcd/Response.hpp | 13 +++++++++++++ etcd/v3/V3Response.hpp | 4 +++- src/Response.cpp | 10 ++++++++++ src/v3/AsyncGRPC.cpp | 26 ++++++++++++++++---------- src/v3/V3Response.cpp | 4 ++++ 5 files changed, 46 insertions(+), 11 deletions(-) diff --git a/etcd/Response.hpp b/etcd/Response.hpp index 25426a3..d1dfd19 100644 --- a/etcd/Response.hpp +++ b/etcd/Response.hpp @@ -134,6 +134,18 @@ class Response { */ Value const& prev_value() const; + /** + * Returns the index-th action of the response to an 'recursive watch' + * operation. Equivalent to actions()[index] + */ + std::string const& action(int index) const; + + /** + * Returns the vector of actions in a directory in response to an 'recursive + * watch' operation. + */ + std::vector const& actions() const; + /** * Returns the index-th value of the response to an 'ls' operation. Equivalent * to values()[index] @@ -220,6 +232,7 @@ class Response { std::string _action; Value _value; Value _prev_value; + std::vector _actions; // for recursive watch Values _values; Keys _keys; int64_t _compact_revision = -1; // for watch diff --git a/etcd/v3/V3Response.hpp b/etcd/v3/V3Response.hpp index c5042b7..573d196 100644 --- a/etcd/v3/V3Response.hpp +++ b/etcd/v3/V3Response.hpp @@ -10,7 +10,7 @@ namespace etcdv3 { class V3Response { public: - V3Response() : error_code(0), index(0){}; + V3Response() : error_code(0), index(0) {}; void set_error_code(int code); int get_error_code() const; std::string const& get_error_message() const; @@ -18,6 +18,7 @@ class V3Response { void set_action(std::string action); int64_t get_index() const; std::string const& get_action() const; + std::vector const& get_actions() const; std::vector const& get_values() const; std::vector const& get_prev_values() const; etcdv3::KeyValue const& get_value() const; @@ -44,6 +45,7 @@ class V3Response { std::string action; etcdv3::KeyValue value; etcdv3::KeyValue prev_value; + std::vector actions; std::vector values; std::vector prev_values; int64_t compact_revision = -1; diff --git a/src/Response.cpp b/src/Response.cpp index 6a6a3d6..d523548 100644 --- a/src/Response.cpp +++ b/src/Response.cpp @@ -35,7 +35,9 @@ etcd::Response::Response(const etcdv3::V3Response& reply, _error_message = reply.get_error_message(); if (reply.has_values()) { auto val = reply.get_values(); + auto action = reply.get_actions(); for (unsigned int index = 0; index < val.size(); index++) { + _actions.push_back(action[index]); _values.push_back(Value(val[index])); _keys.push_back(val[index].kvs.key()); } @@ -96,6 +98,14 @@ etcd::Value const& etcd::Response::value() const { return _value; } etcd::Value const& etcd::Response::prev_value() const { return _prev_value; } +std::vector const& etcd::Response::actions() const { + return _actions; +} + +std::string const& etcd::Response::action(int index) const { + return _actions[index]; +} + etcd::Values const& etcd::Response::values() const { return _values; } etcd::Value const& etcd::Response::value(int index) const { diff --git a/src/v3/AsyncGRPC.cpp b/src/v3/AsyncGRPC.cpp index bc1dcdd..b34bb45 100644 --- a/src/v3/AsyncGRPC.cpp +++ b/src/v3/AsyncGRPC.cpp @@ -283,24 +283,30 @@ void etcdv3::AsyncWatchResponse::ParseResponse(WatchResponse& reply) { auto event = reply.events(cnt); if (mvccpb::Event::EventType::Event_EventType_PUT == event.type()) { if (event.kv().version() == 1) { - action = etcdv3::CREATE_ACTION; + actions.emplace_back(etcdv3::CREATE_ACTION); } else { - action = etcdv3::SET_ACTION; + actions.emplace_back(etcdv3::SET_ACTION); } - value.kvs = event.kv(); + etcdv3::KeyValue kv; + kv.kvs.CopyFrom(event.kv()); + values.push_back(kv); } else if (mvccpb::Event::EventType::Event_EventType_DELETE_ == event.type()) { - action = etcdv3::DELETE_ACTION; - value.kvs = event.kv(); + actions.emplace_back(etcdv3::DELETE_ACTION); + etcdv3::KeyValue kv; + kv.kvs.CopyFrom(event.kv()); + values.push_back(kv); } if (event.has_prev_kv()) { - prev_value.kvs = event.prev_kv(); + etcdv3::KeyValue kv; + kv.kvs.CopyFrom(event.prev_kv()); + prev_values.emplace_back(kv); } - // just store the first occurence of the key in values. - // this is done so tas client will not need to change their behaviour. - // break immediately - break; } + + action = actions[0]; + value = values[0]; + prev_value = prev_values[0]; } etcdv3::AsyncCampaignAction::AsyncCampaignAction( diff --git a/src/v3/V3Response.cpp b/src/v3/V3Response.cpp index 21a235f..36a1c44 100644 --- a/src/v3/V3Response.cpp +++ b/src/v3/V3Response.cpp @@ -21,6 +21,10 @@ void etcdv3::V3Response::set_action(std::string action) { this->action = action; } +std::vector const& etcdv3::V3Response::get_actions() const { + return actions; +} + std::vector const& etcdv3::V3Response::get_values() const { return values; } From c269e76a628d61f5c44eabd16e8eeef5b6244533 Mon Sep 17 00:00:00 2001 From: Sungsik Date: Mon, 2 Jun 2025 23:16:56 +0200 Subject: [PATCH 2/6] Fix bug on deletion: segfault --- src/Response.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/Response.cpp b/src/Response.cpp index d523548..f9f2b0c 100644 --- a/src/Response.cpp +++ b/src/Response.cpp @@ -37,7 +37,9 @@ etcd::Response::Response(const etcdv3::V3Response& reply, auto val = reply.get_values(); auto action = reply.get_actions(); for (unsigned int index = 0; index < val.size(); index++) { - _actions.push_back(action[index]); + if (index < action.size()) { + _actions.push_back(action[index]); + } _values.push_back(Value(val[index])); _keys.push_back(val[index].kvs.key()); } From dd380f4e1fb6b64ba99b364faaede1e17edea36f Mon Sep 17 00:00:00 2001 From: Sungsik Date: Mon, 2 Jun 2025 23:46:23 +0200 Subject: [PATCH 3/6] Add main.cpp for testing --- src/CMakeLists.txt | 3 +++ src/main.cpp | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+) create mode 100644 src/main.cpp diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 19687db..b9d5a95 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -84,3 +84,6 @@ else() install(TARGETS etcd-cpp-api-core etcd-cpp-api EXPORT etcd-targets) endif() + +add_executable(main main.cpp) +target_link_libraries(main PRIVATE etcd-cpp-api) diff --git a/src/main.cpp b/src/main.cpp new file mode 100644 index 0000000..3b9d912 --- /dev/null +++ b/src/main.cpp @@ -0,0 +1,32 @@ +#include +#include + +int main() { + auto etcd_client = etcd::SyncClient("http://127.0.0.1:2379"); + etcd_client.rmdir("/test", true); + etcd_client.put("/test/0", "val_0"); + auto ls_resp = etcd_client.ls("/test"); + etcd_client.put("/test/1", "val_1"); + etcd_client.put("/test/2", "val_2"); + etcd_client.rm("/test/2"); + + auto watcher = etcd::Watcher( + etcd_client, "/test", ls_resp.index() + 1, + [](const etcd::Response& resp) { + auto actions = resp.actions(); + auto values = resp.values(); + std::cout << "size of values: " << values.size() << std::endl; + std::cout << "size of keys: " << resp.keys().size() << std::endl; + for (auto i = 0u; i < values.size(); ++i) { + auto action = actions[i]; + auto val = values[i]; + std::cout << "action: " << action << ", key: " << val.key() + << ", val: " << val.as_string() << std::endl; + // callback( + // response(resp.action(), val.key(), val.as_string())); + } + }, + true /*recursive*/); + std::this_thread::sleep_for(std::chrono::seconds(1)); + return 0; +} From 09f2f29f19dea44fbe2900fa2779fabb0503cf1a Mon Sep 17 00:00:00 2001 From: Sungsik Date: Mon, 2 Jun 2025 23:46:49 +0200 Subject: [PATCH 4/6] Add ${PROJECT_SOURCE_DIR} as include directory: you can include headers like #include --- src/CMakeLists.txt | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index b9d5a95..f3b5716 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -85,5 +85,8 @@ else() EXPORT etcd-targets) endif() +target_include_directories(etcd-cpp-api INTERFACE + $) + add_executable(main main.cpp) target_link_libraries(main PRIVATE etcd-cpp-api) From c75a8b88b6909114433034bc855ca2fbb0c3b371 Mon Sep 17 00:00:00 2001 From: Sungsik Date: Tue, 3 Jun 2025 00:15:58 +0200 Subject: [PATCH 5/6] Fix segfault on AsyncWatchResponse::ParseResponse --- src/v3/AsyncGRPC.cpp | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/v3/AsyncGRPC.cpp b/src/v3/AsyncGRPC.cpp index b34bb45..49ec773 100644 --- a/src/v3/AsyncGRPC.cpp +++ b/src/v3/AsyncGRPC.cpp @@ -304,9 +304,12 @@ void etcdv3::AsyncWatchResponse::ParseResponse(WatchResponse& reply) { } } - action = actions[0]; - value = values[0]; - prev_value = prev_values[0]; + if (not actions.empty()) + action = actions[0]; + if (not values.empty()) + value = values[0]; + if (not prev_values.empty()) + prev_value = prev_values[0]; } etcdv3::AsyncCampaignAction::AsyncCampaignAction( From 0820acb841154e301b6f8dabb274ee6a16937ba9 Mon Sep 17 00:00:00 2001 From: Sungsik Date: Tue, 10 Jun 2025 17:36:54 +0200 Subject: [PATCH 6/6] Apply memory-leak fix: https://github.com/etcd-cpp-apiv3/etcd-cpp-apiv3/issues/289 --- src/v3/Action.cpp | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/v3/Action.cpp b/src/v3/Action.cpp index fff45e4..8a03f3a 100644 --- a/src/v3/Action.cpp +++ b/src/v3/Action.cpp @@ -16,6 +16,12 @@ etcdv3::Action::Action(etcdv3::ActionParameters&& params) { etcdv3::Action::~Action() { cq_.Shutdown(); + void* tag; + bool ok; + while (cq_.Next(&tag, &ok)) { + // Optionally, handle or log the tag/ok if needed + } + // cancel on-the-fly calls context.TryCancel(); }