[PATCH 2/6] engine: pass vclock instead of lsn to collect_garbage callback

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

First, this is consistent with other engine callbacks, such as
checkpoint or backup.

Second, a vclock can be used as a search key in a vclock set,
which in turn can make code more straightforward, e.g. look how
this patch simplifies vy_log_prev_checkpoint().
---
 src/box/box.cc         |  2 +-
 src/box/engine.c       |  9 +++++----
 src/box/engine.h       |  9 +++++----
 src/box/gc.c           |  7 ++-----
 src/box/memtx_engine.c |  5 +++--
 src/box/vinyl.c        |  9 ++++-----
 src/box/vy_log.c       | 36 ++++++++++++++++--------------------
 src/box/vy_log.h       |  4 ++--
 src/box/wal.c          | 41 +++++++++++++++++++++++------------------
 src/box/wal.h          |  9 +++++----
 10 files changed, 66 insertions(+), 65 deletions(-)

diff --git a/src/box/box.cc b/src/box/box.cc
index ab6658fb..21b84991 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -2129,7 +2129,7 @@ box_cfg_xc(void)
 	enum wal_mode wal_mode = box_check_wal_mode(cfg_gets("wal_mode"));
 	if (wal_init(wal_mode, cfg_gets("wal_dir"), wal_max_rows,
 		     wal_max_size, &INSTANCE_UUID, &replicaset.vclock,
-		     vclock_sum(&first_checkpoint->vclock))) {
+		     &first_checkpoint->vclock) != 0) {
 		diag_raise();
 	}
 	gc_set_wal_watcher();
diff --git a/src/box/engine.c b/src/box/engine.c
index 2a30dcdd..409479a3 100644
--- a/src/box/engine.c
+++ b/src/box/engine.c
@@ -148,11 +148,11 @@ engine_abort_checkpoint(void)
 }
 
 int
-engine_collect_garbage(int64_t lsn)
+engine_collect_garbage(const struct vclock *vclock)
 {
 	struct engine *engine;
 	engine_foreach(engine) {
-		if (engine->vtab->collect_garbage(engine, lsn) < 0)
+		if (engine->vtab->collect_garbage(engine, vclock) < 0)
 			return -1;
 	}
 	return 0;
@@ -317,10 +317,11 @@ generic_engine_abort_checkpoint(struct engine *engine)
 }
 
 int
-generic_engine_collect_garbage(struct engine *engine, int64_t lsn)
+generic_engine_collect_garbage(struct engine *engine,
+			       const struct vclock *vclock)
 {
 	(void)engine;
-	(void)lsn;
+	(void)vclock;
 	return 0;
 }
 
diff --git a/src/box/engine.h b/src/box/engine.h
index 5b96c744..87c84e51 100644
--- a/src/box/engine.h
+++ b/src/box/engine.h
@@ -156,7 +156,7 @@ struct engine_vtab {
 	void (*abort_checkpoint)(struct engine *);
 	/**
 	 * Remove files that are not needed to recover
-	 * from checkpoint with @lsn or newer.
+	 * from checkpoint @vclock or newer.
 	 *
 	 * If this function returns a non-zero value, garbage
 	 * collection is aborted, i.e. this method isn't called
@@ -166,7 +166,8 @@ struct engine_vtab {
 	 * fails to delete a snapshot file, because we recover
 	 * checkpoint list by scanning the snapshot directory.
 	 */
-	int (*collect_garbage)(struct engine *engine, int64_t lsn);
+	int (*collect_garbage)(struct engine *engine,
+			       const struct vclock *vclock);
 	/**
 	 * Backup callback. It is supposed to call @cb for each file
 	 * that needs to be backed up in order to restore from the
@@ -337,7 +338,7 @@ void
 engine_abort_checkpoint(void);
 
 int
-engine_collect_garbage(int64_t lsn);
+engine_collect_garbage(const struct vclock *vclock);
 
 int
 engine_backup(const struct vclock *vclock, engine_backup_cb cb, void *cb_arg);
@@ -369,7 +370,7 @@ int generic_engine_begin_checkpoint(struct engine *);
 int generic_engine_wait_checkpoint(struct engine *, const struct vclock *);
 void generic_engine_commit_checkpoint(struct engine *, const struct vclock *);
 void generic_engine_abort_checkpoint(struct engine *);
-int generic_engine_collect_garbage(struct engine *, int64_t);
+int generic_engine_collect_garbage(struct engine *, const struct vclock *);
 int generic_engine_backup(struct engine *, const struct vclock *,
 			  engine_backup_cb, void *);
 void generic_engine_memory_stat(struct engine *, struct engine_memory_stat *);
diff --git a/src/box/gc.c b/src/box/gc.c
index cefe1553..55c36d15 100644
--- a/src/box/gc.c
+++ b/src/box/gc.c
@@ -201,9 +201,6 @@ gc_run(void)
 	if (!run_engine_gc && !run_wal_gc)
 		return; /* nothing to do */
 
-	int64_t wal_lsn = vclock_sum(vclock);
-	int64_t checkpoint_lsn = vclock_sum(&checkpoint->vclock);
-
 	/*
 	 * Engine callbacks may sleep, because they use coio for
 	 * removing files. Make sure we won't try to remove the
@@ -220,14 +217,14 @@ gc_run(void)
 	 */
 	int rc = 0;
 	if (run_engine_gc)
-		rc = engine_collect_garbage(checkpoint_lsn);
+		rc = engine_collect_garbage(&checkpoint->vclock);
 	/*
 	 * Run wal_collect_garbage() even if we don't need to
 	 * delete any WAL files, because we have to apprise
 	 * the WAL thread of the oldest checkpoint signature.
 	 */
 	if (rc == 0)
-		wal_collect_garbage(wal_lsn, checkpoint_lsn);
+		wal_collect_garbage(vclock, &checkpoint->vclock);
 	latch_unlock(&gc.latch);
 }
 
diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index 28afb324..9e05ecf6 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -822,7 +822,7 @@ memtx_engine_abort_checkpoint(struct engine *engine)
 }
 
 static int
-memtx_engine_collect_garbage(struct engine *engine, int64_t lsn)
+memtx_engine_collect_garbage(struct engine *engine, const struct vclock *vclock)
 {
 	struct memtx_engine *memtx = (struct memtx_engine *)engine;
 	/*
@@ -833,7 +833,8 @@ memtx_engine_collect_garbage(struct engine *engine, int64_t lsn)
 	 * That said, we have to abort garbage collection if we
 	 * fail to delete a snap file.
 	 */
-	if (xdir_collect_garbage(&memtx->snap_dir, lsn, XDIR_GC_USE_COIO) != 0)
+	if (xdir_collect_garbage(&memtx->snap_dir, vclock_sum(vclock),
+				 XDIR_GC_USE_COIO) != 0)
 		return -1;
 
 	return 0;
diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index ce81c6ad..1794489d 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -3286,21 +3286,20 @@ vy_gc(struct vy_env *env, struct vy_recovery *recovery,
 }
 
 static int
-vinyl_engine_collect_garbage(struct engine *engine, int64_t lsn)
+vinyl_engine_collect_garbage(struct engine *engine, const struct vclock *vclock)
 {
 	struct vy_env *env = vy_env(engine);
 
 	/* Cleanup old metadata log files. */
-	vy_log_collect_garbage(lsn);
+	vy_log_collect_garbage(vclock);
 
 	/* Cleanup run files. */
-	int64_t signature = vy_log_signature();
-	struct vy_recovery *recovery = vy_recovery_new(signature, 0);
+	struct vy_recovery *recovery = vy_recovery_new(vy_log_signature(), 0);
 	if (recovery == NULL) {
 		say_error("failed to recover vylog for garbage collection");
 		return 0;
 	}
-	vy_gc(env, recovery, VY_GC_DROPPED, lsn);
+	vy_gc(env, recovery, VY_GC_DROPPED, vclock_sum(vclock));
 	vy_recovery_delete(recovery);
 	return 0;
 }
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index 8a8f9d70..c9e0713c 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -202,22 +202,16 @@ vy_log_filename(int64_t signature)
 }
 
 /**
- * Return the lsn of the checkpoint that was taken
- * before the given lsn.
+ * Return the vclock of the checkpoint that was taken
+ * before the given vclock.
  */
-static int64_t
-vy_log_prev_checkpoint(int64_t lsn)
+static const struct vclock *
+vy_log_prev_checkpoint(const struct vclock *vclock)
 {
-	int64_t ret = -1;
-	for (struct vclock *vclock = vclockset_last(&vy_log.dir.index);
-	     vclock != NULL;
-	     vclock = vclockset_prev(&vy_log.dir.index, vclock)) {
-		if (vclock_sum(vclock) < lsn) {
-			ret = vclock_sum(vclock);
-			break;
-		}
-	}
-	return ret;
+	struct vclock *prev = vclockset_psearch(&vy_log.dir.index, vclock);
+	if (prev != NULL && vclock_sum(prev) == vclock_sum(vclock))
+		prev = vclockset_prev(&vy_log.dir.index, prev);
+	return prev;
 }
 
 /** An snprint-style function to print a log record. */
@@ -1076,14 +1070,16 @@ fail:
 }
 
 void
