memtx_tx.c

/*
 * Copyright 2010-2020, Tarantool AUTHORS, please see AUTHORS file.
 *
 * Redistribution and use in source and binary forms, with or
 * without modification, are permitted provided that the following
 * conditions are met:
 *
 * 1. Redistributions of source code must retain the above
 *    copyright notice, this list of conditions and the
 *    following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above
 *    copyright notice, this list of conditions and the following
 *    disclaimer in the documentation and/or other materials
 *    provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
 * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
 * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
 * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
#include "memtx_tx.h"

#include <assert.h>
#include <limits.h>
#include <stddef.h>
#include <stdint.h>

#include "schema_def.h"
#include "small/mempool.h"

enum
{
	/**
	 * Virtual PSN that will be set to del_psn of rolled-back story.
	 * Must be less that any existing "real" PSN.
	 */
	MEMTX_TX_ROLLBACKED_PSN = 1,
};

static_assert((int)MEMTX_TX_ROLLBACKED_PSN < (int)TXN_MIN_PSN,
	      "There must be a range for TX manager's internal use");

/**
 * Link that connects a memtx_story with older and newer stories of the same
 * key in index.
 */
struct memtx_story_link {
	/** Story that was happened after that story was ended. */
	struct memtx_story *newer_story;
	/** Story that was happened before that story was started. */
	struct memtx_story *older_story;
	/** List of interval items @sa gap_item. */
	struct rlist nearby_gaps;
	/**
	 * If the tuple of story is physically in index, here the pointer
	 * to that index is stored.
	 */
	struct index *in_index;
};

/**
 * A part of a history of a value in space.
 * It's a story about a tuple, from the point it was added to space to the
 * point when it was deleted from a space.
 * All stories are linked into a list of stories of the same key of each index.
 */
struct memtx_story {
	/** The story is about this tuple. The tuple is referenced. */
	struct tuple *tuple;
	/**
	 * Statement that introduced this story. Is set to NULL when the
	 * statement's transaction becomes committed. Can also be NULL if we
	 * don't know who introduced that story, the tuple was added by a
	 * transaction that was completed and destroyed some time ago.
	 */
	struct txn_stmt *add_stmt;
	/**
	 * Prepare sequence number of add_stmt's transaction. Is set when
	 * the transaction is prepared. Can be 0 if the transaction is
	 * in progress or we don't know who introduced that story.
	 */
	int64_t add_psn;
	/**
	 * Statement that ended this story. Is set to NULL when the statement's
	 * transaction becomes committed. Can also be NULL if the tuple has not
	 * been deleted yet.
	 */
	struct txn_stmt *del_stmt;
	/**
	 * Prepare sequence number of del_stmt's transaction. Is set when
	 * the transaction is prepared. Can be 0 if the transaction is
	 * in progress or if nobody has deleted the tuple.
	 */
	int64_t del_psn;
	/**
	 * List of trackers - transactions that has read this tuple.
	 */
	struct rlist reader_list;
	/**
	 * Link in tx_manager::all_stories
	 */
	struct rlist in_all_stories;
	/**
	 * Link in space::memtx_tx_stories.
	 */
	struct rlist in_space_stories;
	/**
	 * Number of indexes in this space - and the count of link[].
	 */
	uint32_t index_count;
	/**
	 * Status of story, describes the reason why story cannot be deleted.
	 * It is initialized in memtx_story constructor and is changed only in
	 * memtx_tx_story_gc.
	 */
	enum memtx_tx_story_status status;
	/**
	 * Flag is set when @a tuple is not placed in primary key and
	 * the story is the only reason why @a tuple cannot be deleted.
	 */
	bool tuple_is_retained;
	/**
	 * Link with older and newer stories (and just tuples) for each
	 * index respectively.
	 */
	struct memtx_story_link link[];
};

static uint32_t
memtx_tx_story_key_hash(const struct tuple *a)
{
	uintptr_t u = (uintptr_t)a;
	if (sizeof(uintptr_t) <= sizeof(uint32_t))
		return u;
	else
		return u ^ (u >> 32);
}

#define mh_name _history
#define mh_key_t struct tuple *
#define mh_node_t struct memtx_story *
#define mh_arg_t int
#define mh_hash(a, arg) (memtx_tx_story_key_hash((*(a))->tuple))
#define mh_hash_key(a, arg) (memtx_tx_story_key_hash(a))
#define mh_cmp(a, b, arg) ((*(a))->tuple != (*(b))->tuple)
#define mh_cmp_key(a, b, arg) ((a) != (*(b))->tuple)
#define MH_SOURCE
#include "salad/mhash.h"

/**
 * Record that links transaction and a story that the transaction have read.
 */
struct tx_read_tracker {
	/** The TX that read story. */
	struct txn *reader;
	/** The story that was read by reader. */
	struct memtx_story *story;
	/** Link in story->reader_list. */
	struct rlist in_reader_list;
	/** Link in reader->read_set. */
	struct rlist in_read_set;
	/** Bit field of indexes in which the data was read by reader. */
	uint64_t index_mask;
};

/**
 * An element that stores the fact that some transaction have read
 * a full key and found nothing.
 */
struct point_hole_item {
	/** A link of headless list of items with the same index and key. */
	struct rlist ring;
	/** Link in txn->point_holes_list. */
	struct rlist in_point_holes_list;
	/** Saved index->unique_id. */
	uint32_t index_unique_id;
	/** Precalculated hash for storing in hash table.. */
	uint32_t hash;
	/** Saved txn. */
	struct txn *txn;
	/** Saved key. Points to @a short_key or allocated in txn's region. */
	const char *key;
	/** Saved key len. */
	size_t key_len;
	/** Storage for short key. @key may point here. */
	char short_key[16];
	/** Flag that the hash tables stores pointer to this item. */
	bool is_head;
};

/**
 * An element that stores the fact that some transaction have read
 * a full key and found nothing.
 */
struct gap_item {
	/** A link in memtx_story_link::nearby_gaps OR index::nearby_gaps. */
	struct rlist in_nearby_gaps;
	/** Link in txn->gap_list. */
	struct rlist in_gap_list;
	/** The transaction that read it. */
	struct txn *txn;
	/** The key. Can be NULL. */
	const char *key;
	uint32_t key_len;
	uint32_t part_count;
	/** Search mode. */
	enum iterator_type type;
	/** Storage for short key. @key may point here. */
	char short_key[16];
};

/**
 * An element that stores the fact that some transaction have read
 * a full index.
 */
struct full_scan_item {
	/** A link in index::full_scans. */
	struct rlist in_full_scans;
	/** Link in txn->full_scan_list. */
	struct rlist in_full_scan_list;
	/** The transaction that read it. */
	struct txn *txn;
};

/**
 * Helper structure for searching for point_hole_item in the hash table,
 * @sa point_hole_item_pool.
 */
struct point_hole_key {
	struct index *index;
	struct tuple *tuple;
};

/** Hash calculatore for the key. */
static uint32_t
point_hole_storage_key_hash(struct point_hole_key *key)
{
	struct key_def *def = key->index->def->key_def;
	return key->index->unique_id ^ def->tuple_hash(key->tuple, def);
}

/** point_hole_item comparator. */
static int
point_hole_storage_equal(const struct point_hole_item *obj1,
			 const struct point_hole_item *obj2)
{
	/* Canonical msgpack is comparable by memcmp. */
	if (obj1->index_unique_id != obj2->index_unique_id ||
	    obj1->key_len != obj2->key_len)
		return 1;
	return memcmp(obj1->key, obj2->key, obj1->key_len) != 0;
}

/** point_hole_item comparator with key. */
static int
point_hole_storage_key_equal(const struct point_hole_key *key,
			     const struct point_hole_item *object)
{
	if (key->index->unique_id != object->index_unique_id)
		return 1;
	assert(key->index != NULL);
	assert(key->tuple != NULL);
	struct key_def *def = key->index->def->key_def;
	hint_t oh = def->key_hint(object->key, def->part_count, def);
	hint_t kh = def->tuple_hint(key->tuple, def);
	return def->tuple_compare_with_key(key->tuple, kh, object->key,
					   def->part_count, oh, def);
}

/**
 * Hash table definition for hole read storage.
 * The key is constructed by unique index ID and search key.
 * Actually it stores pointers to point_hole_item structures.
 * If more than one point_hole_item is added to the hash table,
 * it is simply added to the headless list in existing point_hole_item.
 */

#define mh_name _point_holes
#define mh_key_t struct point_hole_key *
#define mh_node_t struct point_hole_item *
#define mh_arg_t int
#define mh_hash(a, arg) ((*(a))->hash)
#define mh_hash_key(a, arg) ( point_hole_storage_key_hash(a) )
#define mh_cmp(a, b, arg) point_hole_storage_equal(*(a), *(b))
#define mh_cmp_key(a, b, arg) point_hole_storage_key_equal((a), *(b))
#define MH_SOURCE
#include "salad/mhash.h"

/**
 * Collect an allocation to memtx_tx_stats.
 */
static inline void
memtx_tx_stats_collect(struct memtx_tx_stats *stats, size_t size)
{
	stats->count++;
	stats->total += size;
}

/**
 * Discard an allocation collected by memtx_tx_stats.
 */
static inline void
memtx_tx_stats_discard(struct memtx_tx_stats *stats, size_t size)
{
	assert(stats->count > 0);
	assert(stats->total >= size);

	stats->count--;
	stats->total -= size;
}

/**
 * Collect allocation statistics.
 */
static inline void
memtx_tx_track_allocation(struct txn *txn, size_t size,
			  enum memtx_tx_alloc_type alloc_type)
{
	assert(alloc_type < MEMTX_TX_ALLOC_TYPE_MAX);
	txn->memtx_tx_alloc_stats[alloc_type] += size;
}

/**
 * Collect deallocation statistics.
 */
static inline void
memtx_tx_track_deallocation(struct txn *txn, size_t size,
			    enum memtx_tx_alloc_type alloc_type)
{
	assert(alloc_type < MEMTX_TX_ALLOC_TYPE_MAX);
	assert(txn->memtx_tx_alloc_stats[alloc_type] >= size);
	txn->memtx_tx_alloc_stats[alloc_type] -= size;
}

/**
 * A wrapper over mempool.
 * Use it instead of mempool to track allocations!
 */
struct memtx_tx_mempool {
	/**
	 * Wrapped mempool.
	 */
	struct mempool pool;
	/**
	 * Each allocation is accounted with this type.
	 */
	enum memtx_tx_alloc_type alloc_type;
};

static inline void
memtx_tx_mempool_create(struct memtx_tx_mempool *mempool, uint32_t objsize,
			enum memtx_tx_alloc_type alloc_type)
{
	mempool_create(&mempool->pool, cord_slab_cache(), objsize);
	mempool->alloc_type = alloc_type;
}

static inline void
memtx_tx_mempool_destroy(struct memtx_tx_mempool *mempool)
{
	mempool_destroy(&mempool->pool);
	mempool->alloc_type = MEMTX_TX_ALLOC_TYPE_MAX;
}

void *
memtx_tx_mempool_alloc(struct txn *txn, struct memtx_tx_mempool *mempool)
{
	void *allocation = mempool_alloc(&mempool->pool);
	if (allocation != NULL) {
		uint32_t size = mempool->pool.objsize;
		memtx_tx_track_allocation(txn, size, mempool->alloc_type);
	}
	return allocation;
}

void
memtx_tx_mempool_free(struct txn *txn, struct memtx_tx_mempool *mempool, void *ptr)
{
	uint32_t size = mempool->pool.objsize;
	memtx_tx_track_deallocation(txn, size, mempool->alloc_type);
	mempool_free(&mempool->pool, ptr);
}

/**
 * Choose memtx_tx_alloc_type for alloc_obj.
 */
static inline enum memtx_tx_alloc_type
memtx_tx_region_object_to_type(enum memtx_tx_alloc_object alloc_obj)
{
	enum memtx_tx_alloc_type alloc_type = MEMTX_TX_ALLOC_TYPE_MAX;
	switch (alloc_obj) {
	case MEMTX_TX_OBJECT_CONFLICT_TRACKER:
	case MEMTX_TX_OBJECT_READ_TRACKER:
		alloc_type = MEMTX_TX_ALLOC_TRACKER;
		break;
	default:
		unreachable();
	};
	assert(alloc_type < MEMTX_TX_ALLOC_TYPE_MAX);
	return alloc_type;
}

/**
 * Alloc object on region. Pass object as enum memtx_tx_alloc_object.
 * Use this method to track txn's allocations!
 */
static inline void *
memtx_tx_region_alloc_object(struct txn *txn,
			     enum memtx_tx_alloc_object alloc_obj)
{
	size_t size = 0;
	void *alloc = NULL;
	enum memtx_tx_alloc_type alloc_type =
		memtx_tx_region_object_to_type(alloc_obj);
	switch (alloc_obj) {
	case MEMTX_TX_OBJECT_READ_TRACKER:
		alloc = region_alloc_object(&txn->region,
					    struct tx_read_tracker, &size);
		break;
	default:
		unreachable();
	}
	assert(alloc_type < MEMTX_TX_ALLOC_TYPE_MAX);
	if (alloc != NULL)
		memtx_tx_track_allocation(txn, size, alloc_type);
	return alloc;
}

/**
 * Tx_region method for allocations of arbitrary size.
 * You must pass allocation type explicitly to categorize an allocation.
 * Use this method to track allocations!
 */
