[PATCH 1/2] xlog: simplify xdir_add_vclock protocol

[Vladimir Davydov <vdavydov.dev@gmail.com>]

We add the vclock of a new snapshot/xlog/vylog file to the corresponding
xdir, because we need it for garbage collection and backup. We use the
xdir_add_vclock() function for that. Currently, the function protocol is
rather abstruse: it expects that the caller allocates a vclock with
malloc() and passes the ownership on to the xdir by calling this
function. This is done that way, because we add a vclock to an xdir
after committing a new xlog file, where we can't fail anymore. Since
malloc() can theoretically fail, we allocate a vclock before writing an
xlog and add it to an xdir only after the xlog is committed. This makes
the code look rather complicated.

Actually, as explained in #3534, malloc() doesn't normally fail so this
complexity is in fact needless and only makes it more difficult to patch
the code. Let's simplify the code by making xdir_add_vclock() allocate
vclock by itself and, since it can't fail, panic on allocation error.
---
https://github.com/tarantool/tarantool/issues/3624
https://github.com/tarantool/tarantool/commits/dv/gh-3624-vy-fix-vylog-backup

 src/box/memtx_engine.c | 26 ++++++++------------------
 src/box/vy_log.c       | 23 ++++-------------------
 src/box/wal.c          | 12 +-----------
 src/box/xlog.c         | 10 ++++++++++
 src/box/xlog.h         |  8 ++------
 5 files changed, 25 insertions(+), 54 deletions(-)

diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index f5ace926..1f80ce54 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -580,7 +580,7 @@ struct checkpoint {
 	struct cord cord;
 	bool waiting_for_snap_thread;
 	/** The vclock of the snapshot file. */
-	struct vclock *vclock;
+	struct vclock vclock;
 	struct xdir dir;
 	/**
 	 * Do nothing, just touch the snapshot file - the
@@ -597,14 +597,7 @@ checkpoint_init(struct checkpoint *ckpt, const char *snap_dirname,
 	ckpt->waiting_for_snap_thread = false;
 	xdir_create(&ckpt->dir, snap_dirname, SNAP, &INSTANCE_UUID);
 	ckpt->snap_io_rate_limit = snap_io_rate_limit;
-	/* May be used in abortCheckpoint() */
-	ckpt->vclock = malloc(sizeof(*ckpt->vclock));
-	if (ckpt->vclock == NULL) {
-		diag_set(OutOfMemory, sizeof(*ckpt->vclock),
-			 "malloc", "vclock");
-		return -1;
-	}
-	vclock_create(ckpt->vclock);
+	vclock_create(&ckpt->vclock);
 	ckpt->touch = false;
 	return 0;
 }
@@ -618,7 +611,6 @@ checkpoint_destroy(struct checkpoint *ckpt)
 	}
 	rlist_create(&ckpt->entries);
 	xdir_destroy(&ckpt->dir);
-	free(ckpt->vclock);
 }
 
 
