[tarantool-patches] [PATCH 1/1] swim: introduce SWIM's anti-entropy component
Vladislav Shpilevoy
v.shpilevoy at tarantool.org
Wed Mar 6 18:13:23 MSK 2019
SWIM - Scalable Weakly-consistent Infection-style Process Group
Membership Protocol. It consists of 2 components: events
dissemination and failure detection, and stores in memory a
table of known remote hosts - members. Also some SWIM
implementations have an additional component: anti-entropy -
periodical broadcast of a random subset of members table.
Dissemination component spreads over the cluster changes occurred
with members. Failure detection constantly searches for failed
dead members. Anti-entropy just sends all known information at
once about a member so as to synchronize it among all other
members in case some events were not disseminated (UDP problems).
Anti-entropy is the most vital component, since it can work
without dissemination and failure detection. But they can not
work properly with out the former. Consider the example: two SWIM
nodes, both are alive. Nothing happens, so the events list is
empty, only pings are being sent periodically. Then a third
node appears. It knows about one of existing nodes. How should
it learn about another one? Sure, its known counterpart can try
to notify another one, but it is UDP, so this event can get lost.
Anti-entropy is an extra simple component, it just piggybacks
random part of members table with each regular round message.
In the example above the new node will learn about the third
one via anti-entropy messages of the second one soon or late.
This is why anti-entropy is the first implemented component.
Part of #3234
---
Branch: https://github.com/tarantool/tarantool/tree/gerold103/gh-3234-swim-anti-entropy
cmake/profile.cmake | 2 +-
src/CMakeLists.txt | 3 +-
src/lib/CMakeLists.txt | 1 +
src/lib/core/diag.h | 2 +
src/lib/core/exception.cc | 23 +
src/lib/core/exception.h | 7 +
src/lib/swim/CMakeLists.txt | 12 +
src/lib/swim/swim.c | 985 ++++++++++++++++++++++++++++++
src/lib/swim/swim.h | 141 +++++
src/lib/swim/swim_ev.c | 51 ++
src/lib/swim/swim_ev.h | 64 ++
src/lib/swim/swim_io.c | 217 +++++++
src/lib/swim/swim_io.h | 225 +++++++
src/lib/swim/swim_proto.c | 331 ++++++++++
src/lib/swim/swim_proto.h | 320 ++++++++++
src/lib/swim/swim_transport.h | 79 +++
src/lib/swim/swim_transport_udp.c | 112 ++++
test/unit/CMakeLists.txt | 8 +
test/unit/swim.c | 243 ++++++++
test/unit/swim.result | 56 ++
test/unit/swim_proto.c | 212 +++++++
test/unit/swim_test_ev.c | 252 ++++++++
test/unit/swim_test_ev.h | 58 ++
test/unit/swim_test_transport.c | 257 ++++++++
test/unit/swim_test_transport.h | 58 ++
test/unit/swim_test_utils.c | 139 +++++
test/unit/swim_test_utils.h | 92 +++
27 files changed, 3948 insertions(+), 2 deletions(-)
create mode 100644 src/lib/swim/CMakeLists.txt
create mode 100644 src/lib/swim/swim.c
create mode 100644 src/lib/swim/swim.h
create mode 100644 src/lib/swim/swim_ev.c
create mode 100644 src/lib/swim/swim_ev.h
create mode 100644 src/lib/swim/swim_io.c
create mode 100644 src/lib/swim/swim_io.h
create mode 100644 src/lib/swim/swim_proto.c
create mode 100644 src/lib/swim/swim_proto.h
create mode 100644 src/lib/swim/swim_transport.h
create mode 100644 src/lib/swim/swim_transport_udp.c
create mode 100644 test/unit/swim.c
create mode 100644 test/unit/swim.result
create mode 100644 test/unit/swim_proto.c
create mode 100644 test/unit/swim_test_ev.c
create mode 100644 test/unit/swim_test_ev.h
create mode 100644 test/unit/swim_test_transport.c
create mode 100644 test/unit/swim_test_transport.h
create mode 100644 test/unit/swim_test_utils.c
create mode 100644 test/unit/swim_test_utils.h
diff --git a/cmake/profile.cmake b/cmake/profile.cmake
index 0ba31fa2c..70fa74694 100644
--- a/cmake/profile.cmake
+++ b/cmake/profile.cmake
@@ -1,5 +1,5 @@
set(CMAKE_REQUIRED_FLAGS "-fprofile-arcs -ftest-coverage")
-check_library_exists("" __gcov_flush "" HAVE_GCOV)
+set(HAVE_GCOV 1)
set(CMAKE_REQUIRED_FLAGS "")
set(ENABLE_GCOV_DEFAULT OFF)
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 481618f20..7c2395517 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -162,7 +162,8 @@ endif()
set_source_files_compile_flags(${server_sources})
add_library(server STATIC ${server_sources})
-target_link_libraries(server core coll http_parser bit uri uuid)
+target_link_libraries(server core coll http_parser bit uri uuid swim swim_udp
+ swim_ev)
# Rule of thumb: if exporting a symbol from a static library, list the
# library here.
diff --git a/src/lib/CMakeLists.txt b/src/lib/CMakeLists.txt
index 39795bbae..b7179d04f 100644
--- a/src/lib/CMakeLists.txt
+++ b/src/lib/CMakeLists.txt
@@ -10,6 +10,7 @@ add_subdirectory(http_parser)
add_subdirectory(core)
add_subdirectory(uuid)
add_subdirectory(coll)
+add_subdirectory(swim)
if(ENABLE_BUNDLED_MSGPUCK)
add_subdirectory(msgpuck EXCLUDE_FROM_ALL)
endif()
diff --git a/src/lib/core/diag.h b/src/lib/core/diag.h
index 885fd1195..78f0cdbdf 100644
--- a/src/lib/core/diag.h
+++ b/src/lib/core/diag.h
@@ -249,6 +249,8 @@ struct error *
BuildSystemError(const char *file, unsigned line, const char *format, ...);
struct error *
BuildCollationError(const char *file, unsigned line, const char *format, ...);
+struct error *
+BuildSwimError(const char *file, unsigned line, const char *format, ...);
struct index_def;
diff --git a/src/lib/core/exception.cc b/src/lib/core/exception.cc
index 8fdb24ad9..6124c70d0 100644
--- a/src/lib/core/exception.cc
+++ b/src/lib/core/exception.cc
@@ -269,6 +269,17 @@ CollationError::CollationError(const char *file, unsigned line,
va_end(ap);
}
+const struct type_info type_SwimError = make_type("SwimError", &type_Exception);
+
+SwimError::SwimError(const char *file, unsigned line, const char *format, ...)
+ : Exception(&type_SwimError, file, line)
+{
+ va_list ap;
+ va_start(ap, format);
+ error_vformat_msg(this, format, ap);
+ va_end(ap);
+}
+
#define BuildAlloc(type) \
void *p = malloc(sizeof(type)); \
if (p == NULL) \
@@ -348,6 +359,18 @@ BuildCollationError(const char *file, unsigned line, const char *format, ...)
return e;
}
+struct error *
+BuildSwimError(const char *file, unsigned line, const char *format, ...)
+{
+ BuildAlloc(SwimError);
+ SwimError *e = new (p) SwimError(file, line, "");
+ va_list ap;
+ va_start(ap, format);
+ error_vformat_msg(e, format, ap);
+ va_end(ap);
+ return e;
+}
+
struct error *
BuildSocketError(const char *file, unsigned line, const char *socketname,
const char *format, ...)
diff --git a/src/lib/core/exception.h b/src/lib/core/exception.h
index ede26643a..4f5b66f2e 100644
--- a/src/lib/core/exception.h
+++ b/src/lib/core/exception.h
@@ -50,6 +50,7 @@ extern const struct type_info type_LuajitError;
extern const struct type_info type_IllegalParams;
extern const struct type_info type_SystemError;
extern const struct type_info type_CollationError;
+extern const struct type_info type_SwimError;
const char *
exception_get_string(struct error *e, const struct method_info *method);
@@ -159,6 +160,12 @@ public:
virtual void raise() { throw this; }
};
+class SwimError: public Exception {
+public:
+ SwimError(const char *file, unsigned line, const char *format, ...);
+ virtual void raise() { throw this; }
+};
+
/**
* Initialize the exception subsystem.
*/
diff --git a/src/lib/swim/CMakeLists.txt b/src/lib/swim/CMakeLists.txt
new file mode 100644
index 000000000..8ba99d803
--- /dev/null
+++ b/src/lib/swim/CMakeLists.txt
@@ -0,0 +1,12 @@
+set(lib_swim_sources swim.c swim_io.c swim_proto.c)
+set(lib_swim_udp_sources swim_transport_udp.c)
+set(lib_swim_ev_sources swim_ev.c)
+
+set_source_files_compile_flags(${lib_swim_sources} ${lib_swim_udp_sources}
+ ${lib_swim_ev_sources})
+add_library(swim STATIC ${lib_swim_sources})
+target_link_libraries(swim core misc uuid)
+add_library(swim_udp STATIC ${lib_swim_udp_sources})
+target_link_libraries(swim_udp core)
+add_library(swim_ev STATIC ${lib_swim_ev_sources})
+target_link_libraries(swim_ev core)
diff --git a/src/lib/swim/swim.c b/src/lib/swim/swim.c
new file mode 100644
index 000000000..19d7935b0
--- /dev/null
+++ b/src/lib/swim/swim.c
@@ -0,0 +1,985 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "swim.h"
+#include "swim_io.h"
+#include "swim_proto.h"
+#include "swim_ev.h"
+#include "uri/uri.h"
+#include "fiber.h"
+#include "msgpuck.h"
+#include "info/info.h"
+#include "assoc.h"
+#include "sio.h"
+
+/**
+ * SWIM - Scalable Weakly-consistent Infection-style Process Group
+ * Membership Protocol. It consists of 2 components: events
+ * dissemination and failure detection, and stores in memory a
+ * table of known remote hosts - members. Also some SWIM
+ * implementations have an additional component: anti-entropy -
+ * periodical broadcast of a random subset of the member table.
+ *
+ * Each SWIM component is different in both protocol payload and
+ * goals, and could even use different messages to send data. But
+ * SWIM describes piggybacking of messages: a ping message can
+ * piggyback a dissemination's one.
+ *
+ * SWIM has a main operating cycle during which it randomly
+ * chooses members from a member table and sends to them events +
+ * ping. Replies are processed out of the main cycle,
+ * asynchronously.
+ *
+ * Random selection provides even network load of ~1 message on
+ * each member per one protocol step regardless of the cluster
+ * size. Without randomness each member would receive a network
+ * load of N messages in each protocol step, where N is the
+ * cluster size.
+ *
+ * To speed up propagation of new information by means of a few
+ * random messages SWIM proposes a kind of fairness: when
+ * selecting a next random member to ping, the protocol prefers
+ * LRU members. In code it would be too complicated, so
+ * Tarantool's implementation is slightly different, easier:
+ *
+ * Tarantool splits protocol operation into rounds. At the
+ * beginning of a round all members are randomly reordered and
+ * linked into a list. At each round step a member is popped from
+ * the list head, a message is sent to it, and then it waits for
+ * the next round. In such implementation all random selection of
+ * the original SWIM is executed once per round. The round is
+ * 'planned', actually. A list is used instead of an array since
+ * new members can be added to its tail without realloc, and dead
+ * members can be removed easily as well.
+ *
+ * Also Tarantool implements the third SWIM component -
+ * anti-entropy. Why is it needed and even vital? Consider the
+ * example: two SWIM nodes, both are alive. Nothing happens, so
+ * the events list is empty, only pings are being sent
+ * periodically. Then a third node appears. It knows about one of
+ * the existing nodes. How can it learn about the rest? Sure,
+ * its known counterpart can try to notify its peer, but it is
+ * UDP, so this event can be lost. Anti-entropy is an extra simple
+ * component, it just piggybacks random part of the member table
+ * with each regular message. In the example above the new node
+ * will learn about the third one via anti-entropy messages from
+ * the second one sooner or later.
+ *
+ * Surprisingly, original SWIM does not describe any addressing,
+ * how to uniquely identify a member. IP/port fallaciously could
+ * be considered as a good unique identifier, but some arguments
+ * below demolish this belief:
+ *
+ * - if instances work in separate containers, they can have
+ * the same IP/port inside a container NATed to a unique
+ * IP/port outside the container;
+ *
+ * - IP/port are likely to change during instance lifecycle.
+ * Once IP/port are changed, a ghost of the old member's
+ * configuration still lives for a while until it is
+ * suspected, dead and GC-ed. Taking into account that ACK
+ * timeout can be tens of seconds, 'Dead Souls' can exist
+ * unpleasantly long.
+ *
+ * Tarantool SWIM implementation uses UUIDs as unique identifiers.
+ * UUID is much more unlikely to change than IP/port. But even if
+ * that happens, dissemination component for a while gossips the
+ * new UUID together with the old one.
+ *
+ * SWIM implementation is split into 3 parts: protocol logic,
+ * transport level, protocol structure.
+ *
+ * - protocol logic consists of how to react on various
+ * events, failure detection pings/acks, how often to send
+ * messages, handles the logic of the three components
+ * (failure detection, anti-entropy, dissemination);
+ *
+ * - transport level handles routing, transport headers,
+ * packet forwarding;
+ *
+ * - protocol structure describes how packet looks in
+ * MessagePack, how sections and headers follow each other.
+ */
+
+enum {
+ /**
+ * How often to send membership messages and pings in
+ * seconds. Nothing special in this concrete default
+ * value.
+ */
+ HEARTBEAT_RATE_DEFAULT = 1,
+};
+
+/**
+ * Return a random number within given boundaries.
+ *
+ * Instead of blindly calculating a modulo, scale the random
+ * number down the given boundaries to preserve the original
+ * distribution. The result belongs to range [start, end].
+ */
+static inline int
+swim_scaled_rand(int start, int end)
+{
+ assert(end >= start);
+ /*
+ * RAND_MAX is likely to be INT_MAX - hardly SWIM will
+ * ever be used in such a huge cluster.
+ */
+ assert(end - start < RAND_MAX);
+ return rand() / (RAND_MAX / (end - start + 1) + 1);
+}
+
+/** Calculate UUID hash to use as a member table key. */
+static inline uint32_t
+swim_uuid_hash(const struct tt_uuid *uuid)
+{
+ return mh_strn_hash((const char *) uuid, UUID_LEN);
+}
+
+/**
+ * A helper to not call tt_static_buf() in all places where it is
+ * used to get string UUID.
+ */
+static inline const char *
+swim_uuid_str(const struct tt_uuid *uuid)
+{
+ char *buf = tt_static_buf();
+ tt_uuid_to_string(uuid, buf);
+ return buf;
+}
+
+/**
+ * A cluster member description. This structure describes the
+ * last known state of an instance. This state is updated
+ * periodically via UDP according to SWIM protocol rules.
+ */
+struct swim_member {
+ /**
+ * Member status. Since the communication goes via UDP,
+ * actual status can be different, as well as different on
+ * other SWIM nodes. But SWIM guarantees that each member
+ * will learn a real status of an instance sometime.
+ */
+ enum swim_member_status status;
+ /**
+ * Address of the instance to which to send UDP packets.
+ */
+ struct sockaddr_in addr;
+ /**
+ * A unique identifier of the member. Is used as a key in
+ * the mebmers table.
+ */
+ struct tt_uuid uuid;
+ /**
+ * Cached hash of the uuid for the members table lookups.
+ */
+ uint32_t hash;
+ /**
+ * Position in a queue of members in the current round.
+ */
+ struct rlist in_round_queue;
+};
+
+#define mh_name _swim_table
+struct mh_swim_table_key {
+ uint32_t hash;
+ const struct tt_uuid *uuid;
+};
+#define mh_key_t struct mh_swim_table_key
+#define mh_node_t struct swim_member *
+#define mh_arg_t void *
+#define mh_hash(a, arg) ((*a)->hash)
+#define mh_hash_key(a, arg) (a.hash)
+#define mh_cmp(a, b, arg) (tt_uuid_compare(&(*a)->uuid, &(*b)->uuid))
+#define mh_cmp_key(a, b, arg) (tt_uuid_compare(a.uuid, &(*b)->uuid))
+#define MH_SOURCE 1
+#include "salad/mhash.h"
+
+/**
+ * SWIM instance. Stores configuration, manages periodical tasks,
+ * rounds. Each member has an object of this type on its host,
+ * while on others it is represented as a struct swim_member
+ * object.
+ */
+struct swim {
+ /**
+ * Global hash of all known members of the cluster. Hash
+ * key is UUID, value is a struct member, describing a
+ * remote instance. Discovered members live here until
+ * they are detected as dead - in such a case they are
+ * removed from the hash after a while.
+ */
+ struct mh_swim_table_t *members;
+ /**
+ * This node. Is used to not send messages to self, it's
+ * meaningless. Also to refute false gossips about self
+ * status.
+ */
+ struct swim_member *self;
+ /**
+ * Members to which a message should be sent next during
+ * this round.
+ */
+ struct rlist round_queue;
+ /** Generator of round step events. */
+ struct ev_timer round_tick;
+ /**
+ * Single round step task. It is impossible to have
+ * multiple round steps in the same SWIM instance at the
+ * same time, so it is single and preallocated per SWIM
+ * instance.
+ */
+ struct swim_task round_step_task;
+ /**
+ * Preallocated buffer to store shuffled members here at
+ * the beginning of each round.
+ */
+ struct swim_member **shuffled;
+ /**
+ * Scheduler of output requests, receiver of incoming
+ * ones.
+ */
+ struct swim_scheduler scheduler;
+ /**
+ * Swim instance is its own iterator. Such iterators are
+ * obviously unstable and fragile, but it is good enough
+ * for a fast non-yielding scan of members table. Each new
+ * public API iterator iterates through member table using
+ * this variable as a storage of the current position.
+ */
+ mh_int_t iterator;
+};
+
+/**
+ * SWIM fd mainly is needed to be printed into the logs in order
+ * to distinguish between different SWIM instances logs.
+ */
+static inline int
+swim_fd(const struct swim *swim)
+{
+ return swim->scheduler.transport.fd;
+}
+
+/**
+ * A helper to get a pointer to a SWIM instance having only a
+ * pointer to its scheduler. It is used by task complete
+ * functions.
+ */
+static inline struct swim *
+swim_by_scheduler(struct swim_scheduler *scheduler)
+{
+ return container_of(scheduler, struct swim, scheduler);
+}
+
+/** Free member's resources. */
+static inline void
+swim_member_delete(struct swim_member *member)
+{
+ assert(rlist_empty(&member->in_round_queue));
+ free(member);
+}
+
+/**
+ * Reserve space for one more member in the member table. Used to
+ * execute a non-failing UUID update.
+ */
+static inline int
+swim_reserve_one_member(struct swim *swim)
+{
+ if (mh_swim_table_reserve(swim->members, mh_size(swim->members) + 1,
+ NULL) != 0) {
+ diag_set(OutOfMemory, sizeof(mh_int_t), "malloc", "node");
+ return -1;
+ }
+ return 0;
+}
+
+/** Create a new member. It is not registered anywhere here. */
+static struct swim_member *
+swim_member_new(const struct sockaddr_in *addr, const struct tt_uuid *uuid,
+ enum swim_member_status status)
+{
+ struct swim_member *member =
+ (struct swim_member *) calloc(1, sizeof(*member));
+ if (member == NULL) {
+ diag_set(OutOfMemory, sizeof(*member), "calloc", "member");
+ return NULL;
+ }
+ member->status = status;
+ member->addr = *addr;
+ member->uuid = *uuid;
+ member->hash = swim_uuid_hash(uuid);
+ rlist_create(&member->in_round_queue);
+ return member;
+}
+
+/**
+ * Remove the member from all queues, hashes, destroy it and free
+ * the memory.
+ */
+static void
+swim_delete_member(struct swim *swim, struct swim_member *member)
+{
+ say_verbose("SWIM %d: member %s is deleted", swim_fd(swim),
+ swim_uuid_str(&member->uuid));
+ 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));
+ mh_swim_table_del(swim->members, rc, NULL);
+ rlist_del_entry(member, in_round_queue);
+ swim_member_delete(member);
+}
+
+/** Find a member by UUID. */
+static inline struct swim_member *
+swim_find_member(struct swim *swim, const struct tt_uuid *uuid)
+{
+ struct mh_swim_table_key key = {swim_uuid_hash(uuid), uuid};
+ mh_int_t node = mh_swim_table_find(swim->members, key, NULL);
+ if (node == mh_end(swim->members))
+ return NULL;
+ return *mh_swim_table_node(swim->members, node);
+}
+
+/**
+ * Register a new member with a specified status. It is not added
+ * to the round queue here. It waits until the current round is
+ * finished, and then is included into a new round. It is done
+ * mainly to not add self into the round queue, because self is
+ * also created via this function.
+ */
+static struct swim_member *
+swim_new_member(struct swim *swim, const struct sockaddr_in *addr,
+ const struct tt_uuid *uuid, enum swim_member_status status)
+{
+ int new_bsize = sizeof(swim->shuffled[0]) *
+ (mh_size(swim->members) + 1);
+ struct swim_member **new_shuffled =
+ (struct swim_member **) realloc(swim->shuffled, new_bsize);
+ if (new_shuffled == NULL) {
+ diag_set(OutOfMemory, new_bsize, "realloc", "new_shuffled");
+ return NULL;
+ }
+ swim->shuffled = new_shuffled;
+ struct swim_member *member = swim_member_new(addr, uuid, status);
+ if (member == NULL)
+ return NULL;
+ assert(swim_find_member(swim, uuid) == NULL);
+ mh_int_t rc = mh_swim_table_put(swim->members,
+ (const struct swim_member **) &member,
+ NULL, NULL);
+ if (rc == mh_end(swim->members)) {
+ swim_member_delete(member);
+ diag_set(OutOfMemory, sizeof(mh_int_t), "malloc", "node");
+ return NULL;
+ }
+ swim_ev_timer_start(loop(), &swim->round_tick);
+ say_verbose("SWIM %d: member %s is added, total is %d", swim_fd(swim),
+ swim_uuid_str(&member->uuid), mh_size(swim->members));
+ return member;
+}
+
+/**
+ * Take all the members from the table and shuffle them randomly.
+ * Is used for forthcoming round planning.
+ */
+static void
+swim_shuffle_members(struct swim *swim)
+{
+ struct mh_swim_table_t *members = swim->members;
+ int i = 0;
+ /*
+ * This shuffling preserves even distribution of a random
+ * sequence. The distribution properties have been
+ * verified by a longevity test.
+ */
+ for (mh_int_t node = mh_first(members), end = mh_end(members);
+ node != end; node = mh_next(members, node), ++i) {
+ swim->shuffled[i] = *mh_swim_table_node(members, node);
+ int j = swim_scaled_rand(0, i);
+ SWAP(swim->shuffled[i], swim->shuffled[j]);
+ }
+}
+
+/**
+ * Shuffle members, build randomly ordered queue of addressees. In
+ * other words, do all round preparation work.
+ */
+static int
+swim_new_round(struct swim *swim)
+{
+ int size = mh_size(swim->members);
+ if (size == 1) {
+ assert(swim->self != NULL);
+ say_verbose("SWIM %d: skip a round - no members",
+ swim_fd(swim));
+ return 0;
+ }
+ say_verbose("SWIM %d: start a new round with %d members", swim_fd(swim),
+ size);
+ swim_shuffle_members(swim);
+ rlist_create(&swim->round_queue);
+ for (int i = 0; i < size; ++i) {
+ if (swim->shuffled[i] != swim->self) {
+ rlist_add_entry(&swim->round_queue, swim->shuffled[i],
+ in_round_queue);
+ }
+ }
+ return 0;
+}
+
+/**
+ * Encode anti-entropy header and random members data as many as
+ * possible to the end of the packet.
+ * @retval 0 Not error, but nothing is encoded.
+ * @retval 1 Something is encoded.
+ */
+static int
+swim_encode_anti_entropy(struct swim *swim, struct swim_packet *packet)
+{
+ struct swim_anti_entropy_header_bin ae_header_bin;
+ struct swim_member_bin member_bin;
+ int size = sizeof(ae_header_bin);
+ char *header = swim_packet_reserve(packet, size);
+ if (header == NULL)
+ return 0;
+ char *pos = header;
+ swim_member_bin_create(&member_bin);
+ struct mh_swim_table_t *t = swim->members;
+ int i = 0, member_count = mh_size(t);
+ int rnd = swim_scaled_rand(0, member_count - 1);
+ for (mh_int_t rc = mh_swim_table_random(t, rnd), end = mh_end(t);
+ i < member_count; ++i) {
+ struct swim_member *m = *mh_swim_table_node(t, rc);
+ int new_size = size + sizeof(member_bin);
+ if (swim_packet_reserve(packet, new_size) == NULL)
+ break;
+ swim_member_bin_fill(&member_bin, &m->addr, &m->uuid,
+ m->status);
+ memcpy(pos + size, &member_bin, sizeof(member_bin));
+ size = new_size;
+ /*
+ * First random member could be choosen too close
+ * to the hash end. Here the cycle is wrapped, if
+ * a packet still has free memory, but the
+ * iterator has already reached the hash end.
+ */
+ rc = mh_next(t, rc);
+ if (rc == end)
+ rc = mh_first(t);
+ }
+ if (i == 0)
+ return 0;
+ swim_packet_advance(packet, size);
+ swim_anti_entropy_header_bin_create(&ae_header_bin, i);
+ memcpy(header, &ae_header_bin, sizeof(ae_header_bin));
+ return 1;
+}
+
+/**
+ * Encode source UUID.
+ * @retval 0 Not error, but nothing is encoded.
+ * @retval 1 Something is encoded.
+ */
+static inline int
+swim_encode_src_uuid(struct swim *swim, struct swim_packet *packet)
+{
+ struct swim_src_uuid_bin uuid_bin;
+ char *pos = swim_packet_alloc(packet, sizeof(uuid_bin));
+ if (pos == NULL)
+ return 0;
+ swim_src_uuid_bin_create(&uuid_bin, &swim->self->uuid);
+ memcpy(pos, &uuid_bin, sizeof(uuid_bin));
+ return 1;
+}
+
+/** Encode SWIM components into a UDP packet. */
+static void
+swim_encode_round_msg(struct swim *swim, struct swim_packet *packet)
+{
+ swim_packet_create(packet);
+ char *header = swim_packet_alloc(packet, 1);
+ int map_size = 0;
+ map_size += swim_encode_src_uuid(swim, packet);
+ map_size += swim_encode_anti_entropy(swim, packet);
+
+ assert(mp_sizeof_map(map_size) == 1 && map_size == 2);
+ mp_encode_map(header, map_size);
+}
+
+/**
+ * Once per specified timeout trigger a next round step. In round
+ * step a next memeber is taken from the round queue and a round
+ * message is sent to him. One member per step.
+ */
+static void
+swim_begin_step(struct ev_loop *loop, struct ev_timer *t, int events)
+{
+ assert((events & EV_TIMER) != 0);
+ (void) events;
+ (void) loop;
+ struct swim *swim = (struct swim *) t->data;
+ if (rlist_empty(&swim->round_queue) && swim_new_round(swim) != 0) {
+ diag_log();
+ return;
+ }
+ /*
+ * Possibly empty, if no members but self are specified.
+ */
+ if (rlist_empty(&swim->round_queue))
+ return;
+
+ swim_encode_round_msg(swim, &swim->round_step_task.packet);
+ struct swim_member *m =
+ rlist_first_entry(&swim->round_queue, struct swim_member,
+ in_round_queue);
+ swim_task_send(&swim->round_step_task, &m->addr, &swim->scheduler);
+}
+
+/**
+ * After a round message is sent, the addressee can be popped from
+ * the queue, and the next step is scheduled.
+ */
+static void
+swim_complete_step(struct swim_task *task,
+ struct swim_scheduler *scheduler, int rc)
+{
+ (void) rc;
+ (void) task;
+ struct swim *swim = swim_by_scheduler(scheduler);
+ swim_ev_timer_start(loop(), &swim->round_tick);
+ rlist_shift_entry(&swim->round_queue, struct swim_member,
+ in_round_queue);
+}
+
+/**
+ * Update member's UUID if it is changed. On UUID change the
+ * member is reinserted into the member table with a new UUID.
+ * @retval 0 Success.
+ * @retval -1 Error. Out of memory or the new UUID is already in
+ * use.
+ */
+static int
+swim_update_member_uuid(struct swim *swim, struct swim_member *member,
+ const struct tt_uuid *new_uuid)
+{
+ if (tt_uuid_is_equal(new_uuid, &member->uuid))
+ return 0;
+ if (swim_find_member(swim, new_uuid) != NULL) {
+ diag_set(SwimError, "duplicate UUID '%s'",
+ swim_uuid_str(new_uuid));
+ return -1;
+ }
+ /*
+ * Reserve before put + delete, because put below can
+ * call rehash, and a reference to the old place in the
+ * hash will taint.
+ */
+ if (swim_reserve_one_member(swim) != 0)
+ return -1;
+ struct mh_swim_table_t *t = swim->members;
+ struct tt_uuid old_uuid = member->uuid;
+ struct mh_swim_table_key key = {member->hash, &old_uuid};
+ mh_int_t old_rc = mh_swim_table_find(t, key, NULL);
+ assert(old_rc != mh_end(t));
+ member->uuid = *new_uuid;
+ member->hash = swim_uuid_hash(new_uuid);
+ mh_int_t new_rc =
+ mh_swim_table_put(t, (const struct swim_member **) &member,
+ NULL, NULL);
+ /* Can not fail - reserved above. */
+ assert(new_rc != mh_end(t));
+ (void) new_rc;
+ /*
+ * Old_rc is still valid, because a possible rehash
+ * happened before put.
+ */
+ mh_swim_table_del(t, old_rc, NULL);
+ say_verbose("SWIM %d: a member has changed its UUID from %s to %s",
+ swim_fd(swim), swim_uuid_str(&old_uuid),
+ swim_uuid_str(new_uuid));
+ return 0;
+}
+
+/** Update member's address.*/
+static inline void
+swim_update_member_addr(struct swim *swim, struct swim_member *member,
+ const struct sockaddr_in *addr)
+{
+ (void) swim;
+ member->addr = *addr;
+}
+
+/**
+ * Update or create a member by its definition, received from a
+ * remote instance.
+ * @retval NULL Error.
+ * @retval New member, or updated old member.
+ */
+static struct swim_member *
+swim_update_member(struct swim *swim, const struct swim_member_def *def)
+{
+ struct swim_member *member = swim_find_member(swim, &def->uuid);
+ if (member == NULL) {
+ member = swim_new_member(swim, &def->addr, &def->uuid,
+ def->status);
+ return member;
+ }
+ struct swim_member *self = swim->self;
+ if (member != self)
+ swim_update_member_addr(swim, member, &def->addr);
+ return member;
+}
+
+/** Decode an anti-entropy message, update member table. */
+static int
+swim_process_anti_entropy(struct swim *swim, const char **pos, const char *end)
+{
+ const char *prefix = "invalid anti-entropy message:";
+ uint32_t size;
+ if (swim_decode_array(pos, end, &size, prefix, "root") != 0)
+ return -1;
+ for (uint64_t i = 0; i < size; ++i) {
+ struct swim_member_def def;
+ if (swim_member_def_decode(&def, pos, end, prefix) != 0)
+ return -1;
+ if (swim_update_member(swim, &def) == NULL) {
+ /*
+ * Not a critical error. Other members
+ * still can be updated.
+ */
+ diag_log();
+ }
+ }
+ return 0;
+}
+
+/** Process a new message. */
+static void
+swim_on_input(struct swim_scheduler *scheduler, const char *pos,
+ const char *end, const struct sockaddr_in *src)
+{
+ (void) src;
+ const char *prefix = "invalid message:";
+ struct swim *swim = swim_by_scheduler(scheduler);
+ struct tt_uuid uuid;
+ uint32_t size;
+ if (swim_decode_map(&pos, end, &size, prefix, "root") != 0)
+ goto error;
+ if (size == 0) {
+ diag_set(SwimError, "%s body can not be empty", prefix);
+ goto error;
+ }
+ uint64_t key;
+ if (swim_decode_uint(&pos, end, &key, prefix, "a key") != 0)
+ goto error;
+ if (key != SWIM_SRC_UUID) {
+ diag_set(SwimError, "%s first key should be source UUID",
+ prefix);
+ goto error;
+ }
+ if (swim_decode_uuid(&uuid, &pos, end, prefix, "source uuid") != 0)
+ goto error;
+ --size;
+ for (uint32_t i = 0; i < size; ++i) {
+ if (swim_decode_uint(&pos, end, &key, prefix, "a key") != 0)
+ goto error;
+ switch(key) {
+ case SWIM_ANTI_ENTROPY:
+ say_verbose("SWIM %d: process anti-entropy",
+ swim_fd(swim));
+ if (swim_process_anti_entropy(swim, &pos, end) != 0)
+ goto error;
+ break;
+ default:
+ diag_set(SwimError, "%s unexpected key", prefix);
+ goto error;
+ }
+ }
+ return;
+error:
+ diag_log();
+}
+
+struct swim *
+swim_new(void)
+{
+ struct swim *swim = (struct swim *) calloc(1, sizeof(*swim));
+ if (swim == NULL) {
+ diag_set(OutOfMemory, sizeof(*swim), "calloc", "swim");
+ return NULL;
+ }
+ swim->members = mh_swim_table_new();
+ if (swim->members == NULL) {
+ free(swim);
+ diag_set(OutOfMemory, sizeof(*swim->members),
+ "mh_swim_table_new", "members");
+ return NULL;
+ }
+ rlist_create(&swim->round_queue);
+ swim_ev_timer_init(&swim->round_tick, swim_begin_step,
+ HEARTBEAT_RATE_DEFAULT, 0);
+ swim->round_tick.data = (void *) swim;
+ swim_task_create(&swim->round_step_task, swim_complete_step, NULL);
+ swim_scheduler_create(&swim->scheduler, swim_on_input);
+ return swim;
+}
+
+/**
+ * Parse URI, filter out everything but IP addresses and ports,
+ * and fill a struct sockaddr_in.
+ */
+static inline int
+swim_uri_to_addr(const char *uri, struct sockaddr_in *addr,
+ const char *prefix)
+{
+ struct sockaddr_storage storage;
+ if (sio_uri_to_addr(uri, (struct sockaddr *) &storage) != 0)
+ return -1;
+ if (storage.ss_family != AF_INET) {
+ diag_set(IllegalParams, "%s only IP sockets are supported",
+ prefix);
+ return -1;
+ }
+ *addr = *((struct sockaddr_in *) &storage);
+ if (addr->sin_addr.s_addr == 0) {
+ diag_set(IllegalParams, "%s INADDR_ANY is not supported",
+ prefix);
+ return -1;
+ }
+ return 0;
+}
+
+int
+swim_cfg(struct swim *swim, const char *uri, double heartbeat_rate,
+ const struct tt_uuid *uuid)
+{
+ const char *prefix = "swim.cfg:";
+ struct sockaddr_in addr;
+ if (uri != NULL && swim_uri_to_addr(uri, &addr, prefix) != 0)
+ return -1;
+ bool is_first_cfg = swim->self == NULL;
+ if (is_first_cfg) {
+ if (uuid == NULL || tt_uuid_is_nil(uuid) || uri == NULL) {
+ diag_set(SwimError, "%s UUID and URI are mandatory in "\
+ "a first config", prefix);
+ return -1;
+ }
+ swim->self = swim_new_member(swim, &addr, uuid, MEMBER_ALIVE);
+ if (swim->self == NULL)
+ return -1;
+ } else if (uuid == NULL || tt_uuid_is_nil(uuid)) {
+ uuid = &swim->self->uuid;
+ } else if (! tt_uuid_is_equal(uuid, &swim->self->uuid)) {
+ if (swim_find_member(swim, uuid) != NULL) {
+ diag_set(SwimError, "%s a member with such UUID "\
+ "already exists", prefix);
+ return -1;
+ }
+ /*
+ * Reserve one cell for reinsertion of self with a
+ * new UUID. Reserve is necessary right here, not
+ * later, for atomic reconfiguration. Without
+ * reservation in that place it is possible that
+ * the instance is bound to a new URI, but failed
+ * to update UUID due to memory issues.
+ */
+ if (swim_reserve_one_member(swim) != 0)
+ return -1;
+ }
+ if (uri != NULL) {
+ /*
+ * Bind is smart - it does nothing if the address
+ * was not changed.
+ */
+ if (swim_scheduler_bind(&swim->scheduler, &addr) != 0) {
+ if (is_first_cfg) {
+ swim_delete_member(swim, swim->self);
+ swim->self = NULL;
+ }
+ return -1;
+ }
+ /*
+ * A real address can be different from a one
+ * passed by user. For example, if 0 port was
+ * specified.
+ */
+ addr = swim->scheduler.transport.addr;
+ }
+ if (swim->round_tick.at != heartbeat_rate && heartbeat_rate > 0)
+ swim_ev_timer_set(&swim->round_tick, heartbeat_rate, 0);
+
+ swim_ev_timer_start(loop(), &swim->round_tick);
+ swim_update_member_addr(swim, swim->self, &addr);
+ int rc = swim_update_member_uuid(swim, swim->self, uuid);
+ /* Reserved above. */
+ assert(rc == 0);
+ (void) rc;
+ return 0;
+}
+
+bool
+swim_is_configured(const struct swim *swim)
+{
+ return swim->self != NULL;
+}
+
+int
+swim_add_member(struct swim *swim, const char *uri, const struct tt_uuid *uuid)
+{
+ const char *prefix = "swim.add_member:";
+ assert(swim_is_configured(swim));
+ if (uri == NULL || uuid == NULL || tt_uuid_is_nil(uuid)) {
+ diag_set(SwimError, "%s URI and UUID are mandatory", prefix);
+ return -1;
+ }
+ struct sockaddr_in addr;
+ if (swim_uri_to_addr(uri, &addr, prefix) != 0)
+ return -1;
+ struct swim_member *member = swim_find_member(swim, uuid);
+ if (member == NULL) {
+ member = swim_new_member(swim, &addr, uuid, MEMBER_ALIVE);
+ return member == NULL ? -1 : 0;
+ }
+ diag_set(SwimError, "%s a member with such UUID already exists",
+ prefix);
+ return -1;
+}
+
+int
+swim_remove_member(struct swim *swim, const struct tt_uuid *uuid)
+{
+ assert(swim_is_configured(swim));
+ const char *prefix = "swim.remove_member:";
+ if (uuid == NULL || tt_uuid_is_nil(uuid)) {
+ diag_set(SwimError, "%s UUiD is mandatory", prefix);
+ return -1;
+ }
+ struct swim_member *member = swim_find_member(swim, uuid);
+ if (member == NULL)
+ return 0;
+ if (member == swim->self) {
+ diag_set(SwimError, "%s can not remove self", prefix);
+ return -1;
+ }
+ swim_delete_member(swim, member);
+ return 0;
+}
+
+void
+swim_info(struct swim *swim, struct info_handler *info)
+{
+ info_begin(info);
+ for (mh_int_t node = mh_first(swim->members),
+ end = mh_end(swim->members); node != end;
+ node = mh_next(swim->members, node)) {
+ struct swim_member *m =
+ *mh_swim_table_node(swim->members, node);
+ info_table_begin(info,
+ sio_strfaddr((struct sockaddr *) &m->addr,
+ sizeof(m->addr)));
+ info_append_str(info, "status",
+ swim_member_status_strs[m->status]);
+ info_append_str(info, "uuid", swim_uuid_str(&m->uuid));
+ info_table_end(info);
+ }
+ info_end(info);
+}
+
+void
+swim_delete(struct swim *swim)
+{
+ swim_scheduler_destroy(&swim->scheduler);
+ swim_ev_timer_stop(loop(), &swim->round_tick);
+ swim_task_destroy(&swim->round_step_task);
+ mh_int_t node = mh_first(swim->members);
+ while (node != mh_end(swim->members)) {
+ struct swim_member *m =
+ *mh_swim_table_node(swim->members, node);
+ swim_delete_member(swim, m);
+ node = mh_first(swim->members);
+ }
+ mh_swim_table_delete(swim->members);
+ free(swim->shuffled);
+}
+
+const struct swim_member *
+swim_self(struct swim *swim)
+{
+ assert(swim_is_configured(swim));
+ return swim->self;
+}
+
+const struct swim_member *
+swim_member_by_uuid(struct swim *swim, const struct tt_uuid *uuid)
+{
+ assert(swim_is_configured(swim));
+ return swim_find_member(swim, uuid);
+}
+
+struct swim_iterator *
+swim_iterator_open(struct swim *swim)
+{
+ assert(swim_is_configured(swim));
+ swim->iterator = mh_first(swim->members);
+ return (struct swim_iterator *) swim;
+}
+
+const struct swim_member *
+swim_iterator_next(struct swim_iterator *iterator)
+{
+ struct swim *swim = (struct swim *) iterator;
+ assert(swim_is_configured(swim));
+ struct mh_swim_table_t *t = swim->members;
+ if (swim->iterator == mh_end(t))
+ return NULL;
+ struct swim_member *m = *mh_swim_table_node(t, swim->iterator);
+ swim->iterator = mh_next(t, swim->iterator);
+ return m;
+}
+
+void
+swim_iterator_close(struct swim_iterator *iterator)
+{
+ (void) iterator;
+}
+
+const char *
+swim_member_uri(const struct swim_member *member)
+{
+ return sio_strfaddr((const struct sockaddr *) &member->addr,
+ sizeof(member->addr));
+}
+
+const struct tt_uuid *
+swim_member_uuid(const struct swim_member *member)
+{
+ return &member->uuid;
+}
diff --git a/src/lib/swim/swim.h b/src/lib/swim/swim.h
new file mode 100644
index 000000000..8b82df051
--- /dev/null
+++ b/src/lib/swim/swim.h
@@ -0,0 +1,141 @@
+#ifndef TARANTOOL_SWIM_H_INCLUDED
+#define TARANTOOL_SWIM_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include <stdbool.h>
+
+#if defined(__cplusplus)
+extern "C" {
+#endif
+
+struct info_handler;
+struct swim;
+struct tt_uuid;
+struct swim_iterator;
+struct swim_member;
+
+/**
+ * Create a new SWIM instance. Do not bind to a port or set any
+ * parameters. Allocation and initialization only.
+ */
+struct swim *
+swim_new(void);
+
+/** Check if a swim instance is configured. */
+bool
+swim_is_configured(const struct swim *swim);
+
+/**
+ * Configure or reconfigure a SWIM instance.
+ *
+ * @param swim SWIM instance to configure.
+ * @param uri URI in the format "ip:port".
+ * @param heartbeat_rate Rate of sending round messages. It does
+ * not mean that each member will be checked each
+ * @heartbeat_rate seconds. It is rather the protocol
+ * speed. Protocol period depends on member count and
+ * @heartbeat_rate.
+ * @param uuid UUID of this instance. Must be unique over the
+ * cluster.
+ *
+ * @retval 0 Success.
+ * @retval -1 Error. Memory, not unique UUID, system error.
+ */
+int
+swim_cfg(struct swim *swim, const char *uri, double heartbeat_rate,
+ const struct tt_uuid *uuid);
+
+/**
+ * Stop listening and broadcasting messages, cleanup all internal
+ * structures, free memory.
+ */
+void
+swim_delete(struct swim *swim);
+
+/** Add a new member. */
+int
+swim_add_member(struct swim *swim, const char *uri, const struct tt_uuid *uuid);
+
+/** Silently remove a member from member table. */
+int
+swim_remove_member(struct swim *swim, const struct tt_uuid *uuid);
+
+/** Dump member statuses into @a info. */
+void
+swim_info(struct swim *swim, struct info_handler *info);
+
+/** Get a SWIM member, describing this instance. */
+const struct swim_member *
+swim_self(struct swim *swim);
+
+/**
+ * Find a member by its UUID in the local member table.
+ * @retval NULL Not found.
+ * @retval not NULL A member.
+ */
+const struct swim_member *
+swim_member_by_uuid(struct swim *swim, const struct tt_uuid *uuid);
+
+/**
+ * Open an iterator to scan the whole member table. The iterator
+ * is not stable. It means, that a caller can not yield between
+ * open and close - iterator position can be lost. Also it is
+ * impossible to open more than one iterator on one SWIM instance
+ * at the same time.
+ */
+struct swim_iterator *
+swim_iterator_open(struct swim *swim);
+
+/**
+ * Get a next SWIM member.
+ * @retval NULL EOF.
+ * @retval not NULL A valid member.
+ */
+const struct swim_member *
+swim_iterator_next(struct swim_iterator *iterator);
+
+/** Close an iterator. */
+void
+swim_iterator_close(struct swim_iterator *iterator);
+
+/** Member's URI. */
+const char *
+swim_member_uri(const struct swim_member *member);
+
+/** Member's UUID. */
+const struct tt_uuid *
+swim_member_uuid(const struct swim_member *member);
+
+#if defined(__cplusplus)
+}
+#endif
+
+#endif /* TARANTOOL_SWIM_H_INCLUDED */
diff --git a/src/lib/swim/swim_ev.c b/src/lib/swim/swim_ev.c
new file mode 100644
index 000000000..f7c464426
--- /dev/null
+++ b/src/lib/swim/swim_ev.c
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "swim_ev.h"
+#include "tarantool_ev.h"
+#include "fiber.h"
+
+double
+swim_time(void)
+{
+ return fiber_clock();
+}
+
+void
+swim_ev_timer_start(struct ev_loop *loop, struct ev_timer *watcher)
+{
+ ev_timer_start(loop, watcher);
+}
+
+void
+swim_ev_timer_stop(struct ev_loop *loop, struct ev_timer *watcher)
+{
+ ev_timer_stop(loop, watcher);
+}
diff --git a/src/lib/swim/swim_ev.h b/src/lib/swim/swim_ev.h
new file mode 100644
index 000000000..8f41d43fe
--- /dev/null
+++ b/src/lib/swim/swim_ev.h
@@ -0,0 +1,64 @@
+#ifndef TARANTOOL_SWIM_EV_H_INCLUDED
+#define TARANTOOL_SWIM_EV_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+struct ev_loop;
+struct ev_timer;
+struct ev_io;
+
+/**
+ * Similar to swim transport, the functions below are compile time
+ * virtualized. Unit tests implement them in one way, and the
+ * server in another.
+ */
+
+double
+swim_time(void);
+
+void
+swim_ev_timer_start(struct ev_loop *loop, struct ev_timer *watcher);
+
+void
+swim_ev_timer_stop(struct ev_loop *loop, struct ev_timer *watcher);
+
+#define swim_ev_init ev_init
+
+#define swim_ev_timer_init ev_timer_init
+
+#define swim_ev_timer_set ev_timer_set
+
+#define swim_ev_io_start ev_io_start
+
+#define swim_ev_io_stop ev_io_stop
+
+#define swim_ev_io_set ev_io_set
+
+#endif /* TARANTOOL_SWIM_EV_H_INCLUDED */
diff --git a/src/lib/swim/swim_io.c b/src/lib/swim/swim_io.c
new file mode 100644
index 000000000..9c16d1ad3
--- /dev/null
+++ b/src/lib/swim/swim_io.c
@@ -0,0 +1,217 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "swim_io.h"
+#include "swim_proto.h"
+#include "swim_ev.h"
+#include "fiber.h"
+#include "sio.h"
+
+/**
+ * Allocate memory for meta. The same as mere alloc, but moves
+ * body pointer.
+ */
+static inline void
+swim_packet_alloc_meta(struct swim_packet *packet, int size)
+{
+ char *tmp = swim_packet_alloc(packet, size);
+ assert(tmp != NULL);
+ (void) tmp;
+ packet->body = packet->pos;
+}
+
+void
+swim_packet_create(struct swim_packet *packet)
+{
+ packet->body = packet->buf;
+ packet->pos = packet->body;
+ swim_packet_alloc_meta(packet, sizeof(struct swim_meta_header_bin));
+}
+
+void
+swim_task_create(struct swim_task *task, swim_task_f complete,
+ swim_task_f cancel)
+{
+ memset(task, 0, sizeof(*task));
+ task->complete = complete;
+ task->cancel = cancel;
+ swim_packet_create(&task->packet);
+ rlist_create(&task->in_queue_output);
+}
+
+/** Put the task into the queue of output tasks. */
+static inline void
+swim_task_schedule(struct swim_task *task, struct swim_scheduler *scheduler)
+{
+ assert(rlist_empty(&task->in_queue_output));
+ rlist_add_tail_entry(&scheduler->queue_output, task, in_queue_output);
+ swim_ev_io_start(loop(), &scheduler->output);
+}
+
+void
+swim_task_send(struct swim_task *task, const struct sockaddr_in *dst,
+ struct swim_scheduler *scheduler)
+{
+ task->dst = *dst;
+ swim_task_schedule(task, scheduler);
+}
+
+/**
+ * Scheduler fd mainly is needed to be printed into the logs in
+ * order to distinguish between different SWIM instances logs.
+ */
+static inline int
+swim_scheduler_fd(const struct swim_scheduler *scheduler)
+{
+ return scheduler->transport.fd;
+}
+
+/**
+ * Dispatch a next output event. Build packet meta and send the
+ * packet.
+ */
+static void
+swim_scheduler_on_output(struct ev_loop *loop, struct ev_io *io, int events);
+
+/**
+ * Dispatch a next input event. Unpack meta, forward a packet or
+ * propagate further to protocol logic.
+ */
+static void
+swim_scheduler_on_input(struct ev_loop *loop, struct ev_io *io, int events);
+
+void
+swim_scheduler_create(struct swim_scheduler *scheduler,
+ swim_scheduler_on_input_f on_input)
+{
+ swim_ev_init(&scheduler->output, swim_scheduler_on_output);
+ scheduler->output.data = (void *) scheduler;
+ swim_ev_init(&scheduler->input, swim_scheduler_on_input);
+ scheduler->input.data = (void *) scheduler;
+ rlist_create(&scheduler->queue_output);
+ scheduler->on_input = on_input;
+ swim_transport_create(&scheduler->transport);
+}
+
+int
+swim_scheduler_bind(struct swim_scheduler *scheduler,
+ const struct sockaddr_in *addr)
+{
+ struct swim_transport *t = &scheduler->transport;
+ if (swim_transport_bind(t, (const struct sockaddr *) addr,
+ sizeof(*addr)) != 0)
+ return -1;
+ swim_ev_io_set(&scheduler->output, t->fd, EV_WRITE);
+ swim_ev_io_set(&scheduler->input, t->fd, EV_READ);
+ swim_ev_io_start(loop(), &scheduler->input);
+ swim_ev_io_start(loop(), &scheduler->output);
+ return 0;
+}
+
+void
+swim_scheduler_destroy(struct swim_scheduler *scheduler)
+{
+ struct swim_task *t, *tmp;
+ /*
+ * Use 'safe', because cancelation can delete the task
+ * from the queue, or even delete the task itself.
+ */
+ rlist_foreach_entry_safe(t, &scheduler->queue_output, in_queue_output,
+ tmp) {
+ if (t->cancel != NULL)
+ t->cancel(t, scheduler, -1);
+ }
+ swim_transport_destroy(&scheduler->transport);
+ swim_ev_io_stop(loop(), &scheduler->output);
+ swim_ev_io_stop(loop(), &scheduler->input);
+}
+
+static void
+swim_scheduler_on_output(struct ev_loop *loop, struct ev_io *io, int events)
+{
+ assert((events & EV_WRITE) != 0);
+ (void) events;
+ struct swim_scheduler *scheduler = (struct swim_scheduler *) io->data;
+ if (rlist_empty(&scheduler->queue_output)) {
+ /*
+ * Possible, if a member pushed a task and then
+ * was deleted together with it.
+ */
+ swim_ev_io_stop(loop, io);
+ return;
+ }
+ struct swim_task *task =
+ rlist_shift_entry(&scheduler->queue_output, struct swim_task,
+ in_queue_output);
+ say_verbose("SWIM %d: send to %s", swim_scheduler_fd(scheduler),
+ sio_strfaddr((struct sockaddr *) &task->dst,
+ sizeof(task->dst)));
+ struct swim_meta_header_bin header;
+ swim_meta_header_bin_create(&header, &scheduler->transport.addr);
+ memcpy(task->packet.meta, &header, sizeof(header));
+ int rc = swim_transport_send(&scheduler->transport, task->packet.buf,
+ task->packet.pos - task->packet.buf,
+ (const struct sockaddr *) &task->dst,
+ sizeof(task->dst));
+ if (rc < 0)
+ diag_log();
+ if (task->complete != NULL)
+ task->complete(task, scheduler, rc);
+}
+
+static void
+swim_scheduler_on_input(struct ev_loop *loop, struct ev_io *io, int events)
+{
+ assert((events & EV_READ) != 0);
+ (void) events;
+ (void) loop;
+ struct swim_scheduler *scheduler = (struct swim_scheduler *) io->data;
+ struct sockaddr_in src;
+ socklen_t len = sizeof(src);
+ char buf[UDP_PACKET_SIZE];
+ ssize_t size = swim_transport_recv(&scheduler->transport, buf,
+ sizeof(buf),
+ (struct sockaddr *) &src, &len);
+ if (size <= 0) {
+ if (size < 0)
+ goto error;
+ return;
+ }
+ say_verbose("SWIM %d: received from %s", swim_scheduler_fd(scheduler),
+ sio_strfaddr((struct sockaddr *) &src, len));
+ struct swim_meta_def meta;
+ const char *pos = buf, *end = pos + size;
+ if (swim_meta_def_decode(&meta, &pos, end) < 0)
+ goto error;
+ scheduler->on_input(scheduler, pos, end, &meta.src);
+ return;
+error:
+ diag_log();
+}
diff --git a/src/lib/swim/swim_io.h b/src/lib/swim/swim_io.h
new file mode 100644
index 000000000..68fb89818
--- /dev/null
+++ b/src/lib/swim/swim_io.h
@@ -0,0 +1,225 @@
+#ifndef TARANTOOL_SWIM_IO_H_INCLUDED
+#define TARANTOOL_SWIM_IO_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "trivia/util.h"
+#include "small/rlist.h"
+#include "salad/stailq.h"
+#include "swim_transport.h"
+#include "tarantool_ev.h"
+#include <stdbool.h>
+#include <arpa/inet.h>
+
+/**
+ * SWIM protocol transport level.
+ */
+
+struct swim_task;
+struct swim_scheduler;
+
+enum {
+ /**
+ * Default MTU is 1500. MTU (when IPv4 is used) consists
+ * of IPv4 header, UDP header, Data. IPv4 has 20 bytes
+ * header, UDP - 8 bytes. So Data = 1500 - 20 - 8 = 1472.
+ * TODO: adapt to other MTUs which can be reduced in some
+ * networks by their admins. Or allow to specify MTU in
+ * configuration.
+ */
+ UDP_PACKET_SIZE = 1472,
+};
+
+/**
+ * UDP packet. Works as an allocator, allowing to fill its body
+ * gradually, while preserving prefix for metadata.
+ *
+ * < - - - -UDP_PACKET_SIZE- - - - ->
+ * +--------+-----------------------+
+ * | meta | body | *free* |
+ * +--------+-----------------------+
+ * ^ ^ ^ ^
+ * meta body pos end
+ * buf
+ */
+struct swim_packet {
+ /** End of the body. */
+ char *pos;
+ /**
+ * Starting position of body in the buffer. Not the same
+ * as buf, because the latter has metadata at the
+ * beginning.
+ */
+ char *body;
+ /**
+ * Alias for swim_packet.buf. Just sugar for code working
+ * with meta.
+ */
+ char meta[0];
+ /** Packet body buffer. */
+ char buf[UDP_PACKET_SIZE];
+ /**
+ * Pointer to the end of the buffer. Just sugar to do not
+ * write 'buf + sizeof(buf)' each time.
+ */
+ char end[0];
+};
+
+/**
+ * Ensure that the packet can fit @a size bytes more. Multiple
+ * reserves of the same size will return the same pointer until
+ * advance is called.
+ */
+static inline char *
+swim_packet_reserve(struct swim_packet *packet, int size)
+{
+ return packet->pos + size > packet->end ? NULL : packet->pos;
+}
+
+/**
+ * Propagate body end pointer. This declares next @a size bytes as
+ * occupied.
+ */
+static inline void
+swim_packet_advance(struct swim_packet *packet, int size)
+{
+ assert(packet->pos + size <= packet->end);
+ packet->pos += size;
+}
+
+/** Reserve + advance. */
+static inline char *
+swim_packet_alloc(struct swim_packet *packet, int size)
+{
+ char *res = swim_packet_reserve(packet, size);
+ if (res == NULL)
+ return NULL;
+ swim_packet_advance(packet, size);
+ return res;
+}
+
+/** Initialize @a packet, reserve some space for meta. */
+void
+swim_packet_create(struct swim_packet *packet);
+
+typedef void (*swim_scheduler_on_input_f)(struct swim_scheduler *scheduler,
+ const char *buf, const char *end,
+ const struct sockaddr_in *src);
+
+/** Planner and executor of input and output operations.*/
+struct swim_scheduler {
+ /** Transport to send/receive packets. */
+ struct swim_transport transport;
+ /**
+ * Function called when a packet is received. It takes
+ * packet body, while meta is handled by transport level
+ * completely.
+ */
+ swim_scheduler_on_input_f on_input;
+ /**
+ * Event dispatcher of incomming messages. Takes them from
+ * the network.
+ */
+ struct ev_io input;
+ /**
+ * Event dispatcher of outcomming messages. Takes tasks
+ * from queue_output.
+ */
+ struct ev_io output;
+ /** Queue of output tasks ready to write now. */
+ struct rlist queue_output;
+};
+
+/** Initialize scheduler. */
+void
+swim_scheduler_create(struct swim_scheduler *scheduler,
+ swim_scheduler_on_input_f on_input);
+
+/**
+ * Bind or rebind the scheduler to an address. In case of rebind
+ * the old socket is closed.
+ */
+int
+swim_scheduler_bind(struct swim_scheduler *scheduler,
+ const struct sockaddr_in *addr);
+
+/** Destroy scheduler, its queues, close the socket. */
+void
+swim_scheduler_destroy(struct swim_scheduler *scheduler);
+
+/**
+ * Each SWIM component in a common case independently may want to
+ * push some data into the network. Dissemination sends events,
+ * failure detection sends pings, acks. Anti-entropy sends member
+ * tables. The intention to send a data is called IO task and is
+ * stored in a queue that is dispatched when output is possible.
+ */
+typedef void (*swim_task_f)(struct swim_task *,
+ struct swim_scheduler *scheduler, int rc);
+
+struct swim_task {
+ /**
+ * Function called when the task has completed. Error code
+ * or 0 are passed as an argument.
+ */
+ swim_task_f complete;
+ /**
+ * Function, called when a scheduler is under destruction,
+ * and it cancels all its tasks.
+ */
+ swim_task_f cancel;
+ /** Packet to send. */
+ struct swim_packet packet;
+ /** Destination address. */
+ struct sockaddr_in dst;
+ /** Place in a queue of tasks. */
+ struct rlist in_queue_output;
+};
+
+/**
+ * Put the task into a queue of tasks. Eventually it will be sent.
+ */
+void
+swim_task_send(struct swim_task *task, const struct sockaddr_in *dst,
+ struct swim_scheduler *scheduler);
+
+/** Initialize the task, without scheduling. */
+void
+swim_task_create(struct swim_task *task, swim_task_f complete,
+ swim_task_f cancel);
+
+/** Destroy the task, pop from the queue. */
+static inline void
+swim_task_destroy(struct swim_task *task)
+{
+ rlist_del_entry(task, in_queue_output);
+}
+
+#endif /* TARANTOOL_SWIM_IO_H_INCLUDED */
\ No newline at end of file
diff --git a/src/lib/swim/swim_proto.c b/src/lib/swim/swim_proto.c
new file mode 100644
index 000000000..aba050eae
--- /dev/null
+++ b/src/lib/swim/swim_proto.c
@@ -0,0 +1,331 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "swim_proto.h"
+#include "msgpuck.h"
+#include "say.h"
+#include "version.h"
+#include "diag.h"
+
+const char *swim_member_status_strs[] = {
+ "alive",
+};
+
+int
+swim_decode_map(const char **pos, const char *end, uint32_t *size,
+ const char *prefix, const char *param_name)
+{
+ if (mp_typeof(**pos) != MP_MAP || *pos == end ||
+ mp_check_map(*pos, end) > 0) {
+ diag_set(SwimError, "%s %s should be a map", prefix,
+ param_name);
+ return -1;
+ }
+ *size = mp_decode_map(pos);
+ return 0;
+}
+
+int
+swim_decode_array(const char **pos, const char *end, uint32_t *size,
+ const char *prefix, const char *param_name)
+{
+ if (mp_typeof(**pos) != MP_ARRAY || *pos == end ||
+ mp_check_array(*pos, end) > 0) {
+ diag_set(SwimError, "%s %s should be an array", prefix,
+ param_name);
+ return -1;
+ }
+ *size = mp_decode_array(pos);
+ return 0;
+}
+
+int
+swim_decode_uint(const char **pos, const char *end, uint64_t *value,
+ const char *prefix, const char *param_name)
+{
+ if (mp_typeof(**pos) != MP_UINT || *pos == end ||
+ mp_check_uint(*pos, end) > 0) {
+ diag_set(SwimError, "%s %s should be a uint", prefix,
+ param_name);
+ return -1;
+ }
+ *value = mp_decode_uint(pos);
+ return 0;
+}
+
+static inline int
+swim_decode_ip(struct sockaddr_in *address, const char **pos, const char *end,
+ const char *prefix, const char *param_name)
+{
+ uint64_t ip;
+ if (swim_decode_uint(pos, end, &ip, prefix, param_name) != 0)
+ return -1;
+ if (ip > UINT32_MAX) {
+ diag_set(SwimError, "%s %s is an invalid IP address", prefix,
+ param_name);
+ return -1;
+ }
+ address->sin_addr.s_addr = ip;
+ return 0;
+}
+
+static inline int
+swim_decode_port(struct sockaddr_in *address, const char **pos, const char *end,
+ const char *prefix, const char *param_name)
+{
+ uint64_t port;
+ if (swim_decode_uint(pos, end, &port, prefix, param_name) != 0)
+ return -1;
+ if (port > UINT16_MAX) {
+ diag_set(SwimError, "%s %s is an invalid port", prefix,
+ param_name);
+ return -1;
+ }
+ address->sin_port = port;
+ return 0;
+}
+
+int
+swim_decode_uuid(struct tt_uuid *uuid, const char **pos, const char *end,
+ const char *prefix, const char *param_name)
+{
+ if (mp_typeof(**pos) != MP_BIN || *pos == end ||
+ mp_check_binl(*pos, end) > 0) {
+ diag_set(SwimError, "%s %s should be bin", prefix,
+ param_name);
+ return -1;
+ }
+ if (mp_decode_binl(pos) != UUID_LEN || *pos + UUID_LEN > end) {
+ diag_set(SwimError, "%s %s is invalid", prefix, param_name);
+ return -1;
+ }
+ memcpy(uuid, *pos, UUID_LEN);
+ *pos += UUID_LEN;
+ return 0;
+}
+
+void
+swim_member_def_create(struct swim_member_def *def)
+{
+ memset(def, 0, sizeof(*def));
+ def->addr.sin_family = AF_INET;
+ def->status = MEMBER_ALIVE;
+}
+
+/**
+ * Decode a MessagePack value of @a key and store it in @a def.
+ * @param key Key to read value of.
+ * @param[in][out] pos Where a value is stored.
+ * @param end End of the buffer.
+ * @param prefix Error message prefix.
+ * @param[out] def Where to store the value.
+ *
+ * @retval 0 Success.
+ * @retval -1 Error.
+ */
+static int
+swim_decode_member_key(enum swim_member_key key, const char **pos,
+ const char *end, const char *prefix,
+ struct swim_member_def *def)
+{
+ uint64_t tmp;
+ switch (key) {
+ case SWIM_MEMBER_STATUS:
+ if (swim_decode_uint(pos, end, &tmp, prefix,
+ "member status") != 0)
+ return -1;
+ if (tmp >= swim_member_status_MAX) {
+ diag_set(SwimError, "%s unknown member status", prefix);
+ return -1;
+ }
+ def->status = (enum swim_member_status) tmp;
+ break;
+ case SWIM_MEMBER_ADDRESS:
+ if (swim_decode_ip(&def->addr, pos, end, prefix,
+ "member address") != 0)
+ return -1;
+ break;
+ case SWIM_MEMBER_PORT:
+ if (swim_decode_port(&def->addr, pos, end, prefix,
+ "member port") != 0)
+ return -1;
+ break;
+ case SWIM_MEMBER_UUID:
+ if (swim_decode_uuid(&def->uuid, pos, end, prefix,
+ "member uuid") != 0)
+ return -1;
+ break;
+ default:
+ unreachable();
+ }
+ return 0;
+}
+
+int
+swim_member_def_decode(struct swim_member_def *def, const char **pos,
+ const char *end, const char *prefix)
+{
+ uint32_t size;
+ if (swim_decode_map(pos, end, &size, prefix, "member") != 0)
+ return -1;
+ swim_member_def_create(def);
+ for (uint32_t j = 0; j < size; ++j) {
+ uint64_t key;
+ if (swim_decode_uint(pos, end, &key, prefix,
+ "member key") != 0)
+ return -1;
+ if (key >= swim_member_key_MAX) {
+ diag_set(SwimError, "%s unknown member key", prefix);
+ return -1;
+ }
+ if (swim_decode_member_key(key, pos, end, prefix, def) != 0)
+ return -1;
+ }
+ if (def->addr.sin_port == 0 || def->addr.sin_addr.s_addr == 0) {
+ diag_set(SwimError, "%s member address is mandatory", prefix);
+ return -1;
+ }
+ if (tt_uuid_is_nil(&def->uuid)) {
+ diag_set(SwimError, "%s member uuid is mandatory", prefix);
+ return -1;
+ }
+ return 0;
+}
+
+void
+swim_src_uuid_bin_create(struct swim_src_uuid_bin *header,
+ const struct tt_uuid *uuid)
+{
+ header->k_uuid = SWIM_SRC_UUID;
+ header->m_uuid = 0xc4;
+ header->m_uuid_len = UUID_LEN;
+ memcpy(header->v_uuid, uuid, UUID_LEN);
+}
+
+void
+swim_anti_entropy_header_bin_create(struct swim_anti_entropy_header_bin *header,
+ uint16_t batch_size)
+{
+ header->k_anti_entropy = SWIM_ANTI_ENTROPY;
+ header->m_anti_entropy = 0xdc;
+ header->v_anti_entropy = mp_bswap_u16(batch_size);
+}
+
+void
+swim_member_bin_fill(struct swim_member_bin *header,
+ const struct sockaddr_in *addr, const struct tt_uuid *uuid,
+ enum swim_member_status status)
+{
+ header->v_status = status;
+ header->v_addr = mp_bswap_u32(addr->sin_addr.s_addr);
+ header->v_port = mp_bswap_u16(addr->sin_port);
+ memcpy(header->v_uuid, uuid, UUID_LEN);
+}
+
+void
+swim_member_bin_create(struct swim_member_bin *header)
+{
+ header->m_header = 0x84;
+ header->k_status = SWIM_MEMBER_STATUS;
+ header->k_addr = SWIM_MEMBER_ADDRESS;
+ header->m_addr = 0xce;
+ header->k_port = SWIM_MEMBER_PORT;
+ header->m_port = 0xcd;
+ header->k_uuid = SWIM_MEMBER_UUID;
+ header->m_uuid = 0xc4;
+ header->m_uuid_len = UUID_LEN;
+}
+
+void
+swim_meta_header_bin_create(struct swim_meta_header_bin *header,
+ const struct sockaddr_in *src)
+{
+ header->m_header = 0x83;
+ header->k_version = SWIM_META_TARANTOOL_VERSION;
+ header->m_version = 0xce;
+ header->v_version = mp_bswap_u32(tarantool_version_id());
+ header->k_addr = SWIM_META_SRC_ADDRESS;
+ header->m_addr = 0xce;
+ header->v_addr = mp_bswap_u32(src->sin_addr.s_addr);
+ header->k_port = SWIM_META_SRC_PORT;
+ header->m_port = 0xcd;
+ header->v_port = mp_bswap_u16(src->sin_port);
+}
+
+int
+swim_meta_def_decode(struct swim_meta_def *def, const char **pos,
+ const char *end)
+{
+ const char *prefix = "invalid meta section:";
+ uint32_t size;
+ if (swim_decode_map(pos, end, &size, prefix, "root") != 0)
+ return -1;
+ memset(def, 0, sizeof(*def));
+ for (uint32_t i = 0; i < size; ++i) {
+ uint64_t key;
+ if (swim_decode_uint(pos, end, &key, prefix, "a key") != 0)
+ return -1;
+ switch (key) {
+ case SWIM_META_TARANTOOL_VERSION:
+ if (swim_decode_uint(pos, end, &key, prefix,
+ "version") != 0)
+ return -1;
+ if (key > UINT32_MAX) {
+ diag_set(SwimError, "%s invalid version, too "\
+ "big", prefix);
+ return -1;
+ }
+ def->version = key;
+ break;
+ case SWIM_META_SRC_ADDRESS:
+ if (swim_decode_ip(&def->src, pos, end, prefix,
+ "source address") != 0)
+ return -1;
+ break;
+ case SWIM_META_SRC_PORT:
+ if (swim_decode_port(&def->src, pos, end, prefix,
+ "source port") != 0)
+ return -1;
+ break;
+ default:
+ diag_set(SwimError, "%s unknown key", prefix);
+ return -1;
+ }
+ }
+ if (def->version == 0) {
+ diag_set(SwimError, "%s version is mandatory", prefix);
+ return -1;
+ }
+ if (def->src.sin_port == 0 || def->src.sin_addr.s_addr == 0) {
+ diag_set(SwimError, "%s source address is mandatory", prefix);
+ return -1;
+ }
+ return 0;
+}
diff --git a/src/lib/swim/swim_proto.h b/src/lib/swim/swim_proto.h
new file mode 100644
index 000000000..300a08c1f
--- /dev/null
+++ b/src/lib/swim/swim_proto.h
@@ -0,0 +1,320 @@
+#ifndef TARANTOOL_SWIM_PROTO_H_INCLUDED
+#define TARANTOOL_SWIM_PROTO_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "trivia/util.h"
+#include "uuid/tt_uuid.h"
+#include <arpa/inet.h>
+#include <stdbool.h>
+
+/**
+ * SWIM binary protocol structures and helpers. Below is a picture
+ * of a SWIM message template:
+ *
+ * +----------Meta section, handled by transport level-----------+
+ * | { |
+ * | SWIM_META_TARANTOOL_VERSION: uint, Tarantool version ID,|
+ * | SWIM_META_SRC_ADDRESS: uint, ip, |
+ * | SWIM_META_SRC_PORT: uint, port |
+ * | } |
+ * +-------------------Protocol logic section--------------------+
+ * | { |
+ * | SWIM_SRC_UUID: 16 byte UUID, |
+ * | |
+ * | AND |
+ * | |
+ * | SWIM_ANTI_ENTROPY: [ |
+ * | { |
+ * | SWIM_MEMBER_STATUS: uint, enum member_status, |
+ * | SWIM_MEMBER_ADDRESS: uint, ip, |
+ * | SWIM_MEMBER_PORT: uint, port, |
+ * | SWIM_MEMBER_UUID: 16 byte UUID |
+ * | }, |
+ * | ... |
+ * | ], |
+ * | } |
+ * +-------------------------------------------------------------+
+ */
+
+enum swim_member_status {
+ /** The instance is ok, responds to requests. */
+ MEMBER_ALIVE = 0,
+ swim_member_status_MAX,
+};
+
+extern const char *swim_member_status_strs[];
+
+/**
+ * SWIM member attributes from anti-entropy and dissemination
+ * messages.
+ */
+struct swim_member_def {
+ struct tt_uuid uuid;
+ struct sockaddr_in addr;
+ enum swim_member_status status;
+};
+
+/** Initialize the definition with default values. */
+void
+swim_member_def_create(struct swim_member_def *def);
+
+/**
+ * Decode member definition from a MessagePack buffer.
+ * @param[out] def Definition to decode into.
+ * @param[in][out] pos Start of the MessagePack buffer.
+ * @param end End of the MessagePack buffer.
+ * @param prefix A prefix of an error message to use for
+ * diag_set, when something is wrong.
+ *
+ * @retval 0 Success.
+ * @retval -1 Error.
+ */
+int
+swim_member_def_decode(struct swim_member_def *def, const char **pos,
+ const char *end, const char *prefix);
+
+/**
+ * Main round messages can carry merged failure detection,
+ * anti-entropy, dissemination messages. With these keys the
+ * components can be distinguished from each other.
+ */
+enum swim_body_key {
+ SWIM_SRC_UUID = 0,
+ SWIM_ANTI_ENTROPY,
+};
+
+/**
+ * One of SWIM packet body components - SWIM_SRC_UUID. It is not
+ * in the meta section, handled by the transport, because the
+ * transport has nothing to do with UUIDs - it operates by IP/port
+ * only. This component shall be first in message's body.
+ */
+struct PACKED swim_src_uuid_bin {
+ /** mp_encode_uint(SWIM_SRC_UUID) */
+ uint8_t k_uuid;
+ /** mp_encode_bin(UUID_LEN) */
+ uint8_t m_uuid;
+ uint8_t m_uuid_len;
+ uint8_t v_uuid[UUID_LEN];
+};
+
+/** Initialize source UUID section. */
+void
+swim_src_uuid_bin_create(struct swim_src_uuid_bin *header,
+ const struct tt_uuid *uuid);
+
+/** {{{ Anti-entropy component */
+
+/**
+ * Attributes of each record of a broadcasted member table. Just
+ * the same as some of struct swim_member attributes.
+ */
+enum swim_member_key {
+ SWIM_MEMBER_STATUS = 0,
+ SWIM_MEMBER_ADDRESS,
+ SWIM_MEMBER_PORT,
+ SWIM_MEMBER_UUID,
+ swim_member_key_MAX,
+};
+
+/** SWIM anti-entropy MessagePack header template. */
+struct PACKED swim_anti_entropy_header_bin {
+ /** mp_encode_uint(SWIM_ANTI_ENTROPY) */
+ uint8_t k_anti_entropy;
+ /** mp_encode_array(...) */
+ uint8_t m_anti_entropy;
+ uint16_t v_anti_entropy;
+};
+
+/** Initialize SWIM_ANTI_ENTROPY header. */
+void
+swim_anti_entropy_header_bin_create(struct swim_anti_entropy_header_bin *header,
+ uint16_t batch_size);
+
+/**
+ * SWIM member MessagePack template. Represents one record in
+ * anti-entropy section.
+ */
+struct PACKED swim_member_bin {
+ /** mp_encode_map(4) */
+ uint8_t m_header;
+
+ /** mp_encode_uint(SWIM_MEMBER_STATUS) */
+ uint8_t k_status;
+ /** mp_encode_uint(enum member_status) */
+ uint8_t v_status;
+
+ /** mp_encode_uint(SWIM_MEMBER_ADDRESS) */
+ uint8_t k_addr;
+ /** mp_encode_uint(addr.sin_addr.s_addr) */
+ uint8_t m_addr;
+ uint32_t v_addr;
+
+ /** mp_encode_uint(SWIM_MEMBER_PORT) */
+ uint8_t k_port;
+ /** mp_encode_uint(addr.sin_port) */
+ uint8_t m_port;
+ uint16_t v_port;
+
+ /** mp_encode_uint(SWIM_MEMBER_UUID) */
+ uint8_t k_uuid;
+ /** mp_encode_bin(UUID_LEN) */
+ uint8_t m_uuid;
+ uint8_t m_uuid_len;
+ uint8_t v_uuid[UUID_LEN];
+};
+
+/** Initialize antri-entropy record. */
+void
+swim_member_bin_create(struct swim_member_bin *header);
+
+/**
+ * Since usually there are many members, it is faster to reset a
+ * few fields in an existing template, then each time create a
+ * new template. So the usage pattern is create(), fill(),
+ * fill() ... .
+ */
+void
+swim_member_bin_fill(struct swim_member_bin *header,
+ const struct sockaddr_in *addr, const struct tt_uuid *uuid,
+ enum swim_member_status status);
+
+/** }}} Anti-entropy component */
+
+/** {{{ Meta component */
+
+/**
+ * Meta component keys, completely handled by the transport level.
+ */
+enum swim_meta_key {
+ /**
+ * Version is now unused, but in future can help in
+ * the protocol improvement, extension.
+ */
+ SWIM_META_TARANTOOL_VERSION = 0,
+ /**
+ * Source IP/port are stored in body of UDP packet despite
+ * the fact that UDP has them in its header. This is
+ * because
+ * - packet body is going to be encrypted, but header
+ * is still open and anybody can catch the packet,
+ * change source IP/port, and therefore execute
+ * man-in-the-middle attack;
+ *
+ * - some network filters can change the address to an
+ * address of a router or another device.
+ */
+ SWIM_META_SRC_ADDRESS,
+ SWIM_META_SRC_PORT,
+};
+
+/**
+ * Each SWIM packet carries meta info, which helps to determine
+ * SWIM protocol version, final packet destination and any other
+ * internal details, not linked with etalon SWIM protocol.
+ *
+ * The meta header is mandatory, preceeds main protocol data as a
+ * separate MessagePack map.
+ */
+struct PACKED swim_meta_header_bin {
+ /** mp_encode_map(3) */
+ uint8_t m_header;
+
+ /** mp_encode_uint(SWIM_META_TARANTOOL_VERSION) */
+ uint8_t k_version;
+ /** mp_encode_uint(tarantool_version_id()) */
+ uint8_t m_version;
+ uint32_t v_version;
+
+ /** mp_encode_uint(SWIM_META_SRC_ADDRESS) */
+ uint8_t k_addr;
+ /** mp_encode_uint(addr.sin_addr.s_addr) */
+ uint8_t m_addr;
+ uint32_t v_addr;
+
+ /** mp_encode_uint(SWIM_META_SRC_PORT) */
+ uint8_t k_port;
+ /** mp_encode_uint(addr.sin_port) */
+ uint8_t m_port;
+ uint16_t v_port;
+};
+
+/** Initialize meta section. */
+void
+swim_meta_header_bin_create(struct swim_meta_header_bin *header,
+ const struct sockaddr_in *src);
+
+/** Meta definition. */
+struct swim_meta_def {
+ /** Tarantool version. */
+ uint32_t version;
+ /** Source of the message. */
+ struct sockaddr_in src;
+};
+
+/**
+ * Decode meta section into its definition object.
+ * @param[out] def Definition to decode into.
+ * @param[in][out] pos MessagePack buffer to decode.
+ * @param end End of the MessagePack buffer.
+ *
+ * @retval 0 Success.
+ * @retval -1 Error.
+ */
+int
+swim_meta_def_decode(struct swim_meta_def *def, const char **pos,
+ const char *end);
+
+/** }}} Meta component */
+
+/**
+ * Helpers to decode some values - map, array, etc with
+ * appropriate checks. All of them set diagnostics on an error
+ * with a specified message prefix and a parameter name.
+ */
+
+int
+swim_decode_map(const char **pos, const char *end, uint32_t *size,
+ const char *prefix, const char *param_name);
+
+int
+swim_decode_array(const char **pos, const char *end, uint32_t *size,
+ const char *prefix, const char *param_name);
+
+int
+swim_decode_uint(const char **pos, const char *end, uint64_t *value,
+ const char *prefix, const char *param_name);
+
+int
+swim_decode_uuid(struct tt_uuid *uuid, const char **pos, const char *end,
+ const char *prefix, const char *param_name);
+
+#endif /* TARANTOOL_SWIM_PROTO_H_INCLUDED */
diff --git a/src/lib/swim/swim_transport.h b/src/lib/swim/swim_transport.h
new file mode 100644
index 000000000..e19dbb6a9
--- /dev/null
+++ b/src/lib/swim/swim_transport.h
@@ -0,0 +1,79 @@
+#ifndef TARANTOOL_SWIM_TRANSPORT_H_INCLUDED
+#define TARANTOOL_SWIM_TRANSPORT_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "trivia/util.h"
+#include <arpa/inet.h>
+
+/** Transport implementation. */
+struct swim_transport {
+ /** Socket. */
+ int fd;
+ /** Socket address. */
+ struct sockaddr_in addr;
+};
+
+/**
+ * Despite there are no transport vtab, those are virtual methods.
+ * But virtualization is handled on compilation time. This header
+ * file has one implementation for server, and another for tests.
+ * Transport source is built as a separate library.
+ *
+ * Methods below for server mostly are just wrappers of
+ * corresponding system calls, working with UDP sockets.
+ */
+
+ssize_t
+swim_transport_send(struct swim_transport *transport, const void *data,
+ size_t size, const struct sockaddr *addr,
+ socklen_t addr_size);
+
+ssize_t
+swim_transport_recv(struct swim_transport *transport, void *buffer, size_t size,
+ struct sockaddr *addr, socklen_t *addr_size);
+
+/**
+ * Bind @a transport to a new address. The old socket, if exists,
+ * is closed. If @a addr is from INET family and has 0 port, then
+ * @a transport will save not 0 port, but a real one, got after
+ * bind() using getsockname().
+ */
+int
+swim_transport_bind(struct swim_transport *transport,
+ const struct sockaddr *addr, socklen_t addr_len);
+
+void
+swim_transport_destroy(struct swim_transport *transport);
+
+void
+swim_transport_create(struct swim_transport *transport);
+
+#endif /* TARANTOOL_SWIM_TRANSPORT_H_INCLUDED */
diff --git a/src/lib/swim/swim_transport_udp.c b/src/lib/swim/swim_transport_udp.c
new file mode 100644
index 000000000..f8fbae102
--- /dev/null
+++ b/src/lib/swim/swim_transport_udp.c
@@ -0,0 +1,112 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "swim_transport.h"
+#include "evio.h"
+#include "diag.h"
+
+ssize_t
+swim_transport_send(struct swim_transport *transport, const void *data,
+ size_t size, const struct sockaddr *addr,
+ socklen_t addr_size)
+{
+ ssize_t ret = sio_sendto(transport->fd, data, size, 0, addr, addr_size);
+ if (ret == -1 && sio_wouldblock(errno))
+ return 0;
+ return ret;
+}
+
+ssize_t
+swim_transport_recv(struct swim_transport *transport, void *buffer, size_t size,
+ struct sockaddr *addr, socklen_t *addr_size)
+{
+ ssize_t ret = sio_recvfrom(transport->fd, buffer, size, 0, addr,
+ addr_size);
+ if (ret == -1 && sio_wouldblock(errno))
+ return 0;
+ return ret;
+}
+
+int
+swim_transport_bind(struct swim_transport *transport,
+ const struct sockaddr *addr, socklen_t addr_len)
+{
+ assert(addr->sa_family == AF_INET);
+ const struct sockaddr_in *new_addr = (const struct sockaddr_in *) addr;
+ const struct sockaddr_in *old_addr = &transport->addr;
+ assert(addr_len == sizeof(*new_addr));
+
+ if (transport->fd != -1 &&
+ new_addr->sin_addr.s_addr == old_addr->sin_addr.s_addr &&
+ new_addr->sin_port == old_addr->sin_port)
+ return 0;
+
+ int fd = sio_socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+ if (fd < 0)
+ return -1;
+ if (sio_bind(fd, (struct sockaddr *) addr, addr_len) != 0 ||
+ evio_setsockopt_server(fd, AF_INET, SOCK_DGRAM) != 0) {
+ if (errno == EADDRINUSE)
+ diag_set(SocketError, sio_socketname(fd), "bind");
+ close(fd);
+ return -1;
+ }
+ int real_port = new_addr->sin_port;
+ if (new_addr->sin_port == 0) {
+ struct sockaddr_in real_addr;
+ addr_len = sizeof(real_addr);
+ if (sio_getsockname(fd, (struct sockaddr *) &real_addr,
+ &addr_len) != 0) {
+ close(fd);
+ return -1;
+ }
+ real_port = real_addr.sin_port;
+ }
+ if (transport->fd != -1)
+ close(transport->fd);
+ transport->fd = fd;
+ transport->addr = *new_addr;
+ transport->addr.sin_port = real_port;
+ return 0;
+}
+
+void
+swim_transport_destroy(struct swim_transport *transport)
+{
+ if (transport->fd != -1)
+ close(transport->fd);
+}
+
+void
+swim_transport_create(struct swim_transport *transport)
+{
+ transport->fd = -1;
+ memset(&transport->addr, 0, sizeof(transport->addr));
+}
diff --git a/test/unit/CMakeLists.txt b/test/unit/CMakeLists.txt
index c13528400..6d5704349 100644
--- a/test/unit/CMakeLists.txt
+++ b/test/unit/CMakeLists.txt
@@ -208,3 +208,11 @@ target_link_libraries(checkpoint_schedule.test m unit)
add_executable(sio.test sio.c)
target_link_libraries(sio.test unit core)
+
+add_executable(swim.test swim.c swim_test_transport.c swim_test_ev.c
+ swim_test_utils.c ${PROJECT_SOURCE_DIR}/src/version.c)
+target_link_libraries(swim.test unit swim)
+
+add_executable(swim_proto.test swim_proto.c swim_test_transport.c swim_test_ev.c
+ swim_test_utils.c ${PROJECT_SOURCE_DIR}/src/version.c)
+target_link_libraries(swim_proto.test unit swim)
diff --git a/test/unit/swim.c b/test/unit/swim.c
new file mode 100644
index 000000000..78a69b7c6
--- /dev/null
+++ b/test/unit/swim.c
@@ -0,0 +1,243 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+
+#include "memory.h"
+#include "fiber.h"
+#include "uuid/tt_uuid.h"
+#include "unit.h"
+#include "swim/swim.h"
+#include "swim_test_transport.h"
+#include "swim_test_ev.h"
+#include "swim_test_utils.h"
+
+/**
+ * Test result is a real returned value of main_f. Fiber_join can
+ * not be used, because it expects if a returned value < 0 then
+ * diag is not empty. But in unit tests it can be violated -
+ * check_plan() does not set diag.
+ */
+static int test_result;
+
+static void
+swim_test_one_link(void)
+{
+ swim_start_test(1);
+ /*
+ * Run a simple cluster of two elements. One of them
+ * learns about another explicitly. Another should add the
+ * former into his table of members.
+ */
+ struct swim_cluster *cluster = swim_cluster_new(2);
+ fail_if(swim_cluster_add_link(cluster, 0, 1) != 0);
+ is(swim_cluster_wait_fullmesh(cluster, 1), 0, "one link");
+ swim_cluster_delete(cluster);
+
+ swim_finish_test();
+}
+
+static void
+swim_test_sequence(void)
+{
+ swim_start_test(1);
+ /*
+ * Run a simple cluster of several elements. Build a
+ * 'forward list' from them. It should turn into fullmesh
+ * in O(N) time. Time is not fixed because of randomness,
+ * so here just in case 2N is used - it should be enough.
+ */
+ struct swim_cluster *cluster = swim_cluster_new(5);
+ for (int i = 0; i < 4; ++i)
+ swim_cluster_add_link(cluster, i, i + 1);
+ is(swim_cluster_wait_fullmesh(cluster, 10), 0, "sequence");
+ swim_cluster_delete(cluster);
+
+ swim_finish_test();
+}
+
+static void
+swim_test_uuid_update(void)
+{
+ swim_start_test(4);
+
+ struct swim_cluster *cluster = swim_cluster_new(2);
+ swim_cluster_add_link(cluster, 0, 1);
+ fail_if(swim_cluster_wait_fullmesh(cluster, 1) != 0);
+ struct swim *s = swim_cluster_node(cluster, 0);
+ struct tt_uuid new_uuid = uuid_nil;
+ new_uuid.time_low = 1000;
+ is(swim_cfg(s, NULL, -1, &new_uuid), 0, "UUID update");
+ is(swim_cluster_wait_fullmesh(cluster, 1), 0,
+ "old UUID is returned back as a 'ghost' member");
+ new_uuid.time_low = 2;
+ is(swim_cfg(s, NULL, -1, &new_uuid), -1,
+ "can not update to an existing UUID - swim_cfg fails");
+ ok(swim_error_check_match("exists"), "diag says 'exists'");
+ swim_cluster_delete(cluster);
+
+ swim_finish_test();
+}
+
+static void
+swim_test_cfg(void)
+{
+ swim_start_test(15);
+
+ struct swim *s = swim_new();
+ assert(s != NULL);
+ is(swim_cfg(s, NULL, -1, NULL), -1, "first cfg failed - no URI");
+ ok(swim_error_check_match("mandatory"), "diag says 'mandatory'");
+ const char *uri = "127.0.0.1:1";
+ is(swim_cfg(s, uri, -1, NULL), -1, "first cfg failed - no UUID");
+ ok(swim_error_check_match("mandatory"), "diag says 'mandatory'");
+ struct tt_uuid uuid = uuid_nil;
+ uuid.time_low = 1;
+ is(swim_cfg(s, uri, -1, &uuid), 0, "configured first time");
+ is(swim_cfg(s, NULL, -1, NULL), 0, "second time can omit URI, UUID");
+ is(swim_cfg(s, NULL, 2, NULL), 0, "hearbeat is dynamic");
+
+ struct swim *s2 = swim_new();
+ assert(s2 != NULL);
+ const char *bad_uri1 = "127.1.1.1.1.1.1:1";
+ const char *bad_uri2 = "google.com:1";
+ const char *bad_uri3 = "unix/:/home/gerold103/any/dir";
+ struct tt_uuid uuid2 = uuid_nil;
+ uuid2.time_low = 2;
+ is(swim_cfg(s2, bad_uri1, -1, &uuid2), -1, "can not use invalid URI");
+ ok(swim_error_check_match("invalid uri"), "diag says 'invalid uri'");
+ is(swim_cfg(s2, bad_uri2, -1, &uuid2), -1, "can not use domain names");
+ ok(swim_error_check_match("invalid uri"), "diag says 'invalid uri'");
+ is(swim_cfg(s2, bad_uri3, -1, &uuid2), -1,
+ "UNIX sockets are not supported");
+ ok(swim_error_check_match("only IP"), "diag says 'only IP'");
+ is(swim_cfg(s2, uri, -1, &uuid2), -1,
+ "can not bind to an occupied port");
+ ok(swim_error_check_match("bind"), "diag says 'bind'");
+ swim_delete(s2);
+ swim_delete(s);
+
+ swim_finish_test();
+}
+
+static void
+swim_test_add_remove(void)
+{
+ swim_start_test(12);
+
+ struct swim_cluster *cluster = swim_cluster_new(2);
+ swim_cluster_add_link(cluster, 0, 1);
+ fail_if(swim_cluster_wait_fullmesh(cluster, 1) != 0);
+ struct swim *s1 = swim_cluster_node(cluster, 0);
+ struct swim *s2 = swim_cluster_node(cluster, 1);
+ const struct swim_member *s2_self = swim_self(s2);
+
+ is(swim_add_member(s1, swim_member_uri(s2_self),
+ swim_member_uuid(s2_self)), -1,
+ "can not add an existing member");
+ ok(swim_error_check_match("already exists"),
+ "diag says 'already exists'");
+
+ const char *bad_uri = "127.0.0101010101";
+ struct tt_uuid uuid = uuid_nil;
+ uuid.time_low = 1000;
+ is(swim_add_member(s1, bad_uri, &uuid), -1,
+ "can not add a invalid uri");
+ ok(swim_error_check_match("invalid uri"), "diag says 'invalid uri'");
+
+ is(swim_remove_member(s2, swim_member_uuid(s2_self)), -1,
+ "can not remove self");
+ ok(swim_error_check_match("can not remove self"),
+ "diag says the same");
+
+ isnt(swim_member_by_uuid(s1, swim_member_uuid(s2_self)), NULL,
+ "find by UUID works");
+ is(swim_remove_member(s1, swim_member_uuid(s2_self)), 0,
+ "now remove one element");
+ is(swim_member_by_uuid(s1, swim_member_uuid(s2_self)), NULL,
+ "and it can not be found anymore");
+
+ is(swim_remove_member(s1, &uuid), 0, "remove of a not existing member");
+
+ is(swim_cluster_is_fullmesh(cluster), false,
+ "after removal the cluster is not in fullmesh");
+ is(swim_cluster_wait_fullmesh(cluster, 1), 0,
+ "but it is back in 1 step");
+ swim_cluster_delete(cluster);
+
+ swim_finish_test();
+}
+
+static int
+main_f(va_list ap)
+{
+ swim_start_test(5);
+
+ (void) ap;
+ struct ev_loop *loop = loop();
+ swim_test_ev_init();
+ swim_test_transport_init();
+
+ swim_test_one_link();
+ swim_test_sequence();
+ swim_test_uuid_update();
+ swim_test_cfg();
+ swim_test_add_remove();
+
+ swim_test_transport_free();
+ swim_test_ev_free();
+
+ test_result = check_plan();
+ footer();
+ return 0;
+}
+
+int
+main()
+{
+ memory_init();
+ fiber_init(fiber_c_invoke);
+ int fd = open("log.txt", O_TRUNC);
+ if (fd != -1)
+ close(fd);
+ say_logger_init("log.txt", 6, 1, "plain", 0);
+
+ struct fiber *main_fiber = fiber_new("main", main_f);
+ fiber_set_joinable(main_fiber, true);
+ assert(main_fiber != NULL);
+ fiber_wakeup(main_fiber);
+ ev_run(loop(), 0);
+ fiber_join(main_fiber);
+
+ say_logger_free();
+ fiber_free();
+ memory_free();
+
+ return test_result;
+}
\ No newline at end of file
diff --git a/test/unit/swim.result b/test/unit/swim.result
new file mode 100644
index 000000000..73e1504a3
--- /dev/null
+++ b/test/unit/swim.result
@@ -0,0 +1,56 @@
+ *** main_f ***
+1..5
+ *** swim_test_one_link ***
+ 1..1
+ ok 1 - one link
+ok 1 - subtests
+ *** swim_test_one_link: done ***
+ *** swim_test_sequence ***
+ 1..1
+ ok 1 - sequence
+ok 2 - subtests
+ *** swim_test_sequence: done ***
+ *** swim_test_uuid_update ***
+ 1..4
+ ok 1 - UUID update
+ ok 2 - old UUID is returned back as a 'ghost' member
+ ok 3 - can not update to an existing UUID - swim_cfg fails
+ ok 4 - diag says 'exists'
+ok 3 - subtests
+ *** swim_test_uuid_update: done ***
+ *** swim_test_cfg ***
+ 1..15
+ ok 1 - first cfg failed - no URI
+ ok 2 - diag says 'mandatory'
+ ok 3 - first cfg failed - no UUID
+ ok 4 - diag says 'mandatory'
+ ok 5 - configured first time
+ ok 6 - second time can omit URI, UUID
+ ok 7 - hearbeat is dynamic
+ ok 8 - can not use invalid URI
+ ok 9 - diag says 'invalid uri'
+ ok 10 - can not use domain names
+ ok 11 - diag says 'invalid uri'
+ ok 12 - UNIX sockets are not supported
+ ok 13 - diag says 'only IP'
+ ok 14 - can not bind to an occupied port
+ ok 15 - diag says 'bind'
+ok 4 - subtests
+ *** swim_test_cfg: done ***
+ *** swim_test_add_remove ***
+ 1..12
+ ok 1 - can not add an existing member
+ ok 2 - diag says 'already exists'
+ ok 3 - can not add a invalid uri
+ ok 4 - diag says 'invalid uri'
+ ok 5 - can not remove self
+ ok 6 - diag says the same
+ ok 7 - find by UUID works
+ ok 8 - now remove one element
+ ok 9 - and it can not be found anymore
+ ok 10 - remove of a not existing member
+ ok 11 - after removal the cluster is not in fullmesh
+ ok 12 - but it is back in 1 step
+ok 5 - subtests
+ *** swim_test_add_remove: done ***
+ *** main_f: done ***
diff --git a/test/unit/swim_proto.c b/test/unit/swim_proto.c
new file mode 100644
index 000000000..639f42b5d
--- /dev/null
+++ b/test/unit/swim_proto.c
@@ -0,0 +1,212 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+
+#include "memory.h"
+#include "fiber.h"
+#include "uuid/tt_uuid.h"
+#include "version.h"
+#include "msgpuck.h"
+#include "unit.h"
+#include "swim/swim_proto.h"
+#include "swim_test_utils.h"
+
+static char buffer[1024 * 1024];
+
+static void
+swim_test_member_def(void)
+{
+ header();
+ plan(12);
+
+ struct swim_member_def mdef;
+ const char *pos = buffer;
+ char *last_valid, *end = mp_encode_array(buffer, 10);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "not map header");
+
+ pos = buffer;
+ end = mp_encode_map(buffer, 4);
+ last_valid = end;
+ end = mp_encode_str(end, "str", 3);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "not uint member key");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, 10000);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "too big member key");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, SWIM_MEMBER_STATUS);
+ end = mp_encode_str(end, "str", 3);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "STATUS is not uint");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, SWIM_MEMBER_STATUS);
+ end = mp_encode_uint(end, 10000);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "invalid STATUS");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, SWIM_MEMBER_ADDRESS);
+ last_valid = end;
+ end = mp_encode_uint(end, (uint64_t)UINT32_MAX + 100);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "invalid address");
+
+ pos = buffer;
+ struct in_addr ipaddr;
+ fail_if(inet_aton("127.0.0.1", &ipaddr) == 0);
+ end = mp_encode_uint(last_valid, ipaddr.s_addr);
+ end = mp_encode_uint(end, SWIM_MEMBER_PORT);
+ last_valid = end;
+ end = mp_encode_uint(end, 100000);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1, "bad port");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, 1);
+ last_valid = end;
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "unexpected buffer end");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, SWIM_MEMBER_UUID);
+ last_valid = end;
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "unexpected buffer end");
+
+ pos = buffer;
+ end = mp_encode_bin(last_valid, (const char *) &uuid_nil,
+ sizeof(uuid_nil));
+ end = mp_encode_uint(end, SWIM_MEMBER_STATUS);
+ end = mp_encode_uint(end, 0);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "uuid is nil/undefined");
+
+ pos = buffer;
+ struct tt_uuid uuid = uuid_nil;
+ uuid.time_low = 1;
+ end = mp_encode_bin(last_valid, (const char *) &uuid, sizeof(uuid));
+ last_valid = end;
+ end = mp_encode_uint(end, SWIM_MEMBER_PORT);
+ end = mp_encode_uint(end, 0);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), -1,
+ "port is 0/undefined");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, SWIM_MEMBER_STATUS);
+ end = mp_encode_uint(end, 0);
+ is(swim_member_def_decode(&mdef, &pos, end, "test"), 0,
+ "normal member def");
+
+ check_plan();
+ footer();
+}
+
+static void
+swim_test_meta(void)
+{
+ header();
+ plan(8);
+
+ struct swim_meta_def mdef;
+ const char *pos = buffer;
+ char *last_valid, *end = mp_encode_array(buffer, 10);
+ is(swim_meta_def_decode(&mdef, &pos, end), -1, "not map header");
+
+ pos = buffer;
+ end = mp_encode_map(buffer, 3);
+ last_valid = end;
+ end = mp_encode_str(end, "str", 3);
+ is(swim_meta_def_decode(&mdef, &pos, end), -1, "not uint meta key");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, 10000);
+ is(swim_meta_def_decode(&mdef, &pos, end), -1, "unknown meta key");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, SWIM_META_TARANTOOL_VERSION);
+ last_valid = end;
+ is(swim_meta_def_decode(&mdef, &pos, end), -1, "unexpected end");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, (uint64_t)UINT32_MAX + 100);
+ is(swim_meta_def_decode(&mdef, &pos, end), -1, "invalid version");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, tarantool_version_id());
+ end = mp_encode_uint(end, SWIM_META_SRC_ADDRESS);
+ struct in_addr ipaddr;
+ fail_if(inet_aton("127.0.0.1", &ipaddr) == 0);
+ end = mp_encode_uint(end, ipaddr.s_addr);
+ last_valid = end;
+ end = mp_encode_uint(end, SWIM_META_SRC_PORT);
+ end = mp_encode_uint(end, 0);
+ is(swim_meta_def_decode(&mdef, &pos, end), -1, "port is 0/undefined");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, SWIM_META_TARANTOOL_VERSION);
+ end = mp_encode_uint(end, 0);
+ is(swim_meta_def_decode(&mdef, &pos, end), -1,
+ "version is 0/undefined");
+
+ pos = buffer;
+ end = mp_encode_uint(last_valid, SWIM_META_SRC_PORT);
+ end = mp_encode_uint(end, 1);
+ is(swim_meta_def_decode(&mdef, &pos, end), 0, "normal meta");
+
+ check_plan();
+ footer();
+}
+
+int
+main()
+{
+ header();
+ plan(2);
+ memory_init();
+ fiber_init(fiber_c_invoke);
+ int fd = open("log.txt", O_TRUNC);
+ if (fd != -1)
+ close(fd);
+ say_logger_init("log.txt", 6, 1, "plain", 0);
+
+ swim_test_member_def();
+ swim_test_meta();
+
+ say_logger_free();
+ fiber_free();
+ memory_free();
+ int rc = check_plan();
+ footer();
+ return rc;
+}
\ No newline at end of file
diff --git a/test/unit/swim_test_ev.c b/test/unit/swim_test_ev.c
new file mode 100644
index 000000000..b70a8b9c8
--- /dev/null
+++ b/test/unit/swim_test_ev.c
@@ -0,0 +1,252 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "swim_test_ev.h"
+#include "swim_test_transport.h"
+#include "trivia/util.h"
+#include "swim/swim_ev.h"
+#include "tarantool_ev.h"
+#define HEAP_FORWARD_DECLARATION
+#include "salad/heap.h"
+#include "assoc.h"
+#include "say.h"
+#include <stdbool.h>
+
+/** Global watch, propagated by new events. */
+static double watch = 0;
+
+/**
+ * Increasing event identifiers are used to preserve order of
+ * events with the same deadline.
+ */
+static int event_id = 0;
+
+/**
+ * Timed event's description and attributes used to process it.
+ * Note, that IO events are not wrapped with that and processed
+ * separately in test transport module.
+ */
+struct swim_test_event {
+ /**
+ * Event mask with which libev watcher's callback should
+ * be invoked on that event.
+ */
+ int revents;
+ /**
+ * Libev watcher. Used to store callback and to find the
+ * event by watcher pointer. It is necessary because SWIM
+ * operates by libev watchers.
+ */
+ struct ev_watcher *watcher;
+ /**
+ * When that event should be invoked according to the fake
+ * watch.
+ */
+ double deadline;
+ /** A link in the events heap. */
+ struct heap_node in_events_heap;
+ /** ID to sort events with the same deadline. */
+ int id;
+};
+
+/**
+ * Heap comparator. Heap's top stores an event with the nearest
+ * deadline and the smallest ID in that deadline.
+ */
+static inline bool
+swim_test_event_less(const struct swim_test_event *e1,
+ const struct swim_test_event *e2)
+{
+ if (e1->deadline == e2->deadline)
+ return e1->id < e2->id;
+ return e1->deadline < e2->deadline;
+}
+
+#define HEAP_NAME events_heap
+#define HEAP_LESS(h, e1, e2) swim_test_event_less(e1, e2)
+#define heap_value_t struct swim_test_event
+#define heap_value_attr in_events_heap
+#include "salad/heap.h"
+
+/** Events heap. Event loop pops them from here. */
+static heap_t events_heap;
+
+/** Libev watcher is matched to exactly one test event here. */
+static struct mh_i64ptr_t *events_hash;
+
+/**
+ * Create a new event which should happen after @a delay and with
+ * the specified events mask on behalf of @a watcher.
+ */
+static void
+swim_test_event_new(struct ev_watcher *watcher, double delay, int revents)
+{
+ struct swim_test_event *e =
+ (struct swim_test_event *) malloc(sizeof(*e));
+ assert(e != NULL);
+ e->watcher = watcher;
+ e->deadline = swim_time() + delay;
+ e->revents = revents;
+ e->id = event_id++;
+ events_heap_insert(&events_heap, e);
+ struct mh_i64ptr_node_t old = {0, NULL}, *old_p = &old;
+ struct mh_i64ptr_node_t node = {(uint64_t) watcher, e};
+ mh_int_t rc = mh_i64ptr_put(events_hash, &node, &old_p, NULL);
+ (void) rc;
+ assert(rc != mh_end(events_hash));
+ assert(old.val == NULL && old.key == 0);
+}
+
+/** Delete event, free resources. */
+static inline void
+swim_test_event_delete(struct swim_test_event *e)
+{
+ events_heap_delete(&events_heap, e);
+ mh_int_t rc = mh_i64ptr_find(events_hash, (uint64_t) e->watcher, NULL);
+ assert(rc != mh_end(events_hash));
+ mh_i64ptr_del(events_hash, rc, NULL);
+ free(e);
+}
+
+/** Find an event by @a watcher. */
+static struct swim_test_event *
+swim_test_event_by_ev(struct ev_watcher *watcher)
+{
+ mh_int_t rc = mh_i64ptr_find(events_hash, (uint64_t) watcher, NULL);
+ if (rc == mh_end(events_hash))
+ return NULL;
+ return (struct swim_test_event *) mh_i64ptr_node(events_hash, rc)->val;
+}
+
+/** Implementation of global time visible in SWIM. */
+double
+swim_time(void)
+{
+ return watch;
+}
+
+/**
+ * Start of a timer generates a delayed event. If a timer is
+ * already started - nothing happens.
+ */
+void
+swim_ev_timer_start(struct ev_loop *loop, struct ev_timer *base)
+{
+ if (swim_test_event_by_ev((struct ev_watcher *) base) != NULL)
+ return;
+ /* Create the periodic watcher and one event. */
+ swim_test_event_new((struct ev_watcher *) base, base->at, EV_TIMER);
+}
+
+/** Time stop cancels the event if the timer is active. */
+void
+swim_ev_timer_stop(struct ev_loop *loop, struct ev_timer *base)
+{
+ /*
+ * Delete the watcher and its events. Should be only one.
+ */
+ struct swim_test_event *e =
+ swim_test_event_by_ev((struct ev_watcher *) base);
+ if (e == NULL)
+ return;
+ swim_test_event_delete(e);
+}
+
+/** Process all the events with the next nearest deadline. */
+void
+swim_do_loop_step(struct ev_loop *loop)
+{
+ say_verbose("Loop watch %f", watch);
+ struct swim_test_event *e = events_heap_top(&events_heap);
+ if (e != NULL) {
+ assert(e->deadline >= watch);
+ /* Multiple events can have the same deadline. */
+ watch = e->deadline;
+ do {
+ int revents = e->revents;
+ struct ev_watcher *w = e->watcher;
+ swim_test_event_delete(e);
+ ev_invoke(loop, w, revents);
+ e = events_heap_top(&events_heap);
+ } while (e != NULL && e->deadline == watch);
+ }
+ /*
+ * After events are processed, it is possible that some of
+ * them generated IO events. Process them too.
+ */
+ do {
+ swim_transport_do_loop_step(loop);
+ /*
+ * Just a single loop + invoke is not enough. At
+ * least two are necessary.
+ *
+ * First loop does nothing since send queues are
+ * empty. First invoke fills send queues.
+ *
+ * Second loop moves messages from send to recv
+ * queues. Second invoke processes messages in
+ * recv queues.
+ *
+ * With indirect messages even 2 cycles is not
+ * enough - processing of one received message can
+ * add a new message into another send queue.
+ */
+ if (ev_pending_count(loop) == 0)
+ break;
+ ev_invoke_pending(loop);
+ } while (true);
+}
+
+void
+swim_test_ev_reset(void)
+{
+ struct swim_test_event *e;
+ while ((e = events_heap_top(&events_heap)) != NULL)
+ swim_test_event_delete(e);
+ assert(mh_size(events_hash) == 0);
+ event_id = 0;
+ watch = 0;
+}
+
+void
+swim_test_ev_init(void)
+{
+ events_hash = mh_i64ptr_new();
+ assert(events_hash != NULL);
+ events_heap_create(&events_heap);
+}
+
+void
+swim_test_ev_free(void)
+{
+ swim_test_ev_reset();
+ events_heap_destroy(&events_heap);
+ mh_i64ptr_delete(events_hash);
+}
diff --git a/test/unit/swim_test_ev.h b/test/unit/swim_test_ev.h
new file mode 100644
index 000000000..788ac06e9
--- /dev/null
+++ b/test/unit/swim_test_ev.h
@@ -0,0 +1,58 @@
+#ifndef TARANTOOL_SWIM_TEST_EV_H_INCLUDED
+#define TARANTOOL_SWIM_TEST_EV_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+struct ev_loop;
+
+/**
+ * SWIM test_ev implements a 'fake' event loop with bogus clock to
+ * speed up events processing while keeping SWIM unaware that it
+ * works in a simulation. Libev is used a little, just to store
+ * some IO events.
+ */
+
+/** Initialize test event processing system. */
+void
+swim_test_ev_init(void);
+
+/** Destroy test event processing system, free resources. */
+void
+swim_test_ev_free(void);
+
+/** Play one step of event loop, process generated events. */
+void
+swim_do_loop_step(struct ev_loop *loop);
+
+/** Destroy pending events, reset global watch. */
+void
+swim_test_ev_reset(void);
+
+#endif /* TARANTOOL_SWIM_TEST_EV_H_INCLUDED */
diff --git a/test/unit/swim_test_transport.c b/test/unit/swim_test_transport.c
new file mode 100644
index 000000000..d49b92885
--- /dev/null
+++ b/test/unit/swim_test_transport.c
@@ -0,0 +1,257 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "swim_test_transport.h"
+#include "swim/swim_transport.h"
+#include "swim/swim_io.h"
+#include "fiber.h"
+#include <errno.h>
+
+enum {
+ /**
+ * All fake sockets have fd >= 1000 in order to prevent
+ * possible intersections with real file descriptors.
+ */
+ FAKE_FD_BASE = 1000,
+ /**
+ * Maximal number of fake file descriptors. Nothing
+ * special about this value and fixed fd table size. It
+ * just simplifies code.
+ */
+ FAKE_FD_NUMBER = 1000,
+};
+
+/** UDP packet wrapper. It is stored in send/recv queues. */
+struct swim_test_packet {
+ /** Source address. */
+ struct sockaddr_in src;
+ /** Destination address. */
+ struct sockaddr_in dst;
+ /** A link in send/recv queue. */
+ struct rlist in_queue;
+ /** Packet data size. */
+ int size;
+ /** Packet data. */
+ char data[0];
+};
+
+/** Wrap @a data into a new packet. */
+static inline struct swim_test_packet *
+swim_test_packet_new(const char *data, int size, const struct sockaddr_in *src,
+ const struct sockaddr_in *dst)
+{
+ struct swim_test_packet *p =
+ (struct swim_test_packet *) malloc(sizeof(*p) + size);
+ assert(p != NULL);
+ rlist_create(&p->in_queue);
+ p->src = *src;
+ p->dst = *dst;
+ p->size = size;
+ memcpy(p->data, data, size);
+ return p;
+}
+
+/** Fake file descriptor. */
+struct swim_fd {
+ /** File descriptor number visible to libev. */
+ int evfd;
+ /**
+ * Link in the list of opened descriptors. Used to feed
+ * them all EV_WRITE.
+ */
+ struct rlist in_opened;
+ /** Queue of received, but not processed packets. */
+ struct rlist recv_queue;
+ /** Queue of sent, but not received packets. */
+ struct rlist send_queue;
+};
+
+/** Table of fake file descriptors. */
+static struct swim_fd swim_fd[FAKE_FD_NUMBER];
+/**
+ * List of opened file descriptors. Used to avoid fullscan of the
+ * table.
+ */
+static RLIST_HEAD(swim_fd_opened);
+
+/** Open a fake file descriptor. */
+static inline int
+swim_fd_open(struct swim_fd *fd)
+{
+ if (! rlist_empty(&fd->in_opened)) {
+ errno = EADDRINUSE;
+ diag_set(SocketError, "test_socket:1", "bind");
+ return -1;
+ }
+ rlist_add_tail_entry(&swim_fd_opened, fd, in_opened);
+ rlist_create(&fd->recv_queue);
+ rlist_create(&fd->send_queue);
+ return 0;
+}
+
+/** Close a fake file descriptor. */
+static inline void
+swim_fd_close(struct swim_fd *fd)
+{
+ struct swim_test_packet *i, *tmp;
+ rlist_foreach_entry_safe(i, &fd->recv_queue, in_queue, tmp)
+ free(i);
+ rlist_foreach_entry_safe(i, &fd->send_queue, in_queue, tmp)
+ free(i);
+ rlist_del_entry(fd, in_opened);
+}
+
+void
+swim_test_transport_init(void)
+{
+ for (int i = 0, evfd = FAKE_FD_BASE; i < FAKE_FD_NUMBER; ++i, ++evfd) {
+ swim_fd[i].evfd = evfd;
+ rlist_create(&swim_fd[i].in_opened);
+ rlist_create(&swim_fd[i].recv_queue);
+ rlist_create(&swim_fd[i].send_queue);
+ }
+}
+
+void
+swim_test_transport_free(void)
+{
+ struct swim_test_packet *p, *tmp;
+ for (int i = 0; i < (int)lengthof(swim_fd); ++i)
+ swim_fd_close(&swim_fd[i]);
+}
+
+/**
+ * Wrap a packet and put into send queue. Packets are popped from
+ * it on EV_WRITE event.
+ */
+ssize_t
+swim_transport_send(struct swim_transport *transport, const void *data,
+ size_t size, const struct sockaddr *addr,
+ socklen_t addr_size)
+{
+ /*
+ * Create packet. Put into sending queue.
+ */
+ (void) addr_size;
+ assert(addr->sa_family == AF_INET);
+ struct swim_test_packet *p =
+ swim_test_packet_new(data, size, &transport->addr,
+ (const struct sockaddr_in *) addr);
+ struct swim_fd *src = &swim_fd[transport->fd - FAKE_FD_BASE];
+ rlist_add_tail_entry(&src->send_queue, p, in_queue);
+ return size;
+}
+
+/**
+ * Move a packet from send to recv queue. The packet is popped and
+ * processed on EV_READ event.
+ */
+ssize_t
+swim_transport_recv(struct swim_transport *transport, void *buffer, size_t size,
+ struct sockaddr *addr, socklen_t *addr_size)
+{
+ /*
+ * Pop a packet from a receving queue.
+ */
+ struct swim_fd *dst = &swim_fd[transport->fd - FAKE_FD_BASE];
+ struct swim_test_packet *p =
+ rlist_shift_entry(&dst->recv_queue, struct swim_test_packet,
+ in_queue);
+ *(struct sockaddr_in *) addr = p->src;
+ *addr_size = sizeof(p->src);
+ ssize_t result = MIN((size_t) p->size, size);
+ memcpy(buffer, p->data, result);
+ free(p);
+ return result;
+}
+
+int
+swim_transport_bind(struct swim_transport *transport,
+ const struct sockaddr *addr, socklen_t addr_len)
+{
+ assert(addr->sa_family == AF_INET);
+ const struct sockaddr_in *new_addr = (const struct sockaddr_in *) addr;
+ int new_fd = ntohs(new_addr->sin_port) + FAKE_FD_BASE;
+ int old_fd = transport->fd;
+ if (old_fd == new_fd)
+ return 0;
+ if (swim_fd_open(&swim_fd[new_fd - FAKE_FD_BASE]) != 0)
+ return -1;
+ transport->fd = new_fd;
+ transport->addr = *new_addr;
+ if (old_fd != -1)
+ swim_fd_close(&swim_fd[old_fd - FAKE_FD_BASE]);
+ return 0;
+}
+
+void
+swim_transport_destroy(struct swim_transport *transport)
+{
+ if (transport->fd != -1)
+ swim_fd_close(&swim_fd[transport->fd - FAKE_FD_BASE]);
+}
+
+void
+swim_transport_create(struct swim_transport *transport)
+{
+ transport->fd = -1;
+ memset(&transport->addr, 0, sizeof(transport->addr));
+}
+
+/** Send one packet to destination's recv queue. */
+static inline void
+swim_fd_send_packet(struct swim_fd *fd)
+{
+ if (rlist_empty(&fd->send_queue))
+ return;
+ struct swim_test_packet *p =
+ rlist_shift_entry(&fd->send_queue, struct swim_test_packet,
+ in_queue);
+ int dst_i = ntohs(p->dst.sin_port);
+ rlist_add_tail_entry(&swim_fd[dst_i].recv_queue, p, in_queue);
+}
+
+void
+swim_transport_do_loop_step(struct ev_loop *loop)
+{
+ struct swim_fd *fd;
+ /*
+ * Reversed because libev invokes events in reversed
+ * order. So this reverse + libev reverse = normal order.
+ */
+ rlist_foreach_entry_reverse(fd, &swim_fd_opened, in_opened) {
+ swim_fd_send_packet(fd);
+ ev_feed_fd_event(loop, fd->evfd, EV_WRITE);
+ }
+ rlist_foreach_entry_reverse(fd, &swim_fd_opened, in_opened) {
+ if (!rlist_empty(&fd->recv_queue))
+ ev_feed_fd_event(loop, fd->evfd, EV_READ);
+ }
+}
diff --git a/test/unit/swim_test_transport.h b/test/unit/swim_test_transport.h
new file mode 100644
index 000000000..c9739e250
--- /dev/null
+++ b/test/unit/swim_test_transport.h
@@ -0,0 +1,58 @@
+#ifndef TARANTOOL_SWIM_TEST_TRANSPORT_H_INCLUDED
+#define TARANTOOL_SWIM_TEST_TRANSPORT_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+struct ev_loop;
+
+/**
+ * SWIM test_transport implements a 'fake' file descriptors table
+ * in user space in orider to get the full control over UDP
+ * sockets, used in core SWIM code. Fake fds are used to provide a
+ * capability to set necessary loss level, delay, reorders.
+ */
+
+/**
+ * Feed EV_WRITE event to all opened descriptors, and EV_READ to
+ * ones, who have not empty recv queue. Move packets from send to
+ * recv queues. No callbacks is invoked. Only events are fed.
+ */
+void
+swim_transport_do_loop_step(struct ev_loop *loop);
+
+/** Initialize test transport system. */
+void
+swim_test_transport_init(void);
+
+/** Destroy test transport system, free resources. */
+void
+swim_test_transport_free(void);
+
+#endif /* TARANTOOL_SWIM_TEST_TRANSPORT_H_INCLUDED */
diff --git a/test/unit/swim_test_utils.c b/test/unit/swim_test_utils.c
new file mode 100644
index 000000000..20775effb
--- /dev/null
+++ b/test/unit/swim_test_utils.c
@@ -0,0 +1,139 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "swim_test_utils.h"
+#include "swim_test_ev.h"
+#include "swim/swim.h"
+#include "uuid/tt_uuid.h"
+#include "trivia/util.h"
+#include "fiber.h"
+
+/**
+ * Cluster is a simple array of SWIM instances assigned to
+ * different URIs.
+ */
+struct swim_cluster {
+ int size;
+ struct swim **node;
+};
+
+struct swim_cluster *
+swim_cluster_new(int size)
+{
+ struct swim_cluster *res = (struct swim_cluster *) malloc(sizeof(*res));
+ assert(res != NULL);
+ int bsize = sizeof(res->node[0]) * size;
+ res->node = (struct swim **) malloc(bsize);
+ assert(res->node != NULL);
+ res->size = size;
+ struct tt_uuid uuid;
+ memset(&uuid, 0, sizeof(uuid));
+ char *uri = tt_static_buf();
+ for (int i = 0; i < size; ++i) {
+ res->node[i] = swim_new();
+ assert(res->node[i] != NULL);
+ sprintf(uri, "127.0.0.1:%d", i + 1);
+ uuid.time_low = i + 1;
+ int rc = swim_cfg(res->node[i], uri, -1, &uuid);
+ assert(rc == 0);
+ (void) rc;
+ }
+ return res;
+}
+
+void
+swim_cluster_delete(struct swim_cluster *cluster)
+{
+ for (int i = 0; i < cluster->size; ++i)
+ swim_delete(cluster->node[i]);
+ free(cluster->node);
+ free(cluster);
+}
+
+int
+swim_cluster_add_link(struct swim_cluster *cluster, int to_id, int from_id)
+{
+ const struct swim_member *from = swim_self(cluster->node[from_id]);
+ return swim_add_member(cluster->node[to_id], swim_member_uri(from),
+ swim_member_uuid(from));
+}
+
+struct swim *
+swim_cluster_node(struct swim_cluster *cluster, int i)
+{
+ assert(i >= 0 && i < cluster->size);
+ return cluster->node[i];
+}
+
+/** Check if @a s1 knows every member of @a s2's table. */
+static inline bool
+swim1_contains_swim2(struct swim *s1, struct swim *s2)
+{
+ struct swim_iterator *it = swim_iterator_open(s1);
+ const struct swim_member *m;
+ while ((m = swim_iterator_next(it)) != NULL) {
+ if (swim_member_by_uuid(s2, swim_member_uuid(m)) == NULL) {
+ swim_iterator_close(it);
+ return false;
+ }
+ }
+ swim_iterator_close(it);
+ return false;
+}
+
+bool
+swim_cluster_is_fullmesh(struct swim_cluster *cluster)
+{
+ struct swim **end = cluster->node + cluster->size;
+ for (struct swim **s1 = cluster->node; s1 < end; ++s1) {
+ for (struct swim **s2 = s1 + 1; s2 < end; ++s2) {
+ if (! swim1_contains_swim2(*s1, *s2) ||
+ ! swim1_contains_swim2(*s2, *s1))
+ return false;
+ }
+ }
+ return true;
+}
+
+int
+swim_cluster_wait_fullmesh(struct swim_cluster *cluster, int max_steps)
+{
+ while (! swim_cluster_is_fullmesh(cluster) && max_steps > 0) {
+ swim_do_loop_step(loop());
+ --max_steps;
+ }
+ return max_steps < 0 ? -1 : 0;
+}
+
+bool
+swim_error_check_match(const char *msg)
+{
+ return strstr(diag_last_error(diag_get())->errmsg, msg) != NULL;
+}
diff --git a/test/unit/swim_test_utils.h b/test/unit/swim_test_utils.h
new file mode 100644
index 000000000..19eea551d
--- /dev/null
+++ b/test/unit/swim_test_utils.h
@@ -0,0 +1,92 @@
+#ifndef TARANTOOL_SWIM_TEST_UTILS_H_INCLUDED
+#define TARANTOOL_SWIM_TEST_UTILS_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ * copyright notice, this list of conditions and the
+ * following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include <stdbool.h>
+
+struct swim_cluster;
+struct swim;
+
+/**
+ * Create a new cluster of SWIM instances. Instances are assigned
+ * URIs like '127.0.0.1:[1 - size]' and UUIDs like
+ * '00...00[1 - size]'. Instances can be got by their ordinal
+ * numbers equal to their port and to last part of UUID.
+ */
+struct swim_cluster *
+swim_cluster_new(int size);
+
+/** Delete all the SWIM instances, and the cluster itself. */
+void
+swim_cluster_delete(struct swim_cluster *cluster);
+
+/** Check that an error in diag contains @a msg. */
+bool
+swim_error_check_match(const char *msg);
+
+/** Get a SWIM instance by its ordinal number. */
+struct swim *
+swim_cluster_node(struct swim_cluster *cluster, int i);
+
+/**
+ * Explicitly add a member of id @a from_id to a member of id
+ * @a to_id.
+ */
+int
+swim_cluster_add_link(struct swim_cluster *cluster, int to_id, int from_id);
+
+/**
+ * Check if in the cluster every instance knowns the about other
+ * instances.
+ */
+bool
+swim_cluster_is_fullmesh(struct swim_cluster *cluster);
+
+/**
+ * Wait for fullmesh at most @a max_steps event loop iterations.
+ */
+int
+swim_cluster_wait_fullmesh(struct swim_cluster *cluster, int max_steps);
+
+#define swim_start_test(n) { \
+ header(); \
+ say_verbose("-------- SWIM start test %s --------", __func__); \
+ plan(n); \
+}
+
+#define swim_finish_test() { \
+ say_verbose("-------- SWIM end test %s --------", __func__); \
+ swim_test_ev_reset(); \
+ check_plan(); \
+ footer(); \
+}
+
+#endif /* TARANTOOL_SWIM_TEST_UTILS_H_INCLUDED */
--
2.17.2 (Apple Git-113)
More information about the Tarantool-patches
mailing list