static inline void *
memtx_tx_region_alloc(struct txn *txn, size_t size,
		      enum memtx_tx_alloc_type alloc_type)
{
	void *allocation = region_alloc(&txn->region, size);
	if (allocation != NULL)
		memtx_tx_track_allocation(txn, size, alloc_type);
	return allocation;
}

/** String representation of enum memtx_tx_alloc_type. */
const char *memtx_tx_alloc_type_strs[MEMTX_TX_ALLOC_TYPE_MAX] = {
	"trackers",
	"conflicts",
};

/** String representation of enum memtx_tx_story_status. */
const char *memtx_tx_story_status_strs[MEMTX_TX_STORY_STATUS_MAX] = {
	"used",
	"read_view",
	"tracking",
};

struct tx_manager
{
	/**
	 * List of all transactions that are in a read view.
	 * New transactions are added to the tail of this list,
	 * so the list is ordered by rv_psn.
	 */
	struct rlist read_view_txs;
	/**
	 * Mempools for tx_story objects with different index count.
	 * It's the only case when we use bare mempool in memtx_tx because
	 * we cannot account story allocation to any particular txn.
	 */
	struct mempool memtx_tx_story_pool[BOX_INDEX_MAX];
	/** Hash table tuple -> memtx_story of that tuple. */
	struct mh_history_t *history;
	/** Mempool for point_hole_item objects. */
	struct memtx_tx_mempool point_hole_item_pool;
	/** Hash table that hold point selects with empty result. */
	struct mh_point_holes_t *point_holes;
	/** Count of elements in point_holes table. */
	size_t point_holes_size;
	/** Mempool for gap_item objects. */
	struct memtx_tx_mempool gap_item_mempoool;
	/** Mempool for full_scan_item objects. */
	struct memtx_tx_mempool full_scan_item_mempool;
	/** List of all memtx_story objects. */
	struct rlist all_stories;
	struct memtx_tx_stats story_stats[MEMTX_TX_STORY_STATUS_MAX];
	struct memtx_tx_stats retained_tuple_stats[MEMTX_TX_STORY_STATUS_MAX];
	/** Iterator that sequentially traverses all memtx_story objects. */
	struct rlist *traverse_all_stories;
	/** The list containing all transactions. */
	struct rlist all_txs;
	/** Accumulated number of GC steps that should be done. */
	size_t must_do_gc_steps;
};

enum {
	/**
	 * Number of iterations that is allowed for TX manager to do for
	 * searching and deleting no more used memtx_tx_stories per creation of
	 * a new story.
	 */
		TX_MANAGER_GC_STEPS_SIZE = 2,
};

/** That's a definition, see declaration for description. */
bool memtx_tx_manager_use_mvcc_engine = false;

/** The one and only instance of tx_manager. */
static struct tx_manager txm;

void
memtx_tx_manager_init()
{
	rlist_create(&txm.read_view_txs);
	for (size_t i = 0; i < BOX_INDEX_MAX; i++) {
		size_t item_size = sizeof(struct memtx_story) +
				   i * sizeof(struct memtx_story_link);
		mempool_create(&txm.memtx_tx_story_pool[i],
			       cord_slab_cache(), item_size);
	}
	txm.history = mh_history_new();
	memtx_tx_mempool_create(&txm.point_hole_item_pool,
				sizeof(struct point_hole_item),
				MEMTX_TX_ALLOC_TRACKER);
	txm.point_holes = mh_point_holes_new();
	memtx_tx_mempool_create(&txm.gap_item_mempoool,
				sizeof(struct gap_item),
				MEMTX_TX_ALLOC_TRACKER);
	memtx_tx_mempool_create(&txm.full_scan_item_mempool,
				sizeof(struct full_scan_item),
				MEMTX_TX_ALLOC_TRACKER);
	txm.point_holes_size = 0;
	rlist_create(&txm.all_stories);
	rlist_create(&txm.all_txs);
	txm.traverse_all_stories = &txm.all_stories;
	txm.must_do_gc_steps = 0;
	memset(&txm.story_stats, 0, sizeof(txm.story_stats));
}

void
memtx_tx_manager_free()
{
	for (size_t i = 0; i < BOX_INDEX_MAX; i++)
		mempool_destroy(&txm.memtx_tx_story_pool[i]);
	mh_history_delete(txm.history);
	memtx_tx_mempool_destroy(&txm.point_hole_item_pool);
	mh_point_holes_delete(txm.point_holes);
	memtx_tx_mempool_destroy(&txm.gap_item_mempoool);
	memtx_tx_mempool_destroy(&txm.full_scan_item_mempool);
}

void
memtx_tx_statistics_collect(struct memtx_tx_statistics *stats)
{
	memset(stats, 0, sizeof(*stats));
	for (size_t i = 0; i < MEMTX_TX_STORY_STATUS_MAX; ++i) {
		stats->stories[i] = txm.story_stats[i];
		stats->retained_tuples[i] = txm.retained_tuple_stats[i];
	}
	if (rlist_empty(&txm.all_txs)) {
		return;
	}
	struct txn *txn;
	size_t txn_count = 0;
	rlist_foreach_entry(txn, &txm.all_txs, in_all_txs) {
		txn_count++;
		for (size_t i = 0; i < MEMTX_TX_ALLOC_TYPE_MAX; ++i) {
			size_t txn_stat = txn->memtx_tx_alloc_stats[i];
			stats->memtx_tx_total[i] += txn_stat;
			if (txn_stat > stats->memtx_tx_max[i])
				stats->memtx_tx_max[i] = txn_stat;
		}
		for (size_t i = 0; i < TX_ALLOC_TYPE_MAX; ++i) {
			size_t txn_stat = txn->alloc_stats[i];
			stats->tx_total[i] += txn_stat;
			if (txn_stat > stats->tx_max[i])
				stats->tx_max[i] = txn_stat;
		}
	}
	stats->txn_count = txn_count;
}

int
memtx_tx_register_txn(struct txn *tx)
{
	int size = 0;
	tx->memtx_tx_alloc_stats =
		region_alloc_array(&tx->region,
				   typeof(*tx->memtx_tx_alloc_stats),
				   MEMTX_TX_ALLOC_TYPE_MAX, &size);
	if (tx->memtx_tx_alloc_stats == NULL)
		return -1;
	memset(tx->memtx_tx_alloc_stats, 0,
	       sizeof(*tx->memtx_tx_alloc_stats) * MEMTX_TX_ALLOC_TYPE_MAX);
	rlist_add_tail(&txm.all_txs, &tx->in_all_txs);
	return 0;
}

void
memtx_tx_acquire_ddl(struct txn *tx)
{
	tx->is_schema_changed = true;
	(void) txn_can_yield(tx, false);
}

void
memtx_tx_abort_all_for_ddl(struct txn *ddl_owner)
{
	struct txn *to_be_aborted;
	rlist_foreach_entry(to_be_aborted, &txm.all_txs, in_all_txs) {
		if (to_be_aborted == ddl_owner)
			continue;
		if (to_be_aborted->status != TXN_INPROGRESS &&
		    to_be_aborted->status != TXN_IN_READ_VIEW)
			continue;
		to_be_aborted->status = TXN_ABORTED;
		txn_set_flags(to_be_aborted, TXN_IS_CONFLICTED);
		say_warn("Transaction committing DDL (id=%lld) has aborted "
			 "another TX (id=%lld)", (long long) ddl_owner->id,
			 (long long) to_be_aborted->id);
	}
}

/**
 * Fix position of @a txn in global read view list to preserve the list to
 * be ordered by rv_psn. Can only move txn to the beginning of the list.
 * The function must be called when a transaction A sends itself to a read view
 * (perhaps a deeper read view in case when it's already in a read view) because
 * it has to skip a statement of another B, prepared transaction.
 * The transaction is always added to the tail of read view list, but in this
 * case there's no guarantee that psn of B is the greatest psn of all prepared
 * transactions, so we have to additionally and push A in the global read view
 * list, jumping over read views with greater rv_psn.
 */
static void
memtx_tx_adjust_position_in_read_view_list(struct txn *txn)
{
	if (txn->in_read_view_txs.prev == &txm.read_view_txs)
		return; /* No transaction before */
	struct txn *prev_txn = rlist_prev_entry(txn, in_read_view_txs);
	if (prev_txn->rv_psn <= txn->rv_psn)
		return; /* The order is already correct. */
	/* Remove from list for a while. */
	rlist_del(&txn->in_read_view_txs);
	while (prev_txn->in_read_view_txs.prev != &txm.read_view_txs) {
		struct txn *scan = rlist_prev_entry(prev_txn, in_read_view_txs);
		if (scan->rv_psn <= txn->rv_psn)
			break;
		prev_txn = scan;
	}
	/* Insert before prev_txn. */
	rlist_add_tail(&prev_txn->in_read_view_txs, &txn->in_read_view_txs);
}

/**
 * Handle conflict when @a victim has read and @a breaker has written the same
 * key, and @a breaker is prepared. The functions must be called in two cases:
 * 1. @a breaker becomes prepared for every victim with non-empty intersection
 * of victim read set / breaker write set.
 * 2. @a victim has to read confirmed value and skips the value that prepared
 * @a breaker wrote.
 * If @a victim is read-only or hasn't made any changes, it should be sent
 * to read view, in which is will not see @a breaker's changes. If @a victim
 * is already in a read view - a read view that does not see every breaker
 * changes is chosen.
 * Otherwise @a victim must be marked as conflicted and aborted on occasion.
 *
 * NB: can trigger story garbage collection.
 */
static void
memtx_tx_handle_conflict(struct txn *breaker, struct txn *victim)
{
	assert(breaker != victim);
	assert(breaker->psn != 0);
	assert(victim->psn == 0);
	if (victim->status != TXN_INPROGRESS &&
	    victim->status != TXN_IN_READ_VIEW) {
		/* Was conflicted by somebody else. */
		return;
	}
	if (stailq_empty(&victim->stmts)) {
		assert((victim->status == TXN_IN_READ_VIEW) ==
		       (victim->rv_psn != 0));
		/* Send to read view, perhaps a deeper one. */
		if (victim->status != TXN_IN_READ_VIEW) {
			victim->status = TXN_IN_READ_VIEW;
			victim->rv_psn = breaker->psn;
			rlist_add_tail(&txm.read_view_txs,
				       &victim->in_read_view_txs);
		} else if (victim->rv_psn > breaker->psn) {
			/*
			 * Note that in every case for every key we may choose
			 * any read view psn between confirmed level and the
			 * oldest prepared transaction that changes that key.
			 * But we choose the latest level because it generally
			 * costs less, and if there are several breakers - we
			 * must sequentially decrease read view level.
			 */
			victim->rv_psn = breaker->psn;
			assert(victim->rv_psn != 0);
		}
		memtx_tx_adjust_position_in_read_view_list(victim);
	} else {
		/* Mark as conflicted. */
		if (victim->status == TXN_IN_READ_VIEW)
			rlist_del(&victim->in_read_view_txs);
		victim->status = TXN_ABORTED;
		txn_set_flags(victim, TXN_IS_CONFLICTED);
	}
}

/**
 * Calculate size of story with its links.
 */
static inline size_t
memtx_story_size(struct memtx_story *story)
{
	struct mempool *pool = &txm.memtx_tx_story_pool[story->index_count];
	return pool->objsize;
}

/**
 * Notify memory manager that a tuple referenced by @a story
 * was replaced from primary key and that is why @a story
 * is the only reason why the tuple cannot be deleted.
 */
static inline void
memtx_tx_story_track_retained_tuple(struct memtx_story *story)
{
	assert(!story->tuple_is_retained);
	assert(story->status < MEMTX_TX_STORY_STATUS_MAX);

	story->tuple_is_retained = true;
	struct memtx_tx_stats *stats = &txm.retained_tuple_stats[story->status];
	size_t tuplesize = tuple_size(story->tuple);
	memtx_tx_stats_collect(stats, tuplesize);
}

/**
 * Notify memory manager that a tuple referenced by @a story
 * was placed to primary key.
 */
static inline void
memtx_tx_story_untrack_retained_tuple(struct memtx_story *story)
{
	assert(story->tuple_is_retained);
	assert(story->status < MEMTX_TX_STORY_STATUS_MAX);

	story->tuple_is_retained = false;
	struct memtx_tx_stats *stats = &txm.retained_tuple_stats[story->status];
	size_t tuplesize = tuple_size(story->tuple);
	memtx_tx_stats_discard(stats, tuplesize);
}

/** Set status of story (see memtx_tx_story_status) */
static inline void
memtx_tx_story_set_status(struct memtx_story *story,
			  enum memtx_tx_story_status new_status)
{
	assert(story->status < MEMTX_TX_STORY_STATUS_MAX);
	enum memtx_tx_story_status old_status = story->status;
	if (old_status == new_status)
		return;
	story->status = new_status;
	struct memtx_tx_stats *old_story_stats = &txm.story_stats[old_status];
	struct memtx_tx_stats *new_story_stats = &txm.story_stats[new_status];
	size_t story_size = memtx_story_size(story);
	memtx_tx_stats_discard(old_story_stats, story_size);
	memtx_tx_stats_collect(new_story_stats, story_size);
	if (story->tuple_is_retained) {
		size_t tuplesize = tuple_size(story->tuple);
		struct memtx_tx_stats *old =
			&txm.retained_tuple_stats[old_status];
		struct memtx_tx_stats *new =
			&txm.retained_tuple_stats[new_status];
		memtx_tx_stats_discard(old, tuplesize);
		memtx_tx_stats_collect(new, tuplesize);
	}
}