@@ -656,7 +648,7 @@ checkpoint_f(va_list ap)
 	struct checkpoint *ckpt = va_arg(ap, struct checkpoint *);
 
 	if (ckpt->touch) {
-		if (xdir_touch_xlog(&ckpt->dir, ckpt->vclock) == 0)
+		if (xdir_touch_xlog(&ckpt->dir, &ckpt->vclock) == 0)
 			return 0;
 		/*
 		 * Failed to touch an existing snapshot, create
@@ -666,7 +658,7 @@ checkpoint_f(va_list ap)
 	}
 
 	struct xlog snap;
-	if (xdir_create_xlog(&ckpt->dir, &snap, ckpt->vclock) != 0)
+	if (xdir_create_xlog(&ckpt->dir, &snap, &ckpt->vclock) != 0)
 		return -1;
 
 	snap.rate_limit = ckpt->snap_io_rate_limit;
@@ -739,7 +731,7 @@ memtx_engine_wait_checkpoint(struct engine *engine,
 	    vclock_compare(&last, vclock) == 0) {
 		memtx->checkpoint->touch = true;
 	}
-	vclock_copy(memtx->checkpoint->vclock, vclock);
+	vclock_copy(&memtx->checkpoint->vclock, vclock);
 
 	if (cord_costart(&memtx->checkpoint->cord, "snapshot",
 			 checkpoint_f, memtx->checkpoint)) {
@@ -771,7 +763,7 @@ memtx_engine_commit_checkpoint(struct engine *engine,
 	small_alloc_setopt(&memtx->alloc, SMALL_DELAYED_FREE_MODE, false);
 
 	if (!memtx->checkpoint->touch) {
-		int64_t lsn = vclock_sum(memtx->checkpoint->vclock);
+		int64_t lsn = vclock_sum(&memtx->checkpoint->vclock);
 		struct xdir *dir = &memtx->checkpoint->dir;
 		/* rename snapshot on completion */
 		char to[PATH_MAX];
@@ -795,9 +787,7 @@ memtx_engine_commit_checkpoint(struct engine *engine,
 	if (xdir_last_vclock(&memtx->snap_dir, &last) < 0 ||
 	    vclock_compare(&last, vclock) != 0) {
 		/* Add the new checkpoint to the set. */
-		xdir_add_vclock(&memtx->snap_dir, memtx->checkpoint->vclock);
-		/* Prevent checkpoint_destroy() from freeing vclock. */
-		memtx->checkpoint->vclock = NULL;
+		xdir_add_vclock(&memtx->snap_dir, &memtx->checkpoint->vclock);
 	}
 
 	checkpoint_destroy(memtx->checkpoint);
@@ -824,7 +814,7 @@ memtx_engine_abort_checkpoint(struct engine *engine)
 	/** Remove garbage .inprogress file. */
 	char *filename =
 		xdir_format_filename(&memtx->checkpoint->dir,
-				     vclock_sum(memtx->checkpoint->vclock),
+				     vclock_sum(&memtx->checkpoint->vclock),
 				     INPROGRESS);
 	(void) coio_unlink(filename);
 
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index 3843cad6..d8613170 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -893,14 +893,9 @@ vy_log_bootstrap(void)
 		return vy_log_rebootstrap();
 
 	/* Add initial vclock to the xdir. */
-	struct vclock *vclock = malloc(sizeof(*vclock));
-	if (vclock == NULL) {
-		diag_set(OutOfMemory, sizeof(*vclock),
-			 "malloc", "struct vclock");
-		return -1;
-	}
-	vclock_create(vclock);
-	xdir_add_vclock(&vy_log.dir, vclock);
+	struct vclock vclock;
+	vclock_create(&vclock);
+	xdir_add_vclock(&vy_log.dir, &vclock);
 	return 0;
 }
 
@@ -1020,15 +1015,6 @@ vy_log_rotate(const struct vclock *vclock)
 		return 0;
 
 	assert(signature > prev_signature);
-
-	struct vclock *new_vclock = malloc(sizeof(*new_vclock));
-	if (new_vclock == NULL) {
-		diag_set(OutOfMemory, sizeof(*new_vclock),
-			 "malloc", "struct vclock");
-		return -1;
-	}
-	vclock_copy(new_vclock, vclock);
-
 	say_verbose("rotating vylog %lld => %lld",
 		    (long long)prev_signature, (long long)signature);
 
@@ -1065,14 +1051,13 @@ vy_log_rotate(const struct vclock *vclock)
 	vclock_copy(&vy_log.last_checkpoint, vclock);
 
 	/* Add the new vclock to the xdir so that we can track it. */
-	xdir_add_vclock(&vy_log.dir, new_vclock);
+	xdir_add_vclock(&vy_log.dir, vclock);
 
 	latch_unlock(&vy_log.latch);
 	say_verbose("done rotating vylog");
 	return 0;
 fail:
 	latch_unlock(&vy_log.latch);
-	free(new_vclock);
 	return -1;
 }
 
diff --git a/src/box/wal.c b/src/box/wal.c
index 41762a59..cd068a4e 100644
--- a/src/box/wal.c
+++ b/src/box/wal.c
@@ -579,26 +579,16 @@ wal_opt_rotate(struct wal_writer *writer)
 	if (xlog_is_open(&writer->current_wal))
 		return 0;
 
-	struct vclock *vclock = (struct vclock *)malloc(sizeof(*vclock));
-	if (vclock == NULL) {
-		diag_set(OutOfMemory, sizeof(*vclock),
-			 "malloc", "struct vclock");
-		diag_log();
-		return -1;
-	}
-	vclock_copy(vclock, &writer->vclock);
-
 	if (xdir_create_xlog(&writer->wal_dir, &writer->current_wal,
 			     &writer->vclock) != 0) {
 		diag_log();
-		free(vclock);
 		return -1;
 	}
 	/*
 	 * Keep track of the new WAL vclock. Required for garbage
 	 * collection, see wal_collect_garbage().
 	 */
-	xdir_add_vclock(&writer->wal_dir, vclock);
+	xdir_add_vclock(&writer->wal_dir, &writer->vclock);
 
 	wal_notify_watchers(writer, WAL_EVENT_ROTATE);
 	return 0;
diff --git a/src/box/xlog.c b/src/box/xlog.c
index e35fb4d9..d399a726 100644
--- a/src/box/xlog.c
+++ b/src/box/xlog.c
@@ -707,6 +707,16 @@ xdir_collect_inprogress(struct xdir *xdir)
 	}
 }
 
+void
+xdir_add_vclock(struct xdir *xdir, const struct vclock *vclock)
+{
+	struct vclock *copy = malloc(sizeof(*vclock));
+	if (copy == NULL)
+		panic("failed to allocate vclock");
+	vclock_copy(copy, vclock);
+	vclockset_insert(&xdir->index, copy);
+}
+
 /* }}} */
 
 
diff --git a/src/box/xlog.h b/src/box/xlog.h
index 476105c2..c2ac4774 100644
--- a/src/box/xlog.h
+++ b/src/box/xlog.h
@@ -208,13 +208,9 @@ xdir_last_vclock(struct xdir *xdir, struct vclock *vclock)
 
 /**
  * Insert a vclock into the file index of a directory.
- * The vclock must be allocated with malloc().
  */
-static inline void
-xdir_add_vclock(struct xdir *xdir, struct vclock *vclock)
-{
-	vclockset_insert(&xdir->index, vclock);
-}
+void
+xdir_add_vclock(struct xdir *xdir, const struct vclock *vclock);
 
 /* }}} */
 
-- 
2.11.0