[PATCH v4 07/12] [RAW] swim: keep encoded round message cached

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

During a SWIM round a message is being handed out consisting of
at most 4 sections. Parts of the message change rarely, by
member attributes update and by removal of some of them. So it is
possible to cache the message and send it during several round
steps in a row. Or even do not rebuild it the whole round.

Part of #3234
---
 src/lib/swim/swim.c    | 21 +++++++++++++++++----
 src/lib/swim/swim_io.h |  7 +++++++
 2 files changed, 24 insertions(+), 4 deletions(-)

diff --git a/src/lib/swim/swim.c b/src/lib/swim/swim.c
index 353e55254..40faa296e 100644
--- a/src/lib/swim/swim.c
+++ b/src/lib/swim/swim.c
@@ -357,6 +357,13 @@ struct swim {
 	struct rlist queue_events;
 };
 
+/** Reset cached round message on any change of any member. */
+static inline void
+cached_round_msg_invalidate(struct swim *swim)
+{
+	swim_packet_create(&swim->round_step_task.packet);
+}
+
 /** Put the member into a list of ACK waiters. */
 static void
 swim_member_wait_ack(struct swim *swim, struct swim_member *member)
@@ -386,6 +393,7 @@ swim_schedule_event(struct swim *swim, struct swim_member *member)
 				     in_queue_events);
 	}
 	member->status_ttl = mh_size(swim->members);
+	cached_round_msg_invalidate(swim);
 }
 
 /**
@@ -458,6 +466,7 @@ static void
 swim_member_delete(struct swim *swim, struct swim_member *member)
 {
 	say_verbose("SWIM: member %s is deleted", swim_uuid_str(&member->uuid));
+	cached_round_msg_invalidate(swim);
 	struct mh_swim_table_key key = {member->hash, &member->uuid};
 	mh_int_t rc = mh_swim_table_find(swim->members, key, NULL);
 	assert(rc != mh_end(swim->members));
@@ -733,8 +742,11 @@ swim_encode_dissemination(struct swim *swim, struct swim_packet *packet)
 
 /** Encode SWIM components into a UDP packet. */
 static void
-swim_encode_round_msg(struct swim *swim, struct swim_packet *packet)
+swim_encode_round_msg(struct swim *swim)
 {
+	if (swim_packet_body_size(&swim->round_step_task.packet) > 0)
+		return;
+	struct swim_packet *packet = &swim->round_step_task.packet;
 	swim_packet_create(packet);
 	char *header = swim_packet_alloc(packet, 1);
 	int map_size = 0;
@@ -766,8 +778,10 @@ swim_decrease_events_ttl(struct swim *swim)
 				 tmp) {
 		if (member->old_uuid_ttl > 0)
 			--member->old_uuid_ttl;
-		if (--member->status_ttl == 0)
+		if (--member->status_ttl == 0) {
 			rlist_del_entry(member, in_queue_events);
+			cached_round_msg_invalidate(swim);
+		}
 	}
 }
 
@@ -791,8 +805,7 @@ swim_round_step_begin(struct ev_loop *loop, struct ev_periodic *p, int events)
 	 */
 	if (rlist_empty(&swim->queue_round))
 		return;
-
-	swim_encode_round_msg(swim, &swim->round_step_task.packet);
+	swim_encode_round_msg(swim);
 	struct swim_member *m =
 		rlist_first_entry(&swim->queue_round, struct swim_member,
 				  in_queue_round);
diff --git a/src/lib/swim/swim_io.h b/src/lib/swim/swim_io.h
index 508d1ef6e..4d694857d 100644
--- a/src/lib/swim/swim_io.h
+++ b/src/lib/swim/swim_io.h
@@ -125,6 +125,13 @@ swim_packet_alloc(struct swim_packet *packet, int size)
 	return res;
 }
 
+/** Size of the packet body. Meta is not counted. */
+static inline int
+swim_packet_body_size(const struct swim_packet *packet)
+{
+	return packet->pos - packet->body;
+}
+
 /** Initialize @a packet, reserve some space for meta. */
 void
 swim_packet_create(struct swim_packet *packet);
-- 
2.17.2 (Apple Git-113)