/**
 * Use this method to ref tuple that belongs to @a story
 * by primary index. Do not use bare tuple_ref!!!
 */
static inline void
memtx_tx_ref_to_primary(struct memtx_story *story)
{
	assert(story != NULL);
	assert(story->tuple_is_retained);
	tuple_ref(story->tuple);
	memtx_tx_story_untrack_retained_tuple(story);
}

/**
 * Use this method to unref tuple that belongs to @a story
 * from primary index. Do not use bare tuple_unref!!!
 */
static inline void
memtx_tx_unref_from_primary(struct memtx_story *story)
{
	assert(story != NULL);
	tuple_unref(story->tuple);
	if (!story->tuple_is_retained)
		memtx_tx_story_track_retained_tuple(story);
}

/**
 * Create a new story and link it with the @a tuple.
 * @return story on success, NULL on error (diag is set).
 */
static struct memtx_story *
memtx_tx_story_new(struct space *space, struct tuple *tuple,
		   bool tuple_is_referenced_to_pk)
{
	txm.must_do_gc_steps += TX_MANAGER_GC_STEPS_SIZE;
	assert(!tuple_has_flag(tuple, TUPLE_IS_DIRTY));
	uint32_t index_count = space->index_count;
	assert(index_count < BOX_INDEX_MAX);
	struct mempool *pool = &txm.memtx_tx_story_pool[index_count];
	struct memtx_story *story = (struct memtx_story *) mempool_alloc(pool);
	size_t item_size = pool->objsize;
	if (story == NULL) {
		diag_set(OutOfMemory, item_size, "mempool_alloc", "story");
		return NULL;
	}
	story->tuple = tuple;

	const struct memtx_story **put_story =
		(const struct memtx_story **) &story;
	struct memtx_story *replaced = NULL;
	struct memtx_story **preplaced = &replaced;
	mh_history_put(txm.history, put_story, &preplaced, 0);
	assert(preplaced == NULL);
	tuple_set_flag(tuple, TUPLE_IS_DIRTY);
	tuple_ref(tuple);
	story->status = MEMTX_TX_STORY_USED;
	struct memtx_tx_stats *stats = &txm.story_stats[story->status];
	memtx_tx_stats_collect(stats, item_size);
	story->tuple_is_retained = false;
	if (!tuple_is_referenced_to_pk)
		memtx_tx_story_track_retained_tuple(story);
	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_tail(&txm.all_stories, &story->in_all_stories);
	rlist_add(&space->memtx_stories, &story->in_space_stories);
	for (uint32_t i = 0; i < index_count; i++) {
		story->link[i].newer_story = story->link[i].older_story = NULL;
		rlist_create(&story->link[i].nearby_gaps);
		story->link[i].in_index = space->index[i];
	}
	return story;
}

static void
memtx_tx_story_delete(struct memtx_story *story)
{
	memtx_tx_stats_discard(&txm.story_stats[story->status],
			       memtx_story_size(story));
	if (story->tuple_is_retained)
		memtx_tx_story_untrack_retained_tuple(story);

	if (story->add_stmt != NULL) {
		assert(story->add_stmt->add_story == story);
		story->add_stmt->add_story = NULL;
		story->add_stmt = NULL;
	}
	while (story->del_stmt != NULL) {
		assert(story->del_stmt->del_story == story);
		story->del_stmt->del_story = NULL;
		struct txn_stmt *next = story->del_stmt->next_in_del_list;
		story->del_stmt->next_in_del_list = NULL;
		story->del_stmt = next;
	}

	if (txm.traverse_all_stories == &story->in_all_stories)
		txm.traverse_all_stories = rlist_next(txm.traverse_all_stories);
	rlist_del(&story->in_all_stories);
	rlist_del(&story->in_space_stories);

	mh_int_t pos = mh_history_find(txm.history, story->tuple, 0);
	assert(pos != mh_end(txm.history));
	mh_history_del(txm.history, pos, 0);

	tuple_clear_flag(story->tuple, TUPLE_IS_DIRTY);
	tuple_unref(story->tuple);

#ifndef NDEBUG
	/* Expecting to delete fully unlinked story. */
	for (uint32_t i = 0; i < story->index_count; i++) {
		assert(story->link[i].newer_story == NULL);
		assert(story->link[i].older_story == NULL);
	}
#endif

	struct mempool *pool = &txm.memtx_tx_story_pool[story->index_count];
	mempool_free(pool, story);
}

/**
 * Find a story of a @a tuple. The story expected to be present (assert).
 */
static struct memtx_story *
memtx_tx_story_get(struct tuple *tuple)
{
	assert(tuple_has_flag(tuple, TUPLE_IS_DIRTY));

	mh_int_t pos = mh_history_find(txm.history, tuple, 0);
	assert(pos != mh_end(txm.history));
	struct memtx_story *story = *mh_history_node(txm.history, pos);
	if (story->add_stmt != NULL)
		assert(story->add_psn == story->add_stmt->txn->psn);
	if (story->del_stmt != NULL)
		assert(story->del_psn == story->del_stmt->txn->psn);
	return story;
}

/**
 * Link that @a story was added by @a stmt.
 */
static void
memtx_tx_story_link_added_by(struct memtx_story *story,
			     struct txn_stmt *stmt)
{
	assert(story->add_stmt == NULL);
	assert(stmt->add_story == NULL);
	story->add_stmt = stmt;
	stmt->add_story = story;
}

/**
 * Unlink that @a story from @a stmt which added it.
 * Effectively undo memtx_tx_story_link_added_by.
 */
static void
memtx_tx_story_unlink_added_by(struct memtx_story *story,
			       struct txn_stmt *stmt)
{
	assert(stmt->add_story == story);
	assert(story->add_stmt == stmt);
	stmt->add_story = NULL;
	story->add_stmt = NULL;
}

/**
 * Link that @a story was deleted by @a stmt.
 */
static void
memtx_tx_story_link_deleted_by(struct memtx_story *story,
			       struct txn_stmt *stmt)
{
	assert(stmt->del_story == NULL);
	assert(stmt->next_in_del_list == NULL);

	stmt->del_story = story;
	stmt->next_in_del_list = story->del_stmt;
	story->del_stmt = stmt;
}

/**
 * Unlink that @a story from @a stmt that deleted it.
 * Effectively undo memtx_tx_story_link_deleted_by.
 */
static void
memtx_tx_story_unlink_deleted_by(struct memtx_story *story,
				 struct txn_stmt *stmt)
{
	assert(stmt->del_story == story);

	/* Find a place in list from which stmt must be deleted. */
	struct txn_stmt **ptr = &story->del_stmt;
	while (*ptr != stmt) {
		ptr = &(*ptr)->next_in_del_list;
		assert(ptr != NULL);
	}
	*ptr = stmt->next_in_del_list;
	stmt->next_in_del_list = NULL;
	stmt->del_story = NULL;
}

/**
 * Link a @a story with @a old_story in @a index (in both directions).
 * @a old_story is allowed to be NULL.
 */
static void
memtx_tx_story_link(struct memtx_story *story,
		    struct memtx_story *old_story,
		    uint32_t idx)
{
	assert(idx < story->index_count);
	struct memtx_story_link *link = &story->link[idx];
	assert(link->older_story == NULL);

	if (old_story == NULL)
		return;

	assert(idx < old_story->index_count);
	struct memtx_story_link *old_link = &old_story->link[idx];
	assert(old_link->newer_story == NULL);

	link->older_story = old_story;
	old_link->newer_story = story;
}

/**
 * Unlink a @a story with @a old_story in @a index (in both directions).
 * Older story is allowed to be NULL.
 */
static void
memtx_tx_story_unlink(struct memtx_story *story,
		      struct memtx_story *old_story,
		      uint32_t idx)
{
	assert(idx < story->index_count);
	struct memtx_story_link *link = &story->link[idx];
	assert(link->older_story == old_story);

	if (old_story == NULL)
		return;

	assert(idx < old_story->index_count);
	struct memtx_story_link *old_link = &old_story->link[idx];
	assert(old_link->newer_story == story);

	link->older_story = NULL;
	old_link->newer_story = NULL;
}

/**
 * Link a @a story with @a old_story in @a index (in both directions),
 * where old_story was at the top of chain. This is a light version of
 * function, intended for the case when the appropriate index IS ALREADY
 * changed, and now stores story->tuple.
 * In addition to linking in list, this function also rebinds gap records
 * to the new top - story - in history chain.
 * @a old_story is allowed to be NULL.
 */
static void
memtx_tx_story_link_top_light(struct memtx_story *new_top,
			      struct memtx_story *old_top,
			      uint32_t idx)
{
	memtx_tx_story_link(new_top, old_top, idx);

	if (old_top == NULL)
		return;

	/* Rebind gap records to the top of the list */
	struct memtx_story_link *new_link = &new_top->link[idx];
	struct memtx_story_link *old_link = &old_top->link[idx];
	rlist_splice(&new_link->nearby_gaps, &old_link->nearby_gaps);
}

/**
 * Link a @a new_top with @a old_top in @a index (in both directions),
 * where old_top was at the top of chain (that means that index itself
 * stores a pointer to old_top->tuple).
 * In addition to linking in list, this function also rebinds gap records
 * to the new top - story - in history chain.
 * This function makes also changes in the index, replacing old_top->tuple
 * with new_top->tuple.
 */
static void
memtx_tx_story_link_top(struct memtx_story *new_top,
			struct memtx_story *old_top,
			uint32_t idx)
{
	assert(old_top != NULL);
	memtx_tx_story_link_top_light(new_top, old_top, idx);

	/* Make the change in index. */
	struct index *index = old_top->link[idx].in_index;
	assert(index != NULL);
	assert(new_top->link[idx].in_index == NULL);
	struct tuple *removed, *unused;
	if (index_replace(index, old_top->tuple, new_top->tuple,
			  DUP_REPLACE, &removed, &unused) != 0) {
		diag_log();
		unreachable();
		panic("failed to rebind story in index");
	}
	assert(old_top->tuple == removed);
	old_top->link[idx].in_index = NULL;
	new_top->link[idx].in_index = index;

	/*
	 * A space holds references to all his tuples.
	 * All tuples that are physically in the primary index are referenced.
	 * Thus we have to reference the tuple that was added to the primar
	 * index and dereference the tuple that was removed from it.
	 */
	if (idx == 0) {
		memtx_tx_ref_to_primary(new_top);
		memtx_tx_unref_from_primary(old_top);
	}
}

/**
 * Unlink a @a story from history chain in @a index (in both directions),
 * where old_top was at the top of chain (that means that index itself
 * stores a pointer to story->tuple).
 * This function makes also changes in the index, replacing old_top->tuple
 * with the correct tuple (the next in chain, maybe NULL).
 */
static void
memtx_tx_story_unlink_top_common(struct memtx_story *story, uint32_t idx)
{
	assert(story != NULL);
	assert(idx < story->index_count);
	struct memtx_story_link *link = &story->link[idx];

	assert(link->newer_story == NULL);
	/*
	 * Note that link[idx].in_index may not be the same as
	 * story->space->index[idx] in case space is going to be deleted
	 * in memtx_tx_on_space_delete(): during space alter operation we
	 * swap all indexes to the new space object and instead use dummy
	 * structs.
	 */
	struct index *index = story->link[idx].in_index;

	struct memtx_story *old_story = link->older_story;
	assert(old_story == NULL || old_story->link[idx].in_index == NULL);
	struct tuple *old_tuple = old_story == NULL ? NULL : old_story->tuple;
	struct tuple *removed, *unused;
	if (index_replace(index, story->tuple, old_tuple,
			  DUP_INSERT, &removed, &unused) != 0) {
		diag_log();
		unreachable();
		panic("failed to rebind story in index");
	}
	assert(story->tuple == removed);
	story->link[idx].in_index = NULL;
	if (old_story != NULL)
		old_story->link[idx].in_index = index;

	/*
	 * A space holds references to all his tuples.
	 * All tuples that are physically in the primary index are referenced.
	 * Thus we have to reference the tuple that was added to the primary
	 * index and dereference the tuple that was removed from it.
	 */
	if (idx == 0) {
		if (old_story != NULL)
			memtx_tx_ref_to_primary(old_story);
		memtx_tx_unref_from_primary(story);
	}
}

/**
 * Unlink a @a story from history chain in @a index (in both directions),
 * where old_top was at the top of chain (that means that index itself
 * stores a pointer to story->tuple).
 * This is a light version of function, intended for the case when the
 * appropriate change in the index will be done later by caller.
 */
static void
memtx_tx_story_unlink_top_common_light(struct memtx_story *story, uint32_t idx)
{
	assert(story != NULL);
	assert(idx < story->index_count);
	struct memtx_story_link *link = &story->link[idx];
	assert(link->newer_story == NULL);
	struct memtx_story *old_story = link->older_story;
	if (old_story != NULL)
		memtx_tx_story_unlink(story, old_story, idx);
}

/**
 * See description of `memtx_tx_story_unlink_top_common_light`.
 * As opposed to `memtx_tx_story_unlink_top_on_space_delete_light`, also
 * rebinds gap records to the new top in history chain.
 */
static void
memtx_tx_story_unlink_top_on_rollback_light(struct memtx_story *story,
					    uint32_t idx)
{
	assert(story != NULL);
	assert(idx < story->index_count);
	struct memtx_story_link *link = &story->link[idx];
	assert(link->newer_story == NULL);
	struct memtx_story *old_story = link->older_story;
	memtx_tx_story_unlink_top_common_light(story, idx);
	if (old_story != NULL) {
		/* Rebind gap records to the new top of the list */
		struct memtx_story_link *old_link = &old_story->link[idx];
		rlist_splice(&old_link->nearby_gaps, &link->nearby_gaps);
	}
}

