[PATCH v1 0/9] various gossiper code cleanups

[Gleb Natapov]

CI: https://jenkins.scylladb.com/job/scylla-master/job/scylla-ci/12447/

Gleb Natapov (9):
gossiper: co-routinize do_send_ack_msg
gossiper: fold get_or_create_endpoint_state into my_endpoint_state
gossiper: drop unneeded this->
gossiper fix weird logic in get_live_members
gossiper do not needlessly call get_endpoint_state_ptr in
handle_major_state_change
gossiper: co-routinize do_send_ack2_msg
gossiper: remove unused code
gossiper: use 1 seconds instead of 1000 milliseconds
gossiper: start failure_detector_loop on shard 0 only

gms/endpoint_state.hh | 18 ------
gms/gossiper.hh | 8 +--
gms/gossiper.cc | 141 +++++++++++++++++++++---------------------
3 files changed, 71 insertions(+), 96 deletions(-)

--
2.46.2

[Gleb Natapov]

my_endpoint_state() is the only called of
get_or_create_endpoint_state() and calling it is the only thing the
function does anyway.
---
gms/gossiper.hh | 7 +------
gms/gossiper.cc | 3 ++-
2 files changed, 3 insertions(+), 7 deletions(-)

diff --git a/gms/gossiper.hh b/gms/gossiper.hh
index 7d785cc91a2..4a2eba8459f 100644
--- a/gms/gossiper.hh
+++ b/gms/gossiper.hh
@@ -491,12 +491,7 @@ class gossiper : public seastar::async_sharded_service<gossiper>, public seastar

future<> real_mark_alive(inet_address addr);
private:
- // FIXME: for now, allow modifying the endpoint_state's heartbeat_state in place
- // Gets or creates endpoint_state for this node
- endpoint_state& get_or_create_endpoint_state(inet_address ep);
- endpoint_state& my_endpoint_state() {
- return get_or_create_endpoint_state(get_broadcast_address());
- }
+ endpoint_state& my_endpoint_state();

// Use with care, as the endpoint_state_ptr in the endpoint_state_map is considered
// immutable, with one exception - the update_timestamp.
diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 87b000dd0be..6ed8e5938c9 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -1535,7 +1535,8 @@ const endpoint_state& gossiper::get_endpoint_state(inet_address ep) const {
return *it->second;
}