-vy_log_collect_garbage(int64_t signature)
+vy_log_collect_garbage(const struct vclock *vclock)
 {
 	/*
 	 * Always keep the previous file, because
 	 * it is still needed for backups.
 	 */
-	signature = vy_log_prev_checkpoint(signature);
-	xdir_collect_garbage(&vy_log.dir, signature, XDIR_GC_USE_COIO);
+	vclock = vy_log_prev_checkpoint(vclock);
+	if (vclock == NULL)
+		return;
+	xdir_collect_garbage(&vy_log.dir, vclock_sum(vclock), XDIR_GC_USE_COIO);
 }
 
 int64_t
@@ -1099,10 +1095,10 @@ vy_log_backup_path(const struct vclock *vclock)
 	 * Use the previous log file, because the current one
 	 * contains records written after the last checkpoint.
 	 */
-	int64_t lsn = vy_log_prev_checkpoint(vclock_sum(vclock));
-	if (lsn < 0)
+	vclock = vy_log_prev_checkpoint(vclock);
+	if (vclock == NULL)
 		return NULL;
-	const char *path = vy_log_filename(lsn);
+	const char *path = vy_log_filename(vclock_sum(vclock));
 	if (access(path, F_OK) == -1 && errno == ENOENT)
 		return NULL; /* vinyl not used */
 	return path;
diff --git a/src/box/vy_log.h b/src/box/vy_log.h
index 7718d9c6..70e25245 100644
--- a/src/box/vy_log.h
+++ b/src/box/vy_log.h
@@ -455,10 +455,10 @@ vy_log_rotate(const struct vclock *vclock);
 
 /**
  * Remove metadata log files that are not needed to recover
- * from the snapshot with the given signature or newer.
+ * from checkpoint @vclock.
  */
 void
-vy_log_collect_garbage(int64_t signature);
+vy_log_collect_garbage(const struct vclock *vclock);
 
 /**
  * Return the signature of the newest vylog to the time.
diff --git a/src/box/wal.c b/src/box/wal.c
index 673762a6..cbf2c121 100644
--- a/src/box/wal.c
+++ b/src/box/wal.c
@@ -120,11 +120,11 @@ struct wal_writer
 	 */
 	struct vclock vclock;
 	/**
-	 * Signature of the oldest checkpoint available on the instance.
+	 * VClock of the oldest checkpoint available on the instance.
 	 * The WAL writer must not delete WAL files that are needed to
 	 * recover from it even if it is running out of disk space.
 	 */
-	int64_t checkpoint_lsn;
+	struct vclock checkpoint_vclock;
 	/** The current WAL file. */
 	struct xlog current_wal;
 	/**
@@ -295,7 +295,8 @@ static void
 wal_writer_create(struct wal_writer *writer, enum wal_mode wal_mode,
 		  const char *wal_dirname, int64_t wal_max_rows,
 		  int64_t wal_max_size, const struct tt_uuid *instance_uuid,
-		  const struct vclock *vclock, int64_t checkpoint_lsn)
+		  const struct vclock *vclock,
+		  const struct vclock *checkpoint_vclock)
 {
 	writer->wal_mode = wal_mode;
 	writer->wal_max_rows = wal_max_rows;
@@ -311,11 +312,8 @@ wal_writer_create(struct wal_writer *writer, enum wal_mode wal_mode,
 	stailq_create(&writer->rollback);
 	cmsg_init(&writer->in_rollback, NULL);
 
-	/* Create and fill writer->vclock. */
-	vclock_create(&writer->vclock);
 	vclock_copy(&writer->vclock, vclock);
-
-	writer->checkpoint_lsn = checkpoint_lsn;
+	vclock_copy(&writer->checkpoint_vclock, checkpoint_vclock);
 	rlist_create(&writer->watchers);
 }
 