/**
 * See description of `memtx_tx_story_unlink_top_common_light`.
 * Since the space is being deleted, we need to simply unlink the story.
 */
static void
memtx_tx_story_unlink_top_on_space_delete_light(struct memtx_story *story,
						uint32_t idx)
{
	memtx_tx_story_unlink_top_common_light(story, idx);
}

/**
 * See description of `memtx_tx_story_unlink_top_common`.
 * Since the space is being deleted, we need to simply unlink the story.
 */
static void
memtx_tx_story_unlink_top_on_space_delete(struct memtx_story *story,
					  uint32_t idx)
{
	memtx_tx_story_unlink_top_common(story, idx);
	memtx_tx_story_unlink_top_on_space_delete_light(story, idx);
}

/**
 * Unlink a @a story from history chain in @a index in both directions.
 * Handles case when the story is not on top of the history: simply remove from
 * list.
 */
static void
memtx_tx_story_unlink_both_common(struct memtx_story *story,
				  uint32_t idx)
{
	assert(idx < story->index_count);
	struct memtx_story_link *link = &story->link[idx];
	struct memtx_story *newer_story = link->newer_story;
	struct memtx_story *older_story = link->older_story;
	memtx_tx_story_unlink(newer_story, story, idx);
	memtx_tx_story_unlink(story, older_story, idx);
	memtx_tx_story_link(newer_story, older_story, idx);
}

/**
 * Unlink a @a story from history chain in @a index in both directions.
 * If the story was in the top of history chain - unlink from top. Otherwise,
 * see description of `memtx_tx_story_unlink_both_common`.
 * Since the space is being deleted, we need to simply unlink the story.
 */
static void
memtx_tx_story_unlink_both_on_space_delete(struct memtx_story *story,
					   uint32_t idx)
{
	assert(idx < story->index_count);
	struct memtx_story_link *link = &story->link[idx];
	if (link->newer_story == NULL) {
		assert(link->in_index != NULL);
		memtx_tx_story_unlink_top_on_space_delete(story, idx);
	} else {
		memtx_tx_story_unlink_both_common(story, idx);
	}
}

/**
 * Change the order of stories in history chain.
 */
static void
memtx_tx_story_reorder(struct memtx_story *story,
		       struct memtx_story *old_story,
		       uint32_t idx)
{
	assert(idx < story->index_count);
	assert(idx < old_story->index_count);
	struct memtx_story_link *link = &story->link[idx];
	struct memtx_story_link *old_link = &old_story->link[idx];
	assert(link->older_story == old_story);
	assert(old_link->newer_story == story);
	struct memtx_story *newer_story = link->newer_story;
	struct memtx_story *older_story = old_link->older_story;

	/*
	 * We have a list of stories, and we have to reorder it.
	 *           What we have                 What we want
	 *      [ index/newer_story ]        [ index/newer_story ]
	 *      [       story       ]        [     old_story     ]
	 *      [     old_story     ]        [       story       ]
	 *      [    older_story    ]        [    older_story    ]
	 */
	memtx_tx_story_unlink(story, old_story, idx);
	memtx_tx_story_unlink(old_story, older_story, idx);
	if (newer_story != NULL) {
		/* Simple relink in list. */
		memtx_tx_story_unlink(newer_story, story, idx);
		memtx_tx_story_link(newer_story, old_story, idx);
		memtx_tx_story_link(old_story, story, idx);
	} else {
		/*
		 * story was in the top of history chain. In terms of reorder,
		 * we have to make old_story the new top of chain. */
		memtx_tx_story_link_top(old_story, story, idx);
	}
	memtx_tx_story_link(story, older_story, idx);
}

/**
 * Unlink @a story from all chains and remove corresponding tuple from
 * indexes if necessary: used in `memtx_tx_on_space_delete` — intentionally
 * violates the top of the history chain invariant (see
 * `memtx_tx_story_full_unlink_story_gc_step`), since all stories are deleted
 * anyways.
 */
static void
memtx_tx_story_full_unlink_on_space_delete(struct memtx_story *story)
{
	for (uint32_t i = 0; i < story->index_count; i++) {
		struct memtx_story_link *link = &story->link[i];
		if (link->newer_story == NULL) {
			/*
			 * We are at the top of the chain. That means
			 * that story->tuple might be in index. If the story
			 * actually deletes the tuple and is present in index,
			 * it must be deleted from index.
			 */
			if (story->del_psn > 0 && link->in_index != NULL) {
				struct index *index = link->in_index;
				struct tuple *removed, *unused;
				if (index_replace(index, story->tuple, NULL,
						  DUP_INSERT,
						  &removed, &unused) != 0) {
					diag_log();
					unreachable();
					panic("failed to rollback change");
				}
				assert(story->tuple == removed);
				link->in_index = NULL;
				/*
				 * All tuples in pk are referenced.
				 * Once removed it must be unreferenced.
				 */
				if (i == 0)
					memtx_tx_unref_from_primary(story);
			}

			memtx_tx_story_unlink(story, link->older_story, i);
		} else {
			/* Just unlink from list */
			link->newer_story->link[i].older_story = link->older_story;
			if (link->older_story != NULL)
				link->older_story->link[i].newer_story =
					link->newer_story;
			link->older_story = NULL;
			link->newer_story = NULL;
		}
	}
}

/**
 * Unlink @a story from all chains and remove corresponding tuple from
 * indexes if necessary: used in garbage collection step and preserves the top
 * of the history chain invariant (as opposed to
 * `memtx_tx_story_full_unlink_on_space_delete`).
 */
static void
memtx_tx_story_full_unlink_story_gc_step(struct memtx_story *story)
{
	for (uint32_t i = 0; i < story->index_count; i++) {
		struct memtx_story_link *link = &story->link[i];
		if (link->newer_story == NULL) {
			/*
			 * We are at the top of the chain. That means
			 * that story->tuple is in index or the story is a
			 * rollbacked one. If the story actually deletes the
			 * tuple and is present in index, it must be deleted
			 * from index.
			 */
			assert(link->in_index != NULL);
			/*
			 * Invariant that the top of the history chain
			 * is always in the index: here we delete
			 * (sic: not replace) a tuple from the index,
			 * and  it must be the last story left in the
			 * history chain, otherwise `link->older_story`
			 * starts to be at the top of the history chain
			 * and is not present in index, which violates
			 * our invariant.
			 */
			assert(link->older_story == NULL);
			if (story->del_psn > 0 && link->in_index != NULL) {
				struct index *index = link->in_index;
				struct tuple *removed, *unused;
				if (index_replace(index, story->tuple, NULL,
						  DUP_INSERT,
						  &removed, &unused) != 0) {
					diag_log();
					unreachable();
					panic("failed to rollback change");
				}
				assert(story->tuple == removed);
				link->in_index = NULL;
				/*
				 * All tuples in pk are referenced.
				 * Once removed it must be unreferenced.
				 */
				if (i == 0)
					memtx_tx_unref_from_primary(story);
			}

			memtx_tx_story_unlink(story, link->older_story, i);
		} else {
			/* Just unlink from list */
			link->newer_story->link[i].older_story =
				link->older_story;
			if (link->older_story != NULL)
				link->older_story->link[i].newer_story =
					link->newer_story;
			link->older_story = NULL;
			link->newer_story = NULL;
		}
	}
}

/**
 * Run one step of a crawler that traverses all stories and removes no more
 * used stories.
 */
void
memtx_tx_story_gc_step()
{
	if (txm.traverse_all_stories == &txm.all_stories) {
		/* We came to the head of the list. */
		txm.traverse_all_stories = txm.traverse_all_stories->next;
		return;
	}

	/*
	 * Lowest read view PSN.
	 * Default value is psn_generator because if it is not so some
	 * stories (stories produced by last txn at least) will be marked as
	 * potentially in read view even though there are no txns in read view.
	 */
	int64_t lowest_rv_psn = psn_generator;
	if (!rlist_empty(&txm.read_view_txs)) {
		struct txn *txn =
			rlist_first_entry(&txm.read_view_txs, struct txn,
					  in_read_view_txs);
		assert(txn->rv_psn != 0);
		lowest_rv_psn = txn->rv_psn;
	}

	struct memtx_story *story =
		rlist_entry(txm.traverse_all_stories, struct memtx_story,
			    in_all_stories);
	txm.traverse_all_stories = txm.traverse_all_stories->next;

	/**
	 * The order in which conditions are checked is important,
	 * see description of enum memtx_tx_story_status.
	 */
	if (story->add_stmt != NULL || story->del_stmt != NULL ||
	    !rlist_empty(&story->reader_list)) {
		memtx_tx_story_set_status(story, MEMTX_TX_STORY_USED);
		/* The story is used directly by some transactions. */
		return;
	}
	if (story->add_psn >= lowest_rv_psn ||
	    story->del_psn >= lowest_rv_psn) {
		memtx_tx_story_set_status(story, MEMTX_TX_STORY_READ_VIEW);
		/* The story can be used by a read view. */
		return;
	}
	for (uint32_t i = 0; i < story->index_count; i++) {
		struct memtx_story_link *link = &story->link[i];
		if (link->newer_story == NULL) {
			assert(link->in_index != NULL);
			/*
			 * We would have to unlink this tuple (and perhaps
			 * delete it from index if story->del_psn > 0), but we
			 * cannot do this since after that `link->older_story`
			 * starts to be at the top of the history chain, and it
			 * is not present in index, which violates our
			 * invariant.
			 */
			if (link->older_story != NULL) {
				memtx_tx_story_set_status(story,
							  MEMTX_TX_STORY_USED);
				return;
			}
		} else if (i > 0 && link->newer_story->add_stmt != NULL) {
			/*
			 * We need to retain the story since the newer story
			 * can get rolled back (this is maintained by delete
			 * statement list in case of primary index).
			 */
			memtx_tx_story_set_status(story,
						  MEMTX_TX_STORY_USED);
			return;
		}
		if (!rlist_empty(&link->nearby_gaps)) {
			memtx_tx_story_set_status(story,
						  MEMTX_TX_STORY_TRACK_GAP);
			/* The story is used for gap tracking. */
			return;
		}
	}

	/* Unlink and delete the story */
	memtx_tx_story_full_unlink_story_gc_step(story);
	memtx_tx_story_delete(story);
}

void
memtx_tx_story_gc()
{
	for (size_t i = 0; i < txm.must_do_gc_steps; i++)
		memtx_tx_story_gc_step();
	txm.must_do_gc_steps = 0;
}

/**
 * Check whether the beginning of a @a story (that is insertion of its tuple)
 * is visible for transaction @a txn.
 * @param is_prepared_ok - whether prepared, not confirmed change is acceptable.
 * @param own_change - return true if the change was made by @txn itself.
 * @return true if the story beginning is visible, false otherwise.
 */
static bool
memtx_tx_story_insert_is_visible(struct memtx_story *story, struct txn *txn,
				 bool is_prepared_ok, bool *is_own_change)
{
	*is_own_change = false;

	if (story->add_stmt != NULL && story->add_stmt->txn == txn) {
		/* Tuple is added by us (@txn). */
		*is_own_change = true;
		return true;
	}

	int64_t rv_psn = INT64_MAX;
	if (txn != NULL && txn->rv_psn != 0)
		rv_psn = txn->rv_psn;

	if (is_prepared_ok && story->add_psn != 0 && story->add_psn < rv_psn)
		return true; /* Tuple is added by another prepared TX. */

	if (story->add_psn != 0 && story->add_stmt == NULL &&
	    story->add_psn < rv_psn)
		return true; /* Tuple is added by committed TX. */

	if (story->add_psn == 0 && story->add_stmt == NULL)
		return true; /* Added long time ago. */

	return false;
}

/**
 * Check whether the end of a @a story (that is deletion of its tuple) is
 * visible for transaction @a txn.
 * @param is_prepared_ok - whether prepared, not confirmed change is acceptable.
 * @param own_change - return true if the change was made by @txn itself.
 * @return true if the story end is visible, false otherwise.
 */
static bool
memtx_tx_story_delete_is_visible(struct memtx_story *story, struct txn *txn,
				 bool is_prepared_ok, bool *is_own_change)
{
	*is_own_change = false;

	bool was_deleted_by_prepared = false;
	struct txn_stmt *dels = story->del_stmt;
	while (dels != NULL) {
		if (dels->txn == txn) {
			/* Tuple is deleted by us (@txn). */
			*is_own_change = true;
			return true;
		}
		if (story->del_psn != 0 && dels->txn->psn == story->del_psn)
			was_deleted_by_prepared = true;
		dels = dels->next_in_del_list;
	}

	int64_t rv_psn = INT64_MAX;
	if (txn != NULL && txn->rv_psn != 0)
		rv_psn = txn->rv_psn;

	if (is_prepared_ok && story->del_psn != 0 && story->del_psn < rv_psn)
		return true; /* Tuple is deleted by prepared TX. */

	if (story->del_psn != 0 && !was_deleted_by_prepared &&
	    story->del_psn < rv_psn)
		return true; /* Tuple is deleted by committed TX. */

	return false;
}

/**
 * Scan a history starting with @a story in @a index for a @a visible_tuple.
 * If @a is_prepared_ok is true that prepared statements are visible for
 * that lookup, and not visible otherwise.
 *
 * `is_own_change` is set to true iff `visible_tuple` was modified (either
 * added or deleted) by `txn`.
 */