-endpoint_state& gossiper::get_or_create_endpoint_state(inet_address ep) {
+endpoint_state& gossiper::my_endpoint_state() {
+ auto ep = get_broadcast_address();
auto it = _endpoint_state_map.find(ep);
if (it == _endpoint_state_map.end()) {
it = _endpoint_state_map.emplace(ep, make_endpoint_state_ptr({})).first;
--
2.46.2

[Gleb Natapov]

---
gms/gossiper.cc | 48 ++++++++++++++++++++++++------------------------
1 file changed, 24 insertions(+), 24 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 6ed8e5938c9..db3fb4b4a88 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -147,8 +147,8 @@ void gossiper::do_sort(utils::chunked_vector<gossip_digest>& g_digest_list) cons
utils::chunked_vector<gossip_digest> diff_digests;
for (auto g_digest : g_digest_list) {
auto ep = g_digest.get_endpoint();
- auto ep_state = this->get_endpoint_state_ptr(ep);
- version_type version = ep_state ? this->get_max_endpoint_state_version(*ep_state) : version_type();
+ auto ep_state = get_endpoint_state_ptr(ep);
+ version_type version = ep_state ? get_max_endpoint_state_version(*ep_state) : version_type();
int32_t diff_version = ::abs((version - g_digest.get_max_version()).value());
diff_digests.emplace_back(gossip_digest(ep, g_digest.get_generation(), version_type(diff_version)));
}
@@ -170,7 +170,7 @@ void gossiper::do_sort(utils::chunked_vector<gossip_digest>& g_digest_list) cons
future<> gossiper::handle_syn_msg(msg_addr from, gossip_digest_syn syn_msg) {
logger.trace("handle_syn_msg():from={},cluster_name:peer={},local={},group0_id:peer={},local={},partitioner_name:peer={},local={}",
from, syn_msg.cluster_id(), get_cluster_name(), syn_msg.group0_id(), get_group0_id(), syn_msg.partioner(), get_partitioner_name());
- if (!this->is_enabled()) {
+ if (!is_enabled()) {
co_return;
}

@@ -288,7 +288,7 @@ static bool should_count_as_msg_processing(const std::map<inet_address, endpoint
future<> gossiper::handle_ack_msg(msg_addr id, gossip_digest_ack ack_msg) {
logger.trace("handle_ack_msg():from={},msg={}", id, ack_msg);

- if (!this->is_enabled() && !this->is_in_shadow_round()) {
+ if (!is_enabled() && !is_in_shadow_round()) {
co_return;
}

@@ -307,13 +307,13 @@ future<> gossiper::handle_ack_msg(msg_addr id, gossip_digest_ack ack_msg) {

if (ep_state_map.size() > 0) {
update_timestamp_for_nodes(ep_state_map);
- co_await this->apply_state_locally(std::move(ep_state_map));
+ co_await apply_state_locally(std::move(ep_state_map));
}

auto from = id;
auto ack_msg_digest = std::move(g_digest_list);
- if (this->is_in_shadow_round()) {
- this->finish_shadow_round();
+ if (is_in_shadow_round()) {
+ finish_shadow_round();
// don't bother doing anything else, we have what we came for
co_return;
}
@@ -381,7 +381,7 @@ future<> gossiper::do_send_ack2_msg(msg_addr from, utils::chunked_vector<gossip_
const auto version = es->get_heart_beat_state().get_generation() > g_digest.get_generation()
? version_type(0)
: g_digest.get_max_version();
- auto local_ep_state_ptr = this->get_state_for_version_bigger_than(addr, version);
+ auto local_ep_state_ptr = get_state_for_version_bigger_than(addr, version);
if (local_ep_state_ptr) {
delta_ep_state_map.emplace(addr, *local_ep_state_ptr);
}
@@ -475,10 +475,10 @@ future<> gossiper::handle_shutdown_msg(inet_address from, std::optional<int64_t>
co_return;
}

- auto permit = co_await this->lock_endpoint(from, null_permit_id);
+ auto permit = co_await lock_endpoint(from, null_permit_id);
if (generation_number_opt) {
debug_validate_gossip_generation(*generation_number_opt);
- auto es = this->get_endpoint_state_ptr(from);
+ auto es = get_endpoint_state_ptr(from);
if (es) {
auto local_generation = es->get_heart_beat_state().get_generation();
logger.info("Got shutdown message from {}, received_generation={}, local_generation={}",
@@ -494,7 +494,7 @@ future<> gossiper::handle_shutdown_msg(inet_address from, std::optional<int64_t>
co_return;
}
}
- co_await this->mark_as_shutdown(from, permit.id());
+ co_await mark_as_shutdown(from, permit.id());
}

future<gossip_get_endpoint_states_response>
@@ -589,8 +589,8 @@ future<> gossiper::send_gossip(gossip_digest_syn message, std::set<inet_address>
future<> gossiper::do_apply_state_locally(gms::inet_address node, endpoint_state remote_state, bool listener_notification) {
// If state does not exist just add it. If it does then add it if the remote generation is greater.
// If there is a generation tie, attempt to break it by heartbeat version.
- auto permit = co_await this->lock_endpoint(node, null_permit_id);
- auto es = this->get_endpoint_state_ptr(node);
+ auto permit = co_await lock_endpoint(node, null_permit_id);
+ auto es = get_endpoint_state_ptr(node);
if (!es && _topo_sm) {
// Even if there is no endpoint for the given IP the message can still belong to existing endpoint that
// was restarted with different IP, so lets try to locate the endpoint by host id as well. Do it in raft
@@ -615,7 +615,7 @@ future<> gossiper::do_apply_state_locally(gms::inet_address node, endpoint_state
if (listener_notification) {
logger.trace("Updating heartbeat state generation to {} from {} for {}", remote_generation, local_generation, node);
// major state change will handle the update by inserting the remote state directly
- co_await this->handle_major_state_change(node, std::move(remote_state), permit.id());
+ co_await handle_major_state_change(node, std::move(remote_state), permit.id());
} else {
logger.debug("Applying remote_state for node {} (remote generation > local generation)", node);
co_await replicate(node, std::move(remote_state), permit.id());
@@ -623,16 +623,16 @@ future<> gossiper::do_apply_state_locally(gms::inet_address node, endpoint_state
} else if (remote_generation == local_generation) {
if (listener_notification) {
// find maximum state
- auto local_max_version = this->get_max_endpoint_state_version(local_state);
- auto remote_max_version = this->get_max_endpoint_state_version(remote_state);
+ auto local_max_version = get_max_endpoint_state_version(local_state);
+ auto remote_max_version = get_max_endpoint_state_version(remote_state);
if (remote_max_version > local_max_version) {
// apply states, but do not notify since there is no major change
- co_await this->apply_new_states(node, std::move(local_state), remote_state, permit.id());
+ co_await apply_new_states(node, std::move(local_state), remote_state, permit.id());
} else {
logger.debug("Ignoring remote version {} <= {} for {}", remote_max_version, local_max_version, node);
}
- if (!is_alive(node) && !this->is_dead_state(get_endpoint_state(node))) { // unless of course, it was dead
- this->mark_alive(node);
+ if (!is_alive(node) && !is_dead_state(get_endpoint_state(node))) { // unless of course, it was dead
+ mark_alive(node);
}
} else {
bool update = false;
@@ -660,7 +660,7 @@ future<> gossiper::do_apply_state_locally(gms::inet_address node, endpoint_state
}
} else {
if (listener_notification) {
- co_await this->handle_major_state_change(node, std::move(remote_state), permit.id());
+ co_await handle_major_state_change(node, std::move(remote_state), permit.id());
} else {
logger.debug("Applying remote_state for node {} (new node)", node);
co_await replicate(node, std::move(remote_state), permit.id());
@@ -683,7 +683,7 @@ future<> gossiper::apply_state_locally(std::map<inet_address, endpoint_state> ma
logger.debug("apply_state_locally_endpoints={}", endpoints);

co_await coroutine::parallel_for_each(endpoints, [this, &map] (auto&& ep) -> future<> {
- if (ep == this->get_broadcast_address() && !this->is_in_shadow_round()) {
+ if (ep == get_broadcast_address() && !is_in_shadow_round()) {
return make_ready_future<>();
}
if (_topo_sm) {
@@ -1101,7 +1101,7 @@ future<> gossiper::replicate_live_endpoints_on_change(foreign_ptr<std::unique_pt
void gossiper::run() {
// Run it in the background.
(void)seastar::with_semaphore(_callback_running, 1, [this] {
- return seastar::async([this, g = this->shared_from_this()] {
+ return seastar::async([this, g = shared_from_this()] {
logger.trace("=== Gossip round START");

//wait on messaging service to start listening
@@ -1117,7 +1117,7 @@ void gossiper::run() {
}

utils::chunked_vector<gossip_digest> g_digests;
- this->make_random_gossip_digest(g_digests);
+ make_random_gossip_digest(g_digests);

if (g_digests.size() > 0) {
gossip_digest_syn message(get_cluster_name(), get_partitioner_name(), g_digests, get_group0_id());
@@ -2127,7 +2127,7 @@ future<> gossiper::advertise_to_nodes(generation_for_nodes advertise_to_nodes) {
}

future<> gossiper::do_shadow_round(std::unordered_set<gms::inet_address> nodes, mandatory is_mandatory) {
- return seastar::async([this, g = this->shared_from_this(), nodes = std::move(nodes), is_mandatory] () mutable {
+ return seastar::async([this, g = shared_from_this(), nodes = std::move(nodes), is_mandatory] () mutable {
nodes.erase(get_broadcast_address());
gossip_get_endpoint_states_request request{{
gms::application_state::STATUS,
--
2.46.2

[Gleb Natapov]

---
gms/gossiper.cc | 18 ++++++++----------
1 file changed, 8 insertions(+), 10 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 68dc609c696..87b000dd0be 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -239,16 +239,14 @@ future<> gossiper::handle_syn_msg(msg_addr from, gossip_digest_syn syn_msg) {
}

future<> gossiper::do_send_ack_msg(msg_addr from, gossip_digest_syn syn_msg) {
- return futurize_invoke([this, from, syn_msg = std::move(syn_msg)] () mutable {
- auto g_digest_list = syn_msg.get_gossip_digests();
- do_sort(g_digest_list);
- utils::chunked_vector<gossip_digest> delta_gossip_digest_list;
- std::map<inet_address, endpoint_state> delta_ep_state_map;
- this->examine_gossiper(g_digest_list, delta_gossip_digest_list, delta_ep_state_map);
- gms::gossip_digest_ack ack_msg(std::move(delta_gossip_digest_list), std::move(delta_ep_state_map));
- logger.debug("Calling do_send_ack_msg to node {}, syn_msg={}, ack_msg={}", from, syn_msg, ack_msg);
- return ser::gossip_rpc_verbs::send_gossip_digest_ack(&_messaging, from, std::move(ack_msg));
- });
+ auto g_digest_list = syn_msg.get_gossip_digests();
+ do_sort(g_digest_list);
+ utils::chunked_vector<gossip_digest> delta_gossip_digest_list;
+ std::map<inet_address, endpoint_state> delta_ep_state_map;
+ examine_gossiper(g_digest_list, delta_gossip_digest_list, delta_ep_state_map);
+ gms::gossip_digest_ack ack_msg(std::move(delta_gossip_digest_list), std::move(delta_ep_state_map));
+ logger.debug("Calling do_send_ack_msg to node {}, syn_msg={}, ack_msg={}", from, syn_msg, ack_msg);
+ co_return co_await ser::gossip_rpc_verbs::send_gossip_digest_ack(&_messaging, from, std::move(ack_msg));
}

static bool should_count_as_msg_processing(const std::map<inet_address, endpoint_state>& map) {
--
2.46.2

[Gleb Natapov]

The code adds a node to a set and then removes it if a condition is met.
Add to the set if the condition is not met instead.
---
gms/gossiper.cc | 5 ++---
1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index db3fb4b4a88..47914cb119b 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -1217,9 +1217,8 @@ std::set<inet_address> gossiper::get_live_members() const {
std::set<inet_address> live_members(_live_endpoints.begin(), _live_endpoints.end());
auto myip = get_broadcast_address();
logger.debug("live_members before={}", live_members);
- live_members.insert(myip);
- if (is_shutdown(myip)) {
- live_members.erase(myip);
+ if (!is_shutdown(myip)) {
+ live_members.insert(myip);
}
logger.debug("live_members after={}", live_members);
return live_members;
--
2.46.2

[Gleb Natapov]

---
gms/gossiper.cc | 44 +++++++++++++++++++++-----------------------
1 file changed, 21 insertions(+), 23 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 7086a88721d..c01821d7b8d 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -365,31 +365,29 @@ future<> gossiper::handle_ack_msg(msg_addr id, gossip_digest_ack ack_msg) {
}

future<> gossiper::do_send_ack2_msg(msg_addr from, utils::chunked_vector<gossip_digest> ack_msg_digest) {
- return futurize_invoke([this, from, ack_msg_digest = std::move(ack_msg_digest)] () mutable {
/* Get the state required to send to this gossipee - construct GossipDigestAck2Message */

- std::map<inet_address, endpoint_state> delta_ep_state_map;

- for (auto g_digest : ack_msg_digest) {
- inet_address addr = g_digest.get_endpoint();
- const auto es = get_endpoint_state_ptr(addr);
- if (!es || es->get_heart_beat_state().get_generation() < g_digest.get_generation()) {
- continue;
- }
- // Local generation for addr may have been increased since the
- // current node sent an initial SYN. Comparing versions across
- // different generations in get_state_for_version_bigger_than
- // could result in losing some app states with smaller versions.
- const auto version = es->get_heart_beat_state().get_generation() > g_digest.get_generation()
- ? version_type(0)
- : g_digest.get_max_version();
- auto local_ep_state_ptr = get_state_for_version_bigger_than(addr, version);
- if (local_ep_state_ptr) {
- delta_ep_state_map.emplace(addr, *local_ep_state_ptr);
- }

+ std::map<inet_address, endpoint_state> delta_ep_state_map;

+ for (auto g_digest : ack_msg_digest) {
+ inet_address addr = g_digest.get_endpoint();
+ const auto es = get_endpoint_state_ptr(addr);
+ if (!es || es->get_heart_beat_state().get_generation() < g_digest.get_generation()) {
+ continue;
}
- gms::gossip_digest_ack2 ack2_msg(std::move(delta_ep_state_map));
- logger.debug("Calling do_send_ack2_msg to node {}, ack_msg_digest={}, ack2_msg={}", from, ack_msg_digest, ack2_msg);
- return ser::gossip_rpc_verbs::send_gossip_digest_ack2(&_messaging, from, std::move(ack2_msg));
- });
+ // Local generation for addr may have been increased since the
+ // current node sent an initial SYN. Comparing versions across
+ // different generations in get_state_for_version_bigger_than
+ // could result in losing some app states with smaller versions.
+ const auto version = es->get_heart_beat_state().get_generation() > g_digest.get_generation()
+ ? version_type(0)
+ : g_digest.get_max_version();

+ auto local_ep_state_ptr = get_state_for_version_bigger_than(addr, version);

+ if (local_ep_state_ptr) {
+ delta_ep_state_map.emplace(addr, *local_ep_state_ptr);
+ }
+ }
+ gms::gossip_digest_ack2 ack2_msg(std::move(delta_ep_state_map));
+ logger.debug("Calling do_send_ack2_msg to node {}, ack_msg_digest={}, ack2_msg={}", from, ack_msg_digest, ack2_msg);
+ co_return co_await ser::gossip_rpc_verbs::send_gossip_digest_ack2(&_messaging, from, std::move(ack2_msg));
}

// Depends on
--
2.46.2

[Gleb Natapov]

---
gms/gossiper.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 2a5432c9516..93c490eabb0 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -1025,7 +1025,7 @@ future<> gossiper::failure_detector_loop() {
try {
while (_live_endpoints.empty() && is_enabled()) {
logger.debug("failure_detector_loop: Wait until live_nodes={} is not empty", _live_endpoints);
- co_await sleep_abortable(std::chrono::milliseconds(1000), _abort_source);
+ co_await sleep_abortable(std::chrono::seconds(1), _abort_source);
}
if (!is_enabled()) {
co_return;
--
2.46.2

[Gleb Natapov]

---
gms/endpoint_state.hh | 18 ------------------
gms/gossiper.cc | 1 -
2 files changed, 19 deletions(-)

diff --git a/gms/endpoint_state.hh b/gms/endpoint_state.hh
index 53d65ecd958..d9f22f96452 100644
--- a/gms/endpoint_state.hh
+++ b/gms/endpoint_state.hh
@@ -32,7 +32,6 @@ class endpoint_state {
application_state_map _application_state;
/* fields below do not get serialized */
clk::time_point _update_timestamp;
- bool _is_normal = false;

public:
bool operator==(const endpoint_state& other) const {
@@ -45,14 +44,12 @@ class endpoint_state {
: _heart_beat_state()
, _update_timestamp(clk::now())
{
- update_is_normal();
}

endpoint_state(heart_beat_state initial_hb_state) noexcept
: _heart_beat_state(initial_hb_state)
, _update_timestamp(clk::now())
{
- update_is_normal();
}

endpoint_state(heart_beat_state&& initial_hb_state,
@@ -61,7 +58,6 @@ class endpoint_state {
, _application_state(application_state)
, _update_timestamp(clk::now())
{
- update_is_normal();
}

// Valid only on shard 0
@@ -95,12 +91,10 @@ class endpoint_state {

void add_application_state(application_state key, versioned_value value) {
_application_state[key] = std::move(value);
- update_is_normal();
}

void add_application_state(const endpoint_state& es) {
_application_state = es._application_state;
- update_is_normal();
}

/* getters and setters */
@@ -133,18 +127,6 @@ class endpoint_state {
return value.substr(0, pos);
}

- bool is_shutdown() const noexcept {
- return get_status() == versioned_value::SHUTDOWN;
- }
-
- bool is_normal() const noexcept {
- return _is_normal;
- }
-
- void update_is_normal() noexcept {
- _is_normal = get_status() == versioned_value::STATUS_NORMAL;
- }
-
bool is_cql_ready() const noexcept;

// Return the value of the HOST_ID application state
diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index c01821d7b8d..2a5432c9516 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -1351,7 +1351,6 @@ future<> gossiper::replicate(inet_address ep, endpoint_state es, permit_id pid)
// Use foreign_ptr<std::unique_ptr> to ensure destroy on remote shards on exception
std::vector<foreign_ptr<endpoint_state_ptr>> ep_states;
ep_states.resize(smp::count);
- es.update_is_normal();
auto p = make_foreign(make_endpoint_state_ptr(std::move(es)));
const auto *eps = p.get();
ep_states[this_shard_id()] = std::move(p);
--
2.46.2

[Gleb Natapov]

The code calls for get_endpoint_state_ptr several times instead of using
the result of the first call. Change it.
---
gms/gossiper.hh | 1 +
gms/gossiper.cc | 18 ++++++++++--------
2 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/gms/gossiper.hh b/gms/gossiper.hh
index 4a2eba8459f..bd92cd44e42 100644
--- a/gms/gossiper.hh
+++ b/gms/gossiper.hh
@@ -658,6 +658,7 @@ class gossiper : public seastar::async_sharded_service<gossiper>, public seastar
public:
bool is_seed(const inet_address& endpoint) const;
bool is_shutdown(const inet_address& endpoint) const;
+ bool is_shutdown(const endpoint_state& eps) const;
bool is_normal(const inet_address& endpoint) const;
bool is_left(const inet_address& endpoint) const;
// Check if a node is in NORMAL or SHUTDOWN status which means the node is
diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 47914cb119b..7086a88721d 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -1858,17 +1858,15 @@ future<> gossiper::handle_major_state_change(inet_address ep, endpoint_state eps
mark_alive(ep);
} else {
logger.debug("Not marking {} alive due to dead state {}", ep, get_gossip_status(eps));
- co_await mark_dead(ep, std::move(ep_state), pid);
+ co_await mark_dead(ep, ep_state, pid);
}

- auto eps_new = get_endpoint_state_ptr(ep);
- if (eps_new) {
- co_await _subscribers.for_each([ep, eps_new, pid] (shared_ptr<i_endpoint_state_change_subscriber> subscriber) {
- return subscriber->on_join(ep, eps_new, pid);
- });
- }
+ co_await _subscribers.for_each([ep, ep_state, pid] (shared_ptr<i_endpoint_state_change_subscriber> subscriber) {
+ return subscriber->on_join(ep, ep_state, pid);
+ });
+
// check this at the end so nodes will learn about the endpoint
- if (is_shutdown(ep)) {
+ if (is_shutdown(*ep_state)) {
co_await mark_as_shutdown(ep, pid);
}
}
@@ -1881,6 +1879,10 @@ bool gossiper::is_shutdown(const inet_address& endpoint) const {
return get_gossip_status(endpoint) == versioned_value::SHUTDOWN;
}

+bool gossiper::is_shutdown(const endpoint_state& eps) const {
+ return get_gossip_status(eps) == versioned_value::SHUTDOWN;
+}
+
bool gossiper::is_normal(const inet_address& endpoint) const {
return get_gossip_status(endpoint) == versioned_value::STATUS_NORMAL;
}
--
2.46.2

[Gleb Natapov]

failure_detector_loop does nothing on all other shards.
---
gms/gossiper.cc | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 93c490eabb0..0e5b538ec49 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -2096,7 +2096,7 @@ future<> gossiper::start_gossiping(gms::generation_type generation_nbr, applicat
co_await container().invoke_on_all([] (gms::gossiper& g) {
g._enabled = true;
});
- co_await container().invoke_on_all([] (gms::gossiper& g) {
+ co_await container().invoke_on(0, [] (gms::gossiper& g) {
g._failure_detector_loop_done = g.failure_detector_loop();
});
}
@@ -2349,7 +2349,7 @@ future<> gossiper::do_stop_gossiping() {
_scheduled_gossip_task.cancel();
// Take the semaphore makes sure existing gossip loop is finished
auto units = co_await get_units(_callback_running, 1);
- co_await container().invoke_on_all([] (auto& g) {
+ co_await container().invoke_on(0, [] (auto& g) {
return std::move(g._failure_detector_loop_done);
});
logger.info("Gossip is now stopped");
--
2.46.2

[Tomasz Grabiec]

LGTM

--
You received this message because you are subscribed to the Google Groups "ScyllaDB development" group.
To unsubscribe from this group and stop receiving emails from it, send an email to scylladb-dev...@googlegroups.com.
To view this discussion visit https://groups.google.com/d/msgid/scylladb-dev/20241027094509.2353523-1-gleb%40scylladb.com.

[Avi Kivity]

This is not a no-op transformation. live_members is initialized to _live_endpoints. The original code would remove a pre-existing endpoint, the new code does not.

[Benny Halevy]

Right. I believe this was the original intention (to always remove myip from live_members on shutdown.
This could work better:

if (is_shutdown(myip)) {
live_members.erase(myip);
} else {
live_members.insert(myip);

}

>
>
>
> --
> You received this message because you are subscribed to the Google Groups "ScyllaDB development" group.
> To unsubscribe from this group and stop receiving emails from it, send an email to scylladb-dev...@googlegroups.com.

> To view this discussion visit https://groups.google.com/d/msgid/scylladb-dev/262e28ae4c710a64986b0400693a3f2b396ed1e9.camel%40scylladb.com.

[Benny Halevy]

On Sun, 2024-10-27 at 11:43 +0200, 'Gleb Natapov' via ScyllaDB development wrote:

nit: no need for co_return as the function returns future<>
and this is the last statement in the function's body.

[Benny Halevy]

On Sun, 2024-10-27 at 11:43 +0200, 'Gleb Natapov' via ScyllaDB development wrote:

previously, we could have lost the endpoint state while co_await'ing mark_dead,
I presume if the node is removed.
Not sure if this can still happen with consistent topology changes,
but maybe it might still happen with gossip topology changes.

If this series only aims at cleanups, I'd avoid functional changes.

[Benny Halevy]

On Sun, 2024-10-27 at 11:43 +0200, 'Gleb Natapov' via ScyllaDB development wrote:

> ---
>  gms/gossiper.cc | 44 +++++++++++++++++++++-----------------------
>  1 file changed, 21 insertions(+), 23 deletions(-)
>
> diff --git a/gms/gossiper.cc b/gms/gossiper.cc
> index 7086a88721d..c01821d7b8d 100644
> --- a/gms/gossiper.cc
> +++ b/gms/gossiper.cc
> @@ -365,31 +365,29 @@ future<> gossiper::handle_ack_msg(msg_addr id, gossip_digest_ack ack_msg) {
>  }
>  
>  future<> gossiper::do_send_ack2_msg(msg_addr from, utils::chunked_vector<gossip_digest> ack_msg_digest) {
> -    return futurize_invoke([this, from, ack_msg_digest = std::move(ack_msg_digest)] () mutable {
>          /* Get the state required to send to this gossipee - construct GossipDigestAck2Message */

nit: it looks like the comment would be over-indented after this change.

nit: no need for co_return

>  }
>  

[Gleb Natapov]

If you look at the history why the ep is retaken it is because long time
ago a reference was returned here and the referenced object may have
changed during preemption, so the code was changed to re-take the
object. Now since we have a shared prt here this patch returns the code
to its original state.

> > -    if (eps_new) {
> > -        co_await _subscribers.for_each([ep, eps_new, pid] (shared_ptr<i_endpoint_state_change_subscriber> subscriber) {
> > -            return subscriber->on_join(ep, eps_new, pid);
> > -        });
> > -    }
> > +    co_await _subscribers.for_each([ep, ep_state, pid] (shared_ptr<i_endpoint_state_change_subscriber> subscriber) {
> > +        return subscriber->on_join(ep, ep_state, pid);
> > +    });
> > +
> >      // check this at the end so nodes will learn about the endpoint
> > -    if (is_shutdown(ep)) {
> > +    if (is_shutdown(*ep_state)) {
> >          co_await mark_as_shutdown(ep, pid);
> >      }
> >  }
> > @@ -1881,6 +1879,10 @@ bool gossiper::is_shutdown(const inet_address& endpoint) const {
> >      return get_gossip_status(endpoint) == versioned_value::SHUTDOWN;
> >  }
> >  
> > +bool gossiper::is_shutdown(const endpoint_state& eps) const {
> > +    return get_gossip_status(eps) == versioned_value::SHUTDOWN;
> > +}
> > +
> >  bool gossiper::is_normal(const inet_address& endpoint) const {
> >      return get_gossip_status(endpoint) == versioned_value::STATUS_NORMAL;
> >  }
> > --
> > 2.46.2
> >
>

--
Gleb.

[Gleb Natapov]

Yes, it is clearer this way.

--
Gleb.

[Gleb Natapov]

Actually I remember now why I changed it this way. As far as I see _live_endpoints
never contains local broadcast address (we do not run failure detector loop against it)
and if this is the case my transformation is correct.

--
Gleb.

[Benny Halevy]

But this changes the observed behavior.
It is out of scope for a cleanup series.

>
> --
> Gleb.

[Benny Halevy]

okay, fair enough

[Gleb Natapov]

How does it change the observed behaviour if _live_endpoints do not
contain local address?

--
Gleb.

[Gleb Natapov]

CI: https://jenkins.scylladb.com/job/scylla-master/job/scylla-ci/12812/

Also in scylla-dev gleb/gossip-cleanup-v2


v1->v2:
- remove two unneeded co_returns
- fix comment indentation

Gleb Natapov (9):
gossiper: co-routinize do_send_ack_msg
gossiper: fold get_or_create_endpoint_state into my_endpoint_state
gossiper: drop unneeded this->
gossiper fix weird logic in get_live_members
gossiper do not needlessly call get_endpoint_state_ptr in
handle_major_state_change
gossiper: co-routinize do_send_ack2_msg
gossiper: remove unused code
gossiper: use 1 seconds instead of 1000 milliseconds
gossiper: start failure_detector_loop on shard 0 only

gms/endpoint_state.hh | 18 ------
gms/gossiper.hh | 8 +--

gms/gossiper.cc | 143 +++++++++++++++++++++---------------------
3 files changed, 72 insertions(+), 97 deletions(-)

--
2.46.2

[Gleb Natapov]

---
gms/gossiper.cc | 18 ++++++++----------
1 file changed, 8 insertions(+), 10 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index dd7ad93f8cf..522fe3806cf 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc

@@ -239,16 +239,14 @@ future<> gossiper::handle_syn_msg(msg_addr from, gossip_digest_syn syn_msg) {
}

future<> gossiper::do_send_ack_msg(msg_addr from, gossip_digest_syn syn_msg) {

- return futurize_invoke([this, from, syn_msg = std::move(syn_msg)] () mutable {

- auto g_digest_list = syn_msg.get_gossip_digests();
- do_sort(g_digest_list);
- utils::chunked_vector<gossip_digest> delta_gossip_digest_list;

- std::map<inet_address, endpoint_state> delta_ep_state_map;

- this->examine_gossiper(g_digest_list, delta_gossip_digest_list, delta_ep_state_map);

- gms::gossip_digest_ack ack_msg(std::move(delta_gossip_digest_list), std::move(delta_ep_state_map));
- logger.debug("Calling do_send_ack_msg to node {}, syn_msg={}, ack_msg={}", from, syn_msg, ack_msg);
- return ser::gossip_rpc_verbs::send_gossip_digest_ack(&_messaging, from, std::move(ack_msg));
- });
+ auto g_digest_list = syn_msg.get_gossip_digests();
+ do_sort(g_digest_list);
+ utils::chunked_vector<gossip_digest> delta_gossip_digest_list;

+ std::map<inet_address, endpoint_state> delta_ep_state_map;

+ examine_gossiper(g_digest_list, delta_gossip_digest_list, delta_ep_state_map);

+ gms::gossip_digest_ack ack_msg(std::move(delta_gossip_digest_list), std::move(delta_ep_state_map));
+ logger.debug("Calling do_send_ack_msg to node {}, syn_msg={}, ack_msg={}", from, syn_msg, ack_msg);

+ co_await ser::gossip_rpc_verbs::send_gossip_digest_ack(&_messaging, from, std::move(ack_msg));

[Gleb Natapov]

my_endpoint_state() is the only called of
get_or_create_endpoint_state() and calling it is the only thing the
function does anyway.
---
gms/gossiper.hh | 7 +------
gms/gossiper.cc | 3 ++-

2 files changed, 3 insertions(+), 7 deletions(-)

diff --git a/gms/gossiper.hh b/gms/gossiper.hh
index 7d785cc91a2..4a2eba8459f 100644
--- a/gms/gossiper.hh
+++ b/gms/gossiper.hh

@@ -491,12 +491,7 @@ class gossiper : public seastar::async_sharded_service<gossiper>, public seastar

future<> real_mark_alive(inet_address addr);
private:
- // FIXME: for now, allow modifying the endpoint_state's heartbeat_state in place
- // Gets or creates endpoint_state for this node
- endpoint_state& get_or_create_endpoint_state(inet_address ep);
- endpoint_state& my_endpoint_state() {
- return get_or_create_endpoint_state(get_broadcast_address());
- }
+ endpoint_state& my_endpoint_state();

// Use with care, as the endpoint_state_ptr in the endpoint_state_map is considered
// immutable, with one exception - the update_timestamp.

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 522fe3806cf..ea2ed75f9cc 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc

[Gleb Natapov]

---
gms/gossiper.cc | 48 ++++++++++++++++++++++++------------------------
1 file changed, 24 insertions(+), 24 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index ea2ed75f9cc..9034fe0de31 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc

@@ -147,8 +147,8 @@ void gossiper::do_sort(utils::chunked_vector<gossip_digest>& g_digest_list) cons
utils::chunked_vector<gossip_digest> diff_digests;
for (auto g_digest : g_digest_list) {
auto ep = g_digest.get_endpoint();
- auto ep_state = this->get_endpoint_state_ptr(ep);
- version_type version = ep_state ? this->get_max_endpoint_state_version(*ep_state) : version_type();
+ auto ep_state = get_endpoint_state_ptr(ep);
+ version_type version = ep_state ? get_max_endpoint_state_version(*ep_state) : version_type();
int32_t diff_version = ::abs((version - g_digest.get_max_version()).value());
diff_digests.emplace_back(gossip_digest(ep, g_digest.get_generation(), version_type(diff_version)));
}
@@ -170,7 +170,7 @@ void gossiper::do_sort(utils::chunked_vector<gossip_digest>& g_digest_list) cons

future<> gossiper::handle_syn_msg(msg_addr from, gossip_digest_syn syn_msg) {

logger.trace("handle_syn_msg():from={},cluster_name:peer={},local={},group0_id:peer={},local={},partitioner_name:peer={},local={}",
from, syn_msg.cluster_id(), get_cluster_name(), syn_msg.group0_id(), get_group0_id(), syn_msg.partioner(), get_partitioner_name());
- if (!this->is_enabled()) {
+ if (!is_enabled()) {
co_return;
}

@@ -288,7 +288,7 @@ static bool should_count_as_msg_processing(const std::map<inet_address, endpoint

future<> gossiper::handle_ack_msg(msg_addr id, gossip_digest_ack ack_msg) {

logger.trace("handle_ack_msg():from={},msg={}", id, ack_msg);

- if (!this->is_enabled() && !this->is_in_shadow_round()) {
+ if (!is_enabled() && !is_in_shadow_round()) {
co_return;
}

@@ -307,13 +307,13 @@ future<> gossiper::handle_ack_msg(msg_addr id, gossip_digest_ack ack_msg) {

if (ep_state_map.size() > 0) {
update_timestamp_for_nodes(ep_state_map);
- co_await this->apply_state_locally(std::move(ep_state_map));
+ co_await apply_state_locally(std::move(ep_state_map));
}

auto from = id;
auto ack_msg_digest = std::move(g_digest_list);
- if (this->is_in_shadow_round()) {
- this->finish_shadow_round();
+ if (is_in_shadow_round()) {
+ finish_shadow_round();
// don't bother doing anything else, we have what we came for
co_return;
}

@@ -381,7 +381,7 @@ future<> gossiper::do_send_ack2_msg(msg_addr from, utils::chunked_vector<gossip_

const auto version = es->get_heart_beat_state().get_generation() > g_digest.get_generation()

? version_type(0)
: g_digest.get_max_version();
- auto local_ep_state_ptr = this->get_state_for_version_bigger_than(addr, version);

+ auto local_ep_state_ptr = get_state_for_version_bigger_than(addr, version);

[Gleb Natapov]

The code adds a node to a set and then removes it if a condition is met.
Add to the set if the condition is not met instead.
---
gms/gossiper.cc | 5 ++---

1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 9034fe0de31..d2c41e09240 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc

@@ -1217,9 +1217,8 @@ std::set<inet_address> gossiper::get_live_members() const {
std::set<inet_address> live_members(_live_endpoints.begin(), _live_endpoints.end());
auto myip = get_broadcast_address();
logger.debug("live_members before={}", live_members);
- live_members.insert(myip);
- if (is_shutdown(myip)) {
- live_members.erase(myip);
+ if (!is_shutdown(myip)) {
+ live_members.insert(myip);
}
logger.debug("live_members after={}", live_members);
return live_members;

--
2.46.2

[Gleb Natapov]

---
gms/gossiper.cc | 46 ++++++++++++++++++++++------------------------
1 file changed, 22 insertions(+), 24 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 0944a3b61b8..1a9cc6caefb 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc
@@ -365,31 +365,29 @@ future<> gossiper::handle_ack_msg(msg_addr id, gossip_digest_ack ack_msg) {
}

future<> gossiper::do_send_ack2_msg(msg_addr from, utils::chunked_vector<gossip_digest> ack_msg_digest) {
- return futurize_invoke([this, from, ack_msg_digest = std::move(ack_msg_digest)] () mutable {
- /* Get the state required to send to this gossipee - construct GossipDigestAck2Message */

- std::map<inet_address, endpoint_state> delta_ep_state_map;

- for (auto g_digest : ack_msg_digest) {

- inet_address addr = g_digest.get_endpoint();
- const auto es = get_endpoint_state_ptr(addr);
- if (!es || es->get_heart_beat_state().get_generation() < g_digest.get_generation()) {
- continue;
- }
- // Local generation for addr may have been increased since the
- // current node sent an initial SYN. Comparing versions across
- // different generations in get_state_for_version_bigger_than
- // could result in losing some app states with smaller versions.

- const auto version = es->get_heart_beat_state().get_generation() > g_digest.get_generation()
- ? version_type(0)
- : g_digest.get_max_version();
- auto local_ep_state_ptr = get_state_for_version_bigger_than(addr, version);

- if (local_ep_state_ptr) {
- delta_ep_state_map.emplace(addr, *local_ep_state_ptr);
- }

+ /* Get the state required to send to this gossipee - construct GossipDigestAck2Message */

+ std::map<inet_address, endpoint_state> delta_ep_state_map;

+ for (auto g_digest : ack_msg_digest) {

+ inet_address addr = g_digest.get_endpoint();
+ const auto es = get_endpoint_state_ptr(addr);
+ if (!es || es->get_heart_beat_state().get_generation() < g_digest.get_generation()) {
+ continue;
}
- gms::gossip_digest_ack2 ack2_msg(std::move(delta_ep_state_map));
- logger.debug("Calling do_send_ack2_msg to node {}, ack_msg_digest={}, ack2_msg={}", from, ack_msg_digest, ack2_msg);
- return ser::gossip_rpc_verbs::send_gossip_digest_ack2(&_messaging, from, std::move(ack2_msg));
- });
+ // Local generation for addr may have been increased since the
+ // current node sent an initial SYN. Comparing versions across
+ // different generations in get_state_for_version_bigger_than
+ // could result in losing some app states with smaller versions.

+ const auto version = es->get_heart_beat_state().get_generation() > g_digest.get_generation()

+ ? version_type(0)
+ : g_digest.get_max_version();

+ auto local_ep_state_ptr = get_state_for_version_bigger_than(addr, version);

+ if (local_ep_state_ptr) {
+ delta_ep_state_map.emplace(addr, *local_ep_state_ptr);
+ }
+ }
+ gms::gossip_digest_ack2 ack2_msg(std::move(delta_ep_state_map));
+ logger.debug("Calling do_send_ack2_msg to node {}, ack_msg_digest={}, ack2_msg={}", from, ack_msg_digest, ack2_msg);

+ co_await ser::gossip_rpc_verbs::send_gossip_digest_ack2(&_messaging, from, std::move(ack2_msg));

[Gleb Natapov]

failure_detector_loop does nothing on all other shards.
---
gms/gossiper.cc | 4 ++--

1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 5a13bb7bc3f..ec8a86ad06c 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc

[Gleb Natapov]

The code calls for get_endpoint_state_ptr several times instead of using
the result of the first call. Change it.
---
gms/gossiper.hh | 1 +
gms/gossiper.cc | 18 ++++++++++--------

2 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/gms/gossiper.hh b/gms/gossiper.hh
index 4a2eba8459f..bd92cd44e42 100644
--- a/gms/gossiper.hh
+++ b/gms/gossiper.hh

@@ -658,6 +658,7 @@ class gossiper : public seastar::async_sharded_service<gossiper>, public seastar
public:
bool is_seed(const inet_address& endpoint) const;
bool is_shutdown(const inet_address& endpoint) const;
+ bool is_shutdown(const endpoint_state& eps) const;
bool is_normal(const inet_address& endpoint) const;
bool is_left(const inet_address& endpoint) const;
// Check if a node is in NORMAL or SHUTDOWN status which means the node is

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index d2c41e09240..0944a3b61b8 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc

@@ -1858,17 +1858,15 @@ future<> gossiper::handle_major_state_change(inet_address ep, endpoint_state eps
mark_alive(ep);
} else {
logger.debug("Not marking {} alive due to dead state {}", ep, get_gossip_status(eps));
- co_await mark_dead(ep, std::move(ep_state), pid);
+ co_await mark_dead(ep, ep_state, pid);
}

- auto eps_new = get_endpoint_state_ptr(ep);

[Gleb Natapov]

---
gms/gossiper.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index a51d7f20881..5a13bb7bc3f 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc

[Gleb Natapov]

diff --git a/gms/gossiper.cc b/gms/gossiper.cc
index 1a9cc6caefb..a51d7f20881 100644
--- a/gms/gossiper.cc
+++ b/gms/gossiper.cc

[Gleb Natapov]

Benni, Avi ping

--
Gleb.

[Gleb Natapov]

On Thu, Oct 31, 2024 at 12:52:33PM +0200, Gleb Natapov wrote:
> Benni, Avi ping
>

Ping^2

--
Gleb.

[Benny Halevy]

1. There's the /gossiper/endpoint/live api
2. storage_service::describe_schema_versions() uses _gossiper.get_live_members()


>
> --
> Gleb.

[Gleb Natapov]

Yes. So how my patch changes the behaviour of those two?

--
Gleb.

[Benny Halevy]

On Wed, 2024-10-30 at 12:58 +0200, 'Gleb Natapov' via ScyllaDB development wrote:
> The code adds a node to a set and then removes it if a condition is met.
> Add to the set if the condition is not met instead.
> ---
>  gms/gossiper.cc | 5 ++---
>  1 file changed, 2 insertions(+), 3 deletions(-)
>
> diff --git a/gms/gossiper.cc b/gms/gossiper.cc
> index 9034fe0de31..d2c41e09240 100644
> --- a/gms/gossiper.cc
> +++ b/gms/gossiper.cc
> @@ -1217,9 +1217,8 @@ std::set<inet_address> gossiper::get_live_members() const {
>      std::set<inet_address> live_members(_live_endpoints.begin(), _live_endpoints.end());
>      auto myip = get_broadcast_address();
>      logger.debug("live_members before={}", live_members);
> -    live_members.insert(myip);
> -    if (is_shutdown(myip)) {
> -        live_members.erase(myip);
> +    if (!is_shutdown(myip)) {
> +        live_members.insert(myip);

Since _live_endpoints is not supposed to have myip this should be equivalent to the original code