[Tarantool-patches] [PATCH 13/16] tx: introduce txm_story

[Vladislav Shpilevoy]

Thanks for the patch!

See 3 comments below.

On 08.07.2020 17:14, Aleksandr Lyapunov wrote:
> ---
>  src/box/txn.c | 725 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
>  src/box/txn.h | 127 ++++++++++
>  2 files changed, 851 insertions(+), 1 deletion(-)
> 
> diff --git a/src/box/txn.c b/src/box/txn.c
> index b9f1737..78c542f 100644
> --- a/src/box/txn.c
> +++ b/src/box/txn.c
> @@ -1095,4 +1141,681 @@ txm_cause_conflict(struct txn *wreaker, struct txn *victim)
>  	rlist_add(&wreaker->conflict_list, &tracker->in_conflict_list);
>  	rlist_add(&wreaker->conflicted_by_list, &tracker->in_conflicted_by_list);
>  	return 0;
> -}
> \ No newline at end of file
> +}
> +
> +struct txm_story *
> +txm_story_new(struct tuple *tuple, uint32_t index_count)
> +{
> +	assert(!tuple->is_dirty);
> +	assert(index_count < BOX_INDEX_MAX);
> +	struct mempool *pool = &txm.txm_story_pool[index_count];
> +	struct txm_story *story = (struct txm_story *)mempool_alloc(pool);
> +	if (story == NULL) {
> +		size_t item_size = sizeof(struct txm_story) +
> +			index_count * sizeof(struct txm_story_link);
> +		diag_set(OutOfMemory, item_size,
> +			 "tx_manager", "tx story");
> +		return story;

1. What if index count of the space will change in a concurrect
DDL transaction?

> +	}
> +	story->tuple = tuple;
> +
> +	const struct txm_story **put_story = (const struct txm_story **)&story;
> +	struct txm_story **empty = NULL;
> +	mh_int_t pos = mh_history_put(txm.history, put_story, &empty, 0);
> +	if (pos == mh_end(txm.history)) {
> +		mempool_free(pool, story);
> +		diag_set(OutOfMemory, pos + 1,
> +			 "tx_manager", "tx history hash table");
> +		return NULL;
> +	}
> +	tuple->is_dirty = true;
> +	tuple_ref(tuple);
> +
> +	story->index_count = index_count;
> +	story->add_stmt = NULL;
> +	story->add_psn = 0;
> +	story->del_stmt = NULL;
> +	story->del_psn = 0;
> +	rlist_create(&story->reader_list);
> +	rlist_add(&txm.all_stories, &story->in_all_stories);
> +	memset(story->link, 0, sizeof(story->link[0]) * index_count);
> +	return story;
> +}
> +
> +static struct txm_story *
> +txm_story_new_add_stmt(struct tuple *tuple, struct txn_stmt *stmt)

2. Probably would be better to use the existing terms: old_tuple and new_tuple.
So the functions would be txm_story_add_new_stmt and txm_story_add_old_stmt. The
same in all the other places. 'add_stmt' -> 'new_stmt'. 'del_stmt' -> 'old_stmt'.