static void
memtx_tx_story_find_visible_tuple(struct memtx_story *story, struct txn *tnx,
				  uint32_t index, bool is_prepared_ok,
				  struct tuple **visible_tuple,
				  bool *is_own_change)
{
	for (; story != NULL; story = story->link[index].older_story) {
		assert(index < story->index_count);
		if (memtx_tx_story_delete_is_visible(story, tnx, is_prepared_ok,
						     is_own_change)) {
			*visible_tuple = NULL;
			return;
		}
		if (memtx_tx_story_insert_is_visible(story, tnx, is_prepared_ok,
						     is_own_change)) {
			*visible_tuple = story->tuple;
			return;
		}
	}
	*visible_tuple = NULL;
}

/**
 * replace_check_dup wrapper, that follows usual return code convention and
 * sets diag in case of error.
 */
static int
memtx_tx_check_dup(struct tuple *new_tuple, struct tuple *old_tuple,
		   struct tuple *dup_tuple, enum dup_replace_mode mode,
		   struct index *index, struct space *space)
{
	uint32_t errcode;
	errcode = replace_check_dup(old_tuple, dup_tuple, mode);
	if (errcode != 0) {
		if (errcode == ER_TUPLE_FOUND) {
			diag_set(ClientError, errcode,
				 index->def->name,
				 space_name(space),
				 tuple_str(dup_tuple),
				 tuple_str(new_tuple));
		} else {
			diag_set(ClientError, errcode, space_name(space));
		}
		return -1;
	}
	return 0;
}

static struct point_hole_item *
point_hole_storage_find(struct index *index, struct tuple *tuple)
{
	struct point_hole_key key;
	key.index = index;
	key.tuple = tuple;
	mh_int_t pos = mh_point_holes_find(txm.point_holes, &key, 0);
	if (pos == mh_end(txm.point_holes))
		return NULL;
	return *mh_point_holes_node(txm.point_holes, pos);
}

static int
memtx_tx_track_read_story(struct txn *txn, struct space *space,
			  struct memtx_story *story, uint64_t index_mask);

/**
 * Check for possible conflict relations during insertion of @a new tuple,
 * and given that it was a real insertion, not the replacement of existing
 * tuple. It's the moment where we can search for stored point hole trackers
 * and find conflict causes.
 */
static int
check_hole(struct space *space, struct memtx_story *story,
	   struct tuple *new_tuple, uint32_t ind)
{
	struct point_hole_item *list =
		point_hole_storage_find(space->index[ind], new_tuple);
	if (list == NULL)
		return 0;

	struct point_hole_item *item = list;
	uint64_t index_mask = 1ull << (ind & 63);
	do {
		if (memtx_tx_track_read_story(item->txn, space, story,
					      index_mask) != 0)
			return -1;
		item = rlist_entry(item->ring.next,
				   struct point_hole_item, ring);
	} while (item != list);
	return 0;
}

/**
 * Record in TX manager that a transaction @txn have read a @tuple in @space.
 *
 * NB: can trigger story garbage collection.
 *
 * @return 0 on success, -1 on memory error.
 */
static int
memtx_tx_track_read(struct txn *txn, struct space *space, struct tuple *tuple);

/**
 * Check that replaced tuples in space's indexes does not violate common
 * replace rules. See memtx_space_replace_all_keys comment.
 * @return 0 on success or -1 on fail.
 *
 * `is_own_change` is set to true iff `old_tuple` was modified (either
 * added or deleted) by `stmt`'s transaction.
 */
static int
check_dup(struct txn_stmt *stmt, struct tuple *new_tuple,
	  struct tuple **directly_replaced, struct tuple **old_tuple,
	  enum dup_replace_mode mode, bool *is_own_change)
{
	struct space *space = stmt->space;
	struct txn *txn = stmt->txn;

	struct tuple *visible_replaced;
	if (directly_replaced[0] == NULL ||
	    !tuple_has_flag(directly_replaced[0], TUPLE_IS_DIRTY)) {
		*is_own_change = false;
		visible_replaced = directly_replaced[0];
	} else {
		struct memtx_story *story =
			memtx_tx_story_get(directly_replaced[0]);
		memtx_tx_story_find_visible_tuple(story, txn, 0, true,
						  &visible_replaced,
						  is_own_change);
	}

	if (memtx_tx_check_dup(new_tuple, *old_tuple, visible_replaced,
			       mode, space->index[0], space) != 0) {
		memtx_tx_track_read(txn, space, visible_replaced);
		return -1;
	}

	for (uint32_t i = 1; i < space->index_count; i++) {
		/*
		 * Check that visible tuple is NULL or the same as in the
		 * primary index, namely replaced[0].
		 */
		if (directly_replaced[i] == NULL)
			continue; /* NULL is OK in any case. */

		struct tuple *visible;
		if (!tuple_has_flag(directly_replaced[i], TUPLE_IS_DIRTY)) {
			visible = directly_replaced[i];
		} else {
			/*
			 * The replaced tuple is dirty. A chain of changes
			 * cannot lead to clean tuple, but in can lead to NULL,
			 * that's the only chance to be OK.
			 */
			struct memtx_story *story =
				memtx_tx_story_get(directly_replaced[i]);
			bool unused;
			memtx_tx_story_find_visible_tuple(story, txn,
							  i, true, &visible,
							  &unused);
		}

		if (memtx_tx_check_dup(new_tuple, visible_replaced, visible,
				       DUP_INSERT, space->index[i],
				       space) != 0) {
			memtx_tx_track_read(txn, space, visible);
			return -1;
		}
	}

	*old_tuple = visible_replaced;
	return 0;
}

static struct gap_item *
memtx_tx_gap_item_new(struct txn *txn, enum iterator_type type,
		      const char *key, uint32_t part_count);

/**
 * Handle insertion to a new place in index. There can be readers which
 * have read from this gap and thus must be sent to read view or conflicted.
 */
static int
memtx_tx_handle_gap_write(struct txn *txn, struct space *space,
			  struct memtx_story *story, struct tuple *tuple,
			  struct tuple *successor, uint32_t ind)
{
	struct index *index = space->index[ind];
	struct full_scan_item *fsc_item, *fsc_tmp;
	struct rlist *fsc_list = &index->full_scans;
	uint64_t index_mask = 1ull << (ind & 63);
	rlist_foreach_entry_safe(fsc_item, fsc_list, in_full_scans, fsc_tmp) {
		if (fsc_item->txn != txn &&
		    (memtx_tx_track_read_story(fsc_item->txn, space, story,
					       index_mask) != 0))
			return -1;
	}
	if (successor != NULL && !tuple_has_flag(successor, TUPLE_IS_DIRTY))
		return 0; /* no gap records */

	struct rlist *list = &index->nearby_gaps;
	if (successor != NULL) {
		assert(tuple_has_flag(successor, TUPLE_IS_DIRTY));
		struct memtx_story *succ_story = memtx_tx_story_get(successor);
		assert(ind < succ_story->index_count);
		list = &succ_story->link[ind].nearby_gaps;
		assert(list->next != NULL && list->prev != NULL);
	}
	struct gap_item *item, *tmp;
	rlist_foreach_entry_safe(item, list, in_nearby_gaps, tmp) {
		int cmp = 0;
		if (item->key != NULL) {
			struct key_def *def = index->def->key_def;
			hint_t oh =
				def->key_hint(item->key, item->part_count, def);
			hint_t kh = def->tuple_hint(tuple, def);
			cmp = def->tuple_compare_with_key(tuple, kh, item->key,
							  item->part_count, oh,
							  def);
		}
		int dir = iterator_direction(item->type);
		bool is_full_key = item->part_count ==
				   index->def->cmp_def->part_count;
		bool is_eq = item->type == ITER_EQ || item->type == ITER_REQ;
		bool is_e = item->type == ITER_LE || item->type == ITER_GE;
		bool need_split = item->key == NULL ||
				  (dir * cmp > 0 && !is_eq) ||
				  (!is_full_key && cmp == 0 && (is_e || is_eq));
		bool need_move = !need_split &&
				 ((dir < 0 && cmp > 0) ||
				  (cmp > 0 && item->type == ITER_EQ) ||
				  (cmp == 0 && ((dir < 0 && is_full_key) ||
						item->type == ITER_LT)));
		bool need_track = need_split ||
				  (is_full_key && cmp == 0 && is_e);
		/* There's no need to track read of own change. */
		if (story->add_stmt != NULL &&
		    story->add_stmt->txn == item->txn)
			need_track = false;
		if (need_track && memtx_tx_track_read_story(item->txn, space,
							    story,
							    index_mask) != 0)
			return -1;
		if (need_split) {
			/*
			 * The insertion divided the gap into two parts.
			 * Old tracker is left in one gap, let's copy tracker
			 * to another.
			 */
			struct gap_item *copy =
				memtx_tx_gap_item_new(item->txn, item->type,
						      item->key,
						      item->part_count);
			if (copy == NULL)
				return -1;

			rlist_add(&story->link[ind].nearby_gaps,
				  &copy->in_nearby_gaps);
		} else if (need_move) {
			/* The tracker must be moved to the left gap. */
			rlist_del(&item->in_nearby_gaps);
			rlist_add(&story->link[ind].nearby_gaps,
				  &item->in_nearby_gaps);
		} else {
			assert((dir > 0 && cmp < 0) ||
			       (cmp < 0 && item->type == ITER_REQ) ||
			       (cmp == 0 && ((dir > 0 && is_full_key) ||
					     item->type == ITER_GT)));
		}
	}
	return 0;
}

/**
 * Helper of @sa memtx_tx_history_add_stmt, does actual work in case when
 * new_tuple != NULL.
 * Just for understanding, that might be:
 * REPLACE, and old_tuple is NULL because it is unknown yet.
 * INSERT, and old_tuple is NULL because there's no such tuple.
 * UPDATE, and old_tuple is not NULL and is the updated tuple.
 */
static int
memtx_tx_history_add_insert_stmt(struct txn_stmt *stmt,
				 struct tuple *old_tuple,
				 struct tuple *new_tuple,
				 enum dup_replace_mode mode,
				 struct tuple **result)
{
	assert(new_tuple != NULL);

	struct space *space = stmt->space;
	struct memtx_story *add_story = NULL;
	struct memtx_story *created_story = NULL, *replaced_story = NULL;

	struct tuple *directly_replaced[space->index_count];
	struct tuple *direct_successor[space->index_count];
	uint32_t directly_replaced_count = 0;

	for (uint32_t i = 0; i < space->index_count; i++) {
		struct index *index = space->index[i];
		struct tuple **replaced = &directly_replaced[i];
		struct tuple **successor = &direct_successor[i];
		if (index_replace(index, NULL, new_tuple,
				  DUP_REPLACE_OR_INSERT,
				  replaced, successor) != 0)
		{
			directly_replaced_count = i;
			goto fail;
		}
	}
	directly_replaced_count = space->index_count;
	struct tuple *replaced = directly_replaced[0];

	/* Check overwritten tuple */
	bool is_own_change = false;
	int rc = check_dup(stmt, new_tuple, directly_replaced,
			   &old_tuple, mode, &is_own_change);
	if (rc != 0)
		goto fail;

	/*
	 * Create add_story and replaced_story if necessary.
	 * Note that despite the tuple is already in pk,
	 * it is not referenced to it, so we pass false as the last argument.
	 */
	add_story = memtx_tx_story_new(space, new_tuple, false);
	if (add_story == NULL)
		goto fail;
	memtx_tx_story_link_added_by(add_story, stmt);

	if (replaced != NULL && !tuple_has_flag(replaced, TUPLE_IS_DIRTY)) {
		/*
		 * Note that despite the tuple is not in pk,
		 * it is referenced to it, so we pass true as the last argument.
		 */
		created_story = memtx_tx_story_new(space, replaced, true);
		if (created_story == NULL)
			goto fail;
		replaced_story = created_story;
		memtx_tx_story_link_top_light(add_story, replaced_story, 0);
	} else if (replaced != NULL) {
		replaced_story = memtx_tx_story_get(replaced);
		memtx_tx_story_link_top_light(add_story, replaced_story, 0);
	} else {
		rc = memtx_tx_handle_gap_write(stmt->txn, space,
					       add_story, new_tuple,
					       direct_successor[0], 0);
		if (rc != 0)
			goto fail;
	}

	/* Collect point phantom read conflicts. */
	for (uint32_t i = 0; i < space->index_count; i++) {
		if (check_hole(space, add_story, new_tuple, i) != 0)
			goto fail;
	}

	if (replaced_story != NULL)
		replaced_story->link[0].in_index = NULL;
	for (uint32_t i = 1; i < space->index_count; i++) {
		if (directly_replaced[i] == NULL) {
			rc = memtx_tx_handle_gap_write(stmt->txn, space,
						       add_story, new_tuple,
						       direct_successor[i], i);
			if (rc != 0)
				goto fail;
			continue;
		}
		assert(tuple_has_flag(directly_replaced[i], TUPLE_IS_DIRTY));
		struct memtx_story *secondary_replaced =
			memtx_tx_story_get(directly_replaced[i]);
		memtx_tx_story_link_top_light(add_story, secondary_replaced, i);
		secondary_replaced->link[i].in_index = NULL;
	}

	if (old_tuple != NULL) {
		assert(tuple_has_flag(old_tuple, TUPLE_IS_DIRTY));

		struct memtx_story *del_story = NULL;
		if (old_tuple == replaced)
			del_story = replaced_story;
		else
			del_story = memtx_tx_story_get(old_tuple);
		memtx_tx_story_link_deleted_by(del_story, stmt);
	} else if (is_own_change)
		stmt->is_pure_insert = true;

