[tarantool-patches] [PATCH 1/2] Lightweight vclock_create and vclock_copy

[Georgy Kirichenko <georgy@tarantool.org>]

Modify only needed part of a vclock structure instead of full structure
writing.

Follow-up #2283
---
 src/box/vclock.c    |  5 +++--
 src/box/vclock.h    | 21 ++++++++++++++++-----
 src/box/xrow.h      |  4 ++--
 test/unit/vclock.cc |  2 +-
 4 files changed, 22 insertions(+), 10 deletions(-)

diff --git a/src/box/vclock.c b/src/box/vclock.c
index b5eb2800b..d4b2ba759 100644
--- a/src/box/vclock.c
+++ b/src/box/vclock.c
@@ -41,7 +41,7 @@ vclock_follow(struct vclock *vclock, uint32_t replica_id, int64_t lsn)
 {
 	assert(lsn >= 0);
 	assert(replica_id < VCLOCK_MAX);
-	int64_t prev_lsn = vclock->lsn[replica_id];
+	int64_t prev_lsn = vclock_get(vclock, replica_id);
 	assert(lsn > prev_lsn);
 	/* Easier add each time than check. */
 	vclock->map |= 1 << replica_id;
@@ -159,7 +159,8 @@ vclock_from_string(struct vclock *vclock, const char *str)
 			errno = 0;
 			lsn = strtoll(p, (char **)  &p, 10);
 			if (errno != 0 || lsn < 0 || lsn > INT64_MAX ||
-			    replica_id >= VCLOCK_MAX || vclock->lsn[replica_id] > 0)
+			    replica_id >= VCLOCK_MAX ||
+			    vclock_get(vclock, replica_id) > 0)
 				goto error;
 			vclock->map |= 1 << replica_id;
 			vclock->lsn[replica_id] = lsn;
diff --git a/src/box/vclock.h b/src/box/vclock.h
index 111e29160..9e97aff27 100644
--- a/src/box/vclock.h
+++ b/src/box/vclock.h
@@ -129,7 +129,8 @@ vclock_iterator_next(struct vclock_iterator *it)
 static inline void
 vclock_create(struct vclock *vclock)
 {
-	memset(vclock, 0, sizeof(*vclock));
+	vclock->signature = 0;
+	vclock->map = 0;
 }
 
 /**
@@ -139,8 +140,8 @@ vclock_create(struct vclock *vclock)
 static inline void
 vclock_clear(struct vclock *vclock)
 {
-	memset(vclock, 0, sizeof(*vclock));
 	vclock->signature = -1;
+	vclock->map = 0;
 }
 
 /**
@@ -158,6 +159,8 @@ vclock_get(const struct vclock *vclock, uint32_t replica_id)
 {
 	if (replica_id >= VCLOCK_MAX)
 		return 0;
+	if ((vclock->map & (1 << replica_id)) == 0)
+		return 0;
 	return vclock->lsn[replica_id];
 }
 
@@ -165,6 +168,8 @@ static inline int64_t
 vclock_inc(struct vclock *vclock, uint32_t replica_id)
 {
 	/* Easier add each time than check. */
+	if ((vclock->map & (1 << replica_id)) == 0)
+		vclock->lsn[replica_id] = 0;
 	vclock->map |= 1 << replica_id;
 	vclock->signature++;
 	return ++vclock->lsn[replica_id];
@@ -173,7 +178,13 @@ vclock_inc(struct vclock *vclock, uint32_t replica_id)
 static inline void
 vclock_copy(struct vclock *dst, const struct vclock *src)
 {
-	*dst = *src;
+	if (src->map == 0) {
+		dst->map = src->map;
+		dst->signature = src->signature;
+		return;
+	}
+	unsigned int max_pos = VCLOCK_MAX - bit_clz_u32(src->map);
+	memcpy(dst, src, offsetof(struct vclock, lsn) + sizeof(*dst->lsn) * max_pos);
 }
 
 static inline uint32_t
@@ -253,8 +264,8 @@ vclock_compare(const struct vclock *a, const struct vclock *b)
 	for (size_t replica_id = bit_iterator_next(&it); replica_id < VCLOCK_MAX;
 	     replica_id = bit_iterator_next(&it)) {
 
-		int64_t lsn_a = a->lsn[replica_id];
-		int64_t lsn_b = b->lsn[replica_id];
+		int64_t lsn_a = vclock_get(a, replica_id);
+		int64_t lsn_b = vclock_get(b, replica_id);
 		le = le && lsn_a <= lsn_b;
 		ge = ge && lsn_a >= lsn_b;
 		if (!ge && !le)
diff --git a/src/box/xrow.h b/src/box/xrow.h
index 719add4f0..2fce83bbc 100644
--- a/src/box/xrow.h
+++ b/src/box/xrow.h
@@ -631,7 +631,7 @@ vclock_follow_xrow(struct vclock* vclock, const struct xrow_header *row)
 {
 	assert(row);
 	assert(row->replica_id < VCLOCK_MAX);
-	if (row->lsn <= vclock->lsn[row->replica_id]) {
+	if (row->lsn <= vclock_get(vclock, row->replica_id)) {
 		struct request req;
 		const char *req_str = "n/a";
 		if (xrow_decode_dml((struct xrow_header *)row, &req, 0) == 0)
@@ -640,7 +640,7 @@ vclock_follow_xrow(struct vclock* vclock, const struct xrow_header *row)
 		panic("LSN for %u is used twice or COMMIT order is broken: "
 		      "confirmed: %lld, new: %lld, req: %s",
 		      (unsigned) row->replica_id,
-		      (long long) vclock->lsn[row->replica_id],
+		      (long long) vclock_get(vclock, row->replica_id),
 		      (long long) row->lsn,
 		      req_str);
 	}
diff --git a/test/unit/vclock.cc b/test/unit/vclock.cc
index 8498eba3b..6a1d3bc27 100644
--- a/test/unit/vclock.cc
+++ b/test/unit/vclock.cc
@@ -308,7 +308,7 @@ test_fromstring_one(const char *str, uint32_t count, const int64_t *lsns)
 	vclock_create(&check);
 	for (uint32_t node_id = 0; node_id < count; node_id++) {
 		if (lsns[node_id] >= 0)
-			check.lsn[node_id] = lsns[node_id];
+			vclock_follow(&check, node_id, lsns[node_id]);
 	}
 
 	return (rc != 0 || vclock_compare(&vclock, &check) != 0);
-- 
2.20.1