[PATCH 3/4] vinyl: store lsn of index drop record in vylog
Vladimir Davydov
vdavydov.dev at gmail.com
Wed May 23 19:10:06 MSK 2018
This is required to rework garbage collection in vinyl.
---
src/box/vinyl.c | 7 +++----
src/box/vy_log.c | 43 ++++++++++++++++++++++++++++++++-----------
src/box/vy_log.h | 17 +++++++++++++----
src/box/vy_lsm.c | 2 +-
4 files changed, 49 insertions(+), 20 deletions(-)
diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index a423e95b..63f1c5ff 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -931,7 +931,6 @@ vy_log_lsm_prune(struct vy_lsm *lsm, int64_t gc_lsn)
static void
vinyl_index_commit_drop(struct index *index, int64_t lsn)
{
- (void)lsn;
struct vy_env *env = vy_env(index->engine);
struct vy_lsm *lsm = vy_lsm(index);
@@ -952,7 +951,7 @@ vinyl_index_commit_drop(struct index *index, int64_t lsn)
vy_log_tx_begin();
vy_log_lsm_prune(lsm, checkpoint_last(NULL));
- vy_log_drop_lsm(lsm->id);
+ vy_log_drop_lsm(lsm->id, lsn);
vy_log_tx_try_commit();
}
@@ -3138,7 +3137,7 @@ vy_send_lsm(struct vy_join_ctx *ctx, struct vy_lsm_recovery_info *lsm_info)
{
int rc = -1;
- if (lsm_info->is_dropped)
+ if (lsm_info->drop_lsn >= 0)
return 0;
/*
@@ -3429,7 +3428,7 @@ vinyl_engine_backup(struct engine *engine, struct vclock *vclock,
int loops = 0;
struct vy_lsm_recovery_info *lsm_info;
rlist_foreach_entry(lsm_info, &recovery->lsms, in_recovery) {
- if (lsm_info->is_dropped)
+ if (lsm_info->drop_lsn >= 0)
continue;
struct vy_run_recovery_info *run_info;
rlist_foreach_entry(run_info, &lsm_info->runs, in_lsm) {
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index d3c2bf6c..4e459459 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -82,6 +82,7 @@ enum vy_log_key {
VY_LOG_KEY_TRUNCATE_COUNT = 11,
VY_LOG_KEY_CREATE_LSN = 12,
VY_LOG_KEY_MODIFY_LSN = 13,
+ VY_LOG_KEY_DROP_LSN = 14,
};
/** vy_log_key -> human readable name. */
@@ -100,6 +101,7 @@ static const char *vy_log_key_name[] = {
[VY_LOG_KEY_TRUNCATE_COUNT] = "truncate_count",
[VY_LOG_KEY_CREATE_LSN] = "create_lsn",
[VY_LOG_KEY_MODIFY_LSN] = "modify_lsn",
+ [VY_LOG_KEY_DROP_LSN] = "drop_lsn",
};
/** vy_log_type -> human readable name. */
@@ -258,6 +260,10 @@ vy_log_record_snprint(char *buf, int size, const struct vy_log_record *record)
SNPRINT(total, snprintf, buf, size, "%s=%"PRIi64", ",
vy_log_key_name[VY_LOG_KEY_MODIFY_LSN],
record->modify_lsn);
+ if (record->drop_lsn > 0)
+ SNPRINT(total, snprintf, buf, size, "%s=%"PRIi64", ",
+ vy_log_key_name[VY_LOG_KEY_DROP_LSN],
+ record->drop_lsn);
if (record->dump_lsn > 0)
SNPRINT(total, snprintf, buf, size, "%s=%"PRIi64", ",
vy_log_key_name[VY_LOG_KEY_DUMP_LSN],
@@ -372,6 +378,11 @@ vy_log_record_encode(const struct vy_log_record *record,
size += mp_sizeof_uint(record->modify_lsn);
n_keys++;
}
+ if (record->drop_lsn > 0) {
+ size += mp_sizeof_uint(VY_LOG_KEY_DROP_LSN);
+ size += mp_sizeof_uint(record->drop_lsn);
+ n_keys++;
+ }
if (record->dump_lsn > 0) {
size += mp_sizeof_uint(VY_LOG_KEY_DUMP_LSN);
size += mp_sizeof_uint(record->dump_lsn);
@@ -448,6 +459,10 @@ vy_log_record_encode(const struct vy_log_record *record,
pos = mp_encode_uint(pos, VY_LOG_KEY_MODIFY_LSN);
pos = mp_encode_uint(pos, record->modify_lsn);
}
+ if (record->drop_lsn > 0) {
+ pos = mp_encode_uint(pos, VY_LOG_KEY_DROP_LSN);
+ pos = mp_encode_uint(pos, record->drop_lsn);
+ }
if (record->dump_lsn > 0) {
pos = mp_encode_uint(pos, VY_LOG_KEY_DUMP_LSN);
pos = mp_encode_uint(pos, record->dump_lsn);
@@ -571,6 +586,9 @@ vy_log_record_decode(struct vy_log_record *record,
case VY_LOG_KEY_MODIFY_LSN:
record->modify_lsn = mp_decode_uint(&pos);
break;
+ case VY_LOG_KEY_DROP_LSN:
+ record->drop_lsn = mp_decode_uint(&pos);
+ break;
case VY_LOG_KEY_DUMP_LSN:
record->dump_lsn = mp_decode_uint(&pos);
break;
@@ -1242,9 +1260,9 @@ vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
lsm->index_id = index_id;
lsm->key_parts = NULL;
lsm->key_part_count = 0;
- lsm->is_dropped = false;
lsm->create_lsn = -1;
lsm->modify_lsn = -1;
+ lsm->drop_lsn = -1;
lsm->dump_lsn = -1;
rlist_create(&lsm->ranges);
rlist_create(&lsm->runs);
@@ -1286,7 +1304,7 @@ vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
}
struct vy_lsm_recovery_info *lsm;
lsm = vy_recovery_lsm_by_index_id(recovery, space_id, index_id);
- if (lsm != NULL && !lsm->is_dropped) {
+ if (lsm != NULL && lsm->drop_lsn < 0) {
diag_set(ClientError, ER_INVALID_VYLOG_FILE,
tt_sprintf("LSM tree %u/%u created twice",
(unsigned)space_id, (unsigned)index_id));
@@ -1329,7 +1347,7 @@ vy_recovery_modify_lsm(struct vy_recovery *recovery, int64_t id,
(long long)id));
return -1;
}
- if (lsm->is_dropped) {
+ if (lsm->drop_lsn >= 0) {
diag_set(ClientError, ER_INVALID_VYLOG_FILE,
tt_sprintf("Update of deleted LSM tree %lld",
(long long)id));
@@ -1355,7 +1373,7 @@ vy_recovery_modify_lsm(struct vy_recovery *recovery, int64_t id,
* Returns 0 on success, -1 if ID not found or LSM tree is already marked.
*/
static int
-vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id)
+vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id, int64_t drop_lsn)
{
struct vy_lsm_recovery_info *lsm;
lsm = vy_recovery_lookup_lsm(recovery, id);
@@ -1365,7 +1383,7 @@ vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id)
(long long)id));
return -1;
}
- if (lsm->is_dropped) {
+ if (lsm->drop_lsn >= 0) {
diag_set(ClientError, ER_INVALID_VYLOG_FILE,
tt_sprintf("LSM tree %lld deleted twice",
(long long)id));
@@ -1386,7 +1404,8 @@ vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id)
return -1;
}
}
- lsm->is_dropped = true;
+ assert(drop_lsn >= 0);
+ lsm->drop_lsn = drop_lsn;
return 0;
}
@@ -1408,7 +1427,7 @@ vy_recovery_dump_lsm(struct vy_recovery *recovery,
(long long)id));
return -1;
}
- if (lsm->is_dropped) {
+ if (lsm->drop_lsn >= 0) {
diag_set(ClientError, ER_INVALID_VYLOG_FILE,
tt_sprintf("Dump of deleted LSM tree %lld",
(long long)id));
@@ -1508,7 +1527,7 @@ vy_recovery_create_run(struct vy_recovery *recovery, int64_t lsm_id,
(long long)lsm_id));
return -1;
}
- if (lsm->is_dropped) {
+ if (lsm->drop_lsn >= 0) {
diag_set(ClientError, ER_INVALID_VYLOG_FILE,
tt_sprintf("Run %lld created for deleted "
"LSM tree %lld", (long long)run_id,
@@ -1829,7 +1848,8 @@ vy_recovery_process_record(struct vy_recovery *recovery,
record->modify_lsn);
break;
case VY_LOG_DROP_LSM:
- rc = vy_recovery_drop_lsm(recovery, record->lsm_id);
+ rc = vy_recovery_drop_lsm(recovery, record->lsm_id,
+ record->drop_lsn);
break;
case VY_LOG_INSERT_RANGE:
rc = vy_recovery_insert_range(recovery, record->lsm_id,
@@ -2132,10 +2152,11 @@ vy_log_append_lsm(struct xlog *xlog, struct vy_lsm_recovery_info *lsm)
}
}
- if (lsm->is_dropped) {
+ if (lsm->drop_lsn >= 0) {
vy_log_record_init(&record);
record.type = VY_LOG_DROP_LSM;
record.lsm_id = lsm->id;
+ record.drop_lsn = lsm->drop_lsn;
if (vy_log_append_record(xlog, &record) != 0)
return -1;
}
@@ -2162,7 +2183,7 @@ vy_log_create(const struct vclock *vclock, struct vy_recovery *recovery)
* (and thus not needed for garbage collection) from the
* log on rotation.
*/
- if (lsm->is_dropped && rlist_empty(&lsm->runs))
+ if (lsm->drop_lsn >= 0 && rlist_empty(&lsm->runs))
continue;
/* Create the log file on the first write. */
diff --git a/src/box/vy_log.h b/src/box/vy_log.h
index 1b2b419f..22c31825 100644
--- a/src/box/vy_log.h
+++ b/src/box/vy_log.h
@@ -71,7 +71,7 @@ enum vy_log_record_type {
VY_LOG_CREATE_LSM = 0,
/**
* Drop an LSM tree.
- * Requires vy_log_record::lsm_id.
+ * Requires vy_log_record::lsm_id, drop_lsn.
*/
VY_LOG_DROP_LSM = 1,
/**
@@ -209,6 +209,11 @@ struct vy_log_record {
int64_t create_lsn;
/** LSN of the WAL row that last modified the LSM tree. */
int64_t modify_lsn;
+ /**
+ * LSN of the WAL row that dropped the LSM tree or -1
+ * if the tree is still active.
+ */
+ int64_t drop_lsn;
/** Max LSN stored on disk. */
int64_t dump_lsn;
/**
@@ -258,12 +263,15 @@ struct vy_lsm_recovery_info {
struct key_part_def *key_parts;
/** Number of key parts. */
uint32_t key_part_count;
- /** True if the LSM tree was dropped. */
- bool is_dropped;
/** LSN of the WAL row that created the LSM tree. */
int64_t create_lsn;
/** LSN of the WAL row that last modified the LSM tree. */
int64_t modify_lsn;
+ /**
+ * LSN of the WAL row that dropped the LSM tree or -1
+ * if the tree is still active.
+ */
+ int64_t drop_lsn;
/** LSN of the last LSM tree dump. */
int64_t dump_lsn;
/**
@@ -540,12 +548,13 @@ vy_log_modify_lsm(int64_t id, const struct key_def *key_def, int64_t modify_lsn)
/** Helper to log a vinyl LSM tree drop. */
static inline void
-vy_log_drop_lsm(int64_t id)
+vy_log_drop_lsm(int64_t id, int64_t drop_lsn)
{
struct vy_log_record record;
vy_log_record_init(&record);
record.type = VY_LOG_DROP_LSM;
record.lsm_id = id;
+ record.drop_lsn = drop_lsn;
vy_log_write(&record);
}
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index edc3b1a4..289d5c40 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -543,7 +543,7 @@ vy_lsm_recover(struct vy_lsm *lsm, struct vy_recovery *recovery,
lsm->id = lsm_info->id;
lsm->commit_lsn = lsm_info->modify_lsn;
- if (lsn < lsm_info->create_lsn || lsm_info->is_dropped) {
+ if (lsn < lsm_info->create_lsn || lsm_info->drop_lsn >= 0) {
/*
* Loading a past incarnation of the LSM tree, i.e.
* the LSM tree is going to dropped during final
--
2.11.0
More information about the Tarantool-patches
mailing list