	if (new_tuple != NULL) {
		/*
		 * A space holds references to all his tuples.
		 * It's made via primary index - all tuples that are physically
		 * in primary index must be referenced (a replaces tuple must
		 * be dereferenced).
		 */
		assert(add_story == memtx_tx_story_get(new_tuple));

		memtx_tx_ref_to_primary(add_story);
		if (directly_replaced[0] != NULL) {
			assert(replaced_story ==
			       memtx_tx_story_get(directly_replaced[0]));
			memtx_tx_unref_from_primary(replaced_story);
		}
	}

	*result = old_tuple;
	if (*result != NULL) {
		/*
		 * The result must be a referenced pointer. The caller must
		 * unreference it by itself.
		 */
		tuple_ref(*result);
	}
	return 0;

fail:
	if (add_story != NULL) {
		for (uint32_t i = 0; i < add_story->index_count; i++) {
			struct memtx_story_link *link = &add_story->link[i];
			assert(link->newer_story == NULL); (void)link;
			memtx_tx_story_unlink_top_on_rollback_light(add_story,
								    i);
		}
		memtx_tx_story_delete(add_story);
	}
	if (created_story)
		memtx_tx_story_delete(created_story);

	while (directly_replaced_count > 0) {
		uint32_t i = --directly_replaced_count;
		struct index *index = space->index[i];
		struct tuple *unused;
		if (index_replace(index, new_tuple, directly_replaced[i],
				  DUP_INSERT, &unused, &unused) != 0) {
			diag_log();
			unreachable();
			panic("failed to rollback change");
		}
	}

	return -1;
}

/**
 * Helper of @sa memtx_tx_history_add_stmt, does actual work in case when
 * new_tuple == NULL and old_tuple is deleted (and obviously not NULL).
 * Just for understanding, that's a DELETE statement.
 */
static int
memtx_tx_history_add_delete_stmt(struct txn_stmt *stmt,
				 struct tuple *old_tuple,
				 struct tuple **result)
{
	struct space *space = stmt->space;
	struct memtx_story *del_story;

	if (tuple_has_flag(old_tuple, TUPLE_IS_DIRTY)) {
		del_story = memtx_tx_story_get(old_tuple);
	} else {
		assert(stmt->txn != NULL);
		/*
		 * The tuple is not dirty therefore it is placed in pk
		 * and is referenced to it.
		 */
		del_story = memtx_tx_story_new(space, old_tuple, true);
		if (del_story == NULL)
			return -1;
	}
	if (!del_story->tuple_is_retained)
		memtx_tx_story_track_retained_tuple(del_story);

	memtx_tx_story_link_deleted_by(del_story, stmt);
	*result = old_tuple;

	if (*result != NULL) {
		/*
		 * The result must be a referenced pointer. The caller must
		 * unreference it by itself.
		 */
		tuple_ref(*result);
	}
	return 0;
}

int
memtx_tx_history_add_stmt(struct txn_stmt *stmt, struct tuple *old_tuple,
			  struct tuple *new_tuple, enum dup_replace_mode mode,
			  struct tuple **result)
{
	assert(stmt != NULL);
	assert(stmt->space != NULL);
	assert(new_tuple != NULL || old_tuple != NULL);
	assert(new_tuple == NULL || !tuple_has_flag(new_tuple, TUPLE_IS_DIRTY));

	memtx_tx_story_gc();
	if (new_tuple != NULL)
		return memtx_tx_history_add_insert_stmt(stmt, old_tuple,
							new_tuple, mode,
							result);
	else
		return memtx_tx_history_add_delete_stmt(stmt, old_tuple,
							result);
}

/*
 * Relink those who delete this story and make them delete older story.
 */
static void
memtx_tx_history_remove_story_del_stmts(struct memtx_story *story)
{
	struct memtx_story *old_story = story->link[0].older_story;
	while (story->del_stmt) {
		struct txn_stmt *del_stmt = story->del_stmt;

		/* Unlink from old list in any case. */
		story->del_stmt = del_stmt->next_in_del_list;
		del_stmt->next_in_del_list = NULL;
		del_stmt->del_story = NULL;

		/* Link to old story's list. */
		if (old_story != NULL)
			memtx_tx_story_link_deleted_by(old_story,
						       del_stmt);
	}
}

/*
 * Push story down the history chain to the level of prepared stories in each
 * index.
 */
static void
memtx_tx_history_sink_story(struct memtx_story *story)
{
	for (uint32_t i = 0; i < story->index_count; ) {
		struct memtx_story *old_story = story->link[i].older_story;
		if (old_story == NULL || old_story->add_psn != 0 ||
		    old_story->add_stmt == NULL) {
			/* Old story is absent or prepared or committed. */
			i++; /* Go to the next index. */
			continue;
		}
		memtx_tx_story_reorder(story, old_story, i);
	}
}

/*
 * Rollback addition of story by statement.
 */
static void
memtx_tx_history_rollback_added_story(struct txn_stmt *stmt)
{
	assert(stmt->add_story != NULL);
	assert(stmt->add_story->tuple == stmt->rollback_info.new_tuple);
	struct memtx_story *story = stmt->add_story;

	memtx_tx_history_remove_story_del_stmts(story);
	story->add_stmt = NULL;
	stmt->add_story = NULL;

	/*
	 * Sink the story to the end of chain and mark is as deleted long
	 * time ago (with some very low del_psn). After that the story will
	 * be invisible to any reader (that's what is needed) and still be
	 * able to store read set, if necessary.
	 */
	for (uint32_t i = 0; i < story->index_count; ) {
		struct memtx_story *old_story = story->link[i].older_story;
		if (old_story == NULL) {
			/* Old story is absent. */
			i++; /* Go to the next index. */
			continue;
		}
		memtx_tx_story_reorder(story, old_story, i);
	}
	story->del_psn = MEMTX_TX_ROLLBACKED_PSN;
}

/*
 * Rollback deletion of story by statement.
 */
static void
memtx_tx_history_rollback_deleted_story(struct txn_stmt *stmt)
{
	struct memtx_story *story = stmt->del_story;
	/*
	 * There can be no more than one prepared statement deleting a story at
	 * any point in time.
	 */
	assert(story->del_psn == 0 || story->del_psn == stmt->txn->psn);
	story->del_psn = 0;
	memtx_tx_story_unlink_deleted_by(story, stmt);
}

void
memtx_tx_history_rollback_stmt(struct txn_stmt *stmt)
{
	if (stmt->add_story != NULL)
		memtx_tx_history_rollback_added_story(stmt);
	if (stmt->del_story != NULL)
		memtx_tx_history_rollback_deleted_story(stmt);
	assert(stmt->add_story == NULL && stmt->del_story == NULL);
}

/*
 * Completely remove statement that adds a story.
 */
static void
memtx_tx_history_remove_added_story(struct txn_stmt *stmt)
{
	assert(stmt->add_story != NULL);
	assert(stmt->add_story->tuple == stmt->rollback_info.new_tuple);
	struct memtx_story *story = stmt->add_story;
	memtx_tx_history_remove_story_del_stmts(story);
	for (uint32_t i = 0; i < story->index_count; i++)
		memtx_tx_story_unlink_both_on_space_delete(story, i);
	memtx_tx_story_unlink_added_by(story, stmt);
}

/*
 * Completely remove statement that deletes a story.
 */
static void
memtx_tx_history_remove_deleted_story(struct txn_stmt *stmt)
{
	memtx_tx_history_rollback_deleted_story(stmt);
}

/*
 * Completely (as opposed to rollback) remove statement from history.
 */
static void
memtx_tx_history_remove_stmt(struct txn_stmt *stmt)
{
	if (stmt->add_story != NULL)
		memtx_tx_history_remove_added_story(stmt);
	if (stmt->del_story != NULL)
		memtx_tx_history_remove_deleted_story(stmt);
	stmt->engine_savepoint = NULL;
}

/**
 * Helper of memtx_tx_history_prepare_stmt. Do the job in case when
 * stmt->add_story != NULL, that is REPLACE, INSERT, UPDATE etc.
 */
static void
memtx_tx_history_prepare_insert_stmt(struct txn_stmt *stmt)
{
	assert(stmt->add_story != NULL);
	/**
	 * History of a key in an index can consist of several stories.
	 * The list of stories is started with a dirty tuple that is in index.
	 * The list begins with several (or zero) of stories that are added by
	 * in-progress transactions, then the list continues with several
	 * (or zero) of prepared stories, which are followed by several
	 * (or zero) of committed stories, followed by rollbacked stories.
	 * We have the following totally ordered set over tuple stories:
	 *
	 * —————————————————————————————————————————————————> serialization time
	 * |- - - - - - - -|— — — — — -|— — — — — |— — — — — — -|— — — — — — — -
	 * | Rolled back   | Committed | Prepared | In-progress | One dirty
	 * |               |           |          |             | story in index
	 * |- - - - - - - -|— — — — — -| — — — — —|— — — — — — -|— — — — — — — —
	 *
	 * If a statement becomes prepared, the story it adds must be 'sunk' to
	 * the level of prepared stories.
	 */
	struct memtx_story *story = stmt->add_story;
	uint32_t index_count = story->index_count;
	memtx_tx_history_sink_story(story);

	if (stmt->del_story == NULL) {
		/*
		 * This statement replaced nothing. That means that before
		 * this preparation there was no visible tuple in index, and
		 * now there is.
		 * There also can be some in-progress transactions that think
		 * that the replaced nothing. Some of them must be aborted
		 * (for example inserts, that must replace nothing), other
		 * must be told that they replace this tuple now.
		 */
		struct memtx_story_link *link = &story->link[0];
		while (link->newer_story != NULL) {
			struct memtx_story *test = link->newer_story;
			link = &test->link[0];
			struct txn_stmt *test_stmt = test->add_stmt;
			if (test_stmt->txn == stmt->txn)
				continue;
			if (test_stmt->is_pure_insert)
				continue;
			if (test_stmt->del_story != NULL) {
				assert(test_stmt->del_story->add_stmt->txn
				       == test_stmt->txn);
				continue;
			}
			if (test_stmt->does_require_old_tuple)
				memtx_tx_handle_conflict(stmt->txn,
							 test_stmt->txn);

			memtx_tx_story_link_deleted_by(story, test_stmt);
		}
	}

	struct memtx_story *old_story = story->link[0].older_story;
	if (stmt->del_story == NULL)
		assert(old_story == NULL || old_story->del_psn != 0);
	else
		assert(old_story != NULL && stmt->del_story == old_story);
	if (old_story != NULL && old_story->del_psn != 0) {
		assert(stmt->del_story == NULL);
		old_story = NULL;
	}
	if (old_story != NULL) {
		/*
		 * There can be some transactions that want to delete old_story.
		 * It can be this transaction.
		 * All other transactions must be aborted or relinked to delete
		 * this tuple.
		 */
		struct txn_stmt **from = &old_story->del_stmt;
		struct txn_stmt **to = &story->del_stmt;
		while (*to != NULL)
			to = &((*to)->next_in_del_list);
		while (*from != NULL) {
			struct txn_stmt *test_stmt = *from;
			assert(test_stmt->del_story == old_story);
			if (test_stmt->txn == stmt->txn) {
				assert(test_stmt == stmt ||
				       test_stmt->add_story == NULL);
				/*
				 * Statement from the same transaction. Go to
				 * the next statement.
				 */
				from = &test_stmt->next_in_del_list;
				continue;
			}
			assert(test_stmt->txn->psn == 0);
			/* Unlink from old list in any case. */
			*from = test_stmt->next_in_del_list;
			test_stmt->next_in_del_list = NULL;
			test_stmt->del_story = NULL;

			if (test_stmt->does_require_old_tuple)
				memtx_tx_handle_conflict(stmt->txn,
							 test_stmt->txn);

			/* Link to story's list. */
			test_stmt->del_story = story;
			*to = test_stmt;
			to = &test_stmt->next_in_del_list;
		}
	}

	for (uint32_t i = 1; i < story->index_count; i++) {
		struct memtx_story_link *link = &story->link[i];
		while (link->newer_story != NULL) {
			struct memtx_story *test = link->newer_story;
			link = &test->link[i];
			struct txn_stmt *test_stmt = test->add_stmt;
			if (test_stmt->txn == stmt->txn)
				continue;
			if (test_stmt->is_pure_insert)
				continue;
			if (test_stmt->del_story == story)
				continue;
			memtx_tx_handle_conflict(stmt->txn, test_stmt->txn);
			/*
			 * Note that it's a secondary index and there's no
			 * need to call memtx_tx_story_link_deleted_by since
			 * all is done by the primary story chain.
			 */
		}
	}

	for (uint32_t i = 0; i < index_count; i++) {
		old_story = story->link[i].older_story;
		if (old_story == NULL)
			continue;

		struct tx_read_tracker *tracker;
		rlist_foreach_entry(tracker, &old_story->reader_list,
				    in_reader_list) {
			if (tracker->reader == stmt->txn)
				continue;
			uint64_t index_mask = 1ull << (i & 63);
			if ((tracker->index_mask & index_mask) == 0)
				continue;
			memtx_tx_handle_conflict(stmt->txn, tracker->reader);
		}
	}

	for (uint32_t i = 0; i < index_count; i++) {
		for (struct memtx_story *read_story = story;
		     read_story != NULL;
		     read_story = read_story->link[i].newer_story) {
			struct tx_read_tracker *tracker;
			rlist_foreach_entry(tracker, &read_story->reader_list,
					    in_reader_list) {
				if (tracker->reader == stmt->txn)
					continue;
				uint64_t index_mask = 1ull << (i & 63);
				if ((tracker->index_mask & index_mask) == 0)
					continue;
				memtx_tx_handle_conflict(stmt->txn,
							 tracker->reader);
			}
		}
	}
}