> +{
> +	struct txm_story *res = txm_story_new(tuple, stmt->space->index_count);
> +	if (res == NULL)
> +		return NULL;
> +	res->add_stmt = stmt;
> +	assert(stmt->add_story == NULL);
> +	stmt->add_story = res;
> +	return res;
> +}
> +
> +static struct txm_story *
> +txm_story_new_del_stmt(struct tuple *tuple, struct txn_stmt *stmt)
> +{
> +	struct txm_story *res = txm_story_new(tuple, stmt->space->index_count);
> +	if (res == NULL)
> +		return NULL;
> +	res->del_stmt = stmt;
> +	assert(stmt->del_story == NULL);
> +	stmt->del_story = res;
> +	return res;
> +}
> +
> +void
> +txm_history_prepare_stmt(struct txn_stmt *stmt)
> +{
> +	assert(stmt->txn->psn != 0);
> +
> +	/* Move story to the past to prepared stories. */
> +
> +	struct txm_story *story = stmt->add_story;
> +	uint32_t index_count = story == NULL ? 0 : story->index_count;
> +	for (uint32_t i = 0; i < index_count; ) {
> +		if (!story->link[i].older.is_story) {
> +			/* tuple is old. */
> +			i++;
> +			continue;
> +		}
> +		struct txm_story *old_story =
> +			story->link[i].older.story;
> +		if (old_story->del_psn != 0) {
> +			/* is psn is set, the change is prepared. */
> +			i++;
> +			continue;
> +		}
> +		if (old_story->add_psn != 0) {
> +			/* is psn is set, the change is prepared. */
> +			i++;
> +			continue;
> +		}
> +
> +		if (old_story->add_stmt != NULL) {
> +			/* ancient */
> +			i++;
> +			continue;
> +		}
> +		if (old_story->add_stmt->txn == stmt->txn) {
> +			/* added by us. */
> +			i++;
> +			continue;
> +		}
> +
> +		if (old_story->add_stmt->preserve_old_tuple || i != 0)
> +			old_story->add_stmt->txn->status = TXN_CONFLICTED;
> +
> +		/* Swap story and old story. */
> +		struct txm_story_link *link = &story->link[i];
> +		if (link->newer_story == NULL) {
> +			/* we have to replace the tuple in index. */
> +			struct tuple *unused;
> +			struct index *index = stmt->space->index[i];
> +			if (index_replace(index, story->tuple,
> +			                  old_story->tuple,
> +			                  DUP_INSERT, &unused) != 0) {
> +				diag_log();
> +				unreachable();
> +				panic("failed to rollback change");
> +			}
> +		} else {
> +			struct txm_story *newer = link->newer_story;
> +			assert(newer->link[i].older.is_story);
> +			assert(newer->link[i].older.story == story);
> +			txm_story_unlink(newer, i);
> +			txm_story_link_story(newer, old_story, i);
> +		}
> +
> +		txm_story_unlink(story, i);
> +
> +		txm_story_unlink(story, i);
> +		if (old_story->link[i].older.is_story) {
> +			struct txm_story *to =
> +				old_story->link[i].older.story;
> +			txm_story_unlink(old_story, i);
> +			txm_story_link_story(story, to, i);
> +		} else {
> +			struct tuple *to =
> +				old_story->link[i].older.tuple;
> +			txm_story_unlink(old_story, i);
> +			txm_story_link_tuple(story, to, i);
> +		}
> +
> +		txm_story_link_story(old_story, story, i);
> +
> +		if (i == 0) {
> +			assert(stmt->del_story == old_story);
> +			assert(story->link[0].older.is_story ||
> +			       story->link[0].older.tuple == NULL);
> +
> +			struct txn_stmt *dels = old_story->del_stmt;
> +			assert(dels != NULL);
> +			do {
> +				if (dels->txn != stmt->txn)
> +					dels->txn->status = TXN_CONFLICTED;
> +				dels->del_story = NULL;
> +				struct txn_stmt *next = dels->next_in_del_list;
> +				dels->next_in_del_list = NULL;
> +				dels = next;
> +			} while (dels != NULL);
> +			old_story->del_stmt = NULL;
> +
> +			if (story->link[0].older.is_story) {
> +				struct txm_story *oldest_story =
> +					story->link[0].older.story;
> +				dels = oldest_story->del_stmt;
> +				while (dels != NULL) {
> +					assert(dels->txn != stmt->txn);
> +					dels->del_story = NULL;
> +					struct txn_stmt *next =
> +						dels->next_in_del_list;
> +					dels->next_in_del_list = NULL;
> +					dels = next;
> +				}
> +				oldest_story->del_stmt = stmt;
> +				stmt->del_story = oldest_story;
> +			}
> +		}
> +	}
> +	if (stmt->add_story != NULL)

3. According to a check in the beginning of the function stmt can
be NULL. But then this will crash.

> +		stmt->add_story->add_psn = stmt->txn->psn;
> +
> +	if (stmt->del_story != NULL)
> +		stmt->del_story->del_psn = stmt->txn->psn;
> +}