@@ -421,14 +419,14 @@ wal_open(struct wal_writer *writer)
 int
 wal_init(enum wal_mode wal_mode, const char *wal_dirname, int64_t wal_max_rows,
 	 int64_t wal_max_size, const struct tt_uuid *instance_uuid,
-	 const struct vclock *vclock, int64_t first_checkpoint_lsn)
+	 const struct vclock *vclock, const struct vclock *first_checkpoint_vclock)
 {
 	assert(wal_max_rows > 1);
 
 	struct wal_writer *writer = &wal_writer_singleton;
 	wal_writer_create(writer, wal_mode, wal_dirname, wal_max_rows,
 			  wal_max_size, instance_uuid, vclock,
-			  first_checkpoint_lsn);
+			  first_checkpoint_vclock);
 
 	/*
 	 * Scan the WAL directory to build an index of all
@@ -546,8 +544,8 @@ wal_checkpoint(struct vclock *vclock, bool rotate)
 struct wal_gc_msg
 {
 	struct cbus_call_msg base;
-	int64_t wal_lsn;
-	int64_t checkpoint_lsn;
+	const struct vclock *wal_vclock;
+	const struct vclock *checkpoint_vclock;
 };
 
 static int
@@ -555,20 +553,21 @@ wal_collect_garbage_f(struct cbus_call_msg *data)
 {
 	struct wal_writer *writer = &wal_writer_singleton;
 	struct wal_gc_msg *msg = (struct wal_gc_msg *)data;
-	writer->checkpoint_lsn = msg->checkpoint_lsn;
-	xdir_collect_garbage(&writer->wal_dir, msg->wal_lsn, 0);
+	vclock_copy(&writer->checkpoint_vclock, msg->checkpoint_vclock);
+	xdir_collect_garbage(&writer->wal_dir, vclock_sum(msg->wal_vclock), 0);
 	return 0;
 }
 
 void
-wal_collect_garbage(int64_t wal_lsn, int64_t checkpoint_lsn)
+wal_collect_garbage(const struct vclock *wal_vclock,
+		    const struct vclock *checkpoint_vclock)
 {
 	struct wal_writer *writer = &wal_writer_singleton;
 	if (writer->wal_mode == WAL_NONE)
 		return;
 	struct wal_gc_msg msg;
-	msg.wal_lsn = wal_lsn;
-	msg.checkpoint_lsn = checkpoint_lsn;
+	msg.wal_vclock = wal_vclock;
+	msg.checkpoint_vclock = checkpoint_vclock;
 	bool cancellable = fiber_set_cancellable(false);
 	cbus_call(&wal_thread.wal_pipe, &wal_thread.tx_prio_pipe, &msg.base,
 		  wal_collect_garbage_f, NULL, TIMEOUT_INFINITY);
@@ -643,6 +642,12 @@ wal_fallocate(struct wal_writer *writer, size_t len)
 	struct errinj *errinj = errinj(ERRINJ_WAL_FALLOCATE, ERRINJ_INT);
 
 	/*
+	 * Max LSN that can be collected in case of ENOSPC -
+	 * we must not delete WALs necessary for recovery.
+	 */
+	int64_t gc_lsn = vclock_sum(&writer->checkpoint_vclock);
+
+	/*
 	 * The actual write size can be greater than the sum size
 	 * of encoded rows (compression, fixheaders). Double the
 	 * given length to get a rough upper bound estimate.
@@ -662,7 +667,7 @@ retry:
 	}
 	if (errno != ENOSPC)
 		goto error;
-	if (!xdir_has_garbage(&writer->wal_dir, writer->checkpoint_lsn))
+	if (!xdir_has_garbage(&writer->wal_dir, gc_lsn))
 		goto error;
 
 	if (warn_no_space) {
@@ -674,7 +679,7 @@ retry:
 	struct diag diag;
 	diag_create(&diag);
 	diag_move(diag_get(), &diag);
-	if (xdir_collect_garbage(&writer->wal_dir, writer->checkpoint_lsn,
+	if (xdir_collect_garbage(&writer->wal_dir, gc_lsn,
 				 XDIR_GC_REMOVE_ONE) != 0) {
 		diag_move(&diag, diag_get());
 		goto error;
diff --git a/src/box/wal.h b/src/box/wal.h
index d4a37c55..808af76e 100644
--- a/src/box/wal.h
+++ b/src/box/wal.h
@@ -58,7 +58,7 @@ wal_thread_start();
 int
 wal_init(enum wal_mode wal_mode, const char *wal_dirname, int64_t wal_max_rows,
 	 int64_t wal_max_size, const struct tt_uuid *instance_uuid,
-	 const struct vclock *vclock, int64_t first_checkpoint_lsn);
+	 const struct vclock *vclock, const struct vclock *first_checkpoint_vclock);
 
 void
 wal_thread_stop();
@@ -175,14 +175,15 @@ int
 wal_checkpoint(struct vclock *vclock, bool rotate);
 
 /**
- * Remove all WAL files whose signature is less than @wal_lsn.
- * Update the oldest checkpoint signature with @checkpoint_lsn.
+ * Remove all WAL files whose signature is less than @wal_vclock.
+ * Update the oldest checkpoint signature with @checkpoint_vclock.
  * WAL thread will delete WAL files that are not needed to
  * recover from the oldest checkpoint if it runs out of disk
  * space.
  */
 void
-wal_collect_garbage(int64_t wal_lsn, int64_t checkpoint_lsn);
+wal_collect_garbage(const struct vclock *wal_vclock,
+		    const struct vclock *checkpoint_vclock);
 
 void
 wal_init_vy_log();
-- 
2.11.0