/**
 * Helper of memtx_tx_history_prepare_stmt. Do the job in case when
 * stmt->add_story == NULL, that is DELETE etc.
 */
static void
memtx_tx_history_prepare_delete_stmt(struct txn_stmt *stmt)
{
	assert(stmt->add_story == NULL);
	assert(stmt->del_story != NULL);

	struct memtx_story *story = stmt->del_story;
	/*
	 * There can be some transactions that want to delete old_story.
	 * All other transactions must be aborted.
	 */
	struct txn_stmt **itr = &story->del_stmt;
	while (*itr != NULL) {
		struct txn_stmt *test_stmt = *itr;
		assert(test_stmt->del_story == story);
		if (test_stmt->txn == stmt->txn) {
			assert(test_stmt == stmt ||
			       test_stmt->add_story == NULL);
			/*
			 * Statement from the same transaction. Go to the next
			 * statement.
			 */
			itr = &test_stmt->next_in_del_list;
			continue;
		}
		assert(test_stmt->txn->psn == 0);
		/* Unlink from old list in any case. */
		*itr = test_stmt->next_in_del_list;
		test_stmt->next_in_del_list = NULL;
		test_stmt->del_story = NULL;
		/* Conflict only in case of dependance. */
		if (test_stmt->does_require_old_tuple)
			memtx_tx_handle_conflict(stmt->txn, test_stmt->txn);
	}

	struct tx_read_tracker *tracker;
	rlist_foreach_entry(tracker, &story->reader_list, in_reader_list) {
		if (tracker->reader == stmt->txn)
			continue;
		memtx_tx_handle_conflict(stmt->txn, tracker->reader);
	}
}

void
memtx_tx_history_prepare_stmt(struct txn_stmt *stmt)
{
	assert(stmt->txn->psn != 0);

	if (stmt->add_story != NULL) {
		memtx_tx_history_prepare_insert_stmt(stmt);
	} else if (stmt->del_story != NULL) {
		memtx_tx_history_prepare_delete_stmt(stmt);
	}

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

/**
 * Clean and clear all read lists of @a txn.
 */
static void
memtx_tx_clear_txn_read_lists(struct txn *txn);

void
memtx_tx_prepare_finalize(struct txn *txn)
{
	/* Just free all other lists - we don't need 'em anymore. */
	memtx_tx_clear_txn_read_lists(txn);
}

void
memtx_tx_history_commit_stmt(struct txn_stmt *stmt, size_t *bsize)
{
	if (stmt->add_story != NULL) {
		assert(stmt->add_story->add_stmt == stmt);
		*bsize += tuple_bsize(stmt->add_story->tuple);
		memtx_tx_story_unlink_added_by(stmt->add_story, stmt);
	}
	if (stmt->del_story != NULL) {
		*bsize -= tuple_bsize(stmt->del_story->tuple);
		memtx_tx_story_unlink_deleted_by(stmt->del_story, stmt);
	}
	memtx_tx_story_gc();
}

/**
 * Helper of @sa memtx_tx_tuple_clarify.
 * Do actual work.
 */
static struct tuple *
memtx_tx_tuple_clarify_impl(struct txn *txn, struct space *space,
			    struct tuple *tuple, struct index *index,
			    uint32_t mk_index, bool is_prepared_ok)
{
	assert(tuple_has_flag(tuple, TUPLE_IS_DIRTY));
	struct memtx_story *story = memtx_tx_story_get(tuple);
	bool own_change = false;
	struct tuple *result = NULL;

	while (true) {
		if (memtx_tx_story_delete_is_visible(story, txn, is_prepared_ok,
						     &own_change)) {
			result = NULL;
			break;
		}
		if (story->del_psn != 0 && story->del_stmt != NULL &&
		    txn != NULL) {
			assert(story->del_psn == story->del_stmt->txn->psn);
			/*
			 * If we skip deletion of a tuple by prepared
			 * transaction then the transaction must be before
			 * prepared in serialization order.
			 * That can be a read view or conflict already.
			 */
			memtx_tx_handle_conflict(story->del_stmt->txn, txn);
		}

		if (memtx_tx_story_insert_is_visible(story, txn, is_prepared_ok,
						     &own_change)) {
			result = story->tuple;
			break;
		}
		if (story->add_psn != 0 && story->add_stmt != NULL &&
		    txn != NULL) {
			assert(story->add_psn == story->add_stmt->txn->psn);
			/*
			 * If we skip addition of a tuple by prepared
			 * transaction then the transaction must be before
			 * prepared in serialization order.
			 * That can be a read view or conflict already.
			 */
			memtx_tx_handle_conflict(story->add_stmt->txn, txn);
		}

		if (story->link[index->dense_id].older_story == NULL)
			break;
		story = story->link[index->dense_id].older_story;
	}
	if (!own_change) {
		/*
		 * If the result tuple exists (is visible) - it is visible in
		 * every index. But if we found a story of deleted tuple - we
		 * should record that only in the given index this transaction
		 * have found nothing by this key.
		 */
		int shift = index->dense_id & 63;
		uint64_t mask = result == NULL ? 1ull << shift : UINT64_MAX;
		memtx_tx_track_read_story(txn, space, story, mask);
	}
	if (mk_index != 0) {
		assert(false); /* TODO: multiindex */
		panic("multikey indexes are not supported int TX manager");
	}
	return result;
}

/**
 * Helper of @sa memtx_tx_tuple_clarify.
 * Detect whether the transaction can see prepared, but unconfirmed commits.
 */
static bool
detect_whether_prepared_ok(struct txn *txn)
{
	if (txn == NULL)
		return false;
	else if (txn->isolation == TXN_ISOLATION_READ_COMMITTED)
		return true;
	else if (txn->isolation == TXN_ISOLATION_READ_CONFIRMED ||
		 txn->isolation == TXN_ISOLATION_LINEARIZABLE)
		return false;
	assert(txn->isolation == TXN_ISOLATION_BEST_EFFORT);
	/*
	 * The best effort that we can make is to determine whether the
	 * transaction is read-only or not. For read only (including autocommit
	 * select, that is txn == NULL) we should see only confirmed changes,
	 * ignoring prepared. For read-write transaction we should see prepared
	 * changes in order to avoid conflicts.
	 */
	return !stailq_empty(&txn->stmts);
}

/**
 * Helper of @sa memtx_tx_tuple_clarify.
 * Detect is_prepared_ok flag and pass the job to memtx_tx_tuple_clarify_impl.
 */
struct tuple *
memtx_tx_tuple_clarify_slow(struct txn *txn, struct space *space,
			    struct tuple *tuple, struct index *index,
			    uint32_t mk_index)
{
	if (!tuple_has_flag(tuple, TUPLE_IS_DIRTY)) {
		memtx_tx_track_read(txn, space, tuple);
		return tuple;
	}
	bool is_prepared_ok = detect_whether_prepared_ok(txn);
	struct tuple *res =
		memtx_tx_tuple_clarify_impl(txn, space, tuple, index, mk_index,
					    is_prepared_ok);
	return res;
}

uint32_t
memtx_tx_index_invisible_count_slow(struct txn *txn,
				    struct space *space, struct index *index)
{
	uint32_t res = 0;
	struct memtx_story *story;
	rlist_foreach_entry(story, &space->memtx_stories, in_space_stories) {
		assert(index->dense_id < story->index_count);
		struct memtx_story_link *link = &story->link[index->dense_id];
		/*
		 * A history chain is represented by the top story, which is
		 * stored in index.
		 */
		if (link->in_index == NULL) {
			assert(link->newer_story != NULL);
			continue;
		}
		assert(link->newer_story == NULL);

		struct tuple *visible = NULL;
		bool is_prepared_ok = detect_whether_prepared_ok(txn);
		bool unused;
		memtx_tx_story_find_visible_tuple(story, txn, index->dense_id,
						  is_prepared_ok, &visible,
						  &unused);
		if (visible == NULL)
			res++;
	}
	memtx_tx_story_gc();
	return res;
}

static void
memtx_tx_delete_gap(struct gap_item *item)
{
	rlist_del(&item->in_gap_list);
	rlist_del(&item->in_nearby_gaps);
	memtx_tx_mempool_free(item->txn, &txm.gap_item_mempoool, item);
}

static void
memtx_tx_full_scan_item_delete(struct full_scan_item *item)
{
	rlist_del(&item->in_full_scan_list);
	rlist_del(&item->in_full_scans);
	memtx_tx_mempool_free(item->txn, &txm.full_scan_item_mempool, item);
}

void
memtx_tx_on_index_delete(struct index *index)
{
	while (!rlist_empty(&index->nearby_gaps)) {
		struct gap_item *item =
			rlist_first_entry(&index->nearby_gaps,
					  struct gap_item,
					  in_nearby_gaps);
		memtx_tx_delete_gap(item);
	}
	while (!rlist_empty(&index->full_scans)) {
		struct full_scan_item *item =
			rlist_first_entry(&index->full_scans,
					  struct full_scan_item,
					  in_full_scans);
		memtx_tx_full_scan_item_delete(item);
	}
	memtx_tx_story_gc();
}

void
memtx_tx_on_space_delete(struct space *space)
{
	while (!rlist_empty(&space->memtx_stories)) {
		struct memtx_story *story
			= rlist_first_entry(&space->memtx_stories,
					    struct memtx_story,
					    in_space_stories);
		/*
		 * Space is to be altered (not necessarily dropped). Since
		 * this operation is considered to be DDL, all other
		 * transactions will be aborted anyway. We can't postpone
		 * rollback till actual call of commit/rollback since stories
		 * should be destroyed immediately.
		 */
		if (story->add_stmt != NULL)
			memtx_tx_history_remove_stmt(story->add_stmt);
		if (story->del_stmt != NULL)
			memtx_tx_history_remove_stmt(story->del_stmt);
		memtx_tx_story_full_unlink_on_space_delete(story);
		memtx_tx_story_delete(story);
	}
}

/**
 * Allocate and initialize tx_read_tracker, return NULL in case of error
 * (diag is set). Links in lists are not initialized though.
 */
static struct tx_read_tracker *
tx_read_tracker_new(struct txn *reader, struct memtx_story *story,
		    uint64_t index_mask)
{
	struct tx_read_tracker *tracker;
	tracker = memtx_tx_region_alloc_object(reader,
					       MEMTX_TX_OBJECT_READ_TRACKER);
	if (tracker == NULL) {
		diag_set(OutOfMemory, sizeof(*tracker), "tx region",
			 "read_tracker");
		return NULL;
	}
	tracker->reader = reader;
	tracker->story = story;
	tracker->index_mask = index_mask;
	return tracker;
}

static int
memtx_tx_track_read_story_slow(struct txn *txn, struct memtx_story *story,
			       uint64_t index_mask)
{
	assert(story != NULL);
	struct tx_read_tracker *tracker = NULL;

	struct rlist *r1 = story->reader_list.next;
	struct rlist *r2 = txn->read_set.next;
	while (r1 != &story->reader_list && r2 != &txn->read_set) {
		tracker = rlist_entry(r1, struct tx_read_tracker,
				      in_reader_list);
		assert(tracker->story == story);
		if (tracker->reader == txn)
			break;
		tracker = rlist_entry(r2, struct tx_read_tracker,
				      in_read_set);
		assert(tracker->reader == txn);
		if (tracker->story == story)
			break;
		tracker = NULL;
		r1 = r1->next;
		r2 = r2->next;
	}
	if (tracker != NULL) {
		/* Move to the beginning of a list for faster further lookups.*/
		rlist_del(&tracker->in_reader_list);
		rlist_del(&tracker->in_read_set);
		tracker->index_mask |= index_mask;
	} else {
		tracker = tx_read_tracker_new(txn, story, index_mask);
		if (tracker == NULL)
			return -1;
	}
	rlist_add(&story->reader_list, &tracker->in_reader_list);
	rlist_add(&txn->read_set, &tracker->in_read_set);
	return 0;
}

static int
memtx_tx_track_read_story(struct txn *txn, struct space *space,
			  struct memtx_story *story, uint64_t index_mask)
{
	if (txn == NULL)
		return 0;
	if (space == NULL)
		return 0;
	if (space->def->opts.is_ephemeral)
		return 0;
	return memtx_tx_track_read_story_slow(txn, story, index_mask);
}

/**
 * Record in TX manager that a transaction @txn have read a @tuple in @space.
 *
 * NB: can trigger story garbage collection.
 *
 * @return 0 on success, -1 on memory error.
 */
static int
memtx_tx_track_read(struct txn *txn, struct space *space, struct tuple *tuple)
{
	if (tuple == NULL)
		return 0;
	if (txn == NULL)
		return 0;
	if (space == NULL)
		return 0;
	if (space->def->opts.is_ephemeral)
		return 0;

	if (tuple_has_flag(tuple, TUPLE_IS_DIRTY)) {
		struct memtx_story *story = memtx_tx_story_get(tuple);
		return memtx_tx_track_read_story(txn, space, story, UINT64_MAX);
	} else {
		/*
		 * The tuple is not dirty therefore it is placed in pk
		 * and is referenced to it.
		 */
		struct memtx_story *story =
			memtx_tx_story_new(space, tuple, true);
		if (story == NULL)
			return -1;
		struct tx_read_tracker *tracker;
		tracker = tx_read_tracker_new(txn, story, UINT64_MAX);
		if (tracker == NULL) {
			memtx_tx_story_delete(story);
			return -1;
		}
		rlist_add(&story->reader_list, &tracker->in_reader_list);
		rlist_add(&txn->read_set, &tracker->in_read_set);
		return 0;
	}
}

/**
 * Create new point_hole_item by given argumnets and put it to hash table.
 */
static int
point_hole_storage_new(struct index *index, const char *key,
		       size_t key_len, struct txn *txn)
{
	struct memtx_tx_mempool *pool = &txm.point_hole_item_pool;
	struct point_hole_item *object = memtx_tx_mempool_alloc(txn, pool);
	if (object == NULL) {
		diag_set(OutOfMemory, sizeof(*object),
			 "mempool_alloc", "point_hole_item");
		return -1;
	}

	rlist_create(&object->ring);
	rlist_create(&object->in_point_holes_list);
	object->txn = txn;
	object->index_unique_id = index->unique_id;
	if (key_len <= sizeof(object->short_key)) {
		object->key = object->short_key;
	} else {
		object->key = memtx_tx_region_alloc(txn, key_len,
						    MEMTX_TX_ALLOC_TRACKER);
		if (object->key == NULL) {
			memtx_tx_mempool_free(txn, pool, object);
			diag_set(OutOfMemory, key_len, "tx region",
				 "point key");
			return -1;
		}
	}
	memcpy((char *)object->key, key, key_len);
	object->key_len = key_len;
	object->is_head = true;

	struct key_def *def = index->def->key_def;
	object->hash = object->index_unique_id ^ def->key_hash(key, def);

	const struct point_hole_item **put =
		(const struct point_hole_item **) &object;
	struct point_hole_item *replaced = NULL;
	struct point_hole_item **preplaced = &replaced;
	mh_point_holes_put(txm.point_holes, put, &preplaced, 0);
	if (preplaced != NULL) {
		/*
		 * The item in hash table was overwitten. It's OK, but
		 * we need replaced item to the item list.
		 * */
		rlist_add(&replaced->ring, &object->ring);
		assert(replaced->is_head);
		replaced->is_head = false;
	} else {
		txm.point_holes_size++;
	}
	rlist_add(&txn->point_holes_list, &object->in_point_holes_list);
	return 0;
}

static void
point_hole_storage_delete(struct point_hole_item *object)
{
	if (!object->is_head) {
		/*
		 * The deleting item is linked list, and the hash table
		 * doesn't point directly to this item. Delete from the
		 * list and that's enough.
		 */
		assert(!rlist_empty(&object->ring));
		rlist_del(&object->ring);
	} else if (!rlist_empty(&object->ring)) {
		/*
		 * Hash table point to this item, but there are more
		 * items in the list. Relink the hash table with any other
		 * item in the list, and delete this item from the list.
		 */
		struct point_hole_item *another =
			rlist_next_entry(object, ring);

		const struct point_hole_item **put =
			(const struct point_hole_item **) &another;
		struct point_hole_item *replaced = NULL;
		struct point_hole_item **preplaced = &replaced;
		mh_point_holes_put(txm.point_holes, put, &preplaced, 0);
		assert(replaced == object);
		rlist_del(&object->ring);
		another->is_head = true;
	} else {
		/*
		 * Hash table point to this item, and it's the last in the
		 * list. We have to remove the item from the hash table.
		 */
		int exist = 0;
		const struct point_hole_item **put =
			(const struct point_hole_item **) &object;
		mh_int_t pos = mh_point_holes_put_slot(txm.point_holes, put,
						       &exist, 0);
		assert(exist);
		mh_point_holes_del(txm.point_holes, pos, 0);
		txm.point_holes_size--;
	}
	rlist_del(&object->in_point_holes_list);
	struct memtx_tx_mempool *pool = &txm.point_hole_item_pool;
	memtx_tx_mempool_free(object->txn, pool, object);
}

/**
 * Record in TX manager that a transaction @a txn have read a nothing
 * from @a space and @a index with @a key.
 * The key is expected to be full, that is has part count equal to part
 * count in unique cmp_key of the index.
 * @return 0 on success, -1 on memory error.
 */
int
memtx_tx_track_point_slow(struct txn *txn, struct index *index, const char *key)
{
	if (txn->status != TXN_INPROGRESS)
		return 0;

	struct key_def *def = index->def->key_def;
	const char *tmp = key;
	for (uint32_t i = 0; i < def->part_count; i++)
		mp_next(&tmp);
	size_t key_len = tmp - key;
	memtx_tx_story_gc();
	return point_hole_storage_new(index, key, key_len, txn);
}

static struct gap_item *
memtx_tx_gap_item_new(struct txn *txn, enum iterator_type type,
		      const char *key, uint32_t part_count)
{
	struct gap_item *item =
		memtx_tx_mempool_alloc(txn, &txm.gap_item_mempoool);
	if (item == NULL) {
		diag_set(OutOfMemory, sizeof(*item), "mempool_alloc", "gap");
		return NULL;
	}

	item->txn = txn;
	item->type = type;
	item->part_count = part_count;
	const char *tmp = key;
	for (uint32_t i = 0; i < part_count; i++)
		mp_next(&tmp);
	item->key_len = tmp - key;
	if (part_count == 0) {
		item->key = NULL;
	} else if (item->key_len <= sizeof(item->short_key)) {
		item->key = item->short_key;
	} else {
		item->key = memtx_tx_region_alloc(txn, item->key_len,
						  MEMTX_TX_ALLOC_TRACKER);
		if (item->key == NULL) {
			memtx_tx_mempool_free(txn,
					      &txm.gap_item_mempoool, item);
			diag_set(OutOfMemory, item->key_len, "tx region",
				 "point key");
			return NULL;
		}
	}
	memcpy((char *)item->key, key, item->key_len);
	rlist_add(&txn->gap_list, &item->in_gap_list);
	return item;
}

/**
 * Record in TX manager that a transaction @a txn have read nothing
 * from @a space and @a index with @a key, somewhere from interval between
 * some unknown predecessor and @a successor.
 * This function must be used for ordered indexes, such as TREE, for queries
 * when iteration type is not EQ or when the key is not full (otherwise
 * it's faster to use memtx_tx_track_point).
 *
 * @return 0 on success, -1 on memory error.
 */
int
memtx_tx_track_gap_slow(struct txn *txn, struct space *space, struct index *index,
			struct tuple *successor, enum iterator_type type,
			const char *key, uint32_t part_count)
{
	if (txn->status != TXN_INPROGRESS)
		return 0;

	struct gap_item *item = memtx_tx_gap_item_new(txn, type, key,
						      part_count);
	if (item == NULL)
		return -1;

	if (successor != NULL) {
		struct memtx_story *story;
		if (tuple_has_flag(successor, TUPLE_IS_DIRTY)) {
			story = memtx_tx_story_get(successor);
		} else {
			/*
			 * The tuple is not dirty therefore it is placed in pk
			 * and is referenced to it.
			 */
			story = memtx_tx_story_new(space, successor, true);
			if (story == NULL) {
				memtx_tx_mempool_free(txn,
						      &txm.gap_item_mempoool,
						      item);
				return -1;
			}
		}
		assert(index->dense_id < story->index_count);
		assert(story->link[index->dense_id].in_index != NULL);
		rlist_add(&story->link[index->dense_id].nearby_gaps,
			  &item->in_nearby_gaps);
	} else {
		rlist_add(&index->nearby_gaps, &item->in_nearby_gaps);
	}
	memtx_tx_story_gc();
	return 0;
}

static struct full_scan_item *
memtx_tx_full_scan_item_new(struct txn *txn)
{
	struct full_scan_item *item =
		(struct full_scan_item *)memtx_tx_mempool_alloc(txn,
			&txm.full_scan_item_mempool);
	if (item == NULL) {
		diag_set(OutOfMemory, sizeof(*item), "mempool_alloc",
			 "full_scan_item");
		return NULL;
	}

	item->txn = txn;
	rlist_add(&txn->full_scan_list, &item->in_full_scan_list);
	return item;
}

/**
 * Record in TX manager that a transaction @a txn have read full @a index.
 * This function must be used for unordered indexes, such as HASH, for queries
 * when iteration type is ALL.
 *
 * @return 0 on success, -1 on memory error.
 */
int
memtx_tx_track_full_scan_slow(struct txn *txn, struct index *index)
{
	if (txn->status != TXN_INPROGRESS)
		return 0;

	struct full_scan_item *item = memtx_tx_full_scan_item_new(txn);
	if (item == NULL)
		return -1;

	rlist_add(&index->full_scans, &item->in_full_scans);
	memtx_tx_story_gc();
	return 0;
}

/**
 * Clean and clear all read lists of @a txn.
 */
static void
memtx_tx_clear_txn_read_lists(struct txn *txn)
{
	while (!rlist_empty(&txn->point_holes_list)) {
		struct point_hole_item *object =
			rlist_first_entry(&txn->point_holes_list,
					  struct point_hole_item,
					  in_point_holes_list);
		point_hole_storage_delete(object);
	}
	while (!rlist_empty(&txn->gap_list)) {
		struct gap_item *item =
			rlist_first_entry(&txn->gap_list,
					  struct gap_item,
					  in_gap_list);
		memtx_tx_delete_gap(item);
	}
	while (!rlist_empty(&txn->full_scan_list)) {
		struct full_scan_item *item =
			rlist_first_entry(&txn->full_scan_list,
					  struct full_scan_item,
					  in_full_scan_list);
		memtx_tx_full_scan_item_delete(item);
	}

	struct tx_read_tracker *tracker, *tmp;
	rlist_foreach_entry_safe(tracker, &txn->read_set,
				 in_read_set, tmp) {
		rlist_del(&tracker->in_reader_list);
		rlist_del(&tracker->in_read_set);
	}
	assert(rlist_empty(&txn->read_set));

	rlist_del(&txn->in_read_view_txs);
}

/**
 * Clean memtx_tx part of @a txm.
 */
void
memtx_tx_clean_txn(struct txn *txn)
{
	memtx_tx_clear_txn_read_lists(txn);

	rlist_del(&txn->in_all_txs);

	memtx_tx_story_gc();
}

static uint32_t
memtx_tx_snapshot_cleaner_hash(const struct tuple *a)
{
	uintptr_t u = (uintptr_t)a;
	if (sizeof(uintptr_t) <= sizeof(uint32_t))
		return u;
	else
		return u ^ (u >> 32);
}

struct memtx_tx_snapshot_cleaner_entry
{
	struct tuple *from;
	struct tuple *to;
};

#define mh_name _snapshot_cleaner
#define mh_key_t struct tuple *
#define mh_node_t struct memtx_tx_snapshot_cleaner_entry
#define mh_arg_t int
#define mh_hash(a, arg) (memtx_tx_snapshot_cleaner_hash((a)->from))
#define mh_hash_key(a, arg) (memtx_tx_snapshot_cleaner_hash(a))
#define mh_cmp(a, b, arg) (((a)->from) != ((b)->from))
#define mh_cmp_key(a, b, arg) ((a) != ((b)->from))
#define MH_SOURCE
#include "salad/mhash.h"

void
memtx_tx_snapshot_cleaner_create(struct memtx_tx_snapshot_cleaner *cleaner,
				 struct space *space)
{
	cleaner->ht = NULL;
	if (rlist_empty(&space->memtx_stories) &&
	    rlist_empty(&space->alter_stmts))
		return;
	struct mh_snapshot_cleaner_t *ht = mh_snapshot_cleaner_new();
	struct memtx_story *story;
	rlist_foreach_entry(story, &space->memtx_stories, in_space_stories) {
		struct tuple *tuple = story->tuple;
		struct tuple *clean =
			memtx_tx_tuple_clarify_impl(NULL, space, tuple,
						    space->index[0], 0, true);
		if (clean == tuple)
			continue;

		struct memtx_tx_snapshot_cleaner_entry entry;
		entry.from = tuple;
		entry.to = clean;
		mh_snapshot_cleaner_put(ht,  &entry, NULL, 0);
	}
	struct space_alter_stmt *alter_stmt;
	rlist_foreach_entry(alter_stmt, &space->alter_stmts, link) {
		struct memtx_tx_snapshot_cleaner_entry entry;
		entry.from = alter_stmt->new_tuple;
		entry.to = alter_stmt->old_tuple;
		mh_snapshot_cleaner_put(ht, &entry, NULL, 0);
	}
	cleaner->ht = ht;
}

struct tuple *
memtx_tx_snapshot_clarify_slow(struct memtx_tx_snapshot_cleaner *cleaner,
			       struct tuple *tuple)
{
	assert(cleaner->ht != NULL);

	struct mh_snapshot_cleaner_t *ht = cleaner->ht;
	while (true) {
		mh_int_t pos =  mh_snapshot_cleaner_find(ht, tuple, 0);
		if (pos == mh_end(ht))
			break;
		struct memtx_tx_snapshot_cleaner_entry *entry =
			mh_snapshot_cleaner_node(ht, pos);
		assert(entry->from == tuple);
		tuple = entry->to;
	}
	return tuple;
}

void
memtx_tx_snapshot_cleaner_destroy(struct memtx_tx_snapshot_cleaner *cleaner)
{
	if (cleaner->ht != NULL)
		mh_snapshot_cleaner_delete(cleaner->ht);
}